From ce315d8c5e7d05e9c9e63cbc204b8d6c07822d88 Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 13 Sep 2026 15:34:56 +0900 Subject: [PATCH] fix(issue-19): bound the subscriber daemon to its workspace, forward SIGTERM, keep sandboxes from leaking it create_session.sh detaches `reconcile.sh --subscribe --idle-timeout 0` per workspace; test sandboxes never reaped it and the daemon neither noticed its workspace vanish nor honoured SIGTERM (bash deferred the signal behind the foreground python child). 202 orphans were found holding public-broker wildcard subscriptions for deleted pytest tmpdirs. - daemon.py / lib.sh: MAM_NO_SUBSCRIBER_DAEMON=1 (literal) skips the spawn but still reports a running daemon; cli prints a stderr note, stdout stays empty. - reconcile.sh: python subscriber runs as a waited background child with a TERM/INT trap (exit 143); the MQTT loop exits when the workspace dir is gone or .mam/monitor.lock is no longer the held inode; SIGTERM/SIGINT handled via the stop event so loop_stop/disconnect run; the polling fallback checks the same and sleeps in a waited child. - conftest: sandbox default MAM_NO_SUBSCRIBER_DAEMON=1, per-sandbox reaper finalizer, session-scope leak guard over the pytest basetemp. - tests: 11 lifecycle tests (tests/test_issue19_subscriber_daemon_lifecycle.py); phase-1 watchdog/daemon characterization tests opt back in. - docs: monitor SKILL.md lifecycle contract + pitfalls, create SKILL.md note. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff --- .agents/skills/lib.sh | 6 +- .agents/skills/lib_py/delegation/cli.py | 3 + .agents/skills/lib_py/delegation/daemon.py | 20 +- .../skills/multi-agent-mux-create/SKILL.md | 2 + .../skills/multi-agent-mux-monitor/SKILL.md | 19 ++ .../scripts/reconcile.sh | 70 +++- tests/conftest.py | 99 +++++- ...est_issue19_subscriber_daemon_lifecycle.py | 316 ++++++++++++++++++ tests/test_issue5_phase1_delegation.py | 13 +- 9 files changed, 536 insertions(+), 12 deletions(-) create mode 100644 tests/test_issue19_subscriber_daemon_lifecycle.py diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index fb8d439..a2fe749 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2007,7 +2007,11 @@ _start_watchdog_legacy() { local pid pid=$(pgrep -f "bash $monitor_script --subscribe" || true) - if [ -z "$pid" ]; then + # Issue #19: MAM_NO_SUBSCRIBER_DAEMON=1 reports a running daemon but never + # starts one (test sandboxes; externally supervised monitors). + if [ -z "$pid" ] && [ "${MAM_NO_SUBSCRIBER_DAEMON:-0}" = "1" ]; then + echo "subscriber daemon not started (MAM_NO_SUBSCRIBER_DAEMON=1)" >&2 + elif [ -z "$pid" ]; then # Start the wildcard monitor subscriber daemon with --idle-timeout 0 (never idle out) # and ensure it runs with $workdir as cwd to anchor relative log paths. local orig_pwd="$PWD" diff --git a/.agents/skills/lib_py/delegation/cli.py b/.agents/skills/lib_py/delegation/cli.py index 6844a09..83a6d6f 100644 --- a/.agents/skills/lib_py/delegation/cli.py +++ b/.agents/skills/lib_py/delegation/cli.py @@ -85,6 +85,9 @@ def main(argv: Optional[List[str]] = None) -> int: # Byte-identical to the legacy shell helper's failure path. print("ERROR: monitor script not found: {}".format(exc.args[0]), file=sys.stderr) return 1 + if not pids and daemon.daemon_disabled(): + print("subscriber daemon not started ({}=1)".format(daemon.NO_DAEMON_ENV), + file=sys.stderr) for pid in pids: print(pid) return 0 diff --git a/.agents/skills/lib_py/delegation/daemon.py b/.agents/skills/lib_py/delegation/daemon.py index 999c94b..630ee9d 100644 --- a/.agents/skills/lib_py/delegation/daemon.py +++ b/.agents/skills/lib_py/delegation/daemon.py @@ -20,6 +20,17 @@ MONITOR_RELPATH = os.path.join( ) LOG_RELPATH = os.path.join(".mam", "multi-agent-mux-monitor.log") +# Issue #19: opt-out switch for the per-workspace subscriber daemon. Test +# sandboxes set it so that `create_session.sh` under test never leaves a +# detached `reconcile.sh --subscribe --idle-timeout 0` behind; an operator can +# set it too when the monitor is supervised elsewhere. Only the literal "1" +# disables the spawn - any other value keeps the normal behaviour. +NO_DAEMON_ENV = "MAM_NO_SUBSCRIBER_DAEMON" + + +def daemon_disabled() -> bool: + return os.environ.get(NO_DAEMON_ENV, "") == "1" + def monitor_script_path(workdir: str) -> str: return os.path.join(workdir, MONITOR_RELPATH) @@ -80,12 +91,19 @@ def ensure_subscriber_daemon(job_id: str, workdir: Optional[str] = None) -> List `job_id` is accepted for signature compatibility and is deliberately unused: the daemon is per-workspace (log paths anchor on `workdir`), exactly as in the shell original. Raises FileNotFoundError when the monitor is absent so - the CLI can reproduce the legacy stderr line and exit 1. + the CLI can reproduce the legacy stderr line and exit 1. With + MAM_NO_SUBSCRIBER_DAEMON=1 (issue #19) nothing is spawned and only the PIDs + of a daemon that already runs are returned. """ workdir = workdir or os.getcwd() monitor_script = monitor_script_path(workdir) if not os.path.isfile(monitor_script): raise FileNotFoundError(monitor_script) + if daemon_disabled(): + # Issue #19: report an already-running daemon if there is one, but + # never start a new one. Callers echo the (possibly empty) PID list. + logger.info("%s=1: not starting the subscriber daemon", NO_DAEMON_ENV) + return find_running(monitor_script) running = find_running(monitor_script) if running: return running diff --git a/.agents/skills/multi-agent-mux-create/SKILL.md b/.agents/skills/multi-agent-mux-create/SKILL.md index 5af6232..e9d51d4 100644 --- a/.agents/skills/multi-agent-mux-create/SKILL.md +++ b/.agents/skills/multi-agent-mux-create/SKILL.md @@ -68,6 +68,8 @@ If any check fails → abort with a non-zero exit and report the reason (automat When running multiple agent sessions alongside other workflows (e.g., cmux, background workers, manual herdr sessions), sharing the default herdr server can lead to session name conflicts, monitoring clutter, and accidental destruction of user sessions via global commands. To prevent this, you can run this skill inside an **isolated herdr session** using the `HERDR_SESSION_NAME` environment variable or the `--herdr-session ` flag (opt-in; alias: `--herdr-server`; legacy env alias: `HERDR_SERVER_NAME`). +On success the script also starts the per-workspace monitor daemon through `start_watchdog` (`reconcile.sh --subscribe --idle-timeout 0`, one per workspace, see the lifecycle contract in [multi-agent-mux-monitor/SKILL.md](../multi-agent-mux-monitor/SKILL.md)). Set `MAM_NO_SUBSCRIBER_DAEMON=1` to skip that spawn when the monitor is supervised elsewhere or the workspace is ephemeral (Issue #19); the `watchdog PID:` line is then empty. + Additionally, you can specify `--herdr-workspace ` (default: workspace slug without `mam-` prefix) to name and group the agent panes inside a dedicated Herdr workspace tab in the Herdr runtime (`herdr workspace create --label` / `rename`) as well as recording it in `.mam/agent-sessions.yaml`. Note that `--herdr-workspace` configures the workspace tab label within the session, whereas `--herdr-session` selects the daemon socket itself. Under the hood this maps to a real, separate herdr **session** (`herdr --session ` — its own socket, its own `agent list`/`workspace list`, completely invisible to the default session and vice versa), not just a workspace label inside the same server. `lib.sh`'s shim bootstraps the named session's server headlessly (`herdr --session server`, backgrounded) the first time it's needed, and scopes every subsequent herdr call to it automatically — this headless bootstrap is what lets it work even when the skill itself is running from inside another herdr-managed pane (a plain interactive `herdr --session ` launch is blocked there by herdr's "nested herdr is disabled" guard; headless `server` mode isn't). diff --git a/.agents/skills/multi-agent-mux-monitor/SKILL.md b/.agents/skills/multi-agent-mux-monitor/SKILL.md index 7c39a27..554dda2 100644 --- a/.agents/skills/multi-agent-mux-monitor/SKILL.md +++ b/.agents/skills/multi-agent-mux-monitor/SKILL.md @@ -69,6 +69,23 @@ Run it under whatever supervisor you already use (a dedicated herdr pane, `nohup or a background job). Nothing else needs to be running for the monitor to work — it reconciles YAML ↔ herdr ↔ disk on its own. +### Daemon lifecycle contract (Issue #19) + +`multi-agent-mux-create` starts one detached `reconcile.sh --subscribe --idle-timeout 0` +per workspace through `lib.sh::start_watchdog` (`lib_py.delegation.daemon`). That daemon: + +- is **one per workspace** — `.mam/monitor.lock` (flock) rejects a second instance, and + `start_watchdog` returns the running PID instead of spawning again; +- **ends on its own only when its workspace is gone** — every wake-up (≤ 5 s on the MQTT + loop, every poll on the broker-down fallback) it checks that the workspace directory + still exists and that `.mam/monitor.lock` is still the inode it holds. A deleted + workspace, a purged `.mam/`, or a re-created lock file makes it disconnect and exit 0; +- **stops promptly on `SIGTERM`/`SIGINT`** — the bash wrapper forwards the signal to the + python subscriber and exits 143. `SIGKILL` is never required; +- is **not started when `MAM_NO_SUBSCRIBER_DAEMON=1`** — the literal `1` only. Use it when + the monitor is supervised elsewhere; a daemon that already runs is still reported. Test + sandboxes set it by default and additionally reap any daemon anchored in the sandbox. + The herdr commands the script issues (`herdr agent list`, `herdr agent get `) are real native herdr commands — do not substitute tmux-era names like `herdr ls` / `herdr list-panes` outside a shell that has sourced `.agents/skills/lib.sh`. @@ -141,6 +158,8 @@ YAML: no such session - **Don't expect `--once` to stay alive** — it does a single pass and exits. Use `--subscribe` for continuous monitoring. - **`--idle-timeout` defaults to 3600s** — a monitor meant to run indefinitely needs `--idle-timeout 0` explicitly, or it will quietly exit after an hour of broker silence. +- **`--idle-timeout 0` is bounded by the workspace, not by time** — the daemon exits when its workspace directory or `.mam/monitor.lock` disappears (Issue #19). Do not delete `.mam/monitor.lock` under a running monitor unless you mean to stop it. +- **Spawning from tests or ephemeral checkouts** — set `MAM_NO_SUBSCRIBER_DAEMON=1` so `create_session.sh` does not detach a daemon that outlives the checkout; the pytest sandbox does this for you. - **The poll interval is a default** — `RECONCILE_POLL_INTERVAL` (15s) is what the broker-down fallback uses. A workspace with 5+ agent sessions can bump it to reduce noise. - **Coalesce repeated drifts** — the same drift re-appears on every pass until it is resolved. A caller that acts on `drifts[]` should compare against the previous pass and act only on *new* entries; the script does not deduplicate for you. - **Don't fight the user's explicit action** — if `multi-agent-mux-stop` is mid-flight and the monitor sees the same session in two states within 5s, prefer the user's most recent action. The monitor should not auto-revert a fresh `terminated` to `running` because of a stale `herdr has-session` check. diff --git a/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh b/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh index 0e7b55d..7a3a288 100755 --- a/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh +++ b/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh @@ -72,12 +72,27 @@ if [ "$SUBSCRIBE" = "1" ]; then # no single interpreter needs both paho and PyYAML (review items 4/5/6). PYBIN="$(_delegate_py_bin)" + # Issue #19: the subscriber runs as a background child and this shell waits + # on it, so SIGTERM/SIGINT reach the daemon promptly. With the child in the + # foreground bash deferred the signal until the child exited on its own, + # which for --idle-timeout 0 was never - only SIGKILL worked. + _sub_pid="" + _sub_stop() { + trap - TERM INT + if [ -n "$_sub_pid" ]; then + kill -TERM "$_sub_pid" 2>/dev/null || true + wait "$_sub_pid" 2>/dev/null || true + fi + exit 143 + } + trap _sub_stop TERM INT + # The MQTT subscribe loop exits 3 to signal "broker unavailable → poll instead". set +e YAML_PATH="$AGENT_SESSIONS_YAML" HOME_DIR="$HOME_DIR" CLAUDE_PROJECT_DIR="$CLAUDE_PROJECT_DIR" LOCAL_BIN="$LOCAL_BIN" \ WORKSPACE_ROOT="$WORKSPACE_ROOT" SUB_TIMEOUT="$SUB_TIMEOUT" SUB_IDLE_TIMEOUT="$SUB_IDLE_TIMEOUT" \ SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" \ - "$PYBIN" - <<'PYEOF' + "$PYBIN" - <<'PYEOF' & import os, sys, json, time, subprocess lib_sh = os.environ.get('LIB_SH', '') @@ -100,6 +115,27 @@ except BlockingIOError: except Exception as e: print(f"MQTT Monitor: failed to acquire monitor lock ({e}). Exiting.", flush=True) sys.exit(1) +lock_ino = os.fstat(lock_file.fileno()).st_ino + + +def _workspace_gone(): + """Issue #19: the daemon serves exactly one workspace. Once that workspace + is deleted (a purged sandbox, an rm -rf'd project) or its `.mam/monitor.lock` + is no longer the inode we hold (a re-created `.mam/`, where a fresh daemon + could now take a new lock), there is nothing left to reconcile - exit + instead of holding a broker subscription forever.""" + try: + root = os.path.abspath(workspace_root) if workspace_root else os.getcwd() + except OSError: + return "cwd no longer exists" + if not os.path.isdir(root): + return f"workspace {root} no longer exists" + try: + if os.stat(lock_file_path).st_ino != lock_ino: + return f"{lock_file_path} was replaced" + except OSError: + return f"{lock_file_path} no longer exists" + return None # Locate skills/multi-agent-mux-delegate-job/scripts to import mqtt_common — relative first, then # an upward walk from cwd. No hardcoded absolute path (review item 6). @@ -271,11 +307,28 @@ if not state['connected']: client.loop_stop() sys.exit(3) -import threading +import threading, signal stop_event = threading.Event() +state['stop_signal'] = None + + +def _on_stop_signal(signum, _frame): + state['stop_signal'] = signum + stop_event.set() + + +signal.signal(signal.SIGTERM, _on_stop_signal) +signal.signal(signal.SIGINT, _on_stop_signal) start = time.time() try: while True: + if state['stop_signal'] is not None: + print(f"MQTT Monitor: signal {state['stop_signal']} received, exiting", flush=True) + break + gone = _workspace_gone() + if gone: + print(f"MQTT Monitor: {gone}; exiting", flush=True) + break now = time.time() wall_left = (timeout - (now - start)) if timeout else None idle_left = (idle_timeout - (now - state['last_msg'])) if idle_timeout else None @@ -302,19 +355,30 @@ finally: pass sys.exit(0) PYEOF + _sub_pid=$! + wait "$_sub_pid" sub_rc=$? + _sub_pid="" set -e if [ "$sub_rc" = "3" ]; then echo "MQTT Monitor: broker unavailable — falling back to polling (interval ${POLL_INTERVAL}s)" >&2 _self="$SKILLS_DIR/multi-agent-mux-monitor/scripts/reconcile.sh" _start=$(date +%s) + # Issue #19: same lifecycle as the MQTT loop - stop when the workspace is + # gone, and sleep in a waited child so SIGTERM/SIGINT are not deferred. + trap 'trap - TERM INT; exit 143' TERM INT while :; do + if [ ! -d "$WORKSPACE_ROOT" ] || [ ! -f "$WORKSPACE_ROOT/.mam/monitor.lock" ]; then + echo "MQTT Monitor: workspace $WORKSPACE_ROOT or its monitor.lock is gone; exiting" >&2 + break + fi bash "$_self" --once --emit-diff >/dev/null 2>&1 || true if [ "$SUB_TIMEOUT" != "0" ] && [ "$(( $(date +%s) - _start ))" -ge "$SUB_TIMEOUT" ]; then break fi - sleep "$POLL_INTERVAL" + sleep "$POLL_INTERVAL" & + wait $! || true done fi exit 0 diff --git a/tests/conftest.py b/tests/conftest.py index 93f144d..6b1798c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,13 +11,89 @@ SKILLS_DIR = str(Path(__file__).resolve().parent.parent / ".agents" / "skills") if SKILLS_DIR not in sys.path: sys.path.insert(0, SKILLS_DIR) +def _subscriber_daemon_pids(root): + """PIDs of `reconcile.sh --subscribe` daemons whose command line lives under + `root`, plus their children (the python subscriber the bash wrapper waits on). + + Issue #19: `create_session.sh` starts one such daemon per workspace, + detached into its own session with `--idle-timeout 0`. A sandbox that ran + it must reap it, or every suite run leaves ~30 daemons holding wildcard + subscriptions on the default public broker behind.""" + import subprocess as _sp + root = str(root) + try: + out = _sp.run(["pgrep", "-f", "reconcile.sh --subscribe"], + capture_output=True, text=True).stdout.split() + except OSError: + return [] + found = [] + for pid in out: + try: + args = _sp.run(["ps", "-o", "args=", "-p", pid], + capture_output=True, text=True).stdout + except OSError: + continue + if root in args: + found.append(int(pid)) + kids = [] + for pid in found: + try: + kids += [int(k) for k in _sp.run(["pgrep", "-P", str(pid)], + capture_output=True, text=True).stdout.split()] + except (OSError, ValueError): + pass + return found + kids + + +def _reap_subscriber_daemons(root): + """Terminate (then kill) every subscriber daemon anchored under `root`. + Returns the PIDs that were still alive before the sweep.""" + import signal as _sig + import time as _t + pids = _subscriber_daemon_pids(root) + for sig in (_sig.SIGTERM, _sig.SIGKILL): + for pid in pids: + try: + os.kill(pid, sig) + except OSError: + pass + deadline = _t.time() + 3 + while _t.time() < deadline and any(_pid_alive(p) for p in pids): + _t.sleep(0.1) + if not any(_pid_alive(p) for p in pids): + break + return pids + + +def _pid_alive(pid): + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + # a zombie still answers kill(0); ps state Z means it is gone for our purposes + import subprocess as _sp + try: + state = _sp.run(["ps", "-o", "stat=", "-p", str(pid)], + capture_output=True, text=True).stdout.strip() + except OSError: + return True + return bool(state) and not state.startswith("Z") + + @pytest.fixture -def mam_sandbox(tmp_path, monkeypatch): +def mam_sandbox(tmp_path, monkeypatch, request): """ Manages a temporary sandboxed directory for .mam/ configuration files and copies the scripts from .agents/skills to prevent contaminating the real codebase. Also overrides environments (HOME, PATH, AGENT_SESSIONS_YAML, etc.) for isolation. """ + # Issue #19: reap any subscriber daemon a test spawned under this sandbox, + # whatever path it came from (create_session.sh, start_watchdog, a direct + # `reconcile.sh --subscribe`). Registered first so it runs after every + # other finalizer of the test. + request.addfinalizer(lambda: _reap_subscriber_daemons(tmp_path)) # 1. Copy the skill scripts to sandboxed tmp_path/skills and tmp_path/.agents/skills src_skills = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".agents", "skills")) shutil.copytree(src_skills, tmp_path / "skills") @@ -52,6 +128,10 @@ def mam_sandbox(tmp_path, monkeypatch): # exist, so importing or running delegation code inside a sandbox never injects the # real .mam.env into this process's os.environ. monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent")) + # Issue #19: never let code under test detach a real subscriber daemon + # (`reconcile.sh --subscribe --idle-timeout 0`) from a sandbox. Tests that + # exercise the daemon itself override this with "0" or delenv. + monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", "1") monkeypatch.delenv("HERDR_SESSION_NAME", raising=False) monkeypatch.delenv("HERDR_SERVER_NAME", raising=False) monkeypatch.delenv("HERDR_WORKSPACE_ID", raising=False) @@ -1274,6 +1354,23 @@ def _drop_ambient_mam_version(): os.environ["MAM_VERSION"] = previous +@pytest.fixture(scope="session", autouse=True) +def subscriber_daemon_leak_guard(tmp_path_factory): + """Issue #19: no `reconcile.sh --subscribe` daemon anchored in this session's + pytest basetemp may survive the suite. Mirrors the #15 registry guard: the + per-sandbox finalizer is the fix, this is the regression alarm. Leftovers + are reaped after being reported so a failing run still cleans up.""" + yield + basetemp = str(Path(tmp_path_factory.getbasetemp()).resolve()) + leaked = _subscriber_daemon_pids(basetemp) + if leaked: + _reap_subscriber_daemons(basetemp) + assert not leaked, ( + "Issue #19: {} subscriber daemon process(es) anchored under {} survived the " + "suite (reaped now): {}".format(len(leaked), basetemp, leaked) + ) + + @pytest.fixture(scope="session", autouse=True) def delegation_tree_provenance_guard(tmp_path_factory): yield diff --git a/tests/test_issue19_subscriber_daemon_lifecycle.py b/tests/test_issue19_subscriber_daemon_lifecycle.py new file mode 100644 index 0000000..af7ba4b --- /dev/null +++ b/tests/test_issue19_subscriber_daemon_lifecycle.py @@ -0,0 +1,316 @@ +#!/usr/bin/env python3 +"""Issue #19 — the per-workspace subscriber daemon must not outlive its workspace. + +`create_session.sh` detaches `reconcile.sh --subscribe --idle-timeout 0` into +its own session for every workspace. Three gaps let ~30 of them pile up per +suite run, each holding a wildcard subscription on the default public broker: + + * nothing in the test sandbox stopped the spawn or reaped the daemon; + * the daemon never noticed its workspace (a deleted pytest tmpdir) was gone; + * SIGTERM did not work - bash kept the python child in the foreground and + deferred the signal until the child exited, which with idle-timeout 0 was + never. + +This file pins the fix from all three sides: the MAM_NO_SUBSCRIBER_DAEMON +opt-out that sandboxes set by default, the daemon's self-termination when its +workspace or `.mam/monitor.lock` disappears, and prompt SIGTERM handling on +both the MQTT loop and the broker-down polling fallback. +""" +from __future__ import annotations + +import os +import re +import signal +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +FIXTURES = Path(__file__).parent / "fixtures" / "issue_5" / "phase1" +DELEGATION_MODES = ("0", "1") + + +def _skills(sandbox: Path) -> Path: + return sandbox / ".agents" / "skills" + + +def _reconcile(sandbox: Path) -> Path: + return _skills(sandbox) / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" + + +def _install_fake_monitor(sandbox: Path) -> Path: + target = _reconcile(sandbox) + target.write_text((FIXTURES / "fake_reconcile.sh").read_text(encoding="utf-8"), encoding="utf-8") + target.chmod(0o755) + return target + + +def _run_helper(sandbox: Path, snippet: str, env=None) -> subprocess.CompletedProcess: + run_env = dict(os.environ) + if env: + run_env.update(env) + return subprocess.run( + ["bash", "-c", 'source "{}" >/dev/null 2>&1; {}'.format(_skills(sandbox) / "lib.sh", snippet)], + capture_output=True, text=True, env=run_env, timeout=60, + ) + + +def _daemons_for(sandbox: Path) -> list: + out = subprocess.run(["pgrep", "-f", "reconcile.sh --subscribe"], capture_output=True, text=True).stdout.split() + hits = [] + for pid in out: + args = subprocess.run(["ps", "-o", "args=", "-p", pid], capture_output=True, text=True).stdout + if str(sandbox) in args: + hits.append(int(pid)) + return hits + + +def _alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + state = subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True).stdout.strip() + return bool(state) and not state.startswith("Z") + + +def _kill(pids) -> None: + for pid in pids: + try: + os.kill(int(pid), signal.SIGKILL) + except (OSError, ValueError): + pass + + +def _start_subscriber(sandbox: Path, log: Path, extra_env: dict) -> subprocess.Popen: + env = dict(os.environ) + env.update(extra_env) + handle = open(log, "ab") + try: + return subprocess.Popen( + ["bash", str(_reconcile(sandbox)), "--subscribe", "--idle-timeout", "0"], + cwd=str(sandbox), env=env, stdin=subprocess.DEVNULL, stdout=handle, stderr=subprocess.STDOUT, + start_new_session=True, + ) + finally: + handle.close() + + +def _wait_for(log: Path, pattern: str, timeout: float) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + if log.exists() and re.search(pattern, log.read_text(errors="replace")): + return True + time.sleep(0.1) + return False + + +def _wait_exit(proc: subprocess.Popen, timeout: float): + try: + return proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + return None + + +def _unreachable_broker_env() -> dict: + # Port 9 (discard) is closed on a stock macOS/Linux host: connect() is + # refused at once, so the daemon drops into its polling fallback quickly. + return {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0", "RECONCILE_POLL_INTERVAL": "1"} + + +# --------------------------------------------------------------------------- +# 1. the sandbox opt-out +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("mode", DELEGATION_MODES) +def test_sandbox_default_never_spawns_a_daemon(mam_sandbox, mode): + """With the conftest default MAM_NO_SUBSCRIBER_DAEMON=1, start_watchdog + prints no PID and leaves no process behind, in both helper bodies.""" + assert os.environ.get("MAM_NO_SUBSCRIBER_DAEMON") == "1" + _install_fake_monitor(mam_sandbox) + res = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(mam_sandbox), env={"MAM_DELEGATION_PY": mode}) + assert res.returncode == 0, res.stderr + assert res.stdout.strip() == "", res.stdout + assert "MAM_NO_SUBSCRIBER_DAEMON=1" in res.stderr + assert _daemons_for(mam_sandbox) == [] + + +def test_opt_out_still_reports_a_daemon_that_already_runs(mam_sandbox, monkeypatch): + """Opting out means "do not start", not "pretend nothing runs": an existing + daemon's PID is still returned so callers keep their dedup contract.""" + from lib_py.delegation import daemon + + monitor = _install_fake_monitor(mam_sandbox) + spawned = [] + try: + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) + first = daemon.ensure_subscriber_daemon("job1", str(mam_sandbox)) + spawned += first + assert len(first) == 1 and _alive(int(first[0])) + + monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", "1") + assert daemon.daemon_disabled() + assert daemon.ensure_subscriber_daemon("job1", str(mam_sandbox)) == first + assert daemon.find_running(str(monitor)) == first + finally: + _kill(spawned) + + +def test_only_the_literal_one_disables_the_spawn(mam_sandbox, monkeypatch): + from lib_py.delegation import daemon + + for value in ("0", "", "yes", "true"): + monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", value) + assert not daemon.daemon_disabled(), value + monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", "1") + assert daemon.daemon_disabled() + + +def _create_session(sandbox: Path, name: str) -> subprocess.CompletedProcess: + script = _skills(sandbox) / "multi-agent-mux-create" / "scripts" / "create_session.sh" + return subprocess.run( + ["bash", str(script), "--workspace", str(sandbox), "--agent", "claude", "--role", "Creator", "--session", name], + capture_output=True, text=True, cwd=str(sandbox), timeout=180, + ) + + +def test_create_session_under_sandbox_leaves_no_daemon(mam_sandbox, mock_herdr, mock_agents): + """The leaking path itself: a full create_session.sh run in a sandbox.""" + res = _create_session(mam_sandbox, "issue19-creator-claude") + assert res.returncode == 0, res.stderr + res.stdout + assert "watchdog PID:" in res.stdout + assert _daemons_for(mam_sandbox) == [] + + +def test_create_session_without_opt_out_spawns_a_daemon_the_reaper_removes(mam_sandbox, mock_herdr, mock_agents, monkeypatch): + """Belt and braces: with the opt-out lifted the historic leak is reproduced + (create_session.sh does detach a daemon), and the sandbox reaper takes it + down - so a test that forgets the opt-out still cannot leak.""" + from conftest import _reap_subscriber_daemons + + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) + res = _create_session(mam_sandbox, "issue19-leaky-creator-claude") + assert res.returncode == 0, res.stderr + res.stdout + pids = _daemons_for(mam_sandbox) + try: + assert pids, "expected create_session.sh to detach a subscriber daemon" + reaped = _reap_subscriber_daemons(mam_sandbox) + assert set(pids) <= set(reaped) + deadline = time.time() + 5 + while time.time() < deadline and any(_alive(p) for p in pids): + time.sleep(0.1) + assert not any(_alive(p) for p in pids) + finally: + _kill(pids) + + +# --------------------------------------------------------------------------- +# 2. the sandbox reaper +# --------------------------------------------------------------------------- + +def test_sandbox_reaper_terminates_daemons_anchored_in_the_sandbox(mam_sandbox, monkeypatch): + from conftest import _reap_subscriber_daemons, _subscriber_daemon_pids + from lib_py.delegation import daemon + + _install_fake_monitor(mam_sandbox) + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) + pid = int(daemon.ensure_subscriber_daemon("job1", str(mam_sandbox))[0]) + try: + assert pid in _subscriber_daemon_pids(mam_sandbox) + reaped = _reap_subscriber_daemons(mam_sandbox) + assert pid in reaped + deadline = time.time() + 5 + while time.time() < deadline and _alive(pid): + time.sleep(0.1) + assert not _alive(pid), "reaper left the daemon running" + assert _subscriber_daemon_pids(mam_sandbox) == [] + finally: + _kill([pid]) + + +# --------------------------------------------------------------------------- +# 3. self-termination and signals - polling fallback (no broker needed) +# --------------------------------------------------------------------------- + +def test_poll_fallback_exits_when_monitor_lock_disappears(mam_sandbox, mock_herdr, monkeypatch): + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) + log = mam_sandbox / "sub.log" + proc = _start_subscriber(mam_sandbox, log, _unreachable_broker_env()) + try: + assert _wait_for(log, r"falling back to polling", 30), log.read_text() + lock = mam_sandbox / ".mam" / "monitor.lock" + assert lock.exists() + lock.unlink() + rc = _wait_exit(proc, 15) + assert rc == 0, (rc, log.read_text()) + assert "monitor.lock is gone" in log.read_text() + finally: + if proc.poll() is None: + proc.kill() + + +def test_poll_fallback_sigterm_exits_promptly(mam_sandbox, mock_herdr, monkeypatch): + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) + monkeypatch.setenv("RECONCILE_POLL_INTERVAL", "60") # a foreground sleep would defer the signal for a minute + log = mam_sandbox / "sub.log" + proc = _start_subscriber(mam_sandbox, log, _unreachable_broker_env() | {"RECONCILE_POLL_INTERVAL": "60"}) + try: + assert _wait_for(log, r"falling back to polling", 30), log.read_text() + time.sleep(1.0) # let the loop reach its sleep + t0 = time.time() + proc.send_signal(signal.SIGTERM) + rc = _wait_exit(proc, 10) + assert rc is not None, "SIGTERM was deferred: daemon still running after 10s" + assert time.time() - t0 < 5, "SIGTERM took too long to act" + assert rc == 143, rc + finally: + if proc.poll() is None: + proc.kill() + + +# --------------------------------------------------------------------------- +# 4. self-termination and signals - MQTT loop (needs a reachable broker) +# --------------------------------------------------------------------------- + +def _start_connected(mam_sandbox, log): + proc = _start_subscriber(mam_sandbox, log, {}) + if not _wait_for(log, r"MQTT Monitor: subscribed to ", 45): + proc.kill() + proc.wait(timeout=10) + pytest.skip("broker unreachable from this host: " + log.read_text()[-300:]) + return proc + + +def test_mqtt_loop_exits_when_workspace_lock_is_replaced(mam_sandbox, monkeypatch): + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) + log = mam_sandbox / "sub.log" + proc = _start_connected(mam_sandbox, log) + try: + lock = mam_sandbox / ".mam" / "monitor.lock" + lock.unlink() + lock.write_text("") # a re-created .mam/: a fresh daemon could now take this lock + rc = _wait_exit(proc, 15) + assert rc == 0, (rc, log.read_text()) + assert re.search(r"monitor\.lock (was replaced|no longer exists); exiting", log.read_text()), log.read_text() + finally: + if proc.poll() is None: + proc.kill() + + +def test_mqtt_loop_sigterm_exits_cleanly(mam_sandbox, monkeypatch): + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) + log = mam_sandbox / "sub.log" + proc = _start_connected(mam_sandbox, log) + try: + t0 = time.time() + proc.send_signal(signal.SIGTERM) + rc = _wait_exit(proc, 10) + assert rc is not None, "SIGTERM was deferred: daemon still running after 10s" + assert time.time() - t0 < 6 + assert rc == 143, (rc, log.read_text()) + assert "signal 15 received, exiting" in log.read_text(), log.read_text() + finally: + if proc.poll() is None: + proc.kill() diff --git a/tests/test_issue5_phase1_delegation.py b/tests/test_issue5_phase1_delegation.py index dd08c8e..d5f410e 100644 --- a/tests/test_issue5_phase1_delegation.py +++ b/tests/test_issue5_phase1_delegation.py @@ -194,7 +194,7 @@ def test_c07_watchdog_spawns_daemon_with_workdir_and_log(mam_sandbox, mode): res = _run_helper( mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir), - env={"MAM_DELEGATION_PY": mode}, + env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"}, ) assert res.returncode == 0, res.stderr pid = res.stdout.strip() @@ -219,12 +219,12 @@ def test_c08_watchdog_second_call_reuses_running_daemon(mam_sandbox, mode): spawned = [] try: first = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir), - env={"MAM_DELEGATION_PY": mode}) + env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"}) pid1 = first.stdout.strip() spawned.append(pid1) assert _alive(pid1) second = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir), - env={"MAM_DELEGATION_PY": mode}) + env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"}) assert second.returncode == 0, second.stderr assert second.stdout.strip() == pid1, (first.stdout, second.stdout) matches = subprocess.run( @@ -259,9 +259,9 @@ def test_c09_start_watchdog_and_ensure_subscriber_daemon_agree(mam_sandbox, mode spawned = [] try: a = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir), - env={"MAM_DELEGATION_PY": mode}) + env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"}) b = _run_helper(mam_sandbox, 'ensure_subscriber_daemon job1 "{}"'.format(workdir), - env={"MAM_DELEGATION_PY": mode}) + env={"MAM_DELEGATION_PY": mode, "MAM_NO_SUBSCRIBER_DAEMON": "0"}) spawned += [a.stdout.strip(), b.stdout.strip()] assert a.returncode == b.returncode == 0, (a.stderr, b.stderr) assert a.stdout.strip() == b.stdout.strip(), (a.stdout, b.stdout) @@ -649,10 +649,11 @@ def test_daemon_missing_monitor_raises_with_the_legacy_path(mam_sandbox): assert exc.value.args[0] == str(daemon.monitor_script_path(str(workdir))) -def test_daemon_spawn_dedup_and_survives_parent_exit(mam_sandbox): +def test_daemon_spawn_dedup_and_survives_parent_exit(mam_sandbox, monkeypatch): """The daemon must outlive the shell that started it and never double-spawn.""" from lib_py.delegation import daemon + monkeypatch.delenv("MAM_NO_SUBSCRIBER_DAEMON", raising=False) # issue #19 sandbox default monitor = _install_fake_monitor(mam_sandbox) workdir = str(mam_sandbox) spawned = []