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

Transport structured diagnostics through the pipeline

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-27 15:29:10 +00:00
parent acb04954eb
commit ba569594a1
20 changed files with 578 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/roadmap/implementation.md
View File

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

21
internal/framework/contracts/contracts.go
View File

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

5
internal/framework/contracts/diagnostics.go
View File

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

18
internal/framework/contracts/diagnostics_test.go
View File

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

10
internal/framework/contracts/typed_pipeline.go
View File

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

132
internal/framework/diagnostics/aggregator.go Normal file
View File

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

71
internal/framework/diagnostics/aggregator_test.go Normal file
View File

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

107
internal/framework/pipeline/diagnostics.go Normal file
View File

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

78
internal/framework/pipeline/diagnostics_test.go Normal file
View File

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

2
internal/framework/pipeline/extractor_registry.go
View File

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

2
internal/framework/pipeline/merger_registry.go
View File

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

9
internal/framework/pipeline/normalizer_registry.go
View File

@@ -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),
}
}

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

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

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

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

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

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

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

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

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

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

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

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

1
internal/framework/pipeline/typed_execution.go
View File

@@ -24,6 +24,7 @@ type erasedMergeArtifact struct {
type erasedTypedResult struct {
Value any
Warnings []contracts.Warning
Diagnostics []contracts.ProducerDiagnostic
Retry *contracts.NormalizeRetry
ModelCandidate *contracts.ModelCandidate
}

24
internal/framework/pipeline/validation_executor.go
View File

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