Files
multi-agent-mux/.agents/skills/lib_py/delegation/cli.py
T
GodopuandClaude Fable 5.1 ce315d8c5e fix(issue-19): bound the subscriber daemon to its workspace, forward SIGTERM, keep sandboxes from leaking it
create_session.sh detaches `reconcile.sh --subscribe --idle-timeout 0` per
workspace; test sandboxes never reaped it and the daemon neither noticed its
workspace vanish nor honoured SIGTERM (bash deferred the signal behind the
foreground python child). 202 orphans were found holding public-broker
wildcard subscriptions for deleted pytest tmpdirs.

- daemon.py / lib.sh: MAM_NO_SUBSCRIBER_DAEMON=1 (literal) skips the spawn but
  still reports a running daemon; cli prints a stderr note, stdout stays empty.
- reconcile.sh: python subscriber runs as a waited background child with a
  TERM/INT trap (exit 143); the MQTT loop exits when the workspace dir is gone
  or .mam/monitor.lock is no longer the held inode; SIGTERM/SIGINT handled via
  the stop event so loop_stop/disconnect run; the polling fallback checks the
  same and sleeps in a waited child.
- conftest: sandbox default MAM_NO_SUBSCRIBER_DAEMON=1, per-sandbox reaper
  finalizer, session-scope leak guard over the pytest basetemp.
- tests: 11 lifecycle tests (tests/test_issue19_subscriber_daemon_lifecycle.py);
  phase-1 watchdog/daemon characterization tests opt back in.
- docs: monitor SKILL.md lifecycle contract + pitfalls, create SKILL.md note.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
2026-09-13 15:34:56 +09:00

100 lines
3.9 KiB
Python

"""`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("--expect-dispatch-token", default=None)
p_pub.add_argument("--attempt-id", type=int, default=None)
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.expect_dispatch_token:
forward += ["--expect-dispatch-token", args.expect_dispatch_token]
if args.attempt_id is not None:
forward += ["--attempt-id", str(args.attempt_id)]
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
if not pids and daemon.daemon_disabled():
print("subscriber daemon not started ({}=1)".format(daemon.NO_DAEMON_ENV),
file=sys.stderr)
for pid in pids:
print(pid)
return 0
return 1
if __name__ == "__main__":
sys.exit(main())