feat(watcher): file-watcher service for auto-mining on file change

Closes the "automining when files are added or updated" half of JP's
ask. The other half (CLI add/remove of mined data) ships in
jphein/mempalace#4 as `mempalace mined` + `mempalace purge --source-file`.

Adds a watchdog-based file watcher that runs inside the daemon's
lifespan. Configured via `PALACE_WATCH_DIRS` env var:

    PALACE_WATCH_DIRS="/home/jp/Projects/realmwatch=wing_realmwatch,
                       /home/jp/Projects/oracle=wing_oracle"

Each entry is `path` or `path=wing`. Bare path derives wing via
`mempalace.config.normalize_wing_name(basename)` to match the
local-spawn miner default. Non-existent paths are warned-and-skipped,
not fatal — a misconfigured env var doesn't kill startup.

Architecture:

* watcher.py — pure-Python module: parse_watch_dirs(),
  _DebouncedMineHandler (subclass of FileSystemEventHandler with a
  per-target debounce timer), WatcherService (lifecycle wrapper around
  watchdog.Observer).
* main.py lifespan — instantiate WatcherService, schedule one
  recursive watch per target, register stop() on shutdown. Failures
  are non-fatal; the daemon serves traffic regardless.
* main.py — `_internal_mine()` async closure runs the same
  `mempalace mine ... --mode projects --wing X` argv as the existing
  /mine endpoint, gated by the same `_mine_sem`.
* main.py — new `GET /watch` endpoint surfaces the running target
  list. No POST/DELETE — runtime add/remove requires daemon restart
  (operator just edits the systemd unit env and restarts).

Debounce: events on the same target collapse via a 2-second per-target
timer. Catches editor write+rename storms and *.swp / *.lock churn.
Also pre-filters by extension allowlist (29 extensions: code, configs,
markdown). Inotify fires for everything; the handler discards
non-watched extensions before the timer schedules.

Path translation: when a watch path lives in a Syncthing-replicated
directory, the daemon-side path is translated through
`PALACE_DAEMON_PATH_MAP` before the mine subprocess runs (same
mechanism used by /mine).

Tests: 11 new unittest cases in tests/test_watcher.py:
- TestParseWatchDirs (7 cases): empty, single-with-derived-wing,
  explicit-wing, multiple, skips-nonexistent, skips-files, env-fallback
- TestDebouncedMineHandler (4 cases): single-event-fires,
  burst-collapses-to-one, skips-directory-events, skips-bad-extensions

Stacked on #1 (path-translation). When that
merges, this PR auto-rebases against main; reviewers see only the
new commits in this PR.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jphein

main.py

@@ -525,10 +525,52 @@ async def lifespan(app: FastAPI):
         asyncio.create_task(_watchdog_loop(wdog_secs))
         logger.info("Systemd watchdog active (interval=%ds, tick=%ds).", wdog_secs, max(10, wdog_secs // 2))
 
+    # File-watcher service: mines configured directories on file change.
+    # Configured via PALACE_WATCH_DIRS (comma-separated path[=wing]).
+    # Mirrors the pattern used by the /mine endpoint above.
+    try:
+        from watcher import WatcherService, make_async_mine_fn, parse_watch_dirs
+
+        targets = parse_watch_dirs()
+        loop = asyncio.get_running_loop()
+
+        async def _internal_mine(path: str, wing: str) -> None:
+            translated = _translate_client_path(path)
+            dir_path = Path(translated)
+            if not dir_path.is_dir():
+                logger.warning("watcher: skipping mine for non-dir path %s", translated)
+                return
+            mempalace_bin = os.path.join(os.path.dirname(sys.executable), "mempalace")
+            argv = [mempalace_bin, "mine", translated, "--mode", "projects", "--wing", wing]
+            # Same pattern as /mine endpoint: list-form argv, no shell.
+            async with _mine_sem:
+                proc = await asyncio.create_subprocess_exec(
+                    *argv,
+                    stdout=asyncio.subprocess.PIPE,
+                    stderr=asyncio.subprocess.PIPE,
+                )
+                stdout, stderr = await proc.communicate()
+            if proc.returncode != 0:
+                logger.warning("watcher mine returned %s for %s", proc.returncode, translated)
+
+        watcher = WatcherService(make_async_mine_fn(loop, _internal_mine))
+        watcher.start(targets)
+        app.state.watcher = watcher
+    except Exception as e:
+        logger.warning("File-watcher startup failed (non-fatal): %s", e)
+        app.state.watcher = None
+
     yield
 
     # --- Shutdown: Silent Save / Flush ---
     logger.info("Lifespan: shutting down, flushing memories...")
+    try:
+        watcher = getattr(app.state, "watcher", None)
+        if watcher is not None:
+            watcher.stop()
+            logger.info("WatcherService stopped.")
+    except Exception:
+        logger.exception("WatcherService stop failed (non-fatal)")
     try:
         # We call mempalace_memories_filed_away which triggers a checkpoint in recent mempalace versions
         await _call({
@@ -1135,6 +1177,22 @@ async def mine(request: Request, x_api_key: str | None = Header(default=None)):
     }
 
 
+@app.get("/watch")
+async def watch_list(x_api_key: str | None = Header(default=None)):
+    """List the directories the file-watcher is currently monitoring.
+
+    Configured at startup via PALACE_WATCH_DIRS env var; runtime add /
+    remove requires a daemon restart. Returns an empty list when the
+    watcher isn't running (env unset, watchdog package missing, or
+    startup failed).
+    """
+    _check_auth(x_api_key)
+    watcher = getattr(app.state, "watcher", None)
+    if watcher is None:
+        return {"running": False, "targets": []}
+    return {"running": True, "targets": watcher.list_targets()}
+
+
 # ── Repair + silent-save ─────────────────────────────────────────────────────
 
 @app.post("/silent-save")

requirements.txt

@@ -1,3 +1,4 @@
 mempalace>=3.3.2
 fastapi>=0.136.0
 uvicorn[standard]>=0.44.0
+watchdog>=3.0.0

tests/test_watcher.py

@@ -0,0 +1,184 @@
+"""Unit tests for the file-watcher service.
+
+Run with::
+
+    python -m unittest tests.test_watcher -v
+
+Tests are pure-function: parse_watch_dirs() reads the env var and
+returns a list, with no observer threads or filesystem watching
+involved. The Observer-side debouncing is tested in isolation via
+the _DebouncedMineHandler class — instantiated with a fake mine
+callback, fed events, asserted on call count after the debounce
+window has passed.
+"""
+import os
+import sys
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_ROOT = os.path.dirname(_HERE)
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+import watcher as _watcher  # noqa: E402
+
+
+class TestParseWatchDirs(unittest.TestCase):
+    def setUp(self):
+        # Use the test-class temp dir so created paths exist for the
+        # is_dir() guard in parse_watch_dirs.
+        import tempfile
+
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = self._tmp.name
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_empty_returns_empty(self):
+        self.assertEqual(_watcher.parse_watch_dirs(""), [])
+        self.assertEqual(_watcher.parse_watch_dirs("   "), [])
+
+    def test_path_only_derives_wing(self):
+        d = os.path.join(self.tmp, "my-project")
+        os.mkdir(d)
+        targets = _watcher.parse_watch_dirs(d)
+        self.assertEqual(len(targets), 1)
+        self.assertEqual(targets[0].wing, "my_project")
+        self.assertEqual(str(targets[0].path), d)
+
+    def test_path_with_explicit_wing(self):
+        d = os.path.join(self.tmp, "anything")
+        os.mkdir(d)
+        targets = _watcher.parse_watch_dirs(f"{d}=wing_specific")
+        self.assertEqual(len(targets), 1)
+        self.assertEqual(targets[0].wing, "wing_specific")
+
+    def test_multiple_entries(self):
+        d1 = os.path.join(self.tmp, "alpha")
+        d2 = os.path.join(self.tmp, "beta")
+        os.mkdir(d1)
+        os.mkdir(d2)
+        targets = _watcher.parse_watch_dirs(f"{d1}=wing_a, {d2}")
+        self.assertEqual(len(targets), 2)
+        wings = {t.wing for t in targets}
+        self.assertEqual(wings, {"wing_a", "beta"})
+
+    def test_skips_nonexistent_paths(self):
+        d = os.path.join(self.tmp, "exists")
+        os.mkdir(d)
+        missing = os.path.join(self.tmp, "missing-dir-never-created")
+        targets = _watcher.parse_watch_dirs(f"{d}=ok,{missing}=bad")
+        self.assertEqual(len(targets), 1)
+        self.assertEqual(targets[0].wing, "ok")
+
+    def test_skips_files(self):
+        f = os.path.join(self.tmp, "regular-file")
+        open(f, "w").close()
+        targets = _watcher.parse_watch_dirs(f"{f}=should_skip")
+        self.assertEqual(targets, [])
+
+    def test_reads_env_var_when_no_arg(self):
+        d = os.path.join(self.tmp, "from-env")
+        os.mkdir(d)
+        with patch.dict(os.environ, {"PALACE_WATCH_DIRS": f"{d}=env_wing"}, clear=True):
+            targets = _watcher.parse_watch_dirs()
+        self.assertEqual(len(targets), 1)
+        self.assertEqual(targets[0].wing, "env_wing")
+
+
+class TestDebouncedMineHandler(unittest.TestCase):
+    """Integration-ish: instantiate the real handler, feed events, watch
+    the debounce timer fire."""
+
+    def _make_event(self, src_path: str, is_directory: bool = False):
+        # Real watchdog event objects need the watchdog dep. Use a
+        # minimal stand-in that satisfies the handler's attribute reads.
+        class _E:
+            def __init__(self):
+                self.src_path = src_path
+                self.is_directory = is_directory
+
+        return _E()
+
+    def test_single_event_fires_after_debounce(self):
+        target = _watcher.WatchTarget(path=_watcher.Path("/x"), wing="w")
+        fired: list[tuple[str, str]] = []
+        cv = threading.Condition()
+
+        def mine_fn(p, w):
+            with cv:
+                fired.append((p, w))
+                cv.notify_all()
+
+        # Shorten the debounce so the test stays fast.
+        original = _watcher._DEBOUNCE_SECONDS
+        _watcher._DEBOUNCE_SECONDS = 0.05
+        try:
+            handler = _watcher._DebouncedMineHandler(target, mine_fn)
+            handler.on_any_event(self._make_event("/x/foo.md"))
+            with cv:
+                cv.wait(timeout=1.0)
+            self.assertEqual(fired, [("/x", "w")])
+        finally:
+            _watcher._DEBOUNCE_SECONDS = original
+
+    def test_burst_collapses_to_one_fire(self):
+        """A storm of events on the same target produces exactly one mine."""
+        target = _watcher.WatchTarget(path=_watcher.Path("/x"), wing="w")
+        fired: list[tuple[str, str]] = []
+        cv = threading.Condition()
+
+        def mine_fn(p, w):
+            with cv:
+                fired.append((p, w))
+                cv.notify_all()
+
+        original = _watcher._DEBOUNCE_SECONDS
+        _watcher._DEBOUNCE_SECONDS = 0.05
+        try:
+            handler = _watcher._DebouncedMineHandler(target, mine_fn)
+            for i in range(20):
+                handler.on_any_event(self._make_event(f"/x/file{i}.py"))
+                time.sleep(0.005)
+            with cv:
+                cv.wait(timeout=1.0)
+            self.assertEqual(len(fired), 1)
+        finally:
+            _watcher._DEBOUNCE_SECONDS = original
+
+    def test_skips_directory_events(self):
+        target = _watcher.WatchTarget(path=_watcher.Path("/x"), wing="w")
+        fired: list[tuple[str, str]] = []
+        original = _watcher._DEBOUNCE_SECONDS
+        _watcher._DEBOUNCE_SECONDS = 0.05
+        try:
+            handler = _watcher._DebouncedMineHandler(target, lambda p, w: fired.append((p, w)))
+            handler.on_any_event(self._make_event("/x/subdir", is_directory=True))
+            time.sleep(0.15)
+            self.assertEqual(fired, [])
+        finally:
+            _watcher._DEBOUNCE_SECONDS = original
+
+    def test_skips_unwatched_extensions(self):
+        target = _watcher.WatchTarget(path=_watcher.Path("/x"), wing="w")
+        fired: list[tuple[str, str]] = []
+        original = _watcher._DEBOUNCE_SECONDS
+        _watcher._DEBOUNCE_SECONDS = 0.05
+        try:
+            handler = _watcher._DebouncedMineHandler(target, lambda p, w: fired.append((p, w)))
+            # Extensions not in _WATCHABLE_EXTENSIONS — editor swap, lock,
+            # build artifact, binary.
+            for path in ("/x/file.swp", "/x/file.lock", "/x/file.pyc", "/x/file.png"):
+                handler.on_any_event(self._make_event(path))
+            time.sleep(0.15)
+            self.assertEqual(fired, [])
+        finally:
+            _watcher._DEBOUNCE_SECONDS = original
+
+
+if __name__ == "__main__":
+    unittest.main()