fix(watcher): refusals end in a terminal state; pin every converted writer (#158)
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:
@@ -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")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user