From 8ff74e1251de789e8091bdbd65e14eca5ebb2bf2 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 16 Sep 2026 18:07:37 +0000 Subject: [PATCH] Prove Notarius cache reuse end to end --- docs/roadmap/implementation.md | 14 +- internal/adapters/notarius/subprocess.go | 2 +- internal/adapters/notarius/subprocess_test.go | 20 + .../stage/notarius_cache_integration_test.go | 379 ++++++++++++++++++ 4 files changed, 413 insertions(+), 2 deletions(-) create mode 100644 internal/stage/notarius_cache_integration_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 4d2fc94..e943f45 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -213,7 +213,19 @@ changes. This stage proves Narratio orchestration, not upstream cache hits. ## Stage 5 — Prove Reuse with Real Notarius and a Local Test Provider -Status: Not started +Status: Complete (2026-09-16) + +Completion: Added an opt-in production-boundary integration test using a local +OpenAI-compatible provider and an absolute `NARRATIO_TEST_NOTARIUS_BINARY`. +It proves checkpoint reuse across distinct Narratio runs, cold execution after +reference changes, partial-failure recovery, forced checkpoint bypass, chunk +plan refresh, and preservation of an unrelated cached plan. It also aligned +receipt validation with Notarius's valid zero-attempt summaries for reused +producers. Tested against locally built `notarius v0.7.0` with +`NARRATIO_TEST_NOTARIUS_BINARY=/tmp/narratio-notarius-v070.g6gID8/notarius +GOWORK=off go test ./internal/stage -run +'^TestNotariusCacheReuseIntegration$' -count=1 -v`; the focused race test, +default `GOWORK=off go test -count=1 ./...`, and adapter/stage race tests pass. Read Notarius v0.7.0 `docs/cli.md`, `docs/config.md`, `docs/operations.md`, `examples/dnd-minimal.config.yml`, `examples/seriatim-minimal-transcript.json`, diff --git a/internal/adapters/notarius/subprocess.go b/internal/adapters/notarius/subprocess.go index ff0e69d..f7c0892 100644 --- a/internal/adapters/notarius/subprocess.go +++ b/internal/adapters/notarius/subprocess.go @@ -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{ diff --git a/internal/adapters/notarius/subprocess_test.go b/internal/adapters/notarius/subprocess_test.go index f1ff8fb..d23349f 100644 --- a/internal/adapters/notarius/subprocess_test.go +++ b/internal/adapters/notarius/subprocess_test.go @@ -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") diff --git a/internal/stage/notarius_cache_integration_test.go b/internal/stage/notarius_cache_integration_test.go new file mode 100644 index 0000000..54639bd --- /dev/null +++ b/internal/stage/notarius_cache_integration_test.go @@ -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) +}