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

Add durable validation provenance

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-27 01:27:25 +00:00
parent 7b5f4ebd42
commit 7f01c3e79e
15 changed files with 508 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
internal/framework/pipeline/debug.go
View File

@@ -42,6 +42,7 @@ type debugTimedEnvelope struct {
LaneID string `json:"lane_id,omitempty"`
ModuleKey string `json:"module_key,omitempty"`
Attempt int `json:"attempt,omitempty"`
AttemptKind string `json:"attempt_kind,omitempty"`
StartedAt time.Time `json:"started_at"`
CompletedAt time.Time `json:"completed_at"`
DurationMS int64 `json:"duration_ms"`
@@ -143,9 +144,37 @@ type debugLLMCallReference struct {
PromptID string `json:"prompt_id,omitempty"`
ProfileID string `json:"profile_id,omitempty"`
Model string `json:"model,omitempty"`
PromptTokens int `json:"prompt_tokens,omitempty"`
CompletionTokens int `json:"completion_tokens,omitempty"`
TotalTokens int `json:"total_tokens,omitempty"`
RepairAttempts int `json:"repair_attempts,omitempty"`
Error bool `json:"error,omitempty"`
}
type debugValidationOutcome struct {
ValidatorName string `json:"validator_name"`
Outcome string `json:"outcome"`
AttemptCount int `json:"attempt_count"`
ReasonCode string `json:"reason_code,omitempty"`
}
type debugProducerAttempt struct {
Number int `json:"number"`
Kind string `json:"kind"`
Outcome string `json:"outcome"`
Validation []debugValidationOutcome `json:"validation,omitempty"`
}
type debugProducerTerminal struct {
ProducerAttemptCount int `json:"producer_attempt_count"`
Attempts []debugProducerAttempt `json:"attempts,omitempty"`
AggregateReasonCodes []string `json:"aggregate_reason_codes,omitempty"`
ValidationComplete bool `json:"validation_complete"`
EffectivePolicy ValidationPolicy `json:"effective_policy"`
TerminalAction string `json:"terminal_action"`
Summary artifacts.ValidationSummary `json:"validation_summary"`
}
type debugValidationCall struct {
ValidatorName string `json:"validator_name"`
Request any `json:"request"`
@@ -250,6 +279,10 @@ func (client *debugLLMClient) CompleteStructured(ctx context.Context, req contra
PromptID: req.PromptID,
ProfileID: debugFirstNonEmptyString(response.ProfileID, req.ProfileID),
Model: debugFirstNonEmptyString(response.Model, debugResponseModel(response)),
PromptTokens: response.PromptTokens,
CompletionTokens: response.CompletionTokens,
TotalTokens: response.TotalTokens,
RepairAttempts: response.RepairAttempts,
Error: err != nil,
}
if scope := debugLLMScopeFromContext(ctx); scope != nil {
@@ -406,6 +439,50 @@ func (r attemptTerminalRecorder) record(payload any, terminalErr error) error {
return terminalErr
}
func debugValidationOutcomes(report validationReport) []debugValidationOutcome {
if len(report.records) == 0 {
return nil
}
outcomes := make([]debugValidationOutcome, 0, len(report.records))
for _, record := range report.records {
outcomes = append(outcomes, debugValidationOutcome{
ValidatorName: record.validatorName,
Outcome: string(record.outcome),
AttemptCount: record.attemptCount,
ReasonCode: record.reasonCode,
})
}
return outcomes
}
func writeProducerTerminalDebug(recorder DebugRecorder, name string, terminal producerAttemptTerminal, policy ValidationPolicy, summary artifacts.ValidationSummary) error {
attempts := make([]debugProducerAttempt, 0, len(terminal.Provenance))
for _, item := range terminal.Provenance {
attempts = append(attempts, debugProducerAttempt{
Number: item.Number,
Kind: string(item.Kind),
Outcome: string(item.Outcome),
Validation: debugValidationOutcomes(item.Validation),
})
}
return writeDebugTimed(recorder, name, debugTimedEnvelope{
Stage: summary.Stage,
StepID: summary.StepID,
LaneID: summary.LaneID,
ModuleKey: summary.ModuleKey,
StartedAt: time.Now().UTC(),
Payload: debugProducerTerminal{
ProducerAttemptCount: summary.ProducerAttemptCount,
Attempts: attempts,
AggregateReasonCodes: append([]string(nil), summary.ReasonCodes...),
ValidationComplete: summary.Status == "complete",
EffectivePolicy: policy,
TerminalAction: string(terminal.Action),
Summary: artifacts.CloneValidationSummary(summary),
},
})
}
func debugContentEnvelope(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) debugBinaryEnvelope {
content = redactSecretBytes(content)
return debugBinaryEnvelope{

78
internal/framework/pipeline/producer_attempts.go
View File

@@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
@@ -123,6 +124,7 @@ func (terminal producerAttemptTerminal) clone() producerAttemptTerminal {
terminal.Warnings = cloneWarnings(terminal.Warnings)
if terminal.Rejection != nil {
rejection := *terminal.Rejection
rejection.Validation = cloneValidationSummaryPtr(rejection.Validation)
terminal.Rejection = &rejection
}
terminal.Validation = cloneValidationReport(terminal.Validation)
@@ -246,7 +248,9 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod
if incomplete := firstIncompleteValidation(report); incomplete != nil {
provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptIncompleteAccepted, Validation: report})
if config.Policy.ValidatorFailure == ValidatorFailureWarnContinue {
return producerAttemptTerminal{Action: producerTerminalIncompleteAccepted, Value: output.Value, Warnings: terminalWarnings(output, report), Validation: report, ValidationIncomplete: true, Provenance: cloneProducerAttemptProvenance(provenance)}, nil
warnings := terminalWarnings(output, report)
warnings = append(warnings, incompleteValidationWarnings(report)...)
return producerAttemptTerminal{Action: producerTerminalIncompleteAccepted, Value: output.Value, Warnings: warnings, Validation: report, ValidationIncomplete: true, Provenance: cloneProducerAttemptProvenance(provenance)}, nil
}
return failedProducerAttempt(provenance), validatorFailureError(*incomplete)
}
@@ -305,6 +309,78 @@ func terminalWarnings(output producerAttemptOutput, report validationReport) []c
return warnings
}
// incompleteValidationWarnings reports only validators that exhausted their
// execution budget. It never reports rejected candidates, and it uses fixed
// text so provider errors and correction content cannot cross this boundary.
func incompleteValidationWarnings(report validationReport) []contracts.Warning {
warnings := make([]contracts.Warning, 0)
for _, record := range report.records {
if record.outcome != validationFailed {
continue
}
warnings = append(warnings, contracts.Warning{
Scope: record.validatorName,
ReasonCode: "validator_execution_incomplete",
Message: "Validator execution did not complete within its configured budget.",
})
}
return warnings
}
// validationSummary projects a terminal state-machine result into the durable
// bounded representation used by manifests, rejections, and receipts.
func validationSummary(terminal producerAttemptTerminal, stage ModuleStage, stepID, laneID, moduleKey, chunkID string, chunkIndex int) artifacts.ValidationSummary {
summary := artifacts.ValidationSummary{
Stage: string(stage),
StepID: stepID,
LaneID: laneID,
ModuleKey: moduleKey,
ChunkID: chunkID,
ChunkIndex: chunkIndex,
ProducerAttemptCount: len(terminal.Provenance),
TerminalAction: string(terminal.Action),
}
switch {
case terminal.Action == producerTerminalRejected:
summary.Status = "rejected"
case terminal.Action == producerTerminalFailed:
summary.Status = "incomplete"
case terminal.ValidationIncomplete:
summary.Status = "incomplete"
default:
summary.Status = "complete"
}
seenValidators := make(map[string]struct{})
seenReasons := make(map[string]struct{})
seenIncomplete := make(map[string]struct{})
for _, record := range terminal.Validation.records {
if record.reasonCode != "" {
if _, exists := seenReasons[record.reasonCode]; !exists {
seenReasons[record.reasonCode] = struct{}{}
summary.ReasonCodes = append(summary.ReasonCodes, record.reasonCode)
}
}
if record.outcome == validationRejected {
if _, exists := seenValidators[record.validatorName]; !exists {
seenValidators[record.validatorName] = struct{}{}
summary.RejectingValidators = append(summary.RejectingValidators, record.validatorName)
}
}
if record.outcome == validationFailed || record.outcome == validationSkipped {
if _, exists := seenIncomplete[record.validatorName]; !exists {
seenIncomplete[record.validatorName] = struct{}{}
summary.IncompleteValidators = append(summary.IncompleteValidators, record.validatorName)
}
}
}
if terminal.Rejection != nil && terminal.Rejection.ReasonCode != "" {
if _, exists := seenReasons[terminal.Rejection.ReasonCode]; !exists {
summary.ReasonCodes = append(summary.ReasonCodes, terminal.Rejection.ReasonCode)
}
}
return summary
}
func failedProducerAttempt(provenance []producerAttemptProvenance) producerAttemptTerminal {
return producerAttemptTerminal{Action: producerTerminalFailed, Provenance: cloneProducerAttemptProvenance(provenance)}
}

56
internal/framework/pipeline/producer_attempts_test.go
View File

@@ -280,6 +280,62 @@ func TestRunProducerAttemptsRejectionWinsOverValidatorFailure(t *testing.T) {
}
}
func TestValidationSummaryIsBoundedAndCorrectedSuccessIsQuiet(t *testing.T) {
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Retries: 1, Policy: DefaultValidationPolicy()}, func(_ context.Context, request producerAttemptRequest) (producerAttemptOutput, error) {
if request.Number == 1 {
return producerAttemptOutput{Value: "first", Candidate: attemptCandidate(t, "sensitive defective response")}, nil
}
return producerAttemptOutput{Value: "corrected", Candidate: attemptCandidate(t, "sensitive corrected response")}, nil
}, func(_ context.Context, output producerAttemptOutput) (validationReport, error) {
if output.Value == "first" {
return validationReport{records: []validationRecord{{validatorName: "first", outcome: validationRejected, reasonCode: "needs_fix", message: "sensitive diagnostic", correctionGuidance: "sensitive guidance"}}}, nil
}
return validationReport{}, nil
})
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
summary := validationSummary(terminal, StageExtract, "step", "lane", "module", "chunk", 3)
if summary.Status != "complete" || summary.ProducerAttemptCount != 2 || summary.TerminalAction != string(producerTerminalAccepted) {
t.Fatalf("summary = %#v", summary)
}
if len(summary.ReasonCodes) != 0 || len(summary.RejectingValidators) != 0 || len(summary.IncompleteValidators) != 0 {
t.Fatalf("corrected success summary retained prior findings: %#v", summary)
}
if len(terminal.Warnings) != 0 {
t.Fatalf("corrected success warnings = %#v, want none", terminal.Warnings)
}
}
func TestWarnContinueRecordsOneWarningForEachExhaustedValidator(t *testing.T) {
report := validationReport{records: []validationRecord{
{validatorName: "first", outcome: validationFailed, attemptCount: 2, failure: errors.New("provider error with sensitive details")},
{validatorName: "second", outcome: validationSkipped, attemptCount: 1, reasonCode: "missing_prerequisite", message: "sensitive skipped detail"},
{validatorName: "third", outcome: validationFailed, attemptCount: 1, failure: errors.New("other provider error")},
}}
terminal, err := runProducerAttempts(context.Background(), producerAttemptConfig{Policy: DefaultValidationPolicy()}, func(context.Context, producerAttemptRequest) (producerAttemptOutput, error) {
return producerAttemptOutput{Value: "candidate"}, nil
}, func(context.Context, producerAttemptOutput) (validationReport, error) {
return report, nil
})
if err != nil {
t.Fatalf("runProducerAttempts() error = %v", err)
}
if terminal.Action != producerTerminalIncompleteAccepted {
t.Fatalf("terminal action = %q", terminal.Action)
}
if got, want := terminal.Warnings, []contracts.Warning{
{Scope: "first", ReasonCode: "validator_execution_incomplete", Message: "Validator execution did not complete within its configured budget."},
{Scope: "third", ReasonCode: "validator_execution_incomplete", Message: "Validator execution did not complete within its configured budget."},
}; !reflect.DeepEqual(got, want) {
t.Fatalf("warnings = %#v, want %#v", got, want)
}
summary := validationSummary(terminal, StageNormalize, "step", "lane", "module", "", 0)
if summary.Status != "incomplete" || !reflect.DeepEqual(summary.IncompleteValidators, []string{"first", "second", "third"}) || !reflect.DeepEqual(summary.ReasonCodes, []string{"missing_prerequisite"}) {
t.Fatalf("summary = %#v", summary)
}
}
func TestRunProducerAttemptsStopsForCancellationAndDebugFailure(t *testing.T) {
tests := []struct {
name string

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

@@ -68,13 +68,14 @@ 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"`
OutputFiles []contracts.OutputFile `json:"-"`
CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"`
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"`
OutputFiles []contracts.OutputFile `json:"-"`
CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"`
ValidationSummaries []artifacts.ValidationSummary `json:"validation_summaries,omitempty"`
}
func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err error) {
@@ -241,6 +242,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
if chunkResult.rejection != nil {
output.Rejected = append(output.Rejected, *chunkResult.rejection)
}
if chunkResult.validation != nil {
output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(*chunkResult.validation))
}
output.Warnings = append(output.Warnings, chunkResult.warnings...)
chunkDebugPayload := map[string]any{
"cache_mode": chunkMode,
@@ -297,6 +301,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
if len(output.Rejected) > 0 {
output.Manifest.ValidationStatus = "rejected"
} else if hasIncompleteValidation(output.ValidationSummaries) {
output.Manifest.ValidationStatus = "incomplete"
} else {
output.Manifest.ValidationStatus = "approved"
}
@@ -401,6 +407,15 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
return output, nil
}
func hasIncompleteValidation(summaries []artifacts.ValidationSummary) bool {
for _, summary := range summaries {
if summary.Status == "incomplete" {
return true
}
}
return false
}
func (r *Runner) runPreparedSteps(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, output *RunOutput) error {
for _, step := range input.Prepared.Steps {
stepInput := input
@@ -720,6 +735,9 @@ func populateOutputManifest(output *RunOutput) {
}
output.Manifest.NormalizedOutputs = normalizedOutputManifests(output.NormalizeOutputs)
output.Manifest.RejectedOutputs = rejectedOutputManifests(output.Rejected)
if len(output.ValidationSummaries) > 0 {
output.Manifest.ValidationSummaries = cloneValidationSummaries(output.ValidationSummaries)
}
if len(output.CheckpointEvents) > 0 {
decisions := make([]artifacts.CheckpointDecisionManifest, 0, len(output.CheckpointEvents))
for _, event := range output.CheckpointEvents {
@@ -769,6 +787,7 @@ func rejectedOutputManifests(rejected []contracts.RejectedOutput) []artifacts.Re
Message: output.Message,
AttemptCount: output.AttemptCount,
DiagnosticArtifactPath: output.DiagnosticArtifactPath,
Validation: cloneValidationSummaryPtr(output.Validation),
})
}
return manifests
@@ -1027,7 +1046,31 @@ func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.Rejec
if len(rejected) == 0 {
return nil
}
return append([]contracts.RejectedOutput(nil), rejected...)
cloned := make([]contracts.RejectedOutput, len(rejected))
for index, item := range rejected {
cloned[index] = item
cloned[index].Validation = cloneValidationSummaryPtr(item.Validation)
}
return cloned
}
func cloneValidationSummaryPtr(summary *artifacts.ValidationSummary) *artifacts.ValidationSummary {
if summary == nil {
return nil
}
cloned := artifacts.CloneValidationSummary(*summary)
return &cloned
}
func cloneValidationSummaries(summaries []artifacts.ValidationSummary) []artifacts.ValidationSummary {
if len(summaries) == 0 {
return nil
}
cloned := make([]artifacts.ValidationSummary, len(summaries))
for index, summary := range summaries {
cloned[index] = artifacts.CloneValidationSummary(summary)
}
return cloned
}
func timePtr(t time.Time) *time.Time {

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

@@ -12,15 +12,16 @@ import (
)
type chunkPlanExecution struct {
accepted bool
chunks []source.Chunk
plan *source.ChunkPlan
warnings []contracts.Warning
rejection *contracts.RejectedOutput
lookup ChunkPlanDecision
record *ChunkPlanRecord
action string
summary artifacts.ChunkPlanSummary
accepted bool
chunks []source.Chunk
plan *source.ChunkPlan
warnings []contracts.Warning
rejection *contracts.RejectedOutput
lookup ChunkPlanDecision
record *ChunkPlanRecord
action string
summary artifacts.ChunkPlanSummary
validation *artifacts.ValidationSummary
}
type generatedChunkPlanCandidate struct {
@@ -83,9 +84,14 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
result.warnings = append(cloneWarnings(record.Warnings), report.Warnings()...)
result.accepted = true
result.setValidation(report.Warnings(), nil, nil)
cachedTerminal := producerAttemptTerminal{Action: producerTerminalAccepted, Validation: report}
if firstIncompleteValidation(report) != nil {
result.summary.ValidationStatus = "incomplete"
cachedTerminal.Action = producerTerminalIncompleteAccepted
cachedTerminal.ValidationIncomplete = true
}
summary := validationSummary(cachedTerminal, StageChunk, "", "", chunker.Key(), "", 0)
result.validation = &summary
return result, nil
}
// A cache hit is not model material. Its rejection is discarded and
@@ -114,7 +120,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
attemptStarted := time.Now().UTC()
attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
attemptTerminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted})
attemptTerminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, AttemptKind: string(request.Kind), StartedAt: attemptStarted})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return producerAttemptOutput{}, attemptTerminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr))
@@ -177,6 +183,10 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
}
return report, candidate.terminal.record(payload, nil)
})
terminalSummary := validationSummary(terminal, StageChunk, "", "", chunker.Key(), "", 0)
if debugErr := writeProducerTerminalDebug(input.Debug, "chunk/terminal.json", terminal, input.pipeline.ChunkValidationPolicy, terminalSummary); debugErr != nil {
return result, debugErr
}
if err != nil {
if result.summary.ValidationStatus == "not_run" {
result.summary.ValidationStatus = "error"
@@ -190,6 +200,10 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
result.rejection.ModuleKey = chunker.Key()
}
result.warnings = cloneWarnings(terminal.Warnings)
result.validation = &terminalSummary
if result.rejection != nil {
result.rejection.Validation = cloneValidationSummaryPtr(result.validation)
}
result.setValidation(terminal.Validation.Warnings(), result.rejection, nil)
return result, nil
}
@@ -212,6 +226,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
result.plan = &candidate.plan
result.chunks = candidate.chunks
result.warnings = cloneWarnings(terminal.Warnings)
result.validation = &terminalSummary
result.setValidation(terminal.Validation.Warnings(), nil, nil)
if terminal.ValidationIncomplete {
result.summary.ValidationStatus = "incomplete"

72
internal/framework/pipeline/runner_concurrent.go
View File

@@ -10,35 +10,38 @@ import (
"sync"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type laneExtractState struct {
index int
prepared preparedLaneExecutor
deps []CheckpointFingerprint
decision CheckpointDecision
values []erasedExtractArtifact
serialized []CheckpointArtifact
warnings []contracts.Warning
rejected []contracts.RejectedOutput
incomplete []int
results map[int]extractJobResult
remaining int
failed bool
terminal bool
output RunOutput
index int
prepared preparedLaneExecutor
deps []CheckpointFingerprint
decision CheckpointDecision
values []erasedExtractArtifact
serialized []CheckpointArtifact
warnings []contracts.Warning
rejected []contracts.RejectedOutput
incomplete []int
validationSummaries []artifacts.ValidationSummary
results map[int]extractJobResult
remaining int
failed bool
terminal bool
output RunOutput
}
type finalizedExtractResults struct {
accepted []erasedExtractArtifact
serialized []CheckpointArtifact
warnings []contracts.Warning
rejected []contracts.RejectedOutput
incomplete []int
decision CheckpointDecision
accepted []erasedExtractArtifact
serialized []CheckpointArtifact
warnings []contracts.Warning
rejected []contracts.RejectedOutput
incomplete []int
validationSummaries []artifacts.ValidationSummary
decision CheckpointDecision
}
func loadExtract(loader CheckpointLoader, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
@@ -65,6 +68,7 @@ type extractJobResult struct {
warnings []contracts.Warning
rejected *contracts.RejectedOutput
validationIncomplete bool
validationSummary *artifacts.ValidationSummary
err error
}
@@ -421,7 +425,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
started := time.Now().UTC()
attemptPath := path.Join("extract", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started})
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, AttemptKind: string(request.Kind), StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr))
@@ -467,10 +471,17 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
return report, candidate.terminal.record(payload, nil)
})
result.err = err
summary := validationSummary(terminalResult, StageExtract, input.stepID, lane.ID, lane.Extract.Module, chunk.ID, chunk.Index)
if debugErr := writeProducerTerminalDebug(input.Debug, path.Join("extract", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), "terminal.json"), terminalResult, lane.ExtractValidationPolicy, summary); debugErr != nil {
result.err = errors.Join(result.err, debugErr)
return result
}
if err == nil && terminalResult.Action == producerTerminalRejected {
result.rejected = terminalResult.Rejection
if result.rejected != nil {
result.rejected.Stage, result.rejected.StepID, result.rejected.LaneID, result.rejected.ModuleKey, result.rejected.ChunkID, result.rejected.ChunkIndex = string(StageExtract), input.stepID, lane.ID, lane.Extract.Module, chunk.ID, chunk.Index
result.validationSummary = &summary
result.rejected.Validation = cloneValidationSummaryPtr(result.validationSummary)
}
result.warnings = cloneWarnings(terminalResult.Warnings)
return result
@@ -496,6 +507,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
result.value, result.serialized = candidate.artifact, stored
result.warnings = cloneWarnings(terminalResult.Warnings)
result.validationIncomplete = terminalResult.ValidationIncomplete
result.validationSummary = &summary
}
return result
}
@@ -509,6 +521,9 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, stepID string, state *l
sort.Ints(indexes)
for _, index := range indexes {
result := state.results[index]
if result.validationSummary != nil {
state.validationSummaries = append(state.validationSummaries, artifacts.CloneValidationSummary(*result.validationSummary))
}
if result.rejected != nil {
state.rejected = append(state.rejected, *result.rejected)
state.warnings = append(state.warnings, result.warnings...)
@@ -537,15 +552,17 @@ func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints C
lane := state.prepared.resolved
local := RunOutput{Manifest: manifestFromPipeline(input)}
results := finalizedExtractResults{
accepted: state.values,
serialized: state.serialized,
warnings: state.warnings,
rejected: state.rejected,
incomplete: state.incomplete,
decision: state.decision,
accepted: state.values,
serialized: state.serialized,
warnings: state.warnings,
rejected: state.rejected,
incomplete: state.incomplete,
validationSummaries: state.validationSummaries,
decision: state.decision,
}
local.Warnings = append(local.Warnings, cloneWarnings(results.warnings)...)
local.Rejected = append(local.Rejected, cloneRejectedOutputs(results.rejected)...)
local.ValidationSummaries = append(local.ValidationSummaries, cloneValidationSummaries(results.validationSummaries)...)
if err := writeDebugTimed(input.Debug, path.Join("extract", fileio.EncodePathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "decision": results.decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
return local, &laneRunError{stage: StageExtract, err: err}
}
@@ -616,6 +633,7 @@ func mergeLaneOutput(dst *RunOutput, src RunOutput) error {
dst.Rejected = append(dst.Rejected, cloneRejectedOutputs(src.Rejected)...)
dst.Warnings = append(dst.Warnings, cloneWarnings(src.Warnings)...)
dst.CheckpointEvents = append(dst.CheckpointEvents, src.CheckpointEvents...)
dst.ValidationSummaries = append(dst.ValidationSummaries, cloneValidationSummaries(src.ValidationSummaries)...)
for i := range dst.Manifest.ArtifactLanes {
for j := range src.Manifest.ArtifactLanes {
if dst.Manifest.ArtifactLanes[i].ID == src.Manifest.ArtifactLanes[j].ID && dst.Manifest.ArtifactLanes[i].StepID == src.Manifest.ArtifactLanes[j].StepID && src.Manifest.ArtifactLanes[j].Metadata != nil {

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

@@ -2,11 +2,14 @@ package pipeline
import (
"context"
"encoding/json"
"fmt"
"reflect"
"strings"
"sync"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
@@ -47,7 +50,8 @@ func TestRunnerCorrectsExtractCandidatesIndependentlyPerChunk(t *testing.T) {
return contracts.ValidationResult{Approved: true}, nil
}
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2})
debug := newCapturedDebugRecorder()
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2, Debug: debug})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
@@ -65,6 +69,38 @@ func TestRunnerCorrectsExtractCandidatesIndependentlyPerChunk(t *testing.T) {
t.Fatalf("chunk %d correction = %#v, want its exact initial response and guidance", index, correction)
}
}
terminal := string(debug.json["extract/notes/chunk-000001/terminal.json"])
if !strings.Contains(terminal, `"kind":"semantic_correction"`) || !strings.Contains(terminal, `"producer_attempt_count":2`) {
t.Fatalf("extract terminal debug = %s, want correction provenance", terminal)
}
if strings.Contains(terminal, "initial-response-0") || strings.Contains(terminal, "corrected-response-0") || strings.Contains(terminal, "correct chunk 0") {
t.Fatalf("extract terminal debug leaked correction content: %s", terminal)
}
manifest, err := json.Marshal(output.Manifest)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(manifest), "initial-response-0") || strings.Contains(string(manifest), "corrected-response-0") || strings.Contains(string(manifest), "correct chunk 0") {
t.Fatalf("manifest leaked correction content: %s", manifest)
}
}
func containsWarnings(have, want []contracts.Warning) bool {
for index := 0; index+len(want) <= len(have); index++ {
if reflect.DeepEqual(have[index:index+len(want)], want) {
return true
}
}
return false
}
func containsValidationSummary(summaries []artifacts.ValidationSummary, status string) bool {
for _, summary := range summaries {
if summary.Status == status {
return true
}
}
return false
}
func TestRunnerContinuesValidationIncompleteExtractWithoutCheckpoint(t *testing.T) {
@@ -87,6 +123,12 @@ func TestRunnerContinuesValidationIncompleteExtractWithoutCheckpoint(t *testing.
if len(output.NormalizeOutputs) != 1 || len(output.Rejected) != 0 {
t.Fatalf("run output = %#v, want accepted incomplete extract", output)
}
if output.Manifest.ValidationStatus != "incomplete" || !containsValidationSummary(output.Manifest.ValidationSummaries, "incomplete") {
t.Fatalf("manifest validation = %#v, want incomplete extract provenance", output.Manifest)
}
if got, want := output.Warnings, []contracts.Warning{{Scope: "typed/check", ReasonCode: "validator_execution_incomplete", Message: "Validator execution did not complete within its configured budget."}}; !containsWarnings(got, want) {
t.Fatalf("warnings = %#v, want %#v", got, want)
}
if len(recorder.checkpoint.Outputs) != 0 || len(recorder.checkpoint.Rejected) != 0 {
t.Fatalf("extract checkpoint = %#v, want no persisted incomplete output", recorder.checkpoint)
}

6
internal/framework/pipeline/runner_semantic_correction_test.go
View File

@@ -116,4 +116,10 @@ func TestRunnerContinuesNormalizeAfterValidatorFailure(t *testing.T) {
if len(output.NormalizeOutputs) != 1 || len(output.Rejected) != 0 || checkpoints.normalizeSucceeded != 0 {
t.Fatalf("run output = %#v checkpoints = %#v, want non-checkpointed incomplete normalized output", output, checkpoints)
}
if output.Manifest.ValidationStatus != "incomplete" || !containsValidationSummary(output.Manifest.ValidationSummaries, "incomplete") {
t.Fatalf("manifest validation = %#v, want incomplete normalization provenance", output.Manifest)
}
if got, want := output.Warnings, []contracts.Warning{{Scope: "unavailable", ReasonCode: "validator_execution_incomplete", Message: "Validator execution did not complete within its configured budget."}}; !containsWarnings(got, want) {
t.Fatalf("warnings = %#v, want %#v", got, want)
}
}

21
internal/framework/pipeline/runner_typed.go
View File

@@ -11,6 +11,7 @@ import (
"reflect"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
@@ -267,7 +268,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
started := time.Now().UTC()
attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", request.Number))
attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: request.Number, StartedAt: started})
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: request.Number, AttemptKind: string(request.Kind), StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr))
@@ -304,6 +305,11 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
}
return report, nil
})
terminalSummary := validationSummary(terminalResult, StageMerge, input.stepID, lane.ID, lane.Merge.Module, "", 0)
if debugErr := writeProducerTerminalDebug(input.Debug, path.Join("merge", fileio.EncodePathComponent(lane.ID), "terminal.json"), terminalResult, lane.MergeValidationPolicy, terminalSummary); debugErr != nil {
_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, debugErr)
return stageResult, debugErr
}
if runErr != nil {
_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, runErr)
return stageResult, runErr
@@ -314,6 +320,8 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
return stageResult, fmt.Errorf("merge attempt terminal is missing rejection")
}
rejected.Stage, rejected.StepID, rejected.LaneID, rejected.ModuleKey = string(StageMerge), input.stepID, lane.ID, lane.Merge.Module
rejected.Validation = &terminalSummary
output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(terminalSummary))
output.Warnings = append(output.Warnings, terminalResult.Warnings...)
output.Rejected = append(output.Rejected, *rejected)
if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejected); err != nil {
@@ -339,6 +347,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
}
merged, serializedMerge = candidate.artifact, stored
mergeWarnings = cloneWarnings(terminalResult.Warnings)
output.ValidationSummaries = append(output.ValidationSummaries, terminalSummary)
stageResult.validationIncomplete = terminalResult.ValidationIncomplete
output.Warnings = append(output.Warnings, mergeWarnings...)
if !stageResult.validationIncomplete {
@@ -390,7 +399,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
started := time.Now().UTC()
attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", request.Number))
attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: request.Number, StartedAt: started})
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: request.Number, AttemptKind: string(request.Kind), StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return producerAttemptOutput{}, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr))
@@ -445,6 +454,11 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
}
return report, nil
})
terminalSummary := validationSummary(terminalResult, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, "", 0)
if debugErr := writeProducerTerminalDebug(input.Debug, path.Join("normalize", fileio.EncodePathComponent(lane.ID), "terminal.json"), terminalResult, lane.NormalizeValidationPolicy, terminalSummary); debugErr != nil {
_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, debugErr)
return stageResult, debugErr
}
if runErr != nil {
_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, runErr)
return stageResult, runErr
@@ -455,6 +469,8 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
return stageResult, fmt.Errorf("normalize attempt terminal is missing rejection")
}
rejected.Stage, rejected.StepID, rejected.LaneID, rejected.ModuleKey = string(StageNormalize), input.stepID, lane.ID, lane.Normalize.Module
rejected.Validation = &terminalSummary
output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(terminalSummary))
output.Warnings = append(output.Warnings, terminalResult.Warnings...)
output.Rejected = append(output.Rejected, *rejected)
if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejected); err != nil {
@@ -482,6 +498,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
}
serializedNormalize = stored
normalizeWarnings = cloneWarnings(terminalResult.Warnings)
output.ValidationSummaries = append(output.ValidationSummaries, terminalSummary)
output.Warnings = append(output.Warnings, normalizeWarnings...)
if !terminalResult.ValidationIncomplete {
if err := recordNormalize(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil {