diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 959d20f..9fa3445 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -88,9 +88,11 @@ fall back to normal execution and are refreshed by the recorder. When workspace debug output is enabled, the CLI passes a debug recorder for the current run ID. The runner writes framework-boundary inputs, outputs, structured LLM calls, validator calls, timing, and retry attempt metadata -through that interface. Debug output is not used for resume and can contain -sensitive source, reference, prompt, and model-output material. Concrete modules -still do not receive workspace paths. +through that interface. Each retry or validator attempt records any LLM calls +made within that attempt in an `llm_calls` array and writes scoped copies of the +canonical LLM call artifacts under the attempt directory. Debug output is not +used for resume and can contain sensitive source, reference, prompt, and +model-output material. Concrete modules still do not receive workspace paths. ## Registries And Module Specs diff --git a/docs/operations.md b/docs/operations.md index c7b3f93..193107f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -96,7 +96,7 @@ When `workspace.resume.enabled: true` and `workspace.directory` is set, runs write checkpoints under: ```text -/checkpoints//-// +/checkpoints//-/// ``` Each workflow step owns its own manifest and payload files. There is no @@ -114,8 +114,9 @@ sensitive local state. A checkpoint is reused only when its workspace schema version, checkpoint identity digest, step status, dependency fingerprints, payload files, and payload digests match the current invocation. Changes to input bytes, resolved -pipeline digest, selected lanes, runtime LLM profile override, or materialized -reference digests invalidate reuse. +pipeline digest, selected lanes, runtime LLM profile override, materialized +reference digests, or other identity material invalidate reuse and use a +separate checkpoint directory. Plain `notarius run` does not reuse checkpoints. It executes the workflow and refreshes checkpoint files when checkpointing is enabled. `notarius run @@ -137,9 +138,12 @@ checkpointing does not write debug output. Debug artifacts include framework-boundary inputs and outputs for source, chunk, extract, merge, normalize, and output work, structured LLM request and response data from Notarius contracts, validator requests and results, timing, -and retry attempt metadata. Debug artifacts may contain source material, -reference material, prompt inputs, model outputs, and other sensitive data. -Obvious credential-shaped values and sensitive map keys are redacted, but debug +and retry attempt metadata. Canonical LLM call artifacts are written under +`llm/call-000N.json`; LLM calls made inside a retry or validator attempt are +also copied under that attempt directory and linked from the attempt +`llm_calls` array. Debug artifacts may contain source material, reference +material, prompt inputs, model outputs, and other sensitive data. Obvious +credential-shaped values and sensitive map keys are redacted, but debug directories should still be protected as sensitive local state. ## Retention @@ -195,7 +199,7 @@ Workspace checkpoint and debug directories can also be removed when no longer needed. Remove exact identity or run directories, for example: ```sh -rm -rf /var/lib/notarius/checkpoints/dnd-session/seriatim-abcdef123456/7890abcd1234 +rm -rf /var/lib/notarius/checkpoints/dnd-session/seriatim-abcdef123456/7890abcd1234/identityabcd1234 rm -rf /var/lib/notarius/debug/run-1234567890 ``` diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 4680be1..b552652 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -2244,6 +2244,8 @@ func TestRunPipelineWritesDebugWhenWorkspaceDebugEnabled(t *testing.T) { "chunk/input.json", "chunk/output.json", "extract/spells/input.json", + "extract/spells/chunk-000001-attempt-01.json", + "extract/spells/chunk-000001-attempt-01/llm-call-0001.json", "extract/spells/output.json", "merge/spells/input.json", "merge/spells/output.json", @@ -2257,6 +2259,10 @@ func TestRunPipelineWritesDebugWhenWorkspaceDebugEnabled(t *testing.T) { t.Fatalf("expected debug artifact %q: %v", name, err) } } + attemptDebug := string(readFile(t, filepath.Join(debugDir, "extract/spells/chunk-000001-attempt-01.json"))) + if !strings.Contains(attemptDebug, `"llm_calls"`) || !strings.Contains(attemptDebug, `"scoped_path"`) { + t.Fatalf("extract attempt debug = %s, want scoped llm_calls", attemptDebug) + } assertPathNotExist(t, filepath.Join(workspaceDir, "checkpoints")) } @@ -3640,7 +3646,8 @@ func onlyCheckpointIdentityDir(t *testing.T, workspaceDir string) string { t.Helper() pipelineDir := onlyChildDir(t, filepath.Join(workspaceDir, "checkpoints")) inputDir := onlyChildDir(t, pipelineDir) - return onlyChildDir(t, inputDir) + pipelineDigestDir := onlyChildDir(t, inputDir) + return onlyChildDir(t, pipelineDigestDir) } func anyDiagnosticsFileContains(t *testing.T, runDirs []string, name string, want string) bool { diff --git a/internal/core/workspace/identity.go b/internal/core/workspace/identity.go index bf17c7d..361dbb1 100644 --- a/internal/core/workspace/identity.go +++ b/internal/core/workspace/identity.go @@ -125,7 +125,15 @@ func (i CheckpointIdentity) RelativePath() (string, error) { if err != nil { return "", fmt.Errorf("checkpoint identity pipeline digest: %w", err) } - return filepath.ToSlash(filepath.Join(pipelineID, inputKey+"-"+sourceComponent, pipelineComponent)), nil + identityDigest := digestPrefix(i.Digest) + if identityDigest == "" { + return "", fmt.Errorf("checkpoint identity digest prefix must not be empty") + } + identityComponent, err := safePathComponent(identityDigest) + if err != nil { + return "", fmt.Errorf("checkpoint identity digest: %w", err) + } + return filepath.ToSlash(filepath.Join(pipelineID, inputKey+"-"+sourceComponent, pipelineComponent, identityComponent)), nil } func identityDigest(identity CheckpointIdentity) (string, error) { diff --git a/internal/core/workspace/identity_test.go b/internal/core/workspace/identity_test.go index cf3c760..d9f5b1f 100644 --- a/internal/core/workspace/identity_test.go +++ b/internal/core/workspace/identity_test.go @@ -118,7 +118,9 @@ func TestCheckpointIdentityPathIsFilesystemSafe(t *testing.T) { if strings.Contains(relative, `\`) || strings.Contains(relative, "..") { t.Fatalf("relative path is not filesystem safe: %q", relative) } - if relative != "campaign~2fmain/seriatim~2finput-abcdef0123456789/1234567890abcdef" { + identityDigest := digestPrefix(identity.Digest) + wantRelative := "campaign~2fmain/seriatim~2finput-abcdef0123456789/1234567890abcdef/" + identityDigest + if relative != wantRelative { t.Fatalf("relative path = %q", relative) } @@ -131,12 +133,34 @@ func TestCheckpointIdentityPathIsFilesystemSafe(t *testing.T) { if err != nil { t.Fatalf("CheckpointDirectory: %v", err) } - want := filepath.Join(root, "checkpoints", "campaign~2fmain", "seriatim~2finput-abcdef0123456789", "1234567890abcdef") + want := filepath.Join(root, "checkpoints", "campaign~2fmain", "seriatim~2finput-abcdef0123456789", "1234567890abcdef", identityDigest) if got != want { t.Fatalf("checkpoint directory = %q, want %q", got, want) } } +func TestCheckpointIdentityPathIncludesInvocationIdentity(t *testing.T) { + base := mustIdentity(t, identityInput()) + changedInput := identityInput() + changedInput.References[0].Digest = "sha256:reference-b" + changed := mustIdentity(t, changedInput) + + if base.Digest == changed.Digest { + t.Fatalf("test setup produced same identity digest: %q", base.Digest) + } + basePath, err := base.RelativePath() + if err != nil { + t.Fatalf("base RelativePath: %v", err) + } + changedPath, err := changed.RelativePath() + if err != nil { + t.Fatalf("changed RelativePath: %v", err) + } + if basePath == changedPath { + t.Fatalf("relative path did not change with invocation identity: %q", basePath) + } +} + func TestCheckpointDirectoryDisabledReturnsEmptyPath(t *testing.T) { settings := Settings{CheckpointsRoot: filepath.Join(t.TempDir(), "checkpoints")} got, err := settings.CheckpointDirectory(mustIdentity(t, identityInput())) diff --git a/internal/framework/llm/scriptorium_client.go b/internal/framework/llm/scriptorium_client.go index b5f888f..0f7e5c7 100644 --- a/internal/framework/llm/scriptorium_client.go +++ b/internal/framework/llm/scriptorium_client.go @@ -108,21 +108,24 @@ func (c *ScriptoriumClient) CompleteStructured(ctx context.Context, req contract if result == nil { return contracts.StructuredCompletionResponse{}, fmt.Errorf("run Scriptorium prompt %q: empty result", promptID) } + response := c.responseFromResult(result) if result.Validation.Status == scriptorium.ValidationFailed || !result.Validation.IsValid { - return contracts.StructuredCompletionResponse{}, fmt.Errorf("run Scriptorium prompt %q: validation failed: %s", promptID, strings.Join(result.Validation.Errors, "; ")) + return response, fmt.Errorf("run Scriptorium prompt %q: validation failed: %s", promptID, strings.Join(result.Validation.Errors, "; ")) } + if len(strings.TrimSpace(string(response.Content))) == 0 { + return response, fmt.Errorf("run Scriptorium prompt %q: empty structured output", promptID) + } + if err := json.Unmarshal(response.Content, out); err != nil { + return response, fmt.Errorf("decode Scriptorium structured output for prompt %q: %w", promptID, err) + } + return response, nil +} +func (c *ScriptoriumClient) responseFromResult(result *scriptorium.RunResult) contracts.StructuredCompletionResponse { content := result.Artifact.Body if len(content) == 0 { content = []byte(result.RawOutput) } - if len(strings.TrimSpace(string(content))) == 0 { - return contracts.StructuredCompletionResponse{}, fmt.Errorf("run Scriptorium prompt %q: empty structured output", promptID) - } - if err := json.Unmarshal(content, out); err != nil { - return contracts.StructuredCompletionResponse{}, fmt.Errorf("decode Scriptorium structured output for prompt %q: %w", promptID, err) - } - profile := artifacts.LLMProfileManifest{ ID: strings.TrimSpace(result.SelectedProfileID), Provider: scriptoriumProviderName, @@ -139,7 +142,7 @@ func (c *ScriptoriumClient) CompleteStructured(ctx context.Context, req contract PromptTokens: result.Usage.PromptTokens, CompletionTokens: result.Usage.CompletionTokens, TotalTokens: result.Usage.TotalTokens, - }, nil + } } func (c *ScriptoriumClient) LLMProfileManifests() []artifacts.LLMProfileManifest { diff --git a/internal/framework/llm/scriptorium_client_test.go b/internal/framework/llm/scriptorium_client_test.go index 1ed2bb8..248f5f4 100644 --- a/internal/framework/llm/scriptorium_client_test.go +++ b/internal/framework/llm/scriptorium_client_test.go @@ -87,7 +87,7 @@ func TestScriptoriumClientValidationFailureReturnsError(t *testing.T) { client := newTestScriptoriumClient(t, &fakeScriptoriumLLM{content: `{"bad":true}`}) var out map[string]any - _, err := client.CompleteStructured(context.Background(), contracts.StructuredCompletionRequest{ + resp, err := client.CompleteStructured(context.Background(), contracts.StructuredCompletionRequest{ PromptID: "adapter.test", SessionID: "session-123", Inputs: contracts.LLMInputSet{ @@ -97,6 +97,28 @@ func TestScriptoriumClientValidationFailureReturnsError(t *testing.T) { if err == nil || !strings.Contains(err.Error(), "validation failed") { t.Fatalf("CompleteStructured() error = %v, want validation failure", err) } + if got := string(resp.Content); got != `{"bad":true}` { + t.Fatalf("response content = %q, want raw failed output", got) + } +} + +func TestScriptoriumClientDecodeFailureReturnsRawResponse(t *testing.T) { + client := newTestScriptoriumClient(t, &fakeScriptoriumLLM{content: `{"ok":true}`}) + + var out []any + resp, err := client.CompleteStructured(context.Background(), contracts.StructuredCompletionRequest{ + PromptID: "adapter.test", + SessionID: "session-123", + Inputs: contracts.LLMInputSet{ + "transcript": contracts.NewLLMInputMaterial("transcript", "application/json", []byte(`{"source":true}`), "", ""), + }, + }, &out) + if err == nil || !strings.Contains(err.Error(), "decode Scriptorium structured output") { + t.Fatalf("CompleteStructured() error = %v, want decode failure", err) + } + if got := string(resp.Content); got != `{"ok":true}` { + t.Fatalf("response content = %q, want raw decode-failed output", got) + } } func TestScriptoriumClientProviderFailureIncludesContextAndRedactsBearerToken(t *testing.T) { diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index fd2cbbc..8ece1f2 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -5,10 +5,12 @@ import ( "crypto/sha256" "encoding/base64" "encoding/hex" + "errors" "fmt" "path" "regexp" "strings" + "sync" "time" "unicode/utf8" @@ -56,15 +58,16 @@ func debugPathComponent(value string) string { } type debugTimedEnvelope struct { - Stage string `json:"stage,omitempty"` - LaneID string `json:"lane_id,omitempty"` - ModuleKey string `json:"module_key,omitempty"` - Attempt int `json:"attempt,omitempty"` - StartedAt time.Time `json:"started_at"` - CompletedAt time.Time `json:"completed_at"` - DurationMS int64 `json:"duration_ms"` - Payload any `json:"payload,omitempty"` - Error string `json:"error,omitempty"` + Stage string `json:"stage,omitempty"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + Attempt int `json:"attempt,omitempty"` + StartedAt time.Time `json:"started_at"` + CompletedAt time.Time `json:"completed_at"` + DurationMS int64 `json:"duration_ms"` + Payload any `json:"payload,omitempty"` + LLMCalls []debugLLMCallReference `json:"llm_calls,omitempty"` + Error string `json:"error,omitempty"` } type debugBinaryEnvelope struct { @@ -168,6 +171,15 @@ type debugStructuredLLMCall struct { Error string `json:"error,omitempty"` } +type debugLLMCallReference struct { + CallID string `json:"call_id"` + CanonicalPath string `json:"canonical_path"` + ScopedPath string `json:"scoped_path,omitempty"` + PromptID string `json:"prompt_id,omitempty"` + ProfileID string `json:"profile_id,omitempty"` + Error bool `json:"error,omitempty"` +} + type debugValidationRequest struct { Stage string `json:"stage"` LaneID string `json:"lane_id,omitempty"` @@ -197,9 +209,19 @@ type debugValidationCall struct { type debugLLMClient struct { inner contracts.StructuredLLMClient recorder DebugRecorder + mu sync.Mutex counter int } +type debugLLMScope struct { + prefix string + parent *debugLLMScope + mu sync.Mutex + calls []debugLLMCallReference +} + +type debugLLMScopeContextKey struct{} + func wrapDebugLLMClient(client contracts.StructuredLLMClient, recorder DebugRecorder) contracts.StructuredLLMClient { if client == nil || recorder == nil || !recorder.Enabled() { return client @@ -208,7 +230,11 @@ func wrapDebugLLMClient(client contracts.StructuredLLMClient, recorder DebugReco } func (client *debugLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { + client.mu.Lock() client.counter++ + callID := fmt.Sprintf("call-%04d", client.counter) + client.mu.Unlock() + started := time.Now().UTC() response, err := client.inner.CompleteStructured(ctx, req, out) completed := time.Now().UTC() @@ -219,7 +245,7 @@ func (client *debugLLMClient) CompleteStructured(ctx context.Context, req contra if err != nil { payload.Error = err.Error() } - writeErr := writeDebugTimed(client.recorder, path.Join("llm", fmt.Sprintf("call-%04d.json", client.counter)), debugTimedEnvelope{ + envelope := debugTimedEnvelope{ Stage: req.StageName, ModuleKey: req.StageName, StartedAt: started, @@ -227,7 +253,25 @@ func (client *debugLLMClient) CompleteStructured(ctx context.Context, req contra DurationMS: completed.Sub(started).Milliseconds(), Payload: payload, Error: payload.Error, - }) + } + canonicalPath := path.Join("llm", callID+".json") + writeErr := writeDebugTimed(client.recorder, canonicalPath, envelope) + callRef := debugLLMCallReference{ + CallID: callID, + CanonicalPath: canonicalPath, + PromptID: req.PromptID, + ProfileID: debugFirstNonEmptyString(response.ProfileID, req.ProfileID), + Error: err != nil, + } + if scope := debugLLMScopeFromContext(ctx); scope != nil { + scopedPath := path.Join(scope.prefix, "llm-"+callID+".json") + callRef.ScopedPath = scopedPath + scopedWriteErr := writeDebugTimed(client.recorder, scopedPath, envelope) + if scopedWriteErr != nil { + writeErr = errors.Join(writeErr, scopedWriteErr) + } + scope.record(callRef) + } if err != nil { return response, err } @@ -237,6 +281,73 @@ func (client *debugLLMClient) CompleteStructured(ctx context.Context, req contra return response, err } +func withDebugLLMScope(ctx context.Context, prefix string) (context.Context, *debugLLMScope) { + if ctx == nil { + ctx = context.Background() + } + prefix = cleanDebugPath(prefix) + scope := &debugLLMScope{ + prefix: prefix, + parent: debugLLMScopeFromContext(ctx), + } + return context.WithValue(ctx, debugLLMScopeContextKey{}, scope), scope +} + +func debugLLMScopeFromContext(ctx context.Context) *debugLLMScope { + if ctx == nil { + return nil + } + scope, _ := ctx.Value(debugLLMScopeContextKey{}).(*debugLLMScope) + return scope +} + +func (scope *debugLLMScope) record(ref debugLLMCallReference) { + if scope == nil { + return + } + scope.mu.Lock() + scope.calls = append(scope.calls, ref) + scope.mu.Unlock() + if scope.parent != nil { + scope.parent.record(ref) + } +} + +func (scope *debugLLMScope) references() []debugLLMCallReference { + if scope == nil { + return nil + } + scope.mu.Lock() + defer scope.mu.Unlock() + if len(scope.calls) == 0 { + return nil + } + out := make([]debugLLMCallReference, len(scope.calls)) + copy(out, scope.calls) + return out +} + +func cleanDebugPath(value string) string { + parts := strings.Split(path.Clean(strings.TrimSpace(value)), "/") + out := make([]string, 0, len(parts)) + for _, part := range parts { + out = append(out, debugPathComponent(part)) + } + if len(out) == 0 { + return "_" + } + return path.Join(out...) +} + +func debugFirstNonEmptyString(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return value + } + } + return "" +} + func (client *debugLLMClient) LLMProfileManifests() []artifacts.LLMProfileManifest { provider, ok := client.inner.(contracts.LLMProfileManifestProvider) if !ok { @@ -261,6 +372,13 @@ func writeDebugTimed(recorder DebugRecorder, name string, envelope debugTimedEnv return recorder.WriteJSON(name, envelope) } +func debugEnvelopeWithLLMCalls(envelope debugTimedEnvelope, scope *debugLLMScope) debugTimedEnvelope { + if scope != nil { + envelope.LLMCalls = scope.references() + } + return envelope +} + 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 929bba0..545daf5 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -205,7 +205,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } chunksAccepted, chunkRejection, err = runWithRetry(ctx, input.Pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { attemptStarted := time.Now().UTC() - chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{ + attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt)) + attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + chunkResult, err := chunker.Chunk(attemptCtx, contracts.ChunkRequest{ Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID, @@ -216,25 +218,44 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err Metadata: input.Metadata, }) if err != nil { - _ = writeDebugTimed(debugRecorder, path.Join("chunk", fmt.Sprintf("attempt-%02d.json", attempt)), debugTimedEnvelope{ + _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(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) } if len(chunkResult.Chunks) == 0 { - return false, nil, fmt.Errorf("chunker %q returned no chunks", chunker.Key()) + err := fmt.Errorf("chunker %q returned no chunks", chunker.Key()) + _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageChunk), + ModuleKey: chunker.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Error: err.Error(), + }, llmScope)) + return false, nil, err } chunks, err := validateAndCanonicalizeChunkResult(doc, chunkResult.Chunks) if err != nil { - return false, nil, fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) + err := fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err) + _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(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 } - validationWarnings, rejection, err := r.validateChunksRaw(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.Pipeline.ChunkReferences.ReferenceSet, input.LLMClient, input.Metadata, input.Pipeline.ValidatorChains, attempt, input.Debug) + validationWarnings, rejection, err := r.validateChunksRaw(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.Pipeline.ChunkReferences.ReferenceSet, input.LLMClient, input.Metadata, input.Pipeline.ValidatorChains, attempt, input.Debug) if err != nil || rejection != nil { - _ = writeDebugTimed(debugRecorder, path.Join("chunk", fmt.Sprintf("attempt-%02d.json", attempt)), debugTimedEnvelope{ + _ = writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, @@ -244,12 +265,12 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err "warnings": append(cloneWarnings(chunkResult.Warnings), validationWarnings...), "rejection": debugRejectedOutputPtr(rejection), }, - }) + }, llmScope)) return false, rejection, err } canonicalChunks = chunks chunkWarnings = append(cloneWarnings(chunkResult.Warnings), validationWarnings...) - if err := writeDebugTimed(debugRecorder, path.Join("chunk", fmt.Sprintf("attempt-%02d.json", attempt)), debugTimedEnvelope{ + if err := writeDebugTimed(debugRecorder, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, @@ -258,7 +279,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err "chunks": debugSourceChunkEnvelopes(chunks), "warnings": chunkWarnings, }, - }); err != nil { + }, llmScope)); err != nil { return false, nil, err } return true, nil, nil @@ -419,7 +440,10 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp var acceptedOutput contracts.ExtractOutput var acceptedWarnings []contracts.Warning accepted, rejection, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - result, err := extractor.Extract(ctx, contracts.ExtractionRequest{ + attemptStarted := time.Now().UTC() + attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d-attempt-%02d", chunk.Index+1, attempt)) + attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + result, err := extractor.Extract(attemptCtx, contracts.ExtractionRequest{ Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), @@ -431,6 +455,14 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp Metadata: input.Metadata, }) if err != nil { + _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageExtract), + LaneID: lane.ID, + ModuleKey: extractor.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Error: err.Error(), + }, llmScope)) return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, extractor.Key(), err) } extractOutput := result.Output @@ -440,7 +472,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp extractOutput.ChunkID = chunk.ID extractOutput.ChunkIndex = chunk.Index extractOutput.Payload.Warnings = append(extractOutput.Payload.Warnings, result.Warnings...) - validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ + validationWarnings, rejection, err := r.validateRaw(attemptCtx, rawValidationTarget{ stage: StageExtract, laneID: lane.ID, moduleKey: extractor.Key(), @@ -461,10 +493,35 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp debug: input.Debug, }) if err != nil || rejection != nil { + _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageExtract), + LaneID: lane.ID, + ModuleKey: extractor.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "output": debugExtractOutputEnvelope(extractOutput), + "warnings": append(cloneWarnings(result.Warnings), validationWarnings...), + "rejection": debugRejectedOutputPtr(rejection), + }, + }, llmScope)) return false, rejection, err } acceptedOutput = cloneExtractOutput(extractOutput) acceptedWarnings = append(cloneWarnings(result.Warnings), validationWarnings...) + if err := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageExtract), + LaneID: lane.ID, + ModuleKey: extractor.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "output": debugExtractOutputEnvelope(extractOutput), + "warnings": acceptedWarnings, + }, + }, llmScope)); err != nil { + return false, nil, err + } return true, nil, nil }) if err != nil { @@ -534,7 +591,10 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err) } mergeAccepted, mergeRejection, err := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - mergeResult, err := merger.Merge(ctx, contracts.MergeRequest{ + attemptStarted := time.Now().UTC() + attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) + attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + mergeResult, err := merger.Merge(attemptCtx, contracts.MergeRequest{ Source: doc, LaneID: lane.ID, ExtractOutputs: cloneExtractOutputs(extractOutputs), @@ -547,6 +607,14 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp Metadata: input.Metadata, }) if err != nil { + _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageMerge), + LaneID: lane.ID, + ModuleKey: merger.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Error: err.Error(), + }, llmScope)) return false, nil, fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, merger.Key(), err) } mergeOutput := mergeResult.Output @@ -554,7 +622,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp mergeOutput.MergerKey = merger.Key() mergeOutput.SourceID = doc.ID mergeOutput.Payload.Warnings = append(mergeOutput.Payload.Warnings, mergeResult.Warnings...) - validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ + validationWarnings, rejection, err := r.validateRaw(attemptCtx, rawValidationTarget{ stage: StageMerge, laneID: lane.ID, moduleKey: merger.Key(), @@ -573,10 +641,35 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp debug: input.Debug, }) if err != nil || rejection != nil { + _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageMerge), + LaneID: lane.ID, + ModuleKey: merger.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "output": debugMergeOutputEnvelope(mergeOutput), + "warnings": append(cloneWarnings(mergeResult.Warnings), validationWarnings...), + "rejection": debugRejectedOutputPtr(rejection), + }, + }, llmScope)) return false, rejection, err } acceptedMerge = cloneMergeOutput(mergeOutput) mergeWarnings = append(cloneWarnings(mergeResult.Warnings), validationWarnings...) + if err := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageMerge), + LaneID: lane.ID, + ModuleKey: merger.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "output": debugMergeOutputEnvelope(mergeOutput), + "warnings": mergeWarnings, + }, + }, llmScope)); err != nil { + return false, nil, err + } return true, nil, nil }) if err != nil { @@ -654,7 +747,10 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err) } normalizeAccepted, normalizeRejection, err := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - normalizeResult, err := normalizer.Normalize(ctx, contracts.NormalizeRequest{ + attemptStarted := time.Now().UTC() + attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) + attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + normalizeResult, err := normalizer.Normalize(attemptCtx, contracts.NormalizeRequest{ Source: doc, LaneID: lane.ID, MergeOutput: cloneMergeOutput(acceptedMerge), @@ -667,6 +763,14 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp Metadata: input.Metadata, }) if err != nil { + _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageNormalize), + LaneID: lane.ID, + ModuleKey: normalizer.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Error: err.Error(), + }, llmScope)) return false, nil, fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err) } normalizeOutput := normalizeResult.Output @@ -674,7 +778,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp normalizeOutput.NormalizerKey = normalizer.Key() normalizeOutput.SourceID = doc.ID normalizeOutput.Payload.Warnings = append(normalizeOutput.Payload.Warnings, normalizeResult.Warnings...) - validationWarnings, rejection, err := r.validateRaw(ctx, rawValidationTarget{ + validationWarnings, rejection, err := r.validateRaw(attemptCtx, rawValidationTarget{ stage: StageNormalize, laneID: lane.ID, moduleKey: normalizer.Key(), @@ -693,10 +797,35 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp debug: input.Debug, }) if err != nil || rejection != nil { + _ = writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageNormalize), + LaneID: lane.ID, + ModuleKey: normalizer.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "output": debugNormalizeOutputEnvelope(normalizeOutput), + "warnings": append(cloneWarnings(normalizeResult.Warnings), validationWarnings...), + "rejection": debugRejectedOutputPtr(rejection), + }, + }, llmScope)) return false, rejection, err } acceptedNormalize = cloneNormalizeOutput(normalizeOutput) normalizeWarnings = append(cloneWarnings(normalizeResult.Warnings), validationWarnings...) + if err := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ + Stage: string(StageNormalize), + LaneID: lane.ID, + ModuleKey: normalizer.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "output": debugNormalizeOutputEnvelope(normalizeOutput), + "warnings": normalizeWarnings, + }, + }, llmScope)); err != nil { + return false, nil, err + } return true, nil, nil }) if err != nil { @@ -845,14 +974,16 @@ func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([ } var warnings []contracts.Warning - for _, validatorBinding := range chain.Validators { + for index, validatorBinding := range chain.Validators { validator, err := r.registries.Validators.Build(validatorBinding.Binding.Module) if err != nil { return nil, nil, fmt.Errorf("build validator %q: %w", validatorBinding.Binding.Module, err) } request := target.validationRequest(validatorBinding.Binding) started := time.Now().UTC() - result, err := validator.Validate(ctx, request) + attemptPath := path.Join("validate", debugPathComponent(string(target.stage)), debugPathComponent(target.laneID), debugPathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, debugPathComponent(validator.Name()), target.attempt)) + validatorCtx, llmScope := withDebugLLMScope(ctx, attemptPath) + result, err := validator.Validate(validatorCtx, request) debugPayload := debugValidationCall{ ValidatorName: validator.Name(), Request: debugValidationRequestEnvelope(request), @@ -861,7 +992,7 @@ func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([ if err != nil { debugPayload.Error = err.Error() } - if debugErr := writeDebugTimed(target.debug, path.Join("validate", debugPathComponent(string(target.stage)), debugPathComponent(target.laneID), debugPathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d.json", len(warnings)+1, debugPathComponent(validator.Name()), target.attempt)), debugTimedEnvelope{ + if debugErr := writeDebugTimed(target.debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{ Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, @@ -869,7 +1000,7 @@ func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([ StartedAt: started, Payload: debugPayload, Error: debugPayload.Error, - }); debugErr != nil { + }, llmScope)); debugErr != nil { return nil, nil, fmt.Errorf("write validation debug artifact: %w", debugErr) } if err != nil { diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index d24cda2..7ae09ae 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -2,6 +2,7 @@ package pipeline import ( "context" + "encoding/base64" "errors" "reflect" "strings" @@ -1289,6 +1290,46 @@ func TestRunRetriesSameModuleInputAfterValidatorRejection(t *testing.T) { } } +func TestRunDebugFailedChunkAttemptReferencesScopedLLMOutput(t *testing.T) { + modules := defaultRunnerModules() + modules.chunker.callLLM = true + modules.chunker.llmPromptID = "runner.chunk" + modules.chunker.err = errors.New("malformed structured output") + recorder := newMemoryDebugRecorder() + + _, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{ + Pipeline: resolvedPipeline(), + LLMClient: debugResponseLLMClient{content: []byte(`{"raw":true}`), profileID: "debug-profile"}, + Debug: recorder, + }) + if err == nil || !strings.Contains(err.Error(), "malformed structured output") { + t.Fatalf("Run() error = %v, want chunk failure", err) + } + + attempt := recorder.envelope(t, "chunk/attempt-01.json") + if len(attempt.LLMCalls) != 1 { + t.Fatalf("llm_calls = %#v, want one scoped call", attempt.LLMCalls) + } + call := attempt.LLMCalls[0] + if call.CallID != "call-0001" || call.CanonicalPath != "llm/call-0001.json" || call.ScopedPath != "chunk/attempt-01/llm-call-0001.json" { + t.Fatalf("llm call reference = %#v, want canonical and scoped paths", call) + } + if call.PromptID != "runner.chunk" || call.ProfileID != "debug-profile" || call.Error { + t.Fatalf("llm call metadata = %#v, want prompt/profile and no call error", call) + } + + scoped := recorder.envelope(t, call.ScopedPath) + scopedPayload, ok := scoped.Payload.(debugStructuredLLMCall) + if !ok { + t.Fatalf("scoped payload type = %T, want debugStructuredLLMCall", scoped.Payload) + } + wantContent := base64.StdEncoding.EncodeToString([]byte(`{"raw":true}`)) + if scopedPayload.Response.Content != wantContent { + t.Fatalf("scoped response content = %q, want %q", scopedPayload.Response.Content, wantContent) + } + _ = recorder.envelope(t, call.CanonicalPath) +} + func TestRunStopsRetryAfterConfiguredAttemptsAndRecordsAttemptCount(t *testing.T) { modules := defaultRunnerModules() validator := &runnerChainValidator{name: "chain-extract", approved: []bool{false}, reason: "bad_extract", message: "extract rejected"} @@ -1915,6 +1956,8 @@ type runnerChunker struct { err error failureErr error failuresBeforeSuccess int + callLLM bool + llmPromptID string manifestMetadata map[string]any requests []contracts.ChunkRequest } @@ -1937,6 +1980,20 @@ func (chunker *runnerChunker) Chunk(ctx context.Context, req contracts.ChunkRequ } return contracts.ChunkResult{}, err } + if chunker.callLLM && req.LLMClient != nil { + promptID := strings.TrimSpace(chunker.llmPromptID) + if promptID == "" { + promptID = "runner.chunk" + } + var out map[string]any + if _, err := req.LLMClient.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ + StageName: chunker.key, + PromptID: promptID, + ProfileID: req.LLMProfile, + }, &out); err != nil { + return contracts.ChunkResult{}, err + } + } return contracts.ChunkResult{ Chunks: chunker.chunks, Warnings: chunker.warnings, @@ -2264,6 +2321,51 @@ func (client fakeLLMClient) CompleteStructured(ctx context.Context, req contract return contracts.StructuredCompletionResponse{}, nil } +type debugResponseLLMClient struct { + content []byte + profileID string + err error +} + +func (client debugResponseLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { + profileID := client.profileID + if profileID == "" { + profileID = req.ProfileID + } + return contracts.StructuredCompletionResponse{ + Content: append([]byte(nil), client.content...), + ProfileID: profileID, + }, client.err +} + +type memoryDebugRecorder struct { + payloads map[string]any +} + +func newMemoryDebugRecorder() *memoryDebugRecorder { + return &memoryDebugRecorder{payloads: map[string]any{}} +} + +func (recorder *memoryDebugRecorder) Enabled() bool { return true } + +func (recorder *memoryDebugRecorder) WriteJSON(name string, payload any) error { + recorder.payloads[name] = payload + return nil +} + +func (recorder *memoryDebugRecorder) envelope(t *testing.T, name string) debugTimedEnvelope { + t.Helper() + payload, ok := recorder.payloads[name] + if !ok { + t.Fatalf("debug artifact %q not written; got %#v", name, recorder.payloads) + } + envelope, ok := payload.(debugTimedEnvelope) + if !ok { + t.Fatalf("debug artifact %q type = %T, want debugTimedEnvelope", name, payload) + } + return envelope +} + type manifestReportingLLMClient struct { fakeLLMClient profiles []artifacts.LLMProfileManifest