Merge pull request 'fix(watcher): writes through the brain root (lands #195 on current main); size cap; long-name case (#158, #198, #199)' (#210) from wt/watcher-write-safety into main

ingestion/cmd/server/main.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/mathiasbq/hyperguild/ingestion/internal/api"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/audit"
+	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/capture"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/classification"
@@ -90,11 +91,15 @@ func (s *claudeSink) Ingest(ctx context.Context, b claudewatcher.Batch) error {
 		sb.WriteString("\n\n")
 	}
 	slug := "session-" + b.Host + "-" + b.SessionID
-	dest := filepath.Join(s.brainDir, "archive", "claude-sessions", b.Host, slug+".md")
+	dest := "archive/claude-sessions/" + b.Host + "/" + slug + ".md"
-	if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
+	// Written through the brain root, never through a symlinked directory
-		return fmt.Errorf("create claude-sessions archive dir: %w", err)
+	// or file (hyperguild#158).
+	bfs, err := brainfs.Open(s.brainDir)
+	if err != nil {
+		return fmt.Errorf("open brain: %w", err)
 	}
-	if err := os.WriteFile(dest, []byte(sb.String()), 0o644); err != nil {
+	defer func() { _ = bfs.Close() }()
+	if err := bfs.PutRegular(dest, []byte(sb.String())); err != nil {
 		return fmt.Errorf("write claude session archive: %w", err)
 	}
 	s.logger.Debug("claude session archived (non-indexed)", "path", dest)

ingestion/cmd/server/main_test.go

@@ -77,6 +77,30 @@ func TestAlertingWarnings(t *testing.T) {
 	}
 }
 
+// hyperguild#158: the claude-session archive was written with os.MkdirAll +
+// os.WriteFile by path, so a symlinked archive directory redirected it.
+func TestClaudeSinkDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) {
+	for name, mk := range map[string]func(t *testing.T, dir string) (link, target string){
+		"outside": func(t *testing.T, _ string) (string, string) { o := t.TempDir(); return o, o },
+		"in-brain": func(t *testing.T, dir string) (string, string) {
+			require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw"), 0o755))
+			return filepath.Join("..", "raw"), filepath.Join(dir, "raw")
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			dir := t.TempDir()
+			link, target := mk(t, dir)
+			require.NoError(t, os.Symlink(link, filepath.Join(dir, "archive")))
+			sink := &claudeSink{brainDir: dir, logger: slog.New(slog.NewTextHandler(os.Stderr, nil))}
+			err := sink.Ingest(context.Background(), claudewatcher.Batch{Host: "koala", SessionID: "abc",
+				Turns: []claudewatcher.Turn{{Type: "assistant", Content: "x"}}})
+			assert.Error(t, err)
+			ents, _ := os.ReadDir(target)
+			assert.Empty(t, ents, "the archive was written at the link's target")
+		})
+	}
+}
+
 func TestEnvFloat(t *testing.T) {
 	tests := []struct {
 		name string

ingestion/internal/brainfs/brainfs.go

@@ -109,7 +109,7 @@ func (b *Brain) ReadDirRegular(relDir string) ([]string, error) {
 // after following a symlink, and a ".." after a symlinked directory would
 // otherwise pass the check and redirect the read or write.
 func cleanRel(rel string) error {
-	if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) || strings.Contains(rel, "\\") {
+	if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) {
 		return fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular)
 	}
 	return nil
@@ -243,6 +243,27 @@ func (b *Brain) RealParent(rel string) error {
 	return b.realDir(path.Dir(rel))
 }
 
+// PutFile opens the brain dir as a root and calls PutRegular, for one-off
+// writers that hold no Brain.
+func PutFile(brainDir, rel string, data []byte) error {
+	b, err := Open(brainDir)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = b.Close() }()
+	return b.PutRegular(rel, data)
+}
+
+// AppendFile opens the brain dir as a root and calls AppendRegular.
+func AppendFile(brainDir, rel string, data []byte) error {
+	b, err := Open(brainDir)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = b.Close() }()
+	return b.AppendRegular(rel, data)
+}
+
 // mkdirReal makes sure every component of rel exists as a real directory,
 // creating missing ones through the root; an existing component that is
 // not a real directory is ErrNotRegular.

ingestion/internal/brainfs/brainfs_test.go

@@ -529,3 +529,20 @@ func TestCreateRegularCreatesOnlyNewFilesThroughRealDirectories(t *testing.T) {
 	got, _ = os.ReadFile(filepath.Join(dir, "wiki", "new", "facts", "n.md"))
 	assert.Equal(t, "one", string(got))
 }
+
+// On Linux a backslash is an ordinary filename byte, not a separator, so a
+// name containing one stays a single in-root component (hyperguild#158: the
+// watcher retried such a note forever while cleanRel refused it).
+func TestReadAndPutRegularTreatABackslashAsAFilenameByte(t *testing.T) {
+	dir := t.TempDir()
+	b, err := Open(dir)
+	require.NoError(t, err)
+	defer func() { _ = b.Close() }()
+	require.NoError(t, b.PutRegular(`a\..\b.md`, []byte("x")))
+	got, err := os.ReadFile(filepath.Join(dir, `a\..\b.md`))
+	require.NoError(t, err, "the name was not kept as one component")
+	assert.Equal(t, "x", string(got))
+	got, err = b.ReadRegular(`a\..\b.md`)
+	require.NoError(t, err)
+	assert.Equal(t, "x", string(got))
+}

ingestion/internal/extract/callsites_test.go

@@ -88,7 +88,6 @@ func TestNoPackageReferencesExtractTextOutsideExtract(t *testing.T) {
 // osOpenAllowed names the functions in guarded files that may call os.Open*
 // on a path that is not a raw/ read.
 var osOpenAllowed = map[string]bool{
-	"internal/watcher/watcher.go:appendWatcherLog": true, // ingest log append, not a raw/ read
 	// Text reads by path; outside this package only TextIn's temp copy reaches
 	// it (TestNoPackageReferencesExtractTextOutsideExtract enforces that).
 	"internal/extract/extract.go:Text": true,

ingestion/internal/extract/extract.go

@@ -2,6 +2,7 @@
 package extract
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -50,6 +51,18 @@ func fileExt(path string) string {
 // external tool (pdf, docmark) are copied to a temp file from the handle
 // opened through the root, then extracted from the copy.
 func TextIn(r *os.Root, rel string) (string, error) {
+	return TextInMax(r, rel, DefaultMaxBytes)
+}
+
+// DefaultMaxBytes bounds what TextIn reads or copies from one file.
+const DefaultMaxBytes = 64 << 20
+
+// ErrTooLarge reports a file that exceeded the read cap while being read: the
+// size seen at an earlier Lstat is not a bound, the file may have grown.
+var ErrTooLarge = errors.New("file exceeds the read cap")
+
+// TextInMax is TextIn with an explicit cap on the bytes read (hyperguild#198).
+func TextInMax(r *os.Root, rel string, max int64) (string, error) {
 	fi, err := r.Lstat(rel)
 	if err != nil {
 		return "", fmt.Errorf("stat %s: %w", rel, err)
@@ -65,10 +78,13 @@ func TextIn(r *os.Root, rel string) (string, error) {
 	ext := strings.ToLower(fileExt(rel))
 	switch ext {
 	case ".md", ".txt":
-		b, err := io.ReadAll(f)
+		b, err := io.ReadAll(io.LimitReader(f, max+1))
 		if err != nil {
 			return "", fmt.Errorf("read %s: %w", rel, err)
 		}
+		if int64(len(b)) > max {
+			return "", fmt.Errorf("%s: %w (%d bytes)", rel, ErrTooLarge, max)
+		}
 		return string(b), nil
 	case ".pdf", ".docx", ".xlsx", ".pptx", ".png", ".jpg", ".jpeg":
 		tmp, err := os.CreateTemp("", "ingest-*"+ext)
@@ -76,10 +92,15 @@ func TextIn(r *os.Root, rel string) (string, error) {
 			return "", fmt.Errorf("temp copy: %w", err)
 		}
 		defer func() { _ = os.Remove(tmp.Name()) }()
-		if _, err := io.Copy(tmp, f); err != nil {
+		n, err := io.Copy(tmp, io.LimitReader(f, max+1))
+		if err != nil {
 			_ = tmp.Close()
 			return "", fmt.Errorf("copy %s: %w", rel, err)
 		}
+		if n > max {
+			_ = tmp.Close()
+			return "", fmt.Errorf("%s: %w (%d bytes)", rel, ErrTooLarge, max)
+		}
 		if err := tmp.Close(); err != nil {
 			return "", fmt.Errorf("copy %s: %w", rel, err)
 		}

ingestion/internal/extract/extract_cap_test.go

@@ -0,0 +1,62 @@
+package extract
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func rootWith(t *testing.T, name string, size int) *os.Root {
+	t.Helper()
+	dir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte(strings.Repeat("x", size)), 0o644))
+	r, err := os.OpenRoot(dir)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = r.Close() })
+	return r
+}
+
+// hyperguild#198: the size check in the watcher is a pre-check on Lstat; a note
+// that grows before it is read must still be capped at the read.
+func TestTextInMax(t *testing.T) {
+	tests := []struct {
+		name    string
+		size    int
+		max     int64
+		wantErr bool
+	}{
+		{"under the cap", 100, 1024, false},
+		{"exactly at the cap", 1024, 1024, false},
+		{"one byte over", 1025, 1024, true},
+		{"far over", 1 << 20, 1024, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := rootWith(t, "n.md", tt.size)
+			got, err := TextInMax(r, "n.md", tt.max)
+			if tt.wantErr {
+				require.Error(t, err)
+				assert.True(t, errors.Is(err, ErrTooLarge))
+				return
+			}
+			require.NoError(t, err)
+			assert.Len(t, got, tt.size)
+		})
+	}
+}
+
+func TestTextInMax_TheBinaryTempCopyIsCappedToo(t *testing.T) {
+	r := rootWith(t, "n.pdf", 4096)
+	_, err := TextInMax(r, "n.pdf", 1024)
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, ErrTooLarge), "the temp copy of a pdf must not exceed the cap either: %v", err)
+}
+
+func TestTextIn_HasADefaultCap(t *testing.T) {
+	assert.Positive(t, DefaultMaxBytes)
+}

ingestion/internal/pipeline/backfill.go

@@ -7,8 +7,6 @@ import (
 	"fmt"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
 	"io/fs"
-	"os"
-	"path/filepath"
 	"strings"
 
 	"github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
@@ -90,8 +88,7 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
 	}
 
 	for relPath, page := range pending {
-		dest := filepath.Join(brainDir, filepath.FromSlash(relPath))
+		if err := bfs.PutRegular(relPath, []byte(page.Content)); err != nil {
-		if err := os.WriteFile(dest, []byte(page.Content), 0o644); err != nil {
 			return 0, fmt.Errorf("write %s: %w", relPath, err)
 		}
 	}

ingestion/internal/pipeline/pipeline.go

@@ -10,8 +10,6 @@ import (
 	"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
 	"io/fs"
-	"os"
-	"path/filepath"
 	"strings"
 	"time"
 
@@ -143,10 +141,6 @@ func buildAndWrite(ctx context.Context, wing string, rawPages []RawPage, sourceS
 	var written []string
 	for _, page := range merged {
 		if !dryRun {
-			dest := filepath.Join(brainDir, filepath.FromSlash(page.Path))
-			if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
-				return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err)
-			}
 			content, conflicts, err := reviseOrCreate(brainDir, page)
 			if err != nil {
 				return Result{}, err
@@ -165,7 +159,9 @@ func buildAndWrite(ctx context.Context, wing string, rawPages []RawPage, sourceS
 			if err != nil {
 				return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err)
 			}
-			if err := os.WriteFile(dest, []byte(stamped), 0o644); err != nil {
+			// Written through the brain root, never through a symlinked
+			// directory or page (hyperguild#158).
+			if err := writeBrainFile(brainDir, page.Path, []byte(stamped)); err != nil {
 				return Result{}, fmt.Errorf("write %s: %w", page.Path, err)
 			}
 		}
@@ -224,6 +220,17 @@ func readSourcePage(brainDir, wing, sourceSlug string) string {
 	return string(raw)
 }
 
+// writeBrainFile creates or replaces one brain-relative file through the
+// brain root (hyperguild#158).
+func writeBrainFile(brainDir, rel string, data []byte) error {
+	bfs, err := brainfs.Open(brainDir)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = bfs.Close() }()
+	return bfs.PutRegular(rel, data)
+}
+
 // readBrainFile reads one brain-relative file through the brain root,
 // regular files only (hyperguild#157).
 func readBrainFile(brainDir, rel string) ([]byte, error) {

ingestion/internal/pipeline/root_test.go

@@ -165,3 +165,53 @@ func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(t *testing.T) {
 	require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref")
 	brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read)
 }
+
+// hyperguild#158: BackfillRefs read a regular concept page, a symlink could
+// be swapped in, and os.WriteFile then followed it and overwrote the link's
+// target. The target is not restored between rounds here, so any
+// write-through shows.
+func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
+	brain := t.TempDir()
+	mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
+	require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "s.md"), []byte("---\ntitle: S\n---\n[[c|C]]\n"), 0o644))
+	page := filepath.Join(brain, "wiki", "concepts", "c.md")
+	target := filepath.Join(brain, "raw", "secret.md")
+	require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644))
+	inside := "---\ntitle: C\n---\nbody\n"
+	require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
+	brainfstest.StartSwap(t, page, inside, target, false)
+	ok := 0
+	for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
+		if _, err := BackfillRefs(context.Background(), brain); err == nil {
+			ok++
+		}
+	}
+	got, _ := os.ReadFile(target)
+	assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink")
+	assert.Positive(t, ok, "positive control: some backfills ran")
+}
+
+// hyperguild#158: buildAndWrite wrote each page with os.WriteFile by path, so
+// a page read as regular and swapped for a symlink before the write was
+// written through. The target is not restored between rounds, so any
+// write-through shows.
+func TestBuildAndWriteNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
+	brain := t.TempDir()
+	mkdirs(t, brain, "wiki/sources", "wiki/concepts", "wiki/entities", "raw")
+	page := filepath.Join(brain, "wiki", "concepts", "zq-concept.md")
+	target := filepath.Join(brain, "raw", "secret.md")
+	require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644))
+	inside := "---\ntitle: Zq Concept\n---\nbody\n"
+	require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
+	pages := []RawPage{{Title: "Zq Concept", Type: "concept", Content: "## About\n\nmore\n"}}
+	brainfstest.StartSwap(t, page, inside, target, false)
+	ok := 0
+	for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
+		if _, err := RunRaw(context.Background(), brain, "zq source", pages, false); err == nil {
+			ok++
+		}
+	}
+	got, _ := os.ReadFile(target)
+	assert.Equal(t, "untouched", string(got), "buildAndWrite wrote through a swapped-in symlink")
+	assert.Positive(t, ok, "positive control: some runs returned without error")
+}

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

@@ -3,6 +3,7 @@ package watcher
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -14,6 +15,7 @@ import (
 	"unicode"
 
 	"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
+	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
 	"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
@@ -25,10 +27,9 @@ type Config struct {
 	Interval time.Duration
 	Pipeline pipeline.Config
 
-	// stuck remembers notes that were ingested but could neither be removed
+	// MaxNoteBytes refuses a note larger than this before anything is read into
-	// from raw/ nor marked (e.g. a read-only raw/), so later polls skip them
+	// memory (hyperguild#198). 0 means defaultMaxNoteBytes.
-	// instead of ingesting them again. In memory only: a restart retries once.
+	MaxNoteBytes int64
-	stuck *sync.Map
 
 	// Resolver, when set, decides each note's wing (hyperguild#205): explicit
 	// subfolder/frontmatter, else classifier + scorer. A note it cannot place
@@ -36,12 +37,17 @@ type Config struct {
 	Resolver *wing.Resolver
 }
 
+// defaultMaxNoteBytes bounds a single note: extraction and the archive copy both
+// hold it in memory.
+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 = extract.ErrTooLarge
+
 // Start launches the watcher in a background goroutine.
 // It returns immediately. The watcher stops when ctx is cancelled.
 func Start(ctx context.Context, cfg Config) {
-	if cfg.stuck == nil {
-		cfg.stuck = &sync.Map{}
-	}
 	go func() {
 		ticker := time.NewTicker(cfg.Interval)
 		defer ticker.Stop()
@@ -124,21 +130,34 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
 			return nil
 		}
 
-		if cfg.stuck != nil {
-			if _, skip := cfg.stuck.Load(path); skip {
-				return nil
-			}
-		}
-
 		// Skip files that have already been processed or permanently failed.
-		if _, err := os.Stat(path + ".processed"); err == nil {
+		// 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 skip, stale := markerBlocks(path); skip {
+			warnMarkerOnce(path)
 			return nil
+		} else {
+			for _, m := range stale {
+				slog.Warn("watcher: ignoring a stale marker older than its note", "marker", m)
+				if rmErr := root.Remove(relOf(rawDir, m)); rmErr != nil {
+					slog.Error("watcher: failed to remove a stale marker", "marker", m, "error", rmErr)
 				}
-		if _, err := os.Stat(path + ".failed"); err == nil {
+			}
+		}
+		// 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,
+		// and each round can cost an LLM call (hyperguild#158).
+		info, ierr := d.Info()
+		if ierr == nil && refusedUnchanged(path, info) {
 			return nil
 		}
 
 		if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil {
+			if ierr == nil && (errors.Is(err, brainfs.ErrNotRegular) || errors.Is(err, errTooLarge)) {
+				rememberRefused(path, info)
+				slog.Error("watcher: note refused; not retried until it changes", "file", filepath.Base(path), "error", err)
+			}
 			errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err))
 		}
 		return nil
@@ -167,7 +186,14 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
 	if err != nil {
 		return fmt.Errorf("stat before ingest: %w", err)
 	}
-	content, err := extract.TextIn(root, rel)
+	limit := cfg.MaxNoteBytes
+	if limit <= 0 {
+		limit = defaultMaxNoteBytes
+	}
+	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.TextInMax(root, rel, limit)
 	if err != nil {
 		return fmt.Errorf("extract text: %w", err)
 	}
@@ -183,20 +209,16 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
 	}
 	_, runErr := pipeline.Run(ctx, pcfg, cfg.BrainDir, content, source, false)
 	if runErr != nil {
-		// Copy to failed/ and leave a .failed marker so we don't retry.
+		// Copy to failed/ and leave a .failed marker so we don't retry. Both
-		failedDir := filepath.Join(cfg.BrainDir, "raw", "failed")
+		// written through the brain root (hyperguild#158).
-		if mkErr := os.MkdirAll(failedDir, 0o755); mkErr != nil {
+		if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed", filename, limit); cpErr != nil {
-			return fmt.Errorf("mkdir failed dir: %w", mkErr)
-		}
-		dest := filepath.Join(failedDir, filename)
-		if cpErr := copyFromRoot(root, rel, dest); cpErr != nil {
 			return fmt.Errorf("copy to failed: %w", cpErr)
 		}
 		if changedSince(root, rel, before) {
 			slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
 			return nil
 		}
-		retire(cfg, root, rel, path, ".failed", runErr.Error())
+		retire(cfg, root, rel, path, ".failed", runErr.Error(), before)
 
 		slog.Warn("watcher: file failed", "file", filename, "error", runErr)
 
@@ -208,25 +230,15 @@ 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.
-	processedDir := filepath.Join(cfg.BrainDir, "raw", "processed", date)
+	// Written through the brain root (hyperguild#158).
-	if err := os.MkdirAll(processedDir, 0o755); err != nil {
+	if err := copyFromRoot(root, rel, cfg.BrainDir, "raw/processed/"+date, filename, limit); err != nil {
-		return fmt.Errorf("mkdir processed dir: %w", err)
-	}
-	dest := filepath.Join(processedDir, filename)
-	if _, err := os.Stat(dest); err == nil {
-		// Archive copy already exists; append timestamp to avoid overwriting.
-		ext := filepath.Ext(filename)
-		base := strings.TrimSuffix(filename, ext)
-		dest = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext)
-	}
-	if err := copyFromRoot(root, rel, dest); err != nil {
 		return fmt.Errorf("copy to processed: %w", err)
 	}
 	if changedSince(root, rel, before) {
 		slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
 		return nil
 	}
-	retire(cfg, root, rel, path, ".processed", date)
+	retire(cfg, root, rel, path, ".processed", date, before)
 
 	slog.Info("watcher: file processed", "file", filename, "source", source)
 	return nil
@@ -246,19 +258,45 @@ func changedSince(root *os.Root, rel string, before os.FileInfo) bool {
 }
 
 // retire removes an archived original from raw/ (inbox zero). If the removal
-// fails it falls back to a marker file so the watcher still skips the file.
+// fails it falls back to a marker written through the brain root (#158), and if
-func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) {
+// even that fails the note is remembered, with the size and mtime it had, so it
+// is not ingested again until it changes.
+func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string, before os.FileInfo) {
 	rmErr := root.Remove(rel)
 	if rmErr == nil {
 		return
 	}
 	slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr)
-	if err := os.WriteFile(path+markerSuffix, []byte(markerBody), 0o644); err != nil {
+	if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+markerSuffix, []byte(markerBody)); err != nil {
-		slog.Error("watcher: failed to write marker; skipping this note until restart", "file", rel, "error", err)
+		slog.Error("watcher: failed to write marker; skipping this note until it changes", "file", rel, "error", err)
-		if cfg.stuck != nil {
+		rememberRefused(path, before)
-			cfg.stuck.Store(path, struct{}{})
 	}
+}
+
+// refused remembers notes whose archive copy or marker write was refused,
+// by path, with the size and mtime they had, for the life of the process.
+var refused sync.Map
+
+type refusedStamp struct {
+	size  int64
+	mtime time.Time
+}
+
+func rememberRefused(path string, info os.FileInfo) {
+	refused.Store(path, refusedStamp{size: info.Size(), mtime: info.ModTime()})
+}
+
+func refusedUnchanged(path string, info os.FileInfo) bool {
+	v, ok := refused.Load(path)
+	if !ok {
+		return false
 	}
+	st := v.(refusedStamp)
+	if st.size == info.Size() && st.mtime.Equal(info.ModTime()) {
+		return true
+	}
+	refused.Delete(path)
+	return false
 }
 
 // deriveSource turns a filename into a human-readable source name.
@@ -285,44 +323,63 @@ func appendWatcherLog(brainDir, filename string, runErr error, date string) erro
 	entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
 		date, filename, runErr.Error())
 
-	logPath := filepath.Join(brainDir, "log.md")
+	if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil {
-	f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
-	if err != nil {
-		return fmt.Errorf("open log: %w", err)
-	}
-
-	if _, err = f.WriteString(entry); err != nil {
-		f.Close() //nolint:errcheck
 		return fmt.Errorf("write log: %w", err)
 	}
-	return f.Close()
+	return nil
 }
 
-// copyFromRoot archives a raw/ file read through the root, refusing anything
+// putBrainFile creates or replaces a brain-relative file through the brain
-// that is not a regular file, so a swapped-in symlink cannot copy an outside
+// root, never through a symlinked directory or file (hyperguild#158).
-// file into raw/processed/ or raw/failed/ (hyperguild#153).
+func putBrainFile(brainDir, rel string, data []byte) error {
-func copyFromRoot(root *os.Root, rel, dst string) error {
+	return brainfs.PutFile(brainDir, rel, data)
-	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() }()
-	out, err := os.Create(dst)
+	data, err := readAllMax(in, max)
+	if errors.Is(err, errTooLarge) {
+		return nil, fmt.Errorf("%s: %w (%d bytes)", rel, errTooLarge, max)
+	}
+	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 err
+		return nil, err
 	}
-	if _, err := io.Copy(out, in); err != nil {
+	if int64(len(data)) > max {
-		_ = out.Close()
+		return nil, errTooLarge
-		return err
 	}
-	return out.Close()
+	return data, nil
 }
 
 // topDir is the first directory of a raw/-relative path, "" for a top-level file.
@@ -362,12 +419,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)
+		var err error
-		if err != nil {
+		data, err = readNoteMax(root, rel, defaultMaxNoteBytes)
-			return fmt.Errorf("park pdf: %w", err)
-		}
-		data, err = io.ReadAll(in)
-		_ = in.Close()
 		if err != nil {
 			return fmt.Errorf("park pdf: %w", err)
 		}
@@ -385,7 +438,7 @@ func parkForReview(cfg Config, root *os.Root, rel, path, filename, content strin
 		slog.Warn("watcher: note changed while it was being resolved; leaving the newer version in raw/", "file", filename)
 		return nil
 	}
-	retire(cfg, root, rel, path, ".processed", date)
+	retire(cfg, root, rel, path, ".processed", date, before)
 	return nil
 }
 
@@ -437,3 +490,54 @@ 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 describes an EARLIER
+// note of the same name, so it does not block this one, and is returned in
+// stale, when it is older than the note or dated implausibly far in the future
+// (clock skew, a peer): those are removed by the caller. Heuristic by nature:
+// a marker at least as new as the note still means "done".
+func markerBlocks(path string) (skip bool, stale []string) {
+	note, err := os.Lstat(path)
+	if err != nil {
+		return false, nil
+	}
+	for _, suffix := range []string{".processed", ".failed"} {
+		m, err := os.Lstat(path + suffix)
+		if err != nil {
+			continue
+		}
+		if !m.Mode().IsRegular() {
+			return true, nil
+		}
+		if m.ModTime().Before(note.ModTime()) || m.ModTime().After(time.Now().Add(markerSkew)) {
+			stale = append(stale, path+suffix)
+			continue
+		}
+		return true, nil
+	}
+	return false, stale
+}
+
+// markerSkew is how far ahead of now a marker may be dated before it is not
+// believed.
+const markerSkew = 10 * time.Minute
+
+// markerWarned remembers notes already reported as held by a marker, so the
+// 30-second poll says it once instead of forever.
+var markerWarned sync.Map
+
+func warnMarkerOnce(path string) {
+	if _, seen := markerWarned.LoadOrStore(path, struct{}{}); !seen {
+		slog.Warn("watcher: note is held by a marker newer than itself and will not be processed; delete the marker or edit the note", "file", path)
+	}
+}
+
+func relOf(rawDir, path string) string {
+	rel, err := filepath.Rel(rawDir, path)
+	if err != nil {
+		return filepath.Base(path)
+	}
+	return rel
+}

ingestion/internal/watcher/watcher_test.go

@@ -4,7 +4,9 @@ package watcher
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"log/slog"
 	"os"
 	"path/filepath"
 	"strings"
@@ -409,12 +411,17 @@ func TestProcessDir_DoesNotReingestANoteItCannotRetire(t *testing.T) {
 			},
 			Schema: "# Schema",
 		},
-		stuck: &sync.Map{},
 	}
 	for i := 0; i < 3; i++ {
 		processDir(context.Background(), cfg, "2026-10-19")
 	}
-	assert.Equal(t, int32(1), calls.Load(), "a note that cannot be retired is ingested once, not once per poll")
+	// The LLM count alone proves nothing: pipeline.Run's source-hash check skips
+	// extraction on an unchanged note, so a re-run is invisible there. What a
+	// re-run does leave behind is another archive copy, every poll.
+	archived, err := filepath.Glob(filepath.Join(rawDir, "processed", "2026-10-19", "*"))
+	require.NoError(t, err)
+	assert.Len(t, archived, 1, "a note that cannot be retired is archived once, not once per poll")
+	assert.Equal(t, int32(1), calls.Load())
 }
 
 // If the archive copy cannot be written, the original is the only copy left:
@@ -470,3 +477,236 @@ func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) {
 	require.NoError(t, err, "a note that changed during ingest must stay in raw/")
 	assert.Equal(t, "newer and longer content", string(got))
 }
+
+// hyperguild#198: an oversized note must not be read into memory. It is left in
+// raw/ (visible, not lost), reported once, and not retried until it changes.
+func TestProcessDir_ANoteOverTheSizeCapIsRefusedOnceAndKept(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	big := filepath.Join(brainDir, "raw", "big.md")
+	require.NoError(t, os.WriteFile(big, []byte(strings.Repeat("x", 2048)), 0o644))
+	var calls atomic.Int32
+	cfg := Config{
+		BrainDir:     brainDir,
+		MaxNoteBytes: 1024,
+		Pipeline: pipeline.Config{
+			Complete: func(ctx context.Context, s, u string) (string, error) {
+				calls.Add(1)
+				return successComplete(pipeline.RawPage{Title: "Big", Type: "source", Content: "x\n"})(ctx, s, u)
+			},
+			Schema: "# Schema",
+		},
+	}
+
+	first := processDir(context.Background(), cfg, "2026-10-20")
+	require.NotEmpty(t, first, "the refusal is reported")
+	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")
+
+	assert.Zero(t, calls.Load(), "no extraction")
+	_, err := os.Stat(big)
+	assert.NoError(t, err, "the note stays in raw/")
+	entries, _ := os.ReadDir(filepath.Join(brainDir, "raw", "processed"))
+	assert.Empty(t, entries, "nothing was archived")
+}
+
+// hyperguild#199: a name near NAME_MAX cannot take a ".processed" suffix. Under
+// inbox zero the marker is only the fallback when removal fails, and any failure
+// there is remembered, so the note is still not handled every poll.
+func TestProcessDir_ALongNameWhoseMarkerCannotBeWrittenIsNotHandledEveryPoll(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	rawDir := filepath.Join(brainDir, "raw")
+	name := strings.Repeat("a", 250) + ".md" // 253 bytes; +".processed" exceeds 255
+	require.NoError(t, os.WriteFile(filepath.Join(rawDir, name), []byte("Content."), 0o644))
+	require.NoError(t, os.MkdirAll(filepath.Join(rawDir, "processed"), 0o755))
+	require.NoError(t, os.Chmod(rawDir, 0o555)) // removal fails, so the marker fallback runs
+	t.Cleanup(func() { _ = os.Chmod(rawDir, 0o755) })
+	cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{
+		Complete: successComplete(pipeline.RawPage{Title: "Long", Type: "source", Content: "x\n"}), Schema: "# Schema"}}
+
+	for i := 0; i < 3; i++ {
+		errs := processDir(context.Background(), cfg, "2026-10-20")
+		// Without the remember step the second poll tries to archive a second
+		// copy under a name that is too long and errors, every poll.
+		assert.Empty(t, errs, "poll %d handled the note again", i)
+	}
+	archived, err := filepath.Glob(filepath.Join(rawDir, "processed", "2026-10-20", "*"))
+	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)
+}
+
+// From the second review of #210: a marker dated in the future (clock skew,
+// Syncthing) must not hide the note for good, and when a marker does block it
+// has to say so, once.
+func TestProcessDir_AFutureDatedMarkerDoesNotHideTheNote(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("x"), 0o644))
+	future := time.Now().Add(72 * time.Hour)
+	require.NoError(t, os.Chtimes(marker, future, future))
+	require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"), []byte("A new idea."), 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)
+		}}}
+	processDir(context.Background(), cfg, "2026-10-20")
+	assert.Equal(t, int32(1), calls.Load(), "a marker from the future is not evidence the note was handled")
+}
+
+func TestProcessDir_BothStaleMarkersAreRemoved(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	rawDir := filepath.Join(brainDir, "raw")
+	old := time.Now().Add(-48 * time.Hour)
+	for _, suffix := range []string{".processed", ".failed"} {
+		require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"+suffix), []byte("x"), 0o644))
+		require.NoError(t, os.Chtimes(filepath.Join(rawDir, "idea.md"+suffix), old, old))
+	}
+	require.NoError(t, os.WriteFile(filepath.Join(rawDir, "idea.md"), []byte("A new idea."), 0o644))
+	cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema",
+		Complete: successComplete(pipeline.RawPage{Title: "Idea", Type: "source", Content: "x\n"})}}
+	processDir(context.Background(), cfg, "2026-10-20")
+	for _, suffix := range []string{".processed", ".failed"} {
+		_, err := os.Lstat(filepath.Join(rawDir, "idea.md"+suffix))
+		assert.ErrorIs(t, err, os.ErrNotExist, "orphan %s marker left behind", suffix)
+	}
+}
+
+func TestProcessDir_AMarkerThatBlocksANoteIsReportedOnce(t *testing.T) {
+	var logs strings.Builder
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil)))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+
+	brainDir := setupBrainDir(t)
+	rawDir := filepath.Join(brainDir, "raw")
+	note := filepath.Join(rawDir, "held.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: blocks
+	cfg := Config{BrainDir: brainDir, Pipeline: pipeline.Config{Schema: "# Schema", Complete: errorComplete}}
+	for i := 0; i < 3; i++ {
+		processDir(context.Background(), cfg, "2026-10-20")
+	}
+	assert.Equal(t, 1, strings.Count(logs.String(), "held by a marker"), "said once, not every poll: %s", logs.String())
+}

ingestion/internal/watcher/writes_redirect_test.go

@@ -0,0 +1,190 @@
+package watcher
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
+)
+
+func okPipeline() pipeline.Config {
+	return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
+		b, _ := json.Marshal([]pipeline.RawPage{{Title: "X", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
+		return string(b), nil
+	}, Schema: "# Schema"}
+}
+
+func failingPipeline() pipeline.Config {
+	return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
+		return "", errors.New("model down")
+	}, Schema: "# Schema"}
+}
+
+// hyperguild#158: the watcher wrote the archive copy under raw/processed or
+// raw/failed, and its log.md entry, by path, so a symlinked archive
+// directory or log redirected the write elsewhere in the brain or outside
+// it. (A pre-planted marker symlink makes the watcher skip the note, so the
+// marker write itself is reachable only by a race.)
+func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) {
+	for name, pcfg := range map[string]pipeline.Config{"processed": okPipeline(), "failed": failingPipeline()} {
+		t.Run(name, func(t *testing.T) {
+			brainDir := setupBrainDir(t)
+			outside := t.TempDir()
+			victim := filepath.Join(brainDir, "wiki", "concepts", "v.md")
+			require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
+			require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
+			require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", name)))
+			_ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pcfg}, "2026-10-05")
+			got, _ := os.ReadFile(victim)
+			assert.Equal(t, "untouched", string(got))
+			ents, _ := os.ReadDir(outside)
+			assert.Empty(t, ents, "the archive copy was written outside the brain")
+		})
+	}
+}
+
+func TestProcessDirWritesTheArchiveInRealDirectoriesAndRetiresTheOriginal(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
+	errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05")
+	assert.Empty(t, errs)
+	// Inbox zero (v0.25.0): no marker is written beside a note that was removed.
+	_, err := os.Stat(filepath.Join(brainDir, "raw", "n.md.processed"))
+	assert.ErrorIs(t, err, os.ErrNotExist, "a marker is only the fallback when the removal fails")
+	_, err = os.Stat(filepath.Join(brainDir, "raw", "n.md"))
+	assert.ErrorIs(t, err, os.ErrNotExist, "positive control: the original is retired")
+	_, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md"))
+	assert.NoError(t, err, "positive control: archive copy written")
+}
+
+// The log.md entry for a failed note is appended through the brain root: with
+// a real raw/failed (so the archive copy succeeds and the log is reached), a
+// symlinked log.md is not written through.
+func TestProcessDirDoesNotAppendTheLogThroughASymlink(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	victim := filepath.Join(brainDir, "wiki", "concepts", "v.md")
+	require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
+	require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
+	require.NoError(t, os.Symlink(filepath.Join("wiki", "concepts", "v.md"), filepath.Join(brainDir, "log.md")))
+	_ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: failingPipeline()}, "2026-10-05")
+	_, err := os.Stat(filepath.Join(brainDir, "raw", "failed", "n.md"))
+	require.NoError(t, err, "positive control: the archive copy was written, so the log append was reached")
+	got, _ := os.ReadFile(victim)
+	assert.Equal(t, "untouched", string(got), "the log was appended through a symlink")
+}
+
+// A marker path holding anything (here a dangling symlink) counts as a
+// marker: the note is skipped and nothing is created at the link's target.
+// (os.Stat followed the link, saw nothing, and processed the note again.)
+func TestProcessDirTreatsAnyEntryAtTheMarkerPathAsAMarker(t *testing.T) {
+	for _, kind := range []string{"processed", "failed"} {
+		t.Run(kind, func(t *testing.T) {
+			brainDir := setupBrainDir(t)
+			outside := filepath.Join(t.TempDir(), "created")
+			require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
+			require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "n.md."+kind)))
+			calls := 0
+			cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
+				Complete: func(context.Context, string, string) (string, error) { calls++; return "[]", nil }, Schema: "# Schema"}}
+			_ = processDir(context.Background(), cfg, "2026-10-05")
+			assert.Zero(t, calls, "the note was processed despite an entry at its marker path")
+			_, err := os.Lstat(outside)
+			assert.True(t, os.IsNotExist(err), "a file was created at the marker link's target")
+		})
+	}
+}
+
+// A link planted at the marker path while pipeline.Run is in flight (in
+// 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) {
+				brainDir := setupBrainDir(t)
+				raw := filepath.Join(brainDir, "raw")
+				marker := filepath.Join(raw, "n.md."+kind)
+				require.NoError(t, os.WriteFile(filepath.Join(raw, "n.md"), []byte("note"), 0o644))
+				// The in-brain target is relative, so os.Root itself would
+				// follow it; only the regular-file check refuses it.
+				victim := filepath.Join(raw, "victim")
+				link := filepath.Join(t.TempDir(), "created")
+				if target == "in-brain" {
+					require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
+					link = "victim"
+				}
+				ok := okPipeline().Complete
+				calls := 0
+				cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{Schema: "# Schema",
+					Complete: func(ctx context.Context, sys, user string) (string, error) {
+						calls++
+						require.NoError(t, os.Symlink(link, marker))
+						if kind == "failed" {
+							return "", errors.New("model down")
+						}
+						return ok(ctx, sys, user)
+					}}}
+				for i := 0; i < 3; i++ {
+					_ = processDir(context.Background(), cfg, "2026-10-05")
+					_ = os.Remove(marker)
+				}
+				assert.Equal(t, 1, calls, "the note was retried after the link was removed")
+				if target == "in-brain" {
+					got, err := os.ReadFile(victim)
+					require.NoError(t, err)
+					assert.Equal(t, "untouched", string(got), "the marker was written through the link")
+				} else {
+					_, err := os.Lstat(link)
+					assert.True(t, os.IsNotExist(err), "the marker was written through the link")
+				}
+			})
+		}
+	}
+}
+
+// A refused archive copy must not retry every interval: the note is
+// remembered and skipped until it changes (a symlinked raw/failed cost one
+// LLM call per round before).
+func TestProcessDirDoesNotRetryANoteWhoseArchiveCopyIsRefused(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
+	require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brainDir, "raw", "failed")))
+	calls := 0
+	cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
+		Complete: func(context.Context, string, string) (string, error) { calls++; return "", errors.New("model down") }, Schema: "# Schema"}}
+	for i := 0; i < 3; i++ {
+		_ = processDir(context.Background(), cfg, "2026-10-05")
+	}
+	assert.Equal(t, 1, calls, "the refused note was retried every round")
+
+	// Once the note changes it is tried again.
+	require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note, edited"), 0o644))
+	future := time.Now().Add(2 * time.Second)
+	require.NoError(t, os.Chtimes(filepath.Join(brainDir, "raw", "n.md"), future, future))
+	_ = processDir(context.Background(), cfg, "2026-10-05")
+	assert.Equal(t, 2, calls, "positive control: a changed note is tried again")
+}
+
+// A note whose name has a backslash is legal on Linux and must be processed
+// (cleanRel refused it, so it was never archived or marked).
+func TestProcessDirProcessesANoteWithABackslashInItsName(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "a\\b.md"), []byte("note"), 0o644))
+	errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05")
+	assert.Empty(t, errs)
+	_, err := os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "a\\b.md"))
+	assert.NoError(t, err, "archived")
+	_, err = os.Stat(filepath.Join(brainDir, "raw", "a\\b.md"))
+	assert.ErrorIs(t, err, os.ErrNotExist, "and retired")
+}