fix(watcher): unique archive copies, read caps at the read, stale markers (review of #210)

From the independent review of #210:
- F1: processed/ and failed/ copies are created O_EXCL with -2,-3 suffixes via
  createUnique; same-named notes from different folders no longer replace one
  another's only surviving copy
- F2: extract.TextInMax / readAllMax cap the bytes actually read (a note can
  grow after the Lstat); extract.TextIn gets a 64 MiB default cap
- F3: a regular marker older than its note describes an earlier note of the
  same name; it no longer hides the new one (warn + remove); non-regular
  markers still count as done (no writes through links)
- F4: the planted-marker-link test is renamed to what it observes
Each has a test that fails without it; the read-cap test uses an endless reader
that errors at 8 MiB so a regression fails instead of exhausting memory.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

ingestion/internal/watcher/park.go

@@ -20,6 +20,13 @@ import (
 // root, never replacing a parked note: on a name clash it tries -2, -3, ...
 // A needs-review that is a symlink, or any non-directory, is refused.
 func createParked(brainDir, filename string, data []byte) error {
+	return createUnique(brainDir, "raw/needs-review", filename, data)
+}
+
+// createUnique writes data to dirRel/filename through the brain root, never
+// replacing an existing file: on a name clash it tries -2, -3, ... A dirRel that
+// is a symlink, or any non-directory, is refused.
+func createUnique(brainDir, dirRel, filename string, data []byte) error {
 	bfs, err := brainfs.Open(brainDir)
 	if err != nil {
 		return err
@@ -32,7 +39,7 @@ func createParked(brainDir, filename string, data []byte) error {
 		if n > 1 {
 			name = fmt.Sprintf("%s-%d%s", stem, n, ext)
 		}
-		err := bfs.CreateRegular("raw/needs-review/"+name, data)
+		err := bfs.CreateRegular(dirRel+"/"+name, data)
 		if err == nil {
 			return nil
 		}
@@ -40,5 +47,5 @@ func createParked(brainDir, filename string, data []byte) error {
 			return err
 		}
 	}
-	return fmt.Errorf("no free name for %q in raw/needs-review", filename)
+	return fmt.Errorf("no free name for %q in %s", filename, dirRel)
 }

ingestion/internal/watcher/watcher.go

@@ -43,7 +43,7 @@ const defaultMaxNoteBytes = 64 << 20
 
 // errTooLarge marks a note refused for its size; like a refused write it is
 // reported once and not retried until the note changes.
-var errTooLarge = errors.New("note too large")
+var errTooLarge = extract.ErrTooLarge
 
 // Start launches the watcher in a background goroutine.
 // It returns immediately. The watcher stops when ctx is cancelled.
@@ -134,11 +134,11 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
 		// Any entry at the marker path counts (Lstat, not Stat): a dangling
 		// marker symlink made os.Stat report nothing and the note was
 		// reprocessed every interval (hyperguild#158).
-		if _, err := os.Lstat(path + ".processed"); err == nil {
-			return nil
-		}
-		if _, err := os.Lstat(path + ".failed"); err == nil {
+		if skip, why := markerBlocks(path); skip {
 			return nil
+		} else if why != "" {
+			slog.Warn("watcher: ignoring a stale marker older than its note", "marker", why)
+			_ = root.Remove(filepath.ToSlash(relOf(rawDir, path)) + staleSuffix(why))
 		}
 		// A note whose archive copy or marker was refused is not retried
 		// until it changes: a refusal has no marker to stop the next round,
@@ -188,7 +188,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
 	if before.Size() > limit {
 		return fmt.Errorf("%s is %d bytes, over the %d byte cap: %w", filename, before.Size(), limit, errTooLarge)
 	}
-	content, err := extract.TextIn(root, rel)
+	content, err := extract.TextInMax(root, rel, limit)
 	if err != nil {
 		return fmt.Errorf("extract text: %w", err)
 	}
@@ -206,7 +206,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
 	if runErr != nil {
 		// Copy to failed/ and leave a .failed marker so we don't retry. Both
 		// written through the brain root (hyperguild#158).
-		if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil {
+		if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed", filename, limit); cpErr != nil {
 			return fmt.Errorf("copy to failed: %w", cpErr)
 		}
 		if changedSince(root, rel, before) {
@@ -226,14 +226,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
 
 	// Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest.
 	// Written through the brain root (hyperguild#158).
-	dest := "raw/processed/" + date + "/" + filename
-	if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil {
-		// Archive copy already exists; append timestamp to avoid overwriting.
-		ext := filepath.Ext(filename)
-		base := strings.TrimSuffix(filename, ext)
-		dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext
-	}
-	if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil {
+	if err := copyFromRoot(root, rel, cfg.BrainDir, "raw/processed/"+date, filename, limit); err != nil {
 		return fmt.Errorf("copy to processed: %w", err)
 	}
 	if changedSince(root, rel, before) {
@@ -337,27 +330,51 @@ func putBrainFile(brainDir, rel string, data []byte) error {
 	return brainfs.PutFile(brainDir, rel, data)
 }
 
-// copyFromRoot archives a raw/ file read through the root, refusing anything
-// that is not a regular file, so a swapped-in symlink cannot copy an outside
-// file into raw/processed/ or raw/failed/ (hyperguild#153).
-func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error {
-	fi, err := root.Lstat(rel)
+// copyFromRoot archives a raw/ file read through the root into dstDir under a
+// name no existing file has (so a note never replaces another's copy: the copy
+// is the only one left once the original is retired), refusing anything that is
+// not a regular file and anything over max bytes at the read, not just at the
+// earlier Lstat (hyperguild#158, #198).
+func copyFromRoot(root *os.Root, rel, brainDir, dstDir, filename string, max int64) error {
+	data, err := readNoteMax(root, rel, max)
 	if err != nil {
 		return err
 	}
+	return createUnique(brainDir, dstDir, filename, data)
+}
+
+// readNoteMax reads a raw/ note through the root, at most max bytes.
+func readNoteMax(root *os.Root, rel string, max int64) ([]byte, error) {
+	fi, err := root.Lstat(rel)
+	if err != nil {
+		return nil, err
+	}
 	if !fi.Mode().IsRegular() {
-		return fmt.Errorf("%s: not a regular file", rel)
+		return nil, fmt.Errorf("%s: not a regular file", rel)
 	}
 	in, _, err := extract.OpenRegular(root, rel)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	defer func() { _ = in.Close() }()
-	data, err := io.ReadAll(in)
-	if err != nil {
-		return err
+	data, err := readAllMax(in, max)
+	if errors.Is(err, errTooLarge) {
+		return nil, fmt.Errorf("%s: %w (%d bytes)", rel, errTooLarge, max)
 	}
-	return putBrainFile(brainDir, dstRel, data)
+	return data, err
+}
+
+// readAllMax reads r but never more than max+1 bytes, so an oversize or growing
+// file costs at most that much memory, and reports errTooLarge past max.
+func readAllMax(r io.Reader, max int64) ([]byte, error) {
+	data, err := io.ReadAll(io.LimitReader(r, max+1))
+	if err != nil {
+		return nil, err
+	}
+	if int64(len(data)) > max {
+		return nil, errTooLarge
+	}
+	return data, nil
 }
 
 // topDir is the first directory of a raw/-relative path, "" for a top-level file.
@@ -397,12 +414,8 @@ func parkForReview(cfg Config, root *os.Root, rel, path, filename, content strin
 	var data []byte
 	if strings.ToLower(filepath.Ext(filename)) == ".pdf" {
 		// Binary: park the bytes untouched; the proposal is in log.md only.
-		in, _, err := extract.OpenRegular(root, rel)
-		if err != nil {
-			return fmt.Errorf("park pdf: %w", err)
-		}
-		data, err = io.ReadAll(in)
-		_ = in.Close()
+		var err error
+		data, err = readNoteMax(root, rel, defaultMaxNoteBytes)
 		if err != nil {
 			return fmt.Errorf("park pdf: %w", err)
 		}
@@ -472,3 +485,38 @@ func cutFence(s string) (block, rest string, found bool) {
 	}
 	return "", "", false
 }
+
+// markerBlocks reports whether a marker next to path means "already handled".
+// Any non-regular entry at the marker path counts (a dangling symlink must not
+// be written through, hyperguild#158). A regular marker older than the note
+// describes an EARLIER note of the same name, so it does not block this one:
+// stale is the marker's path and the caller warns and removes it.
+func markerBlocks(path string) (skip bool, stale string) {
+	note, err := os.Lstat(path)
+	if err != nil {
+		return false, ""
+	}
+	for _, suffix := range []string{".processed", ".failed"} {
+		m, err := os.Lstat(path + suffix)
+		if err != nil {
+			continue
+		}
+		if !m.Mode().IsRegular() || !m.ModTime().Before(note.ModTime()) {
+			return true, ""
+		}
+		stale = path + suffix
+	}
+	return false, stale
+}
+
+func relOf(rawDir, path string) string {
+	rel, err := filepath.Rel(rawDir, path)
+	if err != nil {
+		return filepath.Base(path)
+	}
+	return rel
+}
+
+func staleSuffix(markerPath string) string {
+	return filepath.Ext(markerPath)
+}

ingestion/internal/watcher/watcher_test.go

@@ -4,6 +4,7 @@ package watcher
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -497,7 +498,7 @@ func TestProcessDir_ANoteOverTheSizeCapIsRefusedOnceAndKept(t *testing.T) {
 
 	first := processDir(context.Background(), cfg, "2026-10-20")
 	require.NotEmpty(t, first, "the refusal is reported")
-	assert.Contains(t, first[0].Error(), "too large")
+	assert.Contains(t, first[0].Error(), "over the 1024 byte cap")
 	second := processDir(context.Background(), cfg, "2026-10-20")
 	assert.Empty(t, second, "and reported once, not every poll")
 
@@ -532,3 +533,119 @@ func TestProcessDir_ALongNameWhoseMarkerCannotBeWrittenIsNotHandledEveryPoll(t *
 	require.NoError(t, err)
 	assert.Len(t, archived, 1)
 }
+
+// Three notes named n.md in three folders, one poll. The archive and failed/
+// copies are the only copy once the originals are retired, so none may replace
+// another.
+func TestProcessDir_SameNamedNotesAreAllArchivedNoneOverwritten(t *testing.T) {
+	for _, outcome := range []string{"processed", "failed"} {
+		t.Run(outcome, func(t *testing.T) {
+			brainDir := setupBrainDir(t)
+			for _, d := range []string{"a", "b", "c"} {
+				require.NoError(t, os.MkdirAll(filepath.Join(brainDir, "raw", d), 0o755))
+				require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", d, "n.md"), []byte("CONTENT-"+d), 0o644))
+			}
+			cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema",
+				Complete: successComplete(pipeline.RawPage{Title: "N", Type: "source", Content: "x\n"})}}
+			if outcome == "failed" {
+				cfg.Pipeline.Complete = errorComplete
+			}
+			processDir(context.Background(), cfg, "2026-10-20")
+
+			var all string
+			files, _ := filepath.Glob(filepath.Join(brainDir, "raw", outcome, "*", "*"))
+			more, _ := filepath.Glob(filepath.Join(brainDir, "raw", outcome, "*"))
+			for _, f := range append(files, more...) {
+				if b, err := os.ReadFile(f); err == nil {
+					all += string(b)
+				}
+			}
+			for _, d := range []string{"a", "b", "c"} {
+				assert.Contains(t, all, "CONTENT-"+d, "the copy of note %s was overwritten", d)
+			}
+		})
+	}
+}
+
+// A marker older than the note describes an earlier note of the same name; it
+// must not hide the new one, silently, forever (the legacy .processed markers
+// from before inbox zero are exactly this).
+func TestProcessDir_AStaleMarkerDoesNotHideANewerNoteOfTheSameName(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	rawDir := filepath.Join(brainDir, "raw")
+	marker := filepath.Join(rawDir, "idea.md.processed")
+	require.NoError(t, os.WriteFile(marker, []byte("2026-06-16"), 0o644))
+	old := time.Now().Add(-48 * time.Hour)
+	require.NoError(t, os.Chtimes(marker, old, old))
+	require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"), []byte("A new idea, same file name."), 0o644))
+
+	var calls atomic.Int32
+	cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema",
+		Complete: func(ctx context.Context, s, u string) (string, error) {
+			calls.Add(1)
+			return successComplete(pipeline.RawPage{Title: "Idea", Type: "source", Content: "x\n"})(ctx, s, u)
+		}}}
+	require.Empty(t, processDir(context.Background(), cfg, "2026-10-20"))
+
+	assert.Equal(t, int32(1), calls.Load(), "the new note is processed")
+	_, err := os.Stat(filepath.Join(rawDir, "idea.md"))
+	assert.ErrorIs(t, err, os.ErrNotExist, "and retired")
+	_, err = os.Lstat(marker)
+	assert.ErrorIs(t, err, os.ErrNotExist, "the stale marker is cleaned up")
+}
+
+func TestProcessDir_AMarkerNewerThanTheNoteStillMeansDone(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	rawDir := filepath.Join(brainDir, "raw")
+	note := filepath.Join(rawDir, "stuck.md")
+	require.NoError(t, os.WriteFile(note, []byte("Content."), 0o644))
+	past := time.Now().Add(-time.Hour)
+	require.NoError(t, os.Chtimes(note, past, past))
+	require.NoError(t, os.WriteFile(note+".processed", []byte("d"), 0o644)) // newer than the note
+
+	var calls atomic.Int32
+	cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema",
+		Complete: func(context.Context, string, string) (string, error) { calls.Add(1); return "[]", nil }}}
+	processDir(context.Background(), cfg, "2026-10-20")
+	assert.Zero(t, calls.Load(), "a fallback marker written after the note still stops reprocessing")
+}
+
+// The archive copy is a second read of the note; the cap must hold there too,
+// not only at the earlier Lstat (a note can grow in between).
+func TestReadNoteMax(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "n.md"), []byte(strings.Repeat("x", 100)), 0o644))
+	root, err := os.OpenRoot(dir)
+	require.NoError(t, err)
+	defer func() { _ = root.Close() }()
+
+	got, err := readNoteMax(root, "n.md", 100)
+	require.NoError(t, err)
+	assert.Len(t, got, 100, "exactly at the cap is allowed")
+
+	_, err = readNoteMax(root, "n.md", 99)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, errTooLarge)
+}
+
+// countingReader reports how much was pulled from it: an endless file must be
+// abandoned at max+1 bytes, not slurped.
+type countingReader struct{ n int64 }
+
+func (c *countingReader) Read(p []byte) (int, error) {
+	if c.n > 8<<20 { // a regression must fail the test, not exhaust memory
+		return 0, errors.New("endless reader drained")
+	}
+	for i := range p {
+		p[i] = 'x'
+	}
+	c.n += int64(len(p))
+	return len(p), nil
+}
+
+func TestReadAllMax_StopsReadingAtTheCap(t *testing.T) {
+	c := &countingReader{}
+	_, err := readAllMax(c, 1000)
+	require.ErrorIs(t, err, errTooLarge)
+	assert.LessOrEqual(t, c.n, int64(1000+1+32*1024), "pulled far more than the cap from an endless reader: %d", c.n)
+}

ingestion/internal/watcher/writes_redirect_test.go

@@ -103,11 +103,12 @@ func TestProcessDirTreatsAnyEntryAtTheMarkerPathAsAMarker(t *testing.T) {
 }
 
 // A link planted at the marker path while pipeline.Run is in flight (in
-// production, an LLM call) is never written through, and the note is processed
-// once: under inbox zero the original is removed, so there is no marker write
-// left to redirect, and removing the link (as symlinkguard does) does not make
-// the watcher call the model again.
-func TestProcessDirNeverWritesThroughAMarkerLinkPlantedDuringTheRun(t *testing.T) {
+// production, an LLM call) ends up with its target untouched, and the note is
+// processed once, even after the link is removed (as symlinkguard does). Under
+// inbox zero the original is removed, so the marker write that this scenario
+// was written to redirect is never reached: this test does NOT pin that the
+// marker-fallback write goes through the brain root (see the PR description).
+func TestProcessDirLeavesAMarkerLinkPlantedDuringTheRunUntouchedAndProcessesTheNoteOnce(t *testing.T) {
 	for _, kind := range []string{"processed", "failed"} {
 		for _, target := range []string{"outside", "in-brain"} {
 			t.Run(kind+"/"+target, func(t *testing.T) {