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