fix(ingestion): brain/raw/ readers go through an os.Root; symlinks are inert (#153) (#156)

Co-authored-by: Mathias <mthbqv@gmail.com>

ingestion/internal/watcher/watcher.go

@@ -48,8 +48,25 @@ func Start(ctx context.Context, cfg Config) {
 func processDir(ctx context.Context, cfg Config, date string) []error {
 	rawDir := filepath.Join(cfg.BrainDir, "raw")
 
+	// Every read goes through an os.Root on raw/ (hyperguild#153): raw/ is
+	// writable via Syncthing, and a file swapped for a symlink between the walk
+	// and the read must not lead out of it. A symlinked raw/ is refused.
+	if fi, lerr := os.Lstat(rawDir); lerr != nil {
+		if os.IsNotExist(lerr) {
+			return nil
+		}
+		return []error{fmt.Errorf("stat raw dir: %w", lerr)}
+	} else if fi.Mode()&os.ModeSymlink != 0 || !fi.IsDir() {
+		return []error{fmt.Errorf("raw dir %s is not a real directory; refusing to ingest from it", rawDir)}
+	}
+	root, err := os.OpenRoot(rawDir)
+	if err != nil {
+		return []error{fmt.Errorf("open raw root: %w", err)}
+	}
+	defer func() { _ = root.Close() }()
+
 	var errs []error
-	err := filepath.WalkDir(rawDir, func(path string, d os.DirEntry, err error) error {
+	err = filepath.WalkDir(rawDir, func(path string, d os.DirEntry, err error) error {
 		if err != nil {
 			return err
 		}
@@ -86,6 +103,10 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
 		if strings.HasPrefix(d.Name(), "tunnel-candidates-") {
 			return nil
 		}
+		// Only regular files are notes; a symlink's target may be anywhere.
+		if !d.Type().IsRegular() {
+			return nil
+		}
 
 		// Skip files that have already been processed or permanently failed.
 		if _, err := os.Stat(path + ".processed"); err == nil {
@@ -95,7 +116,7 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
 			return nil
 		}
 
-		if err := processFile(ctx, cfg, path, date); err != nil {
+		if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil {
 			errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err))
 		}
 		return nil
@@ -109,11 +130,15 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
 // processFile reads a file, calls pipeline.Run, copies it to processed/ or failed/,
 // and writes a marker file next to the original so the watcher skips it next poll.
 // The original file is never deleted, keeping Syncthing-connected vaults (e.g. Obsidian) intact.
-func processFile(ctx context.Context, cfg Config, path, date string) error {
+func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, date string) error {
 	filename := filepath.Base(path)
 	source := deriveSource(filename)
+	rel, err := filepath.Rel(rawDir, path)
+	if err != nil {
+		return fmt.Errorf("relative path: %w", err)
+	}
 
-	content, err := extract.Text(path)
+	content, err := extract.TextIn(root, rel)
 	if err != nil {
 		return fmt.Errorf("extract text: %w", err)
 	}
@@ -126,7 +151,7 @@ func processFile(ctx context.Context, cfg Config, path, date string) error {
 			return fmt.Errorf("mkdir failed dir: %w", mkErr)
 		}
 		dest := filepath.Join(failedDir, filename)
-		if cpErr := copyFile(path, dest); cpErr != nil {
+		if cpErr := copyFromRoot(root, rel, dest); cpErr != nil {
 			return fmt.Errorf("copy to failed: %w", cpErr)
 		}
 		if mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil {
@@ -154,7 +179,7 @@ func processFile(ctx context.Context, cfg Config, path, date string) error {
 		base := strings.TrimSuffix(filename, ext)
 		dest = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext)
 	}
-	if err := copyFile(path, dest); err != nil {
+	if err := copyFromRoot(root, rel, dest); err != nil {
 		return fmt.Errorf("copy to processed: %w", err)
 	}
 	if err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil {
@@ -165,26 +190,6 @@ func processFile(ctx context.Context, cfg Config, path, date string) error {
 	return nil
 }
 
-// copyFile copies src to dst, creating dst if it doesn't exist.
-func copyFile(src, dst string) error {
-	in, err := os.Open(src)
-	if err != nil {
-		return fmt.Errorf("open src: %w", err)
-	}
-	defer in.Close() //nolint:errcheck
-
-	out, err := os.Create(dst)
-	if err != nil {
-		return fmt.Errorf("create dst: %w", err)
-	}
-
-	if _, err := io.Copy(out, in); err != nil {
-		out.Close() //nolint:errcheck
-		return fmt.Errorf("copy: %w", err)
-	}
-	return out.Close()
-}
-
 // deriveSource turns a filename into a human-readable source name.
 // "shape-up-book.md" → "Shape Up Book"
 func deriveSource(filename string) string {
@@ -221,3 +226,30 @@ func appendWatcherLog(brainDir, filename string, runErr error, date string) erro
 	}
 	return f.Close()
 }
+
+// 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, dst string) error {
+	fi, err := root.Lstat(rel)
+	if err != nil {
+		return err
+	}
+	if !fi.Mode().IsRegular() {
+		return fmt.Errorf("%s: not a regular file", rel)
+	}
+	in, _, err := extract.OpenRegular(root, rel)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = in.Close() }()
+	out, err := os.Create(dst)
+	if err != nil {
+		return err
+	}
+	if _, err := io.Copy(out, in); err != nil {
+		_ = out.Close()
+		return err
+	}
+	return out.Close()
+}

ingestion/internal/watcher/watcher_test.go

@@ -7,6 +7,9 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"syscall"
 	"testing"
 	"time"
 
@@ -282,3 +285,102 @@ func TestProcessDir_SkipsTunnelCandidateFiles(t *testing.T) {
 	_, err = os.Stat(filepath.Join(brainDir, "raw", "processed", date, "tunnel-candidates-2026-07-19.md"))
 	assert.True(t, os.IsNotExist(err), "tunnel-candidates file should not be copied to processed/")
 }
+
+// hyperguild#153: raw/ is writable via Syncthing, and the watcher ingests
+// whatever lands there with no API call at all. WalkDir yields symlinked
+// files, and extract.Text followed them, so a symlink to any pod-readable
+// file would be extracted into the brain. The watcher now ingests only
+// regular files, read through an os.Root on raw/.
+func TestProcessDir_IngestsTheRealNoteAndSkipsTheSymlinkBesideIt(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	outside := filepath.Join(t.TempDir(), "secret.md")
+	require.NoError(t, os.WriteFile(outside, []byte("SENTINEL-153-WATCHER"), 0o644))
+	// "a-" sorts before "real": a walk that stopped at the first symlink
+	// (filepath.SkipAll — one synced link halting all ingestion) would fail
+	// the call count below (PR #156 review).
+	require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "a-leak.md")))
+	require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "real.md"), []byte("real note"), 0o644))
+
+	var sawSentinel bool
+	calls := 0
+	cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
+		Complete: func(_ context.Context, _, user string) (string, error) {
+			calls++
+			if strings.Contains(user, "SENTINEL-153-WATCHER") {
+				sawSentinel = true
+			}
+			b, _ := json.Marshal([]pipeline.RawPage{{Title: "X", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
+			return string(b), nil
+		},
+		Schema: "# Schema",
+	}}
+	errs := processDir(context.Background(), cfg, time.Now().UTC().Format("2006-01-02"))
+	assert.Empty(t, errs)
+	assert.False(t, sawSentinel, "the symlink target must never reach the pipeline")
+	assert.Equal(t, 1, calls, "exactly the real note is ingested; the symlink is not a note")
+}
+
+// A symlinked raw/ is refused outright: os.OpenRoot would otherwise follow
+// it and confine reads to wherever it points.
+func TestProcessDir_RefusesASymlinkedRawDir(t *testing.T) {
+	brainDir := t.TempDir()
+	elsewhere := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(elsewhere, "x.md"), []byte("SENTINEL-153-RAWLINK"), 0o644))
+	require.NoError(t, os.Symlink(elsewhere, filepath.Join(brainDir, "raw")))
+	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",
+	}}
+	errs := processDir(context.Background(), cfg, time.Now().UTC().Format("2006-01-02"))
+	assert.NotEmpty(t, errs, "a symlinked raw/ must be reported, not silently ingested")
+	assert.Zero(t, calls)
+}
+
+// PR #156: a FIFO swapped into raw/ between the watcher's checks and its
+// read or archive copy must not hang the watcher, which runs unattended.
+func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) {
+	brainDir := setupBrainDir(t)
+	raw := filepath.Join(brainDir, "raw")
+	var stop atomic.Bool
+	go func() {
+		for i := 0; !stop.Load(); i++ {
+			tmp := filepath.Join(raw, ".swap")
+			_ = os.Remove(tmp)
+			if i%2 == 0 {
+				_ = os.WriteFile(tmp, []byte("x"), 0o644)
+			} else {
+				_ = syscall.Mkfifo(tmp, 0o644)
+			}
+			_ = os.Rename(tmp, filepath.Join(raw, "r.md"))
+		}
+	}()
+	cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
+		Complete: func(context.Context, string, string) (string, error) {
+			b, _ := json.Marshal([]pipeline.RawPage{{Title: "R", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
+			return string(b), nil
+		},
+		Schema: "# Schema",
+	}}
+	var iters atomic.Int64
+	done := make(chan struct{})
+	go func() {
+		for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
+			_ = os.Remove(filepath.Join(raw, "r.md.processed"))
+			_ = os.Remove(filepath.Join(raw, "r.md.failed"))
+			_ = processDir(context.Background(), cfg, "2026-10-03")
+			iters.Add(1)
+		}
+		close(done)
+	}()
+	select {
+	case <-done:
+		stop.Store(true)
+	case <-time.After(8 * time.Second):
+		stop.Store(true)
+		if f, err := os.OpenFile(filepath.Join(raw, "r.md"), os.O_WRONLY|syscall.O_NONBLOCK, 0); err == nil {
+			_ = f.Close()
+		}
+		t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load())
+	}
+}