Compare commits

..
Author SHA1 Message Date
mathiasandClaude Opus 5.5 2f3422adf2 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 <noreply@anthropic.com>
2026-10-05 22:54:53 +02:00
14 changed files with 171 additions and 191 deletions
+9 -4
View File
@@ -17,6 +17,7 @@ import (
"github.com/mathiasbq/hyperguild/ingestion/internal/api" "github.com/mathiasbq/hyperguild/ingestion/internal/api"
"github.com/mathiasbq/hyperguild/ingestion/internal/audit" "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/capture"
"github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp" "github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp"
"github.com/mathiasbq/hyperguild/ingestion/internal/classification" "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") sb.WriteString("\n\n")
} }
slug := "session-" + b.Host + "-" + b.SessionID 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) return fmt.Errorf("write claude session archive: %w", err)
} }
s.logger.Debug("claude session archived (non-indexed)", "path", dest) s.logger.Debug("claude session archived (non-indexed)", "path", dest)
+2 -4
View File
@@ -133,9 +133,7 @@ func PromoteNote(brainDir string, opts PromoteOptions) (string, error) {
if err != nil { if err != nil {
return "", err return "", err
} }
// Created with O_EXCL: a note that appeared after the collision check if _, err := putBrainFile(brainDir, dest, []byte(promoted)); err != nil {
// above is never overwritten.
if err := createBrainFile(brainDir, dest, []byte(promoted)); err != nil {
return "", fmt.Errorf("write promoted note: %w", err) return "", fmt.Errorf("write promoted note: %w", err)
} }
if err := removeBrainFile(brainDir, src); err != nil { if err := removeBrainFile(brainDir, src); err != nil {
@@ -198,7 +196,7 @@ func RejectNote(brainDir string, opts RejectOptions) (string, error) {
// Write-then-delete: the source survives any write failure. Both go // Write-then-delete: the source survives any write failure. Both go
// through the brain root (hyperguild#188). // through the brain root (hyperguild#188).
if err := createBrainFile(brainDir, dest, []byte(fm.render()+body)); err != nil { if _, err := putBrainFile(brainDir, dest, []byte(fm.render()+body)); err != nil {
return "", fmt.Errorf("write rejected note: %w", err) return "", fmt.Errorf("write rejected note: %w", err)
} }
if err := removeBrainFile(brainDir, src); err != nil { if err := removeBrainFile(brainDir, src); err != nil {
@@ -58,30 +58,3 @@ func TestStampProvenanceFileRefusesANoteBehindASymlinkedDirectory(t *testing.T)
got, _ = os.ReadFile(real) got, _ = os.ReadFile(real)
assert.Contains(t, string(got), "written_by: x") assert.Contains(t, string(got), "written_by: x")
} }
// The same through an in-brain symlinked wing and raw/rejected, which
// os.Root alone would follow.
func TestPromoteAndRejectRefuseAnInBrainSymlinkedDestinationDirectory(t *testing.T) {
brain := t.TempDir()
raw := filepath.Join(brain, "raw")
require.NoError(t, os.MkdirAll(raw, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(raw, "2026-10-05-n.md"), []byte("---\ntitle: n\n---\nbody\n"), 0o644))
target := filepath.Join(brain, "archive")
require.NoError(t, os.MkdirAll(target, 0o755))
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "archive"), filepath.Join(brain, "wiki", "lw")))
require.NoError(t, os.Symlink(filepath.Join("..", "archive"), filepath.Join(raw, "rejected")))
_, err := api.PromoteNote(brain, api.PromoteOptions{Filename: "2026-10-05-n.md", Wing: "lw", Hall: "facts"})
assert.Error(t, err)
_, err = api.RejectNote(brain, api.RejectOptions{Filename: "2026-10-05-n.md"})
assert.Error(t, err)
ents, _ := os.ReadDir(target)
assert.Empty(t, ents, "a note was written at the in-brain link's target")
_, err = os.Stat(filepath.Join(raw, "2026-10-05-n.md"))
assert.NoError(t, err, "the raw source is kept")
require.NoError(t, os.Remove(filepath.Join(raw, "rejected")))
_, err = api.RejectNote(brain, api.RejectOptions{Filename: "2026-10-05-n.md"})
require.NoError(t, err, "positive control: reject into a real raw/rejected")
}
+7 -6
View File
@@ -44,19 +44,20 @@ func StampProvenanceFile(brainDir, relPath, origin, writtenBy string) error {
return nil return nil
} }
// createBrainFile creates the file at abs (inside brainDir) through the // putBrainFile creates or replaces the file at abs (inside brainDir) through
// brain root and fails if anything is already there. // the brain root, never through a symlinked directory or file (#188).
func createBrainFile(brainDir, abs string, data []byte) error { func putBrainFile(brainDir, abs string, data []byte) (string, error) {
rel, err := filepath.Rel(brainDir, abs) rel, err := filepath.Rel(brainDir, abs)
if err != nil { if err != nil {
return err return "", err
} }
rel = filepath.ToSlash(rel)
bfs, err := brainfs.Open(brainDir) bfs, err := brainfs.Open(brainDir)
if err != nil { if err != nil {
return fmt.Errorf("open brain: %w", err) return "", fmt.Errorf("open brain: %w", err)
} }
defer func() { _ = bfs.Close() }() defer func() { _ = bfs.Close() }()
return bfs.CreateRegular(filepath.ToSlash(rel), data) return rel, bfs.PutRegular(rel, data)
} }
// removeBrainFile removes the file at abs (inside brainDir) through the // removeBrainFile removes the file at abs (inside brainDir) through the
+3
View File
@@ -7,6 +7,7 @@ import (
"fmt" "fmt"
"io" "io"
"io/fs" "io/fs"
"path/filepath"
"sort" "sort"
"strings" "strings"
"time" "time"
@@ -30,6 +31,7 @@ func BuildWingIndex(brainDir, wing string) error {
if w == "" { if w == "" {
return fmt.Errorf("invalid wing %q", wing) return fmt.Errorf("invalid wing %q", wing)
} }
wingDir := filepath.Join(brainDir, "wiki", w)
// Listed and read through the brain root (hyperguild#157): a wing or // Listed and read through the brain root (hyperguild#157): a wing or
// hall that is a symlink is not followed, and notes are regular files. // hall that is a symlink is not followed, and notes are regular files.
bfs, err := brainfs.Open(brainDir) bfs, err := brainfs.Open(brainDir)
@@ -62,6 +64,7 @@ func BuildWingIndex(brainDir, wing string) error {
// Written through the brain root, never through a symlinked _index.md // Written through the brain root, never through a symlinked _index.md
// (hyperguild#158). // (hyperguild#158).
_ = wingDir
return bfs.PutRegular("wiki/"+w+"/_index.md", []byte(b.String())) return bfs.PutRegular("wiki/"+w+"/_index.md", []byte(b.String()))
} }
+21 -19
View File
@@ -156,25 +156,6 @@ func (b *Brain) PutRegular(rel string, data []byte) error {
} else if !errors.Is(err, fs.ErrNotExist) { } else if !errors.Is(err, fs.ErrNotExist) {
return err return err
} }
return b.createNew(rel, data)
}
// CreateRegular creates a new regular file at a brain-relative path, with
// PutRegular's parent rules, and fails with fs.ErrExist if anything is
// already there: for writers that must never overwrite (promote, reject).
func (b *Brain) CreateRegular(rel string, data []byte) error {
if err := cleanRel(rel); err != nil {
return err
}
if err := b.mkdirReal(path.Dir(rel)); err != nil {
return err
}
return b.createNew(rel, data)
}
// createNew creates rel with O_EXCL and writes data only after confirming
// the path is the regular file it created.
func (b *Brain) createNew(rel string, data []byte) error {
f, err := b.root.OpenFile(rel, os.O_WRONLY|os.O_CREATE|os.O_EXCL|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0o644) f, err := b.root.OpenFile(rel, os.O_WRONLY|os.O_CREATE|os.O_EXCL|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0o644)
if err != nil { if err != nil {
return err return err
@@ -243,6 +224,27 @@ func (b *Brain) RealParent(rel string) error {
return b.realDir(path.Dir(rel)) 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, // mkdirReal makes sure every component of rel exists as a real directory,
// creating missing ones through the root; an existing component that is // creating missing ones through the root; an existing component that is
// not a real directory is ErrNotRegular. // not a real directory is ErrNotRegular.
@@ -453,79 +453,3 @@ func TestRealParentRefusesASymlinkedDirectoryOrUncleanPath(t *testing.T) {
assert.ErrorIs(t, b.RealParent("wiki/../raw/x.md"), ErrNotRegular) assert.ErrorIs(t, b.RealParent("wiki/../raw/x.md"), ErrNotRegular)
assert.NoError(t, b.RealParent("wiki/x.md"), "positive control") assert.NoError(t, b.RealParent("wiki/x.md"), "positive control")
} }
// cleanRel must refuse an unclean path on its own: here the ".." resolves,
// after a symlinked directory, to a real regular file, so only the clean
// check stands between the call and that file.
func TestAppendRegularRefusesAnUncleanPathThatResolvesToARealFile(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw", "keep"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "raw", "v.md"), []byte("untouched"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("raw", "keep"), filepath.Join(dir, "indir")))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
assert.ErrorIs(t, b.AppendRegular("indir/../v.md", []byte("x")), ErrNotRegular)
assert.ErrorIs(t, b.PutRegular("indir/../v.md", []byte("x")), ErrNotRegular)
got, _ := os.ReadFile(filepath.Join(dir, "raw", "v.md"))
assert.Equal(t, "untouched", string(got))
}
func TestRemoveRefusesAPathBehindASymlinkedDirectoryAndRemovesOnlyTheLink(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw"), 0o755))
require.NoError(t, os.WriteFile(filepath.Join(dir, "raw", "v.md"), []byte("keep"), 0o644))
require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "raw"), filepath.Join(dir, "wiki", "alias")))
require.NoError(t, os.Symlink(filepath.Join("..", "raw", "v.md"), filepath.Join(dir, "wiki", "link.md")))
require.NoError(t, os.WriteFile(filepath.Join(dir, "wiki", "n.md"), []byte("n"), 0o644))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
assert.ErrorIs(t, b.Remove("wiki/alias/v.md"), ErrNotRegular)
_, err = os.Stat(filepath.Join(dir, "raw", "v.md"))
assert.NoError(t, err, "the file behind the symlinked directory is kept")
require.NoError(t, b.Remove("wiki/link.md"))
_, err = os.Stat(filepath.Join(dir, "raw", "v.md"))
assert.NoError(t, err, "removing a final symlink never removes its target")
require.NoError(t, b.Remove("wiki/n.md"), "positive control")
_, err = os.Stat(filepath.Join(dir, "wiki", "n.md"))
assert.True(t, os.IsNotExist(err))
}
// CreateRegular is the only write path for promote and reject: it creates
// missing parents through the root, refuses an in-brain symlinked parent and
// an unclean path, and never replaces anything already there (fs.ErrExist,
// content unchanged), including a dangling link.
func TestCreateRegularCreatesOnlyNewFilesThroughRealDirectories(t *testing.T) {
dir := t.TempDir()
require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw", "keep"), 0o755))
require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki"), 0o755))
require.NoError(t, os.Symlink(filepath.Join("..", "raw"), filepath.Join(dir, "wiki", "inwing")))
require.NoError(t, os.WriteFile(filepath.Join(dir, "raw", "v.md"), []byte("untouched"), 0o644))
require.NoError(t, os.Symlink(filepath.Join("raw", "keep"), filepath.Join(dir, "indir")))
require.NoError(t, os.WriteFile(filepath.Join(dir, "wiki", "exists.md"), []byte("original"), 0o644))
require.NoError(t, os.Symlink("nowhere.md", filepath.Join(dir, "wiki", "dangling.md")))
b, err := Open(dir)
require.NoError(t, err)
defer func() { _ = b.Close() }()
assert.ErrorIs(t, b.CreateRegular("wiki/inwing/facts/n.md", []byte("x")), ErrNotRegular)
_, err = os.Stat(filepath.Join(dir, "raw", "facts"))
assert.True(t, os.IsNotExist(err), "nothing created through the in-brain link")
assert.ErrorIs(t, b.CreateRegular("indir/../v.md", []byte("x")), ErrNotRegular)
got, _ := os.ReadFile(filepath.Join(dir, "raw", "v.md"))
assert.Equal(t, "untouched", string(got))
assert.ErrorIs(t, b.CreateRegular("wiki/exists.md", []byte("x")), fs.ErrExist)
got, _ = os.ReadFile(filepath.Join(dir, "wiki", "exists.md"))
assert.Equal(t, "original", string(got))
assert.ErrorIs(t, b.CreateRegular("wiki/dangling.md", []byte("x")), fs.ErrExist)
_, err = os.Stat(filepath.Join(dir, "wiki", "nowhere.md"))
assert.True(t, os.IsNotExist(err), "nothing created at the dangling link's target")
require.NoError(t, b.CreateRegular("wiki/new/facts/n.md", []byte("one")), "positive control: create with missing parents")
got, _ = os.ReadFile(filepath.Join(dir, "wiki", "new", "facts", "n.md"))
assert.Equal(t, "one", string(got))
}
@@ -13,9 +13,6 @@ import (
// a symlinked directory inside the brain. // a symlinked directory inside the brain.
func TestEditRefusesANoteBehindASymlinkedDirectory(t *testing.T) { func TestEditRefusesANoteBehindASymlinkedDirectory(t *testing.T) {
b := t.TempDir() b := t.TempDir()
// Edit refuses a tree that is not the sync source before any path
// check, so the test needs a .git to reach the symlink refusal.
require.NoError(t, os.MkdirAll(filepath.Join(b, ".git"), 0o755))
victim := filepath.Join(b, "raw", "facts", "v.md") victim := filepath.Join(b, "raw", "facts", "v.md")
require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755)) require.NoError(t, os.MkdirAll(filepath.Dir(victim), 0o755))
require.NoError(t, os.WriteFile(victim, []byte("old text\n"), 0o644)) require.NoError(t, os.WriteFile(victim, []byte("old text\n"), 0o644))
@@ -25,12 +22,4 @@ func TestEditRefusesANoteBehindASymlinkedDirectory(t *testing.T) {
assert.Error(t, err) assert.Error(t, err)
got, _ := os.ReadFile(victim) got, _ := os.ReadFile(victim)
assert.Equal(t, "old text\n", string(got)) assert.Equal(t, "old text\n", string(got))
real := filepath.Join(b, "wiki", "w", "facts", "r.md")
require.NoError(t, os.MkdirAll(filepath.Dir(real), 0o755))
require.NoError(t, os.WriteFile(real, []byte("old text\n"), 0o644))
_, err = Edit(b, "wiki/w/facts/r.md", "old text", "new text", ScopeNotes)
require.NoError(t, err, "positive control: a note in a real directory is edited")
got, _ = os.ReadFile(real)
assert.Equal(t, "new text\n", string(got))
} }
+1 -4
View File
@@ -7,8 +7,6 @@ import (
"fmt" "fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"io/fs" "io/fs"
"os"
"path/filepath"
"strings" "strings"
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki" "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 { 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) return 0, fmt.Errorf("write %s: %w", relPath, err)
} }
} }
+14 -7
View File
@@ -9,8 +9,6 @@ import (
"fmt" "fmt"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs" "github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"io/fs" "io/fs"
"os"
"path/filepath"
"strings" "strings"
"time" "time"
@@ -128,10 +126,6 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
var written []string var written []string
for _, page := range merged { for _, page := range merged {
if !dryRun { 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) content, conflicts, err := reviseOrCreate(brainDir, page)
if err != nil { if err != nil {
return Result{}, err return Result{}, err
@@ -150,7 +144,9 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
if err != nil { if err != nil {
return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err) 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) return Result{}, fmt.Errorf("write %s: %w", page.Path, err)
} }
} }
@@ -201,6 +197,17 @@ func readSourcePage(brainDir, sourceSlug string) string {
return string(raw) 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, // readBrainFile reads one brain-relative file through the brain root,
// regular files only (hyperguild#157). // regular files only (hyperguild#157).
func readBrainFile(brainDir, rel string) ([]byte, error) { func readBrainFile(brainDir, rel string) ([]byte, error) {
+25
View File
@@ -165,3 +165,28 @@ func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(t *testing.T) {
require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref") 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) 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")
}
@@ -101,6 +101,7 @@ func Apply(content string, s Stamp) (string, error) {
return out, nil return out, nil
} }
// plainGroups keeps the provably plain key groups of a frontmatter block. // plainGroups keeps the provably plain key groups of a frontmatter block.
func plainGroups(block string) string { func plainGroups(block string) string {
lines := strings.Split(strings.TrimSuffix(block, "\n"), "\n") lines := strings.Split(strings.TrimSuffix(block, "\n"), "\n")
+22 -33
View File
@@ -12,6 +12,7 @@ import (
"time" "time"
"unicode" "unicode"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/extract" "github.com/mathiasbq/hyperguild/ingestion/internal/extract"
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" "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) _, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false)
if runErr != nil { 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); 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) 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) 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. // 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 { dest := "raw/processed/" + date + "/" + filename
return fmt.Errorf("mkdir processed dir: %w", err) if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil {
}
dest := filepath.Join(processedDir, filename)
if _, err := os.Stat(dest); err == nil {
// Archive copy already exists; append timestamp to avoid overwriting. // Archive copy already exists; append timestamp to avoid overwriting.
ext := filepath.Ext(filename) ext := filepath.Ext(filename)
base := strings.TrimSuffix(filename, ext) 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) 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) 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", entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
date, filename, runErr.Error()) 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 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 // 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 // that is not a regular file, so a swapped-in symlink cannot copy an outside
// file into raw/processed/ or raw/failed/ (hyperguild#153). // 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) fi, err := root.Lstat(rel)
if err != nil { if err != nil {
return err return err
@@ -243,13 +236,9 @@ func copyFromRoot(root *os.Root, rel, dst string) error {
return err return err
} }
defer func() { _ = in.Close() }() defer func() { _ = in.Close() }()
out, err := os.Create(dst) data, err := io.ReadAll(in)
if err != nil { if err != nil {
return err return err
} }
if _, err := io.Copy(out, in); err != nil { return putBrainFile(brainDir, dstRel, data)
_ = out.Close()
return err
}
return out.Close()
} }
@@ -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")
}