diff --git a/ingestion/cmd/server/main_test.go b/ingestion/cmd/server/main_test.go index 4d1236d..b52f0c7 100644 --- a/ingestion/cmd/server/main_test.go +++ b/ingestion/cmd/server/main_test.go @@ -76,3 +76,27 @@ 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") + }) + } +} diff --git a/ingestion/internal/brainfs/brainfs.go b/ingestion/internal/brainfs/brainfs.go index d8ebeac..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 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/root_test.go b/ingestion/internal/pipeline/root_test.go index d6f4fba..a727e2d 100644 --- a/ingestion/internal/pipeline/root_test.go +++ b/ingestion/internal/pipeline/root_test.go @@ -190,3 +190,28 @@ func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) { 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 wrote the page") +} diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 069590f..348eb47 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -3,12 +3,14 @@ package watcher import ( "context" + "errors" "fmt" "io" "log/slog" "os" "path/filepath" "strings" + "sync" "time" "unicode" @@ -110,14 +112,28 @@ func processDir(ctx context.Context, cfg Config, date string) []error { } // 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 @@ -152,6 +168,9 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return fmt.Errorf("copy to failed: %w", cpErr) } if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil { + if errors.Is(mkErr, brainfs.ErrNotRegular) { + return fmt.Errorf("write .failed marker: %w", mkErr) + } slog.Error("watcher: failed to write .failed marker", "error", mkErr) } @@ -177,6 +196,9 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return fmt.Errorf("copy to processed: %w", err) } if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil { + if errors.Is(err, brainfs.ErrNotRegular) { + return fmt.Errorf("write .processed marker: %w", err) + } slog.Error("watcher: failed to write .processed marker", "error", err) } @@ -184,6 +206,32 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return nil } +// 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 { diff --git a/ingestion/internal/watcher/writes_redirect_test.go b/ingestion/internal/watcher/writes_redirect_test.go index ed46375..fe4efed 100644 --- a/ingestion/internal/watcher/writes_redirect_test.go +++ b/ingestion/internal/watcher/writes_redirect_test.go @@ -33,7 +33,7 @@ func failingPipeline() pipeline.Config { // 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 TestProcessDirDoesNotWriteThroughASymlinkedArchiveDirOrLog(t *testing.T) { +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) @@ -42,12 +42,9 @@ func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDirOrLog(t *testing.T) { 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))) - if name == "failed" { - 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: pcfg}, "2026-10-05") got, _ := os.ReadFile(victim) - assert.Equal(t, "untouched", string(got), "the log was written through a symlink") + assert.Equal(t, "untouched", string(got)) ents, _ := os.ReadDir(outside) assert.Empty(t, ents, "the archive copy was written outside the brain") }) @@ -64,3 +61,70 @@ func TestProcessDirWritesMarkersAndArchiveInRealDirectories(t *testing.T) { _, 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) { + 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.processed"))) + 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 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", "a\\b.md.processed")) + assert.NoError(t, err) +}