From 5f58cdbfa88fead509da36fc4a25bb5f41c580e7 Mon Sep 17 00:00:00 2001 From: Igor Lins e Silva <4753812+igorls@users.noreply.github.com> Date: Fri, 19 Jun 2026 08:33:37 -0300 Subject: [PATCH] fix: unblock daemon PR CI + address review comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI was red on all three platforms for the daemon-mode draft PR. Root causes and fixes: - Linux 3.9 collection error: `_submit_daemon_job`'s `dedupe_key: str | None` parameter annotation is evaluated at def time, and hooks_cli.py has no `from __future__ import annotations` — `str | None` raises TypeError on 3.9. Reverted to `dedupe_key: str = None` (the original, 3.9-safe). The other `int | None` in the file is a function-local annotation, which is never evaluated, so it was never the problem. - macOS/Windows daemon lifecycle flakes: the 3 HTTP-lifecycle tests failed at the 10s readiness deadline on contended CI runners (localhost bind is sub-second locally but took ~5s when it passed on the macOS fleet, >10s when it didn't), and because the server thread never shuts down on timeout, run_server's `os.environ["MEMPALACE_PALACE_PATH"]` + `os.umask(0o077)` mutations leaked into the rest of the suite — poisoning every later test that reads MempalaceConfig().palace_path (the 60+ test_mcp_server cascade on macOS; the at-exit socket hang → SIGINT on Windows). Bumped the readiness deadline to 30s and added a module-scoped snapshot + autouse fixture in test_daemon.py that force-restores the env + umask to the pre-suite baseline after every daemon test, so a leaked server thread can't poison other test files. Gemini review comments (fixed in code, no thread replies per convention): - daemon.py `_connect()` was a bare `sqlite3.connect` whose `with`-block only managed the transaction, not the connection — an unbounded FD leak in a long-lived daemon running thousands of jobs (also the source of the Windows "unclosed database" ResourceWarning noise). Converted to a closing @contextlib.contextmanager. - `QueueStore.finish()` gained `only_if_running`; `_safe_finish` passes it so a late worker finish can't overwrite a shutdown-cancelled job back to succeeded/failed — removes the reliance on process-exit timing. - `DaemonClient.request` wraps the final `json.loads` in try/except JSONDecodeError → DaemonError, so a non-JSON 2xx response surfaces as a structured error instead of a bare JSONDecodeError. - test_sync.py: removed the module-level `import mempalace.mcp_server` and moved the stdout-rebinding side effect into an autouse fixture scoped to TestServiceRunSyncReport, so the embedder/Chroma import chain is no longer forced at collection time for the existing sync tests. Coverage: added focused happy-path tests for service.run_sync early-returns, run_mine backend application + invalid mode, execute_job kind dispatch, run_diary_write arg forwarding, run_mcp_tool write-tool dispatch, and print_job_result — lifts service.py from 57% to 85% so the new files (service 85%, daemon 80%) don't drag the total below the 80% CI gate now that the daemon tests complete and the gate is actually evaluated. --- mempalace/daemon.py | 47 +++++++-- mempalace/hooks_cli.py | 2 +- tests/test_daemon.py | 214 ++++++++++++++++++++++++++++++++++++++++- tests/test_sync.py | 18 ++-- 4 files changed, 266 insertions(+), 15 deletions(-) diff --git a/mempalace/daemon.py b/mempalace/daemon.py index d24e259..3440bd0 100644 --- a/mempalace/daemon.py +++ b/mempalace/daemon.py @@ -8,6 +8,7 @@ execution. from __future__ import annotations import argparse +import contextlib import json import os import secrets @@ -173,10 +174,26 @@ class QueueStore: self._lock = threading.RLock() self._init_db() + @contextlib.contextmanager def _connect(self): + """Open a short-lived sqlite3 connection and close it on exit. + + The bare ``with sqlite3.connect(...)`` context manager only manages the + transaction (commit/rollback) — it does NOT close the connection, so every + QueueStore call in this long-lived daemon process leaked a connection FD. + In a daemon that runs thousands of jobs that is an unbounded FD leak. This + wrapper closes the connection on exit so each call is self-contained. + """ conn = sqlite3.connect(str(self.path), timeout=30) - conn.row_factory = sqlite3.Row - return conn + try: + conn.row_factory = sqlite3.Row + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + conn.close() def _init_db(self) -> None: with self._connect() as conn: @@ -375,13 +392,21 @@ class QueueStore: state: str, result: dict[str, Any] | None = None, error: dict[str, Any] | None = None, + only_if_running: bool = False, ) -> Job: + # ``only_if_running`` guards the worker's finish against a lost race with + # shutdown's cancel: if the active job was already flipped to 'cancelled' + # by _drain_and_cleanup, a late worker finish must NOT overwrite it back to + # 'succeeded'/'failed' (which would un-cancel a job recover_running must + # not re-run). The conditional UPDATE makes the worker's finish a no-op in + # that window instead of relying on process-exit timing. + where = "WHERE id = ?" + (" AND state = 'running'" if only_if_running else "") with self._lock, self._connect() as conn: conn.execute( - """ + f""" UPDATE jobs SET state = ?, finished_at = ?, result_json = ?, error_json = ? - WHERE id = ? + {where} """, ( state, @@ -490,7 +515,9 @@ class DaemonRuntime: def _safe_finish(self, job_id: str, *, state: str, result: dict, error: dict | None) -> None: try: - self.store.finish(job_id, state=state, result=result, error=error) + # only_if_running: if shutdown already cancelled this job, don't + # resurrect it. A finish failure must not kill the worker regardless. + self.store.finish(job_id, state=state, result=result, error=error, only_if_running=True) except Exception: # noqa: BLE001 - a finish failure must not kill the worker pass @@ -783,7 +810,15 @@ class DaemonClient: raise DaemonError(str(payload.get("error", exc))) from exc except OSError as exc: raise DaemonError(str(exc)) from exc - return json.loads(raw) if raw else {} + if not raw: + return {} + try: + return json.loads(raw) + except json.JSONDecodeError as exc: + # A 2xx response with a non-JSON body (empty 200, truncated write, + # proxy HTML) shouldn't surface as a bare JSONDecodeError to callers + # that only know how to handle DaemonError. + raise DaemonError(f"daemon returned non-JSON response: {raw[:200]!r}") from exc def health(self) -> dict[str, Any]: return self.request("GET", "/health") diff --git a/mempalace/hooks_cli.py b/mempalace/hooks_cli.py index 87a95cc..b072ada 100644 --- a/mempalace/hooks_cli.py +++ b/mempalace/hooks_cli.py @@ -545,7 +545,7 @@ def _submit_daemon_job( kind: str, payload: dict, *, - dedupe_key: str | None = None, + dedupe_key: str = None, priority: int = 0, wait: bool = False, timeout: float = 60.0, diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 3854f68..424e081 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -1,3 +1,4 @@ +import os import threading import time @@ -6,6 +7,37 @@ import pytest from mempalace import daemon from mempalace import service +# Env keys run_server mutates from its background thread, plus umask. If a +# lifecycle test times out before the server comes up, run_server's finally +# never runs and those mutations leak into the rest of the suite — every later +# test that reads MempalaceConfig().palace_path sees a stale deleted tmp path and +# fails (the 60+ test cascade seen on slow CI runners). The fixtures below force a +# clean baseline around every daemon test so a leaked thread can't poison the +# process for tests/test_mcp_server.py and friends (which have no such guard). +_LEAK_ENV_KEYS = ("MEMPALACE_PALACE_PATH", "MEMPALACE_BACKEND", "MEMPALACE_BACKEND_EXPLICIT") + + +@pytest.fixture(scope="module") +def _clean_env_snapshot(): + """Capture the true pre-suite values once, before any daemon test runs.""" + return {key: os.environ.get(key) for key in _LEAK_ENV_KEYS} + + +@pytest.fixture(autouse=True) +def _isolate_process_global_state(_clean_env_snapshot): + """Restore the process-global env + umask to the pre-suite baseline after every + daemon test, even if a leaked run_server thread is still holding them mutated. + """ + prev_umask = os.umask(0o022) + os.umask(prev_umask) # read current umask without changing it + yield + for key, value in _clean_env_snapshot.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + os.umask(prev_umask) + def _raise_not_ready(*a, **kw): """Stand-in for DaemonClient when the spawned daemon must never come up.""" @@ -85,7 +117,12 @@ def test_daemon_http_lifecycle_executes_job(tmp_path, monkeypatch): thread.start() client = None - deadline = time.monotonic() + 10 + # 30s: localhost bind is sub-second locally, but contended CI runners (notably + # the macOS GitHub Actions fleet) can take several seconds to bring the server + # up. A too-tight deadline here makes the test flake AND, because the server + # thread never shuts down on timeout, leaks env/umask into the rest of the + # suite (guarded by the _isolate_process_global_state fixture above). + deadline = time.monotonic() + 30 while time.monotonic() < deadline: client = daemon.get_client_if_running(str(palace)) if client is not None: @@ -169,7 +206,7 @@ def _start_server(tmp_path, monkeypatch, execute_fn): ) thread.start() client = None - deadline = time.monotonic() + 10 + deadline = time.monotonic() + 30 while time.monotonic() < deadline: client = daemon.get_client_if_running(str(palace)) if client is not None: @@ -455,3 +492,176 @@ def test_start_daemon_kills_orphan_on_readiness_timeout(tmp_path, monkeypatch): with pytest.raises(daemon.DaemonError): daemon.start_daemon(str(palace), timeout=0.05) assert fake.killed is True + + +# --- service.run_* happy-path coverage --- +# These close the draft PR's follow-up ("Add focused happy-path tests for +# service.run_mine / run_diary_write / run_mcp_tool") and, now that the daemon +# tests complete reliably, keep service.py's coverage above the CI gate. The +# capsys-using tests come first; the two that import mempalace.mcp_server (which +# rebinds sys.stdout) come last and do not use capsys, so the rebind can't break +# capture in this file or later files (capsys activates after the rebind). + + +def test_print_job_result_replays_stdout_stderr_and_returns_exit_code(capsys): + from mempalace import service + + code = service.print_job_result( + {"success": False, "error": "boom", "stdout": "out\n", "stderr": "err\n", "exit_code": 3} + ) + assert code == 3 + captured = capsys.readouterr() + assert "out" in captured.out + assert "err" in captured.err + + +def test_print_job_result_prints_error_to_stderr_when_no_stderr(capsys): + from mempalace import service + + code = service.print_job_result({"success": False, "error": "boom", "exit_code": 1}) + assert code == 1 + captured = capsys.readouterr() + assert "mempalace: boom" in captured.err + + +def test_run_sync_returns_success_when_palace_dir_missing(tmp_path): + from mempalace import service + + result = service.run_sync({"palace_path": str(tmp_path / "nope"), "dry_run": True}) + assert result["success"] is True + assert result["exit_code"] == 0 + + +def test_run_sync_returns_success_when_palace_has_no_backend_artifact(tmp_path): + from mempalace import service + + palace = tmp_path / "palace" + palace.mkdir() + result = service.run_sync({"palace_path": str(palace), "dry_run": True}) + assert result["success"] is True + assert result["exit_code"] == 0 + + +def test_run_mine_invalid_mode_returns_structured_error(tmp_path): + from mempalace import service + + palace = tmp_path / "palace" + palace.mkdir() + out = service.run_mine({"palace_path": str(palace), "mode": "bogus"}) + assert out["success"] is False + assert "invalid mine mode" in out["error"] + assert out["exit_code"] == 2 + + +def test_run_mcp_tool_rejects_non_dict_arguments(): + from mempalace import service + + out = service.run_mcp_tool({"name": "mempalace_add_drawer", "arguments": "nope"}) + assert out["success"] is False + assert "must be an object" in out["error"] + assert out["exit_code"] == 2 + + +def test_run_mcp_tool_dispatches_write_tool(monkeypatch): + import mempalace.mcp_server as mcp + from mempalace import service + + captured = {} + + def fake_handler(**arguments): + captured["arguments"] = arguments + return {"success": True, "written": True} + + monkeypatch.setattr(mcp, "TOOLS", {"mempalace_add_drawer": {"handler": fake_handler}}) + out = service.run_mcp_tool({"name": "mempalace_add_drawer", "arguments": {"x": 1}}) + assert out["success"] is True + assert out["written"] is True + assert out["exit_code"] == 0 + assert captured["arguments"] == {"x": 1} + + +def test_run_diary_write_forwards_args_and_sets_exit_code(monkeypatch): + import mempalace.mcp_server as mcp + from mempalace import service + + captured = {} + + def fake_diary(agent_name, entry, topic, wing): + captured.update(agent_name=agent_name, entry=entry, topic=topic, wing=wing) + return {"success": True} + + monkeypatch.setattr(mcp, "tool_diary_write", fake_diary) + out = service.run_diary_write( + {"agent_name": "alice", "entry": "hello", "topic": "t", "wing": "w"} + ) + assert out["success"] is True + assert out["exit_code"] == 0 + assert captured == {"agent_name": "alice", "entry": "hello", "topic": "t", "wing": "w"} + + +def test_run_mine_applies_backend_before_mode_validation(tmp_path): + """Covers _apply_backend (env set + get_backend_class validation) on the daemon + path; the invalid mode short-circuits before any mining runs.""" + from mempalace import service + + palace = tmp_path / "palace" + palace.mkdir() + out = service.run_mine({"palace_path": str(palace), "mode": "bogus", "backend": "chroma"}) + assert out["success"] is False + assert out["exit_code"] == 2 + + +def test_execute_job_dispatches_diary_write_mcp_tool_and_unknown(monkeypatch): + """Covers execute_job's kind dispatch for diary_write, mcp_tool, and the + unknown-kind fallback.""" + import mempalace.mcp_server as mcp + from mempalace import service + + monkeypatch.setattr(mcp, "tool_diary_write", lambda **kw: {"success": True}) + monkeypatch.setattr( + mcp, "TOOLS", {"mempalace_add_drawer": {"handler": lambda **kw: {"success": True}}} + ) + assert service.execute_job("diary_write", {"entry": "x"})["success"] is True + assert ( + service.execute_job("mcp_tool", {"name": "mempalace_add_drawer", "arguments": {}})[ + "success" + ] + is True + ) + unknown = service.execute_job("bogus_kind", {}) + assert unknown["success"] is False + assert unknown["exit_code"] == 2 + + +def test_run_sync_structured_errors_on_sync_failures(tmp_path, monkeypatch): + """Covers run_sync's three exception handlers (MineAlreadyRunning, ValueError, + generic Exception) so a failing sync_palace returns a structured error instead + of propagating.""" + import mempalace.sync as sync_module + from mempalace import service + from mempalace.palace import MineAlreadyRunning + + palace = tmp_path / "palace" + palace.mkdir() + (palace / "chroma.sqlite3").touch() + + def _raise(exc): + def fn(**kw): + raise exc + + return fn + + monkeypatch.setattr(sync_module, "sync_palace", _raise(MineAlreadyRunning("locked"))) + r = service.run_sync({"palace_path": str(palace), "dry_run": True}) + assert r["success"] is False + assert r["error_class"] == "LockHeldByOtherProcess" + + monkeypatch.setattr(sync_module, "sync_palace", _raise(ValueError("bad scope"))) + r = service.run_sync({"palace_path": str(palace), "dry_run": True}) + assert r["success"] is False + assert r["exit_code"] == 2 + + monkeypatch.setattr(sync_module, "sync_palace", _raise(RuntimeError("boom"))) + r = service.run_sync({"palace_path": str(palace), "dry_run": True}) + assert r["success"] is False + assert "sync failed" in r["error"] diff --git a/tests/test_sync.py b/tests/test_sync.py index e3a34dc..00db448 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -11,12 +11,6 @@ from pathlib import Path import chromadb import pytest -# run_sync imports mempalace.mcp_server lazily; that import initializes the -# embedder, which rebinds sys.stdout and defeats capsys/redirect_stdout for any -# prints after sync_palace returns. Importing it here makes the lazy import a -# cached no-op so the daemon-path report tests can capture run_sync's output. -import mempalace.mcp_server # noqa: F401 - def _seed_drawers(palace_path, repo_path, deleted_path, elsewhere_path): """Populate the drawers collection with 6 entries covering all buckets.""" @@ -1416,6 +1410,18 @@ class TestServiceRunSyncReport: which disturbs sys.stdout and defeats capsys. """ + @pytest.fixture(autouse=True) + def _cache_mcp_server_import(self): + """run_sync lazily imports mempalace.mcp_server, whose import initializes + the embedder and rebinds sys.stdout — defeating capsys for any prints + after sync_palace returns. Lazy-load it here, scoped to just these report + tests (not the whole module at collection time), so the import is a cached + no-op by the time run_sync runs and its report output stays capturable. + """ + import mempalace.mcp_server # noqa: F401 + + yield + def _fake_report(self, **overrides): report = { "scanned": 6,