From ba569594a1045c90024284d007e631835033540d Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 27 Aug 2026 15:29:10 +0000 Subject: [PATCH] Transport structured diagnostics through the pipeline --- docs/roadmap/implementation.md | 2 +- internal/framework/contracts/contracts.go | 21 +-- internal/framework/contracts/diagnostics.go | 5 +- .../framework/contracts/diagnostics_test.go | 18 +++ .../framework/contracts/typed_pipeline.go | 10 +- internal/framework/diagnostics/aggregator.go | 132 ++++++++++++++++++ .../framework/diagnostics/aggregator_test.go | 71 ++++++++++ internal/framework/pipeline/diagnostics.go | 107 ++++++++++++++ .../framework/pipeline/diagnostics_test.go | 78 +++++++++++ .../framework/pipeline/extractor_registry.go | 2 +- .../framework/pipeline/merger_registry.go | 2 +- .../framework/pipeline/normalizer_registry.go | 9 +- .../framework/pipeline/producer_attempts.go | 36 +++-- internal/framework/pipeline/runner.go | 29 ++-- .../framework/pipeline/runner_chunk_plan.go | 39 +++--- .../pipeline/runner_concurrency_test.go | 21 ++- .../framework/pipeline/runner_concurrent.go | 18 ++- internal/framework/pipeline/runner_typed.go | 18 ++- .../framework/pipeline/typed_execution.go | 1 + .../framework/pipeline/validation_executor.go | 24 +++- 20 files changed, 578 insertions(+), 65 deletions(-) create mode 100644 internal/framework/diagnostics/aggregator.go create mode 100644 internal/framework/diagnostics/aggregator_test.go create mode 100644 internal/framework/pipeline/diagnostics.go create mode 100644 internal/framework/pipeline/diagnostics_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 2d89bc18..18bd6b9c 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -235,7 +235,7 @@ collector structure. This stage is appropriately sized for one `gpt-5.6-terra` prompt. -## Stage 2 — Add Framework Diagnostic Transport And Transitional Projection +## Stage 2 ✅ — Add Framework Diagnostic Transport And Transitional Projection ### Goal diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 7d57a91c..8f57d144 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -162,9 +162,10 @@ type ChunkRequest struct { } type ChunkPlanResult struct { - Plan source.ChunkPlan `json:"plan"` - Warnings []Warning `json:"warnings,omitempty"` - ModelCandidate *ModelCandidate `json:"-"` + Plan source.ChunkPlan `json:"plan"` + Warnings []Warning `json:"warnings,omitempty"` + Diagnostics []ProducerDiagnostic `json:"diagnostics,omitempty"` + ModelCandidate *ModelCandidate `json:"-"` } type Chunker interface { @@ -283,12 +284,13 @@ const ( ) type ValidationResult struct { - Approved bool `json:"approved"` - ReasonCode string `json:"reason_code,omitempty"` - Message string `json:"message,omitempty"` - CorrectionGuidance string `json:"-"` - DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` - Warnings []Warning `json:"warnings,omitempty"` + Approved bool `json:"approved"` + ReasonCode string `json:"reason_code,omitempty"` + Message string `json:"message,omitempty"` + CorrectionGuidance string `json:"-"` + DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` + Warnings []Warning `json:"warnings,omitempty"` + Diagnostics []ProducerDiagnostic `json:"diagnostics,omitempty"` } type Warning struct { @@ -302,6 +304,7 @@ type OutputRequest struct { NormalizeOutputs []SerializedOutput `json:"normalize_outputs,omitempty"` Rejected []RejectedOutput `json:"rejected,omitempty"` Warnings []Warning `json:"warnings,omitempty"` + Diagnostics DiagnosticCollection `json:"diagnostics,omitempty"` LLMProfile string `json:"llm_profile,omitempty"` StructuredOutputRepairAttempts *int `json:"structured_output_repair_attempts,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` diff --git a/internal/framework/contracts/diagnostics.go b/internal/framework/contracts/diagnostics.go index d65effbf..301f18e6 100644 --- a/internal/framework/contracts/diagnostics.go +++ b/internal/framework/contracts/diagnostics.go @@ -206,8 +206,9 @@ func CloneProducerDiagnostics(diagnostics []ProducerDiagnostic) []ProducerDiagno // CloneDiagnosticCollection returns independent collection ownership. func CloneDiagnosticCollection(collection DiagnosticCollection) DiagnosticCollection { - collection.Groups = make([]DiagnosticGroup, len(collection.Groups)) - for index, group := range collection.Groups { + groups := collection.Groups + collection.Groups = make([]DiagnosticGroup, len(groups)) + for index, group := range groups { collection.Groups[index] = cloneDiagnosticGroup(group) } return collection diff --git a/internal/framework/contracts/diagnostics_test.go b/internal/framework/contracts/diagnostics_test.go index 09177ce0..8bb004b2 100644 --- a/internal/framework/contracts/diagnostics_test.go +++ b/internal/framework/contracts/diagnostics_test.go @@ -128,6 +128,24 @@ func TestDiagnosticGroupRejectsRepeatedSampleWithEqualChunkIndex(t *testing.T) { } } +func TestCloneDiagnosticCollectionOwnsGroupsAndChunkIndex(t *testing.T) { + chunkIndex := 0 + collection := DiagnosticCollection{Groups: []DiagnosticGroup{{ + Disposition: DiagnosticDispositionAdvisory, + Category: DiagnosticCategoryDataQuality, + ReasonCode: "unresolved", + Origin: DiagnosticOrigin{Stage: DiagnosticOriginStageExtract, StepID: "extract", LaneID: "spells", ModuleKey: "dnd/spells"}, + OccurrenceCount: 1, + Samples: []DiagnosticSample{{Scope: "spells[0]", Message: "Spell was not found", ChunkIndex: &chunkIndex}}, + }}} + cloned := CloneDiagnosticCollection(collection) + collection.Groups[0].Samples[0].Message = "changed" + *collection.Groups[0].Samples[0].ChunkIndex = 1 + if got := cloned.Groups[0].Samples[0]; got.Message != "Spell was not found" || got.ChunkIndex == nil || *got.ChunkIndex != 0 { + t.Fatalf("cloned sample = %#v, want independently owned original", got) + } +} + func validProducerDiagnostic() ProducerDiagnostic { return ProducerDiagnostic{ Disposition: DiagnosticDispositionWarning, diff --git a/internal/framework/contracts/typed_pipeline.go b/internal/framework/contracts/typed_pipeline.go index 3438942f..e74596e3 100644 --- a/internal/framework/contracts/typed_pipeline.go +++ b/internal/framework/contracts/typed_pipeline.go @@ -49,6 +49,7 @@ type TypedExtractionRequest struct { type TypedExtractionResult[T any] struct { Value T Warnings []Warning + Diagnostics []ProducerDiagnostic ModelCandidate *ModelCandidate } @@ -74,6 +75,7 @@ type TypedMergeRequest[T any] struct { type TypedMergeResult[T any] struct { Value T Warnings []Warning + Diagnostics []ProducerDiagnostic ModelCandidate *ModelCandidate } @@ -98,6 +100,7 @@ type TypedNormalizeRequest[T any] struct { type TypedNormalizeResult[T any] struct { Value T Warnings []Warning + Diagnostics []ProducerDiagnostic Retry *NormalizeRetry ModelCandidate *ModelCandidate } @@ -112,9 +115,10 @@ const ( // NormalizeRetry asks the framework to retry normalization while retaining a // safe candidate for acceptance if the retry budget is exhausted. type NormalizeRetry struct { - ReasonCode string - Message string - FallbackWarnings []Warning + ReasonCode string + Message string + FallbackWarnings []Warning + FallbackDiagnostics []ProducerDiagnostic } type Normalizer[T any] interface { diff --git a/internal/framework/diagnostics/aggregator.go b/internal/framework/diagnostics/aggregator.go new file mode 100644 index 00000000..ec97973b --- /dev/null +++ b/internal/framework/diagnostics/aggregator.go @@ -0,0 +1,132 @@ +package diagnostics + +import ( + "errors" + "fmt" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +const ( + MaxWarningGroups = 128 + MaxNonWarningGroups = 256 +) + +// Aggregator merges origin-enriched diagnostics in caller-supplied canonical +// order. Its zero value is ready for use. +type Aggregator struct { + groups []contracts.DiagnosticGroup + indices map[groupKey]int + omitted map[groupKey]struct{} + warningGroups int + nonWarningGroups int + unrepresentedOccurrences int +} + +// Add validates and incorporates one final diagnostic group. Actionable +// warnings cannot overflow; later non-warning groups are represented by exact +// unrepresented-occurrence metadata once their fixed bound is reached. +func (aggregator *Aggregator) Add(group contracts.DiagnosticGroup) error { + if err := group.Validate(); err != nil { + return fmt.Errorf("diagnostic group: %w", err) + } + if aggregator.indices == nil { + aggregator.indices = make(map[groupKey]int) + aggregator.omitted = make(map[groupKey]struct{}) + } + key := groupKeyFromGroup(group) + if index, exists := aggregator.indices[key]; exists { + return aggregator.merge(index, group) + } + if _, exists := aggregator.omitted[key]; exists { + return aggregator.addUnrepresented(group.OccurrenceCount) + } + if group.Disposition == contracts.DiagnosticDispositionWarning { + if aggregator.warningGroups >= MaxWarningGroups { + return errors.New("diagnostic warning groups exceed maximum count") + } + aggregator.warningGroups++ + } else if aggregator.nonWarningGroups >= MaxNonWarningGroups { + aggregator.omitted[key] = struct{}{} + return aggregator.addUnrepresented(group.OccurrenceCount) + } else { + aggregator.nonWarningGroups++ + } + aggregator.indices[key] = len(aggregator.groups) + aggregator.groups = append(aggregator.groups, contracts.CloneDiagnosticCollection(contracts.DiagnosticCollection{Groups: []contracts.DiagnosticGroup{group}}).Groups[0]) + return nil +} + +// Collection returns an independently owned grouped result in first-occurrence +// order. +func (aggregator *Aggregator) Collection() contracts.DiagnosticCollection { + if aggregator == nil { + return contracts.DiagnosticCollection{} + } + return contracts.CloneDiagnosticCollection(contracts.DiagnosticCollection{ + Groups: aggregator.groups, + Truncated: aggregator.unrepresentedOccurrences > 0, + UnrepresentedOccurrenceCount: aggregator.unrepresentedOccurrences, + }) +} + +func (aggregator *Aggregator) merge(index int, incoming contracts.DiagnosticGroup) error { + current := &aggregator.groups[index] + if incoming.OccurrenceCount > maximumInt()-current.OccurrenceCount { + return errors.New("diagnostic occurrence count overflow") + } + current.OccurrenceCount += incoming.OccurrenceCount + for _, sample := range incoming.Samples { + if len(current.Samples) == contracts.MaxDiagnosticSamples || containsGroupSample(current.Samples, sample) { + continue + } + current.Samples = append(current.Samples, cloneGroupSample(sample)) + } + current.OmittedSampleCount = current.OccurrenceCount - len(current.Samples) + return nil +} + +func (aggregator *Aggregator) addUnrepresented(count int) error { + if count > maximumInt()-aggregator.unrepresentedOccurrences { + return errors.New("diagnostic unrepresented occurrence count overflow") + } + aggregator.unrepresentedOccurrences += count + return nil +} + +func containsGroupSample(samples []contracts.DiagnosticSample, candidate contracts.DiagnosticSample) bool { + for _, sample := range samples { + if sample.Scope != candidate.Scope || sample.Message != candidate.Message || sample.ChunkID != candidate.ChunkID { + continue + } + if sample.ChunkIndex == nil || candidate.ChunkIndex == nil { + if sample.ChunkIndex == candidate.ChunkIndex { + return true + } + continue + } + if *sample.ChunkIndex == *candidate.ChunkIndex { + return true + } + } + return false +} + +func cloneGroupSample(sample contracts.DiagnosticSample) contracts.DiagnosticSample { + if sample.ChunkIndex != nil { + value := *sample.ChunkIndex + sample.ChunkIndex = &value + } + return sample +} + +type groupKey struct { + disposition contracts.DiagnosticDisposition + category contracts.DiagnosticCategory + reasonCode string + origin contracts.DiagnosticOrigin +} + +func groupKeyFromGroup(group contracts.DiagnosticGroup) groupKey { + return groupKey{disposition: group.Disposition, category: group.Category, reasonCode: group.ReasonCode, origin: group.Origin} +} diff --git a/internal/framework/diagnostics/aggregator_test.go b/internal/framework/diagnostics/aggregator_test.go new file mode 100644 index 00000000..7f7dcc66 --- /dev/null +++ b/internal/framework/diagnostics/aggregator_test.go @@ -0,0 +1,71 @@ +package diagnostics + +import ( + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func TestAggregatorMergesByOriginAndPreservesFirstOccurrenceOrder(t *testing.T) { + aggregator := Aggregator{} + for _, group := range []contracts.DiagnosticGroup{ + groupForAggregation("first", "message one", contracts.DiagnosticDispositionAdvisory, contracts.DiagnosticCategoryDataQuality, "dnd/spells"), + groupForAggregation("second", "message two", contracts.DiagnosticDispositionWarning, contracts.DiagnosticCategoryFallback, "dnd/items"), + groupForAggregation("third", "message three", contracts.DiagnosticDispositionAdvisory, contracts.DiagnosticCategoryDataQuality, "dnd/spells"), + } { + if err := aggregator.Add(group); err != nil { + t.Fatalf("Add() error = %v", err) + } + } + collection := aggregator.Collection() + if len(collection.Groups) != 2 { + t.Fatalf("group count = %d, want 2", len(collection.Groups)) + } + if got := []string{collection.Groups[0].Origin.ModuleKey, collection.Groups[1].Origin.ModuleKey}; !equalStrings(got, []string{"dnd/spells", "dnd/items"}) { + t.Fatalf("group order = %#v, want first occurrence order", got) + } + if collection.Groups[0].OccurrenceCount != 2 || collection.Groups[0].OmittedSampleCount != 0 { + t.Fatalf("merged group = %#v, want two represented occurrences", collection.Groups[0]) + } + if got := []string{collection.Groups[0].Samples[0].Scope, collection.Groups[0].Samples[1].Scope}; !equalStrings(got, []string{"first", "third"}) { + t.Fatalf("merged samples = %#v, want first occurrence order", got) + } +} + +func TestAggregatorEnforcesWarningBoundAndTruncatesOnlyNonWarnings(t *testing.T) { + warnings := Aggregator{} + for index := 0; index < MaxWarningGroups; index++ { + group := groupForAggregation("scope", "message", contracts.DiagnosticDispositionWarning, contracts.DiagnosticCategoryFallback, "module") + group.ReasonCode = "warning-" + string(rune('a'+index)) + if err := warnings.Add(group); err != nil { + t.Fatalf("warning Add(%d) error = %v", index, err) + } + } + if err := warnings.Add(groupForAggregation("overflow", "overflow", contracts.DiagnosticDispositionWarning, contracts.DiagnosticCategoryFallback, "overflow")); err == nil { + t.Fatal("warning overflow error = nil, want error") + } + + nonWarnings := Aggregator{} + for index := 0; index < MaxNonWarningGroups+1; index++ { + group := groupForAggregation("scope", "message", contracts.DiagnosticDispositionAdvisory, contracts.DiagnosticCategoryDataQuality, "module") + group.ReasonCode = "advisory-" + string(rune('a'+index)) + if err := nonWarnings.Add(group); err != nil { + t.Fatalf("non-warning Add(%d) error = %v", index, err) + } + } + collection := nonWarnings.Collection() + if len(collection.Groups) != MaxNonWarningGroups || !collection.Truncated || collection.UnrepresentedOccurrenceCount != 1 { + t.Fatalf("collection = %#v, want bounded non-warning groups and one unrepresented occurrence", collection) + } +} + +func groupForAggregation(scope string, message string, disposition contracts.DiagnosticDisposition, category contracts.DiagnosticCategory, module string) contracts.DiagnosticGroup { + return contracts.DiagnosticGroup{ + Disposition: disposition, + Category: category, + ReasonCode: "source_unrelated", + Origin: contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageExtract, StepID: "extract", LaneID: "spells", ModuleKey: module}, + OccurrenceCount: 1, + Samples: []contracts.DiagnosticSample{{Scope: scope, Message: message}}, + } +} diff --git a/internal/framework/pipeline/diagnostics.go b/internal/framework/pipeline/diagnostics.go new file mode 100644 index 00000000..6140e55d --- /dev/null +++ b/internal/framework/pipeline/diagnostics.go @@ -0,0 +1,107 @@ +package pipeline + +import ( + "fmt" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + frameworkdiagnostics "gitea.maximumdirect.net/eric/notarius/internal/framework/diagnostics" +) + +func terminalDiagnosticGroups(terminal producerAttemptTerminal, origin contracts.DiagnosticOrigin, chunk *source.Chunk) ([]contracts.DiagnosticGroup, error) { + groups, err := promoteProducerDiagnostics(terminal.Diagnostics, origin, chunk) + if err != nil { + return nil, err + } + for _, record := range terminal.Validation.Diagnostics() { + validatorOrigin := origin + validatorOrigin.ValidatorKey = record.validatorName + promoted, err := promoteProducerDiagnostics([]contracts.ProducerDiagnostic{record.diagnostic}, validatorOrigin, chunk) + if err != nil { + return nil, fmt.Errorf("validator %q diagnostic: %w", record.validatorName, err) + } + groups = append(groups, promoted...) + } + return groups, nil +} + +func promoteProducerDiagnostics(diagnostics []contracts.ProducerDiagnostic, origin contracts.DiagnosticOrigin, chunk *source.Chunk) ([]contracts.DiagnosticGroup, error) { + if err := contracts.ValidateProducerDiagnostics(diagnostics); err != nil { + return nil, err + } + if len(diagnostics) == 0 { + return nil, nil + } + groups := make([]contracts.DiagnosticGroup, len(diagnostics)) + for index, diagnostic := range diagnostics { + group := contracts.DiagnosticGroup{ + Disposition: diagnostic.Disposition, + Category: diagnostic.Category, + ReasonCode: diagnostic.ReasonCode, + Origin: origin, + OccurrenceCount: diagnostic.OccurrenceCount, + Samples: cloneDiagnosticSamples(diagnostic.Samples, chunk), + OmittedSampleCount: diagnostic.OmittedSampleCount, + } + if err := group.Validate(); err != nil { + return nil, err + } + groups[index] = group + } + return groups, nil +} + +func cloneDiagnosticSamples(samples []contracts.DiagnosticSample, chunk *source.Chunk) []contracts.DiagnosticSample { + if len(samples) == 0 { + return nil + } + cloned := make([]contracts.DiagnosticSample, len(samples)) + for index, sample := range samples { + if chunk != nil { + chunkIndex := chunk.Index + sample.ChunkID = chunk.ID + sample.ChunkIndex = &chunkIndex + } else if sample.ChunkIndex != nil { + chunkIndex := *sample.ChunkIndex + sample.ChunkIndex = &chunkIndex + } + cloned[index] = sample + } + return cloned +} + +func appendDiagnosticGroups(output *RunOutput, groups []contracts.DiagnosticGroup) { + if output == nil || len(groups) == 0 { + return + } + cloned := contracts.CloneDiagnosticCollection(contracts.DiagnosticCollection{Groups: groups}).Groups + output.diagnosticGroups = append(output.diagnosticGroups, cloned...) + seen := make(map[contracts.Warning]struct{}, len(output.Warnings)) + for _, warning := range output.Warnings { + seen[warning] = struct{}{} + } + for _, group := range cloned { + for _, sample := range group.Samples { + warning := contracts.Warning{Scope: sample.Scope, ReasonCode: group.ReasonCode, Message: sample.Message} + if _, exists := seen[warning]; exists { + continue + } + output.Warnings = append(output.Warnings, warning) + seen[warning] = struct{}{} + } + } +} + +func finalizeDiagnostics(output *RunOutput) error { + if output == nil { + return nil + } + aggregator := frameworkdiagnostics.Aggregator{} + for _, group := range output.diagnosticGroups { + if err := aggregator.Add(group); err != nil { + return err + } + } + output.Diagnostics = aggregator.Collection() + return nil +} diff --git a/internal/framework/pipeline/diagnostics_test.go b/internal/framework/pipeline/diagnostics_test.go new file mode 100644 index 00000000..5762a0e2 --- /dev/null +++ b/internal/framework/pipeline/diagnostics_test.go @@ -0,0 +1,78 @@ +package pipeline + +import ( + "context" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func TestTerminalDiagnosticGroupsDiscardSupersededAttemptDiagnostics(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", Diagnostics: []contracts.ProducerDiagnostic{producerDiagnostic("discarded", "discarded")}, Candidate: attemptCandidate(t, "first")}, nil + } + return producerAttemptOutput{Value: "second", Diagnostics: []contracts.ProducerDiagnostic{producerDiagnostic("terminal", "terminal")}}, nil + }, func(_ context.Context, output producerAttemptOutput) (validationReport, error) { + if output.Value == "first" { + return validationReport{records: []validationRecord{{validatorName: "validator", outcome: validationRejected, reasonCode: "invalid", correctionGuidance: "Correct it."}}}, nil + } + return validationReport{}, nil + }) + if err != nil { + t.Fatalf("runProducerAttempts() error = %v", err) + } + groups, err := terminalDiagnosticGroups(terminal, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageMerge, StepID: "step", LaneID: "lane", ModuleKey: "module"}, nil) + if err != nil { + t.Fatalf("terminalDiagnosticGroups() error = %v", err) + } + if len(groups) != 1 || groups[0].Samples[0].Scope != "terminal" { + t.Fatalf("groups = %#v, want only terminal attempt diagnostics", groups) + } +} + +func TestTerminalDiagnosticGroupsPreserveValidatorOrigin(t *testing.T) { + terminal := producerAttemptTerminal{Validation: validationReport{records: []validationRecord{{ + validatorName: "dnd/spells/source-relatedness", + outcome: validationApproved, + diagnostics: []contracts.ProducerDiagnostic{producerDiagnostic("spells[0]", "not found")}, + }}}} + groups, err := terminalDiagnosticGroups(terminal, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageExtract, StepID: "extract", LaneID: "spells", ModuleKey: "dnd/spells"}, nil) + if err != nil { + t.Fatalf("terminalDiagnosticGroups() error = %v", err) + } + if len(groups) != 1 || groups[0].Origin.ValidatorKey != "dnd/spells/source-relatedness" { + t.Fatalf("groups = %#v, want validator origin", groups) + } +} + +func TestAppendDiagnosticGroupsProjectsOnlyMissingLegacyWarnings(t *testing.T) { + output := RunOutput{Warnings: []contracts.Warning{{Scope: "spells[0]", ReasonCode: "source_unrelated", Message: "not found"}}} + appendDiagnosticGroups(&output, []contracts.DiagnosticGroup{{ + Disposition: contracts.DiagnosticDispositionAdvisory, + Category: contracts.DiagnosticCategoryDataQuality, + ReasonCode: "source_unrelated", + Origin: contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageExtract, StepID: "extract", LaneID: "spells", ModuleKey: "dnd/spells"}, + OccurrenceCount: 2, + Samples: []contracts.DiagnosticSample{{Scope: "spells[0]", Message: "not found"}, {Scope: "spells[1]", Message: "also not found"}}, + }}) + if len(output.Warnings) != 2 { + t.Fatalf("legacy warnings = %#v, want one original and one projected record", output.Warnings) + } + if err := finalizeDiagnostics(&output); err != nil { + t.Fatalf("finalizeDiagnostics() error = %v", err) + } + if len(output.Diagnostics.Groups) != 1 || output.Diagnostics.Groups[0].OccurrenceCount != 2 { + t.Fatalf("diagnostics = %#v, want grouped collection", output.Diagnostics) + } +} + +func producerDiagnostic(scope string, message string) contracts.ProducerDiagnostic { + return contracts.ProducerDiagnostic{ + Disposition: contracts.DiagnosticDispositionAdvisory, + Category: contracts.DiagnosticCategoryDataQuality, + ReasonCode: "source_unrelated", + OccurrenceCount: 1, + Samples: []contracts.DiagnosticSample{{Scope: scope, Message: message}}, + } +} diff --git a/internal/framework/pipeline/extractor_registry.go b/internal/framework/pipeline/extractor_registry.go index 66ddac21..aaa3392a 100644 --- a/internal/framework/pipeline/extractor_registry.go +++ b/internal/framework/pipeline/extractor_registry.go @@ -71,7 +71,7 @@ func RegisterExtractorBuilder[T any](registry *ExtractorRegistry, spec ModuleSpe if err != nil { return erasedTypedResult{}, fmt.Errorf("clone extraction model candidate: %w", err) } - return erasedTypedResult{Value: result.Value, Warnings: cloneWarnings(result.Warnings), ModelCandidate: candidate}, nil + return erasedTypedResult{Value: result.Value, Warnings: cloneWarnings(result.Warnings), Diagnostics: contracts.CloneProducerDiagnostics(result.Diagnostics), ModelCandidate: candidate}, nil }} if registry.typedEntries == nil { registry.typedEntries = map[string]typedExtractorEntry{} diff --git a/internal/framework/pipeline/merger_registry.go b/internal/framework/pipeline/merger_registry.go index 4fc58462..ae20c7d5 100644 --- a/internal/framework/pipeline/merger_registry.go +++ b/internal/framework/pipeline/merger_registry.go @@ -97,7 +97,7 @@ func RegisterMergerBuilder[T any](registry *MergerRegistry, spec ModuleSpec, val if err != nil { return erasedTypedResult{}, fmt.Errorf("clone merge model candidate: %w", err) } - return erasedTypedResult{Value: result.Value, Warnings: cloneWarnings(result.Warnings), ModelCandidate: candidate}, nil + return erasedTypedResult{Value: result.Value, Warnings: cloneWarnings(result.Warnings), Diagnostics: contracts.CloneProducerDiagnostics(result.Diagnostics), ModelCandidate: candidate}, nil }, } return nil diff --git a/internal/framework/pipeline/normalizer_registry.go b/internal/framework/pipeline/normalizer_registry.go index bd4bab2d..28c9d0c6 100644 --- a/internal/framework/pipeline/normalizer_registry.go +++ b/internal/framework/pipeline/normalizer_registry.go @@ -88,7 +88,7 @@ func RegisterNormalizerBuilder[T any](registry *NormalizerRegistry, spec ModuleS if err != nil { return erasedTypedResult{}, fmt.Errorf("clone normalize model candidate: %w", err) } - return erasedTypedResult{Value: result.Value, Warnings: cloneWarnings(result.Warnings), Retry: cloneNormalizeRetry(result.Retry), ModelCandidate: candidate}, nil + return erasedTypedResult{Value: result.Value, Warnings: cloneWarnings(result.Warnings), Diagnostics: contracts.CloneProducerDiagnostics(result.Diagnostics), Retry: cloneNormalizeRetry(result.Retry), ModelCandidate: candidate}, nil }, } return nil @@ -99,9 +99,10 @@ func cloneNormalizeRetry(retry *contracts.NormalizeRetry) *contracts.NormalizeRe return nil } return &contracts.NormalizeRetry{ - ReasonCode: retry.ReasonCode, - Message: retry.Message, - FallbackWarnings: cloneWarnings(retry.FallbackWarnings), + ReasonCode: retry.ReasonCode, + Message: retry.Message, + FallbackWarnings: cloneWarnings(retry.FallbackWarnings), + FallbackDiagnostics: contracts.CloneProducerDiagnostics(retry.FallbackDiagnostics), } } diff --git a/internal/framework/pipeline/producer_attempts.go b/internal/framework/pipeline/producer_attempts.go index e6e61953..71fce4c5 100644 --- a/internal/framework/pipeline/producer_attempts.go +++ b/internal/framework/pipeline/producer_attempts.go @@ -52,24 +52,26 @@ type producerAttemptRequest struct { // the current value as a safe fallback if its shared budget is exhausted. // Artifact-specific adapters are responsible for validating and populating it. type producerRetryDirective struct { - FallbackWarnings []contracts.Warning + FallbackWarnings []contracts.Warning + FallbackDiagnostics []contracts.ProducerDiagnostic } func (directive *producerRetryDirective) clone() *producerRetryDirective { if directive == nil { return nil } - return &producerRetryDirective{FallbackWarnings: cloneWarnings(directive.FallbackWarnings)} + return &producerRetryDirective{FallbackWarnings: cloneWarnings(directive.FallbackWarnings), FallbackDiagnostics: contracts.CloneProducerDiagnostics(directive.FallbackDiagnostics)} } // producerAttemptOutput is intentionally artifact-neutral. Value remains // opaque to the state machine; Candidate is the attempt-local response that // may support semantic correction. type producerAttemptOutput struct { - Value any - Candidate *contracts.ModelCandidate - Warnings []contracts.Warning - Retry *producerRetryDirective + Value any + Candidate *contracts.ModelCandidate + Warnings []contracts.Warning + Diagnostics []contracts.ProducerDiagnostic + Retry *producerRetryDirective } func (output producerAttemptOutput) clone() (producerAttemptOutput, error) { @@ -78,10 +80,11 @@ func (output producerAttemptOutput) clone() (producerAttemptOutput, error) { return producerAttemptOutput{}, fmt.Errorf("clone model candidate: %w", err) } return producerAttemptOutput{ - Value: output.Value, - Candidate: candidate, - Warnings: cloneWarnings(output.Warnings), - Retry: output.Retry.clone(), + Value: output.Value, + Candidate: candidate, + Warnings: cloneWarnings(output.Warnings), + Diagnostics: contracts.CloneProducerDiagnostics(output.Diagnostics), + Retry: output.Retry.clone(), }, nil } @@ -114,6 +117,7 @@ type producerAttemptTerminal struct { Action producerTerminalAction Value any Warnings []contracts.Warning + Diagnostics []contracts.ProducerDiagnostic Rejection *contracts.RejectedOutput Validation validationReport ValidationIncomplete bool @@ -122,6 +126,7 @@ type producerAttemptTerminal struct { func (terminal producerAttemptTerminal) clone() producerAttemptTerminal { terminal.Warnings = cloneWarnings(terminal.Warnings) + terminal.Diagnostics = contracts.CloneProducerDiagnostics(terminal.Diagnostics) if terminal.Rejection != nil { rejection := *terminal.Rejection rejection.Validation = cloneValidationSummaryPtr(rejection.Validation) @@ -206,6 +211,7 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod } if output.Retry != nil { output.Warnings = append(output.Warnings, cloneWarnings(output.Retry.FallbackWarnings)...) + output.Diagnostics = append(output.Diagnostics, contracts.CloneProducerDiagnostics(output.Retry.FallbackDiagnostics)...) } correctionCandidate, err := contracts.CloneModelCandidate(output.Candidate) if err != nil { @@ -255,13 +261,13 @@ func runProducerAttempts(ctx context.Context, config producerAttemptConfig, prod if config.Policy.ValidatorFailure == ValidatorFailureWarnContinue { 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 producerAttemptTerminal{Action: producerTerminalIncompleteAccepted, Value: output.Value, Warnings: warnings, Diagnostics: cloneProducerDiagnostics(output.Diagnostics), Validation: report, ValidationIncomplete: true, Provenance: cloneProducerAttemptProvenance(provenance)}, nil } return failedProducerAttempt(provenance), validatorFailureError(*incomplete) } provenance = append(provenance, producerAttemptProvenance{Number: number, Kind: kind, Outcome: producerAttemptAccepted, Validation: report}) - return producerAttemptTerminal{Action: producerTerminalAccepted, Value: output.Value, Warnings: terminalWarnings(output, report), Validation: report, Provenance: cloneProducerAttemptProvenance(provenance)}, nil + return producerAttemptTerminal{Action: producerTerminalAccepted, Value: output.Value, Warnings: terminalWarnings(output, report), Diagnostics: cloneProducerDiagnostics(output.Diagnostics), Validation: report, Provenance: cloneProducerAttemptProvenance(provenance)}, nil } return failedProducerAttempt(provenance), errors.New("producer attempt budget was not exhausted deterministically") @@ -290,7 +296,7 @@ func applySemanticTerminalPolicy(policy ValidationPolicy, provenance []producerA rejected := contracts.RejectedOutput{ValidatorName: rejection.validatorName, ReasonCode: rejection.reasonCode, Message: rejection.message, AttemptCount: number, DiagnosticArtifactPath: rejection.diagnosticPath} switch policy.SemanticRejection { case SemanticRejectionRejectOutput: - return producerAttemptTerminal{Action: producerTerminalRejected, Warnings: terminalWarnings(output, report), Rejection: &rejected, Validation: report, ValidationIncomplete: firstIncompleteValidation(report) != nil, Provenance: cloneProducerAttemptProvenance(provenance)}, nil + return producerAttemptTerminal{Action: producerTerminalRejected, Warnings: terminalWarnings(output, report), Diagnostics: cloneProducerDiagnostics(output.Diagnostics), Rejection: &rejected, Validation: report, ValidationIncomplete: firstIncompleteValidation(report) != nil, Provenance: cloneProducerAttemptProvenance(provenance)}, nil case SemanticRejectionFailRun: return failedProducerAttempt(provenance), fmt.Errorf("producer candidate rejected after %d attempt(s): %s", number, rejection.message) default: @@ -314,6 +320,10 @@ func terminalWarnings(output producerAttemptOutput, report validationReport) []c return warnings } +func cloneProducerDiagnostics(diagnostics []contracts.ProducerDiagnostic) []contracts.ProducerDiagnostic { + return contracts.CloneProducerDiagnostics(diagnostics) +} + // 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. diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 7d281a04..4766b197 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -51,6 +51,7 @@ type RunInput struct { LLMProfiles []artifacts.LLMProfileManifest Metadata map[string]any Warnings []contracts.Warning + Diagnostics []contracts.ProducerDiagnostic ChunkCacheMode ChunkCacheMode ChunkPlans ChunkPlanStore Checkpoints CheckpointRecorder @@ -69,16 +70,18 @@ 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"` - ValidationSummaries []artifacts.ValidationSummary `json:"validation_summaries,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"` + Diagnostics contracts.DiagnosticCollection `json:"diagnostics,omitempty"` + OutputFiles []contracts.OutputFile `json:"-"` + CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"` + ValidationSummaries []artifacts.ValidationSummary `json:"validation_summaries,omitempty"` normalizeReuseEligibility map[generatedOutputKey]bool + diagnosticGroups []contracts.DiagnosticGroup } func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err error) { @@ -130,6 +133,11 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.llmClient)) }() output.Warnings = append(output.Warnings, cloneWarnings(input.Warnings)...) + inputDiagnostics, diagnosticErr := promoteProducerDiagnostics(input.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageReferences}, nil) + if diagnosticErr != nil { + return failOutput(output), fmt.Errorf("promote input diagnostics: %w", diagnosticErr) + } + appendDiagnosticGroups(&output, inputDiagnostics) if err := writeDebugTimed(debugRecorder, "run.json", debugTimedEnvelope{ Stage: "run", StartedAt: startedTime(input.StartedAt), @@ -249,6 +257,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(*chunkResult.validation)) } output.Warnings = append(output.Warnings, chunkResult.warnings...) + appendDiagnosticGroups(&output, chunkResult.diagnostics) chunkDebugPayload := map[string]any{ "cache_mode": chunkMode, "lookup": chunkResult.lookup, @@ -311,6 +320,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } populateOutputManifest(&output) output.Manifest.CompletedAt = timePtr(time.Now().UTC()) + if err := finalizeDiagnostics(&output); err != nil { + return failOutput(output), fmt.Errorf("aggregate diagnostics: %w", err) + } encoder := input.Prepared.output if err := attachModuleManifestMetadata(&output, "output", encoder); err != nil { @@ -374,6 +386,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err NormalizeOutputs: cloneSerializedOutputs(output.NormalizeOutputs), Rejected: cloneRejectedOutputs(output.Rejected), Warnings: output.Warnings, + Diagnostics: contracts.CloneDiagnosticCollection(output.Diagnostics), LLMProfile: input.pipeline.Output.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(input.pipeline.Output.StructuredOutputRepairAttempts), Metadata: outputMetadata, diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index cca20478..ba660d64 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -12,24 +12,26 @@ 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 - validation *artifacts.ValidationSummary + accepted bool + chunks []source.Chunk + plan *source.ChunkPlan + warnings []contracts.Warning + diagnostics []contracts.DiagnosticGroup + rejection *contracts.RejectedOutput + lookup ChunkPlanDecision + record *ChunkPlanRecord + action string + summary artifacts.ChunkPlanSummary + validation *artifacts.ValidationSummary } type generatedChunkPlanCandidate struct { - plan source.ChunkPlan - chunks []source.Chunk - record ChunkPlanRecord - producerWarnings []contracts.Warning - terminal *attemptTerminalRecorder + plan source.ChunkPlan + chunks []source.Chunk + record ChunkPlanRecord + producerWarnings []contracts.Warning + producerDiagnostics []contracts.ProducerDiagnostic + terminal *attemptTerminalRecorder } func effectiveChunkCacheMode(mode ChunkCacheMode) ChunkCacheMode { @@ -163,7 +165,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S }, Warnings: cloneWarnings(chunkResult.Warnings), CreatedAt: time.Now().UTC(), } - return producerAttemptOutput{Value: generatedChunkPlanCandidate{plan: plan, chunks: chunks, record: candidate, producerWarnings: cloneWarnings(chunkResult.Warnings), terminal: &attemptTerminal}, Candidate: chunkResult.ModelCandidate, Warnings: cloneWarnings(chunkResult.Warnings)}, nil + return producerAttemptOutput{Value: generatedChunkPlanCandidate{plan: plan, chunks: chunks, record: candidate, producerWarnings: cloneWarnings(chunkResult.Warnings), producerDiagnostics: contracts.CloneProducerDiagnostics(chunkResult.Diagnostics), terminal: &attemptTerminal}, Candidate: chunkResult.ModelCandidate, Warnings: cloneWarnings(chunkResult.Warnings), Diagnostics: contracts.CloneProducerDiagnostics(chunkResult.Diagnostics)}, nil }, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) { candidate, ok := output.Value.(generatedChunkPlanCandidate) if !ok { @@ -190,6 +192,11 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S } return result, err } + diagnostics, diagnosticErr := terminalDiagnosticGroups(terminal, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageChunk, ModuleKey: chunker.Key()}, nil) + if diagnosticErr != nil { + return result, fmt.Errorf("promote chunk diagnostics: %w", diagnosticErr) + } + result.diagnostics = diagnostics if terminal.Action == producerTerminalRejected { result.rejection = terminal.Rejection if result.rejection != nil { diff --git a/internal/framework/pipeline/runner_concurrency_test.go b/internal/framework/pipeline/runner_concurrency_test.go index 0f3aa247..853be2ca 100644 --- a/internal/framework/pipeline/runner_concurrency_test.go +++ b/internal/framework/pipeline/runner_concurrency_test.go @@ -367,7 +367,8 @@ func TestRunnerBoundsExtractJobsAndStabilizesReverseCompletion(t *testing.T) { case <-ctx.Done(): return erasedTypedResult{}, ctx.Err() } - return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index), Warnings: []contracts.Warning{{Scope: fmt.Sprintf("lane-%d/chunk-%d", lane, request.Chunk.Index), ReasonCode: "observed", Message: "ordered"}}}, nil + scope := fmt.Sprintf("lane-%d/chunk-%d", lane, request.Chunk.Index) + return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index), Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: "ordered"}}, Diagnostics: []contracts.ProducerDiagnostic{{Disposition: contracts.DiagnosticDispositionObservation, Category: contracts.DiagnosticCategoryNormalization, ReasonCode: "observed", OccurrenceCount: 1, Samples: []contracts.DiagnosticSample{{Scope: scope, Message: "ordered"}}}}}, nil }) } @@ -404,6 +405,24 @@ func TestRunnerBoundsExtractJobsAndStabilizesReverseCompletion(t *testing.T) { if !reflect.DeepEqual(gotScopes, wantScopes) { t.Fatalf("warning order = %#v, want %#v", gotScopes, wantScopes) } + if len(result.output.Diagnostics.Groups) != 2 { + t.Fatalf("diagnostic groups = %#v, want one deterministic group per lane", result.output.Diagnostics.Groups) + } + for lane, group := range result.output.Diagnostics.Groups { + if group.Origin.LaneID != prepared.Steps[0].lanes[lane].resolved.ID { + t.Fatalf("group origin = %#v, want configured lane %q", group.Origin, prepared.Steps[0].lanes[lane].resolved.ID) + } + indexes := make([]int, len(group.Samples)) + for index, sample := range group.Samples { + if sample.ChunkIndex == nil { + t.Fatalf("sample = %#v, want chunk index", sample) + } + indexes[index] = *sample.ChunkIndex + } + if !reflect.DeepEqual(indexes, []int{0, 1, 2}) { + t.Fatalf("group sample chunk order = %#v, want canonical chunk order", indexes) + } + } } func TestRunnerStartsLaneContinuationWhileOtherLaneExtractsRemain(t *testing.T) { diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 1eae3610..64822dbb 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -25,6 +25,7 @@ type laneExtractState struct { values []erasedExtractArtifact serialized []CheckpointArtifact warnings []contracts.Warning + diagnostics []contracts.DiagnosticGroup rejected []contracts.RejectedOutput incomplete []int validationSummaries []artifacts.ValidationSummary @@ -39,6 +40,7 @@ type finalizedExtractResults struct { accepted []erasedExtractArtifact serialized []CheckpointArtifact warnings []contracts.Warning + diagnostics []contracts.DiagnosticGroup rejected []contracts.RejectedOutput incomplete []int validationSummaries []artifacts.ValidationSummary @@ -68,6 +70,7 @@ type extractJobResult struct { value erasedExtractArtifact serialized CheckpointArtifact warnings []contracts.Warning + diagnostics []contracts.DiagnosticGroup rejected *contracts.RejectedOutput validationIncomplete bool validationSummary *artifacts.ValidationSummary @@ -458,7 +461,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. return producerAttemptOutput{}, terminal.record(payload, attemptErr) } serializedCandidate.ChunkID, serializedCandidate.ChunkIndex, serializedCandidate.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef - return producerAttemptOutput{Value: extractAttemptValue{artifact: artifact, serialized: serializedCandidate, terminal: &terminal}, Candidate: extracted.ModelCandidate, Warnings: attemptWarnings}, nil + return producerAttemptOutput{Value: extractAttemptValue{artifact: artifact, serialized: serializedCandidate, terminal: &terminal}, Candidate: extracted.ModelCandidate, Warnings: attemptWarnings, Diagnostics: contracts.CloneProducerDiagnostics(extracted.Diagnostics)}, nil }, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) { candidate, ok := output.Value.(extractAttemptValue) if !ok { @@ -481,6 +484,14 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. result.err = errors.Join(result.err, debugErr) return result } + if err == nil { + diagnostics, diagnosticErr := terminalDiagnosticGroups(terminalResult, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageExtract, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module}, &chunk) + if diagnosticErr != nil { + result.err = fmt.Errorf("promote extract diagnostics: %w", diagnosticErr) + return result + } + result.diagnostics = diagnostics + } if err == nil && terminalResult.Action == producerTerminalRejected { result.rejected = terminalResult.Rejection if result.rejected != nil { @@ -532,11 +543,13 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, stepID string, state *l if result.rejected != nil { state.rejected = append(state.rejected, *result.rejected) state.warnings = append(state.warnings, result.warnings...) + state.diagnostics = append(state.diagnostics, result.diagnostics...) continue } state.values = append(state.values, result.value) state.serialized = append(state.serialized, result.serialized) state.warnings = append(state.warnings, result.warnings...) + state.diagnostics = append(state.diagnostics, result.diagnostics...) if result.validationIncomplete { state.incomplete = append(state.incomplete, result.chunkIndex) } @@ -563,6 +576,7 @@ func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints C accepted: state.values, serialized: state.serialized, warnings: state.warnings, + diagnostics: state.diagnostics, rejected: state.rejected, incomplete: state.incomplete, validationSummaries: state.validationSummaries, @@ -570,6 +584,7 @@ func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints C reuseEligible: state.reuseEligible, } local.Warnings = append(local.Warnings, cloneWarnings(results.warnings)...) + local.diagnosticGroups = append(local.diagnosticGroups, contracts.CloneDiagnosticCollection(contracts.DiagnosticCollection{Groups: results.diagnostics}).Groups...) 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 { @@ -641,6 +656,7 @@ func mergeLaneOutput(dst *RunOutput, src RunOutput) error { dst.NormalizeOutputs = append(dst.NormalizeOutputs, cloneSerializedOutputs(src.NormalizeOutputs)...) dst.Rejected = append(dst.Rejected, cloneRejectedOutputs(src.Rejected)...) dst.Warnings = append(dst.Warnings, cloneWarnings(src.Warnings)...) + appendDiagnosticGroups(dst, src.diagnosticGroups) dst.CheckpointEvents = append(dst.CheckpointEvents, src.CheckpointEvents...) dst.ValidationSummaries = append(dst.ValidationSummaries, cloneValidationSummaries(src.ValidationSummaries)...) if len(src.normalizeReuseEligibility) > 0 { diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 376a7693..63111957 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -295,7 +295,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints if encodeErr != nil { return producerAttemptOutput{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(warnings)}, fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr)) } - return producerAttemptOutput{Value: mergeAttemptValue{artifact: candidate, candidate: serializedCandidate, terminal: &terminal}, Candidate: result.ModelCandidate, Warnings: warnings}, nil + return producerAttemptOutput{Value: mergeAttemptValue{artifact: candidate, candidate: serializedCandidate, terminal: &terminal}, Candidate: result.ModelCandidate, Warnings: warnings, Diagnostics: contracts.CloneProducerDiagnostics(result.Diagnostics)}, nil }, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) { candidate, ok := output.Value.(mergeAttemptValue) if !ok { @@ -322,6 +322,10 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints } return stageResult, runErr } + terminalDiagnostics, diagnosticErr := terminalDiagnosticGroups(terminalResult, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageMerge, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module}, nil) + if diagnosticErr != nil { + return stageResult, fmt.Errorf("promote merge diagnostics: %w", diagnosticErr) + } if terminalResult.Action == producerTerminalRejected { rejected := terminalResult.Rejection if rejected == nil { @@ -331,6 +335,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints rejected.Validation = &terminalSummary output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(terminalSummary)) output.Warnings = append(output.Warnings, terminalResult.Warnings...) + appendDiagnosticGroups(output, terminalDiagnostics) output.Rejected = append(output.Rejected, *rejected) if stageResult.reuseEligible { if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejected); err != nil { @@ -367,6 +372,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints stageResult.reuseEligible = false } output.Warnings = append(output.Warnings, mergeWarnings...) + appendDiagnosticGroups(output, terminalDiagnostics) if stageResult.reuseEligible { if err := recordMerge(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil { return stageResult, err @@ -446,7 +452,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi } anotherAttempt := request.Number <= lane.Normalize.Retries attemptValue.retry = map[string]any{"reason_code": result.Retry.ReasonCode, "message": result.Retry.Message, "another_attempt": anotherAttempt, "fallback_accepted": !anotherAttempt} - directive = &producerRetryDirective{FallbackWarnings: cloneWarnings(result.Retry.FallbackWarnings)} + directive = &producerRetryDirective{FallbackWarnings: cloneWarnings(result.Retry.FallbackWarnings), FallbackDiagnostics: contracts.CloneProducerDiagnostics(result.Retry.FallbackDiagnostics)} if anotherAttempt { payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(warnings), "retry": attemptValue.retry} if debugErr := terminal.record(payload, nil); debugErr != nil { @@ -454,7 +460,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi } } } - return producerAttemptOutput{Value: attemptValue, Candidate: result.ModelCandidate, Warnings: warnings, Retry: directive}, nil + return producerAttemptOutput{Value: attemptValue, Candidate: result.ModelCandidate, Warnings: warnings, Diagnostics: contracts.CloneProducerDiagnostics(result.Diagnostics), Retry: directive}, nil }, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) { candidate, ok := output.Value.(normalizeAttemptValue) if !ok { @@ -484,6 +490,10 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi } return stageResult, runErr } + terminalDiagnostics, diagnosticErr := terminalDiagnosticGroups(terminalResult, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageNormalize, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module}, nil) + if diagnosticErr != nil { + return stageResult, fmt.Errorf("promote normalize diagnostics: %w", diagnosticErr) + } if terminalResult.Action == producerTerminalRejected { rejected := terminalResult.Rejection if rejected == nil { @@ -493,6 +503,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi rejected.Validation = &terminalSummary output.ValidationSummaries = append(output.ValidationSummaries, artifacts.CloneValidationSummary(terminalSummary)) output.Warnings = append(output.Warnings, terminalResult.Warnings...) + appendDiagnosticGroups(output, terminalDiagnostics) output.Rejected = append(output.Rejected, *rejected) if stageResult.reuseEligible { if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejected); err != nil { @@ -527,6 +538,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi normalizeWarnings = cloneWarnings(terminalResult.Warnings) output.ValidationSummaries = append(output.ValidationSummaries, terminalSummary) output.Warnings = append(output.Warnings, normalizeWarnings...) + appendDiagnosticGroups(output, terminalDiagnostics) if terminalResult.ValidationIncomplete { stageResult.reuseEligible = false } diff --git a/internal/framework/pipeline/typed_execution.go b/internal/framework/pipeline/typed_execution.go index 02835373..58766eae 100644 --- a/internal/framework/pipeline/typed_execution.go +++ b/internal/framework/pipeline/typed_execution.go @@ -24,6 +24,7 @@ type erasedMergeArtifact struct { type erasedTypedResult struct { Value any Warnings []contracts.Warning + Diagnostics []contracts.ProducerDiagnostic Retry *contracts.NormalizeRetry ModelCandidate *contracts.ModelCandidate } diff --git a/internal/framework/pipeline/validation_executor.go b/internal/framework/pipeline/validation_executor.go index 0b9ad842..c2a55b5b 100644 --- a/internal/framework/pipeline/validation_executor.go +++ b/internal/framework/pipeline/validation_executor.go @@ -30,12 +30,14 @@ type validationRecord struct { message string diagnosticPath string warnings []contracts.Warning + diagnostics []contracts.ProducerDiagnostic correctionGuidance string failure error } func (record validationRecord) clone() validationRecord { record.warnings = cloneWarnings(record.warnings) + record.diagnostics = contracts.CloneProducerDiagnostics(record.diagnostics) return record } @@ -62,6 +64,24 @@ func (report validationReport) Warnings() []contracts.Warning { return warnings } +func (report validationReport) Diagnostics() []validationDiagnosticRecord { + var diagnostics []validationDiagnosticRecord + for _, record := range report.records { + if record.outcome != validationApproved && record.outcome != validationRejected { + continue + } + for _, diagnostic := range record.diagnostics { + diagnostics = append(diagnostics, validationDiagnosticRecord{validatorName: record.validatorName, diagnostic: cloneProducerDiagnostics([]contracts.ProducerDiagnostic{diagnostic})[0]}) + } + } + return diagnostics +} + +type validationDiagnosticRecord struct { + validatorName string + diagnostic contracts.ProducerDiagnostic +} + func (report validationReport) FirstRejection() *validationRecord { for _, record := range report.records { if record.outcome == validationRejected { @@ -187,14 +207,14 @@ func executeValidationChain(ctx context.Context, chain preparedValidatorChain, i continue } if invocation.result.Approved { - report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationApproved, attemptCount: attempt, warnings: cloneWarnings(invocation.result.Warnings), diagnosticPath: invocation.result.DiagnosticArtifactPath}) + report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationApproved, attemptCount: attempt, warnings: cloneWarnings(invocation.result.Warnings), diagnostics: cloneProducerDiagnostics(invocation.result.Diagnostics), diagnosticPath: invocation.result.DiagnosticArtifactPath}) break } message := invocation.result.Message if message == "" { message = "output rejected" } - report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationRejected, attemptCount: attempt, reasonCode: invocation.result.ReasonCode, message: message, diagnosticPath: invocation.result.DiagnosticArtifactPath, warnings: cloneWarnings(invocation.result.Warnings), correctionGuidance: invocation.result.CorrectionGuidance}) + report.records = append(report.records, validationRecord{validatorName: binding.Module, outcome: validationRejected, attemptCount: attempt, reasonCode: invocation.result.ReasonCode, message: message, diagnosticPath: invocation.result.DiagnosticArtifactPath, warnings: cloneWarnings(invocation.result.Warnings), diagnostics: cloneProducerDiagnostics(invocation.result.Diagnostics), correctionGuidance: invocation.result.CorrectionGuidance}) break } }