Harden output cache and debug state integration

docs/roadmap/implementation.md

@@ -521,7 +521,7 @@ Internal Boundaries; and all checks pass.
 
 ## Stage 5: Harden cross-surface security and compatibility
 
-**Status:** Not started
+**Status:** Complete
 
 ### Objective
 

internal/cli/run.go

@@ -41,6 +41,7 @@ type Options struct {
 	Now                   func() time.Time
 	UserCacheDir          func() (string, error)
 	ChunkPlanStoreFactory pipeline.ChunkPlanStoreFactory
+	DebugRecorderFactory  func(string) (pipeline.DebugRecorder, error)
 }
 
 type LLMClientFactory func(ctx context.Context, cfg config.Config, profileID string) (contracts.StructuredLLMClient, []artifacts.LLMProfileManifest, error)
@@ -96,6 +97,9 @@ func normalizeOptions(opts Options) (Options, error) {
 	if opts.ChunkPlanStoreFactory == nil {
 		opts.ChunkPlanStoreFactory = chunkplan.NewFilesystemStore
 	}
+	if opts.DebugRecorderFactory == nil {
+		opts.DebugRecorderFactory = frameworkdebug.NewFilesystemRecorder
+	}
 	if isEmptyCatalog(opts.Catalog) && isEmptyRegistries(opts.Registries) {
 		components, err := newProductionComponents()
 		if err != nil {
@@ -220,7 +224,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
 			return 1
 		}
 		runID, debugPath, summary = bundle.RunID(), bundle.Path(), bundle.Summary()
-		debugRecorder, err = frameworkdebug.NewFilesystemRecorder(bundle.TraceRoot())
+		debugRecorder, err = opts.DebugRecorderFactory(bundle.TraceRoot())
 		if err != nil {
 			return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("create debug recorder: %w", err), true)
 		}

internal/cli/state_hardening_test.go

@@ -0,0 +1,528 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
+	"gitea.maximumdirect.net/eric/notarius/internal/core/config"
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/chunkplan"
+	"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"
+)
+
+const stateTestDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+
+func TestRunStateSurfaceMatrix(t *testing.T) {
+	for _, debug := range []bool{false, true} {
+		for _, resume := range []bool{false, true} {
+			for _, mode := range []string{"auto", "bypass", "refresh"} {
+				name := fmt.Sprintf("debug=%t/resume=%t/cache=%s", debug, resume, mode)
+				t.Run(name, func(t *testing.T) {
+					roots := newStateTestRoots(t)
+					harness := newStateTestHarness()
+					opts := harness.options()
+					var storeRoots []string
+					opts.ChunkPlanStoreFactory = func(root string) (pipeline.ChunkPlanStore, error) {
+						storeRoots = append(storeRoots, root)
+						return chunkplan.NewFilesystemStore(root)
+					}
+					result := runStateTest(t, roots, opts, debug, resume, mode)
+					if result.code != 0 {
+						t.Fatalf("code=%d stderr=%q", result.code, result.stderr)
+					}
+					assertStateTestOutput(t, roots.output)
+					if mode == "bypass" {
+						assertAbsent(t, roots.plans)
+						if len(storeRoots) != 0 {
+							t.Fatalf("chunk plan store roots = %v, want none", storeRoots)
+						}
+					} else {
+						assertFile(t, filepath.Join(roots.plans, strings.TrimPrefix(stateTestDigest, "sha256:"), "plan.json"))
+						if len(storeRoots) != 1 || storeRoots[0] != roots.plans {
+							t.Fatalf("chunk plan store roots = %v, want [%q]", storeRoots, roots.plans)
+						}
+					}
+					if resume {
+						assertAnyFile(t, roots.checkpoints)
+						assertRestrictedTree(t, roots.checkpoints)
+					} else {
+						assertAbsent(t, roots.checkpoints)
+					}
+					if debug {
+						bundle := onlyChildDir(t, roots.debug)
+						assertFile(t, filepath.Join(bundle, "summary", "invocation.json"))
+						assertAnyFile(t, filepath.Join(bundle, "trace"))
+						assertRestrictedTree(t, roots.debug)
+					} else {
+						assertAbsent(t, roots.debug)
+					}
+				})
+			}
+		}
+	}
+}
+
+func TestRunKeepsStateRootsIndependentAndReusesSelectedCheckpointRoot(t *testing.T) {
+	roots := newStateTestRoots(t)
+	harness := newStateTestHarness()
+	first := runStateTest(t, roots, harness.options(), true, false, "auto")
+	if first.code != 0 {
+		t.Fatalf("first run code=%d stderr=%q", first.code, first.stderr)
+	}
+	planPath := filepath.Join(roots.plans, strings.TrimPrefix(stateTestDigest, "sha256:"), "plan.json")
+	initialPlan, err := os.ReadFile(planPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	firstBundle := onlyChildDir(t, roots.debug)
+
+	second := runStateTest(t, roots, harness.options(), false, false, "auto")
+	if second.code != 0 {
+		t.Fatalf("second run code=%d stderr=%q", second.code, second.stderr)
+	}
+	if harness.chunkCalls != 1 {
+		t.Fatalf("chunk calls after debug toggle = %d, want 1", harness.chunkCalls)
+	}
+	if got, err := os.ReadFile(planPath); err != nil || !bytes.Equal(got, initialPlan) {
+		t.Fatalf("chunk plan changed after debug toggle: %v", err)
+	}
+	if _, err := os.Stat(firstBundle); err != nil {
+		t.Fatalf("initial debug bundle was removed: %v", err)
+	}
+
+	checkpointRoot := roots.checkpoints
+	seed := runStateTest(t, roots, harness.options(), false, true, "auto")
+	if seed.code != 0 {
+		t.Fatalf("checkpoint seed code=%d stderr=%q", seed.code, seed.stderr)
+	}
+	extractCalls := harness.extractCalls
+	checkpointFiles := readTree(t, checkpointRoot)
+	reused := runStateTest(t, roots, harness.options(), false, true, "auto")
+	if reused.code != 0 {
+		t.Fatalf("checkpoint reuse code=%d stderr=%q", reused.code, reused.stderr)
+	}
+	if harness.extractCalls != extractCalls {
+		t.Fatalf("extract calls after checkpoint reuse = %d, want %d", harness.extractCalls, extractCalls)
+	}
+	if got := readTree(t, checkpointRoot); !sameFiles(got, checkpointFiles) {
+		t.Fatal("reused checkpoint was rewritten")
+	}
+}
+
+func TestRunRecomputesOnlyAfterExplicitChunkPlanRemoval(t *testing.T) {
+	roots := newStateTestRoots(t)
+	harness := newStateTestHarness()
+	first := runStateTest(t, roots, harness.options(), true, false, "auto")
+	if first.code != 0 {
+		t.Fatalf("first run code=%d stderr=%q", first.code, first.stderr)
+	}
+	firstOutput := onlyChildDir(t, roots.output)
+	firstBundle := onlyChildDir(t, roots.debug)
+	entry := filepath.Join(roots.plans, strings.TrimPrefix(stateTestDigest, "sha256:"))
+	if err := os.RemoveAll(entry); err != nil {
+		t.Fatal(err)
+	}
+	second := runStateTest(t, roots, harness.options(), false, false, "auto")
+	if second.code != 0 {
+		t.Fatalf("second run code=%d stderr=%q", second.code, second.stderr)
+	}
+	if harness.chunkCalls != 2 {
+		t.Fatalf("chunk calls = %d, want 2 after removing exact cache entry", harness.chunkCalls)
+	}
+	assertFile(t, filepath.Join(firstOutput, "result.json"))
+	assertFile(t, filepath.Join(firstBundle, "summary", "run-report.json"))
+}
+
+func TestRunRetainsDebugBundlesAcrossFailures(t *testing.T) {
+	t.Run("configuration failure precedes allocation", func(t *testing.T) {
+		root := filepath.Join(t.TempDir(), "debug")
+		var stdout, stderr bytes.Buffer
+		code := RunWithOptions([]string{"run", "sample", "--config", filepath.Join(t.TempDir(), "missing.yml"), "--input", "missing", "--debug", "--debug-dir", root}, &stdout, &stderr, newStateTestHarness().options())
+		if code != 1 || !strings.Contains(stderr.String(), "config file") {
+			t.Fatalf("code=%d stderr=%q", code, stderr.String())
+		}
+		assertAbsent(t, root)
+	})
+
+	for _, failure := range []struct {
+		name     string
+		expected string
+		setup    func(*testing.T, stateTestRoots, *stateTestHarness) Options
+	}{
+		{"resolution", "pipeline \"missing\"", func(t *testing.T, roots stateTestRoots, h *stateTestHarness) Options { return h.options() }},
+		{"pipeline", "synthetic extraction failure", func(t *testing.T, roots stateTestRoots, h *stateTestHarness) Options {
+			h.extractErr = errors.New("synthetic extraction failure")
+			return h.options()
+		}},
+		{"output", "create output directory", func(t *testing.T, roots stateTestRoots, h *stateTestHarness) Options {
+			if err := os.WriteFile(roots.output, []byte("not a directory"), 0o600); err != nil {
+				t.Fatal(err)
+			}
+			return h.options()
+		}},
+		{"summary", "write debug invocation metadata", func(t *testing.T, roots stateTestRoots, h *stateTestHarness) Options {
+			opts := h.options()
+			opts.DebugRecorderFactory = func(traceRoot string) (pipeline.DebugRecorder, error) {
+				if err := os.RemoveAll(filepath.Join(filepath.Dir(traceRoot), "summary")); err != nil {
+					return nil, err
+				}
+				if err := os.WriteFile(filepath.Join(filepath.Dir(traceRoot), "summary"), []byte("blocked"), 0o600); err != nil {
+					return nil, err
+				}
+				return frameworkdebug.NewFilesystemRecorder(traceRoot)
+			}
+			return opts
+		}},
+		{"trace", "trace unavailable", func(t *testing.T, roots stateTestRoots, h *stateTestHarness) Options {
+			opts := h.options()
+			opts.DebugRecorderFactory = func(string) (pipeline.DebugRecorder, error) { return failingDebugRecorder{}, nil }
+			return opts
+		}},
+	} {
+		t.Run(failure.name, func(t *testing.T) {
+			roots := newStateTestRoots(t)
+			harness := newStateTestHarness()
+			opts := failure.setup(t, roots, harness)
+			failureStderr := ""
+			if failure.name == "resolution" {
+				var stdout, stderr bytes.Buffer
+				code := RunWithOptions([]string{"run", "missing", "--config", roots.config, "--input", roots.input, "--debug"}, &stdout, &stderr, opts)
+				if code != 1 {
+					t.Fatalf("code=%d stderr=%q", code, stderr.String())
+				}
+				failureStderr = stderr.String()
+			} else {
+				result := runStateTest(t, roots, opts, true, false, "bypass")
+				if result.code != 1 {
+					t.Fatalf("code=%d stderr=%q", result.code, result.stderr)
+				}
+				failureStderr = result.stderr
+			}
+			if !strings.Contains(failureStderr, failure.expected) || !strings.Contains(failureStderr, "debug=") {
+				t.Fatalf("stderr=%q, want %q and debug path", failureStderr, failure.expected)
+			}
+			bundle := onlyChildDir(t, roots.debug)
+			if !strings.Contains(readAllFiles(t, bundle), "synthetic") && failure.name == "pipeline" {
+				t.Fatal("pipeline failure was not retained in debug bundle")
+			}
+		})
+	}
+}
+
+func TestRunDebugArtifactsRedactSecretsButRetainApplicationData(t *testing.T) {
+	roots := newStateTestRoots(t)
+	t.Setenv("STATE_TEST_UNRELATED_ENV", "HOST_ONLY_SENTINEL")
+	if err := os.WriteFile(filepath.Join(filepath.Dir(roots.input), "unrelated.txt"), []byte("HOST_ONLY_FILE_SENTINEL"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	harness := newStateTestHarness()
+	result := runStateTest(t, roots, harness.options(), true, false, "bypass")
+	if result.code != 0 {
+		t.Fatalf("code=%d stderr=%q", result.code, result.stderr)
+	}
+	bundle := onlyChildDir(t, roots.debug)
+	summary := readAllFiles(t, filepath.Join(bundle, "summary"))
+	trace := readAllFiles(t, filepath.Join(bundle, "trace"))
+	for _, forbidden := range []string{"sk-secretvalue", "Bearer secretvalue", "HOST_ONLY_SENTINEL", "HOST_ONLY_FILE_SENTINEL"} {
+		if strings.Contains(summary, forbidden) || strings.Contains(trace, forbidden) {
+			t.Fatalf("debug bundle contains %q", forbidden)
+		}
+	}
+	if strings.Contains(summary, "application content") {
+		t.Fatal("summary contains raw application input")
+	}
+	if !strings.Contains(trace, "application content") {
+		t.Fatal("trace does not retain expected application input")
+	}
+}
+
+type stateTestRoots struct{ config, input, output, plans, checkpoints, debug string }
+
+func newStateTestRoots(t *testing.T) stateTestRoots {
+	t.Helper()
+	base := t.TempDir()
+	roots := stateTestRoots{input: filepath.Join(base, "input.txt"), output: filepath.Join(base, "output"), plans: filepath.Join(base, "plans"), checkpoints: filepath.Join(base, "checkpoints"), debug: filepath.Join(base, "debug")}
+	if err := os.WriteFile(roots.input, []byte("application content Bearer secretvalue sk-secretvalue"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	roots.config = filepath.Join(base, "config.yml")
+	config := fmt.Sprintf("version: 3\noutput:\n  directory: %q\ncache:\n  chunk_plans:\n    directory: %q\n    mode: auto\n  checkpoints:\n    directory: %q\ndebug:\n  directory: %q\npipelines:\n  sample:\n    input: test/input\n    chunk: test/chunk\n    artifacts:\n      items:\n        extract: test/extract\n        merge: test/merge\n        normalize: test/normalize\n    output: test/output\n", roots.output, roots.plans, roots.checkpoints, roots.debug)
+	if err := os.WriteFile(roots.config, []byte(config), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return roots
+}
+
+type stateTestResult struct {
+	code           int
+	stdout, stderr string
+}
+
+func runStateTest(t *testing.T, roots stateTestRoots, opts Options, debug, resume bool, mode string) stateTestResult {
+	t.Helper()
+	args := []string{"run", "sample", "--config", roots.config, "--input", roots.input, "--chunk_cache", mode}
+	if debug {
+		args = append(args, "--debug")
+	}
+	if resume {
+		args = append(args, "--resume")
+	}
+	var stdout, stderr bytes.Buffer
+	return stateTestResult{RunWithOptions(args, &stdout, &stderr, opts), stdout.String(), stderr.String()}
+}
+
+func assertStateTestOutput(t *testing.T, root string) {
+	t.Helper()
+	output := onlyChildDir(t, root)
+	data, err := os.ReadFile(filepath.Join(output, "result.json"))
+	if err != nil || string(data) != "{\"ok\":true}\n" {
+		t.Fatalf("output = %q, %v", data, err)
+	}
+}
+
+func onlyChildDir(t *testing.T, root string) string {
+	t.Helper()
+	entries, err := os.ReadDir(root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var dirs []string
+	for _, entry := range entries {
+		if entry.IsDir() {
+			dirs = append(dirs, filepath.Join(root, entry.Name()))
+		}
+	}
+	if len(dirs) != 1 {
+		t.Fatalf("directories in %q = %v, want one", root, dirs)
+	}
+	return dirs[0]
+}
+
+func assertFile(t *testing.T, path string) {
+	t.Helper()
+	if info, err := os.Stat(path); err != nil || info.IsDir() {
+		t.Fatalf("file %q: %v", path, err)
+	}
+}
+func assertAbsent(t *testing.T, path string) {
+	t.Helper()
+	if _, err := os.Stat(path); !os.IsNotExist(err) {
+		t.Fatalf("%q exists or stat failed: %v", path, err)
+	}
+}
+func assertAnyFile(t *testing.T, root string) {
+	t.Helper()
+	if text := readAllFiles(t, root); text == "" {
+		t.Fatalf("no files under %q", root)
+	}
+}
+
+func readAllFiles(t *testing.T, root string) string {
+	t.Helper()
+	var content strings.Builder
+	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
+		}
+		content.Write(data)
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	return content.String()
+}
+
+func assertRestrictedTree(t *testing.T, root string) {
+	t.Helper()
+	if runtime.GOOS == "windows" {
+		return
+	}
+	if err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		want := os.FileMode(0o600)
+		if info.IsDir() {
+			want = 0o700
+		}
+		if info.Mode().Perm() != want {
+			return fmt.Errorf("%s has mode %o, want %o", path, info.Mode().Perm(), want)
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func readTree(t *testing.T, root string) map[string][]byte {
+	t.Helper()
+	files := map[string][]byte{}
+	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
+		}
+		relative, err := filepath.Rel(root, path)
+		if err != nil {
+			return err
+		}
+		files[relative] = data
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	return files
+}
+func sameFiles(left, right map[string][]byte) bool {
+	if len(left) != len(right) {
+		return false
+	}
+	for path, data := range left {
+		if !bytes.Equal(data, right[path]) {
+			return false
+		}
+	}
+	return true
+}
+
+type stateTestHarness struct {
+	mu                       sync.Mutex
+	chunkCalls, extractCalls int
+	extractErr               error
+}
+
+func newStateTestHarness() *stateTestHarness { return &stateTestHarness{} }
+func (h *stateTestHarness) options() Options {
+	registries := pipeline.Registries{Inputs: pipeline.NewInputAdapterRegistry(), Chunkers: pipeline.NewChunkerRegistry(), ArtifactCodecs: pipeline.NewArtifactCodecRegistry(), Extractors: pipeline.NewExtractorRegistry(), Mergers: pipeline.NewMergerRegistry(), Normalizers: pipeline.NewNormalizerRegistry(), Outputs: pipeline.NewOutputEncoderRegistry()}
+	if err := pipeline.RegisterArtifactCodec(registries.ArtifactCodecs, stateTestCodec{}); err != nil {
+		panic(err)
+	}
+	if err := registries.Inputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/input", Stage: pipeline.StageInput, Provides: []string{"source"}}, func() (contracts.InputAdapter, error) { return stateTestInput{}, nil }); err != nil {
+		panic(err)
+	}
+	if err := registries.Chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/chunk", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}}, func() (contracts.Chunker, error) { return stateTestChunker{h}, nil }); err != nil {
+		panic(err)
+	}
+	if err := pipeline.RegisterExtractor(registries.Extractors, pipeline.ModuleSpec{Key: "test/extract", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Extractor[stateTestArtifact], error) { return stateTestExtractor{h}, nil }); err != nil {
+		panic(err)
+	}
+	if err := pipeline.RegisterMerger(registries.Mergers, pipeline.ModuleSpec{Key: "test/merge", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Merger[stateTestArtifact], error) { return stateTestMerger{}, nil }); err != nil {
+		panic(err)
+	}
+	if err := pipeline.RegisterNormalizer(registries.Normalizers, pipeline.ModuleSpec{Key: "test/normalize", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: stateTestArtifactKind}, func() (contracts.Normalizer[stateTestArtifact], error) { return stateTestNormalizer{}, nil }); err != nil {
+		panic(err)
+	}
+	if err := registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/output", Stage: pipeline.StageOutput, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) { return stateTestOutput{}, nil }); err != nil {
+		panic(err)
+	}
+	return Options{Catalog: catalogFromRegistries(registries), Registries: registries, LookupEnv: emptyLookup, Now: func() time.Time { return time.Unix(1, 0) }, UserCacheDir: func() (string, error) { return "", errors.New("unexpected user cache lookup") }, LLMClientFactory: func(context.Context, config.Config, string) (contracts.StructuredLLMClient, []artifacts.LLMProfileManifest, error) {
+		return nil, nil, nil
+	}}
+}
+
+type stateTestInput struct{}
+
+func (stateTestInput) Key() string { return "test/input" }
+func (stateTestInput) Parse(_ context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
+	return &source.SourceDocument{ID: "source", Kind: "text", Format: "text/plain", Digest: stateTestDigest, Units: []source.SourceUnit{{ID: 1, Kind: "text", Text: string(req.Raw), Ref: source.SourceRef{SourceID: "source", StartUnitID: 1, EndUnitID: 1}}}}, nil
+}
+
+type stateTestChunker struct{ harness *stateTestHarness }
+
+func (stateTestChunker) Key() string                               { return "test/chunk" }
+func (stateTestChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+func (c stateTestChunker) Plan(_ context.Context, req contracts.ChunkRequest) (contracts.ChunkPlanResult, error) {
+	c.harness.mu.Lock()
+	c.harness.chunkCalls++
+	c.harness.mu.Unlock()
+	return contracts.ChunkPlanResult{Plan: source.ChunkPlan{SourceDigest: req.Source.Digest, Ranges: []source.ChunkRange{{StartUnitID: 1, EndUnitID: 1}}}}, nil
+}
+
+const stateTestArtifactKind contracts.ArtifactKind = "test/artifact"
+
+type stateTestArtifact struct {
+	Value string `json:"value"`
+}
+type stateTestCodec struct{}
+
+func (stateTestCodec) Kind() contracts.ArtifactKind { return stateTestArtifactKind }
+func (stateTestCodec) Schema() contracts.ArtifactSchema {
+	return contracts.ArtifactSchema{ID: "test.artifact", Name: "test_artifact", Version: "v1", JSONSchema: []byte(`{"type":"object"}`)}
+}
+func (stateTestCodec) MediaType() string { return "application/json" }
+func (stateTestCodec) EncodeCandidate(v stateTestArtifact) ([]byte, error) {
+	return []byte(`{"value":"ok"}`), nil
+}
+func (stateTestCodec) Encode(v stateTestArtifact) ([]byte, error) {
+	return []byte(`{"value":"ok"}`), nil
+}
+func (stateTestCodec) Decode([]byte) (stateTestArtifact, error) {
+	return stateTestArtifact{Value: "ok"}, nil
+}
+
+type stateTestExtractor struct{ harness *stateTestHarness }
+
+func (stateTestExtractor) Key() string                               { return "test/extract" }
+func (stateTestExtractor) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+func (e stateTestExtractor) Extract(context.Context, contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[stateTestArtifact], error) {
+	e.harness.mu.Lock()
+	defer e.harness.mu.Unlock()
+	e.harness.extractCalls++
+	if e.harness.extractErr != nil {
+		return contracts.TypedExtractionResult[stateTestArtifact]{}, e.harness.extractErr
+	}
+	return contracts.TypedExtractionResult[stateTestArtifact]{Value: stateTestArtifact{Value: "ok"}}, nil
+}
+
+type stateTestMerger struct{}
+
+func (stateTestMerger) Key() string { return "test/merge" }
+func (stateTestMerger) Merge(_ context.Context, req contracts.TypedMergeRequest[stateTestArtifact]) (contracts.TypedMergeResult[stateTestArtifact], error) {
+	return contracts.TypedMergeResult[stateTestArtifact]{Value: req.ExtractOutputs[0].Value}, nil
+}
+
+type stateTestNormalizer struct{}
+
+func (stateTestNormalizer) Key() string                               { return "test/normalize" }
+func (stateTestNormalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+func (stateTestNormalizer) Normalize(_ context.Context, req contracts.TypedNormalizeRequest[stateTestArtifact]) (contracts.TypedNormalizeResult[stateTestArtifact], error) {
+	return contracts.TypedNormalizeResult[stateTestArtifact]{Value: req.MergeOutput.Value}, nil
+}
+
+type stateTestOutput struct{}
+
+func (stateTestOutput) Key() string { return "test/output" }
+func (stateTestOutput) Encode(context.Context, contracts.OutputRequest) (contracts.OutputResult, error) {
+	return contracts.OutputResult{Files: []contracts.OutputFile{{Name: "result.json", Bytes: []byte("{\"ok\":true}\n")}}}, nil
+}
+
+type failingDebugRecorder struct{}
+
+func (failingDebugRecorder) Enabled() bool                   { return true }
+func (failingDebugRecorder) WriteJSON(string, any) error     { return errors.New("trace unavailable") }
+func (failingDebugRecorder) WriteBytes(string, []byte) error { return errors.New("trace unavailable") }

internal/framework/pipeline/debug.go

@@ -491,10 +491,19 @@ func debugSourceChunkEnvelope(chunk source.Chunk) debugSourceChunk {
 }
 
 func cloneSourceUnitsForDebug(units []source.SourceUnit) []source.SourceUnit {
-	cloned, err := cloneSourceUnits(units)
-	if err != nil {
+	if len(units) == 0 {
 		return nil
 	}
+	cloned := make([]source.SourceUnit, len(units))
+	for i, unit := range units {
+		cloned[i] = source.SourceUnit{
+			ID:       unit.ID,
+			Kind:     unit.Kind,
+			Text:     string(redactSecretBytes([]byte(unit.Text))),
+			Ref:      unit.Ref,
+			Metadata: redactSensitiveMap(unit.Metadata),
+		}
+	}
 	return cloned
 }
 

internal/framework/pipeline/debug_test.go

@@ -2,6 +2,7 @@ package pipeline
 
 import (
 	"encoding/json"
+	"strings"
 	"testing"
 
 	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
@@ -29,6 +30,35 @@ func TestDebugSourceDocumentPreservesUnitReferences(t *testing.T) {
 	}
 }
 
+func TestDebugSourceUnitsRedactSecrets(t *testing.T) {
+	units := []source.SourceUnit{{
+		ID:       1,
+		Kind:     "paragraph",
+		Text:     "application text Bearer secretvalue sk-secretvalue",
+		Ref:      source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1},
+		Metadata: map[string]any{"api_key": "sk-secretvalue"},
+	}}
+
+	got := cloneSourceUnitsForDebug(units)
+	if len(got) != 1 {
+		t.Fatalf("debug unit count = %d, want 1", len(got))
+	}
+	if !strings.Contains(got[0].Text, "application text") {
+		t.Fatalf("debug unit text = %q, want application content retained", got[0].Text)
+	}
+	for _, forbidden := range []string{"secretvalue", "sk-secretvalue"} {
+		if strings.Contains(got[0].Text, forbidden) {
+			t.Fatalf("debug unit text contains %q: %q", forbidden, got[0].Text)
+		}
+	}
+	if got, want := got[0].Metadata["api_key"], "[REDACTED]"; got != want {
+		t.Fatalf("debug unit metadata api_key = %#v, want %q", got, want)
+	}
+	if units[0].Text != "application text Bearer secretvalue sk-secretvalue" {
+		t.Fatalf("source unit text was mutated: %q", units[0].Text)
+	}
+}
+
 func TestDebugSourceChunkPreservesReference(t *testing.T) {
 	doc := validSourceDocument()
 	chunk := source.Chunk{