diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index e9fd69b..f9cb47c 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -228,10 +228,15 @@ normally. Dependency fingerprints and debug content digests use the same stable codec bytes that cross those boundaries. Debug instrumentation wraps run, stage, attempt, validator, and structured LLM -boundaries. Every executed module retry has an attempt envelope containing its -candidate, accepted-attempt warnings, rejection or error, and only the LLM -calls made by that module attempt. Validator attempts retain independent scopes -under `validate/`. Debug-write failures are framework errors; debug data is +boundaries. Every executed chunk, extract, merge, and normalize attempt writes +one terminal envelope for acceptance, validator rejection, module or validator +error, or applicable candidate or final serialization error. The envelope +contains its attempt-local warnings, any available candidate and rejection, +and terminal error text; failures before a candidate exists omit that payload. +Only LLM calls made by the module operation belong to the module attempt. +Validator calls retain independent scopes under `validate/` and are not +duplicated into the module envelope. A failed terminal-envelope write is a +framework error and is joined with any primary attempt error. Debug data is never used as a checkpoint source. Typed artifact debug envelopes are domain-neutral, redact sensitive metadata and bytes through the common debug policy, and record codec identity plus schema and content digests. diff --git a/docs/operations.md b/docs/operations.md index 9ace8e9..49f3bd9 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -128,15 +128,17 @@ checkpointing does not write debug output. Debug artifacts include inputs and outputs for source, chunk, extract, merge, normalize, and output work, structured LLM request and response data, validator requests and results, timing, and retry attempt metadata. LLM calls made inside -a retry or validator attempt -write `prompt-000N.json`, `response-000N.json`, and -`response-content-000N.*` files under that attempt directory and are linked from -the attempt `llm_calls` array. Prompt content is written inline in the prompt -artifact. The response metadata and body use the paired files described above; -the body is pretty-printed JSON when possible and raw text otherwise. Merge and -normalize retries use these stable paths: +a module retry write `prompt-000N.json`, `response-000N.json`, and +`response-content-000N.*` files under that attempt directory and are linked +from its `llm_calls` array. Validator calls use separate attempt scopes under +`validate/` and are not duplicated into the module attempt. Prompt content is +written inline in the prompt artifact. The response metadata and body use the +paired files described above; the body is pretty-printed JSON when possible +and raw text otherwise. Retrying stages use these stable module-attempt paths: ```text +chunk/attempt-.json +extract//chunk-/attempt-.json merge//attempt-.json merge//attempt-/prompt-.json merge//attempt-/response-.json @@ -148,10 +150,19 @@ normalize//attempt-/response-.json normalize//attempt-/response-content-. ``` -Checkpoint-reused merge and normalize work retains the stage-level input and -output artifacts but has no retry-attempt artifacts because no module attempt -executed. Debug artifacts may contain source material, reference material, -prompt inputs, model outputs, and other sensitive data. Typed artifact +Every executed chunk, extract, merge, and normalize attempt has one terminal +envelope recording acceptance, validator rejection, or a module, validator, +candidate-serialization, or final-serialization error as applicable. It +includes attempt-local warnings and any available candidate or rejection. A +failure before a candidate exists has no candidate payload. If the envelope +cannot be persisted, the run reports that debug failure together with any +primary attempt error. + +Checkpoint-reused chunk, extract, merge, and normalize work retains the +stage-level input and output artifacts but has no retry-attempt artifacts +because no module attempt executed. Debug artifacts may contain source +material, reference material, prompt inputs, model outputs, and other sensitive +data. Typed artifact envelopes include domain-neutral codec identity, redacted metadata and content, and digests of the stable codec bytes. API keys are not written, and obvious credential-shaped values and sensitive map keys are redacted, but debug diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index 16d028b..a53fa3b 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -401,6 +401,31 @@ func writeDebugAttempt(recorder DebugRecorder, attemptPath string, envelope debu return writeDebugTimed(recorder, attemptPath+".json", debugEnvelopeWithLLMCalls(envelope, scope)) } +type attemptTerminalRecorder struct { + recorder DebugRecorder + path string + label string + scope *debugLLMScope + envelope debugTimedEnvelope +} + +func newAttemptTerminalRecorder(recorder DebugRecorder, attemptPath, label string, scope *debugLLMScope, envelope debugTimedEnvelope) attemptTerminalRecorder { + return attemptTerminalRecorder{recorder: recorder, path: attemptPath, label: label, scope: scope, envelope: envelope} +} + +func (r attemptTerminalRecorder) record(payload any, terminalErr error) error { + envelope := r.envelope + envelope.Payload = payload + if terminalErr != nil { + envelope.Error = terminalErr.Error() + } + if err := writeDebugAttempt(r.recorder, r.path, envelope, r.scope); err != nil { + debugErr := fmt.Errorf("write %s attempt debug artifact: %w", r.label, err) + return errors.Join(terminalErr, debugErr) + } + return terminalErr +} + func debugContentEnvelope(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) debugBinaryEnvelope { content = redactSecretBytes(content) return debugBinaryEnvelope{ diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 3794cc0..7c52c23 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "mime" "path" @@ -207,6 +208,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err attemptStarted := time.Now().UTC() attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + terminal := newAttemptTerminalRecorder(debugRecorder, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted}) chunkResult, err := chunker.Chunk(attemptCtx, contracts.ChunkRequest{ Source: doc, SourceInput: sourceInput.Clone(), @@ -216,68 +218,32 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Metadata: input.Metadata, }) if err != nil { - _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ - Stage: string(StageChunk), - ModuleKey: chunker.Key(), - Attempt: attempt, - StartedAt: attemptStarted, - Error: err.Error(), - }, llmScope) - return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err) + attemptErr := fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err) + return false, nil, terminal.record(nil, attemptErr) } if len(chunkResult.Chunks) == 0 { - err := fmt.Errorf("chunker %q returned no chunks", chunker.Key()) - _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ - Stage: string(StageChunk), - ModuleKey: chunker.Key(), - Attempt: attempt, - StartedAt: attemptStarted, - Error: err.Error(), - }, llmScope) - return false, nil, err + attemptErr := fmt.Errorf("chunker %q returned no chunks", chunker.Key()) + return false, nil, terminal.record(nil, attemptErr) } chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks) if err != nil { - err := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) - _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ - Stage: string(StageChunk), - ModuleKey: chunker.Key(), - Attempt: attempt, - StartedAt: attemptStarted, - Payload: map[string]any{ - "warnings": cloneWarnings(chunkResult.Warnings), - }, - Error: err.Error(), - }, llmScope) - return false, nil, err + attemptErr := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) + payload := map[string]any{"warnings": debugWarningEnvelopes(chunkResult.Warnings)} + return false, nil, terminal.record(payload, attemptErr) } validationWarnings, rejection, err := 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...) + payload := map[string]any{ + "chunks": debugSourceChunkEnvelopes(chunks), + "warnings": debugWarningEnvelopes(attemptWarnings), + "rejection": debugRejectedOutputPtr(rejection), + } if err != nil || rejection != nil { - _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ - Stage: string(StageChunk), - ModuleKey: chunker.Key(), - Attempt: attempt, - StartedAt: attemptStarted, - Payload: map[string]any{ - "chunks": debugSourceChunkEnvelopes(chunks), - "warnings": append(cloneWarnings(chunkResult.Warnings), validationWarnings...), - "rejection": debugRejectedOutputPtr(rejection), - }, - }, llmScope) - return false, rejection, err + return false, rejection, terminal.record(payload, err) } canonicalChunks = chunks - chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...) - if err := writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ - Stage: string(StageChunk), - ModuleKey: chunker.Key(), - Attempt: attempt, - StartedAt: attemptStarted, - Payload: map[string]any{ - "chunks": debugSourceChunkEnvelopes(chunks), - "warnings": chunkWarnings, - }, - }, llmScope); err != nil { + chunkWarnings = attemptWarnings + if err := terminal.record(payload, nil); err != nil { return false, nil, err } return true, nil, nil @@ -446,11 +412,15 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, if err != nil { debugCall.Error = err.Error() } - if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { - return nil, nil, debugErr - } if err != nil { - return nil, nil, fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err) + validationErr := fmt.Errorf("validate chunks with validator %q: %w", binding.Module, err) + if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { + return warnings, nil, errors.Join(validationErr, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr)) + } + return warnings, nil, validationErr + } + if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil { + return warnings, nil, fmt.Errorf("write chunk validator attempt debug artifact: %w", debugErr) } if !result.Approved { reason := result.ReasonCode @@ -461,7 +431,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, if message == "" { message = "output rejected" } - return nil, &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil + return warnings, &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: binding.Module, ReasonCode: reason, Message: message, AttemptCount: attempt, DiagnosticArtifactPath: result.DiagnosticArtifactPath}, nil } warnings = append(warnings, result.Warnings...) } diff --git a/internal/framework/pipeline/runner_attempt_debug_test.go b/internal/framework/pipeline/runner_attempt_debug_test.go index 72a5474..1d7b5a2 100644 --- a/internal/framework/pipeline/runner_attempt_debug_test.go +++ b/internal/framework/pipeline/runner_attempt_debug_test.go @@ -230,6 +230,7 @@ func TestRunnerRecordsDistinctRetryAttemptsAndPromotesAcceptedWarningsOnly(t *te laneID := lane.resolved.ID firstPath := fmt.Sprintf("%s/%s/attempt-01.json", stage, laneID) secondPath := fmt.Sprintf("%s/%s/attempt-02.json", stage, laneID) + assertAttemptEnvelopeSequence(t, debug, fmt.Sprintf("%s/%s", stage, laneID), 1, 2) first := debug.envelope(t, firstPath) second := debug.envelope(t, secondPath) if len(first.LLMCalls) != 1 || len(second.LLMCalls) != 1 || first.LLMCalls[0].CallID == second.LLMCalls[0].CallID { diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index b82b31c..a8f0c19 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -268,24 +268,36 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. started := time.Now().UTC() attemptPath := path.Join("extract", debugPathComponent(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), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started}) extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: cloneMetadata(input.Metadata)}) if callErr != nil { - _ = writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope) - return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr) + 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) } artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: extracted.Value} - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value}, state.prepared.extractValidators, attempt, input.Debug) + attemptWarnings := cloneWarnings(extracted.Warnings) + serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value) + 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) + } + serializedCandidate.ChunkID, serializedCandidate.ChunkIndex, serializedCandidate.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef + warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, 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, validateErr + return false, rejected, terminal.record(payload, validateErr) } stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value) if encodeErr != nil { - return false, nil, encodeErr + 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) } stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef accepted, serialized = artifact, stored - acceptedWarnings = append(cloneWarnings(extracted.Warnings), warnings...) - if debugErr := writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope); debugErr != nil { + acceptedWarnings = attemptWarnings + if debugErr := terminal.record(payload, nil); debugErr != nil { return false, nil, debugErr } return true, nil, nil diff --git a/internal/framework/pipeline/runner_extract_handoff_test.go b/internal/framework/pipeline/runner_extract_handoff_test.go index 8705d56..4b90fd3 100644 --- a/internal/framework/pipeline/runner_extract_handoff_test.go +++ b/internal/framework/pipeline/runner_extract_handoff_test.go @@ -142,6 +142,11 @@ func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) { if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" { t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings) } + assertAttemptEnvelopeSequence(t, debug, "extract/notes/chunk-000001", 1, 2) + first := debug.envelope(t, "extract/notes/chunk-000001/attempt-01.json") + if first.Error != "" || !strings.Contains(string(debug.json["extract/notes/chunk-000001/attempt-01.json"]), "rejection") { + t.Fatalf("first extract attempt = %#v, want rejection without error", first) + } name := "extract/notes/chunk-000001/attempt-02.json" if !debug.has(name) { t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names()) diff --git a/internal/framework/pipeline/runner_terminal_debug_test.go b/internal/framework/pipeline/runner_terminal_debug_test.go new file mode 100644 index 0000000..dda34fd --- /dev/null +++ b/internal/framework/pipeline/runner_terminal_debug_test.go @@ -0,0 +1,248 @@ +package pipeline + +import ( + "context" + "errors" + "reflect" + "sort" + "strconv" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type terminalChunker struct { + key string + chunks []source.Chunk + warnings []contracts.Warning + err error +} + +func (c terminalChunker) Key() string { return c.key } + +func (terminalChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil } + +func (c terminalChunker) Chunk(context.Context, contracts.ChunkRequest) (contracts.ChunkResult, error) { + return contracts.ChunkResult{Chunks: cloneSourceChunks(c.chunks), Warnings: cloneWarnings(c.warnings)}, c.err +} + +type terminalChunkValidator struct { + result contracts.ValidationResult + err error +} + +func (terminalChunkValidator) Name() string { return "terminal/chunk-validator" } + +func (terminalChunkValidator) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassDeterministic +} + +func (v terminalChunkValidator) Validate(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + return v.result, v.err +} + +func assertAttemptEnvelopeSequence(t *testing.T, debug *capturedDebugRecorder, prefix string, attempts ...int) { + t.Helper() + marker := strings.TrimSuffix(prefix, "/") + "/attempt-" + var got []int + for _, name := range debug.names() { + if !strings.HasPrefix(name, marker) || !strings.HasSuffix(name, ".json") { + continue + } + remainder := strings.TrimSuffix(strings.TrimPrefix(name, marker), ".json") + if strings.Contains(remainder, "/") { + continue + } + value, err := strconv.Atoi(remainder) + if err != nil { + t.Fatalf("parse attempt index from %q: %v", name, err) + } + got = append(got, value) + } + sort.Ints(got) + if !reflect.DeepEqual(got, attempts) { + t.Fatalf("attempt envelopes under %q = %#v, want %#v; names = %#v", prefix, got, attempts, debug.names()) + } +} + +func preparedTerminalDebugPipeline(t *testing.T) (*PreparedPipeline, []source.Chunk) { + t.Helper() + prepared := preparedAttemptDebugPipeline(t) + chunker, ok := prepared.chunker.(*typedTestChunker) + if !ok { + t.Fatalf("prepared chunker = %T, want *typedTestChunker", prepared.chunker) + } + return prepared, cloneSourceChunks(chunker.chunks) +} + +func TestRunnerRecordsChunkTerminalOutcomes(t *testing.T) { + tests := []struct { + name string + moduleError error + validator terminalChunkValidator + wantError string + wantRejection bool + }{ + {name: "accepted", validator: terminalChunkValidator{result: contracts.ValidationResult{Approved: true}}}, + {name: "module error", moduleError: errors.New("chunk module failed"), wantError: "chunk module failed"}, + {name: "validator rejection", validator: terminalChunkValidator{result: contracts.ValidationResult{Approved: false, ReasonCode: "chunk_rejected", Message: "chunk rejected"}}, wantRejection: true}, + {name: "validator error", validator: terminalChunkValidator{err: errors.New("chunk validator failed")}, wantError: "chunk validator failed"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + prepared, chunks := preparedTerminalDebugPipeline(t) + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, chunks: chunks, warnings: []contracts.Warning{{Scope: "chunk", ReasonCode: "observed", Message: "chunk warning"}}, err: tc.moduleError} + prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding("terminal/chunk-validator"), Target: ValidatorTargetChunk}, chunk: tc.validator}} + debug := newCapturedDebugRecorder() + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + assertAttemptEnvelopeSequence(t, debug, "chunk", 1) + envelope := debug.envelope(t, "chunk/attempt-01.json") + if tc.wantError != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantError) || !strings.Contains(envelope.Error, tc.wantError) { + t.Fatalf("Run() error = %v, attempt error = %q; want %q", err, envelope.Error, tc.wantError) + } + } else if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + if tc.wantRejection { + if envelope.Error != "" || len(output.Rejected) != 1 || output.Rejected[0].ReasonCode != "chunk_rejected" { + t.Fatalf("chunk rejection = envelope %#v, outputs %#v", envelope, output.Rejected) + } + } + }) + } +} + +func TestRunnerRecordsExtractTerminalOutcomes(t *testing.T) { + tests := []struct { + name string + moduleError error + validatorErr error + reject bool + candidateFail bool + finalFail bool + wantError string + }{ + {name: "terminal rejection", reject: true}, + {name: "module error", moduleError: errors.New("extract module failed"), wantError: "extract module failed"}, + {name: "validator error", validatorErr: errors.New("extract validator failed"), wantError: "extract validator failed"}, + {name: "candidate codec error", candidateFail: true, wantError: "serialize extract candidate"}, + {name: "final codec error", finalFail: true, wantError: "serialize accepted extract output"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + value := "extract-terminal" + codec := &observedNotesCodec{} + if tc.candidateFail { + codec.candidateError = value + } + if tc.finalFail { + codec.finalError = value + } + installObservedNotesCodec(t, prepared, codec) + installExtractOperation(prepared, 0, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + if tc.moduleError != nil { + return erasedTypedResult{}, tc.moduleError + } + return erasedTypedResult{Value: codecNotes{Items: []string{value}}, Warnings: []contracts.Warning{{Scope: "extract", ReasonCode: "observed", Message: "extract warning"}}}, nil + }) + prepared.lanes[0].extractValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("terminal/extract-validator"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + return contracts.ValidationResult{Approved: !tc.reject, ReasonCode: "extract_rejected", Message: "extract rejected"}, tc.validatorErr + }, + }} + debug := newCapturedDebugRecorder() + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + assertAttemptEnvelopeSequence(t, debug, "extract/notes/chunk-000001", 1) + attemptPath := "extract/notes/chunk-000001/attempt-01.json" + envelope := debug.envelope(t, attemptPath) + if tc.wantError != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantError) || !strings.Contains(envelope.Error, tc.wantError) { + t.Fatalf("Run() error = %v, attempt error = %q; want %q", err, envelope.Error, tc.wantError) + } + } else { + if err != nil { + t.Fatalf("Run() error = %v, want nil rejection", err) + } + if envelope.Error != "" || len(output.Rejected) != 1 || output.Rejected[0].ReasonCode != "extract_rejected" { + t.Fatalf("extract rejection = envelope %#v, outputs %#v", envelope, output.Rejected) + } + } + if tc.moduleError == nil && !strings.Contains(string(debug.json[attemptPath]), "extract warning") { + t.Fatalf("attempt envelope = %s, want attempt warning", debug.json[attemptPath]) + } + }) + } +} + +func TestRunnerJoinsPrimaryAndAttemptWriteErrors(t *testing.T) { + t.Run("chunk", func(t *testing.T) { + prepared, chunks := preparedTerminalDebugPipeline(t) + primary := errors.New("chunk operation failed") + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, chunks: chunks, err: primary} + debug := newCapturedDebugRecorder() + debug.failPath = "chunk/attempt-01.json" + + _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + if err == nil || !errors.Is(err, primary) || !strings.Contains(err.Error(), "write chunk attempt debug artifact") || !strings.Contains(err.Error(), "debug recorder failure") { + t.Fatalf("Run() error = %v, want joined operation and debug errors", err) + } + }) + + t.Run("extract", func(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + primary := errors.New("extract operation failed") + installExtractOperation(prepared, 0, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{}, primary + }) + debug := newCapturedDebugRecorder() + debug.failPath = "extract/notes/chunk-000001/attempt-01.json" + + _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + if err == nil || !errors.Is(err, primary) || !strings.Contains(err.Error(), "write extract attempt debug artifact") || !strings.Contains(err.Error(), "debug recorder failure") { + t.Fatalf("Run() error = %v, want joined operation and debug errors", err) + } + }) +} + +func TestRunnerKeepsExtractModuleAndValidatorLLMCallsIsolated(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + debug := newCapturedDebugRecorder() + client := WithDebugLLMRecording(attemptDebugLLM{}, debug) + installExtractOperation(prepared, 0, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + if err := callAttemptDebugLLM(ctx, client, "extract-module"); err != nil { + return erasedTypedResult{}, err + } + return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil + }) + prepared.lanes[0].extractValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("llm-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(ctx context.Context, _ any, _ typedValidationTarget) (contracts.ValidationResult, error) { + if err := callAttemptDebugLLM(ctx, client, "extract-validator"); err != nil { + return contracts.ValidationResult{}, err + } + return contracts.ValidationResult{Approved: true}, nil + }, + }} + + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}); err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + module := debug.envelope(t, "extract/notes/chunk-000001/attempt-01.json") + validator := debug.envelope(t, "validate/extract/notes/typed~2fextract-notes/01-llm-check-attempt-01.json") + if len(module.LLMCalls) != 1 || !strings.Contains(module.LLMCalls[0].ResponsePath, "extract/notes/chunk-000001/attempt-01/") { + t.Fatalf("module LLM calls = %#v, want extract module call only", module.LLMCalls) + } + if len(validator.LLMCalls) != 1 || !strings.Contains(validator.LLMCalls[0].ResponsePath, "validate/extract/notes/") { + t.Fatalf("validator LLM calls = %#v, want validator call only", validator.LLMCalls) + } + if module.LLMCalls[0].CallID == validator.LLMCalls[0].CallID { + t.Fatalf("module and validator attempts share LLM call %#v", module.LLMCalls) + } +} diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 4c5bdb3..467a775 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -4,6 +4,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "errors" "fmt" "path" "time" @@ -175,50 +176,32 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi started := time.Now().UTC() attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) - attemptEnvelope := func(payload map[string]any, attemptErr error) error { - envelope := debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started, Payload: payload} - if attemptErr != nil { - envelope.Error = attemptErr.Error() - } - return writeDebugAttempt(input.Debug, attemptPath, envelope, llmScope) - } + terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started}) result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(lane.MergeReferences.ReferenceSet), LLMProfile: lane.Merge.LLMProfile, Metadata: cloneMetadata(input.Metadata)}) if callErr != nil { attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr) - if debugErr := attemptEnvelope(nil, attemptErr); debugErr != nil { - return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr) - } - return false, nil, attemptErr + return false, nil, 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) - if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil { - return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr) - } - return false, nil, attemptErr + return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, 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 { - if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil { - return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr) - } - return false, rejected, validateErr + return false, rejected, terminal.record(payload, validateErr) } 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) - if debugErr := attemptEnvelope(payload, attemptErr); debugErr != nil { - return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr) - } - return false, nil, attemptErr + return false, nil, terminal.record(payload, attemptErr) } - if debugErr := attemptEnvelope(payload, nil); debugErr != nil { - return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr) + if debugErr := terminal.record(payload, nil); debugErr != nil { + return false, nil, debugErr } merged, serializedMerge = candidate, stored mergeWarnings = attemptWarnings @@ -273,49 +256,31 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi started := time.Now().UTC() attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) - attemptEnvelope := func(payload map[string]any, attemptErr error) error { - envelope := debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started, Payload: payload} - if attemptErr != nil { - envelope.Error = attemptErr.Error() - } - return writeDebugAttempt(input.Debug, attemptPath, envelope, llmScope) - } + terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started}) 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(lane.NormalizeReferences.ReferenceSet), LLMProfile: lane.Normalize.LLMProfile, Metadata: cloneMetadata(input.Metadata)}) if callErr != nil { attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr) - if debugErr := attemptEnvelope(nil, attemptErr); debugErr != nil { - return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr) - } - return false, nil, attemptErr + return false, nil, 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) - if debugErr := attemptEnvelope(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr); debugErr != nil { - return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr) - } - return false, nil, attemptErr + return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, 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 { - if debugErr := attemptEnvelope(payload, validateErr); debugErr != nil { - return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr) - } - return false, rejected, validateErr + return false, rejected, terminal.record(payload, validateErr) } 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) - if debugErr := attemptEnvelope(payload, attemptErr); debugErr != nil { - return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr) - } - return false, nil, attemptErr + return false, nil, terminal.record(payload, attemptErr) } - if debugErr := attemptEnvelope(payload, nil); debugErr != nil { - return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr) + if debugErr := terminal.record(payload, nil); debugErr != nil { + return false, nil, debugErr } serializedNormalize = stored normalizeWarnings = attemptWarnings @@ -370,6 +335,13 @@ func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, m func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecEntry, target typedValidationTarget, chain preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) { var warnings []contracts.Warning + if len(chain.validators) > 0 && target.candidate == nil { + candidate, err := validationCandidateArtifact(codec, target) + if err != nil { + return nil, nil, fmt.Errorf("serialize %s candidate for validation: %w", target.stage, err) + } + target.candidate = &candidate + } for index, item := range chain.validators { binding := item.resolved.Binding var result contracts.ValidationResult @@ -396,11 +368,15 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE if err != nil { debugCall.Error = err.Error() } - if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { - return nil, nil, fmt.Errorf("write validation debug artifact: %w", debugErr) - } if err != nil { - return nil, nil, fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err) + validationErr := fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err) + if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { + return warnings, nil, errors.Join(validationErr, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr)) + } + return warnings, nil, validationErr + } + if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil { + return warnings, nil, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr) } if !result.Approved { reason := result.ReasonCode @@ -411,7 +387,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE if message == "" { message = "artifact rejected" } - return nil, &contracts.RejectedOutput{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string { + return warnings, &contracts.RejectedOutput{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string { if target.chunk != nil { return target.chunk.ID }