eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Expose chunk plan provenance and safe diagnostics

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-18 00:45:01 +00:00
parent 8ba5228c01
commit 6fc6ce0adb
15 changed files with 363 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/roadmap/implementation.md
View File

@@ -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

8
internal/cli/run.go
View File

@@ -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))
}

30
internal/core/artifacts/artifacts.go
View File

@@ -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"`

32
internal/core/artifacts/artifacts_test.go
View File

@@ -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",

1
internal/core/diagnostics/artifacts.go
View File

@@ -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"

1
internal/core/diagnostics/artifacts_test.go
View File

@@ -10,6 +10,7 @@ func TestArtifactNamesUseExtractionOrientedNames(t *testing.T) {
ArtifactResolvedReferences,
ArtifactSourceDocument,
ArtifactRunManifest,
ArtifactChunkPlan,
ArtifactRunReport,
ArtifactWarnings,
ArtifactErrorLog,

4
internal/core/diagnostics/run_dir.go
View File

@@ -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)
}

13
internal/core/diagnostics/run_dir_test.go
View File

@@ -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) {

4
internal/framework/contracts/contracts.go
View File

@@ -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"`

48
internal/framework/pipeline/runner.go
View File

@@ -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

134
internal/framework/pipeline/runner_chunk_plan.go
View File

@@ -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
}

96
internal/framework/pipeline/runner_chunk_plan_test.go
View File

@@ -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)

4
internal/modules/dnd/chunk/scenes/chunker.go
View File

@@ -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)
}

4
internal/modules/generic/chunk/units/chunker.go
View File

@@ -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
}

23
internal/modules/generic/output/json/encoder.go
View File

@@ -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...)
}