Prove Notarius cache reuse end to end
This commit is contained in:
@@ -282,7 +282,7 @@ func validateValidationSummaries(documents []validationSummaryDocument) ([]Valid
|
||||
if document.Status != "complete" && document.Status != "rejected" && document.Status != "incomplete" {
|
||||
return nil, fmt.Errorf("notarius validation summary status %q is invalid", document.Status)
|
||||
}
|
||||
if document.ProducerAttemptCount == nil || *document.ProducerAttemptCount <= 0 || !validTerminalAction(document.TerminalAction) {
|
||||
if document.ProducerAttemptCount == nil || *document.ProducerAttemptCount < 0 || !validTerminalAction(document.TerminalAction) {
|
||||
return nil, fmt.Errorf("notarius validation summary is missing required fields")
|
||||
}
|
||||
summaries = append(summaries, ValidationSummary{
|
||||
|
||||
@@ -766,6 +766,26 @@ func assertRedactedCredentialBytes(t *testing.T, data []byte, canary string) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateValidationSummariesAcceptsReusedProducerWithoutAttempts(t *testing.T) {
|
||||
zero := 0
|
||||
summaries, err := validateValidationSummaries([]validationSummaryDocument{{
|
||||
Stage: "chunk", Status: "complete", ProducerAttemptCount: &zero, TerminalAction: "accepted",
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatalf("validateValidationSummaries() error = %v", err)
|
||||
}
|
||||
if len(summaries) != 1 || summaries[0].ProducerAttemptCount != 0 {
|
||||
t.Fatalf("summaries = %#v, want one reused producer with zero attempts", summaries)
|
||||
}
|
||||
|
||||
negative := -1
|
||||
if _, err := validateValidationSummaries([]validationSummaryDocument{{
|
||||
Stage: "chunk", Status: "complete", ProducerAttemptCount: &negative, TerminalAction: "accepted",
|
||||
}}); err == nil {
|
||||
t.Fatal("validateValidationSummaries() negative producer attempts error = nil")
|
||||
}
|
||||
}
|
||||
|
||||
func writeValidBundleAndReceipt(t *testing.T, req RunRequest, includeUnknown bool) {
|
||||
t.Helper()
|
||||
bundle := filepath.Join(req.OutputRoot, "notarius-run-1")
|
||||
|
||||
379
internal/stage/notarius_cache_integration_test.go
Normal file
379
internal/stage/notarius_cache_integration_test.go
Normal file
@@ -0,0 +1,379 @@
|
||||
package stage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||
)
|
||||
|
||||
const notariusIntegrationBinaryEnv = "NARRATIO_TEST_NOTARIUS_BINARY"
|
||||
|
||||
// Run with:
|
||||
// NARRATIO_TEST_NOTARIUS_BINARY=/absolute/path/to/notarius GOWORK=off go test ./internal/stage -run '^TestNotariusCacheReuseIntegration$' -count=1 -v
|
||||
func TestNotariusCacheReuseIntegration(t *testing.T) {
|
||||
binary := strings.TrimSpace(os.Getenv(notariusIntegrationBinaryEnv))
|
||||
if binary == "" {
|
||||
t.Skipf("set %s to an absolute Notarius v0.7.0-compatible executable to run this integration test", notariusIntegrationBinaryEnv)
|
||||
}
|
||||
if !filepath.IsAbs(binary) {
|
||||
t.Fatalf("%s = %q, want an absolute path", notariusIntegrationBinaryEnv, binary)
|
||||
}
|
||||
info, err := os.Stat(binary)
|
||||
if err != nil {
|
||||
t.Fatalf("inspect %s: %v", notariusIntegrationBinaryEnv, err)
|
||||
}
|
||||
if !info.Mode().IsRegular() || info.Mode().Perm()&0o111 == 0 {
|
||||
t.Fatalf("%s = %q, want an executable regular file", notariusIntegrationBinaryEnv, binary)
|
||||
}
|
||||
|
||||
t.Run("reuse reference identity and forced refresh", func(t *testing.T) {
|
||||
provider := newNotariusIntegrationProvider(t)
|
||||
env, runManifest, fixture := newNotariusIntegrationFixture(t, binary, provider.URL(), "dnd-session")
|
||||
|
||||
first := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010101Z-a0000001", false)
|
||||
if got := provider.Calls(); got != 2 {
|
||||
t.Fatalf("first provider calls = %d, want 2", got)
|
||||
}
|
||||
if first.ChunkPlan == nil || first.ChunkPlan.Action != "generated" {
|
||||
t.Fatalf("first chunk plan = %#v, want generated", first.ChunkPlan)
|
||||
}
|
||||
|
||||
env.Config.Pipeline.Notarius.Resume = true
|
||||
second := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010102Z-a0000002", false)
|
||||
if got := provider.Calls(); got != 2 {
|
||||
t.Fatalf("resumed provider calls = %d, want unchanged count 2", got)
|
||||
}
|
||||
assertNotariusIntegrationDecision(t, second, "extract", "spells-first", "reused", "checkpoint_reused")
|
||||
assertNotariusIntegrationDecision(t, second, "extract", "spells-second", "reused", "checkpoint_reused")
|
||||
|
||||
oldSnapshot := notariusIntegrationSnapshotPath(t, env, runManifest, artifactpolicy.SourceInputParty)
|
||||
setNotariusIntegrationReference(t, runManifest, fixture.partyPath, "The party now travels with a silver dragon.\n")
|
||||
third := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010103Z-a0000003", false)
|
||||
if got := provider.Calls(); got != 4 {
|
||||
t.Fatalf("changed-reference provider calls = %d, want 4", got)
|
||||
}
|
||||
assertNotariusIntegrationDecision(t, third, "extract", "spells-first", "executed", "checkpoint_missing")
|
||||
newSnapshot := notariusIntegrationSnapshotPath(t, env, runManifest, artifactpolicy.SourceInputParty)
|
||||
if newSnapshot == oldSnapshot {
|
||||
t.Fatalf("changed reference retained snapshot %q", newSnapshot)
|
||||
}
|
||||
for _, path := range []string{oldSnapshot, newSnapshot} {
|
||||
if _, err := os.Stat(path); err != nil {
|
||||
t.Fatalf("snapshot %q is unavailable: %v", path, err)
|
||||
}
|
||||
}
|
||||
|
||||
setNotariusIntegrationTranscript(t, runManifest, fixture.transcriptPath, notariusIntegrationTranscript("session-unrelated", "The wizard studies Magic Missile."))
|
||||
unrelated := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010104Z-a0000004", false)
|
||||
if unrelated.ChunkPlan == nil || unrelated.ChunkPlan.Action != "generated" {
|
||||
t.Fatalf("unrelated chunk plan = %#v, want generated", unrelated.ChunkPlan)
|
||||
}
|
||||
unrelatedPlan := filepath.Join(fixture.chunkRoot, strings.TrimPrefix(unrelated.ChunkPlan.SourceDigest, "sha256:"), "plan.json")
|
||||
unrelatedBefore, err := os.ReadFile(unrelatedPlan)
|
||||
if err != nil {
|
||||
t.Fatalf("read unrelated chunk plan: %v", err)
|
||||
}
|
||||
|
||||
setNotariusIntegrationTranscript(t, runManifest, fixture.transcriptPath, notariusIntegrationTranscript("session-cache", "Aria casts Cure Wounds."))
|
||||
forced := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T010105Z-a0000005", true)
|
||||
if got := provider.Calls(); got != 8 {
|
||||
t.Fatalf("forced provider calls = %d, want 8", got)
|
||||
}
|
||||
if len(forced.CheckpointDecisions) != 0 {
|
||||
t.Fatalf("forced extraction loaded checkpoint state: %#v", forced.CheckpointDecisions)
|
||||
}
|
||||
if forced.ChunkPlan == nil || forced.ChunkPlan.Action != "refreshed" {
|
||||
t.Fatalf("forced chunk plan = %#v, want refreshed", forced.ChunkPlan)
|
||||
}
|
||||
unrelatedAfter, err := os.ReadFile(unrelatedPlan)
|
||||
if err != nil {
|
||||
t.Fatalf("read unrelated chunk plan after refresh: %v", err)
|
||||
}
|
||||
if string(unrelatedAfter) != string(unrelatedBefore) {
|
||||
t.Fatal("forced refresh changed the unrelated input's cached plan")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("partial failure resumes accepted predecessor", func(t *testing.T) {
|
||||
provider := newNotariusIntegrationProvider(t)
|
||||
provider.FailRequest(2)
|
||||
env, runManifest, _ := newNotariusIntegrationFixture(t, binary, provider.URL(), "dnd-recovery")
|
||||
env.Config.Pipeline.Notarius.Resume = true
|
||||
|
||||
runManifest.RunID = "20260916T020101Z-b0000001"
|
||||
env.Force = false
|
||||
if result, err := (extractStage{}).Run(context.Background(), env, runManifest); err == nil || result != nil {
|
||||
t.Fatalf("partial extraction result = %#v error = %v, want failure", result, err)
|
||||
}
|
||||
if got := provider.Calls(); got != 2 {
|
||||
t.Fatalf("failed invocation provider calls = %d, want 2", got)
|
||||
}
|
||||
|
||||
provider.FailRequest(0)
|
||||
recovered := runNotariusIntegrationExtraction(t, env, runManifest, "20260916T020102Z-b0000002", false)
|
||||
if got := provider.Calls(); got != 3 {
|
||||
t.Fatalf("recovery provider calls = %d, want one additional call (3 total)", got)
|
||||
}
|
||||
assertNotariusIntegrationDecision(t, recovered, "extract", "spells-first", "reused", "checkpoint_reused")
|
||||
assertNotariusIntegrationDecision(t, recovered, "extract", "spells-second", "executed", "")
|
||||
})
|
||||
}
|
||||
|
||||
type notariusIntegrationProvider struct {
|
||||
server *httptest.Server
|
||||
calls atomic.Int64
|
||||
failAt atomic.Int64
|
||||
}
|
||||
|
||||
func newNotariusIntegrationProvider(t *testing.T) *notariusIntegrationProvider {
|
||||
t.Helper()
|
||||
provider := ¬ariusIntegrationProvider{}
|
||||
provider.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
|
||||
call := provider.calls.Add(1)
|
||||
if request.Method != http.MethodPost || request.URL.Path != "/v1/chat/completions" {
|
||||
t.Errorf("provider request = %s %s, want POST /v1/chat/completions", request.Method, request.URL.Path)
|
||||
http.Error(w, "unexpected request", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
if provider.failAt.Load() == call {
|
||||
http.Error(w, "controlled provider failure", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"id":"local-completion","object":"chat.completion","created":1,"model":"local-model","choices":[{"index":0,"message":{"role":"assistant","content":"{\"spell_casts\":[]}"},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":4,"total_tokens":14}}`))
|
||||
}))
|
||||
t.Cleanup(provider.server.Close)
|
||||
return provider
|
||||
}
|
||||
|
||||
func (p *notariusIntegrationProvider) URL() string { return p.server.URL }
|
||||
func (p *notariusIntegrationProvider) Calls() int64 { return p.calls.Load() }
|
||||
func (p *notariusIntegrationProvider) FailRequest(request int64) { p.failAt.Store(request) }
|
||||
|
||||
type notariusIntegrationFixture struct {
|
||||
transcriptPath string
|
||||
partyPath string
|
||||
chunkRoot string
|
||||
}
|
||||
|
||||
func newNotariusIntegrationFixture(t *testing.T, binary, endpoint, pipelineID string) (*Env, *manifest.Manifest, notariusIntegrationFixture) {
|
||||
t.Helper()
|
||||
env, runManifest, _ := setupExtractEnv(t)
|
||||
initialOutput := artifacts.SessionRunNotariusOutputRootForCampaign(
|
||||
env.Config.Pipeline.Workspace.Root, runManifest.Campaign, runManifest.SessionID, runManifest.RunID,
|
||||
)
|
||||
if err := os.RemoveAll(initialOutput); err != nil {
|
||||
t.Fatalf("remove synthetic output fixture: %v", err)
|
||||
}
|
||||
|
||||
root := env.Config.Pipeline.Workspace.Root
|
||||
profilePath := filepath.Join(root, "local-profile.yml")
|
||||
if err := os.WriteFile(profilePath, []byte("id: local-test\nbackend: local\nmodel: local-model\ntimeout_seconds: 30\n"), 0o600); err != nil {
|
||||
t.Fatalf("write local profile: %v", err)
|
||||
}
|
||||
chunkRoot := filepath.Join(root, "notarius-cache", "chunk-plans")
|
||||
checkpointRoot := filepath.Join(root, "notarius-cache", "checkpoints")
|
||||
configPath := filepath.Join(root, "notarius-integration.yml")
|
||||
content := fmt.Sprintf(`version: 5
|
||||
promptkit:
|
||||
profile_file: %q
|
||||
local_backend:
|
||||
endpoint: %q
|
||||
concurrency_limit: 1
|
||||
concurrency:
|
||||
total_llm: 1
|
||||
stage_workers:
|
||||
extract: 1
|
||||
output:
|
||||
directory: %q
|
||||
cache:
|
||||
chunk_plans:
|
||||
mode: auto
|
||||
directory: %q
|
||||
checkpoints:
|
||||
enabled: true
|
||||
directory: %q
|
||||
pipelines:
|
||||
%s:
|
||||
llm_profile: local-test
|
||||
input:
|
||||
module: seriatim
|
||||
chunk:
|
||||
module: generic
|
||||
output:
|
||||
module: json
|
||||
steps:
|
||||
- id: collect-first
|
||||
artifacts:
|
||||
spells-first:
|
||||
extract:
|
||||
module: dnd/spells
|
||||
normalize:
|
||||
module: dnd/spells
|
||||
- id: collect-second
|
||||
artifacts:
|
||||
spells-second:
|
||||
extract:
|
||||
module: dnd/spells
|
||||
normalize:
|
||||
module: dnd/spells
|
||||
`, profilePath, endpoint+"/v1", filepath.Join(root, "unused-output"), chunkRoot, checkpointRoot, pipelineID)
|
||||
if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil {
|
||||
t.Fatalf("write Notarius integration config: %v", err)
|
||||
}
|
||||
|
||||
transcriptPath := filepath.Join(sessionPathsForEnv(env, runManifest.SessionID).ArtifactsDir, "trimmed.from-manifest.json")
|
||||
setNotariusIntegrationTranscript(t, runManifest, transcriptPath, notariusIntegrationTranscript("session-cache", "Aria casts Cure Wounds."))
|
||||
partyPath := recordPreparedExtractInput(t, env, runManifest, artifactpolicy.SourceInputParty, "The party includes Aria the cleric.\n")
|
||||
env.Notarius = notarius.NewSubprocessRunner()
|
||||
env.Config.Pipeline.Notarius = &config.NotariusConfig{
|
||||
Enabled: true, Binary: binary, ConfigPath: configPath, PipelineID: pipelineID,
|
||||
Timeout: "2m", WorkingDirectory: root, ChunkCache: "auto",
|
||||
References: map[string]string{"party": artifactpolicy.SourceInputParty},
|
||||
Outputs: map[string]config.NotariusOutputConfig{
|
||||
"spells_first": {
|
||||
LaneID: "spells-first", MediaType: "application/json", ModuleKey: "dnd/spells",
|
||||
SchemaID: "notarius.dnd.spells", SchemaVersion: "v1",
|
||||
},
|
||||
"spells_second": {
|
||||
LaneID: "spells-second", MediaType: "application/json", ModuleKey: "dnd/spells",
|
||||
SchemaID: "notarius.dnd.spells", SchemaVersion: "v1",
|
||||
},
|
||||
},
|
||||
}
|
||||
return env, runManifest, notariusIntegrationFixture{transcriptPath: transcriptPath, partyPath: partyPath, chunkRoot: chunkRoot}
|
||||
}
|
||||
|
||||
func notariusIntegrationTranscript(id, text string) []byte {
|
||||
return []byte(fmt.Sprintf(`{"metadata":{"id":%q,"title":"Cache integration"},"segments":[{"id":1,"start":0,"end":4,"speaker":"Aria","text":%q}]}`, id, text))
|
||||
}
|
||||
|
||||
func setNotariusIntegrationTranscript(t *testing.T, runManifest *manifest.Manifest, path string, content []byte) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(path, content, 0o644); err != nil {
|
||||
t.Fatalf("write integration transcript: %v", err)
|
||||
}
|
||||
checksum, err := artifacts.SHA256File(path)
|
||||
if err != nil {
|
||||
t.Fatalf("checksum integration transcript: %v", err)
|
||||
}
|
||||
record := runManifest.Stages["trim"]
|
||||
if record == nil || len(record.Outputs) != 1 {
|
||||
t.Fatalf("trim record = %#v, want one output", record)
|
||||
}
|
||||
record.Outputs[0].Kind = artifactmodel.TranscriptOutputKindFinalTrimmed
|
||||
record.Outputs[0].SourceID = artifactmodel.SourceTranscriptFinalTrimmed
|
||||
record.Outputs[0].LocalPath = path
|
||||
record.Outputs[0].Checksum = checksum
|
||||
}
|
||||
|
||||
func setNotariusIntegrationReference(t *testing.T, runManifest *manifest.Manifest, path, content string) {
|
||||
t.Helper()
|
||||
descriptor, ok := artifactpolicy.DescribePreparedInputSource(artifactpolicy.SourceInputParty)
|
||||
if !ok {
|
||||
t.Fatal("prepared party input descriptor is unavailable")
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
|
||||
t.Fatalf("write integration reference: %v", err)
|
||||
}
|
||||
checksum, err := artifacts.SHA256File(path)
|
||||
if err != nil {
|
||||
t.Fatalf("checksum integration reference: %v", err)
|
||||
}
|
||||
for index := range runManifest.Inputs {
|
||||
if runManifest.Inputs[index].Kind == descriptor.ManifestKind {
|
||||
runManifest.Inputs[index].Checksum = checksum
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("prepared party input not found in %#v", runManifest.Inputs)
|
||||
}
|
||||
|
||||
func notariusIntegrationSnapshotPath(t *testing.T, env *Env, runManifest *manifest.Manifest, sourceID string) string {
|
||||
t.Helper()
|
||||
descriptor, ok := artifactpolicy.DescribePreparedInputSource(sourceID)
|
||||
if !ok {
|
||||
t.Fatalf("prepared input descriptor %q is unavailable", sourceID)
|
||||
}
|
||||
for _, input := range runManifest.Inputs {
|
||||
if input.Kind != descriptor.ManifestKind {
|
||||
continue
|
||||
}
|
||||
path, err := artifacts.SessionNotariusReferenceSnapshotPath(
|
||||
sessionPathsForEnv(env, runManifest.SessionID), input.Checksum, filepath.Base(input.Path),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("resolve integration snapshot: %v", err)
|
||||
}
|
||||
return path
|
||||
}
|
||||
t.Fatalf("prepared input %q not found", sourceID)
|
||||
return ""
|
||||
}
|
||||
|
||||
func runNotariusIntegrationExtraction(t *testing.T, env *Env, runManifest *manifest.Manifest, runID string, force bool) notariusIntegrationRunManifest {
|
||||
t.Helper()
|
||||
runManifest.RunID = runID
|
||||
env.Force = force
|
||||
result, err := (extractStage{}).Run(context.Background(), env, runManifest)
|
||||
if err != nil {
|
||||
receiptPath := artifacts.SessionRunNotariusReceiptPathForCampaign(env.Config.Pipeline.Workspace.Root, runManifest.Campaign, runManifest.SessionID, runID)
|
||||
if data, readErr := os.ReadFile(receiptPath); readErr == nil {
|
||||
t.Logf("failed Notarius receipt: %s", data)
|
||||
}
|
||||
t.Fatalf("extract run %q: %v", runID, err)
|
||||
}
|
||||
bundleRoot, ok := result.Metadata["bundle_root"].(string)
|
||||
if !ok || bundleRoot == "" {
|
||||
t.Fatalf("extract metadata bundle_root = %#v", result.Metadata["bundle_root"])
|
||||
}
|
||||
data, err := os.ReadFile(filepath.Join(bundleRoot, "manifest.json"))
|
||||
if err != nil {
|
||||
t.Fatalf("read Notarius manifest: %v", err)
|
||||
}
|
||||
var published notariusIntegrationRunManifest
|
||||
if err := json.Unmarshal(data, &published); err != nil {
|
||||
t.Fatalf("decode Notarius manifest: %v", err)
|
||||
}
|
||||
return published
|
||||
}
|
||||
|
||||
type notariusIntegrationRunManifest struct {
|
||||
ChunkPlan *struct {
|
||||
Action string `json:"action"`
|
||||
SourceDigest string `json:"source_digest"`
|
||||
} `json:"chunk_plan"`
|
||||
CheckpointDecisions []struct {
|
||||
Stage string `json:"stage"`
|
||||
LaneID string `json:"lane_id"`
|
||||
Category string `json:"category"`
|
||||
ReasonCode string `json:"reason_code"`
|
||||
} `json:"checkpoint_decisions"`
|
||||
}
|
||||
|
||||
func assertNotariusIntegrationDecision(t *testing.T, published notariusIntegrationRunManifest, stageName, laneID, category, reasonCode string) {
|
||||
t.Helper()
|
||||
for _, decision := range published.CheckpointDecisions {
|
||||
if decision.Stage == stageName && decision.LaneID == laneID {
|
||||
if decision.Category != category || reasonCode != "" && decision.ReasonCode != reasonCode {
|
||||
t.Fatalf("checkpoint decision for %s/%s = %#v, want category %q reason %q", stageName, laneID, decision, category, reasonCode)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("checkpoint decision for %s/%s not found in %#v", stageName, laneID, published.CheckpointDecisions)
|
||||
}
|
||||
Reference in New Issue
Block a user