From 77c44f298c60d69f17d436ad77f74408b3089130 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 22:54:53 +0200 Subject: [PATCH] fix: pipeline, backfill, watcher and archive writes go through the brain root (#158 part B) The remaining writers created or rewrote brain files by path: - pipeline buildAndWrite: os.MkdirAll + os.WriteFile of every page. - BackfillRefs: os.WriteFile of each concept page; a page swapped for a symlink after the read was written through (the race test overwrote an in-brain target before this change). - watcher: the raw/processed and raw/failed archive copies (os.Create), the .processed/.failed markers and the log.md entry (os.OpenFile). - cmd/server: the claude-session archive. All now go through brainfs PutRegular/AppendRegular (new package-level PutFile/AppendFile for one-off writers: the watcher's call-site lint bans .Open calls in that file so raw/ reads stay on OpenRegular). Stacked on #189 (needs PutRegular/AppendRegular). Co-Authored-By: Claude Opus 5.5 --- ingestion/cmd/server/main.go | 13 ++-- ingestion/internal/brainfs/brainfs.go | 21 ++++++ ingestion/internal/pipeline/backfill.go | 5 +- ingestion/internal/pipeline/pipeline.go | 21 ++++-- ingestion/internal/pipeline/root_test.go | 25 +++++++ ingestion/internal/watcher/watcher.go | 55 +++++++--------- .../internal/watcher/writes_redirect_test.go | 66 +++++++++++++++++++ 7 files changed, 158 insertions(+), 48 deletions(-) create mode 100644 ingestion/internal/watcher/writes_redirect_test.go diff --git a/ingestion/cmd/server/main.go b/ingestion/cmd/server/main.go index db721e4..62dd95a 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" @@ -89,11 +90,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/internal/brainfs/brainfs.go b/ingestion/internal/brainfs/brainfs.go index f7f423d..d8ebeac 100644 --- a/ingestion/internal/brainfs/brainfs.go +++ b/ingestion/internal/brainfs/brainfs.go @@ -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/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 c368d7f..76f97fb 100644 --- a/ingestion/internal/pipeline/pipeline.go +++ b/ingestion/internal/pipeline/pipeline.go @@ -9,8 +9,6 @@ import ( "fmt" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "io/fs" - "os" - "path/filepath" "strings" "time" @@ -128,10 +126,6 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br 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 @@ -150,7 +144,9 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br 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) } } @@ -201,6 +197,17 @@ func readSourcePage(brainDir, 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 b9f7b6c..d6f4fba 100644 --- a/ingestion/internal/pipeline/root_test.go +++ b/ingestion/internal/pipeline/root_test.go @@ -165,3 +165,28 @@ 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") +} diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index b5c41b3..069590f 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -12,6 +12,7 @@ import ( "time" "unicode" + "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "github.com/mathiasbq/hyperguild/ingestion/internal/extract" "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" ) @@ -145,16 +146,12 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d _, runErr := pipeline.Run(ctx, cfg.Pipeline, 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 mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil { + if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil { slog.Error("watcher: failed to write .failed marker", "error", mkErr) } @@ -168,21 +165,18 @@ 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 err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil { + if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil { slog.Error("watcher: failed to write .processed marker", "error", err) } @@ -214,23 +208,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 @@ -243,13 +236,9 @@ 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) } diff --git a/ingestion/internal/watcher/writes_redirect_test.go b/ingestion/internal/watcher/writes_redirect_test.go new file mode 100644 index 0000000..ed46375 --- /dev/null +++ b/ingestion/internal/watcher/writes_redirect_test.go @@ -0,0 +1,66 @@ +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 TestProcessDirDoesNotWriteThroughASymlinkedArchiveDirOrLog(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))) + 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") + ents, _ := os.ReadDir(outside) + assert.Empty(t, ents, "the archive copy was written outside the brain") + }) + } +} + +func TestProcessDirWritesMarkersAndArchiveInRealDirectories(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) + _, err := os.Stat(filepath.Join(brainDir, "raw", "n.md.processed")) + assert.NoError(t, err, "positive control: marker written") + _, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md")) + assert.NoError(t, err, "positive control: archive copy written") +}