diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index ba7072c0..e4a687b4 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -621,7 +621,7 @@ Chunk, extract, merge, and normalize all use the same attempt and validation semantics, with no nested normalizer retry loop. This stage is one Terra prompt. -## Stage 14 — Finalize Provenance, Checkpoints, Debug, And Durable Results +## Stage 14 ✅ — Finalize Provenance, Checkpoints, Debug, And Durable Results ### Goal diff --git a/internal/cli/run_result.go b/internal/cli/run_result.go index a7e76d3a..a54816a4 100644 --- a/internal/cli/run_result.go +++ b/internal/cli/run_result.go @@ -7,22 +7,24 @@ import ( "path/filepath" "strings" + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) const runResultSchemaVersion = "notarius.run-result.v1" type runResult struct { - SchemaVersion string `json:"schema_version"` - RunID string `json:"run_id"` - PipelineID string `json:"pipeline_id"` - OutputDirectory string `json:"output_directory"` - IndexFile string `json:"index_file,omitempty"` - NormalizedOutputCount int `json:"normalized_output_count"` - RejectedOutputCount int `json:"rejected_output_count"` - WarningCount int `json:"warning_count"` - ValidationStatus string `json:"validation_status"` - DebugDirectory string `json:"debug_directory,omitempty"` + SchemaVersion string `json:"schema_version"` + RunID string `json:"run_id"` + PipelineID string `json:"pipeline_id"` + OutputDirectory string `json:"output_directory"` + IndexFile string `json:"index_file,omitempty"` + NormalizedOutputCount int `json:"normalized_output_count"` + RejectedOutputCount int `json:"rejected_output_count"` + WarningCount int `json:"warning_count"` + ValidationStatus string `json:"validation_status"` + ValidationSummaries []artifacts.ValidationSummary `json:"validation_summaries,omitempty"` + DebugDirectory string `json:"debug_directory,omitempty"` } func newRunResult(resolved pipeline.ResolvedPipeline, output pipeline.RunOutput, outputDirectory, debugDirectory string) (runResult, error) { @@ -59,6 +61,7 @@ func newRunResult(resolved pipeline.ResolvedPipeline, output pipeline.RunOutput, RejectedOutputCount: len(output.Rejected), WarningCount: len(output.Warnings), ValidationStatus: output.Manifest.ValidationStatus, + ValidationSummaries: cloneValidationSummaries(output.Manifest.ValidationSummaries), } if strings.TrimSpace(debugDirectory) != "" { @@ -85,6 +88,17 @@ func newRunResult(resolved pipeline.ResolvedPipeline, output pipeline.RunOutput, return result, nil } +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 encodeRunResult(result runResult) ([]byte, error) { encoded, err := json.Marshal(result) if err != nil { diff --git a/internal/cli/run_result_test.go b/internal/cli/run_result_test.go index 6b289f7a..db0c9515 100644 --- a/internal/cli/run_result_test.go +++ b/internal/cli/run_result_test.go @@ -55,6 +55,9 @@ func TestRunResultEncodesRequiredFieldsAndCounts(t *testing.T) { if got := decoded["warning_count"]; got != float64(1) { t.Fatalf("warning_count = %v", got) } + if got := decoded["validation_summaries"]; got != nil { + t.Fatalf("validation_summaries = %#v, want omitted when empty", got) + } if got := decoded["output_directory"]; got != filepath.Join(mustWorkingDirectory(t), "relative-output") { t.Fatalf("output_directory = %q", got) } @@ -112,6 +115,26 @@ func TestRunResultOmitsIndexFileForOtherOutputModules(t *testing.T) { } } +func TestRunResultProjectsOwnedValidationSummaries(t *testing.T) { + output := testRunOutput() + output.Manifest.ValidationSummaries = []artifacts.ValidationSummary{{Status: "incomplete", IncompleteValidators: []string{"validator"}, ProducerAttemptCount: 1, TerminalAction: "warn_continue"}} + result, err := newRunResult(testResolvedPipeline(pipeline.DefaultOutputModule), output, "output", "") + if err != nil { + t.Fatal(err) + } + output.Manifest.ValidationSummaries[0].IncompleteValidators[0] = "caller mutation" + if got := result.ValidationSummaries[0].IncompleteValidators; len(got) != 1 || got[0] != "validator" { + t.Fatalf("result validation summaries = %#v", result.ValidationSummaries) + } + encoded, err := encodeRunResult(result) + if err != nil { + t.Fatal(err) + } + if !bytes.Contains(encoded, []byte(`"validation_summaries":[{"status":"incomplete","incomplete_validators":["validator"],"producer_attempt_count":1,"terminal_action":"warn_continue"}]`)) { + t.Fatalf("encoded result = %s", encoded) + } +} + func TestRunResultRequiresOneProductionIndexFile(t *testing.T) { tests := []struct { name string diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index f96d0b5a..b7dd6c0c 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -93,17 +93,44 @@ type NormalizedOutputManifest struct { } type RejectedOutputManifest struct { - Stage string `json:"stage"` - StepID string `json:"step_id,omitempty"` - LaneID string `json:"lane_id,omitempty"` - ModuleKey string `json:"module_key,omitempty"` - ChunkID string `json:"chunk_id,omitempty"` - ChunkIndex int `json:"chunk_index,omitempty"` - ValidatorName string `json:"validator_name,omitempty"` - ReasonCode string `json:"reason_code,omitempty"` - Message string `json:"message,omitempty"` - AttemptCount int `json:"attempt_count,omitempty"` - DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` + Stage string `json:"stage"` + StepID string `json:"step_id,omitempty"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + ChunkID string `json:"chunk_id,omitempty"` + ChunkIndex int `json:"chunk_index,omitempty"` + ValidatorName string `json:"validator_name,omitempty"` + ReasonCode string `json:"reason_code,omitempty"` + Message string `json:"message,omitempty"` + AttemptCount int `json:"attempt_count,omitempty"` + DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` + Validation *ValidationSummary `json:"validation,omitempty"` +} + +// ValidationSummary is the bounded, durable outcome of validating one +// producer result. It deliberately contains identities and stable codes, not +// model responses, corrective guidance, validator diagnostics, or payloads. +type ValidationSummary struct { + Stage string `json:"stage,omitempty"` + StepID string `json:"step_id,omitempty"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + ChunkID string `json:"chunk_id,omitempty"` + ChunkIndex int `json:"chunk_index,omitempty"` + Status string `json:"status"` + RejectingValidators []string `json:"rejecting_validators,omitempty"` + ReasonCodes []string `json:"reason_codes,omitempty"` + IncompleteValidators []string `json:"incomplete_validators,omitempty"` + ProducerAttemptCount int `json:"producer_attempt_count"` + TerminalAction string `json:"terminal_action"` +} + +// CloneValidationSummary returns an independently owned durable summary. +func CloneValidationSummary(summary ValidationSummary) ValidationSummary { + summary.RejectingValidators = append([]string(nil), summary.RejectingValidators...) + summary.ReasonCodes = append([]string(nil), summary.ReasonCodes...) + summary.IncompleteValidators = append([]string(nil), summary.IncompleteValidators...) + return summary } type CheckpointDecisionManifest struct { @@ -163,6 +190,7 @@ type RunManifest struct { References []ReferenceProvenance `json:"references,omitempty"` NormalizedOutputs []NormalizedOutputManifest `json:"normalized_outputs,omitempty"` RejectedOutputs []RejectedOutputManifest `json:"rejected_outputs,omitempty"` + ValidationSummaries []ValidationSummary `json:"validation_summaries,omitempty"` CheckpointDecisions []CheckpointDecisionManifest `json:"checkpoint_decisions,omitempty"` LLMProfiles []LLMProfileManifest `json:"llm_profiles,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go index 7d40fe34..7bce66b7 100644 --- a/internal/framework/checkpoint/recorder.go +++ b/internal/framework/checkpoint/recorder.go @@ -10,6 +10,7 @@ import ( "strings" "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" @@ -344,7 +345,15 @@ 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 + if item.Validation != nil { + summary := artifacts.CloneValidationSummary(*item.Validation) + cloned[index].Validation = &summary + } + } + return cloned } func cloneMetadata(metadata map[string]any) map[string]any { diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index ba2d2342..7d57a91c 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -326,17 +326,18 @@ type OutputEncoder interface { } type RejectedOutput struct { - Stage string `json:"stage"` - StepID string `json:"step_id,omitempty"` - LaneID string `json:"lane_id,omitempty"` - ModuleKey string `json:"module_key,omitempty"` - ChunkID string `json:"chunk_id,omitempty"` - ChunkIndex int `json:"chunk_index,omitempty"` - ValidatorName string `json:"validator_name,omitempty"` - ReasonCode string `json:"reason_code,omitempty"` - Message string `json:"message"` - AttemptCount int `json:"attempt_count,omitempty"` - DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` + Stage string `json:"stage"` + StepID string `json:"step_id,omitempty"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + ChunkID string `json:"chunk_id,omitempty"` + ChunkIndex int `json:"chunk_index,omitempty"` + ValidatorName string `json:"validator_name,omitempty"` + ReasonCode string `json:"reason_code,omitempty"` + Message string `json:"message"` + AttemptCount int `json:"attempt_count,omitempty"` + DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` + Validation *artifacts.ValidationSummary `json:"validation,omitempty"` } type ManifestMetadataProvider interface { diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index 7a336266..f5fd1cbc 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -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{ diff --git a/internal/framework/pipeline/producer_attempts.go b/internal/framework/pipeline/producer_attempts.go index 2c484a88..d1550aa3 100644 --- a/internal/framework/pipeline/producer_attempts.go +++ b/internal/framework/pipeline/producer_attempts.go @@ -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)} } diff --git a/internal/framework/pipeline/producer_attempts_test.go b/internal/framework/pipeline/producer_attempts_test.go index b9fb1afe..c623855f 100644 --- a/internal/framework/pipeline/producer_attempts_test.go +++ b/internal/framework/pipeline/producer_attempts_test.go @@ -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 diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 40342171..cf7a3722 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -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 { diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index d7348898..cd9aeb19 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -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" diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 6341d98b..d9c50ab5 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -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 { diff --git a/internal/framework/pipeline/runner_extract_correction_test.go b/internal/framework/pipeline/runner_extract_correction_test.go index 8419f4c0..f3f5904d 100644 --- a/internal/framework/pipeline/runner_extract_correction_test.go +++ b/internal/framework/pipeline/runner_extract_correction_test.go @@ -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) } diff --git a/internal/framework/pipeline/runner_semantic_correction_test.go b/internal/framework/pipeline/runner_semantic_correction_test.go index f46cf38a..4ff5105e 100644 --- a/internal/framework/pipeline/runner_semantic_correction_test.go +++ b/internal/framework/pipeline/runner_semantic_correction_test.go @@ -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) + } } diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 9deaf711..c488b90a 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -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 {