feat(issue-5): DelegationService, cli router, ensure_subscriber_daemon; lib.sh helpers delegate to python

Adds the python side of the delegation helpers and switches lib.sh over to it. daemon.py port of start_watchdog. Same monitor path, same pgrep pattern, so a daemon started by either implementation is found by the other; same stderr wording and exit 1 when reconcile.sh is missing; same "print the PID" contract, multiple pgrep matches included, since the shell echoed them all and trimming would be a silent change. start_new_session detaches it, and SIGHUP is ignored across the exec so it outlives its parent exactly as nohup did. cli.py thin router. submit/publish forward to the existing main() of the module that already owned the behaviour, so exit codes, stdout shape and side effects cannot drift between the shell path and the script path. Only ensure-daemon carries logic of its own. service.py DelegationService facade, with an optional persistent connection for the phase 2a engine. CLI and shell always use one-shot. lib.sh keeps the three legacy bodies verbatim as _*_legacy behind MAM_DELEGATION_PY=0, which is the instant rollback switch. PYTHONPATH is passed explicitly so `-m` resolves lib_py from the tree this lib.sh belongs to, which is what makes a frozen loop run frozen delegation code. start_watchdog survives as an alias for ensure_subscriber_daemon. The wrapper and run_loop.sh are untouched; reconcile.sh gains only the ensure_env_loaded line from the previous commit. Verified: C-01..C-09 pass against both the legacy and the python bodies, the daemon dedups and survives its parent, the cli reproduces the legacy failure line exactly, and publish failure semantics (status synced, seq consumed, audit records published=false) are identical through the shim and the cli. tier1+tier2 109 passed; bash -n clean on all six shell files. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
2026-09-06 15:42:09 +09:00
1 parent fb3498e23b
commit 78cb9ecfa9
@@ -1873,7 +1873,7 @@ _delegate_script() {
 # delegate_submit_job <prompt> <agent> <agent_session>
 #
 # Register a job in the multi-agent-mux-delegate-job registry. Prints the new JID on stdout.
-delegate_submit_job() {
+_delegate_submit_job_legacy() {
  local prompt="$1" agent="$2" session="$3"
  local py_bin registry_py
  py_bin="$(_delegate_py_bin)"
@@ -1894,7 +1894,7 @@ delegate_submit_job() {
 # inline .venv-walk + publish_event.py blocks that were duplicated across
 # create/delete/resume (review item 7). Non-fatal by contract: an empty job id,
 # a missing script, or a broker failure never aborts the caller.
-delegate_publish_event() {
+_delegate_publish_event_legacy() {
  local job_id="$1" event="$2" detail="${3:-}"
  [ -n "$job_id" ] || return 0
  local py_bin pub
@@ -1909,7 +1909,7 @@ delegate_publish_event() {
 # The watchdog re-spawns the subscriber every 2 minutes (or whatever hard
 # limit we set) and exits automatically when the JOB reaches terminal state.
 # Returns the watchdog PID via stdout.
-start_watchdog() {
+_start_watchdog_legacy() {
  local job_id="$1"
  local workdir="${2:-$PWD}"
  local monitor_script="$workdir/.agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh"
@@ -1937,6 +1937,59 @@ start_watchdog() {
  echo "$pid"
 }

+# ---------------------------------------------------------------------------
+# Python-backed delegation helpers (issue #5 phase 1).
+#
+# The bodies above are kept verbatim as _*_legacy and are still reachable with
+# MAM_DELEGATION_PY=0, which is the instant rollback switch: no redeploy, no
+# revert. They are removed in phase 4, once the python path has soaked.
+#
+# PYTHONPATH is passed explicitly (same pattern as env_python) so `-m` resolves
+# lib_py from the tree this lib.sh belongs to -- that is what makes a frozen
+# loop run frozen delegation code.
+# ---------------------------------------------------------------------------
+
+# delegate_submit_job <prompt> <agent> <agent_session>
+# Register a job. Prints the new JID on stdout and nothing else.
+delegate_submit_job() {
+  if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _delegate_submit_job_legacy "$@"; return $?; fi
+  local prompt="$1" agent="$2" session="$3"
+  local registry_py
+  registry_py="$(_delegate_script registry.py)"
+  if [ -z "$registry_py" ] || [ ! -f "$registry_py" ]; then
+    echo "ERROR: multi-agent-mux-delegate-job registry.py not found under .agents/skills/" >&2
+    return 1
+  fi
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+    submit --prompt "$prompt" --agent "$agent" --agent-session "$session"
+}
+
+# delegate_publish_event <job_id> <event> [detail]
+# Non-fatal by contract: an empty job id, a missing script tree, or a broker
+# failure must never abort the caller.
+delegate_publish_event() {
+  local job_id="$1" event="$2" detail="${3:-}"
+  [ -n "$job_id" ] || return 0
+  if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _delegate_publish_event_legacy "$@"; return $?; fi
+  local pub
+  pub="$(_delegate_script publish_event.py)"
+  [ -n "$pub" ] && [ -f "$pub" ] || return 0
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+    publish --job "$job_id" --event "$event" --detail "$detail" || true
+}
+
+# ensure_subscriber_daemon <job_id> [workdir]
+# Start the monitor subscriber for this workspace unless one is already running.
+# Prints the PID(s) on stdout, exactly as start_watchdog always did.
+ensure_subscriber_daemon() {
+  if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _start_watchdog_legacy "$@"; return $?; fi
+  PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \
+    ensure-daemon --job "$1" --workdir "${2:-$PWD}"
+}
+
+# start_watchdog — retained name; create_session.sh and operator docs use it.
+start_watchdog() { ensure_subscriber_daemon "$@"; }
+
 # wait_for_tui_ready <session_name> <agent>
 # Waits up to 30 seconds for the agent's TUI to render its welcome screen.
 # Returns 0 on readiness, 1 on timeout, 2 on unobservable/headless.
@@ -16,3 +16,7 @@ Package rules, enforced by tests:
    sibling import would silently mix trees.
 """

+from lib_py.delegation.daemon import ensure_subscriber_daemon  # noqa: E402,F401
+from lib_py.delegation.service import DelegationService        # noqa: E402,F401
+
+__all__ = ["DelegationService", "ensure_subscriber_daemon"]
@@ -0,0 +1,7 @@
+"""Allow `python -m lib_py.delegation` as a synonym for the cli router."""
+import sys
+
+from lib_py.delegation.cli import main
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,90 @@
+"""`python -m lib_py.delegation.cli` — the router the shell helpers call.
+
+Deliberately thin: every subcommand delegates to the existing `main()` of the
+module that already owned that behaviour, so exit codes, stdout shape and side
+effects cannot drift between the shell path and the script path. The one piece
+of real logic here is `ensure-daemon`, which reproduces the shell helper's
+stderr wording and exit status for a missing monitor script.
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from typing import List, Optional
+
+from lib_py.delegation import daemon, mqtt_common, publish_event, registry
+
+logger = logging.getLogger("delegate_job.cli")
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="lib_py.delegation.cli",
+        description="delegation helpers invoked by lib.sh",
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    p_submit = sub.add_parser("submit", help="register a job; prints the job id")
+    p_submit.add_argument("--prompt", required=True)
+    p_submit.add_argument("--agent", default="claude-code")
+    p_submit.add_argument("--agent-session", default="herdr:claude")
+    p_submit.add_argument("--registry-dir", default=None)
+
+    p_pub = sub.add_parser("publish", help="publish a job lifecycle event")
+    p_pub.add_argument("--job", required=True)
+    p_pub.add_argument("--event", default="progress", choices=publish_event.VALID_EVENTS)
+    p_pub.add_argument("--detail", default="")
+    p_pub.add_argument("--data", default=None)
+    p_pub.add_argument("--retained", action="store_true")
+    p_pub.add_argument("--registry-dir", default=None)
+    p_pub.add_argument("--attempts", type=int, default=3)
+    p_pub.add_argument("-v", "--verbose", action="store_true")
+
+    p_daemon = sub.add_parser("ensure-daemon", help="start the subscriber daemon if absent")
+    p_daemon.add_argument("--job", required=True)
+    p_daemon.add_argument("--workdir", default=None)
+
+    return parser
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    mqtt_common.ensure_env_loaded()
+    args = _build_parser().parse_args(argv)
+
+    if args.command == "submit":
+        forward = ["register", "--prompt", args.prompt,
+                   "--agent", args.agent, "--agent-session", args.agent_session]
+        if args.registry_dir:
+            forward = ["--registry-dir", args.registry_dir] + forward
+        return registry.main(forward)
+
+    if args.command == "publish":
+        forward = ["--job", args.job, "--event", args.event,
+                   "--detail", args.detail, "--attempts", str(args.attempts)]
+        if args.data is not None:
+            forward += ["--data", args.data]
+        if args.retained:
+            forward += ["--retained"]
+        if args.registry_dir:
+            forward += ["--registry-dir", args.registry_dir]
+        if args.verbose:
+            forward += ["--verbose"]
+        return publish_event.main(forward)
+
+    if args.command == "ensure-daemon":
+        try:
+            pids = daemon.ensure_subscriber_daemon(args.job, args.workdir)
+        except FileNotFoundError as exc:
+            # Byte-identical to the legacy shell helper's failure path.
+            print("ERROR: monitor script not found: {}".format(exc.args[0]), file=sys.stderr)
+            return 1
+        for pid in pids:
+            print(pid)
+        return 0
+
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,86 @@
+"""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")
+
+
+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:
+    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.
+    """
+    workdir = workdir or os.getcwd()
+    monitor_script = monitor_script_path(workdir)
+    if not os.path.isfile(monitor_script):
+        raise FileNotFoundError(monitor_script)
+    running = find_running(monitor_script)
+    if running:
+        return running
+    return [str(spawn(monitor_script, workdir))]
@@ -0,0 +1,67 @@
+"""DelegationService — one façade over registering, publishing and supervising.
+
+Phase 1 keeps every behaviour in the existing entry points; this only groups
+them so a long-running caller (the phase 2a engine) has a single object to hold,
+optionally with a persistent MQTT connection. CLI and shell callers always use
+`persistent=False`, because a process that exits after one publish gains nothing
+from connection reuse.
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+
+from lib_py.delegation import daemon, mqtt_common, publish_event, registry
+
+logger = logging.getLogger("delegate_job.service")
+
+
+class DelegationService(object):
+    def __init__(self, registry_dir=None, persistent=False):
+        self.registry_dir = registry_dir or mqtt_common.DEFAULT_REGISTRY_DIR
+        self.persistent = persistent
+        self._publisher = None
+
+    # -- lifecycle ---------------------------------------------------------
+    def __enter__(self):
+        if self.persistent:
+            from lib_py.delegation.publisher import PersistentPublisher
+            self._publisher = PersistentPublisher()
+            self._publisher.open()
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+        return False
+
+    def close(self):
+        if self._publisher is not None:
+            self._publisher.close()
+            self._publisher = None
+
+    # -- operations --------------------------------------------------------
+    def submit_job(self, prompt, agent="claude-code", agent_session="herdr:claude", **kw):
+        return registry.register_job(
+            prompt=prompt, agent=agent, agent_session=agent_session,
+            registry_dir=self.registry_dir, **kw
+        )
+
+    def publish_event(self, job_id, event, detail="", data=None, attempts=3):
+        """Publish one event. Returns the same exit status as the CLI (0 / 2)."""
+        prepared = publish_event.prepare_event(
+            job_id=job_id, event=event, detail=detail,
+            data=data, registry_dir=self.registry_dir,
+        )
+        publisher = self._publisher
+        if publisher is None:
+            from lib_py.delegation.publisher import OneShotPublisher
+            publisher = OneShotPublisher()
+        ok, error = publisher.publish_with_retry(prepared, attempts=attempts)
+        publish_event.record_outcome(prepared, self.registry_dir, ok, error)
+        return 0 if ok else 2
+
+    def ensure_subscriber_daemon(self, job_id, workdir=None):
+        return daemon.ensure_subscriber_daemon(job_id, workdir)
+
+    def load_job(self, job_id) -> Dict[str, Any]:
+        return mqtt_common.load_job(job_id, self.registry_dir)
@@ -607,3 +607,116 @@ def test_k6_subprocess_executes_the_copy_it_was_given(shim_trees):
    assert probe.returncode == 0, probe.stderr
    assert str(shim_trees / "sandbox") in probe.stdout, probe.stdout

+
+# ==============================================================================
+# 5. P1-02 — daemon module, cli router, shell-to-python equivalence
+# ==============================================================================
+
+def _delegation_env(mam_sandbox):
+    """Env for running the package out of the sandbox copy."""
+    env = dict(os.environ)
+    env["PYTHONPATH"] = str(_sandbox_skills(mam_sandbox))
+    env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent")
+    return env
+
+
+def test_daemon_missing_monitor_raises_with_the_legacy_path(mam_sandbox):
+    from lib_py.delegation import daemon
+
+    workdir = Path(mam_sandbox) / "no_monitor"
+    workdir.mkdir()
+    with pytest.raises(FileNotFoundError) as exc:
+        daemon.ensure_subscriber_daemon("job1", str(workdir))
+    assert exc.value.args[0] == str(daemon.monitor_script_path(str(workdir)))
+
+
+def test_daemon_spawn_dedup_and_survives_parent_exit(mam_sandbox):
+    """The daemon must outlive the shell that started it and never double-spawn."""
+    from lib_py.delegation import daemon
+
+    monitor = _install_fake_monitor(mam_sandbox)
+    workdir = str(mam_sandbox)
+    spawned = []
+    try:
+        first = daemon.ensure_subscriber_daemon("job1", workdir)
+        spawned += first
+        assert len(first) == 1 and _alive(first[0])
+        assert (Path(workdir) / ".mam" / "multi-agent-mux-monitor.log").exists()
+        second = daemon.ensure_subscriber_daemon("job1", workdir)
+        assert second == first, "a running daemon must be reused, not replaced"
+        # detached into its own session, so a parent's process-group signal misses it
+        pgid = os.getpgid(int(first[0]))
+        assert pgid != os.getpgid(os.getpid())
+        assert daemon.find_running(str(monitor)) == first
+    finally:
+        _kill(spawned)
+
+
+def test_cli_ensure_daemon_reproduces_the_legacy_failure_line(mam_sandbox):
+    workdir = Path(mam_sandbox) / "no_monitor_cli"
+    workdir.mkdir()
+    res = subprocess.run(
+        [sys.executable, "-m", "lib_py.delegation.cli",
+         "ensure-daemon", "--job", "j1", "--workdir", str(workdir)],
+        capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
+    )
+    assert res.returncode == 1
+    expected = str(workdir / ".agents" / "skills" / "multi-agent-mux-monitor"
+                   / "scripts" / "reconcile.sh")
+    assert res.stderr.strip() == "ERROR: monitor script not found: {}".format(expected)
+    assert res.stdout == ""
+
+
+def test_cli_submit_prints_only_the_job_id(mam_sandbox):
+    reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
+    reg_dir.mkdir(parents=True, exist_ok=True)
+    res = subprocess.run(
+        [sys.executable, "-m", "lib_py.delegation.cli", "submit",
+         "--prompt", "hello", "--agent", "claude-code",
+         "--agent-session", "herdr:claude", "--registry-dir", str(reg_dir)],
+        capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
+    )
+    assert res.returncode == 0, res.stderr
+    job_id = res.stdout.strip()
+    assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout)
+    assert res.stdout == job_id + "\n", "stdout must carry the id and nothing else"
+    assert (reg_dir / (job_id + ".json")).exists()
+
+
+@pytest.mark.parametrize("entry", ["script", "cli"])
+def test_publish_failure_semantics_match_across_entry_points(mam_sandbox, entry):
+    """G-1/G-2/G-4 semantics must be identical through the shim and through the cli.
+
+    Broker points at a non-routable port, so nothing touches the network."""
+    reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
+    reg_dir.mkdir(parents=True, exist_ok=True)
+    env = _delegation_env(mam_sandbox)
+    job_id = subprocess.run(
+        [sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "p",
+         "--registry-dir", str(reg_dir)],
+        capture_output=True, text=True, env=env, timeout=60,
+    ).stdout.strip()
+
+    record = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
+    record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False}
+    (reg_dir / (job_id + ".json")).write_text(json.dumps(record), encoding="utf-8")
+
+    if entry == "cli":
+        cmd = [sys.executable, "-m", "lib_py.delegation.cli", "publish"]
+    else:
+        cmd = [sys.executable, str(SHIM_DIR / "publish_event.py")]
+    res = subprocess.run(
+        cmd + ["--job", job_id, "--event", "completed", "--detail", "done",
+               "--attempts", "1", "--registry-dir", str(reg_dir)],
+        capture_output=True, text=True, env=env, cwd=str(mam_sandbox), timeout=120,
+    )
+    assert res.returncode == 2, (res.stdout, res.stderr)
+    assert res.stdout == "", "publish must stay silent on stdout"
+    after = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
+    assert after["status"] == "completed", "status syncs even when the publish fails"
+    assert after.get("last_seq") == 1, "seq is consumed on failure by design"
+    audit = (Path(mam_sandbox) / ".mam" / "delegate_job_logs" / job_id / "events.ndjson")
+    events = [json.loads(l) for l in audit.read_text(encoding="utf-8").splitlines() if l.strip()]
+    published = [e for e in events if e.get("event") == "published"]
+    assert published and published[-1]["published"] is False
+    assert published[-1]["publish_error"]