diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index fb8f828..dc3c7f6 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2280,6 +2280,39 @@ except Exception: ' <<< "$raw" 2>/dev/null || true } +# _sks_job_started — Issue #18 delivery signal: the job's own +# `started` (registry status left `pending`, or a started line in the job's +# events.log). Reads the registry the wrapper exports (SKS_REGISTRY_DIR), else +# MAM_JOBS_DIR, else $WORKSPACE_ROOT/.mam/jobs. "adhoc" (no job) is never started. +_sks_job_started() { + local job_id="${1:-adhoc}" rd + [ -n "$job_id" ] && [ "$job_id" != "adhoc" ] || return 1 + rd="${SKS_REGISTRY_DIR:-${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}}" + "$(_delegate_py_bin)" - "$rd" "$job_id" <<'PYEOF' +import json, os, sys +rd, jid = sys.argv[1], sys.argv[2] +rec = os.path.join(rd, jid + ".json") +try: + st = json.load(open(rec)).get("status") + if st and st != "pending": + sys.exit(0) +except Exception: + pass +log = os.path.join(rd, jid + ".events.log") +try: + with open(log) as fh: + for line in fh: + try: + if json.loads(line).get("event") == "started": + sys.exit(0) + except Exception: + continue +except OSError: + pass +sys.exit(1) +PYEOF +} + # _wait_session_gone [max_sec=5] # Reactive loop to wait until a herdr session goes away (happy path returns early). _wait_session_gone() { @@ -2494,9 +2527,6 @@ send_keys_safe() { # Native herdr 0.8+ fast path: agent prompt handles atomic submission if [ "${SKS_PROMPT_WAIT:-1}" = "1" ]; then local p_out="" p_rc=0 - # ISSUE-14: pre-submit snapshot for the delivery-confirmation poll below. - local rpc_pre_pane="" - [ "$_q_rc" != "2" ] && rpc_pre_pane=$(_pane_capture "$sess") local _p_tmp _p_tmp=$(mktemp 2>/dev/null || echo "/tmp/sks_p_${sess}_$$") if _sks_herdr agent prompt "$agent_target" "$text" --wait --until working --timeout 5000 >"$_p_tmp" 2>&1; then @@ -2534,29 +2564,22 @@ send_keys_safe() { if [ "$(_sks_agent_status "$sess" "$agent_target")" = "working" ]; then return 0 fi - # ISSUE-14: `timeout` / `agent_prompt_stalled` mean herdr ACCEPTED the - # submission and sent the input - only its post-submit status watch came - # up empty. An agent whose agent_status never leaves `idle` mid-turn - # (codex under herdr 0.9.0) therefore failed every delivered prompt here: - # the job was closed as agent_blocked and the agent's own `started` / - # `completed`, published seconds later, were refused as late. For an - # accepted submission keep observing for SKS_DELIVERY_GRACE_S (default - # 15) and take any positive signal: agent_status working, a busy token - # the pane did not show before the prompt, or the pane moving on from - # its pre-submit snapshot. No signal at all still fails closed (rc 4). + # ISSUE-14 / ISSUE-18: `timeout` / `agent_prompt_stalled` mean herdr + # ACCEPTED the submission and sent the input - only its post-submit + # status watch came up empty (codex under herdr 0.9.0 never reports + # `working`). For an accepted submission keep observing for + # SKS_DELIVERY_GRACE_S (default 15) and take a *protocol* signal only: + # agent_status working, or the job's own `started` in the registry + # (_sks_job_started). Pane changes and busy tokens are no longer + # delivery evidence (Issue #18: they are CLI-version-specific screen + # scraping that a headless or remote agent never produces). No signal + # still fails closed (rc 4); the wrapper then waits for `started` up to + # the job's ack timeout T before finalising the inject as failed. if printf '%s\n' "$p_out" | grep -Eq '"code": *"(timeout|agent_prompt_stalled)"'; then local g_deadline=$(( $(date +%s) + ${SKS_DELIVERY_GRACE_S:-15} )) - local pre_busy_rpc=0 cur_pane - printf '%s\n' "$rpc_pre_pane" | grep -Fq "esc to interrupt" && pre_busy_rpc=1 while [ "$(date +%s)" -lt "$g_deadline" ]; do [ "$(_sks_agent_status "$sess" "$agent_target")" = "working" ] && return 0 - if [ "$_q_rc" != "2" ]; then - cur_pane=$(_pane_capture "$sess") - if [ "$pre_busy_rpc" -eq 0 ] && printf '%s\n' "$cur_pane" | grep -Fq "esc to interrupt"; then - return 0 - fi - [ -n "$cur_pane" ] && [ "$cur_pane" != "$rpc_pre_pane" ] && return 0 - fi + _sks_job_started "$job_id" && return 0 sleep 1 done fi diff --git a/.agents/skills/lib_py/delegation/cli.py b/.agents/skills/lib_py/delegation/cli.py index 83a6d6f..0a9a634 100644 --- a/.agents/skills/lib_py/delegation/cli.py +++ b/.agents/skills/lib_py/delegation/cli.py @@ -35,6 +35,8 @@ def _build_parser() -> argparse.ArgumentParser: 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("--heartbeat", type=int, default=None, metavar="N", + help="start the automatic heartbeat timer (Issue #18)") p_pub.add_argument("--data", default=None) p_pub.add_argument("--retained", action="store_true") p_pub.add_argument("--registry-dir", default=None) @@ -74,6 +76,8 @@ def main(argv: Optional[List[str]] = None) -> int: forward += ["--expect-dispatch-token", args.expect_dispatch_token] if args.attempt_id is not None: forward += ["--attempt-id", str(args.attempt_id)] + if getattr(args, "heartbeat", None) is not None: + forward += ["--heartbeat", str(args.heartbeat)] if args.verbose: forward += ["--verbose"] return publish_event.main(forward) diff --git a/.agents/skills/lib_py/delegation/heartbeat.py b/.agents/skills/lib_py/delegation/heartbeat.py new file mode 100644 index 0000000..5586dfc --- /dev/null +++ b/.agents/skills/lib_py/delegation/heartbeat.py @@ -0,0 +1,239 @@ +"""heartbeat.py — the agent-side automatic heartbeat timer (Issue #18). + +The liveness contract obliges an agent to publish `progress` at least every N +seconds while it works. Relying on the brief to say so meant an agent that +forgot, or one whose turn was one long tool call, was closed as unresponsive +while it was fine. This timer publishes the heartbeat for it. + +`publish_event.py --event started --heartbeat N` starts one detached timer per +job; it runs in its own session so the agent's shell may return at once. The +timer stops on its own when + + * the job record reaches a terminal status (completed / error / cancelled), + * the record's dispatch_token or attempt_id changes (the job was + re-dispatched: this attempt's heartbeats would be refused anyway), + * the job record disappears, or + * its lifetime exceeds the job's wall-clock timeout plus one period. + +A terminal publish also stops it directly through the pidfile +(`/.heartbeat.pid`) so no heartbeat can trail a `completed`. + +Each beat is an ordinary `publish_event` call in this process: it takes the +next seq, is signed with the job's auth_token, carries the dispatch token and +attempt id the timer was started with, and is recorded in events.log before +the wire publish - exactly what the orchestrator's disk and wire liveness +checks read. +""" +from __future__ import annotations + +import argparse +import logging +import os +import signal +import subprocess +import sys +import threading +import time +from typing import List, Optional + +from lib_py.delegation import mqtt_common +from lib_py.delegation.mqtt_common import default_registry_dir, load_job + +logger = logging.getLogger("delegate_job.heartbeat") + +TERMINAL_STATUSES = ("completed", "error", "cancelled") +HEARTBEAT_DETAIL = "heartbeat" +MAX_IN_FLIGHT = 8 + + +def pidfile_path(job_id: str, registry_dir: str) -> str: + return os.path.join(registry_dir, "{}.heartbeat.pid".format(job_id)) + + +def _pid_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +def running_timer(job_id: str, registry_dir: str) -> Optional[int]: + """PID of a live timer for this job, or None.""" + try: + with open(pidfile_path(job_id, registry_dir)) as fh: + pid = int(fh.read().strip() or 0) + except (OSError, ValueError): + return None + if pid > 0 and _pid_alive(pid): + return pid + return None + + +def stop_timer(job_id: str, registry_dir: str) -> bool: + """Terminate the job's timer if one runs. Returns True when one was signalled.""" + pid = running_timer(job_id, registry_dir) + path = pidfile_path(job_id, registry_dir) + if pid is None: + try: + os.unlink(path) + except OSError: + pass + return False + try: + os.kill(pid, signal.SIGTERM) + except OSError: + pass + try: + os.unlink(path) + except OSError: + pass + return True + + +def start_timer(job_id: str, registry_dir: str, interval: int, + dispatch_token: Optional[str] = None, attempt_id: Optional[int] = None, + python: Optional[str] = None) -> int: + """Spawn the detached timer process and return its pid. + + A timer that already runs for the job is stopped first: a second `started` + for the same job (a retry of the same command) must not double the rate. + """ + stop_timer(job_id, registry_dir) + cmd: List[str] = [python or sys.executable, "-m", "lib_py.delegation.heartbeat", + "--job", job_id, "--registry-dir", registry_dir, + "--interval", str(int(interval))] + if dispatch_token: + cmd += ["--expect-dispatch-token", dispatch_token] + if attempt_id is not None: + cmd += ["--attempt-id", str(int(attempt_id))] + env = dict(os.environ) + # The timer imports lib_py from the same tree as this process. + pkg_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + env["PYTHONPATH"] = pkg_root + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "") + log_path = os.path.join(registry_dir, "{}.heartbeat.log".format(job_id)) + with open(log_path, "ab") as log: + proc = subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, + start_new_session=True, env=env, cwd=os.getcwd()) + with open(pidfile_path(job_id, registry_dir), "w") as fh: + fh.write(str(proc.pid)) + logger.info("heartbeat timer pid %d for job %s every %ss", proc.pid, job_id, interval) + return proc.pid + + +def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id) -> Optional[str]: + try: + rec = load_job(job_id, registry_dir) + except FileNotFoundError: + return "job record gone" + if rec.get("status") in TERMINAL_STATUSES: + return "job is {}".format(rec.get("status")) + if dispatch_token and rec.get("dispatch_token") != dispatch_token: + return "dispatch token changed (re-dispatched)" + if attempt_id is not None and rec.get("attempt_id", rec.get("iteration", 1)) != attempt_id: + return "attempt changed (re-dispatched)" + return None + + +def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None, + attempt_id=None, max_lifetime: Optional[float] = None) -> int: + """The timer loop. Runs until a stop condition holds.""" + from lib_py.delegation import publish_event # late import: the CLI imports us + + interval = max(1, int(interval)) + if max_lifetime is None: + try: + max_lifetime = float(load_job(job_id, registry_dir).get("timeout_sec", 3600)) + interval + except FileNotFoundError: + return 1 + stop = {"flag": False} + + def _on_term(_signum, _frame): + stop["flag"] = True + + signal.signal(signal.SIGTERM, _on_term) + signal.signal(signal.SIGINT, _on_term) + + started = time.monotonic() + own_pidfile = pidfile_path(job_id, registry_dir) + # Fixed cadence: the next beat is due `interval` after the previous one + # was *due*, not after its publish returned, so a slow broker (seconds per + # ACK) does not stretch the period the contract promises. Each beat is + # recorded on disk before its wire publish, so the on-disk cadence holds + # even when the wire lags. Sleep first: `started` was just published. + next_due = time.monotonic() + interval + in_flight: List[threading.Thread] = [] + while not stop["flag"]: + while time.monotonic() < next_due and not stop["flag"]: + time.sleep(min(0.5, max(0.0, next_due - time.monotonic()))) + if stop["flag"]: + break + next_due += interval + if next_due < time.monotonic(): + next_due = time.monotonic() + interval # fell far behind: resync + reason = _should_stop(job_id, registry_dir, dispatch_token, attempt_id) + if reason: + logger.info("heartbeat timer for %s stopping: %s", job_id, reason) + break + if time.monotonic() - started > max_lifetime: + logger.info("heartbeat timer for %s stopping: lifetime %.0fs exceeded", job_id, max_lifetime) + break + argv = ["--registry-dir", registry_dir, "--job", job_id, "--event", "progress", + "--detail", HEARTBEAT_DETAIL, "--attempts", "2"] + if dispatch_token: + argv += ["--expect-dispatch-token", dispatch_token] + if attempt_id is not None: + argv += ["--attempt-id", str(attempt_id)] + # Each beat publishes in its own thread: the local record (seq, HMAC, + # events.log) is taken at once, and the wire publish - seconds per ACK + # on a slow broker - must not delay the next beat. In-flight beats are + # capped so a dead broker cannot pile up threads without bound. + in_flight[:] = [t for t in in_flight if t.is_alive()] + if len(in_flight) >= MAX_IN_FLIGHT: + logger.warning("heartbeat for %s skipped: %d publishes still in flight", job_id, len(in_flight)) + continue + + def _beat(argv=argv): + try: + rc = publish_event.main(argv) + except Exception as exc: # pragma: no cover - defensive + logger.warning("heartbeat publish raised: %s", exc) + rc = 2 + if rc == 4: + logger.info("heartbeat publish refused (stale attempt); timer for %s will stop", job_id) + stop["flag"] = True + + t = threading.Thread(target=_beat, daemon=True) + t.start() + in_flight.append(t) + for t in in_flight: + t.join(timeout=15) + try: + with open(own_pidfile) as fh: + if fh.read().strip() == str(os.getpid()): + os.unlink(own_pidfile) + except (OSError, ValueError): + pass + return 0 + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description="automatic heartbeat timer for a delegated job") + parser.add_argument("--job", required=True) + parser.add_argument("--registry-dir", default=default_registry_dir()) + parser.add_argument("--interval", type=int, required=True, help="heartbeat period N in seconds") + parser.add_argument("--expect-dispatch-token", default=None) + parser.add_argument("--attempt-id", type=int, default=None) + parser.add_argument("--max-lifetime", type=float, default=None) + parser.add_argument("-v", "--verbose", action="store_true") + args = parser.parse_args(argv) + mqtt_common.ensure_env_loaded() + mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.INFO) + return run(args.job, args.registry_dir, args.interval, args.expect_dispatch_token, + args.attempt_id, args.max_lifetime) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index d350c07..9f6966d 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -44,6 +44,8 @@ from lib_py.delegation.mqtt_common import ( logger = logging.getLogger("delegate_job.job_subscriber") TERMINAL_EVENTS = ("completed", "error") +TERMINAL_STATUSES = ("completed", "error", "cancelled") +STATE_EVENT = "state" def _format_line(topic: str, payload: Dict[str, Any]) -> str: @@ -133,6 +135,46 @@ class _Watcher: # by the transition boundary as an idempotent ack, so a replay after # restart cannot double-apply either. self.seen_intents = {} + # Issue #18: jobs for which at least one wire event was accepted. A seq + # gap is only meaningful after that - the initial last_seq comes from + # the registry and may be behind the wire (or ahead of a fresh remote). + self.accepted_any: Set[str] = set() + + def _note_seq_gap(self, jid: str, expected: int, got: int, via: str, topic: str) -> None: + """Issue #18: a seq the subscriber never saw is a detected loss.""" + logger.warning("seq gap for job %s: expected %d, got %d (via %s)", jid, expected, got, via) + mqtt_common.append_event(jid, { + "event": "seq_gap", + "expected_seq": expected, + "seq": got, + "via": via, + "topic": topic, + }) + + def _on_state(self, msg, payload: Dict[str, Any], jid: str) -> None: + """A retained `/state` message (Issue #18): learn the job's + status and last seq at (re)connect without waiting for the next + heartbeat. Signed and token-checked like an event. Delivered to the + main loop as a state item so the idle window resets and a terminal + status ends the wait.""" + seq = payload.get("seq") + if not isinstance(seq, int): + logger.warning("drop state for job %s: missing or invalid seq", jid) + return + last = self.last_seq.get(jid, 0) + if jid in self.accepted_any and seq > last + 1: + self._note_seq_gap(jid, last + 1, seq, "state", msg.topic) + if seq > last: + self.last_seq[jid] = seq + mqtt_common.append_event(jid, { + "event": "received", + "source_event": STATE_EVENT, + "seq": seq, + "topic": msg.topic, + "timestamp": payload.get("timestamp"), + "detail": "status={} last_event={}".format(payload.get("status"), payload.get("last_event")), + }) + self.events.put((msg.topic, payload)) def on_message(self, _client, _userdata, msg) -> None: # --- defensive parsing ------------------------------------------- @@ -175,6 +217,9 @@ class _Watcher: logger.warning("drop event for job %s: attempt_id %r is older than expected %r", jid, ev_att, expected_att) return + if payload.get("event") == STATE_EVENT: + self._on_state(msg, payload, jid) + return # --- replay attack defense: check monotonic sequence --- seq = payload.get("seq") if seq is None or not isinstance(seq, int): @@ -210,7 +255,10 @@ class _Watcher: logger.warning("drop event for job %s: seq %d is not monotonically increasing (last %d)", jid, seq, self.last_seq.get(jid, 0)) return + if jid in self.accepted_any and seq > self.last_seq.get(jid, 0) + 1: + self._note_seq_gap(jid, self.last_seq.get(jid, 0) + 1, seq, "events", msg.topic) self.last_seq[jid] = seq + self.accepted_any.add(jid) # Recorded only once the event is actually accepted. A late packet that the # sequence guard refuses must not mark its intent as seen, or the genuine # retry that follows would be mistaken for the replay of an event this @@ -232,6 +280,22 @@ class _Watcher: self.events.put((msg.topic, payload)) +def _sync_remote_status(jid: str, status: str, registry_dir: str, + dispatch_token: Optional[str]) -> None: + """Issue #18: mirror a wire event into the orchestrator's registry for a + job whose agent shares no disk with it. Goes through the same transition + boundary as the publisher's own status sync; a refused (late/duplicate) + transition is logged, never raised.""" + try: + res = mqtt_common.transition_job_status( + jid, registry_dir, _source="job_subscriber", status=status, + expected_dispatch_token=dispatch_token) + if not res.committed and not res.idempotent: + logger.warning("remote status sync refused for %s -> %s: %s", jid, status, res.reason) + except Exception as exc: # pragma: no cover - best effort + logger.warning("remote status sync failed for %s: %s", jid, exc) + + def _collect_jobs(args) -> List[Dict[str, Any]]: """Resolve the list of job records this invocation should watch.""" if args.wait_any: @@ -276,6 +340,13 @@ def _run_subscriber(args) -> int: base_job = jobs[0] wall_timeout = args.timeout if args.timeout is not None else float(base_job.get("timeout_sec", 3600)) idle_timeout = args.idle_timeout if args.idle_timeout is not None else float(base_job.get("idle_timeout_sec", 120)) + # Issue #18: until the first accepted event the "last event" is the + # injection itself, which happens after this subscriber started and may + # take up to the ack timeout T to produce `started` (a remote runner has + # to receive the dispatch first). The idle budget before the first event + # is therefore T + K x N; from the first event on it is K x N. + ack_timeout = args.ack_timeout if args.ack_timeout is not None else float(base_job.get("ack_timeout_sec", 0) or 0) + first_event_seen = False # All watched jobs share a broker in practice; connect using the first # job's broker and subscribe to each job's events topic. @@ -283,16 +354,31 @@ def _run_subscriber(args) -> int: client = make_client("subscriber", config) client.on_message = watcher.on_message + # Issue #18: a remote job (no shared disk with the agent) has nobody else + # to move its registry record - the subscriber does it from the wire. + remote_sync = bool(args.remote) or any(j.get("transport") == "remote" for j in jobs) + subscribed_topics = [] + state_topics = [] for job in jobs: prefix = job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"]) subscribed_topics.append(f"{prefix}/events") + state_topics.append(f"{prefix}/state") + + connects = {"count": 0} def on_connect(_c, _u, _flags, reason_code, _props): if mqtt_common.reason_code_value(reason_code) != 0: logger.error("broker connection failed: rc=%s", reason_code) return - for topic in subscribed_topics: + connects["count"] += 1 + if connects["count"] > 1: + # Issue #18: paho reconnected on its own; re-subscribing here is + # what makes the retained state message arrive again at once. + logger.warning("broker reconnected (connect #%d); re-subscribing", connects["count"]) + for jid in expected_ids: + mqtt_common.append_event(jid, {"event": "reconnected", "connect_count": connects["count"]}) + for topic in subscribed_topics + state_topics: _c.subscribe(topic, qos=1) logger.info("subscribed to %s", topic) @@ -361,7 +447,7 @@ def _run_subscriber(args) -> int: return 2 if not subscribed: last_event = now - idle_left = idle_timeout - (now - last_event) + idle_left = (idle_timeout if first_event_seen else idle_timeout + ack_timeout) - (now - last_event) if idle_left <= 0: _check_disk_fallback(pending, args.registry_dir, terminal, dispatch_tokens, attempts) if not pending: @@ -386,10 +472,30 @@ def _run_subscriber(args) -> int: last_event = time.monotonic() retention_checked = True - print(_format_line(topic, payload), flush=True) + first_event_seen = True jid = payload["job_id"] event = payload.get("event") + if event == STATE_EVENT: + # Issue #18: retained job state. Any state resets the idle + # window (the agent is alive enough to have published); a + # terminal status ends the wait exactly like the terminal event. + status = payload.get("status") + if status in TERMINAL_STATUSES and jid not in terminal: + synth = "completed" if status == "completed" else "error" + payload = dict(payload, event=synth, source="state", + detail="terminal status '{}' recovered from retained state".format(status)) + print(_format_line(topic, payload), flush=True) + terminal[jid] = synth + pending.discard(jid) + if remote_sync: + _sync_remote_status(jid, status, args.registry_dir, dispatch_tokens.get(jid)) + continue + + print(_format_line(topic, payload), flush=True) + if remote_sync and event in ("started", "completed", "error", "cancelled"): + _sync_remote_status(jid, {"started": "running"}.get(event, event), + args.registry_dir, dispatch_tokens.get(jid)) if event in TERMINAL_EVENTS: if jid in terminal: # Already finalised: ignore duplicates / late reorders. @@ -421,6 +527,9 @@ def main(argv=None) -> int: help="wall-clock budget in seconds (default: job.timeout_sec or 3600)") parser.add_argument("--idle-timeout", type=float, default=None, help="max seconds with no new event (default: job.idle_timeout_sec or 120)") + parser.add_argument("--ack-timeout", type=float, default=None, + help="extra idle budget before the first event, for the injection to " + "produce `started` (default: job.ack_timeout_sec; Issue #18)") parser.add_argument("--expect-retention", action="store_true", help="warn if no retained terminal event arrives promptly") parser.add_argument("--expect-dispatch-token", default=None, @@ -428,6 +537,10 @@ def main(argv=None) -> int: parser.add_argument("--attempt-id", type=int, default=None, help="minimum attempt id to accept events for") parser.add_argument("--registry-dir", default=default_registry_dir()) + parser.add_argument("--remote", action="store_true", + help="the agent shares no disk with this registry: mirror started/" + "terminal events into the job record (Issue #18; automatic " + "for jobs registered with --transport remote)") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args(argv) diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index 2a4a807..677891a 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -29,6 +29,7 @@ import logging import sys from typing import Any, Dict, Optional +from lib_py.delegation import heartbeat as heartbeat_timer from lib_py.delegation import mqtt_common from lib_py.delegation import registry from lib_py.delegation.mqtt_common import ( @@ -92,15 +93,49 @@ def build_payload( return payload -def _publish_once(config, topic: str, body: bytes, retain: bool) -> None: +def _publish_once(config, topic: str, body: bytes, retain: bool, extra=()) -> None: """Connect, publish one QoS-1 message, wait for the broker ACK, disconnect. Delegates to the transport module so the CLI path and the service path share one implementation. Keeping a second copy here is what let the persistent transport's handshake and not-connected bugs exist in only one of them. Raises on any failure so ``with_retry`` can re-run the whole sequence. + + ``extra`` (Issue #18): further ``(topic, body, retain)`` messages sent over + the same connection after the event - the retained job-state message, so + it costs no second broker handshake. """ - OneShotPublisher().publish_raw(config, topic, body, retain) + OneShotPublisher().publish_raw(config, topic, body, retain, extra=extra) + + +def state_topic_for(job: Dict[str, Any]) -> str: + prefix = job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"]) + return f"{prefix}/state" + + +def build_state_payload(job: Dict[str, Any], prepared: "PreparedEvent") -> Dict[str, Any]: + """The retained job-state message published next to every event (Issue #18). + + A subscriber that (re)connects receives it at once and learns the job's + status and the last seq without waiting for the next heartbeat; a seq it + never saw in between is a detected loss, not a silent one. Signed like an + event so a forged state cannot close a job. + """ + status = EVENT_TO_STATUS.get(prepared.event) or "running" + data = {k: v for k, v in (prepared.payload.get("data") or {}).items() + if k in ("dispatch_token", "attempt_id")} + payload = build_payload(job_id=prepared.job_id, seq=prepared.seq, event="state", + detail=prepared.event, data=data, auth_token=job.get("auth_token")) + payload["status"] = status + payload["last_event"] = prepared.event + if job.get("auth_token"): + # re-sign: `status`/`last_event` were added after build_payload signed + payload["data"].pop("hmac_sig", None) + sign_payload = {k: v for k, v in payload.items() if k != "data"} + sign_payload["data"] = dict(payload["data"]) + msg = json.dumps(sign_payload, sort_keys=True, separators=(",", ":")).encode() + payload["data"]["hmac_sig"] = hmac.new(job["auth_token"].encode(), msg, hashlib.sha256).hexdigest() + return payload class PreparedEvent(object): @@ -256,9 +291,19 @@ def main(argv=None) -> int: "job's current iteration)") parser.add_argument("--expect-dispatch-token", default=None, help="expected dispatch token for CAS verification") + parser.add_argument("--heartbeat", type=int, default=None, metavar="N", + help="after publishing, start a detached timer that publishes a " + "`progress` heartbeat every N seconds until the job ends " + "(Issue #18; not with a terminal event)") + parser.add_argument("--no-state", action="store_true", + help="do not publish the retained /state message") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args(argv) + if args.heartbeat is not None and (args.event in TERMINAL_EVENTS or args.heartbeat < 1): + logger.error("--heartbeat needs a positive period and a non-terminal event") + return 1 + mqtt_common.ensure_env_loaded() mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.WARNING) @@ -341,6 +386,12 @@ def main(argv=None) -> int: check_token, actual_tok) return 4 + # Issue #18: a terminal event ends the heartbeat obligation. Stop the timer + # before taking this event's seq so no heartbeat can follow `completed`. + if args.event in TERMINAL_EVENTS: + if heartbeat_timer.stop_timer(job_id, args.registry_dir): + logger.info("stopped the heartbeat timer for job %s", job_id) + try: prepared = prepare_event( job_id=job_id, @@ -356,6 +407,12 @@ def main(argv=None) -> int: logger.error("%s", exc) return 1 + extra_messages = [] + if not args.no_state and rec is not None: + state_payload = build_state_payload(rec, prepared) + extra_messages.append((state_topic_for(rec), + json.dumps(state_payload, ensure_ascii=False).encode("utf-8"), True)) + # Local first: the event reaches events.log before the publish can block on # the broker ACK and its retries (see record_local_event). record_local_event(prepared, args.registry_dir) @@ -370,7 +427,7 @@ def main(argv=None) -> int: publish_ok = True publish_error: Optional[str] = None try: - publish(prepared.config, prepared.topic, prepared.body, prepared.retain) + publish(prepared.config, prepared.topic, prepared.body, prepared.retain, extra_messages) except Exception as exc: publish_ok = False publish_error = str(exc) @@ -388,6 +445,22 @@ def main(argv=None) -> int: expected_attempt=sync_expected_attempt, local_event_recorded=True) + # Issue #18: the automatic heartbeat. Started whether or not the wire + # publish succeeded - the agent is working either way, and each beat lands + # in events.log before its own publish, which is what local liveness reads. + if args.heartbeat is not None: + hb_token = sync_expected_token + hb_attempt = sync_expected_attempt + if rec is not None: + hb_token = hb_token or rec.get("dispatch_token") + hb_attempt = hb_attempt if hb_attempt is not None else rec.get("attempt_id", rec.get("iteration", 1)) + try: + pid = heartbeat_timer.start_timer(job_id, args.registry_dir, args.heartbeat, + dispatch_token=hb_token, attempt_id=hb_attempt) + logger.info("heartbeat timer started (pid %d, every %ss)", pid, args.heartbeat) + except OSError as exc: + logger.error("could not start the heartbeat timer: %s", exc) + if not publish_ok: return 2 diff --git a/.agents/skills/lib_py/delegation/publisher.py b/.agents/skills/lib_py/delegation/publisher.py index 612e0d9..a921fe1 100644 --- a/.agents/skills/lib_py/delegation/publisher.py +++ b/.agents/skills/lib_py/delegation/publisher.py @@ -79,11 +79,13 @@ class OneShotPublisher(_BasePublisher): def publish(self, prepared): self.publish_raw(prepared.config, prepared.topic, prepared.body, prepared.retain) - def publish_raw(self, config, topic: str, body: bytes, retain: bool) -> None: + def publish_raw(self, config, topic: str, body: bytes, retain: bool, extra=()) -> None: """Connect, publish, wait for the ACK, disconnect. The single implementation of one-shot publishing: ``publish_event``'s CLI path calls straight into this, so the two entry points cannot drift. + ``extra`` messages (``(topic, body, retain)`` tuples, Issue #18's retained + job state) go out over the same connection, each awaited like the first. """ client = self.client_factory("publisher", config) self.connect_count += 1 @@ -104,6 +106,8 @@ class OneShotPublisher(_BasePublisher): if mqtt_common.reason_code_value(connected["rc"]) != 0: raise ConnectionError("broker refused connection: rc={}".format(connected["rc"])) _send(client, topic, body, retain) + for x_topic, x_body, x_retain in (extra or ()): + _send(client, x_topic, x_body, x_retain) finally: client.loop_stop() try: diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index 189ecb8..b774f21 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -47,6 +47,22 @@ def generate_job_id(bits: int = 32) -> str: return uuid.uuid4().hex[:nibbles] +# Issue #18: the liveness contract's defaults. An agent heartbeats (publishes +# `progress`) at least every N seconds; the orchestrator treats K missed +# heartbeats (K x N seconds since the last event) as unresponsive; `started` +# must arrive within T seconds of the prompt injection. K x N with these +# defaults equals the historical 120 s idle timeout, so a caller that never +# asked for the contract sees the same idle window as before. +DEFAULT_HEARTBEAT_SEC = 30 +DEFAULT_HEARTBEAT_MISSES = 4 +DEFAULT_ACK_TIMEOUT_SEC = 60 + + +def liveness_window(heartbeat_sec: int, heartbeat_misses: int) -> int: + """K x N: seconds without any event after which a job is unresponsive.""" + return max(1, int(heartbeat_sec)) * max(1, int(heartbeat_misses)) + + def register_job( prompt: str, agent: str = "claude-code", @@ -54,7 +70,7 @@ def register_job( role: str = "Worker", broker: Optional[Dict[str, Any]] = None, timeout_sec: int = 3600, - idle_timeout_sec: int = 120, + idle_timeout_sec: Optional[int] = None, registry_dir: str = DEFAULT_REGISTRY_DIR, job_id: Optional[str] = None, expected_artifacts: Optional[List[str]] = None, @@ -64,13 +80,31 @@ def register_job( reviewer: Optional[str] = None, reviewer_session: Optional[str] = None, max_iterations: int = 5, + heartbeat_sec: int = DEFAULT_HEARTBEAT_SEC, + heartbeat_misses: int = DEFAULT_HEARTBEAT_MISSES, + ack_timeout_sec: int = DEFAULT_ACK_TIMEOUT_SEC, + dispatch_token: Optional[str] = None, + attempt_id: int = 1, + transport: str = "herdr", ) -> str: """Create a new ``pending`` job record and return its id. ``broker`` defaults to the current environment's resolved broker block, so the registry alone is enough for ``publish_event.py`` to connect later. + + Issue #18: ``idle_timeout_sec`` left as None derives from the liveness + contract (``heartbeat_misses x heartbeat_sec``); an explicit value is kept + as given. ``dispatch_token`` / ``attempt_id`` are minted here unless a + remote runner mirrors a job the orchestrator already registered, in which + case it passes the orchestrator's so the events it publishes carry the + identity the orchestrator's subscriber expects. ``transport`` records how + the prompt reaches the agent (``herdr`` pane or ``remote`` dispatch topic). """ job_id = job_id or generate_job_id(bits) + heartbeat_sec = max(1, int(heartbeat_sec)) + heartbeat_misses = max(1, int(heartbeat_misses)) + if idle_timeout_sec is None: + idle_timeout_sec = liveness_window(heartbeat_sec, heartbeat_misses) if broker is None: broker = broker_config_from_env().to_registry_block() if auth_token is None: @@ -101,8 +135,13 @@ def register_job( "reviewer_session": reviewer_session, "max_iterations": int(max_iterations), "iteration": 1, - "dispatch_token": uuid.uuid4().hex, - "attempt_id": 1, + "dispatch_token": dispatch_token or uuid.uuid4().hex, + "attempt_id": int(attempt_id), + # Issue #18 liveness contract + "heartbeat_sec": heartbeat_sec, + "heartbeat_misses": heartbeat_misses, + "ack_timeout_sec": max(1, int(ack_timeout_sec)), + "transport": transport, } with registry_lock(registry_dir): if mqtt_common._job_path(job_id, registry_dir).exists(): @@ -276,7 +315,20 @@ def _build_parser() -> argparse.ArgumentParser: p_reg.add_argument("--agent-session", default="herdr:claude") p_reg.add_argument("--role", default="Worker", help="logical role for the delegated agent (e.g. Worker, Planner, Reviewer)") p_reg.add_argument("--timeout", type=int, default=3600) - p_reg.add_argument("--idle-timeout", type=int, default=120) + p_reg.add_argument("--idle-timeout", type=int, default=None, + help="seconds without an event before the job is unresponsive " + "(default: --heartbeat-misses x --heartbeat-sec)") + p_reg.add_argument("--heartbeat-sec", type=int, default=DEFAULT_HEARTBEAT_SEC, + help="agent heartbeat period N in seconds (Issue #18)") + p_reg.add_argument("--heartbeat-misses", type=int, default=DEFAULT_HEARTBEAT_MISSES, + help="missed heartbeats K before the job counts as unresponsive") + p_reg.add_argument("--ack-timeout", type=int, default=DEFAULT_ACK_TIMEOUT_SEC, + help="seconds T after injection within which `started` must arrive") + p_reg.add_argument("--transport", default="herdr", choices=["herdr", "remote"], + help="how the prompt reaches the agent") + p_reg.add_argument("--dispatch-token", default=None, help=argparse.SUPPRESS) + p_reg.add_argument("--attempt-id", type=int, default=1, help=argparse.SUPPRESS) + p_reg.add_argument("--job-id", default=None, help=argparse.SUPPRESS) p_reg.add_argument("--bits", type=int, default=32, help="32 (PoC) or 128 (prod)") p_reg.add_argument("--artifact", action="append", default=[], dest="artifacts") p_reg.add_argument("--auth-token", default=None, help="HMAC auth token for the job (auto-generated if secure broker is detected)") @@ -430,6 +482,13 @@ def main(argv: Optional[List[str]] = None) -> int: reviewer=args.reviewer, reviewer_session=args.reviewer_session, max_iterations=args.max_iterations, + heartbeat_sec=args.heartbeat_sec, + heartbeat_misses=args.heartbeat_misses, + ack_timeout_sec=args.ack_timeout, + transport=args.transport, + dispatch_token=args.dispatch_token, + attempt_id=args.attempt_id, + job_id=args.job_id, ) print(job_id) return 0 diff --git a/.agents/skills/lib_py/delegation/remote.py b/.agents/skills/lib_py/delegation/remote.py new file mode 100644 index 0000000..c14560f --- /dev/null +++ b/.agents/skills/lib_py/delegation/remote.py @@ -0,0 +1,381 @@ +"""remote.py — deliver a job to an agent on another host over the broker (Issue #18). + +The herdr transport injects the prompt into a pane on this machine and reads +the agent's `started`/`progress`/`completed` events back over the broker. A +remote agent shares neither the pane nor the orchestrator's disk, so the +prompt itself has to travel over the wire too: + + orchestrator --retained `/dispatch`--> remote runner + remote runner --`/events` + `/state`--> orchestrator's subscriber + +`publish_dispatch` (wrapper side) publishes one retained dispatch message +carrying the brief, the job's auth_token / dispatch_token / attempt_id and the +liveness contract (heartbeat N, misses K, ack T). `run_runner` (remote side) +subscribes for its session name, mirrors the job into its own registry, +publishes `started` with the automatic heartbeat, runs the agent command, and +publishes the terminal event. Every event it publishes is signed with the +job's auth_token and carries the orchestrator's dispatch identity, so the +orchestrator's unchanged subscriber accepts it; the subscriber's `--remote` +mode mirrors those events into the orchestrator's registry. + +Trust: a dispatch makes a runner execute a command, so it is signed with a +pre-shared secret (`MAM_DISPATCH_SECRET`, from each host's own .mam.env) and a +runner ignores anything unsigned or mis-signed. The auth_token travels inside +that signed message; use this only on a private broker (see MESSAGING.md). +""" +from __future__ import annotations + +import argparse +import hashlib +import hmac +import json +import logging +import os +import queue +import shlex +import socket +import subprocess +import sys +import time +from typing import Any, Dict, Optional + +from lib_py.delegation import mqtt_common, registry +from lib_py.delegation.mqtt_common import ( + DEFAULT_TOPIC_ROOT, + SCHEMA_VERSION, + broker_config_from_env, + broker_config_from_job, + default_registry_dir, + load_job, + make_client, +) +from lib_py.delegation.publisher import OneShotPublisher + +logger = logging.getLogger("delegate_job.remote") + +DISPATCH_KIND = "dispatch" +SECRET_ENV = "MAM_DISPATCH_SECRET" +SESSION_PREFIX = "remote:" + + +def is_remote_session(agent_session: Optional[str]) -> bool: + return bool(agent_session) and str(agent_session).startswith(SESSION_PREFIX) + + +def session_name(agent_session: str) -> str: + return agent_session[len(SESSION_PREFIX):] if is_remote_session(agent_session) else agent_session + + +def dispatch_topic_for(job: Dict[str, Any]) -> str: + prefix = job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"]) + return f"{prefix}/dispatch" + + +def _canonical(payload: Dict[str, Any]) -> bytes: + body = {k: v for k, v in payload.items() if k != "sig"} + return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +def sign_dispatch(payload: Dict[str, Any], secret: str) -> Dict[str, Any]: + payload = dict(payload) + payload["sig"] = hmac.new(secret.encode("utf-8"), _canonical(payload), hashlib.sha256).hexdigest() + return payload + + +def verify_dispatch(payload: Dict[str, Any], secret: Optional[str]) -> bool: + """True when the dispatch carries a valid signature for ``secret``. With + no secret configured on the runner side nothing is accepted.""" + if not secret: + return False + sig = payload.get("sig") + if not isinstance(sig, str): + return False + expected = hmac.new(secret.encode("utf-8"), _canonical(payload), hashlib.sha256).hexdigest() + return hmac.compare_digest(sig, expected) + + +def build_dispatch(job: Dict[str, Any], session: str, instructions: str) -> Dict[str, Any]: + return { + "schema_version": SCHEMA_VERSION, + "kind": DISPATCH_KIND, + "job_id": job["job_id"], + "session": session, + "instructions": instructions, + "brief": job.get("prompt", ""), + "role": job.get("role"), + "agent": job.get("agent"), + "auth_token": job.get("auth_token"), + "dispatch_token": job.get("dispatch_token"), + "attempt_id": job.get("attempt_id", job.get("iteration", 1)), + "timeout_sec": job.get("timeout_sec", 3600), + "idle_timeout_sec": job.get("idle_timeout_sec"), + "heartbeat_sec": job.get("heartbeat_sec", registry.DEFAULT_HEARTBEAT_SEC), + "heartbeat_misses": job.get("heartbeat_misses", registry.DEFAULT_HEARTBEAT_MISSES), + "ack_timeout_sec": job.get("ack_timeout_sec", registry.DEFAULT_ACK_TIMEOUT_SEC), + "topic_prefix": job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"]), + "origin": socket.gethostname(), + "timestamp": mqtt_common._utcnow(), + } + + +def publish_dispatch(job_id: str, registry_dir: str, session: str, instructions: str, + secret: Optional[str] = None, allow_unsigned: bool = False) -> str: + """Publish the retained dispatch for ``job_id``; returns the topic.""" + job = load_job(job_id, registry_dir) + secret = secret or os.environ.get(SECRET_ENV) + payload = build_dispatch(job, session, instructions) + if secret: + payload = sign_dispatch(payload, secret) + elif not allow_unsigned: + raise RuntimeError("{} is not set: refusing to publish an unsigned dispatch " + "(set it in .mam.env on both hosts, or pass --allow-unsigned)".format(SECRET_ENV)) + topic = dispatch_topic_for(job) + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + mqtt_common.with_retry(OneShotPublisher().publish_raw, attempts=3, + exceptions=(OSError, TimeoutError, ConnectionError, ValueError))( + broker_config_from_job(job), topic, body, True) + mqtt_common.append_event(job_id, {"event": "dispatched", "topic": topic, "session": session, + "attempt_id": payload["attempt_id"]}) + logger.info("dispatched job %s to remote session %s on %s", job_id, session, topic) + return topic + + +def clear_dispatch(config, topic: str) -> None: + """Retire a retained dispatch (empty retained payload) so a runner that + restarts does not execute it again.""" + try: + OneShotPublisher().publish_raw(config, topic, b"", True) + except Exception as exc: # pragma: no cover - best effort + logger.warning("could not clear dispatch %s: %s", topic, exc) + + +# --------------------------------------------------------------------------- +# remote side +# --------------------------------------------------------------------------- + +def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> str: + """Create the local job record a remote agent publishes against. + + The record carries the orchestrator's job_id, auth_token, dispatch_token + and attempt_id, so `publish_event` on this host produces events the + orchestrator's subscriber accepts. A previous attempt's record (a + re-dispatch) is retired first: creation is the only write allowed here. + """ + jid = d["job_id"] + path = mqtt_common._job_path(jid, registry_dir) + if path.exists(): + old = load_job(jid, registry_dir) + if (old.get("dispatch_token") == d.get("dispatch_token") + and old.get("status") not in ("completed", "error", "cancelled")): + return jid # same attempt, already mirrored (duplicate delivery) + with mqtt_common.registry_lock(registry_dir): + path.unlink() + log = mqtt_common._job_path(jid, registry_dir).with_suffix(".events.log") + if log.exists(): + log.unlink() + registry.register_job( + prompt=d.get("brief", ""), + agent=d.get("agent") or "remote", + agent_session=SESSION_PREFIX + session, + role=d.get("role") or "Worker", + timeout_sec=int(d.get("timeout_sec") or 3600), + idle_timeout_sec=d.get("idle_timeout_sec"), + registry_dir=registry_dir, + job_id=jid, + auth_token=d.get("auth_token"), + heartbeat_sec=int(d.get("heartbeat_sec") or registry.DEFAULT_HEARTBEAT_SEC), + heartbeat_misses=int(d.get("heartbeat_misses") or registry.DEFAULT_HEARTBEAT_MISSES), + ack_timeout_sec=int(d.get("ack_timeout_sec") or registry.DEFAULT_ACK_TIMEOUT_SEC), + dispatch_token=d.get("dispatch_token"), + attempt_id=int(d.get("attempt_id") or 1), + transport="remote", + ) + brief_dir = os.path.join(registry_dir, jid) + os.makedirs(brief_dir, exist_ok=True) + with open(os.path.join(brief_dir, "brief.md"), "w", encoding="utf-8") as fh: + fh.write(d.get("brief", "")) + return jid + + +def _publish(jid: str, registry_dir: str, event: str, detail: str, d: Dict[str, Any], + heartbeat: Optional[int] = None) -> int: + from lib_py.delegation import publish_event + argv = ["--registry-dir", registry_dir, "--job", jid, "--event", event, "--detail", detail] + if d.get("dispatch_token"): + argv += ["--expect-dispatch-token", d["dispatch_token"]] + if d.get("attempt_id") is not None: + argv += ["--attempt-id", str(d["attempt_id"])] + if heartbeat: + argv += ["--heartbeat", str(heartbeat)] + return publish_event.main(argv) + + +def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str, + exec_cmd: Optional[str], python: Optional[str] = None) -> int: + """Run one dispatched job to its terminal event. Returns the agent rc.""" + jid = mirror_job(d, registry_dir, session) + heartbeat = int(d.get("heartbeat_sec") or registry.DEFAULT_HEARTBEAT_SEC) + rc_started = _publish(jid, registry_dir, "started", "remote runner {} claimed the job".format(session), + d, heartbeat=heartbeat) + if rc_started not in (0, 2): + logger.error("could not publish started for %s (rc %s)", jid, rc_started) + return 1 + job = load_job(jid, registry_dir) + rc = 0 + if exec_cmd: + py = python or sys.executable + skills_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + env = dict(os.environ) + env["PYTHONPATH"] = skills_root + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "") + env.update({ + "JOB_ID": jid, + "MAM_JOB_ID": jid, + "MAM_JOB_REGISTRY_DIR": registry_dir, + "MAM_JOB_BRIEF": os.path.join(registry_dir, jid, "brief.md"), + "MAM_JOB_INSTRUCTIONS": d.get("instructions", ""), + "MAM_PUBLISH_CMD": " ".join(shlex.quote(x) for x in [ + py, "-m", "lib_py.delegation.publish_event", "--registry-dir", registry_dir, "--job", jid] + + (["--expect-dispatch-token", d["dispatch_token"]] if d.get("dispatch_token") else []) + + (["--attempt-id", str(d["attempt_id"])] if d.get("attempt_id") is not None else [])), + }) + timeout = float(job.get("timeout_sec", 3600)) + try: + res = subprocess.run(exec_cmd, shell=True, env=env, capture_output=True, text=True, + timeout=timeout, cwd=os.getcwd()) + rc = res.returncode + tail = (res.stdout or "").strip().splitlines() + detail = tail[-1][:200] if tail else "exit {}".format(rc) + if rc != 0: + err = (res.stderr or "").strip().splitlines() + detail = "exit {}: {}".format(rc, (err[-1] if err else detail)[:200]) + except subprocess.TimeoutExpired: + rc, detail = 124, "agent command exceeded timeout_sec {}".format(int(timeout)) + cur = load_job(jid, registry_dir) + if cur.get("status") not in ("completed", "error", "cancelled"): + _publish(jid, registry_dir, "completed" if rc == 0 else "error", detail, d) + return rc + # No command: an agent on this host publishes the terminal event itself. + deadline = time.monotonic() + float(job.get("timeout_sec", 3600)) + while time.monotonic() < deadline: + cur = load_job(jid, registry_dir) + if cur.get("status") in ("completed", "error", "cancelled"): + return 0 if cur.get("status") == "completed" else 1 + time.sleep(1.0) + _publish(jid, registry_dir, "error", "deadline_timeout (remote runner)", d) + return 2 + + +def run_runner(session: str, registry_dir: str, exec_cmd: Optional[str] = None, once: bool = False, + secret: Optional[str] = None, allow_unsigned: bool = False, + timeout: Optional[float] = None, python: Optional[str] = None) -> int: + secret = secret or os.environ.get(SECRET_ENV) + if not secret and not allow_unsigned: + logger.error("%s is not set: a runner must not execute unsigned dispatches " + "(pass --allow-unsigned only on a private broker)", SECRET_ENV) + return 1 + cfg = broker_config_from_env() + client = make_client("remote-runner", cfg) + inbox: "queue.Queue[Dict[str, Any]]" = queue.Queue() + topic = f"{DEFAULT_TOPIC_ROOT}/+/dispatch" + + def on_connect(_c, _u, _flags, reason_code, _props): + if mqtt_common.reason_code_value(reason_code) != 0: + logger.error("broker connection failed: rc=%s", reason_code) + return + _c.subscribe(topic, qos=1) + print(f"SUBSCRIBED {topic}", flush=True) + + def on_message(_c, _u, msg): + if not msg.payload: + return # a cleared (retired) dispatch + try: + d = json.loads(msg.payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + logger.warning("drop unparseable dispatch on %s", msg.topic) + return + if not isinstance(d, dict) or d.get("kind") != DISPATCH_KIND or d.get("schema_version") != SCHEMA_VERSION: + return + if d.get("session") != session: + return + if secret and not verify_dispatch(d, secret): + logger.warning("drop dispatch for %s: bad or missing signature", d.get("job_id")) + return + if not secret and not allow_unsigned: + return + d["_topic"] = msg.topic + inbox.put(d) + + client.on_connect = on_connect + client.on_message = on_message + client.reconnect_delay_set(min_delay=1, max_delay=16) + mqtt_common.with_retry(lambda: client.connect(cfg.host, cfg.port, cfg.keepalive), + attempts=3, base_delay=0.2, max_delay=1.0)() + client.loop_start() + deadline = time.monotonic() + timeout if timeout else None + handled = 0 + try: + while True: + wait = 1.0 if deadline is None else max(0.0, min(1.0, deadline - time.monotonic())) + try: + d = inbox.get(timeout=wait) + except queue.Empty: + if deadline is not None and time.monotonic() >= deadline: + logger.warning("runner timeout after %d job(s)", handled) + return 0 if handled else 2 + continue + print("DISPATCH job={} attempt={} from={}".format(d.get("job_id"), d.get("attempt_id"), d.get("origin")), + flush=True) + # Retire the retained dispatch before running so a restart cannot replay it. + clear_dispatch(cfg, d["_topic"]) + rc = handle_dispatch(d, registry_dir, session, exec_cmd, python=python) + handled += 1 + print("DONE job={} rc={}".format(d.get("job_id"), rc), flush=True) + if once: + return 0 + finally: + client.loop_stop() + try: + client.disconnect() + except Exception: + pass + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description="remote transport for delegated jobs (Issue #18)") + sub = parser.add_subparsers(dest="command", required=True) + p_d = sub.add_parser("dispatch", help="publish the retained dispatch for a registered job") + p_d.add_argument("--job", required=True) + p_d.add_argument("--registry-dir", default=default_registry_dir()) + p_d.add_argument("--session", required=True, help="remote session name (without the remote: prefix)") + p_d.add_argument("--instructions", default="") + p_d.add_argument("--allow-unsigned", action="store_true") + p_r = sub.add_parser("run", help="run a remote session: execute dispatches addressed to it") + p_r.add_argument("--session", required=True) + p_r.add_argument("--registry-dir", default=default_registry_dir()) + p_r.add_argument("--exec", dest="exec_cmd", default=None, + help="shell command run per job (env: JOB_ID, MAM_JOB_BRIEF, MAM_PUBLISH_CMD, ...)") + p_r.add_argument("--once", action="store_true", help="exit after one job") + p_r.add_argument("--timeout", type=float, default=None, help="exit after this many seconds") + p_r.add_argument("--allow-unsigned", action="store_true") + p_r.add_argument("--python", default=None) + parser.add_argument("-v", "--verbose", action="store_true") + args = parser.parse_args(argv) + mqtt_common.ensure_env_loaded() + mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.INFO) + if args.command == "dispatch": + try: + print(publish_dispatch(args.job, args.registry_dir, args.session, args.instructions, + allow_unsigned=args.allow_unsigned)) + except (FileNotFoundError, RuntimeError) as exc: + logger.error("%s", exc) + return 1 + except Exception as exc: + logger.error("dispatch publish failed: %s", exc) + return 2 + return 0 + return run_runner(args.session, args.registry_dir, args.exec_cmd, args.once, + allow_unsigned=args.allow_unsigned, timeout=args.timeout, python=args.python) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.agents/skills/lib_py/loop/runner.py b/.agents/skills/lib_py/loop/runner.py index 24dfcf8..57b0634 100644 --- a/.agents/skills/lib_py/loop/runner.py +++ b/.agents/skills/lib_py/loop/runner.py @@ -241,6 +241,32 @@ class LoopRunner: return "codex" return "claude" + def session_label(self, name: str) -> str: + """`herdr:` for a pane-backed session, `remote:` for a row + registered with `transport: remote` (Issue #18: the prompt then goes + over the broker and no herdr/pane/lsof/ps path is taken).""" + state = getattr(self, "_sessions_state", None) + if state is None: + state = self.load_sessions_state() + self._sessions_state = state + for s in state.get("herdr_sessions", []): + if s.get("name") == name and str(s.get("transport") or "").lower() == "remote": + return f"remote:{name}" + return f"herdr:{name}" + + def contract_args(self) -> List[str]: + """Liveness-contract knobs passed through to every delegate job + (Issue #18): MAM_HEARTBEAT_SEC (N), MAM_HEARTBEAT_MISSES (K), + MAM_ACK_TIMEOUT_S (T). The wrapper derives --idle-timeout as K x N.""" + args: List[str] = [] + for env_name, flag in (("MAM_HEARTBEAT_SEC", "--heartbeat"), + ("MAM_HEARTBEAT_MISSES", "--heartbeat-misses"), + ("MAM_ACK_TIMEOUT_S", "--ack-timeout")): + val = os.environ.get(env_name) + if val and val.isdigit() and int(val) > 0: + args += [flag, val] + return args + def resolve_all_reviewers(self, state: Dict[str, Any]) -> List[str]: """Resolve running reviewer sessions excluding creator.""" reviewers = [] @@ -266,12 +292,12 @@ class LoopRunner: def _run_cmd(p_text: str): cmd = [ "bash", str(wrapper_script), "submit", - "--agent-session", f"herdr:{agent_session}", + "--agent-session", self.session_label(agent_session), "--agent", agent_type, "--type", "direct", "--role", role, "--prompt", p_text, - ] + ] + self.contract_args() env = dict(os.environ) if "DELEGATE_JOB_PYTHON" not in env: env["DELEGATE_JOB_PYTHON"] = sys.executable diff --git a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md b/.agents/skills/multi-agent-mux-delegate-job/SKILL.md index fb78042..cc6db36 100644 --- a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md +++ b/.agents/skills/multi-agent-mux-delegate-job/SKILL.md @@ -40,7 +40,16 @@ multi-agent-mux-delegate-job submit \ --agent-session herdr: \ --prompt "Task description or instructions here" \ --role \ - --timeout 3600 --idle-timeout 120 + --timeout 3600 [--heartbeat 30 --heartbeat-misses 4 --ack-timeout 60] [--idle-timeout ] + # Liveness contract (Issue #18): the agent heartbeats every N s (automatic), + # the job is unresponsive after K x N s (--idle-timeout overrides), `started` + # must arrive within T s of the injection or the inject fails closed. + +# 1b) Same, for an agent on another host (no herdr pane, no shared disk): +# the brief travels over the broker as a signed retained dispatch and a +# `remote.py run --session ` runner on that host executes it. +multi-agent-mux-delegate-job submit --agent remote --agent-session remote: \ + --prompt "Task description" --role Worker # needs MAM_DISPATCH_SECRET on both hosts # 2) Submit a job with a feedback loop (Worker-Reviewer Loop) multi-agent-mux-delegate-job submit \ @@ -77,8 +86,8 @@ Under fail-closed operation (`MAM_DELEGATE_FAIL_CLOSED=1`, default), the CLI ret | Exit Code | Meaning | Final Registry Status | Failure Behavior & Resilience | |:---:|:---:|:---:|---| | `0` | Success | `completed` | Job succeeded and completed event received or verified on disk with approval (`PASS`/`AGREE`). | -| `1` | Failure | `error` | Execution failure / Agent injection error (`inject_agent`). | -| `2` | Timeout | `error` (`deadline_timeout`) | Subscriber timeout (total budget or idle timeout reached without terminal event). **R1 Working-Agent Resilience**: If the agent in Herdr is still actively `working`, the wrapper does not immediately fail-close; it re-observes disk status at intervals up to the wall-clock `--timeout` deadline before fail-closing. | +| `1` | Failure | `error` | Execution failure / delivery failure: no `started` within `--ack-timeout` T s after the injection (`agent_blocked`; the prompt is never re-injected), or the input never reached the pane. | +| `2` | Timeout | `error` (`deadline_timeout`) | Subscriber timeout (total budget or idle timeout reached without terminal event). **R1 liveness grace (Issue #18)**: after the subscriber idles out the wrapper keeps re-observing the disk while the job's events.log shows an event at most K x N (`--idle-timeout`) old - the automatic heartbeat counts - up to the wall-clock `--timeout`. herdr status and the pane are logged as a diagnostic only. | | `3` | Broker Error | `error` (`deadline_timeout`) | Broker unavailable (`MQTT_BROKER_UNAVAILABLE`) and terminal status not found on disk during deadline. | | `5` | Session Busy | `cancelled` (`session_busy`) | Precondition failed / `session_busy`: Preflight check (`preflight_agent`) or TUI idle handshake wait failed (prompt never delivered, retryable). | @@ -131,11 +140,12 @@ a fresh interpreter — but a test harness could. Every agent participating in the delegation contract must follow the same lifecycle publishing protocol using `publish_event.py`: -1. **On Start**: Publish `started` event. - `python3 .agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py --job "$JOB_ID" --event started` +1. **On Start** (first thing, within the ack timeout T): Publish `started` and arm the heartbeat timer. + `python3 .agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py --job "$JOB_ID" --event started --heartbeat ` + The timer publishes `progress` every N s until your terminal event (Issue #18); the brief and the injected instructions carry the exact command. 2. **On Tool/Permission Prompt**: Publish `permission_required` event. `python3 ... --job "$JOB_ID" --event permission_required --detail ":"` -3. **On Progress Update (Optional)**: Publish `progress` event. +3. **On Progress Update**: Publish `progress` at check-points (the heartbeat timer already satisfies the every-N-seconds obligation). `python3 ... --job "$JOB_ID" --event progress --detail ""` 4. **On Success**: Publish `completed` event. `python3 ... --job "$JOB_ID" --event completed --detail ""` (Reviewer should include `"PASS"` in the detail to approve). diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 4cdb786..b040430 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -76,11 +76,17 @@ multi-agent-mux-delegate-job [options] submit --agent --prompt [--workdir ] [--agent-session