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,
same `pgrep -f` pre-check pattern, 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.

    Uses the exact `pgrep -f` pattern the shell helper used, so a daemon started
    by either implementation is found by the other. 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(
            ["pgrep", "-f", "bash {} --subscribe".format(monitor_script)],
            capture_output=True, text=True,
        )
    except OSError as exc:  # pragma: no cover - pgrep missing
        logger.warning("pgrep unavailable: %s", exc)
        return []
    return [line for line in res.stdout.split() if line.strip()]


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