From a5bbfea9b99bdd77fbf51aba8ec9c9265e45b7d4 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 8 Jul 2026 03:14:59 +0000 Subject: [PATCH] Write workspace debug artifacts during runs --- docs/config.md | 3 +- docs/internal/diagnostics.md | 1 + docs/internal/overview.md | 5 +- docs/internal/pipeline.md | 5 + docs/operations.md | 21 + internal/cli/run.go | 6 + internal/cli/run_test.go | 211 +++++++++- internal/framework/debug/recorder.go | 34 ++ internal/framework/pipeline/debug.go | 563 ++++++++++++++++++++++++++ internal/framework/pipeline/runner.go | 291 ++++++++++++- 10 files changed, 1134 insertions(+), 6 deletions(-) create mode 100644 internal/framework/debug/recorder.go create mode 100644 internal/framework/pipeline/debug.go diff --git a/docs/config.md b/docs/config.md index d5bf37c..aeae3f9 100644 --- a/docs/config.md +++ b/docs/config.md @@ -356,7 +356,8 @@ When `workspace.resume.enabled` is true, runs write stage-owned checkpoint artifacts under `/checkpoints/`. `notarius run --resume` can reuse valid checkpoints from a compatible invocation. -Debug artifact writers are not part of the current workflow. +When `workspace.debug.enabled` is true, runs write per-invocation debug +artifacts under `/debug//`. ## Diagnostics diff --git a/docs/internal/diagnostics.md b/docs/internal/diagnostics.md index cfdc026..d6b3347 100644 --- a/docs/internal/diagnostics.md +++ b/docs/internal/diagnostics.md @@ -34,6 +34,7 @@ Implemented artifact names: - `effective-config.json` - `resolved-pipeline.json` - `resolved-references.json` +- `checkpoint-events.json` - `source-document.json` - `run-manifest.json` - `run-report.json` diff --git a/docs/internal/overview.md b/docs/internal/overview.md index bf11c54..3fd99bb 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -44,9 +44,10 @@ production modules. encoders, and structured LLM clients. - `internal/framework/checkpoint`: workspace-backed checkpoint recorder and checkpoint payload envelope serialization. +- `internal/framework/debug`: workspace-backed debug artifact writer. - `internal/framework/pipeline`: module registries, module specs, profile - resolution, capability checks, run orchestration, checkpoint recorder - boundaries, warnings, validation, and manifest population. + resolution, capability checks, run orchestration, checkpoint and debug + recorder boundaries, warnings, validation, and manifest population. - `internal/framework/llm`: Scriptorium-backed structured-output client, prompt/schema asset registry, scheduler, schema registry, and secret redaction. diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 4e1db7f..4616617 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -83,6 +83,11 @@ status, identity digest, dependency fingerprints, payload files, and payload digests validate for the current invocation. Missing or invalid checkpoints 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. Concrete modules still do not receive workspace paths. + ## Registries And Module Specs `pipeline.Registries` holds concrete constructors for execution. A diff --git a/docs/operations.md b/docs/operations.md index 8a3c2af..3e5f331 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -111,6 +111,27 @@ 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. +## Debug + +When `workspace.debug.enabled: true` and `workspace.directory` is set, runs +write debug artifacts under: + +```text +/debug// +``` + +Debug output is per invocation. It is independent of checkpointing and is not +used for resume. Enabling debug does not write checkpoints, and enabling resume +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 +directories should still be protected as sensitive local state. + ## Retention Diagnostics retention is configured with `workspace.diagnostics.retention`, diff --git a/internal/cli/run.go b/internal/cli/run.go index 4748966..1425adb 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -21,6 +21,7 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/core/workspace" "gitea.maximumdirect.net/eric/notarius/internal/framework/checkpoint" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + frameworkdebug "gitea.maximumdirect.net/eric/notarius/internal/framework/debug" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) @@ -190,6 +191,10 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if *resume && !workspaceSettings.ResumeEnabled { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("--resume requires workspace.resume.enabled: true")) } + debugRecorder, err := frameworkdebug.NewWorkspaceRecorder(workspaceSettings, runID) + if err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("create debug recorder: %w", err)) + } catalog, err := effectiveCatalog(opts) if err != nil { @@ -279,6 +284,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i Warnings: referenceWarnings, Checkpoints: checkpointRecorder, Checkpoint: checkpointLoader, + Debug: debugRecorder, }) if err != nil { if output.Manifest.PipelineID != "" && runDir != nil { diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 11effa6..4680be1 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -3,12 +3,14 @@ package cli import ( "bytes" "context" + "encoding/base64" "encoding/json" "errors" "io/fs" "os" "path/filepath" "reflect" + "regexp" "sort" "strings" "testing" @@ -2219,6 +2221,121 @@ func TestRunPipelineWritesCheckpointsWhenWorkspaceResumeEnabled(t *testing.T) { assertPathNotExist(t, filepath.Join(workspaceDir, "debug")) } +func TestRunPipelineWritesDebugWhenWorkspaceDebugEnabled(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + outputDir := t.TempDir() + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceDebugEnabled("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", outputDir}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + debugDir := onlyChildDir(t, filepath.Join(workspaceDir, "debug")) + for _, name := range []string{ + "run.json", + "source/input.json", + "source/output.json", + "chunk/input.json", + "chunk/output.json", + "extract/spells/input.json", + "extract/spells/output.json", + "merge/spells/input.json", + "merge/spells/output.json", + "normalize/spells/input.json", + "normalize/spells/output.json", + "output/input.json", + "output/output.json", + "llm/call-0001.json", + } { + if _, err := os.Stat(filepath.Join(debugDir, name)); err != nil { + t.Fatalf("expected debug artifact %q: %v", name, err) + } + } + assertPathNotExist(t, filepath.Join(workspaceDir, "checkpoints")) +} + +func TestRunPipelineDebugAndResumeCanBeEnabledIndependently(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + outputDir := t.TempDir() + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceDiagnosticsAndStateEnabled("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", outputDir}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + if code != 0 { + t.Fatalf("seed RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + code = RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", outputDir, "--resume"}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + if code != 0 { + t.Fatalf("resume RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if entries := childDirs(t, filepath.Join(workspaceDir, "checkpoints")); len(entries) != 1 { + t.Fatalf("workspace checkpoint pipeline dirs = %v, want one", entries) + } + if entries := childDirs(t, filepath.Join(workspaceDir, "debug")); len(entries) != 2 { + t.Fatalf("workspace debug run dirs = %v, want two", entries) + } +} + +func TestRunPipelineDebugRedactsObviousSecrets(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + outputDir := t.TempDir() + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceDebugEnabled("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + client := newFakeRunLLMClient(false) + client.payload = map[string]any{ + "spell_casts": []map[string]any{ + { + "caster": "Aria", + "spell": "sk-secretvalue", + "effect": "Bearer secretvalue", + "narrative_description": "Aria casts a spell.", + "source_refs": []map[string]any{ + {"source_id": "session-alpha", "start_unit_id": 1, "end_unit_id": 1}, + }, + }, + }, + } + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", outputDir}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(client, nil), + }) + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + assertDebugTreeDoesNotContain(t, onlyChildDir(t, filepath.Join(workspaceDir, "debug")), "sk-secretvalue", "Bearer secretvalue") +} + +func TestWorkspaceStateRootsDoNotOverlap(t *testing.T) { + workspaceDir := filepath.Join(t.TempDir(), "workspace") + outputDir := t.TempDir() + configPath := writeTestConfig(t, mvpConfigYAMLWithWorkspaceDiagnosticsAndStateEnabled("dnd-session", workspaceDir, "always")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", outputDir}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + assertDistinctRoots(t, filepath.Join(workspaceDir, "diagnostics"), filepath.Join(workspaceDir, "checkpoints"), filepath.Join(workspaceDir, "debug")) +} + func TestRunPipelineResumeRequiresWorkspaceResumeEnabled(t *testing.T) { workspaceDir := filepath.Join(t.TempDir(), "workspace") outputDir := t.TempDir() @@ -2504,7 +2621,9 @@ func TestRunPipelineDiagnosticsDirFlagOverridesWorkspaceDiagnosticsOnly(t *testi if entries := childDirs(t, filepath.Join(workspaceDir, "checkpoints")); len(entries) != 1 { t.Fatalf("workspace checkpoint pipeline dirs = %v, want one", entries) } - assertPathNotExist(t, filepath.Join(workspaceDir, "debug")) + if entries := childDirs(t, filepath.Join(workspaceDir, "debug")); len(entries) != 1 { + t.Fatalf("workspace debug run dirs = %v, want one", entries) + } } func TestExampleFixtureConfigValidateAndPipelinesList(t *testing.T) { @@ -3054,6 +3173,24 @@ pipelines: ` } +func mvpConfigYAMLWithWorkspaceDebugEnabled(pipelineID, workspaceDir, retention string) string { + return `version: 2 +workspace: + directory: ` + workspaceDir + ` + diagnostics: + enabled: true + retention: ` + retention + ` + debug: + enabled: true +pipelines: + ` + pipelineID + `: + input: seriatim + artifacts: + spells: + extract: dnd/spells +` +} + func mvpConfigYAMLWithWorkspaceResumeAndChunker(pipelineID, workspaceDir, retention, chunker string) string { return `version: 2 workspace: @@ -3523,6 +3660,78 @@ func anyDiagnosticsFileContains(t *testing.T, runDirs []string, name string, wan return false } +func assertDistinctRoots(t *testing.T, roots ...string) { + t.Helper() + for i, first := range roots { + for _, second := range roots[i+1:] { + firstAbs, err := filepath.Abs(first) + if err != nil { + t.Fatalf("resolve %q: %v", first, err) + } + secondAbs, err := filepath.Abs(second) + if err != nil { + t.Fatalf("resolve %q: %v", second, err) + } + if firstAbs == secondAbs { + t.Fatalf("workspace roots overlap exactly: %q", firstAbs) + } + firstRel, err := filepath.Rel(firstAbs, secondAbs) + if err != nil { + t.Fatalf("rel %q %q: %v", firstAbs, secondAbs, err) + } + secondRel, err := filepath.Rel(secondAbs, firstAbs) + if err != nil { + t.Fatalf("rel %q %q: %v", secondAbs, firstAbs, err) + } + if !strings.HasPrefix(firstRel, ".."+string(filepath.Separator)) && firstRel != ".." { + t.Fatalf("workspace root %q contains %q", firstAbs, secondAbs) + } + if !strings.HasPrefix(secondRel, ".."+string(filepath.Separator)) && secondRel != ".." { + t.Fatalf("workspace root %q contains %q", secondAbs, firstAbs) + } + } + } +} + +var debugBase64FieldPattern = regexp.MustCompile(`"(?:content_base64|content)"\s*:\s*"([^"]*)"`) + +func assertDebugTreeDoesNotContain(t *testing.T, root string, forbidden ...string) { + t.Helper() + if err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { + if err != nil { + return err + } + if entry.IsDir() { + return nil + } + data, err := os.ReadFile(path) + if err != nil { + return err + } + text := string(data) + for _, value := range forbidden { + if strings.Contains(text, value) { + t.Fatalf("debug artifact %q contains forbidden value %q", path, value) + } + } + for _, match := range debugBase64FieldPattern.FindAllStringSubmatch(text, -1) { + decoded, err := base64.StdEncoding.DecodeString(match[1]) + if err != nil { + continue + } + decodedText := string(decoded) + for _, value := range forbidden { + if strings.Contains(decodedText, value) { + t.Fatalf("debug artifact %q decoded content contains forbidden value %q", path, value) + } + } + } + return nil + }); err != nil { + t.Fatalf("walk debug tree %q: %v", root, err) + } +} + func seedWorkspaceCheckpoint(t *testing.T, configPath string, inputPath string, extraArgs []string) { t.Helper() client := newFakeRunLLMClient(false) diff --git a/internal/framework/debug/recorder.go b/internal/framework/debug/recorder.go new file mode 100644 index 0000000..506cdb9 --- /dev/null +++ b/internal/framework/debug/recorder.go @@ -0,0 +1,34 @@ +package debug + +import ( + "strings" + + coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +type WorkspaceRecorder struct { + root string +} + +func NewWorkspaceRecorder(settings coreworkspace.Settings, runID string) (pipeline.DebugRecorder, error) { + root, err := settings.DebugRunDirectory(runID) + if err != nil { + return nil, err + } + if strings.TrimSpace(root) == "" { + return pipeline.NoopDebugRecorder(), nil + } + return &WorkspaceRecorder{root: root}, nil +} + +func (r *WorkspaceRecorder) Enabled() bool { + return r != nil && strings.TrimSpace(r.root) != "" +} + +func (r *WorkspaceRecorder) WriteJSON(name string, payload any) error { + if !r.Enabled() { + return nil + } + return coreworkspace.WriteJSON(r.root, name, payload) +} diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go new file mode 100644 index 0000000..fd2cbbc --- /dev/null +++ b/internal/framework/pipeline/debug.go @@ -0,0 +1,563 @@ +package pipeline + +import ( + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "fmt" + "path" + "regexp" + "strings" + "time" + "unicode/utf8" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type DebugRecorder interface { + Enabled() bool + WriteJSON(name string, payload any) error +} + +type noopDebugRecorder struct{} + +func NoopDebugRecorder() DebugRecorder { return noopDebugRecorder{} } + +func (noopDebugRecorder) Enabled() bool { return false } +func (noopDebugRecorder) WriteJSON(string, any) 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"` + 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"` +} + +type debugBinaryEnvelope struct { + ContentBase64 string `json:"content_base64,omitempty"` + ContentDigest string `json:"content_digest,omitempty"` + MediaType string `json:"media_type,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} + +type debugRawPayload struct { + Content debugBinaryEnvelope `json:"content"` +} + +type debugSourceInput struct { + SourceID string `json:"source_id,omitempty"` + Path string `json:"path,omitempty"` + Raw debugBinaryEnvelope `json:"raw,omitempty"` + Options map[string]any `json:"options,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type debugSourceDocument struct { + ID string `json:"id"` + Kind string `json:"kind"` + Format string `json:"format,omitempty"` + Digest string `json:"digest,omitempty"` + Units []source.SourceUnit `json:"units,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type debugSourceChunk struct { + ID string `json:"id"` + SourceID string `json:"source_id"` + Index int `json:"index"` + StartUnitID int `json:"start_unit_id"` + EndUnitID int `json:"end_unit_id"` + Content debugBinaryEnvelope `json:"content"` + Units []source.SourceUnit `json:"units,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type debugExtractOutput struct { + LaneID string `json:"lane_id"` + ExtractorKey string `json:"extractor_key"` + SourceID string `json:"source_id"` + ChunkID string `json:"chunk_id"` + ChunkIndex int `json:"chunk_index"` + Schema contracts.ResponseSchema `json:"schema,omitempty"` + Payload debugBinaryEnvelope `json:"payload"` +} + +type debugMergeOutput struct { + LaneID string `json:"lane_id"` + MergerKey string `json:"merger_key"` + SourceID string `json:"source_id,omitempty"` + Schema contracts.ResponseSchema `json:"schema,omitempty"` + Payload debugBinaryEnvelope `json:"payload"` +} + +type debugNormalizeOutput struct { + LaneID string `json:"lane_id"` + NormalizerKey string `json:"normalizer_key"` + SourceID string `json:"source_id,omitempty"` + Schema contracts.ResponseSchema `json:"schema,omitempty"` + Payload debugBinaryEnvelope `json:"payload"` +} + +type debugLLMInputMaterial struct { + Name string `json:"name"` + MediaType string `json:"media_type,omitempty"` + Content string `json:"content_base64,omitempty"` + Digest string `json:"digest,omitempty"` + OriginURI string `json:"origin_uri,omitempty"` + SizeBytes int64 `json:"size_bytes,omitempty"` +} + +type debugStructuredCompletionRequest struct { + StageName string `json:"stage_name"` + PromptID string `json:"prompt_id,omitempty"` + PromptVersion string `json:"prompt_version,omitempty"` + ProfileID string `json:"profile_id,omitempty"` + SessionID string `json:"session_id,omitempty"` + Inputs map[string]debugLLMInputMaterial `json:"inputs,omitempty"` + Vars map[string]any `json:"vars,omitempty"` +} + +type debugStructuredCompletionResponse struct { + Content string `json:"content,omitempty"` + Provider string `json:"provider,omitempty"` + Model string `json:"model,omitempty"` + ProfileID string `json:"profile_id,omitempty"` + PromptTokens int `json:"prompt_tokens,omitempty"` + CompletionTokens int `json:"completion_tokens,omitempty"` + TotalTokens int `json:"total_tokens,omitempty"` +} + +type debugStructuredLLMCall struct { + Request debugStructuredCompletionRequest `json:"request"` + Response debugStructuredCompletionResponse `json:"response,omitempty"` + Error string `json:"error,omitempty"` +} + +type debugValidationRequest struct { + Stage string `json:"stage"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key"` + SourceID string `json:"source_id,omitempty"` + SessionID string `json:"session_id,omitempty"` + LLMProfile string `json:"llm_profile,omitempty"` + Options map[string]any `json:"options,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + Schema contracts.ResponseSchema `json:"schema,omitempty"` + Payload *debugBinaryEnvelope `json:"payload,omitempty"` + ChunkID string `json:"chunk_id,omitempty"` + ChunkIndex int `json:"chunk_index,omitempty"` + Chunk *debugSourceChunk `json:"chunk,omitempty"` + Chunks []debugSourceChunk `json:"chunks,omitempty"` + ExtractOutputs []debugExtractOutput `json:"extract_outputs,omitempty"` + MergeOutput *debugMergeOutput `json:"merge_output,omitempty"` +} + +type debugValidationCall struct { + ValidatorName string `json:"validator_name"` + Request debugValidationRequest `json:"request"` + Result contracts.ValidationResult `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +type debugLLMClient struct { + inner contracts.StructuredLLMClient + recorder DebugRecorder + counter int +} + +func wrapDebugLLMClient(client contracts.StructuredLLMClient, recorder DebugRecorder) contracts.StructuredLLMClient { + if client == nil || recorder == nil || !recorder.Enabled() { + return client + } + return &debugLLMClient{inner: client, recorder: recorder} +} + +func (client *debugLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { + client.counter++ + started := time.Now().UTC() + response, err := client.inner.CompleteStructured(ctx, req, out) + completed := time.Now().UTC() + payload := debugStructuredLLMCall{ + Request: debugCompletionRequest(req), + Response: debugCompletionResponse(response), + } + if err != nil { + payload.Error = err.Error() + } + writeErr := writeDebugTimed(client.recorder, path.Join("llm", fmt.Sprintf("call-%04d.json", client.counter)), debugTimedEnvelope{ + Stage: req.StageName, + ModuleKey: req.StageName, + StartedAt: started, + CompletedAt: completed, + DurationMS: completed.Sub(started).Milliseconds(), + Payload: payload, + Error: payload.Error, + }) + if err != nil { + return response, err + } + if writeErr != nil { + return response, fmt.Errorf("write LLM debug artifact: %w", writeErr) + } + return response, err +} + +func (client *debugLLMClient) LLMProfileManifests() []artifacts.LLMProfileManifest { + provider, ok := client.inner.(contracts.LLMProfileManifestProvider) + if !ok { + return nil + } + return provider.LLMProfileManifests() +} + +func writeDebugTimed(recorder DebugRecorder, name string, envelope debugTimedEnvelope) error { + if recorder == nil || !recorder.Enabled() { + return nil + } + if envelope.CompletedAt.IsZero() { + envelope.CompletedAt = time.Now().UTC() + } + if envelope.StartedAt.IsZero() { + envelope.StartedAt = envelope.CompletedAt + } + if envelope.DurationMS == 0 { + envelope.DurationMS = envelope.CompletedAt.Sub(envelope.StartedAt).Milliseconds() + } + return recorder.WriteJSON(name, envelope) +} + +func debugContentEnvelope(content []byte, mediaType string, metadata map[string]any, warnings []contracts.Warning) debugBinaryEnvelope { + content = redactSecretBytes(content) + return debugBinaryEnvelope{ + ContentBase64: base64.StdEncoding.EncodeToString(content), + ContentDigest: debugContentDigest(content), + MediaType: mediaType, + Metadata: redactSensitiveMap(metadata), + Warnings: cloneWarnings(warnings), + } +} + +func debugPayloadEnvelope(payload contracts.RawPayload) debugBinaryEnvelope { + return debugContentEnvelope(payload.Content, payload.MediaType, payload.Metadata, payload.Warnings) +} + +func debugSourceDocumentEnvelope(doc *source.SourceDocument) *debugSourceDocument { + if doc == nil { + return nil + } + return &debugSourceDocument{ + ID: doc.ID, + Kind: doc.Kind, + Format: doc.Format, + Digest: doc.Digest, + Units: cloneSourceUnits(doc.Units), + Metadata: redactSensitiveMap(doc.Metadata), + } +} + +func debugSourceChunkEnvelope(chunk contracts.SourceChunk) debugSourceChunk { + return debugSourceChunk{ + ID: chunk.ID, + SourceID: chunk.SourceID, + Index: chunk.Index, + StartUnitID: chunk.StartUnitID, + EndUnitID: chunk.EndUnitID, + Content: debugContentEnvelope(chunk.Content, chunk.MediaType, chunk.Metadata, nil), + Units: cloneSourceUnits(chunk.Units), + Metadata: redactSensitiveMap(chunk.Metadata), + } +} + +func debugSourceChunkEnvelopes(chunks []contracts.SourceChunk) []debugSourceChunk { + if len(chunks) == 0 { + return nil + } + out := make([]debugSourceChunk, 0, len(chunks)) + for _, chunk := range chunks { + out = append(out, debugSourceChunkEnvelope(chunk)) + } + return out +} + +func debugExtractOutputEnvelope(output contracts.ExtractOutput) debugExtractOutput { + output.Schema.JSONSchema = nil + return debugExtractOutput{ + LaneID: output.LaneID, + ExtractorKey: output.ExtractorKey, + SourceID: output.SourceID, + ChunkID: output.ChunkID, + ChunkIndex: output.ChunkIndex, + Schema: output.Schema, + Payload: debugPayloadEnvelope(output.Payload), + } +} + +func debugExtractOutputEnvelopes(outputs []contracts.ExtractOutput) []debugExtractOutput { + if len(outputs) == 0 { + return nil + } + out := make([]debugExtractOutput, 0, len(outputs)) + for _, output := range outputs { + out = append(out, debugExtractOutputEnvelope(output)) + } + return out +} + +func debugMergeOutputEnvelope(output contracts.MergeOutput) debugMergeOutput { + output.Schema.JSONSchema = nil + return debugMergeOutput{ + LaneID: output.LaneID, + MergerKey: output.MergerKey, + SourceID: output.SourceID, + Schema: output.Schema, + Payload: debugPayloadEnvelope(output.Payload), + } +} + +func debugNormalizeOutputEnvelope(output contracts.NormalizeOutput) debugNormalizeOutput { + output.Schema.JSONSchema = nil + return debugNormalizeOutput{ + LaneID: output.LaneID, + NormalizerKey: output.NormalizerKey, + SourceID: output.SourceID, + Schema: output.Schema, + Payload: debugPayloadEnvelope(output.Payload), + } +} + +func debugNormalizeOutputEnvelopes(outputs []contracts.NormalizeOutput) []debugNormalizeOutput { + if len(outputs) == 0 { + return nil + } + out := make([]debugNormalizeOutput, 0, len(outputs)) + for _, output := range outputs { + out = append(out, debugNormalizeOutputEnvelope(output)) + } + return out +} + +type debugOutputFile struct { + Name string `json:"name"` + ContentType string `json:"content_type,omitempty"` + Content debugBinaryEnvelope `json:"content"` +} + +func debugOutputFiles(files []contracts.OutputFile) []debugOutputFile { + if len(files) == 0 { + return nil + } + out := make([]debugOutputFile, 0, len(files)) + for _, file := range files { + out = append(out, debugOutputFile{ + Name: file.Name, + ContentType: file.ContentType, + Content: debugContentEnvelope(file.Bytes, file.ContentType, nil, nil), + }) + } + return out +} + +func debugCompletionRequest(req contracts.StructuredCompletionRequest) debugStructuredCompletionRequest { + inputs := make(map[string]debugLLMInputMaterial, len(req.Inputs)) + for key, material := range req.Inputs { + inputs[key] = debugLLMInputMaterial{ + Name: material.Name, + MediaType: material.MediaType, + Content: base64.StdEncoding.EncodeToString(redactSecretBytes(material.Content)), + Digest: material.Digest, + OriginURI: material.OriginURI, + SizeBytes: material.SizeBytes, + } + } + if len(inputs) == 0 { + inputs = nil + } + return debugStructuredCompletionRequest{ + StageName: req.StageName, + PromptID: req.PromptID, + PromptVersion: req.PromptVersion, + ProfileID: req.ProfileID, + SessionID: req.SessionID, + Inputs: inputs, + Vars: redactSensitiveMap(req.Vars), + } +} + +func debugCompletionResponse(response contracts.StructuredCompletionResponse) debugStructuredCompletionResponse { + return debugStructuredCompletionResponse{ + Content: base64.StdEncoding.EncodeToString(redactSecretBytes(response.Content)), + Provider: response.Provider, + Model: response.Model, + ProfileID: response.ProfileID, + PromptTokens: response.PromptTokens, + CompletionTokens: response.CompletionTokens, + TotalTokens: response.TotalTokens, + } +} + +func debugValidationRequestEnvelope(req contracts.ValidationRequest) debugValidationRequest { + req.Schema.JSONSchema = nil + out := debugValidationRequest{ + Stage: req.Stage, + LaneID: req.LaneID, + ModuleKey: req.ModuleKey, + SourceID: req.SourceID, + SessionID: req.SessionID, + LLMProfile: req.LLMProfile, + Options: redactSensitiveMap(req.Options), + Metadata: redactSensitiveMap(req.Metadata), + Schema: req.Schema, + ChunkID: req.ChunkID, + ChunkIndex: req.ChunkIndex, + } + payload := debugPayloadEnvelope(req.Payload) + out.Payload = &payload + if req.Chunk != nil { + chunk := debugSourceChunkEnvelope(*req.Chunk) + out.Chunk = &chunk + } + out.Chunks = debugSourceChunkEnvelopes(req.Chunks) + out.ExtractOutputs = debugExtractOutputEnvelopes(req.ExtractOutputs) + if len(req.MergeOutput.Payload.Content) > 0 || req.MergeOutput.LaneID != "" { + merge := debugMergeOutputEnvelope(req.MergeOutput) + out.MergeOutput = &merge + } + return out +} + +func debugValidationResultEnvelope(result contracts.ValidationResult) contracts.ValidationResult { + result.Message = string(redactSecretBytes([]byte(result.Message))) + result.DiagnosticArtifactPath = string(redactSecretBytes([]byte(result.DiagnosticArtifactPath))) + for i := range result.Warnings { + result.Warnings[i].Message = string(redactSecretBytes([]byte(result.Warnings[i].Message))) + } + return result +} + +func debugRejectedOutputEnvelope(rejected contracts.RejectedOutput) contracts.RejectedOutput { + rejected.Message = string(redactSecretBytes([]byte(rejected.Message))) + rejected.DiagnosticArtifactPath = string(redactSecretBytes([]byte(rejected.DiagnosticArtifactPath))) + return rejected +} + +func debugRejectedOutputPtr(rejected *contracts.RejectedOutput) any { + if rejected == nil { + return nil + } + out := debugRejectedOutputEnvelope(*rejected) + return out +} + +func debugRejectedOutputEnvelopes(rejected []contracts.RejectedOutput) []contracts.RejectedOutput { + if len(rejected) == 0 { + return nil + } + out := make([]contracts.RejectedOutput, 0, len(rejected)) + for _, item := range rejected { + out = append(out, debugRejectedOutputEnvelope(item)) + } + return out +} + +func debugContentDigest(content []byte) string { + sum := sha256.Sum256(content) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +var secretPatterns = []*regexp.Regexp{ + regexp.MustCompile(`(?i)bearer\s+[a-z0-9._~+/=-]{8,}`), + regexp.MustCompile(`(?i)sk-[a-z0-9_-]{8,}`), +} + +func redactSecretBytes(content []byte) []byte { + if len(content) == 0 || !utf8.Valid(content) { + return append([]byte(nil), content...) + } + text := string(content) + for _, pattern := range secretPatterns { + text = pattern.ReplaceAllString(text, "[REDACTED]") + } + return []byte(text) +} + +func redactSensitiveMap(values map[string]any) map[string]any { + if len(values) == 0 { + return nil + } + out := make(map[string]any, len(values)) + for key, value := range values { + if sensitiveKey(key) { + out[key] = "[REDACTED]" + continue + } + out[key] = redactSensitiveValue(value) + } + return out +} + +func redactSensitiveValue(value any) any { + switch typed := value.(type) { + case string: + return string(redactSecretBytes([]byte(typed))) + case map[string]any: + return redactSensitiveMap(typed) + case map[string]string: + out := make(map[string]string, len(typed)) + for key, value := range typed { + if sensitiveKey(key) { + out[key] = "[REDACTED]" + } else { + out[key] = string(redactSecretBytes([]byte(value))) + } + } + return out + default: + return value + } +} + +func sensitiveKey(key string) bool { + key = strings.ToLower(key) + return strings.Contains(key, "api_key") || + strings.Contains(key, "apikey") || + strings.Contains(key, "authorization") || + strings.Contains(key, "bearer") || + strings.Contains(key, "password") || + strings.Contains(key, "secret") || + strings.Contains(key, "token") +} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 67f962c..929bba0 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -50,6 +50,7 @@ type RunInput struct { Warnings []contracts.Warning Checkpoints CheckpointRecorder Checkpoint CheckpointLoader + Debug DebugRecorder } type RunOutput struct { @@ -81,10 +82,27 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if checkpointLoader == nil { checkpointLoader = NoopCheckpointLoader() } + debugRecorder := input.Debug + if debugRecorder == nil { + debugRecorder = NoopDebugRecorder() + } + input.Debug = debugRecorder + input.LLMClient = wrapDebugLLMClient(input.LLMClient, debugRecorder) defer func() { output.Manifest.LLMProfiles = mergeLLMProfileManifests(input.LLMProfiles, llmProfileManifests(input.LLMClient)) }() output.Warnings = append(output.Warnings, cloneWarnings(input.Warnings)...) + if err := writeDebugTimed(debugRecorder, "run.json", debugTimedEnvelope{ + Stage: "run", + StartedAt: startedTime(input.StartedAt), + Payload: map[string]any{ + "pipeline_id": input.Pipeline.ID, + "pipeline_digest": input.Pipeline.Digest, + "run_id": output.Manifest.RunID, + }, + }); err != nil { + return failOutput(output), fmt.Errorf("write debug run artifact: %w", err) + } adapter, err := r.registries.Inputs.Build(input.Pipeline.Input.Module) if err != nil { @@ -94,6 +112,21 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key()) recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision) doc := sourceCheckpoint.Document + sourceStarted := time.Now().UTC() + if err := writeDebugTimed(debugRecorder, "source/input.json", debugTimedEnvelope{ + Stage: "source", + ModuleKey: adapter.Key(), + StartedAt: sourceStarted, + Payload: debugSourceInput{ + SourceID: input.SourceID, + Path: input.Path, + Raw: debugContentEnvelope(input.RawInput, sourceInputMediaType(input.Path), nil, nil), + Options: redactSensitiveMap(input.Pipeline.Input.Options), + Metadata: redactSensitiveMap(input.Metadata), + }, + }); err != nil { + return failOutput(output), fmt.Errorf("write source debug artifact: %w", err) + } if !sourceDecision.Reused { if err := checkpoints.SourceRunning(adapter.Key()); err != nil { return failOutput(output), fmt.Errorf("write source checkpoint: %w", err) @@ -118,6 +151,18 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("write source checkpoint: %w", err) } } + if err := writeDebugTimed(debugRecorder, "source/output.json", debugTimedEnvelope{ + Stage: "source", + ModuleKey: adapter.Key(), + StartedAt: sourceStarted, + Payload: map[string]any{ + "reused": sourceDecision.Reused, + "decision": sourceDecision, + "document": debugSourceDocumentEnvelope(doc), + }, + }); err != nil { + return failOutput(output), fmt.Errorf("write source debug artifact: %w", err) + } sourceInput := sourceInputMaterial(input.Path, input.RawInput) sessionID := resolvedSessionID(input.SessionID, doc.ID) output.Manifest.Metadata = manifestMetadataWithSessionID(output.Manifest.Metadata, sessionID) @@ -132,6 +177,22 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err var chunkWarnings []contracts.Warning chunkCheckpoint, chunkDecision := checkpointLoader.Chunk(chunker.Key(), doc.Digest) recordCheckpointEvent(&output, checkpointLoader, string(StageChunk), "", chunker.Key(), chunkDecision) + chunkStarted := time.Now().UTC() + if err := writeDebugTimed(debugRecorder, "chunk/input.json", debugTimedEnvelope{ + Stage: string(StageChunk), + ModuleKey: chunker.Key(), + StartedAt: chunkStarted, + Payload: map[string]any{ + "reused": chunkDecision.Reused, + "decision": chunkDecision, + "source": debugSourceDocumentEnvelope(doc), + "source_input": debugContentEnvelope(sourceInput.Content, sourceInput.MediaType, nil, nil), + "options": redactSensitiveMap(input.Pipeline.Chunk.Options), + "metadata": redactSensitiveMap(input.Metadata), + }, + }); err != nil { + return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err) + } chunksAccepted := chunkDecision.Reused var chunkRejection *contracts.RejectedOutput if chunkDecision.Reused { @@ -143,6 +204,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", 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{ Source: doc, SourceInput: sourceInput.Clone(), @@ -154,6 +216,13 @@ 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{ + Stage: string(StageChunk), + ModuleKey: chunker.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Error: err.Error(), + }) return false, nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err) } if len(chunkResult.Chunks) == 0 { @@ -163,12 +232,35 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err if err != nil { return false, nil, fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), 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) + 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) if err != nil || rejection != nil { + _ = writeDebugTimed(debugRecorder, path.Join("chunk", fmt.Sprintf("attempt-%02d.json", attempt)), debugTimedEnvelope{ + Stage: string(StageChunk), + ModuleKey: chunker.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "chunks": debugSourceChunkEnvelopes(chunks), + "warnings": append(cloneWarnings(chunkResult.Warnings), validationWarnings...), + "rejection": debugRejectedOutputPtr(rejection), + }, + }) 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{ + Stage: string(StageChunk), + ModuleKey: chunker.Key(), + Attempt: attempt, + StartedAt: attemptStarted, + Payload: map[string]any{ + "chunks": debugSourceChunkEnvelopes(chunks), + "warnings": chunkWarnings, + }, + }); err != nil { + return false, nil, err + } return true, nil, nil }) if err != nil { @@ -187,6 +279,23 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } } } + chunkDebugPayload := map[string]any{ + "reused": chunkDecision.Reused, + "accepted": chunksAccepted, + "chunks": debugSourceChunkEnvelopes(canonicalChunks), + "warnings": chunkWarnings, + } + if chunkRejection != nil { + chunkDebugPayload["rejection"] = debugRejectedOutputEnvelope(*chunkRejection) + } + if err := writeDebugTimed(debugRecorder, "chunk/output.json", debugTimedEnvelope{ + Stage: string(StageChunk), + ModuleKey: chunker.Key(), + StartedAt: chunkStarted, + Payload: chunkDebugPayload, + }); err != nil { + return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err) + } if chunksAccepted { for _, lane := range input.Pipeline.ArtifactLanes { @@ -209,6 +318,22 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("build output encoder %q: %w", input.Pipeline.Output.Module, err) } attachModuleManifestMetadata(&output, "output", encoder) + outputStarted := time.Now().UTC() + if err := writeDebugTimed(debugRecorder, "output/input.json", debugTimedEnvelope{ + Stage: string(StageOutput), + ModuleKey: encoder.Key(), + StartedAt: outputStarted, + Payload: map[string]any{ + "manifest": output.Manifest, + "normalize_outputs": debugNormalizeOutputEnvelopes(output.NormalizeOutputs), + "rejected": debugRejectedOutputEnvelopes(output.Rejected), + "warnings": output.Warnings, + "options": redactSensitiveMap(input.Pipeline.Output.Options), + "metadata": redactSensitiveMap(input.Metadata), + }, + }); err != nil { + return failOutput(output), fmt.Errorf("write output debug artifact: %w", err) + } encoded, err := encoder.Encode(ctx, contracts.OutputRequest{ Manifest: output.Manifest, NormalizeOutputs: cloneNormalizeOutputs(output.NormalizeOutputs), @@ -227,6 +352,17 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), fmt.Errorf("validate output files from encoder %q: %w", encoder.Key(), err) } output.OutputFiles = files + if err := writeDebugTimed(debugRecorder, "output/output.json", debugTimedEnvelope{ + Stage: string(StageOutput), + ModuleKey: encoder.Key(), + StartedAt: outputStarted, + Payload: map[string]any{ + "files": debugOutputFiles(files), + "warnings": encoded.Warnings, + }, + }); err != nil { + return failOutput(output), fmt.Errorf("write output debug artifact: %w", err) + } return output, nil } @@ -252,6 +388,23 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp extractDependencies := digestFingerprints("chunks", joinedChunkDigest(chunks)) extractCheckpoint, extractDecision := checkpointLoader.Extract(lane.ID, extractor.Key(), extractDependencies) recordCheckpointEvent(output, checkpointLoader, string(StageExtract), lane.ID, extractor.Key(), extractDecision) + extractStarted := time.Now().UTC() + if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{ + Stage: string(StageExtract), + LaneID: lane.ID, + ModuleKey: extractor.Key(), + StartedAt: extractStarted, + Payload: map[string]any{ + "reused": extractDecision.Reused, + "decision": extractDecision, + "source": debugSourceDocumentEnvelope(doc), + "chunks": debugSourceChunkEnvelopes(chunks), + "options": redactSensitiveMap(lane.Extract.Options), + "metadata": redactSensitiveMap(input.Metadata), + }, + }); err != nil { + return fmt.Errorf("write extract debug artifact for lane %q: %w", lane.ID, err) + } if extractDecision.Reused { extractOutputs = cloneExtractOutputs(extractCheckpoint.Outputs) extractWarnings = cloneWarnings(extractCheckpoint.Warnings) @@ -305,6 +458,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp metadata: input.Metadata, chains: input.Pipeline.ValidatorChains, attempt: attempt, + debug: input.Debug, }) if err != nil || rejection != nil { return false, rejection, err @@ -330,6 +484,20 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err) } } + if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{ + Stage: string(StageExtract), + LaneID: lane.ID, + ModuleKey: extractor.Key(), + StartedAt: extractStarted, + Payload: map[string]any{ + "reused": extractDecision.Reused, + "outputs": debugExtractOutputEnvelopes(extractOutputs), + "rejected": debugRejectedOutputEnvelopes(output.Rejected[extractRejectedStart:]), + "warnings": extractWarnings, + }, + }); err != nil { + return fmt.Errorf("write extract debug artifact for lane %q: %w", lane.ID, err) + } if len(extractOutputs) == 0 { return nil @@ -340,6 +508,23 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp mergeDependencies := rawOutputDigests(extractPayloads(extractOutputs)) mergeCheckpoint, mergeDecision := checkpointLoader.Merge(lane.ID, merger.Key(), mergeDependencies) recordCheckpointEvent(output, checkpointLoader, string(StageMerge), lane.ID, merger.Key(), mergeDecision) + mergeStarted := time.Now().UTC() + if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{ + Stage: string(StageMerge), + LaneID: lane.ID, + ModuleKey: merger.Key(), + StartedAt: mergeStarted, + Payload: map[string]any{ + "reused": mergeDecision.Reused, + "decision": mergeDecision, + "source": debugSourceDocumentEnvelope(doc), + "extract_outputs": debugExtractOutputEnvelopes(extractOutputs), + "options": redactSensitiveMap(lane.Merge.Options), + "metadata": redactSensitiveMap(input.Metadata), + }, + }); err != nil { + return fmt.Errorf("write merge debug artifact for lane %q: %w", lane.ID, err) + } if mergeDecision.Reused { acceptedMerge = cloneMergeOutput(mergeCheckpoint.Output) mergeWarnings = cloneWarnings(mergeCheckpoint.Warnings) @@ -385,6 +570,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp metadata: input.Metadata, chains: input.Pipeline.ValidatorChains, attempt: attempt, + debug: input.Debug, }) if err != nil || rejection != nil { return false, rejection, err @@ -402,6 +588,19 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp if err := checkpoints.MergeRejected(lane.ID, merger.Key(), mergeDependencies, *mergeRejection); err != nil { return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err) } + if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{ + Stage: string(StageMerge), + LaneID: lane.ID, + ModuleKey: merger.Key(), + StartedAt: mergeStarted, + Payload: map[string]any{ + "accepted": false, + "rejection": debugRejectedOutputEnvelope(*mergeRejection), + "warnings": mergeWarnings, + }, + }); err != nil { + return fmt.Errorf("write merge debug artifact for lane %q: %w", lane.ID, err) + } return nil } output.Warnings = append(output.Warnings, mergeWarnings...) @@ -409,12 +608,43 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp return fmt.Errorf("write merge checkpoint for lane %q: %w", lane.ID, err) } } + if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{ + Stage: string(StageMerge), + LaneID: lane.ID, + ModuleKey: merger.Key(), + StartedAt: mergeStarted, + Payload: map[string]any{ + "reused": mergeDecision.Reused, + "accepted": true, + "output": debugMergeOutputEnvelope(acceptedMerge), + "warnings": mergeWarnings, + }, + }); err != nil { + return fmt.Errorf("write merge debug artifact for lane %q: %w", lane.ID, err) + } var acceptedNormalize contracts.NormalizeOutput var normalizeWarnings []contracts.Warning normalizeDependencies := rawOutputDigests([]contracts.RawPayload{acceptedMerge.Payload}) normalizeCheckpoint, normalizeDecision := checkpointLoader.Normalize(lane.ID, normalizer.Key(), normalizeDependencies) recordCheckpointEvent(output, checkpointLoader, string(StageNormalize), lane.ID, normalizer.Key(), normalizeDecision) + normalizeStarted := time.Now().UTC() + if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{ + Stage: string(StageNormalize), + LaneID: lane.ID, + ModuleKey: normalizer.Key(), + StartedAt: normalizeStarted, + Payload: map[string]any{ + "reused": normalizeDecision.Reused, + "decision": normalizeDecision, + "source": debugSourceDocumentEnvelope(doc), + "merge_output": debugMergeOutputEnvelope(acceptedMerge), + "options": redactSensitiveMap(lane.Normalize.Options), + "metadata": redactSensitiveMap(input.Metadata), + }, + }); err != nil { + return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err) + } if normalizeDecision.Reused { acceptedNormalize = cloneNormalizeOutput(normalizeCheckpoint.Output) normalizeWarnings = cloneWarnings(normalizeCheckpoint.Warnings) @@ -460,6 +690,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp metadata: input.Metadata, chains: input.Pipeline.ValidatorChains, attempt: attempt, + debug: input.Debug, }) if err != nil || rejection != nil { return false, rejection, err @@ -477,6 +708,19 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp if err := checkpoints.NormalizeRejected(lane.ID, normalizer.Key(), normalizeDependencies, *normalizeRejection); err != nil { return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err) } + if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{ + Stage: string(StageNormalize), + LaneID: lane.ID, + ModuleKey: normalizer.Key(), + StartedAt: normalizeStarted, + Payload: map[string]any{ + "accepted": false, + "rejection": debugRejectedOutputEnvelope(*normalizeRejection), + "warnings": normalizeWarnings, + }, + }); err != nil { + return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err) + } return nil } output.Warnings = append(output.Warnings, normalizeWarnings...) @@ -484,6 +728,20 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, checkpoints Checkp return fmt.Errorf("write normalize checkpoint for lane %q: %w", lane.ID, err) } } + if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{ + Stage: string(StageNormalize), + LaneID: lane.ID, + ModuleKey: normalizer.Key(), + StartedAt: normalizeStarted, + Payload: map[string]any{ + "reused": normalizeDecision.Reused, + "accepted": true, + "output": debugNormalizeOutputEnvelope(acceptedNormalize), + "warnings": normalizeWarnings, + }, + }); err != nil { + return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err) + } output.NormalizeOutputs = append(output.NormalizeOutputs, acceptedNormalize) return nil } @@ -509,6 +767,7 @@ type rawValidationTarget struct { metadata map[string]any chains []ResolvedValidatorChain attempt int + debug DebugRecorder } func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, *contracts.RejectedOutput, error)) (bool, *contracts.RejectedOutput, error) { @@ -558,7 +817,7 @@ func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, return false, lastRejection, nil } -func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []contracts.SourceChunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, llmClient contracts.StructuredLLMClient, metadata map[string]any, chains []ResolvedValidatorChain, attempt int) ([]contracts.Warning, *contracts.RejectedOutput, error) { +func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []contracts.SourceChunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, llmClient contracts.StructuredLLMClient, metadata map[string]any, chains []ResolvedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) { return r.validateRaw(ctx, rawValidationTarget{ stage: StageChunk, moduleKey: moduleKey, @@ -572,6 +831,7 @@ func (r *Runner) validateChunksRaw(ctx context.Context, doc *source.SourceDocume metadata: metadata, chains: chains, attempt: attempt, + debug: debug, }) } @@ -591,7 +851,27 @@ func (r *Runner) validateRaw(ctx context.Context, target rawValidationTarget) ([ 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) + debugPayload := debugValidationCall{ + ValidatorName: validator.Name(), + Request: debugValidationRequestEnvelope(request), + Result: debugValidationResultEnvelope(result), + } + 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{ + Stage: string(target.stage), + LaneID: target.laneID, + ModuleKey: target.moduleKey, + Attempt: target.attempt, + StartedAt: started, + Payload: debugPayload, + Error: debugPayload.Error, + }); debugErr != nil { + return nil, nil, fmt.Errorf("write validation debug artifact: %w", debugErr) + } if err != nil { return nil, nil, fmt.Errorf("validate raw %s output with validator %q: %w", target.stage, validator.Name(), err) } @@ -733,6 +1013,13 @@ func validateRunInput(input RunInput) error { return nil } +func startedTime(t time.Time) time.Time { + if t.IsZero() { + return time.Now().UTC() + } + return t.UTC() +} + func manifestFromPipeline(input RunInput) artifacts.RunManifest { startedAt := input.StartedAt if startedAt.IsZero() {