multi-agent-mux/tests/test_issue19_subscriber_daemon_lifecycle.py

#!/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:
    """Same argv-token selection as the conftest reaper (review R-2)."""
    from conftest import _is_subscriber_daemon_argv

    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 _is_subscriber_daemon_argv(args, str(sandbox)):
            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_reaper_selects_only_the_daemon_argv_shape(tmp_path):
    """Review R-2: a process that quotes the words in a brief, or lives under
    a sibling root that is a string prefix, must never be selected."""
    from conftest import _is_subscriber_daemon_argv

    root = str(tmp_path / "sbx1")
    daemon = "bash {}/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh --subscribe --idle-timeout 0".format(root)
    assert _is_subscriber_daemon_argv(daemon, root)
    assert _is_subscriber_daemon_argv("/bin/sh " + daemon.split(" ", 1)[1], root)
    # sibling root that is a prefix of this one
    assert not _is_subscriber_daemon_argv(daemon, root[:-1])
    assert not _is_subscriber_daemon_argv(daemon, root + "0")
    # brief-carrying wrapper that merely quotes the words and the path
    quoting = "python3 -m lib_py.delegation.cli submit --prompt 'note: reconcile.sh --subscribe under {}/.agents/skills/x'".format(root)
    assert not _is_subscriber_daemon_argv(quoting, root)
    # the real reconcile.sh in a different mode
    assert not _is_subscriber_daemon_argv(daemon.replace("--subscribe", "--once"), root)
    assert not _is_subscriber_daemon_argv("", root)

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