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

Add retry-scoped merge and normalize debugging

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 13:43:25 +00:00
parent 35e45f0914
commit fbc3d9add6
7 changed files with 558 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
docs/internal/pipeline.md
View File

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

26
docs/operations.md
View File

@@ -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/<lane-id>/attempt-<NN>.json
merge/<lane-id>/attempt-<NN>/prompt-<NNNN>.json
merge/<lane-id>/attempt-<NN>/response-<NNNN>.json
merge/<lane-id>/attempt-<NN>/response-content-<NNNN>.<ext>
normalize/<lane-id>/attempt-<NN>.json
normalize/<lane-id>/attempt-<NN>/prompt-<NNNN>.json
normalize/<lane-id>/attempt-<NN>/response-<NNNN>.json
normalize/<lane-id>/attempt-<NN>/response-content-<NNNN>.<ext>
```
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.

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

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

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

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

430
internal/framework/pipeline/runner_attempt_debug_test.go Normal file
View File

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

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

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

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

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