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

Preserve terminal rejection warnings

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-09 01:19:55 +00:00
parent 14bfae216d
commit 8e669a1f14
7 changed files with 254 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
internal/cli/assembled_spell_pipeline_contract_test.go
View File

@@ -134,7 +134,7 @@ func TestAssembledSpellPipelineHonorsNormalizeValidatorOverride(t *testing.T) {
}
}
func TestAssembledSpellPipelineRejectsUnknownSpellWithoutPromotingAttemptWarning(t *testing.T) {
func TestAssembledSpellPipelinePromotesTerminalUnknownSpellWarning(t *testing.T) {
registries, resolved, _ := assembledSpellPipeline(t, assembledSpellPipelineOptions{unknownSpell: true})
prepared, err := pipeline.Prepare(resolved, registries, pipeline.ModuleDependencies{})
if err != nil {
@@ -161,10 +161,8 @@ func TestAssembledSpellPipelineRejectsUnknownSpellWithoutPromotingAttemptWarning
if !reflect.DeepEqual(rejectedFile.Rejected, output.Rejected) {
t.Fatalf("rejected file = %#v, run rejections = %#v, want durable rejection diagnostic", rejectedFile.Rejected, output.Rejected)
}
for _, warning := range output.Warnings {
if warning.ReasonCode == spellnormalize.ReasonCodeSpellNameUnresolved {
t.Fatalf("warnings = %#v, want rejected-attempt warning to remain non-durable", output.Warnings)
}
if len(output.Warnings) != 1 || output.Warnings[0].ReasonCode != spellnormalize.ReasonCodeSpellNameUnresolved || output.Warnings[0].Scope != "spell_casts[0]" {
t.Fatalf("warnings = %#v, want terminal normalize catalog warning", output.Warnings)
}
}

2
internal/cli/spell_catalog_retry_contract_test.go
View File

@@ -92,7 +92,7 @@ func TestProductionSpellCatalogValidationRetries(t *testing.T) {
t.Fatalf("rejection = %#v, want exhausted unknown-spell rejection", rejection)
}
if len(output.Warnings) != 0 {
t.Fatalf("warnings = %#v, want no warnings from rejected attempts", output.Warnings)
t.Fatalf("warnings = %#v, want no emitted warnings from rejected attempts", output.Warnings)
}
return
}

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

@@ -240,9 +240,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
if chunkResult.rejection != nil {
output.Rejected = append(output.Rejected, *chunkResult.rejection)
}
if chunkResult.accepted || chunkResult.lookup.Status == ChunkPlanHit {
output.Warnings = append(output.Warnings, chunkResult.warnings...)
}
output.Warnings = append(output.Warnings, chunkResult.warnings...)
chunkDebugPayload := map[string]any{
"cache_mode": chunkMode,
"lookup": chunkResult.lookup,
@@ -422,45 +420,52 @@ func (r *Runner) runPreparedSteps(ctx context.Context, input RunInput, checkpoin
return nil
}
func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, *contracts.RejectedOutput, error)) (bool, *contracts.RejectedOutput, error) {
type retryAttemptResult struct {
accepted bool
rejection *contracts.RejectedOutput
warnings []contracts.Warning
}
func runWithRetry(ctx context.Context, retries int, run func(attempt int) (retryAttemptResult, error)) (retryAttemptResult, error) {
attempts := retries + 1
var last *contracts.RejectedOutput
var last retryAttemptResult
for attempt := 1; attempt <= attempts; attempt++ {
if err := ctx.Err(); err != nil {
return false, nil, err
return retryAttemptResult{}, err
}
accepted, rejection, err := run(attempt)
result, err := run(attempt)
if err != nil {
var debugErr *attemptDebugPersistenceError
if errors.As(err, &debugErr) {
return false, nil, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
return retryAttemptResult{}, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
}
if attempt == attempts {
return false, nil, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
return retryAttemptResult{}, fmt.Errorf("failed after %d attempt(s): %w", attempt, err)
}
if ctxErr := ctx.Err(); ctxErr != nil {
return false, nil, ctxErr
return retryAttemptResult{}, ctxErr
}
continue
}
if accepted {
return true, nil, nil
if result.accepted {
return result, nil
}
if rejection != nil {
if result.rejection != nil {
rejection := *result.rejection
rejection.AttemptCount = attempt
last = rejection
last = retryAttemptResult{rejection: &rejection, warnings: cloneWarnings(result.warnings)}
}
if ctxErr := ctx.Err(); ctxErr != nil {
return false, nil, ctxErr
return retryAttemptResult{}, ctxErr
}
if attempt == attempts {
if last == nil {
last = &contracts.RejectedOutput{ReasonCode: "output_rejected", Message: "output rejected", AttemptCount: attempt}
if last.rejection == nil {
last.rejection = &contracts.RejectedOutput{ReasonCode: "output_rejected", Message: "output rejected", AttemptCount: attempt}
}
return false, last, nil
return last, nil
}
}
return false, last, nil
return last, nil
}
func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []source.Chunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, metadata map[string]any, prepared preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) {

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

@@ -84,14 +84,14 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
}
var producerWarnings []contracts.Warning
accepted, rejection, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
retryResult, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (retryAttemptResult, error) {
attemptStarted := time.Now().UTC()
attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr))
}
chunkResult, callErr := chunker.Plan(attemptCtx, contracts.ChunkRequest{
Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID,
@@ -99,21 +99,21 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: requestMetadata,
})
if callErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr))
}
plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, chunkResult.Plan)
if validationErr != nil {
attemptErr := fmt.Errorf("validate chunk plan from chunker %q: %w", chunker.Key(), validationErr)
payload := map[string]any{"plan": debugChunkPlanEnvelope(chunkResult.Plan), "warnings": debugWarningEnvelopes(chunkResult.Warnings)}
return false, nil, terminal.record(payload, attemptErr)
return retryAttemptResult{}, terminal.record(payload, attemptErr)
}
planDigest, digestErr := source.DigestChunkPlan(plan)
if digestErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr))
}
producerMetadata, _, metadataErr := moduleManifestMetadata(chunker)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr))
}
profile := ""
if input.pipeline.ChunkExecutionClass == contracts.ExecutionClassLLMBacked {
@@ -137,7 +137,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
action = "bypassed"
}
if candidateErr := result.setCandidate(candidate, action); candidateErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone generated chunk plan record: %w", candidateErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone generated chunk plan record: %w", candidateErr))
}
validationWarnings, rejected, validationErr := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
@@ -145,9 +145,16 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
"plan": debugChunkPlanEnvelope(plan), "materialized_chunks": debugSourceChunkEnvelopes(chunks),
"warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected),
}
if validationErr != nil || rejected != nil {
if validationErr != nil {
result.setValidation(validationWarnings, rejected, validationErr)
return false, rejected, terminal.record(payload, validationErr)
return retryAttemptResult{}, terminal.record(payload, validationErr)
}
if rejected != nil {
result.setValidation(validationWarnings, rejected, nil)
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
}
return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil
}
result.chunks = chunks
result.plan = &plan
@@ -155,9 +162,9 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
producerWarnings = cloneWarnings(chunkResult.Warnings)
result.setValidation(validationWarnings, nil, nil)
if debugErr := terminal.record(payload, nil); debugErr != nil {
return false, nil, debugErr
return retryAttemptResult{}, debugErr
}
return true, nil, nil
return retryAttemptResult{accepted: true}, nil
})
if err != nil {
if result.summary.ValidationStatus == "not_run" {
@@ -165,9 +172,10 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
}
return result, err
}
result.accepted = accepted
result.rejection = rejection
if !accepted {
result.accepted = retryResult.accepted
result.rejection = retryResult.rejection
if !retryResult.accepted {
result.warnings = cloneWarnings(retryResult.warnings)
return result, nil
}

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

@@ -254,6 +254,9 @@ func (e *laneEngine) startContinuationWorkers() {
}
}
// collect is the sole owner of pending, launch, and completion accounting. It
// must drain closed extract results and one completion for every launched
// continuation, even after cancellation.
func (e *laneEngine) collect() {
resultChannel := (<-chan extractJobResult)(e.results)
for resultChannel != nil || len(e.pending) > 0 || e.completed < e.launched {
@@ -407,20 +410,20 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
var accepted erasedExtractArtifact
var serialized CheckpointArtifact
var acceptedWarnings []contracts.Warning
ok, rejection, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
retryResult, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (retryAttemptResult, error) {
started := time.Now().UTC()
attemptPath := path.Join("extract", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr))
}
extractReferences := operationReferenceSet(input, lane.ExtractReferences)
extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(extractReferences), LLMProfile: lane.Extract.LLMProfile, Metadata: requestMetadata})
if callErr != nil {
attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
return false, nil, terminal.record(nil, attemptErr)
return retryAttemptResult{}, terminal.record(nil, attemptErr)
}
artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: extracted.Value}
attemptWarnings := cloneWarnings(extracted.Warnings)
@@ -428,31 +431,38 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize extract candidate for lane %q chunk %q: %w", lane.ID, chunk.ID, encodeErr)
payload := map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}
return false, nil, terminal.record(payload, attemptErr)
return retryAttemptResult{}, terminal.record(payload, attemptErr)
}
serializedCandidate.ChunkID, serializedCandidate.ChunkIndex, serializedCandidate.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: extractReferences, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value, candidate: &serializedCandidate}, state.prepared.extractValidators, attempt, input.Debug)
attemptWarnings = append(attemptWarnings, warnings...)
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
if validateErr != nil || rejected != nil {
return false, rejected, terminal.record(payload, validateErr)
if validateErr != nil {
return retryAttemptResult{}, terminal.record(payload, validateErr)
}
if rejected != nil {
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
}
return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil
}
stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize accepted extract output for lane %q chunk %q: %w", lane.ID, chunk.ID, encodeErr)
return false, nil, terminal.record(payload, attemptErr)
return retryAttemptResult{}, terminal.record(payload, attemptErr)
}
stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
accepted, serialized = artifact, stored
acceptedWarnings = attemptWarnings
if debugErr := terminal.record(payload, nil); debugErr != nil {
return false, nil, debugErr
return retryAttemptResult{}, debugErr
}
return true, nil, nil
return retryAttemptResult{accepted: true}, nil
})
result.err = err
if err == nil && !ok {
result.rejected = rejection
if err == nil && !retryResult.accepted {
result.rejected = retryResult.rejection
result.warnings = cloneWarnings(retryResult.warnings)
return result
}
result.value, result.serialized, result.warnings = accepted, serialized, acceptedWarnings
@@ -470,6 +480,7 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, stepID string, state *l
result := state.results[index]
if result.rejected != nil {
state.rejected = append(state.rejected, *result.rejected)
state.warnings = append(state.warnings, result.warnings...)
continue
}
state.values = append(state.values, result.value)

142
internal/framework/pipeline/runner_rejection_warnings_test.go Normal file
View File

@@ -0,0 +1,142 @@
package pipeline
import (
"context"
"fmt"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type warningChunker struct {
key string
plan source.ChunkPlan
calls int
}
func (c *warningChunker) Key() string { return c.key }
func (*warningChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil }
func (c *warningChunker) Plan(context.Context, contracts.ChunkRequest) (contracts.ChunkPlanResult, error) {
c.calls++
return contracts.ChunkPlanResult{
Plan: source.CloneChunkPlan(c.plan),
Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", c.calls), ReasonCode: "operation", Message: "operation warning"}},
}, nil
}
type chunkValidationFunc struct {
name string
validate func(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error)
}
func (v chunkValidationFunc) Name() string { return v.name }
func (chunkValidationFunc) ExecutionClass() contracts.ExecutionClass {
return contracts.ExecutionClassDeterministic
}
func (v chunkValidationFunc) Validate(ctx context.Context, request contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
return v.validate(ctx, request)
}
func TestRunnerPromotesOnlyTerminalRejectionWarnings(t *testing.T) {
for _, target := range []ModuleStage{StageChunk, StageExtract, StageMerge, StageNormalize} {
t.Run(string(target), func(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
lane := &prepared.Steps[0].lanes[0]
attempts := 0
first := func() contracts.ValidationResult {
return contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("validator-%d", attempts), ReasonCode: "validator", Message: "validator warning"}}}
}
reject := func() contracts.ValidationResult {
return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "rejected"}
}
debug := newCapturedDebugRecorder()
recorder := &extractCaptureRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
switch target {
case StageChunk:
chunker := prepared.chunker.(*typedTestChunker)
prepared.chunker = &warningChunker{key: prepared.resolved.Chunk.Module, plan: source.CloneChunkPlan(chunker.plan)}
prepared.resolved.Chunk.Retries = 1
prepared.chunkValidators.validators = []preparedValidator{
{resolved: ResolvedValidator{Binding: Binding("warning-approval"), Target: ValidatorTargetChunk}, chunk: chunkValidationFunc{name: "warning-approval", validate: func(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
return first(), nil
}}},
{resolved: ResolvedValidator{Binding: Binding("warning-rejection"), Target: ValidatorTargetChunk}, chunk: chunkValidationFunc{name: "warning-rejection", validate: func(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
return reject(), nil
}}},
}
chunkerWithWarnings := prepared.chunker.(*warningChunker)
first = func() contracts.ValidationResult {
return contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("validator-%d", chunkerWithWarnings.calls), ReasonCode: "validator", Message: "validator warning"}}}
}
case StageExtract:
lane.resolved.Extract.Retries = 1
installExtractOperation(prepared, 0, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
attempts++
return erasedTypedResult{Value: codecNotes{Items: []string{"extract"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil
})
lane.extractValidators.validators = rejectionWarningTypedValidators(first, reject)
case StageMerge:
lane.resolved.Merge.Retries = 1
lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) {
attempts++
return erasedTypedResult{Value: codecNotes{Items: []string{"merge"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil
}
lane.mergeValidators.validators = rejectionWarningTypedValidators(first, reject)
case StageNormalize:
lane.resolved.Normalize.Retries = 1
lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
attempts++
return erasedTypedResult{Value: codecNotes{Items: []string{"normalize"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil
}
lane.normalizeValidators.validators = rejectionWarningTypedValidators(first, reject)
}
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Debug: debug})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
wantScopes := []string{"operation-2", "validator-2"}
if got := rejectionWarningScopes(output.Warnings); !reflect.DeepEqual(got, wantScopes) {
t.Fatalf("published warning scopes = %#v, want %#v", got, wantScopes)
}
if len(output.Rejected) != 1 || output.Rejected[0].AttemptCount != 2 {
t.Fatalf("rejections = %#v, want final rejection after two attempts", output.Rejected)
}
if target == StageExtract && !reflect.DeepEqual(rejectionWarningScopes(recorder.checkpoint.Warnings), wantScopes) {
t.Fatalf("extract checkpoint warnings = %#v, want %#v", recorder.checkpoint.Warnings, wantScopes)
}
attemptPath := fmt.Sprintf("%s/notes/attempt-01.json", target)
if target == StageChunk {
attemptPath = "chunk/attempt-01.json"
} else if target == StageExtract {
attemptPath = "extract/notes/chunk-000001/attempt-01.json"
}
if !strings.Contains(string(debug.json[attemptPath]), "operation-1") || !strings.Contains(string(debug.json[attemptPath]), "validator-1") {
t.Fatalf("first attempt debug = %s, want discarded warnings", debug.json[attemptPath])
}
})
}
}
func rejectionWarningTypedValidators(first func() contracts.ValidationResult, reject func() contracts.ValidationResult) []preparedValidator {
return []preparedValidator{
{resolved: ResolvedValidator{Binding: Binding("warning-approval"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
return first(), nil
}},
{resolved: ResolvedValidator{Binding: Binding("warning-rejection"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) {
return reject(), nil
}},
}
}
func rejectionWarningScopes(warnings []contracts.Warning) []string {
scopes := make([]string, len(warnings))
for index := range warnings {
scopes[index] = warnings[index].Scope
}
return scopes
}

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

@@ -249,52 +249,59 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil {
return stageResult, err
}
ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
retryResult, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (retryAttemptResult, error) {
started := time.Now().UTC()
attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr))
}
result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(mergeReferences), LLMProfile: lane.Merge.LLMProfile, Metadata: requestMetadata})
if callErr != nil {
attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr)
return false, nil, terminal.record(nil, attemptErr)
return retryAttemptResult{}, terminal.record(nil, attemptErr)
}
candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value}
attemptWarnings := cloneWarnings(result.Warnings)
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr)
return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
return retryAttemptResult{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
}
warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: mergeReferences, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug)
attemptWarnings = append(attemptWarnings, warnings...)
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)}
if validateErr != nil || rejected != nil {
return false, rejected, terminal.record(payload, validateErr)
if validateErr != nil {
return retryAttemptResult{}, terminal.record(payload, validateErr)
}
if rejected != nil {
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
}
return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil
}
stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize accepted merge output for lane %q: %w", lane.ID, encodeErr)
return false, nil, terminal.record(payload, attemptErr)
return retryAttemptResult{}, terminal.record(payload, attemptErr)
}
if debugErr := terminal.record(payload, nil); debugErr != nil {
return false, nil, debugErr
return retryAttemptResult{}, debugErr
}
merged, serializedMerge = candidate, stored
mergeWarnings = attemptWarnings
return true, nil, nil
return retryAttemptResult{accepted: true}, nil
})
if runErr != nil {
_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, runErr)
return stageResult, runErr
}
if !ok {
output.Rejected = append(output.Rejected, *rejection)
if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejection); err != nil {
if !retryResult.accepted {
output.Warnings = append(output.Warnings, retryResult.warnings...)
output.Rejected = append(output.Rejected, *retryResult.rejection)
if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *retryResult.rejection); err != nil {
return stageResult, err
}
stageResult.terminal = true
@@ -344,30 +351,30 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil {
return stageResult, err
}
ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
retryResult, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (retryAttemptResult, error) {
started := time.Now().UTC()
attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
if metadataErr != nil {
return false, nil, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr))
return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr))
}
result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(normalizeReferences), LLMProfile: lane.Normalize.LLMProfile, Metadata: requestMetadata})
if callErr != nil {
attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr)
return false, nil, terminal.record(nil, attemptErr)
return retryAttemptResult{}, terminal.record(nil, attemptErr)
}
attemptWarnings := cloneWarnings(result.Warnings)
serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr)
return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
return retryAttemptResult{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr)
}
var retryPayload map[string]any
if result.Retry != nil {
if err := validateNormalizeRetry(result.Retry); err != nil {
return false, nil, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err))
return retryAttemptResult{}, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err))
}
retryRemaining := attempt <= lane.Normalize.Retries
retryPayload = map[string]any{
@@ -378,7 +385,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
}
if retryRemaining {
payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "retry": retryPayload}
return false, nil, terminal.record(payload, nil)
return retryAttemptResult{}, terminal.record(payload, nil)
}
attemptWarnings = append(attemptWarnings, cloneWarnings(result.Retry.FallbackWarnings)...)
}
@@ -388,28 +395,35 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
if retryPayload != nil {
payload["retry"] = retryPayload
}
if validateErr != nil || rejected != nil {
return false, rejected, terminal.record(payload, validateErr)
if validateErr != nil {
return retryAttemptResult{}, terminal.record(payload, validateErr)
}
if rejected != nil {
if debugErr := terminal.record(payload, nil); debugErr != nil {
return retryAttemptResult{}, debugErr
}
return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil
}
stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value)
if encodeErr != nil {
attemptErr := fmt.Errorf("serialize accepted normalize output for lane %q: %w", lane.ID, encodeErr)
return false, nil, terminal.record(payload, attemptErr)
return retryAttemptResult{}, terminal.record(payload, attemptErr)
}
if debugErr := terminal.record(payload, nil); debugErr != nil {
return false, nil, debugErr
return retryAttemptResult{}, debugErr
}
serializedNormalize = stored
normalizeWarnings = attemptWarnings
return true, nil, nil
return retryAttemptResult{accepted: true}, nil
})
if runErr != nil {
_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, runErr)
return stageResult, runErr
}
if !ok {
output.Rejected = append(output.Rejected, *rejection)
if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejection); err != nil {
if !retryResult.accepted {
output.Warnings = append(output.Warnings, retryResult.warnings...)
output.Rejected = append(output.Rejected, *retryResult.rejection)
if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *retryResult.rejection); err != nil {
return stageResult, err
}
return stageResult, nil