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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff

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

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()

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 = []