multi-agent-mux/.agents/skills/lib_py/delegation/daemon.py

"""Subscriber daemon supervision — the python port of lib.sh's `start_watchdog`.

Behaviour is byte-for-byte compatible with the shell original (issue #5 phase 1,
D5): same monitor path, same stderr wording and exit status when it is missing,
the same daemons found by the pre-check (matched exactly since Issue #72, see
find_running), same log file, same "print the PID" contract.
Callers capture stdout with $( ), so nothing else may be printed there.
"""
from __future__ import annotations

import logging
import os
import signal
import subprocess
from typing import List, Optional

logger = logging.getLogger("delegate_job.daemon")

MONITOR_RELPATH = os.path.join(
    ".agents", "skills", "multi-agent-mux-monitor", "scripts", "reconcile.sh"
)
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)


def find_running(monitor_script: str) -> List[str]:
    """PIDs of subscribers already running for this monitor script.

    A subscriber is a process whose command line is `<bash> <monitor_script>
    --subscribe[ ...]`, argv0 a bash, as both this module and the shell helper
    start it. Issue #72: matched exactly (lib.sh _monitor_pids is the shell
    twin the legacy helper uses), not as a `pgrep -f` regex, because the update skill signals what this returns - a daemon of
    another workspace the unescaped path also matched, or a process that only
    names the path in an argument, is not this workspace's monitor. Multiple
    matches are returned in full: the legacy helper echoed every line and
    callers only ever echo it back, so trimming here would be a silent
    behaviour change.
    """
    try:
        res = subprocess.run(
            ["ps", "-A", "-ww", "-o", "pid=", "-o", "args="],
            capture_output=True, text=True,
        )
    except OSError as exc:  # pragma: no cover - ps missing
        logger.warning("ps unavailable: %s", exc)
        return []
    want = "{} --subscribe".format(monitor_script)
    pids = []
    for line in res.stdout.splitlines():
        parts = line.strip().split(" ", 1)
        if len(parts) != 2:
            continue
        argv0, _, rest = parts[1].partition(" ")
        if os.path.basename(argv0) == "bash" and (rest == want or rest.startswith(want + " ")):
            pids.append(parts[0])
    return pids


def _ignore_sighup():  # pragma: no cover - runs in the forked child
    """Match `nohup`: SIG_IGN survives execve, so the daemon outlives its parent."""
    signal.signal(signal.SIGHUP, signal.SIG_IGN)


def spawn(monitor_script: str, workdir: str) -> int:
    """Start the subscriber detached and return its pid.

    One deliberate difference from the shell original: the log directory is
    created if missing. The legacy `>>` redirect failed outright when `.mam/`
    did not exist, so this is strictly more robust, not a contract change.
    """
    log_file = os.path.join(workdir, LOG_RELPATH)
    os.makedirs(os.path.dirname(log_file), exist_ok=True)
    log_handle = open(log_file, "ab")
    try:
        proc = subprocess.Popen(
            ["bash", monitor_script, "--subscribe", "--idle-timeout", "0"],
            cwd=workdir,
            stdin=subprocess.DEVNULL,
            stdout=log_handle,
            stderr=subprocess.STDOUT,
            start_new_session=True,
            preexec_fn=_ignore_sighup,
        )
    finally:
        log_handle.close()
    return proc.pid


def ensure_subscriber_daemon(job_id: str, workdir: Optional[str] = None) -> List[str]:
    """Return the PIDs of the running subscriber, starting one if needed.

    `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. 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
    return [str(spawn(monitor_script, workdir))]