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 {

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"))