fix: address Copilot review feedback on PRs #3 + #4

Seven findings, batched into one cleanup PR.

PR #3 (file-watcher) findings:

* watcher.py:30 — drop unused ``import time``.
* watcher.py:336 — type ``_log_future_exception`` parameter as
  ``concurrent.futures.Future`` (NOT ``asyncio.Future``); that's
  what ``asyncio.run_coroutine_threadsafe`` returns. Catch its
  cancellation/state errors plus the asyncio variants so the
  callback can't itself crash on a concurrent cancellation.
* main.py:560 — surface stderr/stdout when watcher mine returns
  non-zero. The rc alone hides 'No mempalace.yaml found' / python
  tracebacks operators need to diagnose.
* main.py:1202 — ``GET /watch`` now double-checks
  ``watcher.is_running`` so a thread crash that flips is_running
  to False between startup and now is reflected in the endpoint
  response. Lifespan still gates publication; this adds defense
  in depth.

PR #4 (mine-queue) findings:

* main.py drain replay — re-apply optional ``extract`` and
  ``limit`` fields. The prior drain dropped them silently, so a
  queue entry with those fields got replayed without. Live /mine
  accepts them; replay must too.
* main.py drain replay — enforce the same safety guards as the
  live /mine endpoint: ``isinstance(dir, str)``, absolute-path,
  no ``..`` traversal, ``mode in {convos, projects}``, ``extract
  in {exchange, general}`` if set, ``limit`` int-coercible if
  set. Without these a queue entry could smuggle through values
  the live endpoint would 400.

Tests: +2 — replay-extract-and-limit (asserts the new fields
make it onto the subprocess argv) and skips-invalid-payload-fields
(5 invalid-payload variants, each correctly skipped pre-spawn).
35 daemon tests pass.

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

Author: jphein

main.py

@@ -385,17 +385,58 @@ async def _drain_pending_mines() -> int:
                 failed_lines.append(line)
         # Replay in original order
         unique_entries.reverse()
+        # Same valid-mode/extract sets as the live /mine endpoint —
+        # apply them on replay too so a queue entry can't smuggle through
+        # a value the live endpoint would reject (Copilot finding on
+        # jphein/palace-daemon#4).
+        VALID_MODES = {"convos", "projects"}
+        VALID_EXTRACTS = {"exchange", "general"}
         for line, entry in unique_entries:
             try:
                 payload = entry["payload"]
-                directory = _translate_client_path(payload["dir"])
-                if not Path(directory).is_dir():
+                raw_dir = payload.get("dir")
+                if not isinstance(raw_dir, str) or not raw_dir:
+                    _log.warning("drain-mine: skipping entry — invalid 'dir'")
+                    continue
+                directory = _translate_client_path(raw_dir)
+                dir_path = Path(directory)
+                # Same path-shape gate as /mine (absolute + no traversal).
+                if not dir_path.is_absolute() or ".." in dir_path.parts:
+                    _log.warning(
+                        "drain-mine: skipping %s — non-absolute or contains '..'", raw_dir
+                    )
+                    continue
+                if not dir_path.is_dir():
                     _log.warning("drain-mine: skipping %s — not a directory", directory)
                     continue
                 wing = payload.get("wing", "general")
                 mode = payload.get("mode", "convos")
+                if mode not in VALID_MODES:
+                    _log.warning("drain-mine: skipping %s — invalid mode %r", directory, mode)
+                    continue
+                extract = payload.get("extract")
+                if extract is not None and extract not in VALID_EXTRACTS:
+                    _log.warning(
+                        "drain-mine: skipping %s — invalid extract %r", directory, extract
+                    )
+                    continue
+                limit = payload.get("limit")
+                if limit is not None:
+                    try:
+                        limit = int(limit)
+                    except (TypeError, ValueError):
+                        _log.warning(
+                            "drain-mine: skipping %s — invalid limit %r", directory, limit
+                        )
+                        continue
                 mempalace_bin = os.path.join(os.path.dirname(sys.executable), "mempalace")
                 cmd = [mempalace_bin, "mine", directory, "--mode", mode, "--wing", wing]
+                # Re-apply optional fields the original /mine accepted but
+                # the prior drain dropped silently (Copilot finding on #4).
+                if extract:
+                    cmd += ["--extract", extract]
+                if limit:
+                    cmd += ["--limit", str(limit)]
                 async with _mine_sem:
                     proc = await asyncio.create_subprocess_exec(
                         *cmd,
@@ -406,7 +447,12 @@ async def _drain_pending_mines() -> int:
                 if proc.returncode == 0:
                     count += 1
                 else:
-                    _log.warning("drain-mine: replay returned %s for %s", proc.returncode, directory)
+                    _log.warning(
+                        "drain-mine: replay returned %s for %s\n  stderr: %s",
+                        proc.returncode,
+                        directory,
+                        (stderr or b"").decode(errors="replace")[:300],
+                    )
                     failed_lines.append(line)
             except Exception:
                 _log.exception("drain-mine: entry replay raised")
@@ -659,7 +705,17 @@ async def _internal_mine(path: str, wing: str) -> None:
                 )
                 stdout, stderr = await proc.communicate()
             if proc.returncode != 0:
-                logger.warning("watcher mine returned %s for %s", proc.returncode, path)
+                # Surface the actual subprocess output — the rc alone hides
+                # 'No mempalace.yaml found' / 'directory not readable' /
+                # python tracebacks that operators need to diagnose.
+                # Closes Copilot finding on jphein/palace-daemon#3.
+                logger.warning(
+                    "watcher mine returned %s for %s\n  stderr: %s\n  stdout: %s",
+                    proc.returncode,
+                    path,
+                    (stderr or b"").decode(errors="replace")[:500],
+                    (stdout or b"").decode(errors="replace")[-500:],
+                )
 
         watcher = WatcherService(make_async_mine_fn(loop, _internal_mine))
         watcher.start(targets)
@@ -1325,7 +1381,12 @@ async def watch_list(x_api_key: str | None = Header(default=None)):
     """
     _check_auth(x_api_key)
     watcher = getattr(app.state, "watcher", None)
-    if watcher is None:
+    # Belt + suspenders: lifespan only publishes app.state.watcher when
+    # is_running, but check it again here so a thread crash that flips
+    # is_running to False between startup and now is reflected in the
+    # endpoint's running= field. Closes Copilot finding on
+    # jphein/palace-daemon#3.
+    if watcher is None or not getattr(watcher, "is_running", False):
         return {"running": False, "targets": []}
     return {"running": True, "targets": watcher.list_targets()}
 

tests/test_mine_queue.py

@@ -129,6 +129,62 @@ async def test_drain_empty_queue_returns_zero(self):
         count = await main._drain_pending_mines()
         self.assertEqual(count, 0)
 
+    async def test_drain_replays_extract_and_limit_options(self):
+        """Closes Copilot finding on jphein/palace-daemon#4 — drain
+        previously dropped optional ``extract`` / ``limit`` fields,
+        so a queue entry that included them got replayed without."""
+        await main._enqueue_pending_mine({
+            "dir": "/a", "wing": "wa", "mode": "convos",
+            "extract": "exchange", "limit": 100,
+        })
+
+        captured_argv = []
+
+        async def _fake_subprocess(*args, **kwargs):
+            captured_argv.append(list(args))
+            proc = MagicMock()
+            proc.communicate = AsyncMock(return_value=(b"", b""))
+            proc.returncode = 0
+            return proc
+
+        with patch("asyncio.create_subprocess_exec", side_effect=_fake_subprocess):
+            count = await main._drain_pending_mines()
+
+        self.assertEqual(count, 1)
+        self.assertEqual(len(captured_argv), 1)
+        argv = captured_argv[0]
+        # extract and limit make it onto the replay command
+        self.assertIn("--extract", argv)
+        self.assertEqual(argv[argv.index("--extract") + 1], "exchange")
+        self.assertIn("--limit", argv)
+        self.assertEqual(argv[argv.index("--limit") + 1], "100")
+
+    async def test_drain_skips_invalid_payload_fields(self):
+        """Closes Copilot finding on jphein/palace-daemon#4 — drain
+        previously skipped only is_dir() check; now also enforces
+        the same valid-mode / valid-extract / int-limit / no-traversal
+        guards as the live /mine endpoint."""
+        for bad in (
+            {"dir": "../../etc/passwd", "wing": "wa", "mode": "convos"},  # traversal
+            {"dir": "/a", "wing": "wa", "mode": "wrong-mode"},  # invalid mode
+            {"dir": "/a", "wing": "wa", "mode": "convos", "extract": "wrong"},  # invalid extract
+            {"dir": "/a", "wing": "wa", "mode": "convos", "limit": "not-a-number"},  # invalid limit
+            {"dir": None, "wing": "wa", "mode": "convos"},  # invalid dir type
+        ):
+            await main._enqueue_pending_mine(bad)
+
+        async def _fake_subprocess(*args, **kwargs):
+            proc = MagicMock()
+            proc.communicate = AsyncMock(return_value=(b"", b""))
+            proc.returncode = 0
+            return proc
+
+        with patch("asyncio.create_subprocess_exec", side_effect=_fake_subprocess) as spawn:
+            count = await main._drain_pending_mines()
+        # All five entries skipped, no subprocess spawned, count = 0
+        self.assertEqual(count, 0)
+        self.assertEqual(spawn.call_count, 0)
+
 
 if __name__ == "__main__":
     unittest.main()

watcher.py

@@ -24,10 +24,10 @@
 from __future__ import annotations
 
 import asyncio
+import concurrent.futures
 import logging
 import os
 import threading
-import time
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable
@@ -319,18 +319,23 @@ def list_targets(self) -> list[dict[str, str]]:
         return [{"path": str(t.path), "wing": t.wing} for t in self._targets]
 
 
-def _log_future_exception(future: "asyncio.Future") -> None:
+def _log_future_exception(future: "concurrent.futures.Future") -> None:
     """Surface exceptions raised inside the scheduled mine coroutine.
 
-    `run_coroutine_threadsafe` returns a Future the caller must observe;
-    if the coroutine raises and the Future is dropped, the exception is
-    swallowed silently. Attaching this callback ensures watcher-driven
-    mine failures show up in the daemon log instead of disappearing.
-    Closes Copilot finding on jphein/palace-daemon#2.
+    ``asyncio.run_coroutine_threadsafe`` returns a
+    ``concurrent.futures.Future`` (NOT ``asyncio.Future``) — the
+    callback receives the cross-thread variant. Catch its
+    cancellation/state errors plus the asyncio variants so the
+    callback can't itself crash on a concurrent cancellation.
+    Closes Copilot finding on jphein/palace-daemon#3.
     """
     try:
         exc = future.exception()
-    except (asyncio.CancelledError, asyncio.InvalidStateError):
+    except (
+        concurrent.futures.CancelledError,
+        concurrent.futures.InvalidStateError,
+        asyncio.CancelledError,
+    ):
         return
     if exc is not None:
         _log.error("watcher-scheduled mine raised: %r", exc, exc_info=exc)