From e61e522662be3058c0910b4efc1ed22ead7e6221 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 27 Jul 2026 18:17:54 +0000 Subject: [PATCH] Build evidence context for output encoders --- internal/framework/contracts/contracts.go | 1 + .../framework/pipeline/evidence_output.go | 103 +++++++ .../pipeline/evidence_output_test.go | 252 ++++++++++++++++++ internal/framework/pipeline/runner.go | 15 ++ 4 files changed, 371 insertions(+) create mode 100644 internal/framework/pipeline/evidence_output.go create mode 100644 internal/framework/pipeline/evidence_output_test.go diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 49a2d05..4cd3688 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -292,6 +292,7 @@ type OutputRequest struct { LLMProfile string `json:"llm_profile,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` ChunkMap *SerializedArtifact `json:"chunk_map,omitempty"` + EvidenceContext *SerializedArtifact `json:"evidence_context,omitempty"` } type OutputFile struct { diff --git a/internal/framework/pipeline/evidence_output.go b/internal/framework/pipeline/evidence_output.go new file mode 100644 index 0000000..b2079f2 --- /dev/null +++ b/internal/framework/pipeline/evidence_output.go @@ -0,0 +1,103 @@ +package pipeline + +import ( + "fmt" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext" +) + +// debugEvidenceContextSummary intentionally contains only publication-safe +// identifiers and aggregate counts. The evidence document itself can include +// source text and must never be written to this debug envelope. +type debugEvidenceContextSummary struct { + SourceID string `json:"source_id"` + SelectedLanes []string `json:"selected_lanes"` + WindowUnits int `json:"window_units"` + ContextCount int `json:"context_count"` + UnitCount int `json:"unit_count"` + SourceDigest string `json:"source_digest"` +} + +// buildOutputEvidenceContext projects the prepared output policy from accepted +// normalized artifacts. It is intentionally separate from lane execution so +// checkpointed normalized outputs use the same reconstruction path. +func buildOutputEvidenceContext(prepared *PreparedPipeline, doc *source.SourceDocument, outputs []contracts.SerializedOutput) (*contracts.SerializedArtifact, *debugEvidenceContextSummary, error) { + if prepared == nil || prepared.evidencePlan == nil { + return nil, nil, nil + } + if doc == nil { + return nil, nil, fmt.Errorf("evidence context output: source document is unavailable") + } + if prepared.artifactCodecs == nil { + return nil, nil, fmt.Errorf("evidence context output: artifact codecs are unavailable") + } + + byLane := make(map[string]contracts.SerializedOutput, len(outputs)) + for _, output := range outputs { + laneID := strings.TrimSpace(output.LaneID) + if _, exists := byLane[laneID]; exists { + return nil, nil, fmt.Errorf("evidence context output: accepted normalized outputs contain duplicate lane %q", laneID) + } + byLane[laneID] = contracts.CloneSerializedOutput(output) + } + + request := evidencecontext.BuildRequest{ + Source: doc, + WindowUnits: prepared.evidencePlan.policy.WindowUnits, + SelectedLanes: append([]string(nil), prepared.evidencePlan.policy.LaneIDs...), + LaneEvidence: make([]evidencecontext.LaneEvidence, 0, len(prepared.evidencePlan.lanes)), + } + for _, lane := range prepared.evidencePlan.lanes { + output, ok := byLane[lane.laneID] + if !ok { + continue + } + if output.SourceID != doc.ID { + return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized output source is incompatible", lane.laneID) + } + if output.Artifact.Kind != lane.kind { + return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized output artifact kind is incompatible", lane.laneID) + } + value, err := prepared.artifactCodecs.Decode(contracts.CloneSerializedArtifact(output.Artifact)) + if err != nil { + return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized artifact cannot be decoded", lane.laneID) + } + references, err := lane.project(value) + if err != nil { + return nil, nil, fmt.Errorf("evidence context output lane %q: accepted normalized artifact cannot be projected", lane.laneID) + } + request.LaneEvidence = append(request.LaneEvidence, evidencecontext.LaneEvidence{ + LaneID: lane.laneID, + SourceRefs: append([]source.SourceRef(nil), references...), + }) + } + + document, err := evidencecontext.Build(request) + if err != nil { + return nil, nil, fmt.Errorf("evidence context output: accepted evidence references are invalid") + } + content, err := evidencecontext.New().Encode(document) + if err != nil { + return nil, nil, fmt.Errorf("evidence context output: evidence context serialization failed") + } + artifact := &contracts.SerializedArtifact{ + Kind: evidencecontext.ArtifactKind, + Schema: evidencecontext.New().Schema(), + MediaType: evidencecontext.MediaType, + Content: content, + } + summary := debugEvidenceContextSummary{ + SourceID: document.SourceID, + SelectedLanes: append([]string(nil), document.SelectedLanes...), + WindowUnits: document.WindowUnits, + ContextCount: len(document.Contexts), + SourceDigest: document.SourceDigest, + } + for _, context := range document.Contexts { + summary.UnitCount += len(context.Units) + } + return contracts.CloneSerializedArtifactPointer(artifact), &summary, nil +} diff --git a/internal/framework/pipeline/evidence_output_test.go b/internal/framework/pipeline/evidence_output_test.go new file mode 100644 index 0000000..44c40cc --- /dev/null +++ b/internal/framework/pipeline/evidence_output_test.go @@ -0,0 +1,252 @@ +package pipeline + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/evidencecontext" +) + +type capturingEvidenceOutput struct { + requests []contracts.OutputRequest +} + +func (*capturingEvidenceOutput) Key() string { return "capture/evidence-context" } + +func (output *capturingEvidenceOutput) Encode(_ context.Context, request contracts.OutputRequest) (contracts.OutputResult, error) { + request.EvidenceContext = contracts.CloneSerializedArtifactPointer(request.EvidenceContext) + output.requests = append(output.requests, request) + return contracts.OutputResult{}, nil +} + +func installEvidencePlan(prepared *PreparedPipeline, window int, selected []string, project func(codecNotes) ([]source.SourceRef, error)) { + lanes := make([]preparedEvidenceLane, 0, len(selected)) + for _, laneID := range selected { + for _, step := range prepared.Steps { + for _, lane := range step.lanes { + if lane.resolved.ID != laneID { + continue + } + lanes = append(lanes, preparedEvidenceLane{ + laneID: laneID, + kind: lane.resolved.ArtifactKind, + project: func(value any) ([]source.SourceRef, error) { + notes, ok := value.(codecNotes) + if !ok { + return nil, errors.New("unexpected artifact type") + } + return project(notes) + }, + }) + } + } + } + prepared.evidencePlan = &preparedEvidencePlan{policy: EvidenceContextPolicy{Enabled: true, WindowUnits: window, LaneIDs: append([]string(nil), selected...)}, lanes: lanes} +} + +func setNormalizedNotes(prepared *PreparedPipeline, values map[string]codecNotes) { + for stepIndex := range prepared.Steps { + for laneIndex := range prepared.Steps[stepIndex].lanes { + lane := &prepared.Steps[stepIndex].lanes[laneIndex] + value, ok := values[lane.resolved.ID] + if !ok { + continue + } + lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{Value: value}, nil + } + } + } +} + +func decodeCapturedEvidence(t *testing.T, output *capturingEvidenceOutput) evidencecontext.Document { + t.Helper() + if len(output.requests) != 1 || output.requests[0].EvidenceContext == nil { + t.Fatalf("output requests = %#v, want one evidence context", output.requests) + } + value, err := evidencecontext.New().Decode(output.requests[0].EvidenceContext.Content) + if err != nil { + t.Fatalf("Decode(evidence context): %v", err) + } + return value +} + +func TestRunnerBuildsEvidenceContextFromSelectedNormalizedOutputs(t *testing.T) { + prepared := preparedOrderedPipeline(t, 3, orderedLaneSpec{id: "alpha", profile: "notes"}, orderedLaneSpec{id: "beta", profile: "notes"}) + encoder := &capturingEvidenceOutput{} + prepared.output = encoder + setNormalizedNotes(prepared, map[string]codecNotes{"alpha": {Items: []string{"one"}}, "beta": {Items: []string{"two"}}}) + installEvidencePlan(prepared, 1, []string{"alpha", "beta", "inactive"}, func(notes codecNotes) ([]source.SourceRef, error) { + switch notes.Items[0] { + case "one": + return []source.SourceRef{{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}, nil + case "two": + return []source.SourceRef{{SourceID: "source", StartUnitID: 2, EndUnitID: 2}}, nil + default: + return nil, nil + } + }) + debug := newCapturedDebugRecorder() + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}); err != nil { + t.Fatalf("Run() error = %v", err) + } + + value := decodeCapturedEvidence(t, encoder) + if !reflect.DeepEqual(value.SelectedLanes, []string{"alpha", "beta", "inactive"}) || len(value.Contexts) != 1 || len(value.Contexts[0].Units) != 3 { + t.Fatalf("evidence context = %#v, want selected union", value) + } + if got := value.Contexts[0].EvidenceRefs; len(got) != 2 || got[0].LaneID != "alpha" || got[1].LaneID != "beta" { + t.Fatalf("evidence refs = %#v, want both selected lanes", got) + } + debugJSON := string(debug.json["output/evidence-context.json"]) + if strings.Contains(debugJSON, "text-1") || strings.Contains(debugJSON, "metadata") || !strings.Contains(debugJSON, `"context_count":1`) || !strings.Contains(debugJSON, `"unit_count":3`) { + t.Fatalf("evidence debug envelope = %s, want only allowlisted summary", debugJSON) + } +} + +func TestRunnerEvidenceContextOmitsAbsentAndRejectedLanes(t *testing.T) { + prepared := preparedOrderedPipeline(t, 2, orderedLaneSpec{id: "present", profile: "notes"}, orderedLaneSpec{id: "rejected", profile: "notes"}) + encoder := &capturingEvidenceOutput{} + prepared.output = encoder + setNormalizedNotes(prepared, map[string]codecNotes{"present": {Items: []string{"present"}}, "rejected": {Items: []string{"rejected"}}}) + prepared.Steps[1].lanes[0].mergeValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("reject"), 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 + }, + }} + installEvidencePlan(prepared, 0, []string{"absent", "present", "rejected"}, func(notes codecNotes) ([]source.SourceRef, error) { + if len(notes.Items) > 0 && notes.Items[0] == "present" { + return []source.SourceRef{{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}, nil + } + return []source.SourceRef{{SourceID: "source", StartUnitID: 2, EndUnitID: 2}}, nil + }) + result, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(result.Rejected) != 1 || result.Rejected[0].LaneID != "rejected" { + t.Fatalf("rejections = %#v, want rejected lane unchanged", result.Rejected) + } + value := decodeCapturedEvidence(t, encoder) + if len(value.Contexts) != 1 || len(value.Contexts[0].EvidenceRefs) != 1 || value.Contexts[0].EvidenceRefs[0].LaneID != "present" { + t.Fatalf("evidence context = %#v, want present lane only", value) + } +} + +func TestRunnerEvidenceContextFailurePreventsOutputEncoding(t *testing.T) { + prepared := preparedConcurrentPipeline(t, 1) + encoder := &capturingEvidenceOutput{} + prepared.output = encoder + setNormalizedNotes(prepared, map[string]codecNotes{"notes": {Items: []string{"invalid"}}}) + installEvidencePlan(prepared, 0, []string{"notes"}, func(codecNotes) ([]source.SourceRef, error) { + return []source.SourceRef{{SourceID: "source", StartUnitID: 99, EndUnitID: 99}}, nil + }) + result, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) + if err == nil || !strings.Contains(err.Error(), "accepted evidence references are invalid") { + t.Fatalf("Run() error = %v, want evidence context failure", err) + } + if len(encoder.requests) != 0 || result.Manifest.ValidationStatus != "failed" || len(result.Rejected) != 0 { + t.Fatalf("output requests = %#v manifest = %#v rejected = %#v, want failed run before output encoding", encoder.requests, result.Manifest, result.Rejected) + } +} + +func TestBuildOutputEvidenceContextRejectsIncompatibleAcceptedOutputs(t *testing.T) { + prepared := preparedConcurrentPipeline(t, 1) + doc := prepared.input.(*typedTestInput).doc + installEvidencePlan(prepared, 0, []string{"notes"}, func(codecNotes) ([]source.SourceRef, error) { + return []source.SourceRef{{SourceID: doc.ID, StartUnitID: 1, EndUnitID: 1}}, nil + }) + artifact, err := prepared.artifactCodecs.Encode("test/notes", codecNotes{Items: []string{"valid"}}) + if err != nil { + t.Fatal(err) + } + valid := contracts.SerializedOutput{LaneID: "notes", SourceID: doc.ID, Artifact: artifact} + for _, test := range []struct { + name string + outputs []contracts.SerializedOutput + mutate func(*contracts.SerializedOutput) + want string + }{ + {name: "duplicate lane", outputs: []contracts.SerializedOutput{valid, valid}, want: "duplicate lane"}, + {name: "foreign source", outputs: []contracts.SerializedOutput{valid}, mutate: func(output *contracts.SerializedOutput) { output.SourceID = "other" }, want: "source is incompatible"}, + {name: "wrong artifact kind", outputs: []contracts.SerializedOutput{valid}, mutate: func(output *contracts.SerializedOutput) { output.Artifact.Kind = "test/score" }, want: "artifact kind is incompatible"}, + {name: "invalid artifact payload", outputs: []contracts.SerializedOutput{valid}, mutate: func(output *contracts.SerializedOutput) { output.Artifact.Content = []byte("not JSON") }, want: "cannot be decoded"}, + } { + t.Run(test.name, func(t *testing.T) { + outputs := append([]contracts.SerializedOutput(nil), test.outputs...) + for index := range outputs { + outputs[index] = contracts.CloneSerializedOutput(outputs[index]) + } + if test.mutate != nil { + test.mutate(&outputs[0]) + } + _, _, err := buildOutputEvidenceContext(prepared, doc, outputs) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("buildOutputEvidenceContext() error = %v, want %q", err, test.want) + } + if strings.Contains(err.Error(), "not JSON") { + t.Fatalf("buildOutputEvidenceContext() exposed artifact payload: %v", err) + } + }) + } +} + +func TestEvidenceContextOutputRequestOwnsArtifactBytes(t *testing.T) { + artifact := &contracts.SerializedArtifact{Content: []byte("original"), Metadata: map[string]any{"source": "original"}} + request := contracts.OutputRequest{EvidenceContext: contracts.CloneSerializedArtifactPointer(artifact)} + request.EvidenceContext.Content[0] = 'X' + request.EvidenceContext.Metadata["source"] = "changed" + if string(artifact.Content) != "original" || artifact.Metadata["source"] != "original" { + t.Fatalf("output request evidence context aliases source artifact: %#v", artifact) + } +} + +func TestRunnerEvidenceContextRebuildsFromAcceptedCheckpoint(t *testing.T) { + prepared := preparedConcurrentPipeline(t, 1) + encoder := &capturingEvidenceOutput{} + prepared.output = encoder + doc := prepared.input.(*typedTestInput).doc + lane := prepared.Steps[0].lanes[0] + stored, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Normalize.Module, doc.ID, codecNotes{Items: []string{"stored"}}) + if err != nil { + t.Fatal(err) + } + loader := newAcceptedCheckpointLoader() + key := CheckpointLaneKey(lane.resolved.StepID, lane.resolved.ID) + loader.accepted[key] = NormalizeCheckpoint{Output: stored} + loader.acceptedDecision[key] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) + installEvidencePlan(prepared, 0, []string{"notes"}, func(notes codecNotes) ([]source.SourceRef, error) { + if !reflect.DeepEqual(notes.Items, []string{"stored"}) { + return nil, errors.New("checkpoint artifact was not projected") + } + return []source.SourceRef{{SourceID: doc.ID, StartUnitID: 1, EndUnitID: 1}}, nil + }) + policy := CheckpointExecutionPolicy{RequireReusableLanes: map[string]struct{}{key: {}}} + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}); err != nil { + t.Fatalf("Run() error = %v", err) + } + value := decodeCapturedEvidence(t, encoder) + if len(value.Contexts) != 1 || value.Contexts[0].EvidenceRefs[0].LaneID != "notes" { + t.Fatalf("evidence context = %#v, want checkpointed normalized output", value) + } +} + +func TestRunnerSkipsEvidenceContextWhenOutputDoesNotOptIn(t *testing.T) { + prepared := preparedConcurrentPipeline(t, 1) + encoder := &capturingEvidenceOutput{} + prepared.output = encoder + projected := 0 + prepared.evidencePlan = nil + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}); err != nil { + t.Fatalf("Run() error = %v", err) + } + if projected != 0 || len(encoder.requests) != 1 || encoder.requests[0].EvidenceContext != nil { + t.Fatalf("projected = %d requests = %#v, want no evidence work", projected, encoder.requests) + } +} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 82cd730..bc3f8ab 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -294,6 +294,20 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), err } outputStarted := time.Now().UTC() + evidenceArtifact, evidenceSummary, err := buildOutputEvidenceContext(input.Prepared, doc, output.NormalizeOutputs) + if err != nil { + return failOutput(output), err + } + if evidenceSummary != nil { + if err := writeDebugTimed(debugRecorder, "output/evidence-context.json", debugTimedEnvelope{ + Stage: string(StageOutput), + ModuleKey: encoder.Key(), + StartedAt: outputStarted, + Payload: *evidenceSummary, + }); err != nil { + return failOutput(output), fmt.Errorf("write evidence context debug artifact: %w", err) + } + } outputDebugPayload := map[string]any{ "manifest": output.Manifest, "normalize_outputs": debugSerializedOutputEnvelopes(output.NormalizeOutputs), @@ -325,6 +339,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err LLMProfile: input.pipeline.Output.LLMProfile, Metadata: outputMetadata, ChunkMap: contracts.CloneSerializedArtifactPointer(acceptedChunkMap), + EvidenceContext: contracts.CloneSerializedArtifactPointer(evidenceArtifact), }) output.Warnings = append(output.Warnings, encoded.Warnings...) if err != nil {