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

Encode checkpoint and debug path identities

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-09 00:41:28 +00:00
parent 2ad9283148
commit cda7a61b47
12 changed files with 182 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
docs/internal/state.md
View File

@@ -47,6 +47,9 @@ The serialized
they do not describe a current public state surface.
Ordered-step lane checkpoints include the step identity in their storage scope.
Accepted step and lane identities are encoded injectively before becoming
filesystem path components, while ordinary safe identifiers retain their
readable paths.
When a later lane consumes a generated artifact, its dependency fingerprints
include the producer's artifact kind, complete schema identity, media type,
canonical content digest, and size. Ordinary resume compares those fingerprints

32
internal/core/fileio/fileio.go
View File

@@ -10,6 +10,38 @@ import (
"strings"
)
// EncodePathComponent returns a filesystem-safe, injective representation of
// one logical path component.
func EncodePathComponent(value string) string {
if value == "" {
return "%"
}
const hexadecimal = "0123456789ABCDEF"
var out strings.Builder
for index := 0; index < len(value); index++ {
byteValue := value[index]
switch {
case byteValue >= 'a' && byteValue <= 'z', byteValue >= 'A' && byteValue <= 'Z', byteValue >= '0' && byteValue <= '9', byteValue == '-', byteValue == '_':
out.WriteByte(byteValue)
case byteValue == '.' && safePathDot(value, index):
out.WriteByte(byteValue)
default:
out.WriteByte('%')
out.WriteByte(hexadecimal[byteValue>>4])
out.WriteByte(hexadecimal[byteValue&0x0f])
}
}
return out.String()
}
func safePathDot(value string, index int) bool {
if value == "." || value == ".." {
return false
}
return (index == 0 || value[index-1] != '.') && (index+1 == len(value) || value[index+1] != '.')
}
func SafePath(root, name string) (string, error) {
root = strings.TrimSpace(root)
if root == "" {

33
internal/core/fileio/fileio_test.go
View File

@@ -15,6 +15,39 @@ func TestSafePathRejectsUnsafeNames(t *testing.T) {
}
}
func TestEncodePathComponentIsInjectiveAndSafe(t *testing.T) {
root := t.TempDir()
seen := make(map[string]string)
for _, test := range []struct {
value string
want string
}{
{value: "", want: "%"},
{value: ".", want: "%2E"},
{value: "..", want: "%2E%2E"},
{value: "_", want: "_"},
{value: "a..b", want: "a%2E%2Eb"},
{value: "safe.identifier-9", want: "safe.identifier-9"},
{value: "left/right", want: "left%2Fright"},
{value: "%", want: "%25"},
{value: "~", want: "%7E"},
{value: " a ", want: "%20a%20"},
{value: "é", want: "%C3%A9"},
} {
got := EncodePathComponent(test.value)
if got != test.want {
t.Errorf("EncodePathComponent(%q) = %q, want %q", test.value, got, test.want)
}
if previous, ok := seen[got]; ok {
t.Errorf("EncodePathComponent(%q) = %q, collides with %q", test.value, got, previous)
}
seen[got] = test.value
if _, err := SafePath(root, "components/"+got); err != nil {
t.Errorf("EncodePathComponent(%q) produced unsafe component %q: %v", test.value, got, err)
}
}
}
func TestWriteBytesIsAtomicAndUsesRequestedModes(t *testing.T) {
root := t.TempDir()
if err := WriteBytes(root, "nested/value", []byte("value"), 0o700, 0o600); err != nil {

31
internal/framework/checkpoint/recorder.go
View File

@@ -475,36 +475,9 @@ func laneManifestPath(stage string, stepID string, laneID string) string {
func lanePayloadPath(stage string, stepID string, laneID string, file string) string {
if strings.TrimSpace(stepID) == "" {
return path.Join(stage, checkpointPathComponent(laneID), file)
return path.Join(stage, fileio.EncodePathComponent(laneID), file)
}
return path.Join(stage, checkpointPathComponent(stepID), checkpointPathComponent(laneID), file)
}
func checkpointPathComponent(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "_"
}
var b strings.Builder
for _, r := range value {
switch {
case r >= 'a' && r <= 'z':
b.WriteRune(r)
case r >= 'A' && r <= 'Z':
b.WriteRune(r)
case r >= '0' && r <= '9':
b.WriteRune(r)
case r == '-' || r == '_' || r == '.':
b.WriteRune(r)
default:
b.WriteString(fmt.Sprintf("~%x", r))
}
}
out := b.String()
if out == "." || out == ".." || strings.Contains(out, "..") {
return "_"
}
return out
return path.Join(stage, fileio.EncodePathComponent(stepID), fileio.EncodePathComponent(laneID), file)
}
func contentDigest(content []byte) string {

52
internal/framework/checkpoint/recorder_test.go
View File

@@ -5,6 +5,8 @@ import (
"path/filepath"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
@@ -66,6 +68,56 @@ func TestStepAwareRecorderAndLoaderIsolateLaneState(t *testing.T) {
}
}
func TestStepAwareCheckpointPreservesDistinctDotIdentities(t *testing.T) {
root := t.TempDir()
identity := testIdentity(t)
recorder, err := NewFilesystemRecorder(root, identity)
if err != nil {
t.Fatal(err)
}
stepRecorder := recorder.(pipeline.StepCheckpointRecorder)
for _, test := range []struct {
stepID string
content string
}{
{stepID: ".", content: `{"identity":"dot"}`},
{stepID: "..", content: `{"identity":"dot-dot"}`},
} {
artifact := pipeline.CheckpointArtifact{
LaneID: "lane", ModuleKey: "normalize-module", SourceID: "source", ChunkID: "chunk", ChunkRef: source.SourceRef{SourceID: "source", StartUnitID: 1, EndUnitID: 1}, SchemaDigest: "sha256:schema",
Artifact: contracts.SerializedArtifact{Kind: "kind", Schema: contracts.ArtifactSchema{ID: "schema", Name: "Schema", Version: "1"}, MediaType: "application/json", Content: []byte(test.content)},
}
if err := stepRecorder.NormalizeSucceededForStep(test.stepID, "lane", "normalize-module", nil, artifact, nil); err != nil {
t.Fatalf("record %q: %v", test.stepID, err)
}
}
loader, err := NewFilesystemLoader(root, identity)
if err != nil {
t.Fatal(err)
}
for _, test := range []struct {
stepID string
content string
path string
}{
{stepID: ".", content: `{"identity":"dot"}`, path: "%2E"},
{stepID: "..", content: `{"identity":"dot-dot"}`, path: "%2E%2E"},
} {
loaded, decision := loader.AcceptedNormalize(test.stepID, "lane", "normalize-module")
if !decision.Reused || string(loaded.Output.Artifact.Content) != test.content {
t.Errorf("load %q = %#v, decision=%#v", test.stepID, loaded, decision)
}
relative, err := identity.RelativePath()
if err != nil {
t.Fatal(err)
}
if _, err := os.Stat(filepath.Join(root, relative, "normalize", test.path, "lane", "manifest.json")); err != nil {
t.Errorf("checkpoint for %q: %v", test.stepID, err)
}
}
}
func TestCheckpointSchemaCompatibilityIdentifiers(t *testing.T) {
if WorkspaceSchemaVersion != "notarius.workspace.v3" || WorkspaceSchemaVersionV2 != "notarius.workspace.v2" || WorkspaceSchemaVersionV1 != "notarius.workspace.v1" {
t.Fatal("checkpoint schema identifiers are incorrect")

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

@@ -17,6 +17,7 @@ import (
"unicode/utf8"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
@@ -34,32 +35,6 @@ func NoopDebugRecorder() DebugRecorder { return noopDebugRecorder{} }
func (noopDebugRecorder) Enabled() bool { return false }
func (noopDebugRecorder) WriteJSON(string, any) error { return nil }
func (noopDebugRecorder) WriteBytes(string, []byte) error { return nil }
func debugPathComponent(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "_"
}
var b strings.Builder
for _, r := range value {
switch {
case r >= 'a' && r <= 'z':
b.WriteRune(r)
case r >= 'A' && r <= 'Z':
b.WriteRune(r)
case r >= '0' && r <= '9':
b.WriteRune(r)
case r == '-' || r == '_' || r == '.':
b.WriteRune(r)
default:
b.WriteString(fmt.Sprintf("~%x", r))
}
}
out := b.String()
if out == "." || out == ".." || strings.Contains(out, "..") {
return "_"
}
return out
}
type debugTimedEnvelope struct {
Stage string `json:"stage,omitempty"`
@@ -242,7 +217,7 @@ func (client *debugLLMClient) CompleteStructured(ctx context.Context, req contra
}
scopePrefix := cleanDebugPath(req.StageName)
if scopePrefix == "_" {
if req.StageName == "" {
scopePrefix = "llm"
}
if scope := debugLLMScopeFromContext(ctx); scope != nil {
@@ -310,7 +285,6 @@ func withDebugLLMScope(ctx context.Context, prefix string) (context.Context, *de
if ctx == nil {
ctx = context.Background()
}
prefix = cleanDebugPath(prefix)
scope := &debugLLMScope{
prefix: prefix,
parent: debugLLMScopeFromContext(ctx),
@@ -322,7 +296,6 @@ func withIsolatedDebugLLMScope(ctx context.Context, prefix string) (context.Cont
if ctx == nil {
ctx = context.Background()
}
prefix = cleanDebugPath(prefix)
scope := &debugLLMScope{prefix: prefix}
return context.WithValue(ctx, debugLLMScopeContextKey{}, scope), scope
}
@@ -362,15 +335,12 @@ func (scope *debugLLMScope) references() []debugLLMCallReference {
}
func cleanDebugPath(value string) string {
parts := strings.Split(path.Clean(strings.TrimSpace(value)), "/")
parts := strings.Split(value, "/")
out := make([]string, 0, len(parts))
for _, part := range parts {
out = append(out, debugPathComponent(part))
out = append(out, fileio.EncodePathComponent(part))
}
if len(out) == 0 {
return "_"
}
return path.Join(out...)
return strings.Join(out, "/")
}
func debugFirstNonEmptyString(values ...string) string {

39
internal/framework/pipeline/debug_test.go
View File

@@ -1,13 +1,52 @@
package pipeline
import (
"context"
"encoding/json"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestCleanDebugPathPreservesRawComponents(t *testing.T) {
for _, test := range []struct {
value string
want string
}{
{value: "", want: "%"},
{value: ".", want: "%2E"},
{value: "..", want: "%2E%2E"},
{value: "a//b", want: "a/%/b"},
{value: "/a/", want: "%/a/%"},
{value: " a ", want: "%20a%20"},
} {
if got := cleanDebugPath(test.value); got != test.want {
t.Errorf("cleanDebugPath(%q) = %q, want %q", test.value, got, test.want)
}
}
}
func TestDebugLLMPathsKeepDotIdentitiesDistinct(t *testing.T) {
recorder := newCapturedDebugRecorder()
client := WithDebugLLMRecording(attemptDebugLLM{}, recorder)
for _, test := range []struct {
stageName string
path string
}{
{stageName: ".", path: "%2E/response-0001.json"},
{stageName: "..", path: "%2E%2E/response-0002.json"},
} {
if _, err := client.CompleteStructured(context.Background(), contracts.StructuredCompletionRequest{StageName: test.stageName}, nil); err != nil {
t.Fatalf("CompleteStructured(%q): %v", test.stageName, err)
}
if !recorder.has(test.path) {
t.Errorf("debug artifact %q was not written; names = %#v", test.path, recorder.names())
}
}
}
func TestDebugSourceDocumentPreservesUnitReferences(t *testing.T) {
doc := validSourceDocument()
envelope := debugSourceDocumentEnvelope(doc)

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

@@ -15,6 +15,7 @@ import (
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/chunkmap"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
@@ -437,7 +438,7 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument,
for index, item := range prepared.validators {
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))
attemptPath := path.Join("validate", fileio.EncodePathComponent(string(StageChunk)), "", fileio.EncodePathComponent(moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, fileio.EncodePathComponent(binding.Module), attempt))
validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath)
var result contracts.ValidationResult
requestMetadata, cloneErr := cloneMetadata(metadata)

2
internal/framework/pipeline/runner_attempt_debug_test.go
View File

@@ -351,7 +351,7 @@ func TestRunnerKeepsValidatorLLMCallsOutOfModuleAttempt(t *testing.T) {
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"
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)

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

@@ -10,6 +10,7 @@ import (
"sync"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
@@ -408,7 +409,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source.
var acceptedWarnings []contracts.Warning
ok, rejection, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
started := time.Now().UTC()
attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
attemptPath := path.Join("extract", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
@@ -498,10 +499,10 @@ func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints C
}
local.Warnings = append(local.Warnings, cloneWarnings(results.warnings)...)
local.Rejected = append(local.Rejected, cloneRejectedOutputs(results.rejected)...)
if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "decision": results.decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
if err := writeDebugTimed(input.Debug, path.Join("extract", fileio.EncodePathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "decision": results.decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
return local, &laneRunError{stage: StageExtract, err: err}
}
if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "outputs": debugCheckpointArtifacts(results.serialized), "rejected": debugRejectedOutputEnvelopes(results.rejected), "warnings": debugWarningEnvelopes(results.warnings)}}); err != nil {
if err := writeDebugTimed(input.Debug, path.Join("extract", fileio.EncodePathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "outputs": debugCheckpointArtifacts(results.serialized), "rejected": debugRejectedOutputEnvelopes(results.rejected), "warnings": debugWarningEnvelopes(results.warnings)}}); err != nil {
return local, &laneRunError{stage: StageExtract, err: err}
}
if len(results.accepted) == 0 {

2
internal/framework/pipeline/runner_terminal_debug_test.go
View File

@@ -329,7 +329,7 @@ func TestRunnerKeepsExtractModuleAndValidatorLLMCallsIsolated(t *testing.T) {
t.Fatalf("Run() error = %v, want nil", err)
}
module := debug.envelope(t, "extract/notes/chunk-000001/attempt-01.json")
validator := debug.envelope(t, "validate/extract/notes/typed~2fextract-notes/01-llm-check-attempt-01.json")
validator := debug.envelope(t, "validate/extract/notes/typed%2Fextract-notes/01-llm-check-attempt-01.json")
if len(module.LLMCalls) != 1 || !strings.Contains(module.LLMCalls[0].ResponsePath, "extract/notes/chunk-000001/attempt-01/") {
t.Fatalf("module LLM calls = %#v, want extract module call only", module.LLMCalls)
}

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

@@ -10,6 +10,7 @@ import (
"path"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
@@ -232,7 +233,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
return stageResult, err
}
mergeDecision = mergeResolution.decision
if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
if err := writeDebugTimed(input.Debug, path.Join("merge", fileio.EncodePathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
return stageResult, err
}
var merged erasedMergeArtifact
@@ -249,7 +250,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
}
ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
started := time.Now().UTC()
attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
@@ -303,7 +304,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints
return stageResult, err
}
}
if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil {
if err := writeDebugTimed(input.Debug, path.Join("merge", fileio.EncodePathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil {
return stageResult, err
}
stageResult.artifact = merged
@@ -329,7 +330,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
return stageResult, err
}
normalizeDecision = normalizeResolution.decision
if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugCheckpointArtifact(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
if err := writeDebugTimed(input.Debug, path.Join("normalize", fileio.EncodePathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugCheckpointArtifact(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
return stageResult, err
}
var serializedNormalize CheckpointArtifact
@@ -344,7 +345,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
}
ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
started := time.Now().UTC()
attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt))
attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started})
requestMetadata, metadataErr := cloneMetadata(input.Metadata)
@@ -417,7 +418,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi
return stageResult, err
}
}
if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil {
if err := writeDebugTimed(input.Debug, path.Join("normalize", fileio.EncodePathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil {
return stageResult, err
}
stageResult.serialized = serializedNormalize
@@ -469,7 +470,7 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE
var result contracts.ValidationResult
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))
attemptPath := path.Join("validate", fileio.EncodePathComponent(string(target.stage)), fileio.EncodePathComponent(target.laneID), fileio.EncodePathComponent(target.moduleKey), fmt.Sprintf("%02d-%s-attempt-%02d", index+1, fileio.EncodePathComponent(binding.Module), attempt))
validatorCtx, llmScope := withIsolatedDebugLLMScope(ctx, attemptPath)
requestTarget := target
requestTarget.sourceInput = target.sourceInput.Clone()