diff --git a/mcp_server/hooks/session_lifecycle.py b/mcp_server/hooks/session_lifecycle.py index 5c321170..14ebdff4 100644 --- a/mcp_server/hooks/session_lifecycle.py +++ b/mcp_server/hooks/session_lifecycle.py @@ -34,6 +34,7 @@ from __future__ import annotations import json +import os import sys from datetime import datetime, timezone from typing import Any @@ -226,6 +227,29 @@ def _tombstone_session_registry() -> None: _log(f"session registry tombstone skipped (non-fatal): {exc}") +def _deregister_groomer_coordinator() -> None: + """Best-effort SessionEnd deregistration for the shared groomer (#171). + + precondition: called from a SessionEnd hook's python process — the same + process whose pid ``SessionStart`` registered via ``os.getpid()``. + postcondition: this session's registration is removed from the per-store + ``GroomerCoordinator``; if it was the LAST live session, the groomer's + single-instance marker is cleared (last-exit stop). Never raises — must + not block the profile update / consolidation that follows. + """ + try: + from mcp_server.infrastructure.groomer_coordinator import ( + GroomerCoordinator, + resolve_store_key, + ) + + coord = GroomerCoordinator(resolve_store_key()) + if coord.stop_if_last(os.getpid()): + _log("groomer coordinator: last session exited, groomer stopped") + except Exception as exc: + _log(f"groomer coordinator deregister skipped (non-fatal): {exc}") + + def process_event(event: dict[str, Any] | None) -> None: """Process a single session lifecycle event. @@ -272,6 +296,11 @@ def main() -> None: # window ended regardless of whether stdin carries a usable event. _tombstone_session_registry() + # Groomer coordinator deregistration (#171): decrement the session + # count; last exit stops the shared groomer. Also unconditional and + # independent of the event payload — the window ending is what matters. + _deregister_groomer_coordinator() + if sys.stdin.isatty(): _log("No stdin data (TTY mode), exiting") return diff --git a/mcp_server/hooks/session_start.py b/mcp_server/hooks/session_start.py index 657ad94e..185b1549 100644 --- a/mcp_server/hooks/session_start.py +++ b/mcp_server/hooks/session_start.py @@ -777,29 +777,90 @@ def _auto_wire_pipeline() -> None: ) +def _spawn_consolidate_cycle() -> int | None: + """Spawn the detached ``consolidate_background`` worker; return its pid. + + precondition: none. postcondition: a fully-detached subprocess (own + process group, stdio → ``consolidate.log``) is started and its pid + returned, or None if the spawn itself failed. The worker runs decay, + compression, CLS, memify, cascade, homeostatic, emergence cycles plus + autonomous wiki maintenance — this is the SAME cycle as before; #171 + changes only WHO decides to start it, never what it does. + """ + plugin_root = os.environ.get("CLAUDE_PLUGIN_ROOT") or str( + Path(__file__).resolve().parents[2] + ) + launcher = Path(plugin_root) / "scripts" / "launcher.py" + py = ( + __import__("shutil").which("python3") + or __import__("shutil").which("python") + or sys.executable + ) + if launcher.exists(): + cmd = [py, str(launcher), "mcp_server.hooks.consolidate_background"] + else: + # Fall back to direct -m invocation (dev source is the package root). + cmd = [py, "-m", "mcp_server.hooks.consolidate_background"] + + log_path = Path.home() / ".claude" / "methodology" / "consolidate.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + proc = subprocess.Popen( # noqa: S603 — cmd built from trusted sources + cmd, + stdin=subprocess.DEVNULL, + stdout=open(log_path, "a"), + stderr=subprocess.STDOUT, + start_new_session=True, + ) + _log(f"background consolidate spawned → {log_path}") + return proc.pid + + def _maybe_background_consolidate() -> None: - """Spawn a detached ``consolidate`` cycle when the stamp is stale. + """Ensure ONE consolidate cycle runs per period across N sessions (#171). The consolidate handler must NEVER be invoked manually by the user - (directive 2026-05-18). SessionStart owns the trigger: if the last - successful run was more than ``CORTEX_CONSOLIDATE_TTL_HOURS`` ago - (default 6h), spawn a detached subprocess that: - - * Runs decay, compression, CLS, memify, cascade, homeostatic, - emergence cycles. - * Runs autonomous wiki maintenance (stub purge + classifier-reject - purge + coverage / drift audit). - * Updates the stamp at ``~/.claude/methodology/.last_consolidate``. - * Logs to ``~/.claude/methodology/consolidate.log``. - - Spawn is fully detached (own process group, stdio redirected to the - log) so SessionStart returns immediately. The user opens a session, - Cortex catches up silently in the background. The next session sees - the freshly-consolidated state. - - Failure is silent: a consolidate that crashes leaves the stamp - untouched so the next session retries. A persistent failure surfaces - in the log file (operators can `tail -f` it). + (directive 2026-05-18). SessionStart owns the trigger, but the trigger + is now session-counted, not per-session: this window registers with the + per-store ``GroomerCoordinator`` and asks it to ensure a cycle. The + coordinator writes the period stamp under a per-store lock BEFORE + spawning, so two concurrent sessions produce exactly one cycle per + ``CORTEX_CONSOLIDATE_TTL_HOURS`` window (default 6h) — fixing the old + ``"(in-flight)"``-marker race where a second session read the stamp as + never-run and spawned a duplicate. + + Degrade honestly (#171): if the coordinator path raises for ANY reason, + fall back to the legacy per-session stamp spawn with a logged NOTICE — + never silently skip grooming entirely. + """ + try: + from mcp_server.infrastructure.groomer_coordinator import ( + GroomerCoordinator, + resolve_store_key, + ) + + coord = GroomerCoordinator(resolve_store_key()) + coord.register(os.getpid()) + outcome = coord.ensure_cycle( + period_hours=_CONSOLIDATE_TTL_HOURS, + spawn_fn=_spawn_consolidate_cycle, + ) + _log(f"groomer coordinator: {outcome}") + except Exception as exc: + _log( + f"NOTICE: groomer coordinator unavailable ({exc}); falling back " + "to legacy per-session consolidate spawn" + ) + _legacy_background_consolidate() + + +def _legacy_background_consolidate() -> None: + """Pre-#171 per-session stamp spawn — the honest degrade path. + + Kept as the fallback the coordinator degrades to (never a silent skip). + Spawns the cycle when the global ``.last_consolidate`` stamp is older + than the TTL. Retains the crude ``"(in-flight)"`` marker: under this + path (coordinator wholly unavailable) it is still strictly better than + no guard at all. """ try: from mcp_server.hooks.consolidate_background import ( @@ -815,35 +876,6 @@ def _maybe_background_consolidate() -> None: if age_hours < _CONSOLIDATE_TTL_HOURS: return # Fresh enough; skip. - # Locate the launcher (same as background reanalyze). - plugin_root = os.environ.get("CLAUDE_PLUGIN_ROOT") or str( - Path(__file__).resolve().parents[2] - ) - launcher = Path(plugin_root) / "scripts" / "launcher.py" - if not launcher.exists(): - # Fall back to direct python invocation with PYTHONPATH — - # works when the dev source is the package root. - launcher = None - - py = ( - __import__("shutil").which("python3") - or __import__("shutil").which("python") - or sys.executable - ) - if launcher is not None: - cmd = [ - py, - str(launcher), - "mcp_server.hooks.consolidate_background", - ] - else: - cmd = [py, "-m", "mcp_server.hooks.consolidate_background"] - - log_path = Path.home() / ".claude" / "methodology" / "consolidate.log" - log_path.parent.mkdir(parents=True, exist_ok=True) - # Touch the stamp so a *second* SessionStart racing this one - # doesn't spawn a duplicate worker. The background worker - # overwrites the stamp on its own completion. try: STAMP_PATH.parent.mkdir(parents=True, exist_ok=True) STAMP_PATH.write_text( @@ -855,14 +887,7 @@ def _maybe_background_consolidate() -> None: ) except OSError: pass - subprocess.Popen( # noqa: S603 — cmd built from trusted sources - cmd, - stdin=subprocess.DEVNULL, - stdout=open(log_path, "a"), - stderr=subprocess.STDOUT, - start_new_session=True, - ) - _log(f"background consolidate spawned → {log_path}") + _spawn_consolidate_cycle() except Exception as exc: _log(f"background consolidate skipped: {exc}") diff --git a/mcp_server/infrastructure/groomer_coordinator.py b/mcp_server/infrastructure/groomer_coordinator.py new file mode 100644 index 00000000..8455e308 --- /dev/null +++ b/mcp_server/infrastructure/groomer_coordinator.py @@ -0,0 +1,309 @@ +"""Session-counted shared groomer coordinator (issue #171). + +Problem it fixes: ``session_start`` used to spawn the 6-hour consolidate +cycle PER session, gated only by a global ``.last_consolidate`` stamp whose +in-flight marker (``"...T... (in-flight)"``) is *unparseable* by +``read_stamp`` — so a second session opening while the first cycle is in +flight reads the stamp as ``None`` (never-run), judges it stale, and spawns +a DUPLICATE cycle against the same store. + +This replaces that race with a session-counted, per-store coordinator (CBM's +session-coordination semantics, narrowed to Cortex's one-groomer-per-store +need — NOT a general service daemon): + +* **first session ensures the groomer runs** — ``ensure_cycle`` spawns only + when the period has elapsed AND no cycle is already running. +* **each session registers / deregisters** — ``register`` on SessionStart, + ``deregister`` on SessionEnd; one liveness-validated file per session. +* **last exit stops it** — ``stop_if_last`` clears the active marker (and + invokes an injected ``stop_fn``) when the final session deregisters. +* **exactly one cycle per period across N sessions** — the period stamp is + written UNDER a per-store lock as a *valid parseable* timestamp, so a + concurrent session that later takes the lock reads it fresh and skips. The + lock serialises simultaneous decisions; the stamp serialises sequential. + +Crash safety: a ``kill -9``'d session leaks its registration file but not its +liveness — ``live_session_count`` sweeps dead-pid registrations before +counting (mirroring ``session_registry.purge_dead_entries``), and the +single-instance guard is a pid file validated by liveness, never a bare flag. + +Layer: infrastructure (all I/O). Policy half of the policy/mechanism split; +fs + lock primitives live in ``groomer_coordinator_io.py``. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path + +from mcp_server.infrastructure.groomer_coordinator_io import ( + atomic_write_json, + atomic_write_text, + decision_lock, + parse_iso, + pid_alive, +) +from mcp_server.shared.platform import cache_dir + +# Outcomes of ``ensure_cycle`` — a small closed set of strings so callers +# and tests can branch/observe without importing an Enum across the +# boundary. source: design #171 (this module). +STARTED = "started" +SKIPPED_FRESH = "skipped_fresh" # period not yet elapsed +SKIPPED_RUNNING = "skipped_running" # a cycle is already in flight (single-instance) +SKIPPED_LOCKED = "skipped_locked" # another session holds the decision lock right now + +_SCHEMA_VERSION = 1 # registration-file schema; unknown versions ignored on read. + + +def resolve_store_key(env: dict[str, str] | None = None) -> str: + """Filesystem-safe key identifying the store this coordinator guards. + + precondition: none. postcondition: returns a stable 16-hex-char token + derived from the resolved store identity — the SQLite DB path on the + SQLite backend, the ``DATABASE_URL`` on PostgreSQL — so two windows + against the SAME store share a coordinator dir and two windows against + DIFFERENT stores never collide. Degrades to a fixed ``"default"`` key + (never raises) when backend resolution fails, so coordination still + happens (one shared coordinator) rather than silently splitting. + """ + import hashlib + import os + + e = env if env is not None else dict(os.environ) + try: + from mcp_server.infrastructure.backend_marker import effective_backend + from mcp_server.infrastructure.memory_config import get_memory_settings + + settings = get_memory_settings() + # str(...) coercion: MemorySettings is a pydantic BaseSettings whose + # attributes the type checker resolves as Unknown; these values ARE + # strings, and coercing makes ``identity`` a definite ``str`` (never + # ``str | None``) so the hash below is well-typed. ``get(k) or dflt`` + # (not ``get(k, dflt)``) keeps the same narrowing for the env case. + if effective_backend(e) == "sqlite": + identity = str(settings.SQLITE_FALLBACK_PATH) + else: + identity = e.get("DATABASE_URL") or str(settings.DATABASE_URL) + except Exception: + return "default" + return hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16] + + +class GroomerCoordinator: + """Per-store session-counting coordinator for the consolidate cycle. + + Construct with an explicit ``store_key`` (see ``resolve_store_key``); + ``root`` is overridable for tests. All state lives under + ``root//``: ``sessions/.json`` registrations, a + ``groomer.lock`` decision lock, a ``groomer.pid`` single-instance + guard, a ``.last_consolidate`` period stamp, and an append-only + ``runs.log`` (NDJSON) for operational duplicate-run observability. + """ + + def __init__(self, store_key: str, *, root: Path | None = None) -> None: + base_root = ( + root if root is not None else cache_dir() / "cortex" / "groomer-coordinator" + ) + self.base_dir = base_root / store_key + self.sessions_dir = self.base_dir / "sessions" + self.lock_path = self.base_dir / "groomer.lock" + self.pid_path = self.base_dir / "groomer.pid" + self.stamp_path = self.base_dir / ".last_consolidate" + self.log_path = self.base_dir / "runs.log" + + # ── session registration (crash-safe counting) ────────────────────── + + def register(self, session_pid: int) -> bool: + """Register a live session. Idempotent per pid. + + postcondition: ``sessions/.json`` exists holding + ``{v, pid, registered_at}``, written atomically. Returns False + (never raises) on I/O failure, so the caller can degrade to legacy + per-session behaviour with a logged NOTICE. + """ + payload = { + "v": _SCHEMA_VERSION, + "pid": int(session_pid), + "registered_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + } + return atomic_write_json( + self.sessions_dir / f"{int(session_pid)}.json", payload + ) + + def deregister(self, session_pid: int) -> None: + """Remove a session's registration. Idempotent; never raises.""" + try: + (self.sessions_dir / f"{int(session_pid)}.json").unlink() + except OSError: + pass + + def live_session_count(self) -> int: + """Count live registrations, reclaiming dead-pid files first. + + postcondition: every ``sessions/.json`` whose ``pid`` is no + longer alive (crash / kill -9) is unlinked; returns the count of + the survivors. Never raises — an unreadable dir counts as zero. + """ + live = 0 + for path, pid in self._iter_session_files(): + if pid_alive(pid): + live += 1 + else: + try: + path.unlink() + except OSError: + pass + return live + + def _iter_session_files(self): + """Yield ``(path, pid)`` for valid registration files. Never raises.""" + try: + entries = list(self.sessions_dir.iterdir()) + except OSError: + return + for entry in entries: + if entry.suffix != ".json": + continue + try: + yield entry, int(entry.stem) + except ValueError: + continue + + # ── single-instance guard ─────────────────────────────────────────── + + def is_groomer_running(self) -> bool: + """True iff ``groomer.pid`` names a live process (liveness-validated, + not a bare flag). A stale pid file from a crashed cycle names a dead + pid and reads as not-running. Never raises.""" + try: + raw = self.pid_path.read_text(encoding="utf-8").strip() + return pid_alive(int(raw)) + except (OSError, ValueError): + return False + + # ── the exactly-one-per-period gate ───────────────────────────────── + + def ensure_cycle( + self, *, period_hours: float, spawn_fn, now: datetime | None = None + ) -> str: + """Ensure at most one grooming cycle runs per ``period_hours``. + + precondition: ``spawn_fn()`` starts the consolidate cycle and + returns its pid (or None). postcondition: returns exactly one of + ``STARTED`` / ``SKIPPED_FRESH`` / ``SKIPPED_RUNNING`` / + ``SKIPPED_LOCKED``. ``spawn_fn`` is called AT MOST once, and only on + ``STARTED``. Under the per-store lock the period stamp is (re)written + as a valid ISO timestamp BEFORE spawning, so any concurrent session + that subsequently takes the lock observes a fresh stamp and returns + ``SKIPPED_FRESH`` — the invariant guaranteeing one cycle per period + across N sessions. invariant: no path both writes the stamp and + returns a SKIPPED_*. + """ + self.base_dir.mkdir(parents=True, exist_ok=True) + now = now or datetime.now(timezone.utc) + with decision_lock(self.lock_path) as acquired: + if not acquired: + return SKIPPED_LOCKED + if self.is_groomer_running(): + self._log_run(SKIPPED_RUNNING, now) + return SKIPPED_RUNNING + if not self._period_elapsed(period_hours, now): + return SKIPPED_FRESH + # Commit the period BEFORE spawning: the stamp is the barrier. + self._write_stamp(now) + pid = spawn_fn() + if pid is not None: + atomic_write_text(self.pid_path, str(int(pid))) + self._log_run(STARTED, now, session_pid=pid) + return STARTED + + def _period_elapsed(self, period_hours: float, now: datetime) -> bool: + last = self._read_stamp() + if last is None: + return True + return (now - last).total_seconds() / 3600.0 >= period_hours + + # ── last-exit stop ────────────────────────────────────────────────── + + def stop_if_last(self, session_pid: int, *, stop_fn=None) -> bool: + """Deregister ``session_pid``; if it was the last live session, stop. + + postcondition: the session's registration is removed; when no live + session remains, the ``groomer.pid`` single-instance marker is + cleared and ``stop_fn`` (if given) is invoked. Returns True iff the + stop path fired (this was the last session). Never raises. + """ + self.deregister(session_pid) + if self.live_session_count() > 0: + return False + try: + self.pid_path.unlink() + except OSError: + pass + if stop_fn is not None: + try: + stop_fn() + except Exception: + pass + self._log_run("stopped_last_exit", datetime.now(timezone.utc)) + return True + + # ── observability (24h zero-duplication evidence) ─────────────────── + + def count_cycles_since(self, since: datetime) -> int: + """Number of ``STARTED`` cycles logged at/after ``since``. + + Operational evidence for the issue's "zero duplicated consolidate + runs over 24h" criterion: with coordination working, this returns at + most one per period window. Never raises — a missing/corrupt log line + is skipped, not fatal. + """ + count = 0 + try: + lines = self.log_path.read_text(encoding="utf-8").splitlines() + except OSError: + return 0 + for line in lines: + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + if rec.get("outcome") != STARTED: + continue + ts = parse_iso(rec.get("ts")) + if ts is not None and ts >= since: + count += 1 + return count + + def _log_run( + self, outcome: str, now: datetime, *, session_pid: int | None = None + ) -> None: + rec: dict[str, object] = { + "ts": now.isoformat(timespec="seconds"), + "outcome": outcome, + } + if session_pid is not None: + rec["session_pid"] = int(session_pid) + try: + self.base_dir.mkdir(parents=True, exist_ok=True) + with self.log_path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(rec) + "\n") + except OSError: + pass + + # ── stamp primitives ──────────────────────────────────────────────── + + def _read_stamp(self) -> datetime | None: + try: + raw = self.stamp_path.read_text(encoding="utf-8").strip() + except OSError: + return None + return parse_iso(raw) + + def _write_stamp(self, now: datetime) -> None: + # Valid ISO only — NEVER an unparseable "(in-flight)" suffix (that + # was exactly the #171 duplication bug: read-back returned None → + # concurrent re-spawn). source: design #171 (this module). + atomic_write_text(self.stamp_path, now.isoformat(timespec="seconds")) diff --git a/mcp_server/infrastructure/groomer_coordinator_io.py b/mcp_server/infrastructure/groomer_coordinator_io.py new file mode 100644 index 00000000..86be20e6 --- /dev/null +++ b/mcp_server/infrastructure/groomer_coordinator_io.py @@ -0,0 +1,155 @@ +"""Low-level I/O + lock primitives for the groomer coordinator (issue #171). + +Mechanism half of the policy/mechanism split (coding-standards §1.1 SRP): +``groomer_coordinator.py`` owns the session-counting POLICY; this module +owns the filesystem + cross-platform locking MECHANISM it stands on — +pid-liveness, atomic replace, ISO parsing, and a non-blocking per-store +lock. Kept separate so the policy file reasons about coordination without +carrying the platform ``fcntl``/``msvcrt`` split inline. + +Layer: infrastructure (all I/O). Imports shared/ + stdlib only. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from datetime import datetime, timezone +from pathlib import Path + + +def pid_alive(pid: int) -> bool: + """True iff ``pid`` currently names a live process. Never raises. + + A ``PermissionError`` (pid exists, owned by another user) counts as + alive — same discipline as ``session_registry._pid_alive`` (a peer infra + module); duplicated rather than importing a private symbol. + """ + if pid <= 0: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + return True + + +def parse_iso(raw: str | None) -> datetime | None: + """Parse an ISO-8601 stamp to an aware UTC datetime, or None. Never raises.""" + if not raw: + return None + try: + ts = datetime.fromisoformat(raw) + except ValueError: + return None + return ts.replace(tzinfo=timezone.utc) if ts.tzinfo is None else ts + + +def atomic_write_text(path: Path, text: str) -> bool: + """Atomically replace ``path`` with ``text`` (tmp + ``os.replace``). + + postcondition: a concurrent reader observes either the old content or + the new content in full, never a torn write. Returns False (never + raises) on any I/O failure. + """ + try: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(prefix=path.name + ".tmp.", dir=str(path.parent)) + try: + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(text) + except Exception: + os.unlink(tmp) + raise + os.replace(tmp, path) + except OSError: + return False + return True + + +def atomic_write_json(path: Path, payload: dict) -> bool: + """Atomically replace ``path`` with the JSON of ``payload``.""" + return atomic_write_text(path, json.dumps(payload)) + + +class decision_lock: + """Non-blocking per-store lock context manager. + + ``with decision_lock(path) as acquired:`` yields True when this holder + won the lock and False when another holder already owns it (contended + simultaneous decision). Advisory ``flock`` on POSIX, mandatory + ``msvcrt.locking`` on Windows — same split as ``pipeline_install_lock`` + (source: RAPPORT_INSTALLATION_CORTEX_WINDOWS.md §5.4). Never raises out + of ``__enter__``: a lock-open failure degrades to "not acquired" so the + caller skips rather than crashes. + """ + + def __init__(self, path: Path) -> None: + self._path = path + self._fd: int | None = None + + def __enter__(self) -> bool: + try: + self._path.parent.mkdir(parents=True, exist_ok=True) + self._fd = os.open(str(self._path), os.O_RDWR | os.O_CREAT, 0o644) + except OSError: + self._fd = None + return False + if _try_lock(self._fd): + return True + os.close(self._fd) + self._fd = None + return False + + def __exit__(self, *exc) -> None: + if self._fd is None: + return + _unlock(self._fd) + try: + os.close(self._fd) + except OSError: + pass + self._fd = None + + +# Direct ``sys.platform`` comparison (not the ``IS_WINDOWS`` alias): the type +# checker statically prunes the unreachable branch per its configured target +# platform, so ``msvcrt``/``fcntl`` are each only analysed where they exist — +# an imported bool alias would not enable that narrowing. +if sys.platform == "win32": + import msvcrt + + def _try_lock(fd: int) -> bool: + try: + msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) + return True + except OSError: + return False + + def _unlock(fd: int) -> None: + try: + msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) + except OSError: + pass + +else: + import fcntl + + def _try_lock(fd: int) -> bool: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except (BlockingIOError, OSError): + return False + + def _unlock(fd: int) -> None: + try: + fcntl.flock(fd, fcntl.LOCK_UN) + except OSError: + pass diff --git a/tests_py/infrastructure/test_groomer_coordinator.py b/tests_py/infrastructure/test_groomer_coordinator.py new file mode 100644 index 00000000..f342e553 --- /dev/null +++ b/tests_py/infrastructure/test_groomer_coordinator.py @@ -0,0 +1,286 @@ +"""Tests for the session-counted shared groomer coordinator (issue #171). + +Done criteria proven here: + * two concurrent sessions produce EXACTLY ONE grooming cycle per period + (``test_two_sessions_one_cycle_per_period`` + the real two-process + ``test_two_processes_one_cycle`` for genuine lock contention); + * a killed (dead-pid) session's registration is reclaimed by the next + session's liveness sweep and grooming continues + (``test_crash_dead_pid_reclaimed`` / ``test_crash_does_not_block_cycle``); + * deregistering all sessions stops the groomer + (``test_last_exit_stops_groomer``). +""" + +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import datetime, timedelta, timezone + +import pytest + +from mcp_server.infrastructure.groomer_coordinator import ( + SKIPPED_FRESH, + SKIPPED_RUNNING, + STARTED, + GroomerCoordinator, + resolve_store_key, +) + + +def test_resolve_store_key_resolves_real_identity_not_default(): + """The SQLite store key must be a real 16-hex hash of the store path, NOT + the ``"default"`` degrade sentinel. Guards the regression where an + invented import symbol made ``resolve_store_key`` raise ImportError and + silently collapse every store onto one shared ``"default"`` key.""" + key = resolve_store_key({"CORTEX_MEMORY_STORE_BACKEND": "sqlite"}) + assert key != "default" + assert len(key) == 16 + assert all(c in "0123456789abcdef" for c in key) + + +def test_resolve_store_key_distinct_pg_urls_distinct_keys(): + """Two different PostgreSQL DATABASE_URLs resolve to different keys — the + per-store isolation the coordinator depends on.""" + a = resolve_store_key({"DATABASE_URL": "postgresql://h/db_a"}) + b = resolve_store_key({"DATABASE_URL": "postgresql://h/db_b"}) + assert a != b and a != "default" and b != "default" + + +_PERIOD_H = 6.0 + + +def _coord(tmp_path, store_key="store-a"): + return GroomerCoordinator(store_key, root=tmp_path) + + +class _SpawnCounter: + """A fake ``spawn_fn``: counts calls, returns a live pid (this process's + own, so ``is_groomer_running`` reads it as alive) unless ``dead`` is set.""" + + def __init__(self, *, dead: bool = False): + self.calls = 0 + self.dead = dead + + def __call__(self): + self.calls += 1 + # A dead pid lets us exercise the single-instance guard NOT latching + # on a crashed cycle; a live pid (os.getpid) latches it. + return -1 if self.dead else os.getpid() + + +# ── done criterion 1: exactly one cycle per period across N sessions ───── + + +def test_two_sessions_one_cycle_per_period(tmp_path): + """Two sessions against the same store: the first ensures the cycle, the + second sees a fresh period stamp and skips — spawn runs exactly once.""" + store = resolve_store_key({"CORTEX_MEMORY_STORE_BACKEND": "sqlite"}) + a = _coord(tmp_path, store) + b = _coord(tmp_path, store) + spawn = _SpawnCounter() + + a.register(1111) + b.register(2222) + + now = datetime.now(timezone.utc) + r1 = a.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=now) + r2 = b.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=now) + + assert r1 == STARTED + assert r2 in (SKIPPED_FRESH, SKIPPED_RUNNING) + assert spawn.calls == 1 # EXACTLY ONE grooming cycle this period + + +def test_new_cycle_after_period_elapses(tmp_path): + """Once the period elapses, the next ensure spawns again — the gate is a + period, not a permanent latch.""" + c = _coord(tmp_path) + spawn = _SpawnCounter(dead=True) # dead pid → no single-instance latch + now = datetime.now(timezone.utc) + + assert c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=now) == STARTED + # Still within the period → skipped. + assert ( + c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=now) == SKIPPED_FRESH + ) + later = now + timedelta(hours=_PERIOD_H + 0.1) + assert c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=later) == STARTED + assert spawn.calls == 2 + + +def test_single_instance_guard_blocks_second_spawn_same_period(tmp_path): + """A still-running cycle (live pid) blocks a second spawn even at the + period edge — single-instance enforcement via pid liveness.""" + c = _coord(tmp_path) + spawn = _SpawnCounter() # live pid → latches groomer.pid + now = datetime.now(timezone.utc) + assert c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=now) == STARTED + assert c.is_groomer_running() is True + later = now + timedelta(hours=_PERIOD_H + 1) # period elapsed... + # ...but the previous cycle is still alive → single-instance skip. + assert ( + c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=later) + == SKIPPED_RUNNING + ) + assert spawn.calls == 1 + + +# ── done criterion 2: crash-safe registration reclaim ──────────────────── + + +def test_crash_dead_pid_reclaimed(tmp_path): + """A registration for a dead pid is swept and does not count as live.""" + c = _coord(tmp_path) + c.register(999999999) # a pid that is not alive + c.register(os.getpid()) # a live one + assert c.live_session_count() == 1 # dead one reclaimed + # The dead registration file is gone after the sweep. + assert not (c.sessions_dir / "999999999.json").exists() + + +def test_crash_does_not_leave_groomer_latched(tmp_path): + """A crashed cycle (dead groomer.pid) must not latch the single-instance + guard forever — the next period's ensure spawns again.""" + c = _coord(tmp_path) + dead = _SpawnCounter(dead=True) + now = datetime.now(timezone.utc) + assert c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=dead, now=now) == STARTED + # groomer.pid names a dead pid → not running. + assert c.is_groomer_running() is False + later = now + timedelta(hours=_PERIOD_H + 0.1) + assert c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=dead, now=later) == STARTED + assert dead.calls == 2 # grooming continued after the crash + + +def test_crash_next_session_still_grooms(tmp_path): + """Simulated kill -9: session A registers then dies; session B's sweep + reclaims A and B can still ensure a cycle.""" + store = "crash-store" + a = _coord(tmp_path, store) + a.register(888888888) # A (will be treated as dead) + b = _coord(tmp_path, store) + b.register(os.getpid()) + assert b.live_session_count() == 1 # A reclaimed + spawn = _SpawnCounter(dead=True) + assert b.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn) == STARTED + + +# ── done criterion 3: last exit stops the groomer ──────────────────────── + + +def test_last_exit_stops_groomer(tmp_path): + """Deregister all sessions → the last one triggers stop; earlier ones + do not.""" + c = _coord(tmp_path) + c.register(1) + c.register(2) + # Simulate both pids being alive by using real live pids is overkill; + # instead assert the stop fires only when no live registration remains. + # Use dead pids so live_session_count reflects only what we deregister. + stopped = {"n": 0} + + def stop_fn(): + stopped["n"] += 1 + + # pids 1 and 2 are (almost certainly) not this test's live children; + # live_session_count sweeps them, so the FIRST stop_if_last already + # finds zero live and fires. Model the count explicitly with live pids: + c2 = _coord(tmp_path, "last-exit-store") + p_live = os.getpid() + c2.register(p_live) + # Register a second *live* pid: spawn a short-lived sleeper. + sleeper = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + try: + c2.register(sleeper.pid) + # First exit (this process pid): sleeper still alive → NOT last. + assert c2.stop_if_last(p_live, stop_fn=stop_fn) is False + assert stopped["n"] == 0 + # Second exit (sleeper): now zero live → last exit stops. + assert c2.stop_if_last(sleeper.pid, stop_fn=stop_fn) is True + assert stopped["n"] == 1 + finally: + sleeper.terminate() + sleeper.wait(timeout=5) + + +def test_stop_clears_single_instance_marker(tmp_path): + """Last-exit stop clears groomer.pid so a stale marker cannot survive + the last session.""" + c = _coord(tmp_path) + spawn = _SpawnCounter() # latches groomer.pid with a live pid + c.register(os.getpid()) + c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn) + assert c.pid_path.exists() + c.stop_if_last(os.getpid()) + assert not c.pid_path.exists() + + +# ── observability: duplicate-run counter over a window ─────────────────── + + +def test_cycle_count_over_window_is_one_per_period(tmp_path): + """The runs.log-backed counter reports exactly one STARTED per period — + the mechanism for the issue's 24h zero-duplication operational evidence.""" + c = _coord(tmp_path) + spawn = _SpawnCounter(dead=True) + base = datetime.now(timezone.utc) - timedelta(hours=24) + # 24h / 6h period = at most 4 cycles; drive 24 hourly session opens. + for h in range(24): + now = base + timedelta(hours=h) + for _sess in range(3): # three concurrent-ish sessions per hour + c.ensure_cycle(period_hours=_PERIOD_H, spawn_fn=spawn, now=now) + started = c.count_cycles_since(base - timedelta(minutes=1)) + # 24h split into 6h windows → exactly 4 cycles, never 72 (3*24). + assert started == 4 + + +# ── real two-process contention (genuine concurrency) ──────────────────── + +_CHILD = """ +import sys +from datetime import datetime, timezone +from mcp_server.infrastructure.groomer_coordinator import GroomerCoordinator, STARTED +root = sys.argv[1] +from pathlib import Path +c = GroomerCoordinator("proc-store", root=Path(root)) +now = datetime.now(timezone.utc) +# A spawn_fn that appends a marker file so the parent can count real spawns. +def spawn(): + p = Path(root) / "spawns" + p.parent.mkdir(parents=True, exist_ok=True) + with p.open("a") as f: + f.write("x") + return -1 # dead pid: no single-instance latch, so ONLY the period + # stamp+lock decide — the strict test of the barrier. +out = c.ensure_cycle(period_hours=6.0, spawn_fn=spawn, now=now) +print(out) +""" + + +@pytest.mark.skipif(os.name == "nt", reason="flock semantics differ on Windows") +def test_two_processes_one_cycle(tmp_path): + """Two real OS processes race ``ensure_cycle`` against one store. The + per-store lock + stamp barrier admit exactly one spawn.""" + root = tmp_path / "coord" + env = {**os.environ, "PYTHONPATH": os.getcwd()} + procs = [ + subprocess.Popen( + [sys.executable, "-c", _CHILD, str(root)], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + for _ in range(2) + ] + outs = [p.communicate(timeout=60) for p in procs] + for p in procs: + assert p.returncode == 0, outs + + spawns_file = root / "spawns" + spawn_count = len(spawns_file.read_text()) if spawns_file.exists() else 0 + assert spawn_count == 1, f"expected exactly one spawn, got {spawn_count}: {outs}" + started = sum(1 for out, _err in outs if out.strip() == STARTED) + assert started == 1