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

Update the debug workflow to provide raw LLM output

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-07 23:08:14 -05:00
parent ae65b95374
commit 3011dd91ca
10 changed files with 476 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
docs/internal/pipeline.md
View File

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

18
docs/operations.md
View File

@@ -96,7 +96,7 @@ When `workspace.resume.enabled: true` and `workspace.directory` is set, runs
write checkpoints under:
```text
<workspace.directory>/checkpoints/<pipeline-id>/<input-key>-<source-digest>/<pipeline-digest>/
<workspace.directory>/checkpoints/<pipeline-id>/<input-key>-<source-or-input-digest>/<pipeline-digest>/<identity-digest>/
```
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
```

9
internal/cli/run_test.go
View File

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

10
internal/core/workspace/identity.go
View File

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

28
internal/core/workspace/identity_test.go
View File

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

21
internal/framework/llm/scriptorium_client.go
View File

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

24
internal/framework/llm/scriptorium_client_test.go
View File

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

140
internal/framework/pipeline/debug.go
View File

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

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

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

102
internal/framework/pipeline/runner_test.go
View File

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