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

Carry accepted chunk maps through the runner

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-23 14:58:14 +00:00
parent 66415fd1fa
commit 97c9a8e5ce
6 changed files with 196 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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