From 51a36efb6b6515dda517e6b34ac5bcf9b32de83c Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 18 Jul 2026 00:20:56 +0000 Subject: [PATCH] Integrate chunk plan caching into the runner --- docs/roadmap/implementation.md | 2 +- internal/cli/run_test.go | 3 +- internal/core/workspace/manifest.go | 6 - internal/core/workspace/manifest_test.go | 12 - internal/framework/checkpoint/loader.go | 64 --- internal/framework/checkpoint/recorder.go | 101 ----- .../framework/checkpoint/recorder_test.go | 149 ++----- internal/framework/chunkplan/store.go | 2 +- internal/framework/pipeline/checkpoint.go | 21 - .../framework/pipeline/chunk_plan_store.go | 2 + internal/framework/pipeline/runner.go | 145 +++---- .../runner_candidate_encoding_test.go | 2 +- .../framework/pipeline/runner_chunk_plan.go | 137 +++++++ .../pipeline/runner_chunk_plan_test.go | 368 ++++++++++++++++++ .../pipeline/synchronized_collaborators.go | 17 - 15 files changed, 590 insertions(+), 441 deletions(-) create mode 100644 internal/framework/pipeline/runner_chunk_plan.go create mode 100644 internal/framework/pipeline/runner_chunk_plan_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 2a0fe14..91b6ce0 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -531,7 +531,7 @@ runs still use Stage 2’s generate-and-materialize behavior. ## Stage 4: Integrate cache policy and remove chunk checkpoints -**Status:** Not started +**Status:** Complete ### Objective diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index fd8c173..aba0f40 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -2333,8 +2333,6 @@ func TestRunPipelineWritesCheckpointsWhenWorkspaceResumeEnabled(t *testing.T) { for _, name := range []string{ "source/manifest.json", "source/source-document.json", - "chunk/manifest.json", - "chunk/chunks.json", "extract/spells/manifest.json", "extract/spells/outputs.json", "merge/spells/manifest.json", @@ -2346,6 +2344,7 @@ func TestRunPipelineWritesCheckpointsWhenWorkspaceResumeEnabled(t *testing.T) { t.Fatalf("expected checkpoint artifact %q: %v", name, err) } } + assertPathNotExist(t, filepath.Join(checkpointDir, "chunk")) assertPathNotExist(t, filepath.Join(workspaceDir, "debug")) } diff --git a/internal/core/workspace/manifest.go b/internal/core/workspace/manifest.go index d3cbb39..c1a9e1e 100644 --- a/internal/core/workspace/manifest.go +++ b/internal/core/workspace/manifest.go @@ -11,7 +11,6 @@ type StageName string const ( StageSource StageName = "source" - StageChunk StageName = "chunk" StageExtract StageName = "extract" StageMerge StageName = "merge" StageNormalize StageName = "normalize" @@ -55,11 +54,6 @@ type SourceManifest struct { SourceID string `json:"source_id,omitempty"` } -type ChunkManifest struct { - StageManifest - ChunkCount int `json:"chunk_count,omitempty"` -} - type ExtractLaneManifest struct { StageManifest ChunkCount int `json:"chunk_count,omitempty"` diff --git a/internal/core/workspace/manifest_test.go b/internal/core/workspace/manifest_test.go index ab80841..fd509d0 100644 --- a/internal/core/workspace/manifest_test.go +++ b/internal/core/workspace/manifest_test.go @@ -42,18 +42,6 @@ func TestManifestJSONRoundTrips(t *testing.T) { } }) - t.Run("chunk", func(t *testing.T) { - manifest := ChunkManifest{ - StageManifest: populatedManifest(StageChunk, "", "generic", started, completed), - ChunkCount: 3, - } - var got ChunkManifest - roundTripManifest(t, manifest, &got) - if got.ChunkCount != manifest.ChunkCount || got.Stage != StageChunk { - t.Fatalf("round trip chunk manifest = %+v", got) - } - }) - t.Run("extract", func(t *testing.T) { manifest := ExtractLaneManifest{ StageManifest: populatedManifest(StageExtract, "spells", "dnd/spells", started, completed), diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index a11bbe4..427131c 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -58,36 +58,6 @@ func (l *WorkspaceLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, p return pipeline.SourceCheckpoint{Document: &doc}, reusedDecision() } -func (l *WorkspaceLoader) Chunk(moduleKey string, sourceDigest string) (pipeline.ChunkCheckpoint, pipeline.CheckpointDecision) { - expectedDependencies := digestFingerprints("source_document", sourceDigest) - var manifest coreworkspace.ChunkManifest - if decision := l.readJSON("chunk/manifest.json", &manifest); !decision.Reused { - return pipeline.ChunkCheckpoint{}, decision - } - if decision := l.validateManifest(manifest.StageManifest, coreworkspace.StageChunk, "", moduleKey, coreworkspace.StatusSucceeded, expectedDependencies); !decision.Reused { - return pipeline.ChunkCheckpoint{}, decision - } - var payload chunksEnvelope - if decision := l.readJSON("chunk/chunks.json", &payload); !decision.Reused { - return pipeline.ChunkCheckpoint{}, decision - } - chunks, err := sourceChunksFromEnvelope(payload.Chunks) - if err != nil { - return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint payload is invalid: %v", err) - } - if len(chunks) == 0 { - return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint payload has no chunks") - } - outputDigests, err := chunkOutputDigests(chunks) - if err != nil { - return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint output cannot be digested: %v", err) - } - if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), outputDigests) { - return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint output digests do not match payload") - } - return pipeline.ChunkCheckpoint{Chunks: chunks, Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() -} - func (l *WorkspaceLoader) Extract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) { var manifest coreworkspace.ExtractLaneManifest if d := l.readJSON(laneManifestPath("extract", laneID), &manifest); !d.Reused { @@ -232,40 +202,6 @@ func (l *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManif return reusedDecision() } -func sourceChunksFromEnvelope(values []chunkEnvelope) ([]source.Chunk, error) { - if len(values) == 0 { - return nil, nil - } - out := make([]source.Chunk, 0, len(values)) - for _, value := range values { - content, err := contentFromEnvelope(value.Content) - if err != nil { - return nil, err - } - annotations, err := source.CanonicalizeChunkAnnotations(value.Annotations) - if err != nil { - return nil, fmt.Errorf("canonicalize chunk %q annotations: %w", value.ID, err) - } - planAnnotations, err := source.CanonicalizeChunkAnnotations(value.PlanAnnotations) - if err != nil { - return nil, fmt.Errorf("canonicalize chunk %q plan_annotations: %w", value.ID, err) - } - out = append(out, source.Chunk{ - ID: value.ID, - SourceID: value.SourceID, - Index: value.Index, - Ref: value.Ref, - Content: content, - MediaType: value.Content.MediaType, - Units: cloneSourceUnits(value.Units), - Metadata: cloneMetadata(value.Metadata), - Annotations: annotations, - PlanAnnotations: planAnnotations, - }) - } - return out, nil -} - func contentFromEnvelope(value binaryEnvelope) ([]byte, error) { content, err := base64.StdEncoding.DecodeString(value.ContentBase64) if err != nil { diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go index 1a12732..8ee7d8d 100644 --- a/internal/framework/checkpoint/recorder.go +++ b/internal/framework/checkpoint/recorder.go @@ -65,54 +65,6 @@ func (r *WorkspaceRecorder) SourceFailed(moduleKey string, err error) error { return r.writeManifest("source/manifest.json", coreworkspace.SourceManifest{StageManifest: manifest}) } -func (r *WorkspaceRecorder) ChunkRunning(moduleKey string, sourceDigest string) error { - manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusRunning) - manifest.ModuleKey = moduleKey - manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) - manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest}) -} - -func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string, chunks []source.Chunk, warnings []contracts.Warning) error { - outputDigests, err := chunkOutputDigests(chunks) - if err != nil { - return fmt.Errorf("digest chunk checkpoint output: %w", err) - } - payload := chunksEnvelope{Chunks: chunkEnvelopes(chunks), Warnings: cloneWarnings(warnings)} - if err := r.writePayload("chunk/chunks.json", payload); err != nil { - return err - } - manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceeded) - manifest.ModuleKey = moduleKey - manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) - manifest.OutputDigests = workspaceFingerprints(outputDigests) - manifest.ValidationStatus = validationStatusString(warnings, nil) - manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{ - StageManifest: manifest, - ChunkCount: len(chunks), - }) -} - -func (r *WorkspaceRecorder) ChunkRejected(moduleKey string, sourceDigest string, rejected contracts.RejectedOutput) error { - manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceededWithRejections) - manifest.ModuleKey = moduleKey - manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) - manifest.ValidationStatus = "rejected" - manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) - manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest}) -} - -func (r *WorkspaceRecorder) ChunkFailed(moduleKey string, sourceDigest string, err error) error { - manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusFailed) - manifest.ModuleKey = moduleKey - manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest)) - manifest.CompletedAt = timePtr(r.timestamp()) - manifest.Metadata = errorMetadata(err) - return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest}) -} - func (r *WorkspaceRecorder) ExtractRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { manifest := r.laneManifest(coreworkspace.StageExtract, coreworkspace.StatusRunning, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) @@ -245,23 +197,6 @@ type sourceDocumentEnvelope struct { Document source.SourceDocument `json:"document"` } -type chunksEnvelope struct { - Chunks []chunkEnvelope `json:"chunks"` - Warnings []contracts.Warning `json:"warnings,omitempty"` -} - -type chunkEnvelope struct { - ID string `json:"id"` - SourceID string `json:"source_id"` - Index int `json:"index"` - Ref source.SourceRef `json:"ref"` - Content binaryEnvelope `json:"content"` - Units []source.SourceUnit `json:"units,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` - Annotations source.ChunkAnnotations `json:"annotations,omitempty"` - PlanAnnotations source.ChunkAnnotations `json:"plan_annotations,omitempty"` -} - type binaryEnvelope struct { ContentBase64 string `json:"content_base64,omitempty"` ContentDigest string `json:"content_digest,omitempty"` @@ -315,27 +250,6 @@ func artifactOutputDigests(outputs []pipeline.CheckpointArtifact) []pipeline.Che return normalizeFingerprints(values) } -func chunkEnvelopes(chunks []source.Chunk) []chunkEnvelope { - if len(chunks) == 0 { - return nil - } - out := make([]chunkEnvelope, 0, len(chunks)) - for _, chunk := range chunks { - out = append(out, chunkEnvelope{ - ID: chunk.ID, - SourceID: chunk.SourceID, - Index: chunk.Index, - Ref: chunk.Ref, - Content: binaryEnvelopeFromContent(chunk.Content, chunk.MediaType, chunk.Metadata, nil), - Units: cloneSourceUnits(chunk.Units), - Metadata: cloneMetadata(chunk.Metadata), - Annotations: source.CloneChunkAnnotations(chunk.Annotations), - PlanAnnotations: source.CloneChunkAnnotations(chunk.PlanAnnotations), - }) - } - return out -} - func binaryEnvelopeFromContent(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) binaryEnvelope { return binaryEnvelope{ ContentBase64: base64.StdEncoding.EncodeToString(content), @@ -394,21 +308,6 @@ func cloneMetadata(metadata map[string]any) map[string]any { return out } -func chunkOutputDigests(chunks []source.Chunk) ([]pipeline.CheckpointFingerprint, error) { - values := make([]pipeline.CheckpointFingerprint, 0, len(chunks)) - for _, chunk := range chunks { - digest, err := source.DigestChunk(chunk) - if err != nil { - return nil, fmt.Errorf("chunk %q: %w", chunk.ID, err) - } - values = append(values, pipeline.CheckpointFingerprint{ - Name: chunk.ID, - Value: digest, - }) - } - return normalizeFingerprints(values), nil -} - func digestFingerprints(name string, digest string) []pipeline.CheckpointFingerprint { digest = strings.TrimSpace(digest) if digest == "" { diff --git a/internal/framework/checkpoint/recorder_test.go b/internal/framework/checkpoint/recorder_test.go index 3003aad..304b7e1 100644 --- a/internal/framework/checkpoint/recorder_test.go +++ b/internal/framework/checkpoint/recorder_test.go @@ -1,11 +1,9 @@ package checkpoint import ( - "encoding/base64" "encoding/json" "os" "path/filepath" - "reflect" "strings" "testing" @@ -25,24 +23,6 @@ func TestWorkspaceRecorderWritesSuccessfulCheckpointFiles(t *testing.T) { Digest: "sha256:source", Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}}, } - chunks := []source.Chunk{ - { - ID: "chunk-1", - SourceID: "source-1", - Index: 0, - Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}, - Content: []byte("chunk content"), - MediaType: "text/plain", - Units: doc.Units, - Annotations: source.ChunkAnnotations{ - "same": json.RawMessage(`{"range":1}`), - }, - PlanAnnotations: source.ChunkAnnotations{ - "same": json.RawMessage(`{"plan":2}`), - }, - }, - } - if err := recorder.SourceRunning("seriatim"); err != nil { t.Fatalf("SourceRunning: %v", err) } @@ -54,60 +34,6 @@ func TestWorkspaceRecorderWritesSuccessfulCheckpointFiles(t *testing.T) { if _, err := os.Stat(filepath.Join(root, "source", "source-document.json")); err != nil { t.Fatalf("expected source checkpoint payload: %v", err) } - - if err := recorder.ChunkRunning("generic", doc.Digest); err != nil { - t.Fatalf("ChunkRunning: %v", err) - } - if err := recorder.ChunkSucceeded("generic", doc.Digest, chunks, nil); err != nil { - t.Fatalf("ChunkSucceeded: %v", err) - } - assertManifestStatus(t, filepath.Join(root, "chunk", "manifest.json"), coreworkspace.StatusSucceeded) - var chunkPayload struct { - Chunks []struct { - Annotations source.ChunkAnnotations `json:"annotations"` - PlanAnnotations source.ChunkAnnotations `json:"plan_annotations"` - Content struct { - ContentBase64 string `json:"content_base64"` - ContentDigest string `json:"content_digest"` - } `json:"content"` - } `json:"chunks"` - } - readJSON(t, filepath.Join(root, "chunk", "chunks.json"), &chunkPayload) - if len(chunkPayload.Chunks) != 1 { - t.Fatalf("checkpoint chunks = %#v, want one", chunkPayload.Chunks) - } - decoded, err := base64.StdEncoding.DecodeString(chunkPayload.Chunks[0].Content.ContentBase64) - if err != nil { - t.Fatalf("decode chunk content: %v", err) - } - if string(decoded) != "chunk content" { - t.Fatalf("chunk content = %q, want original content", decoded) - } - if got, want := chunkPayload.Chunks[0].Content.ContentDigest, contentDigest([]byte("chunk content")); got != want { - t.Fatalf("content digest = %q, want %q", got, want) - } - if !jsonEqual(t, chunkPayload.Chunks[0].Annotations["same"], json.RawMessage(`{"range":1}`)) || !jsonEqual(t, chunkPayload.Chunks[0].PlanAnnotations["same"], json.RawMessage(`{"plan":2}`)) { - t.Fatalf("checkpoint annotations = %s / %s", chunkPayload.Chunks[0].Annotations["same"], chunkPayload.Chunks[0].PlanAnnotations["same"]) - } - loaded, decision := (&WorkspaceLoader{root: root}).Chunk("generic", doc.Digest) - if !decision.Reused || len(loaded.Chunks) != 1 { - t.Fatalf("loaded chunk checkpoint = %#v, decision = %#v", loaded, decision) - } - if string(loaded.Chunks[0].Annotations["same"]) != `{"range":1}` || string(loaded.Chunks[0].PlanAnnotations["same"]) != `{"plan":2}` { - t.Fatalf("loaded canonical annotations = %s / %s", loaded.Chunks[0].Annotations["same"], loaded.Chunks[0].PlanAnnotations["same"]) - } -} - -func jsonEqual(t *testing.T, left, right []byte) bool { - t.Helper() - var leftValue, rightValue any - if err := json.Unmarshal(left, &leftValue); err != nil { - t.Fatalf("decode left JSON: %v", err) - } - if err := json.Unmarshal(right, &rightValue); err != nil { - t.Fatalf("decode right JSON: %v", err) - } - return reflect.DeepEqual(leftValue, rightValue) } func TestWorkspaceArtifactCheckpointsRoundTripCodecIdentityAndBytes(t *testing.T) { @@ -158,24 +84,6 @@ func TestWorkspaceLoaderInvalidatesMissingCorruptAndMismatchedCheckpoints(t *tes } }) - t.Run("dependency mismatch", func(t *testing.T) { - root := t.TempDir() - recorder := newTestRecorder(t, root) - chunks := []source.Chunk{{ - ID: "chunk-1", - SourceID: "source-1", - Content: []byte("chunk content"), - MediaType: "text/plain", - }} - if err := recorder.ChunkSucceeded("generic", "sha256:source-a", chunks, nil); err != nil { - t.Fatalf("ChunkSucceeded: %v", err) - } - loader := &WorkspaceLoader{root: root} - if _, decision := loader.Chunk("generic", "sha256:source-b"); decision.Reused || !strings.Contains(decision.Reason, "dependency") { - t.Fatalf("decision = %#v, want dependency invalidation", decision) - } - }) - t.Run("incompatible workspace schema remains untouched", func(t *testing.T) { root := t.TempDir() recorder := newTestRecorder(t, root) @@ -207,28 +115,41 @@ func TestWorkspaceLoaderInvalidatesMissingCorruptAndMismatchedCheckpoints(t *tes } }) - t.Run("corrupt payload", func(t *testing.T) { - root := t.TempDir() - recorder := newTestRecorder(t, root) - chunks := []source.Chunk{{ - ID: "chunk-1", - SourceID: "source-1", - Content: []byte("chunk content"), - MediaType: "text/plain", - }} - if err := recorder.ChunkSucceeded("generic", "sha256:source", chunks, nil); err != nil { - t.Fatalf("ChunkSucceeded: %v", err) - } - payloadPath := filepath.Join(root, "chunk", "chunks.json") - data := strings.ReplaceAll(string(readFile(t, payloadPath)), contentDigest([]byte("chunk content")), "sha256:bad") - if err := os.WriteFile(payloadPath, []byte(data), 0o644); err != nil { - t.Fatalf("corrupt chunk payload: %v", err) - } - loader := &WorkspaceLoader{root: root} - if _, decision := loader.Chunk("generic", "sha256:source"); decision.Reused || !strings.Contains(decision.Reason, "invalid") { - t.Fatalf("decision = %#v, want corrupt payload invalidation", decision) - } - }) +} + +func TestWorkspaceCheckpointsIgnoreLegacyChunkFiles(t *testing.T) { + root := t.TempDir() + legacyManifest := []byte(`{"legacy":"manifest"}`) + legacyPayload := []byte(`{"legacy":"chunks"}`) + legacyDir := filepath.Join(root, "chunk") + if err := os.MkdirAll(legacyDir, 0o700); err != nil { + t.Fatal(err) + } + manifestPath := filepath.Join(legacyDir, "manifest.json") + payloadPath := filepath.Join(legacyDir, "chunks.json") + if err := os.WriteFile(manifestPath, legacyManifest, 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(payloadPath, legacyPayload, 0o600); err != nil { + t.Fatal(err) + } + + doc := &source.SourceDocument{ID: "source-1", Kind: "document", Format: "text/plain", Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}}} + doc.Digest, _ = source.DigestDocument(doc) + recorder := newTestRecorder(t, root) + if err := recorder.SourceSucceeded("seriatim", doc); err != nil { + t.Fatal(err) + } + loaded, decision := (&WorkspaceLoader{root: root}).Source("seriatim") + if !decision.Reused || loaded.Document == nil || loaded.Document.Digest != doc.Digest { + t.Fatalf("source checkpoint = %#v decision = %#v", loaded, decision) + } + if got := readFile(t, manifestPath); string(got) != string(legacyManifest) { + t.Fatalf("legacy manifest changed: %s", got) + } + if got := readFile(t, payloadPath); string(got) != string(legacyPayload) { + t.Fatalf("legacy payload changed: %s", got) + } } func TestWorkspaceRecorderRecordsRejectedExtractOutputs(t *testing.T) { diff --git a/internal/framework/chunkplan/store.go b/internal/framework/chunkplan/store.go index f076ca8..304904f 100644 --- a/internal/framework/chunkplan/store.go +++ b/internal/framework/chunkplan/store.go @@ -14,7 +14,7 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -const SchemaVersion = "notarius.chunk-plan.v1" +const SchemaVersion = pipeline.ChunkPlanSchemaVersion type filesystemStore struct { root string diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 6f21df3..429e817 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -20,10 +20,6 @@ type CheckpointRecorder interface { SourceRunning(moduleKey string) error SourceSucceeded(moduleKey string, doc *source.SourceDocument) error SourceFailed(moduleKey string, err error) error - ChunkRunning(moduleKey string, sourceDigest string) error - ChunkSucceeded(moduleKey string, sourceDigest string, chunks []source.Chunk, warnings []contracts.Warning) error - ChunkRejected(moduleKey string, sourceDigest string, rejected contracts.RejectedOutput) error - ChunkFailed(moduleKey string, sourceDigest string, err error) error ExtractRunning(laneID string, moduleKey string, dependencies []CheckpointFingerprint) error ExtractSucceeded(laneID string, moduleKey string, dependencies []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error ExtractFailed(laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error @@ -54,11 +50,6 @@ type SourceCheckpoint struct { Document *source.SourceDocument } -type ChunkCheckpoint struct { - Chunks []source.Chunk - Warnings []contracts.Warning -} - // CheckpointArtifact is the durable, domain-neutral value stored at a lane // checkpoint boundary. type CheckpointArtifact struct { @@ -90,7 +81,6 @@ type NormalizeCheckpoint struct { type CheckpointLoader interface { Enabled() bool Source(moduleKey string) (SourceCheckpoint, CheckpointDecision) - Chunk(moduleKey string, sourceDigest string) (ChunkCheckpoint, CheckpointDecision) Extract(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) Merge(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) Normalize(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) @@ -105,14 +95,6 @@ func NoopCheckpointLoader() CheckpointLoader { return noopCheckpointLoader{} func (noopCheckpointRecorder) SourceRunning(string) error { return nil } func (noopCheckpointRecorder) SourceSucceeded(string, *source.SourceDocument) error { return nil } func (noopCheckpointRecorder) SourceFailed(string, error) error { return nil } -func (noopCheckpointRecorder) ChunkRunning(string, string) error { return nil } -func (noopCheckpointRecorder) ChunkSucceeded(string, string, []source.Chunk, []contracts.Warning) error { - return nil -} -func (noopCheckpointRecorder) ChunkRejected(string, string, contracts.RejectedOutput) error { - return nil -} -func (noopCheckpointRecorder) ChunkFailed(string, string, error) error { return nil } func (noopCheckpointRecorder) ExtractRunning(string, string, []CheckpointFingerprint) error { return nil } @@ -149,9 +131,6 @@ func (noopCheckpointLoader) Enabled() bool { return false } func (noopCheckpointLoader) Source(string) (SourceCheckpoint, CheckpointDecision) { return SourceCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} } -func (noopCheckpointLoader) Chunk(string, string) (ChunkCheckpoint, CheckpointDecision) { - return ChunkCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} -} func (noopCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { return ExtractCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} } diff --git a/internal/framework/pipeline/chunk_plan_store.go b/internal/framework/pipeline/chunk_plan_store.go index e33efb1..cdccbc8 100644 --- a/internal/framework/pipeline/chunk_plan_store.go +++ b/internal/framework/pipeline/chunk_plan_store.go @@ -8,6 +8,8 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) +const ChunkPlanSchemaVersion = "notarius.chunk-plan.v1" + type ChunkPlanProducer struct { InputModule string `json:"input_module"` ChunkModule string `json:"chunk_module"` diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index b2b1ccd..19f68eb 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -38,19 +38,21 @@ func New() *Runner { } type RunInput struct { - Prepared *PreparedPipeline - SourceID string - Path string - RawInput []byte - SessionID string - RunID string - StartedAt time.Time - LLMProfiles []artifacts.LLMProfileManifest - Metadata map[string]any - Warnings []contracts.Warning - Checkpoints CheckpointRecorder - Checkpoint CheckpointLoader - Debug DebugRecorder + Prepared *PreparedPipeline + SourceID string + Path string + RawInput []byte + SessionID string + RunID string + StartedAt time.Time + LLMProfiles []artifacts.LLMProfileManifest + Metadata map[string]any + Warnings []contracts.Warning + ChunkCacheMode ChunkCacheMode + ChunkPlans ChunkPlanStore + Checkpoints CheckpointRecorder + Checkpoint CheckpointLoader + Debug DebugRecorder // ExtractWorkers bounds run-wide extract jobs. Values less than one use a // single worker so direct framework callers retain deterministic behavior. ExtractWorkers int @@ -174,19 +176,14 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err chunker := input.Prepared.chunker attachModuleManifestMetadata(&output, "chunker", chunker) - var canonicalChunks []source.Chunk - var generatedPlan *source.ChunkPlan - var chunkWarnings []contracts.Warning - chunkCheckpoint, chunkDecision := checkpointLoader.Chunk(chunker.Key(), doc.Digest) - recordCheckpointEvent(&output, checkpointLoader, string(StageChunk), "", chunker.Key(), chunkDecision) chunkStarted := time.Now().UTC() + chunkMode := effectiveChunkCacheMode(input.ChunkCacheMode) if err := writeDebugTimed(debugRecorder, "chunk/input.json", debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), StartedAt: chunkStarted, Payload: map[string]any{ - "reused": chunkDecision.Reused, - "decision": chunkDecision, + "cache_mode": chunkMode, "source": debugSourceDocumentEnvelope(doc), "source_input": debugContentEnvelope(sourceInput.Content, sourceInput.MediaType, nil, nil), "options": redactSensitiveMap(input.pipeline.Chunk.Options), @@ -195,85 +192,28 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err }); err != nil { return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err) } - chunksAccepted := chunkDecision.Reused - var chunkRejection *contracts.RejectedOutput - if chunkDecision.Reused { - canonicalChunks = cloneSourceChunks(chunkCheckpoint.Chunks) - chunkWarnings = cloneWarnings(chunkCheckpoint.Warnings) - output.Warnings = append(output.Warnings, chunkWarnings...) - } else { - if err := checkpoints.ChunkRunning(chunker.Key(), doc.Digest); err != nil { - return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) - } - chunksAccepted, chunkRejection, err = runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - attemptStarted := time.Now().UTC() - attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) - attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) - terminal := newAttemptTerminalRecorder(debugRecorder, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted}) - chunkResult, err := chunker.Plan(attemptCtx, contracts.ChunkRequest{ - Source: doc, - SourceInput: sourceInput.Clone(), - SessionID: sessionID, - References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet), - LLMProfile: input.pipeline.Chunk.LLMProfile, - Metadata: input.Metadata, - }) - if err != nil { - attemptErr := fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err) - return false, nil, terminal.record(nil, attemptErr) - } - plan, chunks, err := validateAndMaterializeChunkPlan(doc, chunkResult.Plan) - if err != nil { - attemptErr := fmt.Errorf("validate chunk plan from chunker %q: %w", chunker.Key(), err) - payload := map[string]any{"plan": debugChunkPlanEnvelope(chunkResult.Plan), "warnings": debugWarningEnvelopes(chunkResult.Warnings)} - return false, nil, terminal.record(payload, attemptErr) - } - validationWarnings, rejection, err := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug) - attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...) - payload := map[string]any{ - "plan": debugChunkPlanEnvelope(plan), - "materialized_chunks": debugSourceChunkEnvelopes(chunks), - "warnings": debugWarningEnvelopes(attemptWarnings), - "rejection": debugRejectedOutputPtr(rejection), - } - if err != nil || rejection != nil { - return false, rejection, terminal.record(payload, err) - } - canonicalChunks = chunks - generatedPlan = &plan - chunkWarnings = attemptWarnings - if err := terminal.record(payload, nil); err != nil { - return false, nil, err - } - return true, nil, nil - }) - if err != nil { - _ = checkpoints.ChunkFailed(chunker.Key(), doc.Digest, err) - return failOutput(output), err - } - if !chunksAccepted { - output.Rejected = append(output.Rejected, *chunkRejection) - if err := checkpoints.ChunkRejected(chunker.Key(), doc.Digest, *chunkRejection); err != nil { - return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) - } - } else { - output.Warnings = append(output.Warnings, chunkWarnings...) - if err := checkpoints.ChunkSucceeded(chunker.Key(), doc.Digest, canonicalChunks, chunkWarnings); err != nil { - return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err) - } - } + chunkResult, err := r.runChunkPlan(ctx, input, doc, sourceInput, sessionID) + if err != nil { + return failOutput(output), err + } + if chunkResult.rejection != nil { + output.Rejected = append(output.Rejected, *chunkResult.rejection) + } + if chunkResult.accepted || chunkResult.lookup.Status == ChunkPlanHit { + output.Warnings = append(output.Warnings, chunkResult.warnings...) } chunkDebugPayload := map[string]any{ - "reused": chunkDecision.Reused, - "accepted": chunksAccepted, - "materialized_chunks": debugSourceChunkEnvelopes(canonicalChunks), - "warnings": chunkWarnings, + "cache_mode": chunkMode, + "lookup": chunkResult.lookup, + "accepted": chunkResult.accepted, + "materialized_chunks": debugSourceChunkEnvelopes(chunkResult.chunks), + "warnings": chunkResult.warnings, } - if generatedPlan != nil { - chunkDebugPayload["plan"] = debugChunkPlanEnvelope(*generatedPlan) + if chunkResult.plan != nil { + chunkDebugPayload["plan"] = debugChunkPlanEnvelope(*chunkResult.plan) } - if chunkRejection != nil { - chunkDebugPayload["rejection"] = debugRejectedOutputEnvelope(*chunkRejection) + if chunkResult.rejection != nil { + chunkDebugPayload["rejection"] = debugRejectedOutputEnvelope(*chunkResult.rejection) } if err := writeDebugTimed(debugRecorder, "chunk/output.json", debugTimedEnvelope{ Stage: string(StageChunk), @@ -284,8 +224,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err) } - if chunksAccepted { - laneOutput, laneErr := r.runLanes(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, canonicalChunks) + if chunkResult.accepted { + laneOutput, laneErr := r.runLanes(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks) mergeLaneOutput(&output, laneOutput) if laneErr != nil { return failOutput(output), laneErr @@ -473,6 +413,13 @@ func validateRunInput(input RunInput) error { if input.Prepared == nil { return fmt.Errorf("prepared pipeline must not be nil") } + mode := effectiveChunkCacheMode(input.ChunkCacheMode) + if err := mode.Validate(); err != nil { + return err + } + if mode != ChunkCacheBypass && input.ChunkPlans == nil { + return fmt.Errorf("chunk plan store is required for %q mode", mode) + } return validateResolvedPipeline(input.Prepared.resolved) } @@ -546,10 +493,6 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest { References: ReferenceProvenance(pipeline), LLMProfiles: cloneLLMProfiles(input.LLMProfiles), } - // The runner does not currently maintain a cache or idempotency key. Reference - // digests are recorded in manifest provenance and intentionally kept separate - // from source_digests. - for _, lane := range pipeline.ArtifactLanes { laneManifest := artifacts.ArtifactLaneManifest{ ID: lane.ID, diff --git a/internal/framework/pipeline/runner_candidate_encoding_test.go b/internal/framework/pipeline/runner_candidate_encoding_test.go index 3f77a63..58ad3f8 100644 --- a/internal/framework/pipeline/runner_candidate_encoding_test.go +++ b/internal/framework/pipeline/runner_candidate_encoding_test.go @@ -209,7 +209,7 @@ func TestRunnerFinalEncodesAcceptedCandidatesOnce(t *testing.T) { for i, event := range output.CheckpointEvents { gotStages[i] = event.Stage } - wantStages := []string{"source", string(StageChunk), string(StageExtract), string(StageMerge), string(StageNormalize)} + wantStages := []string{"source", string(StageExtract), string(StageMerge), string(StageNormalize)} if !reflect.DeepEqual(gotStages, wantStages) { t.Fatalf("checkpoint event stages = %#v, want %#v", gotStages, wantStages) } diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go new file mode 100644 index 0000000..22b560a --- /dev/null +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -0,0 +1,137 @@ +package pipeline + +import ( + "context" + "fmt" + "path" + "time" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type chunkPlanExecution struct { + accepted bool + chunks []source.Chunk + plan *source.ChunkPlan + warnings []contracts.Warning + rejection *contracts.RejectedOutput + lookup ChunkPlanDecision +} + +func effectiveChunkCacheMode(mode ChunkCacheMode) ChunkCacheMode { + if mode == "" { + return ChunkCacheBypass + } + parsed, _ := ParseChunkCacheMode(string(mode)) + return parsed +} + +// Chunk-plan lookup and publication happen serially before lane workers start. +// The runner therefore does not add synchronization around the store. +func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string) (chunkPlanExecution, error) { + mode := effectiveChunkCacheMode(input.ChunkCacheMode) + chunker := input.Prepared.chunker + result := chunkPlanExecution{lookup: ChunkPlanDecision{Reason: "chunk plan lookup skipped"}} + + if mode == ChunkCacheAuto { + record, decision, err := input.ChunkPlans.Load(doc.Digest) + result.lookup = decision + if err != nil { + return result, fmt.Errorf("load chunk plan: %w", err) + } + switch decision.Status { + case ChunkPlanHit: + plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, record.Plan) + if validationErr == nil { + validationWarnings, rejection, err := r.validateChunks(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, 1, input.Debug) + result.plan = &plan + result.chunks = chunks + result.warnings = append(cloneWarnings(record.Warnings), validationWarnings...) + result.rejection = rejection + result.accepted = rejection == nil && err == nil + return result, err + } + result.lookup = ChunkPlanDecision{Status: ChunkPlanInvalid, Reason: "stored chunk plan cannot be materialized against the current source"} + case ChunkPlanMissing, ChunkPlanInvalid: + // Generate below. + default: + return result, fmt.Errorf("load chunk plan returned unsupported status %q", decision.Status) + } + } + + var producerWarnings []contracts.Warning + accepted, rejection, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + attemptStarted := time.Now().UTC() + attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) + attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted}) + chunkResult, callErr := chunker.Plan(attemptCtx, contracts.ChunkRequest{ + Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID, + References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet), + LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: input.Metadata, + }) + if callErr != nil { + return false, nil, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr)) + } + plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, chunkResult.Plan) + if validationErr != nil { + attemptErr := fmt.Errorf("validate chunk plan from chunker %q: %w", chunker.Key(), validationErr) + payload := map[string]any{"plan": debugChunkPlanEnvelope(chunkResult.Plan), "warnings": debugWarningEnvelopes(chunkResult.Warnings)} + return false, nil, terminal.record(payload, attemptErr) + } + validationWarnings, rejected, validationErr := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug) + attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...) + payload := map[string]any{ + "plan": debugChunkPlanEnvelope(plan), "materialized_chunks": debugSourceChunkEnvelopes(chunks), + "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected), + } + if validationErr != nil || rejected != nil { + return false, rejected, terminal.record(payload, validationErr) + } + result.chunks = chunks + result.plan = &plan + result.warnings = attemptWarnings + producerWarnings = cloneWarnings(chunkResult.Warnings) + if debugErr := terminal.record(payload, nil); debugErr != nil { + return false, nil, debugErr + } + return true, nil, nil + }) + if err != nil { + return result, err + } + result.accepted = accepted + result.rejection = rejection + if !accepted { + return result, nil + } + + if mode == ChunkCacheAuto || mode == ChunkCacheRefresh { + planDigest, err := source.DigestChunkPlan(*result.plan) + if err != nil { + return result, fmt.Errorf("digest generated chunk plan: %w", err) + } + producerMetadata, _ := moduleManifestMetadata(chunker) + record := ChunkPlanRecord{ + SchemaVersion: ChunkPlanSchemaVersion, + SourceDigest: doc.Digest, + PlanDigest: planDigest, + Plan: source.CloneChunkPlan(*result.plan), + Producer: ChunkPlanProducer{ + InputModule: input.Prepared.input.Key(), + ChunkModule: chunker.Key(), + LLMProfile: input.pipeline.Chunk.LLMProfile, + References: append([]artifacts.ReferenceProvenance(nil), referenceTargetProvenance(input.pipeline.ChunkReferences)...), + Metadata: cloneMetadata(producerMetadata), + }, + Warnings: cloneWarnings(producerWarnings), + CreatedAt: time.Now().UTC(), + } + if err := input.ChunkPlans.Save(record); err != nil { + return result, fmt.Errorf("save chunk plan: %w", err) + } + } + return result, nil +} diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go new file mode 100644 index 0000000..1994d67 --- /dev/null +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -0,0 +1,368 @@ +package pipeline + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type recordingChunkPlanStore struct { + record ChunkPlanRecord + decision ChunkPlanDecision + loadErr error + saveErr error + loads int + saves int + saved ChunkPlanRecord +} + +func (s *recordingChunkPlanStore) Load(string) (ChunkPlanRecord, ChunkPlanDecision, error) { + s.loads++ + return s.record, s.decision, s.loadErr +} + +func (s *recordingChunkPlanStore) Save(record ChunkPlanRecord) error { + s.saves++ + s.saved = record + return s.saveErr +} + +type countingChunkValidator struct { + calls int + result contracts.ValidationResult + err error +} + +func (*countingChunkValidator) Name() string { return "test/counting-chunks" } +func (*countingChunkValidator) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassDeterministic +} +func (v *countingChunkValidator) Validate(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + v.calls++ + return v.result, v.err +} + +type manifestChunker struct { + terminalChunker + metadata map[string]any +} + +func (c manifestChunker) ManifestMetadata() map[string]any { return c.metadata } + +type llmCountingChunker struct { + terminalChunker + llmCalls *int +} + +func (c llmCountingChunker) Plan(ctx context.Context, request contracts.ChunkRequest) (contracts.ChunkPlanResult, error) { + (*c.llmCalls)++ + return c.terminalChunker.Plan(ctx, request) +} + +type retryingChunker struct { + key string + plan source.ChunkPlan + calls int +} + +func (c *retryingChunker) Key() string { return c.key } +func (*retryingChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil } +func (c *retryingChunker) Plan(context.Context, contracts.ChunkRequest) (contracts.ChunkPlanResult, error) { + c.calls++ + if c.calls == 1 { + return contracts.ChunkPlanResult{Warnings: []contracts.Warning{{Scope: "discarded", ReasonCode: "retry", Message: "discarded warning"}}}, errors.New("retry generation") + } + return contracts.ChunkPlanResult{Plan: source.CloneChunkPlan(c.plan), Warnings: []contracts.Warning{{Scope: "accepted", ReasonCode: "observed", Message: "accepted warning"}}}, nil +} + +type dependencyLoader struct { + CheckpointLoader + extractDependencies []CheckpointFingerprint +} + +func (l *dependencyLoader) Extract(_ string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + l.extractDependencies = append([]CheckpointFingerprint(nil), dependencies...) + return ExtractCheckpoint{}, CheckpointDecision{Reason: "not found"} +} + +func TestRunnerChunkPlanModeMatrix(t *testing.T) { + tests := []struct { + name string + mode ChunkCacheMode + decision ChunkPlanDecision + wantLoads int + wantSaves int + wantCalls int + }{ + {name: "auto hit", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanHit}, wantLoads: 1}, + {name: "auto missing", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanMissing}, wantLoads: 1, wantSaves: 1, wantCalls: 1}, + {name: "auto invalid", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanInvalid}, wantLoads: 1, wantSaves: 1, wantCalls: 1}, + {name: "bypass", mode: ChunkCacheBypass, wantCalls: 1}, + {name: "empty bypass", wantCalls: 1}, + {name: "refresh", mode: ChunkCacheRefresh, wantSaves: 1, wantCalls: 1}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + calls := 0 + llmCalls := 0 + prepared.chunker = llmCountingChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}, llmCalls: &llmCalls} + store := &recordingChunkPlanStore{record: chunkPlanRecord(t, prepared, plan), decision: tc.decision} + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: tc.mode, ChunkPlans: store}); err != nil { + t.Fatalf("Run() error = %v", err) + } + if store.loads != tc.wantLoads || store.saves != tc.wantSaves || calls != tc.wantCalls { + t.Fatalf("calls = load %d save %d module %d, want %d %d %d", store.loads, store.saves, calls, tc.wantLoads, tc.wantSaves, tc.wantCalls) + } + }) + } +} + +func TestRunnerValidatesChunkPlanPolicyInputs(t *testing.T) { + for _, tc := range []struct { + name string + mode ChunkCacheMode + want string + }{ + {name: "auto requires store", mode: ChunkCacheAuto, want: "store is required"}, + {name: "refresh requires store", mode: ChunkCacheRefresh, want: "store is required"}, + {name: "invalid mode", mode: "sometimes", want: "not supported"}, + } { + t.Run(tc.name, func(t *testing.T) { + prepared, _ := preparedTerminalDebugPipeline(t) + _, err := New().Run(context.Background(), RunInput{Prepared: prepared, ChunkCacheMode: tc.mode}) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("Run() error = %v, want %q", err, tc.want) + } + }) + } +} + +func TestRunnerChunkPlanStoreErrorsAreFrameworkErrors(t *testing.T) { + for _, tc := range []struct { + name string + mode ChunkCacheMode + decision ChunkPlanDecision + loadErr error + saveErr error + wantError string + wantCalls int + }{ + {name: "load", mode: ChunkCacheAuto, loadErr: errors.New("read failed"), wantError: "load chunk plan", wantCalls: 0}, + {name: "save", mode: ChunkCacheRefresh, saveErr: errors.New("write failed"), wantError: "save chunk plan", wantCalls: 1}, + } { + t.Run(tc.name, func(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + calls := 0 + llmCalls := 0 + prepared.chunker = llmCountingChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}, llmCalls: &llmCalls} + store := &recordingChunkPlanStore{decision: tc.decision, loadErr: tc.loadErr, saveErr: tc.saveErr} + _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: tc.mode, ChunkPlans: store}) + if err == nil || !strings.Contains(err.Error(), tc.wantError) || calls != tc.wantCalls { + t.Fatalf("Run() error = %v, module calls = %d, want %q and %d", err, calls, tc.wantError, tc.wantCalls) + } + }) + } +} + +func TestRunnerRegeneratesStructurallyInvalidHit(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + calls := 0 + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls} + invalid := chunkPlanRecord(t, prepared, plan) + invalid.Plan.Ranges[0].StartUnitID = 999 + store := &recordingChunkPlanStore{record: invalid, decision: ChunkPlanDecision{Status: ChunkPlanHit}} + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store}); err != nil { + t.Fatal(err) + } + if calls != 1 || store.loads != 1 || store.saves != 1 { + t.Fatalf("calls = module %d load %d save %d, want 1 1 1", calls, store.loads, store.saves) + } +} + +func TestRunnerRetriesBeforePublishingAcceptedPlan(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.Chunk.Retries = 1 + chunker := &retryingChunker{key: prepared.resolved.Chunk.Module, plan: plan} + prepared.chunker = chunker + store := &recordingChunkPlanStore{decision: ChunkPlanDecision{Status: ChunkPlanMissing}} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store}) + if err != nil { + t.Fatal(err) + } + if chunker.calls != 2 || store.saves != 1 { + t.Fatalf("module calls = %d saves = %d, want 2 and 1", chunker.calls, store.saves) + } + if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" || len(store.saved.Warnings) != 1 || store.saved.Warnings[0].Scope != "accepted" { + t.Fatalf("output warnings = %#v stored warnings = %#v", output.Warnings, store.saved.Warnings) + } +} + +func TestRunnerAutoHitValidatesOnceWithoutRegenerationOrMutation(t *testing.T) { + tests := []struct { + name string + result contracts.ValidationResult + validatorErr error + wantError string + wantReject bool + }{ + {name: "warning", result: contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: "current", ReasonCode: "observed", Message: "current warning"}}}}, + {name: "rejection", result: contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "rejected hit"}, wantReject: true}, + {name: "error", validatorErr: errors.New("validator failed"), wantError: "validator failed"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + calls := 0 + llmCalls := 0 + prepared.chunker = llmCountingChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}, llmCalls: &llmCalls} + validator := &countingChunkValidator{result: tc.result, err: tc.validatorErr} + prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}} + record := chunkPlanRecord(t, prepared, plan) + record.Warnings = []contracts.Warning{{Scope: "stored", ReasonCode: "observed", Message: "stored warning"}} + store := &recordingChunkPlanStore{record: record, decision: ChunkPlanDecision{Status: ChunkPlanHit}} + debug := newCapturedDebugRecorder() + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store, Debug: debug}) + if tc.wantError != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantError) { + t.Fatalf("Run() error = %v, want %q", err, tc.wantError) + } + } else if err != nil { + t.Fatalf("Run() error = %v", err) + } + if validator.calls != 1 || calls != 0 || llmCalls != 0 || store.loads != 1 || store.saves != 0 { + t.Fatalf("calls = validator %d module %d llm %d load %d save %d", validator.calls, calls, llmCalls, store.loads, store.saves) + } + assertAttemptEnvelopeSequence(t, debug, "chunk") + if tc.wantError == "" { + encoded := string(debug.json["chunk/output.json"]) + if !strings.Contains(encoded, `"status":"hit"`) || !strings.Contains(encoded, `"plan":`) || !strings.Contains(encoded, `"materialized_chunks":`) { + t.Fatalf("chunk hit debug = %s", encoded) + } + } + if tc.wantReject && (len(output.Rejected) != 1 || output.Rejected[0].ReasonCode != "rejected") { + t.Fatalf("rejected = %#v", output.Rejected) + } + if tc.wantError == "" && len(output.Warnings) != 1+len(tc.result.Warnings) { + t.Fatalf("warnings = %#v, want stored warning once plus current warnings", output.Warnings) + } + }) + } +} + +func TestRunnerPublishesOnlyAcceptedGeneratedPlans(t *testing.T) { + for _, tc := range []struct { + name string + moduleErr error + validator contracts.ValidationResult + cancel bool + wantCalls int + wantReject bool + }{ + {name: "module error", moduleErr: errors.New("generation failed"), wantCalls: 2}, + {name: "validator rejection", validator: contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "rejected generated plan"}, wantCalls: 2, wantReject: true}, + {name: "cancellation", cancel: true}, + } { + t.Run(tc.name, func(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.Chunk.Retries = 1 + calls := 0 + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls, err: tc.moduleErr} + if tc.wantReject { + validator := &countingChunkValidator{result: tc.validator} + prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}} + } + ctx := context.Background() + if tc.cancel { + cancelled, cancel := context.WithCancel(ctx) + cancel() + ctx = cancelled + } + store := &recordingChunkPlanStore{decision: ChunkPlanDecision{Status: ChunkPlanInvalid}} + output, err := New().Run(ctx, RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store}) + if tc.wantReject { + if err != nil || len(output.Rejected) != 1 { + t.Fatalf("Run() error = %v rejected = %#v", err, output.Rejected) + } + } else if err == nil { + t.Fatal("Run() error = nil") + } + if calls != tc.wantCalls || store.saves != 0 { + t.Fatalf("module calls = %d, saves = %d; want %d, 0", calls, store.saves, tc.wantCalls) + } + }) + } +} + +func TestRunnerStoresProducerProvenanceAndProducerWarnings(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.Chunk.LLMProfile = "chunk-profile" + prepared.resolved.ChunkReferences = ResolvedReferenceTarget{ + Stage: StageChunk, + ReferenceSet: contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ + "guide": {Items: []contracts.ReferenceItem{{SlotName: "guide", Digest: "sha256:guide", Origin: contracts.ReferenceOrigin{Type: "file", URI: "file:///guide.txt"}, Content: []byte("sensitive")}}}, + }}, + } + prepared.chunker = manifestChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, warnings: []contracts.Warning{{Scope: "producer", ReasonCode: "observed", Message: "producer warning"}}}, metadata: map[string]any{"prompt_id": "chunk/prompt"}} + validator := &countingChunkValidator{result: contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: "validator", ReasonCode: "observed", Message: "validator warning"}}}} + prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}} + store := &recordingChunkPlanStore{} + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: store}); err != nil { + t.Fatal(err) + } + producer := store.saved.Producer + if producer.InputModule != prepared.input.Key() || producer.ChunkModule != prepared.chunker.Key() || producer.LLMProfile != "chunk-profile" || producer.Metadata["prompt_id"] != "chunk/prompt" { + t.Fatalf("producer = %#v", producer) + } + if len(producer.References) != 1 || producer.References[0].Digest != "sha256:guide" { + t.Fatalf("producer references = %#v", producer.References) + } + if len(store.saved.Warnings) != 1 || store.saved.Warnings[0].Scope != "producer" { + t.Fatalf("stored warnings = %#v, want only producer warning", store.saved.Warnings) + } +} + +func TestRunnerRefreshChangesDownstreamChunkFingerprint(t *testing.T) { + doc := typedTestDocumentWithUnits(2) + prepared := preparedConcurrentPipeline(t, 1) + prepared.lanes = prepared.lanes[:1] + prepared.resolved.ArtifactLanes = prepared.resolved.ArtifactLanes[:1] + prepared.ArtifactLanes = prepared.ArtifactLanes[:1] + prepared.input = &typedTestInput{key: prepared.resolved.Input.Module, doc: doc} + + run := func(plan source.ChunkPlan) []CheckpointFingerprint { + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan} + loader := &dependencyLoader{CheckpointLoader: NoopCheckpointLoader()} + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: &recordingChunkPlanStore{}, Checkpoint: loader}); err != nil { + t.Fatal(err) + } + return loader.extractDependencies + } + separate := typedTestPlan(doc) + combined := source.ChunkPlan{SourceDigest: doc.Digest, Ranges: []source.ChunkRange{{StartUnitID: 1, EndUnitID: 2}}} + if first, second := run(separate), run(combined); reflect.DeepEqual(first, second) { + t.Fatalf("downstream fingerprints did not change: %#v", first) + } +} + +func chunkPlanRecord(t *testing.T, prepared *PreparedPipeline, plan source.ChunkPlan) ChunkPlanRecord { + t.Helper() + digest, err := source.DigestChunkPlan(plan) + if err != nil { + t.Fatal(err) + } + return ChunkPlanRecord{ + SchemaVersion: ChunkPlanSchemaVersion, + SourceDigest: plan.SourceDigest, + PlanDigest: digest, + Plan: source.CloneChunkPlan(plan), + Producer: ChunkPlanProducer{InputModule: prepared.input.Key(), ChunkModule: prepared.chunker.Key()}, + CreatedAt: time.Now().UTC(), + } +} diff --git a/internal/framework/pipeline/synchronized_collaborators.go b/internal/framework/pipeline/synchronized_collaborators.go index 2a00c92..d818400 100644 --- a/internal/framework/pipeline/synchronized_collaborators.go +++ b/internal/framework/pipeline/synchronized_collaborators.go @@ -61,11 +61,6 @@ func (l *lockedCheckpointLoader) Source(key string) (SourceCheckpoint, Checkpoin defer l.mu.Unlock() return l.inner.Source(key) } -func (l *lockedCheckpointLoader) Chunk(key, digest string) (ChunkCheckpoint, CheckpointDecision) { - l.mu.Lock() - defer l.mu.Unlock() - return l.inner.Chunk(key, digest) -} func (l *lockedCheckpointLoader) Extract(lane, key string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { l.mu.Lock() defer l.mu.Unlock() @@ -107,18 +102,6 @@ func (r *lockedCheckpointRecorder) SourceSucceeded(key string, doc *source.Sourc func (r *lockedCheckpointRecorder) SourceFailed(key string, err error) error { return r.call(func() error { return r.inner.SourceFailed(key, err) }) } -func (r *lockedCheckpointRecorder) ChunkRunning(key, digest string) error { - return r.call(func() error { return r.inner.ChunkRunning(key, digest) }) -} -func (r *lockedCheckpointRecorder) ChunkSucceeded(key, digest string, chunks []source.Chunk, warnings []contracts.Warning) error { - return r.call(func() error { return r.inner.ChunkSucceeded(key, digest, chunks, warnings) }) -} -func (r *lockedCheckpointRecorder) ChunkRejected(key, digest string, rejected contracts.RejectedOutput) error { - return r.call(func() error { return r.inner.ChunkRejected(key, digest, rejected) }) -} -func (r *lockedCheckpointRecorder) ChunkFailed(key, digest string, err error) error { - return r.call(func() error { return r.inner.ChunkFailed(key, digest, err) }) -} func (r *lockedCheckpointRecorder) ExtractRunning(lane, key string, deps []CheckpointFingerprint) error { return r.call(func() error { return r.inner.ExtractRunning(lane, key, deps) }) }