From fbc3d9add6965b2561ce6323ac27bfcc0344a7fd Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 17 Jul 2026 13:43:25 +0000 Subject: [PATCH] Add retry-scoped merge and normalize debugging --- docs/internal/pipeline.md | 12 +- docs/operations.md | 26 +- internal/framework/pipeline/debug.go | 13 + internal/framework/pipeline/runner.go | 24 +- .../pipeline/runner_attempt_debug_test.go | 430 ++++++++++++++++++ .../framework/pipeline/runner_concurrent.go | 4 +- internal/framework/pipeline/runner_typed.go | 92 +++- 7 files changed, 558 insertions(+), 43 deletions(-) create mode 100644 internal/framework/pipeline/runner_attempt_debug_test.go diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index b6b5d18..36bda2a 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -223,11 +223,13 @@ 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. Context scopes associate nested LLM calls with the module or -validator attempt that made them. Debug-write failures are framework errors; -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. +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 +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. Checkpoint identity, physical layout, reuse behavior, and debug artifact handling are operator contracts in [Operations](../operations.md). Serialization diff --git a/docs/operations.md b/docs/operations.md index e5340f1..9ace8e9 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -133,11 +133,27 @@ 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. 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 +the body is pretty-printed JSON when possible and raw text otherwise. Merge and +normalize retries use these stable paths: + +```text +merge//attempt-.json +merge//attempt-/prompt-.json +merge//attempt-/response-.json +merge//attempt-/response-content-. + +normalize//attempt-.json +normalize//attempt-/prompt-.json +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 +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 directories should still be protected as sensitive local state. diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index a721cab..16d028b 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -302,6 +302,15 @@ func withDebugLLMScope(ctx context.Context, prefix string) (context.Context, *de return context.WithValue(ctx, debugLLMScopeContextKey{}, scope), scope } +func withIsolatedDebugLLMScope(ctx context.Context, prefix string) (context.Context, *debugLLMScope) { + if ctx == nil { + ctx = context.Background() + } + prefix = cleanDebugPath(prefix) + scope := &debugLLMScope{prefix: prefix} + return context.WithValue(ctx, debugLLMScopeContextKey{}, scope), scope +} + func debugLLMScopeFromContext(ctx context.Context) *debugLLMScope { if ctx == nil { return nil @@ -388,6 +397,10 @@ func debugEnvelopeWithLLMCalls(envelope debugTimedEnvelope, scope *debugLLMScope return envelope } +func writeDebugAttempt(recorder DebugRecorder, attemptPath string, envelope debugTimedEnvelope, scope *debugLLMScope) error { + return writeDebugTimed(recorder, attemptPath+".json", debugEnvelopeWithLLMCalls(envelope, scope)) +} + 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 8af964e..3f9c395 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -217,30 +217,30 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Metadata: input.Metadata, }) if err != nil { - _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted, Error: err.Error(), - }, llmScope)) + }, llmScope) return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err) } if len(chunkResult.Chunks) == 0 { err := fmt.Errorf("chunker %q returned no chunks", chunker.Key()) - _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted, Error: err.Error(), - }, llmScope)) + }, llmScope) return false, nil, err } chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks) if err != nil { err := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) - _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, @@ -249,12 +249,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err "warnings": cloneWarnings(chunkResult.Warnings), }, Error: err.Error(), - }, llmScope)) + }, llmScope) return false, nil, err } validationWarnings, rejection, err := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug) if err != nil || rejection != nil { - _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + _ = writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, @@ -264,12 +264,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err "warnings": append(cloneWarnings(chunkResult.Warnings), validationWarnings...), "rejection": debugRejectedOutputPtr(rejection), }, - }, llmScope)) + }, llmScope) return false, rejection, err } canonicalChunks = chunks chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...) - if err := writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + if err := writeDebugAttempt(debugRecorder, attemptPath, debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, @@ -278,7 +278,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err "chunks": debugSourceChunkEnvelopes(chunks), "warnings": chunkWarnings, }, - }, llmScope)); err != nil { + }, llmScope); err != nil { return false, nil, err } return true, nil, nil @@ -431,7 +431,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, binding := item.resolved.Binding started := time.Now().UTC() attemptPath := path.Join("validate", debugPathComponent(string(StageChunk)), "", debugPathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt)) - validatorCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath) var result contracts.ValidationResult switch item.resolved.Target { case ValidatorTargetChunk: @@ -447,7 +447,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, if err != nil { debugCall.Error = err.Error() } - if debugErr := writeDebugTimed(debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope)); debugErr != nil { + 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 { diff --git a/internal/framework/pipeline/runner_attempt_debug_test.go b/internal/framework/pipeline/runner_attempt_debug_test.go new file mode 100644 index 0000000..72a5474 --- /dev/null +++ b/internal/framework/pipeline/runner_attempt_debug_test.go @@ -0,0 +1,430 @@ +package pipeline + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type capturedDebugRecorder struct { + mu sync.Mutex + json map[string][]byte + bytes map[string][]byte + failPath string +} + +func newCapturedDebugRecorder() *capturedDebugRecorder { + return &capturedDebugRecorder{json: make(map[string][]byte), bytes: make(map[string][]byte)} +} + +func (*capturedDebugRecorder) Enabled() bool { return true } + +func (r *capturedDebugRecorder) WriteJSON(name string, payload any) error { + if name == r.failPath { + return errors.New("debug recorder failure") + } + data, err := json.Marshal(payload) + if err != nil { + return err + } + r.mu.Lock() + r.json[name] = data + r.mu.Unlock() + return nil +} + +func (r *capturedDebugRecorder) WriteBytes(name string, data []byte) error { + if name == r.failPath { + return errors.New("debug recorder failure") + } + r.mu.Lock() + r.bytes[name] = append([]byte(nil), data...) + r.mu.Unlock() + return nil +} + +func (r *capturedDebugRecorder) has(name string) bool { + r.mu.Lock() + defer r.mu.Unlock() + _, jsonOK := r.json[name] + _, bytesOK := r.bytes[name] + return jsonOK || bytesOK +} + +func (r *capturedDebugRecorder) names() []string { + r.mu.Lock() + defer r.mu.Unlock() + names := make([]string, 0, len(r.json)+len(r.bytes)) + for name := range r.json { + names = append(names, name) + } + for name := range r.bytes { + names = append(names, name) + } + sort.Strings(names) + return names +} + +func (r *capturedDebugRecorder) envelope(t *testing.T, name string) debugTimedEnvelope { + t.Helper() + r.mu.Lock() + data := append([]byte(nil), r.json[name]...) + r.mu.Unlock() + if len(data) == 0 { + t.Fatalf("debug envelope %q was not written; names = %#v", name, r.names()) + } + var envelope debugTimedEnvelope + if err := json.Unmarshal(data, &envelope); err != nil { + t.Fatalf("decode debug envelope %q: %v", name, err) + } + return envelope +} + +type attemptDebugLLM struct{} + +func (attemptDebugLLM) CompleteStructured(_ context.Context, request contracts.StructuredCompletionRequest, _ any) (contracts.StructuredCompletionResponse, error) { + return contracts.StructuredCompletionResponse{ + Content: json.RawMessage(`{"accepted":true}`), + Provider: "test", + Model: "test-model", + ProfileID: request.ProfileID, + Debug: &contracts.LLMDebugMaterial{ + Prompt: &contracts.LLMDebugPrompt{PromptID: request.PromptID, Messages: []contracts.LLMDebugMessage{{Role: "user", Content: "test"}}}, + Response: &contracts.LLMDebugResponse{ + Content: `{"accepted":true}`, + PromptID: request.PromptID, + ModelName: "test-model", + }, + }, + }, nil +} + +func preparedAttemptDebugPipeline(t *testing.T) *PreparedPipeline { + t.Helper() + prepared := preparedConcurrentPipeline(t, 1) + prepared.lanes = prepared.lanes[:1] + prepared.resolved.ArtifactLanes = prepared.resolved.ArtifactLanes[:1] + prepared.ArtifactLanes = prepared.ArtifactLanes[:1] + prepared.lanes[0].mergeValidators = preparedValidatorChain{} + prepared.lanes[0].normalizeValidators = preparedValidatorChain{} + return prepared +} + +func callAttemptDebugLLM(ctx context.Context, client contracts.StructuredLLMClient, name string) error { + _, err := client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{StageName: "unscoped-" + name, PromptID: name, ProfileID: "test"}, nil) + return err +} + +func TestRunnerWritesAttemptScopedMergeAndNormalizeDebug(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + debug := newCapturedDebugRecorder() + client := WithDebugLLMRecording(attemptDebugLLM{}, debug) + prepared.lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + if err := callAttemptDebugLLM(ctx, client, "merge"); err != nil { + return erasedTypedResult{}, err + } + return erasedTypedResult{Value: codecNotes{Items: []string{"merged"}}, Warnings: []contracts.Warning{{Scope: "merge", ReasonCode: "observed", Message: "merge warning"}}}, nil + } + prepared.lanes[0].typed.normalize = func(ctx context.Context, _ any, _ contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + if err := callAttemptDebugLLM(ctx, client, "normalize"); err != nil { + return erasedTypedResult{}, err + } + return erasedTypedResult{Value: codecNotes{Items: []string{"normalized"}}, Warnings: []contracts.Warning{{Scope: "normalize", ReasonCode: "observed", Message: "normalize warning"}}}, nil + } + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + if len(output.NormalizeOutputs) != 1 { + t.Fatalf("normalize outputs = %d, want one", len(output.NormalizeOutputs)) + } + + wantPaths := []string{ + "merge/notes/input.json", + "merge/notes/attempt-01.json", + "merge/notes/attempt-01/prompt-0001.json", + "merge/notes/attempt-01/response-0001.json", + "merge/notes/attempt-01/response-content-0001.json", + "merge/notes/output.json", + "normalize/notes/input.json", + "normalize/notes/attempt-01.json", + "normalize/notes/attempt-01/prompt-0002.json", + "normalize/notes/attempt-01/response-0002.json", + "normalize/notes/attempt-01/response-content-0002.json", + "normalize/notes/output.json", + } + for _, name := range wantPaths { + if !debug.has(name) { + t.Errorf("debug artifact %q missing; names = %#v", name, debug.names()) + } + } + for _, name := range debug.names() { + if strings.HasPrefix(name, "unscoped-merge/") || strings.HasPrefix(name, "unscoped-normalize/") { + t.Errorf("LLM call used fallback debug path %q", name) + } + } + mergeEnvelope := debug.envelope(t, "merge/notes/attempt-01.json") + normalizeEnvelope := debug.envelope(t, "normalize/notes/attempt-01.json") + if len(mergeEnvelope.LLMCalls) != 1 || mergeEnvelope.LLMCalls[0].ResponsePath != "merge/notes/attempt-01/response-0001.json" { + t.Fatalf("merge LLM calls = %#v, want attempt-scoped call", mergeEnvelope.LLMCalls) + } + if len(normalizeEnvelope.LLMCalls) != 1 || normalizeEnvelope.LLMCalls[0].ResponsePath != "normalize/notes/attempt-01/response-0002.json" { + t.Fatalf("normalize LLM calls = %#v, want attempt-scoped call", normalizeEnvelope.LLMCalls) + } +} + +func TestRunnerRecordsDistinctRetryAttemptsAndPromotesAcceptedWarningsOnly(t *testing.T) { + for _, stage := range []ModuleStage{StageMerge, StageNormalize} { + t.Run(string(stage), func(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + debug := newCapturedDebugRecorder() + client := WithDebugLLMRecording(attemptDebugLLM{}, debug) + lane := &prepared.lanes[0] + attempts := 0 + operation := func(ctx context.Context) (erasedTypedResult, error) { + attempts++ + if err := callAttemptDebugLLM(ctx, client, string(stage)); err != nil { + return erasedTypedResult{}, err + } + scope := "accepted" + if attempts == 1 { + scope = "discarded" + } + return erasedTypedResult{Value: codecNotes{Items: []string{scope}}, Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: scope}}}, nil + } + validatorCalls := 0 + validator := preparedValidator{ + resolved: ResolvedValidator{Binding: Binding("retry-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + validatorCalls++ + return contracts.ValidationResult{Approved: validatorCalls > 1, ReasonCode: "retry", Message: "retry candidate"}, nil + }, + } + switch stage { + case StageMerge: + lane.resolved.Merge.Retries = 1 + lane.mergeValidators.validators = []preparedValidator{validator} + lane.typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + return operation(ctx) + } + case StageNormalize: + lane.resolved.Normalize.Retries = 1 + lane.normalizeValidators.validators = []preparedValidator{validator} + lane.typed.normalize = func(ctx context.Context, _ any, _ contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + return operation(ctx) + } + } + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + laneID := lane.resolved.ID + firstPath := fmt.Sprintf("%s/%s/attempt-01.json", stage, laneID) + secondPath := fmt.Sprintf("%s/%s/attempt-02.json", stage, laneID) + 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 { + t.Fatalf("retry LLM calls = first %#v, second %#v; want distinct calls", first.LLMCalls, second.LLMCalls) + } + if !strings.Contains(string(debug.json[firstPath]), "discarded") || !strings.Contains(string(debug.json[firstPath]), "rejection") { + t.Fatalf("first attempt envelope = %s, want discarded warning and rejection", debug.json[firstPath]) + } + if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" { + t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings) + } + }) + } +} + +func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) { + tests := []struct { + name string + configure func(*PreparedPipeline) + path string + wantError string + wantBody string + }{ + { + name: "merge module error", + configure: func(prepared *PreparedPipeline) { + prepared.lanes[0].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{}, errors.New("merge exploded") + } + }, + path: "merge/notes/attempt-01.json", + wantError: "merge exploded", + }, + { + name: "normalize validator error", + configure: func(prepared *PreparedPipeline) { + prepared.lanes[0].normalizeValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("error-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + return contracts.ValidationResult{}, errors.New("validator exploded") + }, + }} + }, + path: "normalize/notes/attempt-01.json", + wantError: "validator exploded", + wantBody: "output", + }, + { + name: "merge final rejection", + configure: func(prepared *PreparedPipeline) { + prepared.lanes[0].mergeValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("reject-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, + typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "not accepted"}, nil + }, + }} + }, + path: "merge/notes/attempt-01.json", + wantBody: "rejection", + }, + { + name: "normalize serialization error", + configure: func(prepared *PreparedPipeline) { + prepared.lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{Value: "wrong artifact type"}, nil + } + }, + path: "normalize/notes/attempt-01.json", + wantError: "serialize normalize candidate", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + debug := newCapturedDebugRecorder() + tc.configure(prepared) + output, runErr := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + envelope := debug.envelope(t, tc.path) + if tc.wantError != "" { + if runErr == nil || !strings.Contains(runErr.Error(), tc.wantError) || !strings.Contains(envelope.Error, tc.wantError) { + t.Fatalf("run error = %v, envelope error = %q; want %q", runErr, envelope.Error, tc.wantError) + } + } else if runErr != nil { + t.Fatalf("Run() error = %v, want nil rejection outcome", runErr) + } + if tc.wantBody != "" && !strings.Contains(string(debug.json[tc.path]), tc.wantBody) { + t.Fatalf("attempt envelope = %s, want %q", debug.json[tc.path], tc.wantBody) + } + if tc.name == "merge final rejection" && len(output.Rejected) != 1 { + t.Fatalf("rejected outputs = %#v, want one", output.Rejected) + } + }) + } +} + +func TestRunnerKeepsValidatorLLMCallsOutOfModuleAttempt(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + debug := newCapturedDebugRecorder() + client := WithDebugLLMRecording(attemptDebugLLM{}, debug) + prepared.lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + if err := callAttemptDebugLLM(ctx, client, "merge-module"); err != nil { + return erasedTypedResult{}, err + } + return erasedTypedResult{Value: codecNotes{Items: []string{"merged"}}}, nil + } + prepared.lanes[0].mergeValidators.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, "merge-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) + } + moduleEnvelope := debug.envelope(t, "merge/notes/attempt-01.json") + validatorPath := "validate/merge/notes/typed~2fmerge/01-llm-check-attempt-01.json" + validatorEnvelope := debug.envelope(t, validatorPath) + if len(moduleEnvelope.LLMCalls) != 1 || !strings.Contains(moduleEnvelope.LLMCalls[0].ResponsePath, "merge/notes/attempt-01/") { + t.Fatalf("module LLM calls = %#v, want module call only", moduleEnvelope.LLMCalls) + } + if len(validatorEnvelope.LLMCalls) != 1 || !strings.Contains(validatorEnvelope.LLMCalls[0].ResponsePath, "validate/merge/notes/") { + t.Fatalf("validator LLM calls = %#v, want validator call only", validatorEnvelope.LLMCalls) + } + if moduleEnvelope.LLMCalls[0].CallID == validatorEnvelope.LLMCalls[0].CallID { + t.Fatalf("module and validator envelopes reference the same call: %#v", moduleEnvelope.LLMCalls) + } +} + +type attemptReuseLoader struct { + CheckpointLoader + laneID string + merge MergeCheckpoint + normalize NormalizeCheckpoint +} + +func (l attemptReuseLoader) Enabled() bool { return true } + +func (l attemptReuseLoader) Merge(laneID, _ string, _ []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { + if laneID == l.laneID { + return l.merge, CheckpointDecision{Reused: true, Reason: "test reuse"} + } + return MergeCheckpoint{}, CheckpointDecision{Reason: "not found"} +} + +func (l attemptReuseLoader) Normalize(laneID, _ string, _ []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { + if laneID == l.laneID { + return l.normalize, CheckpointDecision{Reused: true, Reason: "test reuse"} + } + return NormalizeCheckpoint{}, CheckpointDecision{Reason: "not found"} +} + +func TestRunnerCheckpointReuseDoesNotSynthesizeModuleAttempts(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + lane := prepared.lanes[0] + merge, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Merge.Module, "source", codecNotes{Items: []string{"merged"}}) + if err != nil { + t.Fatalf("checkpointArtifact(merge): %v", err) + } + normalize, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Normalize.Module, "source", codecNotes{Items: []string{"normalized"}}) + if err != nil { + t.Fatalf("checkpointArtifact(normalize): %v", err) + } + loader := attemptReuseLoader{ + CheckpointLoader: NoopCheckpointLoader(), + laneID: lane.resolved.ID, + merge: MergeCheckpoint{Output: merge}, + normalize: NormalizeCheckpoint{Output: normalize}, + } + debug := newCapturedDebugRecorder() + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug, Checkpoint: loader}); err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + for _, name := range debug.names() { + if (strings.HasPrefix(name, "merge/notes/attempt-") || strings.HasPrefix(name, "normalize/notes/attempt-")) && name != "" { + t.Fatalf("checkpoint reuse synthesized module attempt artifact %q", name) + } + } + for _, name := range []string{"merge/notes/input.json", "merge/notes/output.json", "normalize/notes/input.json", "normalize/notes/output.json"} { + if !debug.has(name) { + t.Errorf("checkpoint reuse missing stage-level artifact %q", name) + } + } +} + +func TestRunnerTreatsModuleAttemptDebugWriteFailureAsFrameworkError(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + debug := newCapturedDebugRecorder() + debug.failPath = "merge/notes/attempt-01.json" + _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + if err == nil || !strings.Contains(err.Error(), "write merge attempt debug artifact") || !strings.Contains(err.Error(), "debug recorder failure") { + t.Fatalf("Run() error = %v, want merge attempt debug failure", err) + } +} diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 82b326d..e2c9da6 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -265,7 +265,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) 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 { - _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope)) + _ = 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) } artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: extracted.Value} @@ -280,7 +280,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef accepted, serialized = artifact, stored acceptedWarnings = append(cloneWarnings(extracted.Warnings), warnings...) - if debugErr := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(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 { + 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 { return false, nil, debugErr } return true, nil, nil diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 147e8ac..098fd1f 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -197,7 +197,7 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) result, 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 { - _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope)) + _ = 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) } artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: result.Value} @@ -212,7 +212,7 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef accepted, serializedAccepted = artifact, stored acceptedWarnings = append(cloneWarnings(result.Warnings), warnings...) - if debugErr := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(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 { + 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 { return false, nil, debugErr } return true, nil, nil @@ -276,21 +276,48 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C return err } ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - result, callErr := typed.merge(ctx, 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)}) + 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) + } + 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 { - return false, nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr) + 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 } candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value} - warnings, rejected, validateErr := r.validateTypedArtifact(ctx, 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}, prepared.mergeValidators, attempt, input.Debug) + stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value) + attemptWarnings := cloneWarnings(result.Warnings) + 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 + } + 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}, prepared.mergeValidators, attempt, input.Debug) + attemptWarnings = append(attemptWarnings, warnings...) + payload := map[string]any{"output": debugCheckpointArtifact(stored), "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 } - stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value) - if encodeErr != nil { - return false, nil, encodeErr + if debugErr := attemptEnvelope(payload, nil); debugErr != nil { + return false, nil, fmt.Errorf("write merge attempt debug artifact: %w", debugErr) } merged, serializedMerge = candidate, stored - mergeWarnings = append(cloneWarnings(result.Warnings), warnings...) + mergeWarnings = attemptWarnings return true, nil, nil }) if runErr != nil { @@ -339,20 +366,47 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C return err } ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - result, callErr := typed.normalize(ctx, 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 { - return false, nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr) + 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) } - warnings, rejected, validateErr := r.validateTypedArtifact(ctx, 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}, prepared.normalizeValidators, attempt, input.Debug) - if validateErr != nil || rejected != nil { - return false, rejected, validateErr + 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 } stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value) + attemptWarnings := cloneWarnings(result.Warnings) if encodeErr != nil { - return false, nil, encodeErr + 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 + } + 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}, prepared.normalizeValidators, attempt, input.Debug) + attemptWarnings = append(attemptWarnings, warnings...) + payload := map[string]any{"output": debugCheckpointArtifact(stored), "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 + } + if debugErr := attemptEnvelope(payload, nil); debugErr != nil { + return false, nil, fmt.Errorf("write normalize attempt debug artifact: %w", debugErr) } serializedNormalize = stored - normalizeWarnings = append(cloneWarnings(result.Warnings), warnings...) + normalizeWarnings = attemptWarnings return true, nil, nil }) if runErr != nil { @@ -410,7 +464,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE var err error started := time.Now().UTC() attemptPath := path.Join("validate", debugPathComponent(string(target.stage)), debugPathComponent(target.laneID), debugPathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(binding.Module), attempt)) - validatorCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath) switch item.resolved.Target { case ValidatorTargetTyped: target.llmProfile = binding.LLMProfile @@ -430,7 +484,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE if err != nil { debugCall.Error = err.Error() } - if debugErr := writeDebugTimed(debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope)); debugErr != nil { + 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 {