diff --git a/ingestion/cmd/server/main.go b/ingestion/cmd/server/main.go index a08322c..997dcea 100644 --- a/ingestion/cmd/server/main.go +++ b/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") - if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil { - return fmt.Errorf("create claude-sessions archive dir: %w", err) + dest := "archive/claude-sessions/" + b.Host + "/" + slug + ".md" + // Written through the brain root, never through a symlinked directory + // 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) diff --git a/ingestion/cmd/server/main_test.go b/ingestion/cmd/server/main_test.go index 8f6f125..af50176 100644 --- a/ingestion/cmd/server/main_test.go +++ b/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 diff --git a/ingestion/internal/brainfs/brainfs.go b/ingestion/internal/brainfs/brainfs.go index f7f423d..f083f50 100644 --- a/ingestion/internal/brainfs/brainfs.go +++ b/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. diff --git a/ingestion/internal/brainfs/brainfs_test.go b/ingestion/internal/brainfs/brainfs_test.go index 0f5570b..5ac4f4d 100644 --- a/ingestion/internal/brainfs/brainfs_test.go +++ b/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)) +} diff --git a/ingestion/internal/extract/callsites_test.go b/ingestion/internal/extract/callsites_test.go index c544e2e..293ec54 100644 --- a/ingestion/internal/extract/callsites_test.go +++ b/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, diff --git a/ingestion/internal/pipeline/backfill.go b/ingestion/internal/pipeline/backfill.go index 1217f57..8774f9a 100644 --- a/ingestion/internal/pipeline/backfill.go +++ b/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 := os.WriteFile(dest, []byte(page.Content), 0o644); err != nil { + if err := bfs.PutRegular(relPath, []byte(page.Content)); err != nil { return 0, fmt.Errorf("write %s: %w", relPath, err) } } diff --git a/ingestion/internal/pipeline/pipeline.go b/ingestion/internal/pipeline/pipeline.go index 7c39691..9269f3a 100644 --- a/ingestion/internal/pipeline/pipeline.go +++ b/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) { diff --git a/ingestion/internal/pipeline/root_test.go b/ingestion/internal/pipeline/root_test.go index 1e985a8..ab79540 100644 --- a/ingestion/internal/pipeline/root_test.go +++ b/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") +} diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 2a3f67f..d7cadb3 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/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,11 +27,6 @@ type Config struct { Interval time.Duration Pipeline pipeline.Config - // stuck remembers notes that were ingested but could neither be removed - // from raw/ nor marked (e.g. a read-only raw/), so later polls skip them - // instead of ingesting them again. In memory only: a restart retries once. - stuck *sync.Map - // Resolver, when set, decides each note's wing (hyperguild#205): explicit // subfolder/frontmatter, else classifier + scorer. A note it cannot place // is parked in raw/needs-review/ for a human. Nil keeps the retired layout. @@ -39,9 +36,6 @@ type Config struct { // 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 +118,29 @@ 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 _, err := os.Lstat(path + ".processed"); err == nil { return nil } - if _, err := os.Stat(path + ".failed"); err == nil { + if _, err := os.Lstat(path + ".failed"); err == nil { + return 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) { + 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 @@ -183,20 +185,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. - failedDir := filepath.Join(cfg.BrainDir, "raw", "failed") - if mkErr := os.MkdirAll(failedDir, 0o755); mkErr != nil { - return fmt.Errorf("mkdir failed dir: %w", mkErr) - } - dest := filepath.Join(failedDir, filename) - if cpErr := copyFromRoot(root, rel, dest); cpErr != 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 { 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 +206,22 @@ 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) - if err := os.MkdirAll(processedDir, 0o755); err != nil { - return fmt.Errorf("mkdir processed dir: %w", err) - } - dest := filepath.Join(processedDir, filename) - if _, err := os.Stat(dest); err == nil { + // 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 = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext) + dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext } - if err := copyFromRoot(root, rel, dest); err != nil { + if err := copyFromRoot(root, rel, cfg.BrainDir, 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,21 +241,47 @@ 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. -func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) { +// fails it falls back to a marker written through the brain root (#158), and if +// 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 { - slog.Error("watcher: failed to write marker; skipping this note until restart", "file", rel, "error", err) - if cfg.stuck != nil { - cfg.stuck.Store(path, struct{}{}) - } + 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 it changes", "file", rel, "error", err) + rememberRefused(path, before) } } +// 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. // "shape-up-book.md" → "Shape Up Book" func deriveSource(filename string) string { @@ -285,23 +306,22 @@ 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") - 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 + if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil { return fmt.Errorf("write log: %w", err) } - return f.Close() + return nil +} + +// putBrainFile creates or replaces a brain-relative file through the brain +// root, never through a symlinked directory or file (hyperguild#158). +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, dst string) error { +func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error { fi, err := root.Lstat(rel) if err != nil { return err @@ -314,15 +334,11 @@ func copyFromRoot(root *os.Root, rel, dst string) error { return err } defer func() { _ = in.Close() }() - out, err := os.Create(dst) + data, err := io.ReadAll(in) if err != nil { return err } - if _, err := io.Copy(out, in); err != nil { - _ = out.Close() - return err - } - return out.Close() + return putBrainFile(brainDir, dstRel, data) } // topDir is the first directory of a raw/-relative path, "" for a top-level file. @@ -385,7 +401,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 } diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 1f5670d..f76d562 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -409,12 +409,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: diff --git a/ingestion/internal/watcher/writes_redirect_test.go b/ingestion/internal/watcher/writes_redirect_test.go new file mode 100644 index 0000000..9649a0c --- /dev/null +++ b/ingestion/internal/watcher/writes_redirect_test.go @@ -0,0 +1,189 @@ +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) 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) { + 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") +}