diff --git a/internal/cli/state_hardening_test.go b/internal/cli/state_hardening_test.go index e76a349..f44a5a0 100644 --- a/internal/cli/state_hardening_test.go +++ b/internal/cli/state_hardening_test.go @@ -24,7 +24,7 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) -const stateTestDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +const stateTestDigest = "sha256:e511d8906649b78eb639b11215fa57a9652a1a64f4aefa3ed68320dbda46f439" func TestRunStateSurfaceMatrix(t *testing.T) { for _, debug := range []bool{false, true} { @@ -857,7 +857,13 @@ type stateTestInput struct{} func (stateTestInput) Key() string { return "test/input" } func (stateTestInput) Parse(_ context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) { - return &source.SourceDocument{ID: "source", Kind: "text", Format: "text/plain", Digest: stateTestDigest, Units: []source.SourceUnit{{ID: 1, Kind: "text", Text: string(req.Raw), Ref: source.SourceRef{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}}}, nil + doc := &source.SourceDocument{ID: "source", Kind: "text", Format: "text/plain", Units: []source.SourceUnit{{ID: 1, Kind: "text", Text: string(req.Raw), Ref: source.SourceRef{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}}} + digest, err := source.DigestDocument(doc) + if err != nil { + return nil, err + } + doc.Digest = digest + return doc, nil } type stateTestChunker struct{ harness *stateTestHarness } diff --git a/internal/framework/contracts/artifact.go b/internal/framework/contracts/artifact.go index f296b82..644d12e 100644 --- a/internal/framework/contracts/artifact.go +++ b/internal/framework/contracts/artifact.go @@ -69,6 +69,16 @@ func CloneSerializedArtifact(artifact SerializedArtifact) SerializedArtifact { return artifact } +// CloneSerializedArtifactPointer returns an independently owned artifact when +// one is present. +func CloneSerializedArtifactPointer(artifact *SerializedArtifact) *SerializedArtifact { + if artifact == nil { + return nil + } + cloned := CloneSerializedArtifact(*artifact) + return &cloned +} + func CloneSerializedOutput(output SerializedOutput) SerializedOutput { output.Artifact = CloneSerializedArtifact(output.Artifact) return output diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 763b5eb..49a2d05 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -291,6 +291,7 @@ type OutputRequest struct { Warnings []Warning `json:"warnings,omitempty"` LLMProfile string `json:"llm_profile,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` + ChunkMap *SerializedArtifact `json:"chunk_map,omitempty"` } type OutputFile struct { diff --git a/internal/framework/contracts/contracts_test.go b/internal/framework/contracts/contracts_test.go index d6669a6..04f592d 100644 --- a/internal/framework/contracts/contracts_test.go +++ b/internal/framework/contracts/contracts_test.go @@ -26,6 +26,25 @@ func TestCloneReferenceSlotsEmptyInputReturnsNil(t *testing.T) { } } +func TestCloneSerializedArtifactPointerOwnsArtifactData(t *testing.T) { + original := &SerializedArtifact{ + Kind: "test/artifact", + Schema: ArtifactSchema{ID: "test.schema", JSONSchema: []byte(`{"type":"object"}`)}, + Content: []byte(`{"items":[]}`), + Metadata: map[string]any{"count": 1}, + } + cloned := CloneSerializedArtifactPointer(original) + original.Schema.JSONSchema[0] = '[' + original.Content[0] = '[' + original.Metadata["count"] = 2 + if cloned == nil || string(cloned.Schema.JSONSchema) != `{"type":"object"}` || string(cloned.Content) != `{"items":[]}` || cloned.Metadata["count"] != 1 { + t.Fatalf("CloneSerializedArtifactPointer() = %#v, want independent artifact data", cloned) + } + if CloneSerializedArtifactPointer(nil) != nil { + t.Fatal("CloneSerializedArtifactPointer(nil) must return nil") + } +} + func TestCloneReferenceSlotsPreservesFields(t *testing.T) { slots := []ReferenceSlot{ { diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 66daf49..87e11ab 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -16,6 +16,7 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/chunkmap" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) @@ -251,6 +252,28 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err) } + var acceptedChunkMap *contracts.SerializedArtifact + if chunkResult.accepted { + if chunkResult.plan == nil || chunkResult.record == nil { + return failOutput(output), fmt.Errorf("accepted chunk plan is missing plan or producer record") + } + artifact, buildErr := chunkmap.Serialize(chunkmap.BuildRequest{ + Source: doc, + Plan: *chunkResult.plan, + Chunks: chunkResult.chunks, + RequestedChunker: input.pipeline.Chunk.Module, + Producer: chunkmap.Producer{ + InputModule: chunkResult.record.Producer.InputModule, + ChunkModule: chunkResult.record.Producer.ChunkModule, + LLMProfile: chunkResult.record.Producer.LLMProfile, + }, + }) + if buildErr != nil { + return failOutput(output), fmt.Errorf("serialize accepted chunk map: %w", buildErr) + } + acceptedChunkMap = &artifact + } + if chunkResult.accepted { if err := r.runPreparedSteps(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks, &output); err != nil { return failOutput(output), err @@ -270,18 +293,22 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), err } outputStarted := time.Now().UTC() + outputDebugPayload := map[string]any{ + "manifest": output.Manifest, + "normalize_outputs": debugSerializedOutputEnvelopes(output.NormalizeOutputs), + "rejected": debugRejectedOutputEnvelopes(output.Rejected), + "warnings": output.Warnings, + "options": redactSensitiveMap(input.pipeline.Output.Options), + "metadata": redactSensitiveMap(input.Metadata), + } + if acceptedChunkMap != nil { + outputDebugPayload["chunk_map"] = debugSerializedOutputEnvelope(contracts.SerializedOutput{Artifact: *acceptedChunkMap}) + } if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{ Stage: string(StageOutput), ModuleKey: encoder.Key(), StartedAt: outputStarted, - Payload: map[string]any{ - "manifest": output.Manifest, - "normalize_outputs": debugSerializedOutputEnvelopes(output.NormalizeOutputs), - "rejected": debugRejectedOutputEnvelopes(output.Rejected), - "warnings": output.Warnings, - "options": redactSensitiveMap(input.pipeline.Output.Options), - "metadata": redactSensitiveMap(input.Metadata), - }, + Payload: outputDebugPayload, }); err != nil { return failOutput(output), fmt.Errorf("write output debug artifact: %w", err) } @@ -296,6 +323,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Warnings: output.Warnings, LLMProfile: input.pipeline.Output.LLMProfile, Metadata: outputMetadata, + ChunkMap: contracts.CloneSerializedArtifactPointer(acceptedChunkMap), }) output.Warnings = append(output.Warnings, encoded.Warnings...) if err != nil { diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go index 539cf8b..641a2c9 100644 --- a/internal/framework/pipeline/runner_chunk_plan_test.go +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -10,6 +10,7 @@ import ( "time" "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/chunkmap" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) @@ -40,6 +41,17 @@ type countingChunkValidator struct { err error } +type capturingChunkMapOutput struct { + requests []contracts.OutputRequest +} + +func (*capturingChunkMapOutput) Key() string { return "capture/chunk-map" } + +func (output *capturingChunkMapOutput) Encode(_ context.Context, request contracts.OutputRequest) (contracts.OutputResult, error) { + output.requests = append(output.requests, request) + return contracts.OutputResult{}, nil +} + func (*countingChunkValidator) Name() string { return "test/counting-chunks" } func (*countingChunkValidator) ExecutionClass() contracts.ExecutionClass { return contracts.ExecutionClassDeterministic @@ -205,6 +217,116 @@ func TestRunnerChunkPlanHitUsesStoredProducerProvenance(t *testing.T) { } } +func TestRunnerProvidesAcceptedChunkMapToOutput(t *testing.T) { + for _, test := range []struct { + name string + configure func(*PreparedPipeline, source.ChunkPlan) (RunInput, chunkmap.Producer) + }{ + { + name: "bypassed plan", + configure: func(prepared *PreparedPipeline, _ source.ChunkPlan) (RunInput, chunkmap.Producer) { + return RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheBypass}, chunkmap.Producer{ + InputModule: prepared.input.Key(), ChunkModule: prepared.chunker.Key(), + } + }, + }, + { + name: "cached plan", + configure: func(prepared *PreparedPipeline, plan source.ChunkPlan) (RunInput, chunkmap.Producer) { + prepared.resolved.Chunk.Module = "chunk/requested" + record := chunkPlanRecord(t, prepared, plan) + record.Producer.InputModule = "input/original" + record.Producer.ChunkModule = "chunk/original" + record.Producer.LLMProfile = "original-profile" + return RunInput{ + Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, + ChunkPlans: &recordingChunkPlanStore{record: record, decision: ChunkPlanDecision{Status: ChunkPlanHit}}, + }, chunkmap.Producer{InputModule: "input/original", ChunkModule: "chunk/original", LLMProfile: "original-profile"} + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + encoder := &capturingChunkMapOutput{} + prepared.output = encoder + input, wantProducer := test.configure(prepared, plan) + if _, err := New().Run(context.Background(), input); err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(encoder.requests) != 1 || encoder.requests[0].ChunkMap == nil { + t.Fatalf("output requests = %#v, want one accepted chunk map", encoder.requests) + } + artifact := encoder.requests[0].ChunkMap + if artifact.Kind != chunkmap.ArtifactKind || artifact.Schema.ID != chunkmap.SchemaID || artifact.MediaType != chunkmap.MediaType { + t.Fatalf("chunk map artifact = %#v, want fixed serialized identity", artifact) + } + value, err := chunkmap.New().Decode(artifact.Content) + if err != nil { + t.Fatalf("decode output chunk map: %v", err) + } + if value.RequestedChunker != prepared.resolved.Chunk.Module || value.Producer != wantProducer { + t.Fatalf("chunk map = %#v, want current request and producer %#v", value, wantProducer) + } + }) + } +} + +func TestRunnerOmitsChunkMapForRejectedPlan(t *testing.T) { + prepared, _ := preparedTerminalDebugPipeline(t) + encoder := &capturingChunkMapOutput{} + prepared.output = encoder + prepared.chunkValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("reject/chunk"), Target: ValidatorTargetChunk}, + chunk: &countingChunkValidator{result: contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "not accepted"}}, + }} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(output.Rejected) != 1 || len(encoder.requests) != 1 || encoder.requests[0].ChunkMap != nil { + t.Fatalf("output = %#v requests = %#v, want rejected plan without chunk map", output.Rejected, encoder.requests) + } +} + +func TestRunnerRetainsChunkMapAfterLaneRejection(t *testing.T) { + prepared, _ := preparedTerminalDebugPipeline(t) + encoder := &capturingChunkMapOutput{} + prepared.output = encoder + prepared.Steps[0].lanes[0].mergeValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("reject/lane"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "not accepted"}, nil + }, + }} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(output.Rejected) != 1 || len(encoder.requests) != 1 || encoder.requests[0].ChunkMap == nil { + t.Fatalf("output = %#v requests = %#v, want lane rejection with accepted chunk map", output.Rejected, encoder.requests) + } +} + +func TestRunnerChunkMapRequestDoesNotAliasStoredPlan(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + encoder := &capturingChunkMapOutput{} + prepared.output = encoder + record := chunkPlanRecord(t, prepared, plan) + record.Plan.Annotations = source.ChunkAnnotations{"dnd/scenes": json.RawMessage(`{"title":"Opening"}`)} + store := &recordingChunkPlanStore{record: record, decision: ChunkPlanDecision{Status: ChunkPlanHit}} + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store}); err != nil { + t.Fatalf("Run() error = %v", err) + } + store.record.Plan.Annotations["dnd/scenes"][0] = '[' + value, err := chunkmap.New().Decode(encoder.requests[0].ChunkMap.Content) + if err != nil { + t.Fatalf("decode captured chunk map after store mutation: %v", err) + } + if string(value.PlanAnnotations["dnd/scenes"]) != `{"title":"Opening"}` { + t.Fatalf("captured chunk map aliases stored plan: %s", value.PlanAnnotations["dnd/scenes"]) + } +} + func TestRunnerChunkPlanManifestRetainsCandidateOnRejection(t *testing.T) { prepared, _ := preparedTerminalDebugPipeline(t) validator := &countingChunkValidator{result: contracts.ValidationResult{Approved: false, ReasonCode: "policy", Message: "no"}}