diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index d3703e5..ad4c4b3 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -709,7 +709,7 @@ refresh have exact state semantics, and the repository is green. ## Stage 6: Expose producer provenance and safe diagnostics -**Status:** Not started +**Status:** Complete ### Objective diff --git a/internal/cli/run.go b/internal/cli/run.go index 9bd8aeb..cadadf8 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -331,6 +331,9 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if err != nil { if output.Manifest.PipelineID != "" && runDir != nil { _ = runDir.WriteRunManifest(output.Manifest) + if output.ChunkPlan != nil { + _ = runDir.WriteChunkPlan(*output.ChunkPlan) + } _ = runDir.WriteCheckpointEvents(output.CheckpointEvents) } return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("run pipeline %q: %w", pipelineID, err)) @@ -340,6 +343,11 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if err := writeDiagnostics(runDir, func() error { return runDir.WriteRunManifest(output.Manifest) }); err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics run manifest: %w", err)) } + if output.ChunkPlan != nil { + if err := writeDiagnostics(runDir, func() error { return runDir.WriteChunkPlan(*output.ChunkPlan) }); err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics chunk plan: %w", err)) + } + } if err := writeDiagnostics(runDir, func() error { return runDir.WriteWarnings(output.Warnings) }); err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics warnings: %w", err)) } diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index 8a87d15..53a1adc 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -69,12 +69,42 @@ type RejectedOutputManifest struct { DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` } +type ChunkPlanManifest struct { + Mode string `json:"mode"` + Action string `json:"action,omitempty"` + SourceDigest string `json:"source_digest,omitempty"` + PlanDigest string `json:"plan_digest,omitempty"` + PlanSchemaVersion string `json:"plan_schema_version,omitempty"` + RequestedModule string `json:"requested_module"` + ProducerInputModule string `json:"producer_input_module,omitempty"` + ProducerModule string `json:"producer_module,omitempty"` + ProducerLLMProfile string `json:"producer_llm_profile,omitempty"` + ProducerReferences []ReferenceProvenance `json:"producer_references,omitempty"` + ProducerMetadata map[string]any `json:"producer_metadata,omitempty"` + CreatedAt *time.Time `json:"created_at,omitempty"` +} + +// ChunkPlanSummary is deliberately limited to cache and validation decisions. +// It must never contain plan units, source content, annotations, or model I/O. +type ChunkPlanSummary struct { + Mode string `json:"mode"` + SourceDigest string `json:"source_digest,omitempty"` + CandidateDigest string `json:"candidate_digest,omitempty"` + RequestedModule string `json:"requested_module"` + LookupStatus string `json:"lookup_status"` + LookupReason string `json:"lookup_reason,omitempty"` + Action string `json:"action,omitempty"` + ValidationStatus string `json:"validation_status"` + PublicationStatus string `json:"publication_status"` +} + type RunManifest struct { RunID string `json:"run_id,omitempty"` PipelineID string `json:"pipeline_id,omitempty"` PipelineDigest string `json:"pipeline_digest,omitempty"` InputModule string `json:"input_module,omitempty"` Chunker string `json:"chunker,omitempty"` + ChunkPlan *ChunkPlanManifest `json:"chunk_plan,omitempty"` SourceDigests []string `json:"source_digests,omitempty"` Extractors []string `json:"extractors,omitempty"` Merger string `json:"merger,omitempty"` diff --git a/internal/core/artifacts/artifacts_test.go b/internal/core/artifacts/artifacts_test.go index 48c51e0..1888a29 100644 --- a/internal/core/artifacts/artifacts_test.go +++ b/internal/core/artifacts/artifacts_test.go @@ -2,6 +2,7 @@ package artifacts import ( "encoding/json" + "strings" "testing" ) @@ -16,6 +17,37 @@ func TestRunManifestOmitsEmptyOptionalFields(t *testing.T) { } } +func TestRunManifestChunkPlanIsAdditiveAndOmitsPlanContent(t *testing.T) { + manifest := RunManifest{ChunkPlan: &ChunkPlanManifest{ + Mode: "auto", Action: "reused", SourceDigest: "sha256:source", PlanDigest: "sha256:plan", + PlanSchemaVersion: "notarius.chunk-plan.v1", RequestedModule: "chunk/current", + ProducerInputModule: "input/original", ProducerModule: "chunk/original", + }} + encoded, err := json.Marshal(manifest) + if err != nil { + t.Fatal(err) + } + text := string(encoded) + for _, want := range []string{`"chunk_plan"`, `"action":"reused"`, `"requested_module":"chunk/current"`, `"producer_module":"chunk/original"`} { + if !strings.Contains(text, want) { + t.Fatalf("manifest JSON %s does not contain %s", text, want) + } + } + for _, forbidden := range []string{`"plan"`, `"units"`, `"annotations"`} { + if strings.Contains(text, forbidden) { + t.Fatalf("manifest JSON contains forbidden field %s: %s", forbidden, text) + } + } + + var legacy RunManifest + if err := json.Unmarshal([]byte(`{"pipeline_id":"legacy"}`), &legacy); err != nil { + t.Fatal(err) + } + if legacy.PipelineID != "legacy" || legacy.ChunkPlan != nil { + t.Fatalf("legacy manifest = %#v", legacy) + } +} + func TestRunManifestIncludesPipelineAndArtifactLaneFields(t *testing.T) { manifest := RunManifest{ PipelineID: "pipeline-1", diff --git a/internal/core/diagnostics/artifacts.go b/internal/core/diagnostics/artifacts.go index 8dfadf4..74e8689 100644 --- a/internal/core/diagnostics/artifacts.go +++ b/internal/core/diagnostics/artifacts.go @@ -8,6 +8,7 @@ const ( ArtifactCheckpointEvents = "checkpoint-events.json" ArtifactSourceDocument = "source-document.json" ArtifactRunManifest = "run-manifest.json" + ArtifactChunkPlan = "chunk-plan.json" ArtifactRunReport = "run-report.json" ArtifactWarnings = "warnings.json" ArtifactErrorLog = "error.log" diff --git a/internal/core/diagnostics/artifacts_test.go b/internal/core/diagnostics/artifacts_test.go index 813924c..8f9e4e9 100644 --- a/internal/core/diagnostics/artifacts_test.go +++ b/internal/core/diagnostics/artifacts_test.go @@ -10,6 +10,7 @@ func TestArtifactNamesUseExtractionOrientedNames(t *testing.T) { ArtifactResolvedReferences, ArtifactSourceDocument, ArtifactRunManifest, + ArtifactChunkPlan, ArtifactRunReport, ArtifactWarnings, ArtifactErrorLog, diff --git a/internal/core/diagnostics/run_dir.go b/internal/core/diagnostics/run_dir.go index 86474eb..bc60047 100644 --- a/internal/core/diagnostics/run_dir.go +++ b/internal/core/diagnostics/run_dir.go @@ -167,6 +167,10 @@ func (r *RunDirectory) WriteRunManifest(manifest artifacts.RunManifest) error { return r.WriteJSONArtifact(ArtifactRunManifest, manifest) } +func (r *RunDirectory) WriteChunkPlan(summary artifacts.ChunkPlanSummary) error { + return r.WriteJSONArtifact(ArtifactChunkPlan, summary) +} + func (r *RunDirectory) WriteRunReport(payload any) error { return r.WriteJSONArtifact(ArtifactRunReport, payload) } diff --git a/internal/core/diagnostics/run_dir_test.go b/internal/core/diagnostics/run_dir_test.go index bdcb6f4..8716392 100644 --- a/internal/core/diagnostics/run_dir_test.go +++ b/internal/core/diagnostics/run_dir_test.go @@ -200,6 +200,9 @@ func TestWriteTypedArtifacts(t *testing.T) { if err := runDir.WriteRunManifest(artifacts.RunManifest{RunID: "run-1"}); err != nil { t.Fatalf("WriteRunManifest: %v", err) } + if err := runDir.WriteChunkPlan(artifacts.ChunkPlanSummary{Mode: "auto", RequestedModule: "chunk/test", LookupStatus: "invalid", LookupReason: "stored chunk plan failed validation", ValidationStatus: "not_run", PublicationStatus: "not_published"}); err != nil { + t.Fatalf("WriteChunkPlan: %v", err) + } if err := runDir.WriteRunReport(map[string]any{"ok": true}); err != nil { t.Fatalf("WriteRunReport: %v", err) } @@ -213,6 +216,7 @@ func TestWriteTypedArtifacts(t *testing.T) { ArtifactResolvedReferences, ArtifactSourceDocument, ArtifactRunManifest, + ArtifactChunkPlan, ArtifactRunReport, ArtifactWarnings, } { @@ -220,6 +224,15 @@ func TestWriteTypedArtifacts(t *testing.T) { t.Fatalf("expected artifact %q: %v", name, err) } } + chunkSummary, err := os.ReadFile(filepath.Join(runDir.Path(), ArtifactChunkPlan)) + if err != nil { + t.Fatal(err) + } + for _, forbidden := range []string{"units", "annotations", "source content", "prompt", "response", "raw invalid"} { + if strings.Contains(string(chunkSummary), forbidden) { + t.Fatalf("chunk plan summary leaked %q: %s", forbidden, chunkSummary) + } + } } func TestWriteRedactedEffectiveConfigWritesPayloadReturnedByProvider(t *testing.T) { diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 1b1c498..ae59fb6 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -214,6 +214,10 @@ const ( ExecutionClassLLMBacked ExecutionClass = "llm_backed" ) +type ChunkExecutionClassProvider interface { + ExecutionClass() ExecutionClass +} + type ValidationResult struct { Approved bool `json:"approved"` ReasonCode string `json:"reason_code,omitempty"` diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 19f68eb..fd42487 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -63,6 +63,7 @@ type RunInput struct { type RunOutput struct { Manifest artifacts.RunManifest `json:"manifest"` + ChunkPlan *artifacts.ChunkPlanSummary `json:"chunk_plan,omitempty"` NormalizeOutputs []contracts.SerializedOutput `json:"normalize_outputs,omitempty"` Rejected []contracts.RejectedOutput `json:"rejected,omitempty"` Warnings []contracts.Warning `json:"warnings,omitempty"` @@ -81,6 +82,11 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err input.llmClient = input.Prepared.dependencies.LLM output.Manifest = manifestFromPipeline(input) + output.ChunkPlan = &artifacts.ChunkPlanSummary{ + Mode: string(effectiveChunkCacheMode(input.ChunkCacheMode)), RequestedModule: input.pipeline.Chunk.Module, + LookupStatus: "skipped", LookupReason: "chunk plan lookup skipped", + ValidationStatus: "not_run", PublicationStatus: "not_requested", + } checkpoints := input.Checkpoints if checkpoints == nil { checkpoints = NoopCheckpointRecorder() @@ -193,6 +199,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err) } chunkResult, err := r.runChunkPlan(ctx, input, doc, sourceInput, sessionID) + applyChunkPlanExecution(&output, chunkResult) if err != nil { return failOutput(output), err } @@ -481,10 +488,14 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest { pipeline := input.pipeline manifest := artifacts.RunManifest{ - PipelineID: pipeline.ID, - PipelineDigest: pipeline.Digest, - InputModule: pipeline.Input.Module, - Chunker: pipeline.Chunk.Module, + PipelineID: pipeline.ID, + PipelineDigest: pipeline.Digest, + InputModule: pipeline.Input.Module, + Chunker: pipeline.Chunk.Module, + ChunkPlan: &artifacts.ChunkPlanManifest{ + Mode: string(effectiveChunkCacheMode(input.ChunkCacheMode)), + RequestedModule: pipeline.Chunk.Module, + }, OutputEncoder: pipeline.Output.Module, ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)), ValidatorChains: validatorChainManifests(pipeline.ValidatorChains), @@ -673,11 +684,38 @@ func cloneMetadata(metadata map[string]any) map[string]any { } out := make(map[string]any, len(metadata)) for key, value := range metadata { - out[key] = value + out[key] = cloneMetadataValue(value) } return out } +func cloneMetadataValue(value any) any { + switch typed := value.(type) { + case map[string]any: + return cloneMetadata(typed) + case []any: + out := make([]any, len(typed)) + for i := range typed { + out[i] = cloneMetadataValue(typed[i]) + } + return out + case json.RawMessage: + return append(json.RawMessage(nil), typed...) + case []byte: + return append([]byte(nil), typed...) + case []string: + return append([]string(nil), typed...) + case map[string]string: + out := make(map[string]string, len(typed)) + for key, item := range typed { + out[key] = item + } + return out + default: + return value + } +} + func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest { if len(profiles) == 0 { return nil diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index 22b560a..c16b625 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -18,6 +18,9 @@ type chunkPlanExecution struct { warnings []contracts.Warning rejection *contracts.RejectedOutput lookup ChunkPlanDecision + record *ChunkPlanRecord + action string + summary artifacts.ChunkPlanSummary } func effectiveChunkCacheMode(mode ChunkCacheMode) ChunkCacheMode { @@ -33,33 +36,50 @@ func effectiveChunkCacheMode(mode ChunkCacheMode) ChunkCacheMode { 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"}} + result := chunkPlanExecution{ + lookup: ChunkPlanDecision{Reason: "chunk plan lookup skipped"}, + summary: artifacts.ChunkPlanSummary{ + Mode: string(mode), SourceDigest: doc.Digest, RequestedModule: chunker.Key(), LookupStatus: "skipped", + LookupReason: "chunk plan lookup skipped", ValidationStatus: "not_run", PublicationStatus: "not_requested", + }, + } if mode == ChunkCacheAuto { record, decision, err := input.ChunkPlans.Load(doc.Digest) result.lookup = decision + result.summary.LookupStatus = string(decision.Status) + result.summary.LookupReason = decision.Reason if err != nil { + result.summary.LookupStatus = "skipped" + result.summary.LookupReason = "chunk plan lookup failed" return result, fmt.Errorf("load chunk plan: %w", err) } switch decision.Status { case ChunkPlanHit: plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, record.Plan) if validationErr == nil { + result.setCandidate(record, "reused") 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 + result.setValidation(validationWarnings, rejection, err) return result, err } result.lookup = ChunkPlanDecision{Status: ChunkPlanInvalid, Reason: "stored chunk plan cannot be materialized against the current source"} + result.summary.LookupStatus = "invalid" + result.summary.LookupReason = result.lookup.Reason case ChunkPlanMissing, ChunkPlanInvalid: // Generate below. default: return result, fmt.Errorf("load chunk plan returned unsupported status %q", decision.Status) } } + if mode == ChunkCacheAuto || mode == ChunkCacheRefresh { + result.summary.PublicationStatus = "not_published" + } var producerWarnings []contracts.Warning accepted, rejection, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { @@ -81,6 +101,33 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S payload := map[string]any{"plan": debugChunkPlanEnvelope(chunkResult.Plan), "warnings": debugWarningEnvelopes(chunkResult.Warnings)} return false, nil, terminal.record(payload, attemptErr) } + planDigest, digestErr := source.DigestChunkPlan(plan) + if digestErr != nil { + return false, nil, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr)) + } + producerMetadata, _ := moduleManifestMetadata(chunker) + profile := "" + if provider, ok := chunker.(contracts.ChunkExecutionClassProvider); ok && provider.ExecutionClass() == contracts.ExecutionClassLLMBacked { + profile = input.pipeline.Chunk.LLMProfile + } + candidate := ChunkPlanRecord{ + SchemaVersion: ChunkPlanSchemaVersion, SourceDigest: doc.Digest, PlanDigest: planDigest, + Plan: source.CloneChunkPlan(plan), + Producer: ChunkPlanProducer{ + InputModule: input.Prepared.input.Key(), ChunkModule: chunker.Key(), LLMProfile: profile, + References: append([]artifacts.ReferenceProvenance(nil), referenceTargetProvenance(input.pipeline.ChunkReferences)...), + Metadata: cloneMetadata(producerMetadata), + }, + Warnings: cloneWarnings(chunkResult.Warnings), CreatedAt: time.Now().UTC(), + } + action := "generated" + if mode == ChunkCacheRefresh { + action = "refreshed" + } + if mode == ChunkCacheBypass { + action = "bypassed" + } + result.setCandidate(candidate, action) 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{ @@ -88,18 +135,23 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected), } if validationErr != nil || rejected != nil { + result.setValidation(validationWarnings, rejected, validationErr) return false, rejected, terminal.record(payload, validationErr) } result.chunks = chunks result.plan = &plan result.warnings = attemptWarnings producerWarnings = cloneWarnings(chunkResult.Warnings) + result.setValidation(validationWarnings, nil, nil) if debugErr := terminal.record(payload, nil); debugErr != nil { return false, nil, debugErr } return true, nil, nil }) if err != nil { + if result.summary.ValidationStatus == "not_run" { + result.summary.ValidationStatus = "error" + } return result, err } result.accepted = accepted @@ -109,29 +161,69 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S } 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(), - } + record := cloneChunkPlanRecord(*result.record) + record.Warnings = cloneWarnings(producerWarnings) if err := input.ChunkPlans.Save(record); err != nil { return result, fmt.Errorf("save chunk plan: %w", err) } + result.summary.PublicationStatus = "published" } return result, nil } + +func (result *chunkPlanExecution) setCandidate(record ChunkPlanRecord, action string) { + cloned := cloneChunkPlanRecord(record) + result.record = &cloned + result.action = action + result.summary.Action = action + result.summary.SourceDigest = record.SourceDigest + result.summary.CandidateDigest = record.PlanDigest +} + +func (result *chunkPlanExecution) setValidation(warnings []contracts.Warning, rejection *contracts.RejectedOutput, err error) { + switch { + case err != nil: + result.summary.ValidationStatus = "error" + case rejection != nil: + result.summary.ValidationStatus = "rejected" + case len(warnings) > 0: + result.summary.ValidationStatus = "approved_with_warnings" + default: + result.summary.ValidationStatus = "approved" + } +} + +func cloneChunkPlanRecord(record ChunkPlanRecord) ChunkPlanRecord { + record.Plan = source.CloneChunkPlan(record.Plan) + record.Producer.References = append([]artifacts.ReferenceProvenance(nil), record.Producer.References...) + record.Producer.Metadata = cloneMetadata(record.Producer.Metadata) + record.Warnings = cloneWarnings(record.Warnings) + return record +} + +func applyChunkPlanExecution(output *RunOutput, result chunkPlanExecution) { + if output == nil { + return + } + summary := result.summary + output.ChunkPlan = &summary + if output.Manifest.ChunkPlan == nil { + return + } + manifest := output.Manifest.ChunkPlan + manifest.Action = result.action + if result.record == nil { + return + } + record := result.record + manifest.SourceDigest = record.SourceDigest + manifest.PlanDigest = record.PlanDigest + manifest.PlanSchemaVersion = record.SchemaVersion + manifest.ProducerInputModule = record.Producer.InputModule + manifest.ProducerModule = record.Producer.ChunkModule + manifest.ProducerLLMProfile = record.Producer.LLMProfile + manifest.ProducerReferences = append([]artifacts.ReferenceProvenance(nil), record.Producer.References...) + manifest.ProducerMetadata = cloneMetadata(record.Producer.Metadata) + createdAt := record.CreatedAt + manifest.CreatedAt = &createdAt +} diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go index 1994d67..4410f96 100644 --- a/internal/framework/pipeline/runner_chunk_plan_test.go +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -54,17 +54,30 @@ type manifestChunker struct { } func (c manifestChunker) ManifestMetadata() map[string]any { return c.metadata } +func (manifestChunker) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassLLMBacked +} type llmCountingChunker struct { terminalChunker llmCalls *int } +func (llmCountingChunker) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassLLMBacked +} + func (c llmCountingChunker) Plan(ctx context.Context, request contracts.ChunkRequest) (contracts.ChunkPlanResult, error) { (*c.llmCalls)++ return c.terminalChunker.Plan(ctx, request) } +type deterministicChunker struct{ terminalChunker } + +func (deterministicChunker) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassDeterministic +} + type retryingChunker struct { key string plan source.ChunkPlan @@ -93,19 +106,20 @@ func (l *dependencyLoader) Extract(_ string, _ string, dependencies []Checkpoint func TestRunnerChunkPlanModeMatrix(t *testing.T) { tests := []struct { - name string - mode ChunkCacheMode - decision ChunkPlanDecision - wantLoads int - wantSaves int - wantCalls int + name string + mode ChunkCacheMode + decision ChunkPlanDecision + wantLoads int + wantSaves int + wantCalls int + wantAction string }{ - {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}, + {name: "auto hit", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanHit}, wantLoads: 1, wantAction: "reused"}, + {name: "auto missing", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanMissing}, wantLoads: 1, wantSaves: 1, wantCalls: 1, wantAction: "generated"}, + {name: "auto invalid", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanInvalid}, wantLoads: 1, wantSaves: 1, wantCalls: 1, wantAction: "generated"}, + {name: "bypass", mode: ChunkCacheBypass, wantCalls: 1, wantAction: "bypassed"}, + {name: "empty bypass", wantCalls: 1, wantAction: "bypassed"}, + {name: "refresh", mode: ChunkCacheRefresh, wantSaves: 1, wantCalls: 1, wantAction: "refreshed"}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { @@ -114,16 +128,58 @@ func TestRunnerChunkPlanModeMatrix(t *testing.T) { 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 { + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: tc.mode, ChunkPlans: store}) + if 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) } + if output.Manifest.ChunkPlan == nil || output.Manifest.ChunkPlan.Action != tc.wantAction || output.ChunkPlan == nil || output.ChunkPlan.Action != tc.wantAction { + t.Fatalf("chunk plan manifest = %#v summary = %#v, want action %q", output.Manifest.ChunkPlan, output.ChunkPlan, tc.wantAction) + } }) } } +func TestRunnerChunkPlanHitUsesStoredProducerProvenance(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.Chunk.Module = "chunk/requested" + prepared.resolved.Chunk.LLMProfile = "current-profile" + record := chunkPlanRecord(t, prepared, plan) + record.Producer.InputModule = "input/original" + record.Producer.ChunkModule = "chunk/original" + record.Producer.LLMProfile = "original-profile" + record.Producer.Metadata = map[string]any{"prompt": map[string]any{"id": "original"}} + store := &recordingChunkPlanStore{record: record, decision: ChunkPlanDecision{Status: ChunkPlanHit}} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store}) + if err != nil { + t.Fatal(err) + } + manifest := output.Manifest.ChunkPlan + if manifest.RequestedModule != "chunk/requested" || manifest.ProducerModule != "chunk/original" || manifest.ProducerLLMProfile != "original-profile" { + t.Fatalf("chunk plan manifest = %#v", manifest) + } + storedNested := record.Producer.Metadata["prompt"].(map[string]any) + storedNested["id"] = "mutated" + if got := manifest.ProducerMetadata["prompt"].(map[string]any)["id"]; got != "original" { + t.Fatalf("producer metadata changed through stored alias: %v", got) + } +} + +func TestRunnerChunkPlanManifestRetainsCandidateOnRejection(t *testing.T) { + prepared, _ := preparedTerminalDebugPipeline(t) + validator := &countingChunkValidator{result: contracts.ValidationResult{Approved: false, ReasonCode: "policy", Message: "no"}} + prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: &recordingChunkPlanStore{}}) + if err != nil { + t.Fatal(err) + } + if output.Manifest.ChunkPlan.Action != "refreshed" || output.Manifest.ChunkPlan.PlanDigest == "" || output.ChunkPlan.ValidationStatus != "rejected" || output.ChunkPlan.PublicationStatus != "not_published" { + t.Fatalf("manifest = %#v summary = %#v", output.Manifest.ChunkPlan, output.ChunkPlan) + } +} + func TestRunnerValidatesChunkPlanPolicyInputs(t *testing.T) { for _, tc := range []struct { name string @@ -328,6 +384,20 @@ func TestRunnerStoresProducerProvenanceAndProducerWarnings(t *testing.T) { } } +func TestRunnerOmitsProducerProfileForDeterministicChunker(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.Chunk.LLMProfile = "configured-but-unused" + prepared.chunker = deterministicChunker{terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan}} + store := &recordingChunkPlanStore{} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: store}) + if err != nil { + t.Fatal(err) + } + if store.saved.Producer.LLMProfile != "" || output.Manifest.ChunkPlan.ProducerLLMProfile != "" { + t.Fatalf("deterministic producer profile = stored %q manifest %q", store.saved.Producer.LLMProfile, output.Manifest.ChunkPlan.ProducerLLMProfile) + } +} + func TestRunnerRefreshChangesDownstreamChunkFingerprint(t *testing.T) { doc := typedTestDocumentWithUnits(2) prepared := preparedConcurrentPipeline(t, 1) diff --git a/internal/modules/dnd/chunk/scenes/chunker.go b/internal/modules/dnd/chunk/scenes/chunker.go index c328862..b3b72de 100644 --- a/internal/modules/dnd/chunk/scenes/chunker.go +++ b/internal/modules/dnd/chunk/scenes/chunker.go @@ -51,6 +51,10 @@ func (c *Chunker) Key() string { return Key } +func (*Chunker) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassLLMBacked +} + func (c *Chunker) ReferenceSlots() []contracts.ReferenceSlot { return shared.ReferenceSlots(referenceSlotDescriptions) } diff --git a/internal/modules/generic/chunk/units/chunker.go b/internal/modules/generic/chunk/units/chunker.go index 02e1997..d908703 100644 --- a/internal/modules/generic/chunk/units/chunker.go +++ b/internal/modules/generic/chunk/units/chunker.go @@ -38,6 +38,10 @@ func (c *Chunker) Key() string { return Key } +func (*Chunker) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassDeterministic +} + func (c *Chunker) ReferenceSlots() []contracts.ReferenceSlot { return nil } diff --git a/internal/modules/generic/output/json/encoder.go b/internal/modules/generic/output/json/encoder.go index d563280..3f9082f 100644 --- a/internal/modules/generic/output/json/encoder.go +++ b/internal/modules/generic/output/json/encoder.go @@ -273,11 +273,32 @@ func cloneMetadata(metadata map[string]any) map[string]any { } out := make(map[string]any, len(metadata)) for key, value := range metadata { - out[key] = value + out[key] = cloneJSONMetadataValue(value) } return out } +func cloneJSONMetadataValue(value any) any { + switch typed := value.(type) { + case map[string]any: + return cloneMetadata(typed) + case []any: + out := make([]any, len(typed)) + for i := range typed { + out[i] = cloneJSONMetadataValue(typed[i]) + } + return out + case stdjson.RawMessage: + return append(stdjson.RawMessage(nil), typed...) + case []byte: + return append([]byte(nil), typed...) + case []string: + return append([]string(nil), typed...) + default: + return value + } +} + func encoderErrorf(format string, args ...any) error { return fmt.Errorf("json output encoder: "+format, args...) }