From 6bf61d75c03dc2fd85a5a3c8084ffd43bdaf7bd9 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 19 Oct 2026 20:59:06 +0200 Subject: [PATCH 1/4] =?UTF-8?q?feat(watcher):=20inbox=20zero=20=E2=80=94?= =?UTF-8?q?=20archived=20raw/=20originals=20are=20removed,=20not=20copied?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After a note is archived into raw/processed/ or raw/failed/, remove the original from raw/ so raw/ only holds unprocessed input. The .processed / .failed marker is now only a fallback when the removal itself fails. Existing markers are still honoured. Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/watcher/watcher.go | 30 +++++++++++++++------- ingestion/internal/watcher/watcher_test.go | 17 +++++------- 2 files changed, 28 insertions(+), 19 deletions(-) diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index b5c41b3..204da31 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -127,9 +127,12 @@ func processDir(ctx context.Context, cfg Config, date string) []error { return errs } -// processFile reads a file, calls pipeline.Run, copies it to processed/ or failed/, -// and writes a marker file next to the original so the watcher skips it next poll. -// The original file is never deleted, keeping Syncthing-connected vaults (e.g. Obsidian) intact. +// processFile reads a file, calls pipeline.Run, and archives it into processed/ +// or failed/. The original is then removed from raw/ so that raw/ only ever +// holds unprocessed input (inbox zero). The removal syncs to Syncthing peers; +// the archive copy under raw/processed/ or raw/failed/ is what they keep. A +// marker is written next to the original only if the removal fails, so a +// stuck file is not re-ingested every poll. func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, date string) error { filename := filepath.Base(path) source := deriveSource(filename) @@ -154,9 +157,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if cpErr := copyFromRoot(root, rel, dest); cpErr != nil { return fmt.Errorf("copy to failed: %w", cpErr) } - if mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil { - slog.Error("watcher: failed to write .failed marker", "error", mkErr) - } + retire(root, rel, path+".failed", runErr.Error()) slog.Warn("watcher: file failed", "file", filename, "error", runErr) @@ -182,14 +183,25 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if err := copyFromRoot(root, rel, dest); err != nil { return fmt.Errorf("copy to processed: %w", err) } - if err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil { - slog.Error("watcher: failed to write .processed marker", "error", err) - } + retire(root, rel, path+".processed", date) slog.Info("watcher: file processed", "file", filename, "source", source) return nil } +// retire removes an archived original from raw/ (inbox zero). If the removal +// fails it falls back to a marker file so the watcher still skips the file. +func retire(root *os.Root, rel, marker, markerBody string) { + rmErr := root.Remove(rel) + if rmErr == nil { + return + } + slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr) + if err := os.WriteFile(marker, []byte(markerBody), 0o644); err != nil { + slog.Error("watcher: failed to write marker", "marker", marker, "error", err) + } +} + // 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/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 899b2bf..4e2c478 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -89,13 +89,12 @@ func TestStart_ProcessesFile(t *testing.T) { } require.True(t, found, "file should be copied to processed/") - // Original file should still exist (copy, not move — keeps Obsidian vault intact). + // Inbox zero: once archived, the original leaves raw/ and no marker is + // left behind, so what remains in raw/ is exactly the unprocessed set. _, err := os.Stat(rawFile) - assert.NoError(t, err, "original file should remain in raw/") - - // A .processed marker should exist next to the original. + assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once archived") _, err = os.Stat(rawFile + ".processed") - assert.NoError(t, err, ".processed marker should be written") + assert.ErrorIs(t, err, os.ErrNotExist, "no .processed marker: the archive copy is the record") // Wiki page should exist. wikiPath := filepath.Join(brainDir, "wiki", "sources", "shape-up-book.md") @@ -142,13 +141,11 @@ func TestStart_MovesToFailedOnError(t *testing.T) { } require.True(t, found, "file should be copied to failed/") - // Original file should still exist (copy, not move — keeps Obsidian vault intact). + // Inbox zero: a quarantined original leaves raw/ too, no marker beside it. _, err := os.Stat(rawFile) - assert.NoError(t, err, "original file should remain in raw/") - - // A .failed marker should exist next to the original. + assert.ErrorIs(t, err, os.ErrNotExist, "original should be removed from raw/ once quarantined") _, err = os.Stat(rawFile + ".failed") - assert.NoError(t, err, ".failed marker should be written") + assert.ErrorIs(t, err, os.ErrNotExist, "no .failed marker: failed/ is the record") // log.md should contain a watcher error entry. logContent, err := os.ReadFile(filepath.Join(brainDir, "log.md")) -- 2.54.0 From 87ae78252786d10c6f3150797f585ef74d14f4a5 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 19 Oct 2026 21:06:05 +0200 Subject: [PATCH 2/4] fix(watcher): don't re-ingest a note that cannot be retired; pin archive-failure keeps original When both Remove and the marker write fail (read-only raw/), remember the path in memory so later polls skip it instead of writing another archive copy every tick. Adds a test that a failed archive copy leaves the original in raw/, mutation-checked against removing the early return. Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/watcher/watcher.go | 28 +++++++++-- ingestion/internal/watcher/watcher_test.go | 57 ++++++++++++++++++++++ 2 files changed, 80 insertions(+), 5 deletions(-) diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 204da31..5feb3fd 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "unicode" @@ -21,11 +22,19 @@ type Config struct { BrainDir string Interval time.Duration Pipeline pipeline.Config + + // stuck remembers notes that were ingested but could neither be removed + // from raw/ nor marked (e.g. a read-only raw/), so later polls skip them + // instead of ingesting them again. In memory only: a restart retries once. + stuck *sync.Map } // Start launches the watcher in a background goroutine. // It returns immediately. The watcher stops when ctx is cancelled. func Start(ctx context.Context, cfg Config) { + if cfg.stuck == nil { + cfg.stuck = &sync.Map{} + } go func() { ticker := time.NewTicker(cfg.Interval) defer ticker.Stop() @@ -108,6 +117,12 @@ func processDir(ctx context.Context, cfg Config, date string) []error { return nil } + if cfg.stuck != nil { + if _, skip := cfg.stuck.Load(path); skip { + return nil + } + } + // Skip files that have already been processed or permanently failed. if _, err := os.Stat(path + ".processed"); err == nil { return nil @@ -157,7 +172,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if cpErr := copyFromRoot(root, rel, dest); cpErr != nil { return fmt.Errorf("copy to failed: %w", cpErr) } - retire(root, rel, path+".failed", runErr.Error()) + retire(cfg, root, rel, path, ".failed", runErr.Error()) slog.Warn("watcher: file failed", "file", filename, "error", runErr) @@ -183,7 +198,7 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if err := copyFromRoot(root, rel, dest); err != nil { return fmt.Errorf("copy to processed: %w", err) } - retire(root, rel, path+".processed", date) + retire(cfg, root, rel, path, ".processed", date) slog.Info("watcher: file processed", "file", filename, "source", source) return nil @@ -191,14 +206,17 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d // retire removes an archived original from raw/ (inbox zero). If the removal // fails it falls back to a marker file so the watcher still skips the file. -func retire(root *os.Root, rel, marker, markerBody string) { +func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) { rmErr := root.Remove(rel) if rmErr == nil { return } slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr) - if err := os.WriteFile(marker, []byte(markerBody), 0o644); err != nil { - slog.Error("watcher: failed to write marker", "marker", marker, "error", err) + if err := os.WriteFile(path+markerSuffix, []byte(markerBody), 0o644); err != nil { + slog.Error("watcher: failed to write marker; skipping this note until restart", "file", rel, "error", err) + if cfg.stuck != nil { + cfg.stuck.Store(path, struct{}{}) + } } } diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 4e2c478..9af0e7d 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -387,3 +387,60 @@ func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) { t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load()) } } + +// A note whose removal AND marker write both fail (read-only raw/) must not be +// re-ingested on every poll: each pass would write another archive copy. +func TestProcessDir_DoesNotReingestANoteItCannotRetire(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + require.NoError(t, os.WriteFile(filepath.Join(rawDir, "stuck.md"), []byte("Content."), 0o644)) + // processed/ must exist and stay writable; only raw/ itself is locked. + require.NoError(t, os.MkdirAll(filepath.Join(rawDir, "processed"), 0o755)) + require.NoError(t, os.Chmod(rawDir, 0o555)) + t.Cleanup(func() { _ = os.Chmod(rawDir, 0o755) }) + + var calls atomic.Int32 + cfg := Config{ + BrainDir: brainDir, + Pipeline: pipeline.Config{ + Complete: func(ctx context.Context, s, u string) (string, error) { + calls.Add(1) + return successComplete(pipeline.RawPage{Title: "Stuck", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u) + }, + Schema: "# Schema", + }, + stuck: &sync.Map{}, + } + for i := 0; i < 3; i++ { + processDir(context.Background(), cfg, "2026-10-19") + } + assert.Equal(t, int32(1), calls.Load(), "a note that cannot be retired is ingested once, not once per poll") +} + +// If the archive copy cannot be written, the original is the only copy left: +// it must stay in raw/. +func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) { + brainDir := setupBrainDir(t) + rawDir := filepath.Join(brainDir, "raw") + note := filepath.Join(rawDir, "keep-me.md") + require.NoError(t, os.WriteFile(note, []byte("Content."), 0o644)) + // The dated archive dir exists but is read-only: MkdirAll succeeds, the + // copy itself (os.Create) fails. + dated := filepath.Join(rawDir, "processed", "2026-10-19") + require.NoError(t, os.MkdirAll(dated, 0o755)) + require.NoError(t, os.Chmod(dated, 0o555)) + t.Cleanup(func() { _ = os.Chmod(dated, 0o755) }) + + cfg := Config{ + BrainDir: brainDir, + Pipeline: pipeline.Config{ + Complete: successComplete(pipeline.RawPage{Title: "Keep Me", Type: "source", Subtype: "article", Content: "x\n"}), + Schema: "# Schema", + }, + } + errs := processDir(context.Background(), cfg, "2026-10-19") + + assert.NotEmpty(t, errs, "the failed archive is reported") + _, err := os.Stat(note) + assert.NoError(t, err, "original stays in raw/ when it could not be archived") +} -- 2.54.0 From 97ec9e7ce8420437281da19189e06ea66fd55888 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 19 Oct 2026 21:06:46 +0200 Subject: [PATCH 3/4] fix(watcher): keep a note that changed while it was being ingested The archive holds the content read at the start; if Syncthing replaced the note during the (minutes-long) LLM pass, removing the original would drop the newer content. Compare size+mtime before removing; leave it for the next poll. Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/watcher/watcher.go | 23 +++++++++++++++++++++ ingestion/internal/watcher/watcher_test.go | 24 ++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 5feb3fd..f46bf07 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -156,6 +156,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return fmt.Errorf("relative path: %w", err) } + before, err := root.Lstat(rel) + if err != nil { + return fmt.Errorf("stat before ingest: %w", err) + } content, err := extract.TextIn(root, rel) if err != nil { return fmt.Errorf("extract text: %w", err) @@ -172,6 +176,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if cpErr := copyFromRoot(root, rel, dest); cpErr != nil { return fmt.Errorf("copy to failed: %w", cpErr) } + if changedSince(root, rel, before) { + slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename) + return nil + } retire(cfg, root, rel, path, ".failed", runErr.Error()) slog.Warn("watcher: file failed", "file", filename, "error", runErr) @@ -198,12 +206,27 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d if err := copyFromRoot(root, rel, dest); err != nil { return fmt.Errorf("copy to processed: %w", err) } + if changedSince(root, rel, before) { + slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename) + return nil + } retire(cfg, root, rel, path, ".processed", date) slog.Info("watcher: file processed", "file", filename, "source", source) return nil } +// changedSince reports whether the note at rel differs (size or mtime) from +// before. A note replaced mid-ingest (Syncthing) has newer content than the +// archive copy, so it must not be removed. +func changedSince(root *os.Root, rel string, before os.FileInfo) bool { + now, err := root.Lstat(rel) + if err != nil { + return true + } + return now.Size() != before.Size() || !now.ModTime().Equal(before.ModTime()) +} + // retire removes an archived original from raw/ (inbox zero). If the removal // fails it falls back to a marker file so the watcher still skips the file. func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) { diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 9af0e7d..4b0f547 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -444,3 +444,27 @@ func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) { _, err := os.Stat(note) assert.NoError(t, err, "original stays in raw/ when it could not be archived") } + +// If the note is replaced (e.g. by Syncthing) while it is being ingested, the +// archive holds the OLD content; removing the original would lose the new one. +func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) { + brainDir := setupBrainDir(t) + note := filepath.Join(brainDir, "raw", "edited.md") + require.NoError(t, os.WriteFile(note, []byte("old"), 0o644)) + + cfg := Config{ + BrainDir: brainDir, + Pipeline: pipeline.Config{ + Complete: func(ctx context.Context, s, u string) (string, error) { + require.NoError(t, os.WriteFile(note, []byte("newer and longer content"), 0o644)) + return successComplete(pipeline.RawPage{Title: "Edited", Type: "source", Subtype: "article", Content: "x\n"})(ctx, s, u) + }, + Schema: "# Schema", + }, + } + processDir(context.Background(), cfg, "2026-10-19") + + got, err := os.ReadFile(note) + require.NoError(t, err, "a note that changed during ingest must stay in raw/") + assert.Equal(t, "newer and longer content", string(got)) +} -- 2.54.0 From e6e14102ee318f612ac49ba203e2ce2a067a3174 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 19 Oct 2026 21:09:13 +0200 Subject: [PATCH 4/4] docs(watcher): state what changedSince can and cannot detect Co-Authored-By: Claude Sonnet 5.5 --- ingestion/internal/watcher/watcher.go | 6 ++++-- ingestion/internal/watcher/watcher_test.go | 4 +++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index f46bf07..2a951a3 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -217,8 +217,10 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d } // changedSince reports whether the note at rel differs (size or mtime) from -// before. A note replaced mid-ingest (Syncthing) has newer content than the -// archive copy, so it must not be removed. +// before. A note replaced mid-ingest (Syncthing) was extracted from older +// content than it now holds, so it must not be removed. Limits: a same-size +// rewrite with a restored mtime is not detected (the new bytes then survive +// only in the archive copy), nor a change between this check and Remove. func changedSince(root *os.Root, rel string, before os.FileInfo) bool { now, err := root.Lstat(rel) if err != nil { diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index 4b0f547..1f5670d 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -446,7 +446,9 @@ func TestProcessDir_KeepsTheOriginalWhenTheArchiveCopyFails(t *testing.T) { } // If the note is replaced (e.g. by Syncthing) while it is being ingested, the -// archive holds the OLD content; removing the original would lose the new one. +// wiki page was extracted from the old content; removing the original would +// leave the new content only in the archive and never re-ingest it. Observes +// size/mtime change only: a same-size rewrite with a restored mtime is not caught. func TestProcessDir_KeepsANoteThatChangedWhileItWasBeingIngested(t *testing.T) { brainDir := setupBrainDir(t) note := filepath.Join(brainDir, "raw", "edited.md") -- 2.54.0