fix(watcher): refusals end in a terminal state; pin every converted writer (#158)
CI / Lint / Test / Vet (pull_request) Successful in 1m6s
CI / Mirror to GitHub (pull_request) Skipped

Review round 1 on PR #195:
- A refused marker or archive write left no marker, so the note was
  reprocessed every interval (archive copies piling up; an LLM call per
  round on the failure path). The marker check now uses Lstat (any entry
  at the marker path counts, so a dangling marker link skips the note
  instead of looping), and a note whose archive copy or marker write is
  refused is remembered (path, size, mtime) and not retried until it
  changes.
- cleanRel no longer refuses a backslash: it is a legal filename byte on
  Linux, not a traversal, and a raw note named with one was never
  archived or marked.
- Tests now reach every converted writer: the log.md append (real
  raw/failed so the log is reached, symlinked log.md), a dangling marker
  link, the refused-retry loop with a positive control, a backslash name,
  a buildAndWrite swap race, and the claude-session archive with in-brain
  and outside symlinked archive dirs. The stale lint allowlist entry for
  appendWatcherLog is removed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-05 23:40:19 +02:00
co-authored by Claude Opus 5.5
parent 77c44f298c
commit 1a2a93d2f4
6 changed files with 169 additions and 9 deletions
+24
View File
@@ -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")
})
}
}
+1 -1
View File
@@ -109,7 +109,7 @@ func (b *Brain) ReadDirRegular(relDir string) ([]string, error) {
// after following a symlink, and a ".." after a symlinked directory would // after following a symlink, and a ".." after a symlinked directory would
// otherwise pass the check and redirect the read or write. // otherwise pass the check and redirect the read or write.
func cleanRel(rel string) error { 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 fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular)
} }
return nil return nil
@@ -88,7 +88,6 @@ func TestNoPackageReferencesExtractTextOutsideExtract(t *testing.T) {
// osOpenAllowed names the functions in guarded files that may call os.Open* // osOpenAllowed names the functions in guarded files that may call os.Open*
// on a path that is not a raw/ read. // on a path that is not a raw/ read.
var osOpenAllowed = map[string]bool{ 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 // Text reads by path; outside this package only TextIn's temp copy reaches
// it (TestNoPackageReferencesExtractTextOutsideExtract enforces that). // it (TestNoPackageReferencesExtractTextOutsideExtract enforces that).
"internal/extract/extract.go:Text": true, "internal/extract/extract.go:Text": true,
+25
View File
@@ -190,3 +190,28 @@ func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink") assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink")
assert.Positive(t, ok, "positive control: some backfills ran") 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")
}
+50 -2
View File
@@ -3,12 +3,14 @@ package watcher
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"io" "io"
"log/slog" "log/slog"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
"sync"
"time" "time"
"unicode" "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. // 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 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 return nil
} }
if err := processFile(ctx, cfg, root, rawDir, path, date); err != 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)) errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err))
} }
return nil 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) return fmt.Errorf("copy to failed: %w", cpErr)
} }
if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil { 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) 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) return fmt.Errorf("copy to processed: %w", err)
} }
if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil { 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) 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 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. // deriveSource turns a filename into a human-readable source name.
// "shape-up-book.md" → "Shape Up Book" // "shape-up-book.md" → "Shape Up Book"
func deriveSource(filename string) string { func deriveSource(filename string) string {
@@ -33,7 +33,7 @@ func failingPipeline() pipeline.Config {
// directory or log redirected the write elsewhere in the brain or outside // 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 // it. (A pre-planted marker symlink makes the watcher skip the note, so the
// marker write itself is reachable only by a race.) // 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()} { for name, pcfg := range map[string]pipeline.Config{"processed": okPipeline(), "failed": failingPipeline()} {
t.Run(name, func(t *testing.T) { t.Run(name, func(t *testing.T) {
brainDir := setupBrainDir(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(victim, []byte("untouched"), 0o644))
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 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))) 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") _ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pcfg}, "2026-10-05")
got, _ := os.ReadFile(victim) 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) ents, _ := os.ReadDir(outside)
assert.Empty(t, ents, "the archive copy was written outside the brain") 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")) _, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md"))
assert.NoError(t, err, "positive control: archive copy written") 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)
}