diff --git a/.agents/skills/lib_py/delegation/heartbeat.py b/.agents/skills/lib_py/delegation/heartbeat.py index 3988fde..338cf1e 100644 --- a/.agents/skills/lib_py/delegation/heartbeat.py +++ b/.agents/skills/lib_py/delegation/heartbeat.py @@ -15,8 +15,13 @@ timer stops on its own when * 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`. +A terminal publish also stops it directly, so no heartbeat can trail a +`completed` (review round 2, codex R3): it raises an attempt-scoped barrier +under the registry lock that allocates seqs (a beat either took its seq before +it or is refused when it asks for one), stops the timer through the pidfile +(`/.heartbeat.pid`), and waits for the timer to drain its +in-flight beats - killing it after `MAM_HEARTBEAT_DRAIN_S` - before it takes +its own seq. The timer is bound to the agent's life (review B2): it watches one process - `--watch-pid`, or by default the nearest ancestor of the publishing process @@ -36,6 +41,7 @@ checks read. from __future__ import annotations import argparse +import json import logging import os import signal @@ -154,8 +160,31 @@ def _is_timer_process(pid: int, job_id: str) -> bool: return "lib_py.delegation.heartbeat" in command and job_id in command -def stop_timer(job_id: str, registry_dir: str) -> bool: - """Terminate the job's timer if one runs. Returns True when one was signalled.""" +DRAIN_ENV = "MAM_HEARTBEAT_DRAIN_S" +# Short on purpose: the barrier already keeps any beat from taking a seq after +# the terminal, so the drain only lets an idle timer exit cleanly. A beat still +# on the wire is cancelled with the timer (its seq is older than the terminal's) +# rather than holding back the terminal event - which may be the orchestrator +# closing the job - for seconds of broker round trips. +DEFAULT_DRAIN_S = 1.0 + + +def drain_seconds() -> float: + """How long a terminal publish waits for the timer to drain (env override).""" + try: + return max(0.0, float(os.environ.get(DRAIN_ENV, DEFAULT_DRAIN_S))) + except ValueError: + return DEFAULT_DRAIN_S + + +def stop_timer(job_id: str, registry_dir: str, drain: Optional[float] = None) -> bool: + """Terminate the job's timer if one runs. Returns True when one was signalled. + + ``drain`` (seconds): also wait for the timer to exit - on TERM it stops + scheduling and joins its in-flight beats - and SIGKILL it if it has not + gone by then, so none of its beats can still publish afterwards. A terminal + publish uses this (review round 2, codex R3): "TERM sent" is not "timer + stopped".""" pid = running_timer(job_id, registry_dir) path = pidfile_path(job_id, registry_dir) if pid is None or not _is_timer_process(pid, job_id): @@ -168,6 +197,19 @@ def stop_timer(job_id: str, registry_dir: str) -> bool: os.kill(pid, signal.SIGTERM) except OSError: pass + if drain is not None: + deadline = time.monotonic() + drain + while time.monotonic() < deadline and _pid_alive(pid): + time.sleep(0.1) + if _pid_alive(pid): + logger.warning("heartbeat timer %d for %s did not drain within %.0fs; killing it", pid, job_id, drain) + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline and _pid_alive(pid): + time.sleep(0.05) try: os.unlink(path) except OSError: @@ -175,6 +217,40 @@ def stop_timer(job_id: str, registry_dir: str) -> bool: return True +def barrier_path(job_id: str, registry_dir: str) -> str: + return os.path.join(registry_dir, "{}.heartbeat.barrier".format(job_id)) + + +def raise_barrier(job_id: str, registry_dir: str, dispatch_token=None, attempt_id=None) -> None: + """Mark this attempt as ending (review round 2, codex R3). + + Written under the registry lock - the lock `next_seq` takes - so a beat + either took its seq before this point (and the terminal publisher then + waits for the timer to drain) or meets the barrier when it asks for a seq + and is refused without consuming one.""" + with mqtt_common.registry_lock(registry_dir): + with open(barrier_path(job_id, registry_dir), "w") as fh: + json.dump({"dispatch_token": dispatch_token, "attempt_id": attempt_id}, fh) + + +def barrier_blocks(job_id: str, registry_dir: str, dispatch_token=None, attempt_id=None) -> bool: + """True when a terminal barrier was raised for this attempt (a later + attempt, with its own token, is not blocked). Check it under the registry + lock when the answer must be atomic with a seq.""" + try: + with open(barrier_path(job_id, registry_dir)) as fh: + barrier = json.load(fh) + except (OSError, ValueError): + return False + if not isinstance(barrier, dict): + return True + if dispatch_token and barrier.get("dispatch_token"): + return barrier["dispatch_token"] == dispatch_token + if attempt_id is not None and barrier.get("attempt_id") is not None: + return int(barrier["attempt_id"]) == int(attempt_id) + 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, watch_pid: Optional[int] = None) -> int: @@ -222,6 +298,8 @@ def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id, return "job record gone" if rec.get("status") in TERMINAL_STATUSES: return "job is {}".format(rec.get("status")) + if barrier_blocks(job_id, registry_dir, dispatch_token, attempt_id): + return "the terminal barrier for this attempt is raised" 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: @@ -290,12 +368,15 @@ def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None, def _beat(argv=argv): try: - rc = publish_event.main(argv) + # _beat=True: admitted under the seq lock, never after this + # attempt's terminal barrier (review round 2, codex R3) + rc = publish_event.main(argv, _beat=True) 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) + logger.info("heartbeat publish refused (terminal barrier, terminal record or stale attempt); " + "timer for %s will stop", job_id) stop["flag"] = True t = threading.Thread(target=_beat, daemon=True) diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 3c1d747..86f028a 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -168,6 +168,12 @@ class _Watcher: last = max(self.last_seq.get(jid, 0), self.state_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) + # Review round 2 (codex R4): a state renews liveness only when it carries + # news - a seq beyond every event and state already accepted. The same + # retained snapshot re-delivered on every reconnect, or an older one + # replayed, still recovers a terminal status in the main loop, but it is + # no evidence that the agent is alive now. + renews = seq > last if seq > self.state_seq.get(jid, 0): self.state_seq[jid] = seq mqtt_common.append_event(jid, { @@ -178,7 +184,7 @@ class _Watcher: "timestamp": payload.get("timestamp"), "detail": "status={} last_event={}".format(payload.get("status"), payload.get("last_event")), }) - self.events.put((msg.topic, payload)) + self.events.put((msg.topic, dict(payload, _renews_liveness=renews))) def on_message(self, _client, _userdata, msg) -> None: # --- defensive parsing ------------------------------------------- @@ -474,16 +480,18 @@ def _run_subscriber(args) -> int: _check_disk_fallback(pending, args.registry_dir, terminal, dispatch_tokens, attempts) continue - last_event = time.monotonic() retention_checked = True - first_event_seen = True + if payload.pop("_renews_liveness", True): + last_event = time.monotonic() + 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. + # Issue #18: retained job state. It renewed the idle window + # above only if it advanced the seq (review round 2, codex R4: + # a re-delivered or older snapshot is not liveness); 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" diff --git a/.agents/skills/lib_py/delegation/mqtt_common.py b/.agents/skills/lib_py/delegation/mqtt_common.py index a413779..d506b47 100644 --- a/.agents/skills/lib_py/delegation/mqtt_common.py +++ b/.agents/skills/lib_py/delegation/mqtt_common.py @@ -515,12 +515,25 @@ def _update_job_status_legacy(job_id: str, registry_dir: str = DEFAULT_REGISTRY_ return record -def next_seq(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> int: +class AdmissionRefused(Exception): + """A publish was refused under the registry lock before it took a seq.""" + + +def next_seq(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, admit=None) -> int: """Return the next monotonic sequence number for a job, persisted in the record's ``last_seq`` field so it stays consistent across process restarts. - First call returns 1.""" + First call returns 1. + + ``admit`` (optional) is called with the record under the same lock; a + non-empty return refuses the seq (``AdmissionRefused``) and writes nothing. + Heartbeat beats use it so that a terminal barrier and the seq allocation + are one atomic decision (review round 2, codex R3).""" with registry_lock(registry_dir): record = load_job(job_id, registry_dir) + if admit is not None: + reason = admit(record) + if reason: + raise AdmissionRefused(reason) seq = int(record.get("last_seq", 0)) + 1 record["last_seq"] = seq record["updated_at"] = _utcnow() diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index b049f96..4b00c04 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -122,6 +122,10 @@ def build_state_payload(job: Dict[str, Any], prepared: "PreparedEvent") -> Dict[ event so a forged state cannot close a job. """ status = EVENT_TO_STATUS.get(prepared.event) or "running" + if prepared.event not in TERMINAL_EVENTS and job.get("status") in TERMINAL_EVENTS: + # A late non-terminal event must not publish a retained `running` over + # the terminal snapshot (review round 2, codex R3). + status = job["status"] 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", @@ -164,18 +168,20 @@ class PreparedEvent(object): def prepare_event(job_id, event, detail="", data=None, registry_dir=DEFAULT_REGISTRY_DIR, retained=False, - dispatch_token=None, attempt_id=None) -> PreparedEvent: + dispatch_token=None, attempt_id=None, admit=None) -> PreparedEvent: """Load the job, consume the next seq, and build the signed payload. Consuming the seq here (before any transport runs) is deliberate: a failed publish must still burn the number, so a replayed packet can never carry a - seq the subscriber would accept. + seq the subscriber would accept. ``admit`` is checked under the lock that + allocates the seq (``mqtt_common.next_seq``); a refusal raises + ``AdmissionRefused`` and consumes nothing. """ job = load_job(job_id, registry_dir) config = broker_config_from_job(job) topic = job.get("topic_prefix") topic = f"{topic}/events" if topic else events_topic_for(job_id) - seq = next_seq(job_id, registry_dir) + seq = next_seq(job_id, registry_dir) if admit is None else next_seq(job_id, registry_dir, admit=admit) payload_data = dict(data) if data else {} eff_dispatch_token = dispatch_token @@ -265,7 +271,28 @@ def _resolve_job_id(args) -> Optional[str]: return args.job -def main(argv=None) -> int: +def _beat_admission(job_id, registry_dir, dispatch_token, attempt_id): + """Admission for one heartbeat beat, evaluated under the lock that allocates + its seq (review round 2, codex R3): never after this attempt's terminal + barrier, on a terminal record, or after a re-dispatch.""" + def admit(record): + status = record.get("status") + if status in TERMINAL_EVENTS: + return "job is {}".format(status) + if heartbeat_timer.barrier_blocks(job_id, registry_dir, dispatch_token, attempt_id): + return "the terminal barrier for this attempt is raised" + if dispatch_token and record.get("dispatch_token") != dispatch_token: + return "dispatch token changed (re-dispatched)" + if attempt_id is not None and record.get("attempt_id", record.get("iteration", 1)) != attempt_id: + return "attempt changed (re-dispatched)" + return None + return admit + + +def main(argv=None, _beat: bool = False) -> int: + """CLI entry point. ``_beat`` is set only by the heartbeat timer + (``heartbeat.run``): the beat is admitted under the seq lock + (``_beat_admission``) and a refusal returns 4, which stops the timer.""" parser = argparse.ArgumentParser(description="Publish a Job event to MQTT") target = parser.add_mutually_exclusive_group(required=True) target.add_argument("--job", help="job id to publish for") @@ -389,10 +416,19 @@ 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`. + # Issue #18: a terminal event ends the heartbeat obligation. Review round 2 + # (codex R3): first raise this attempt's barrier - under the lock that + # allocates seqs, so no beat can take a seq after it - then stop the timer + # and wait for it to drain its in-flight beats (killing it if it does not), + # and only then take this event's seq. No heartbeat can follow the terminal + # event, on disk or on the wire. if args.event in TERMINAL_EVENTS: - if heartbeat_timer.stop_timer(job_id, args.registry_dir): + if rec is not None: + heartbeat_timer.raise_barrier( + job_id, args.registry_dir, + dispatch_token=args.expect_dispatch_token or actual_tok, + attempt_id=args.attempt_id if args.attempt_id is not None else rec_attempt) + if heartbeat_timer.stop_timer(job_id, args.registry_dir, drain=heartbeat_timer.drain_seconds()): logger.info("stopped the heartbeat timer for job %s", job_id) try: @@ -405,7 +441,11 @@ def main(argv=None) -> int: retained=args.retained, dispatch_token=args.expect_dispatch_token, attempt_id=args.attempt_id, + admit=_beat_admission(job_id, args.registry_dir, check_token, check_attempt) if _beat else None, ) + except mqtt_common.AdmissionRefused as exc: + logger.info("heartbeat for job %s not published: %s", job_id, exc) + return 4 except FileNotFoundError as exc: logger.error("%s", exc) return 1 diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index 8d3de48..e5e4be3 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -88,6 +88,7 @@ def register_job( dispatch_token: Optional[str] = None, attempt_id: int = 1, transport: str = "herdr", + supersede_older_attempt: bool = False, ) -> str: """Create a new ``pending`` job record and return its id. @@ -147,7 +148,17 @@ def register_job( } with registry_lock(registry_dir): if mqtt_common._job_path(job_id, registry_dir).exists(): - raise FileExistsError(f"job already exists: {job_id}") + if not supersede_older_attempt: + raise FileExistsError(f"job already exists: {job_id}") + # A remote runner adopting the orchestrator's next attempt (review + # round 2, codex R1). Decided under this lock, so a stale or + # conflicting dispatch can never roll the record back. + old = mqtt_common.load_job(job_id, registry_dir) + old_attempt = int(old.get("attempt_id", old.get("iteration", 1)) or 1) + if int(attempt_id) <= old_attempt: + raise FileExistsError( + f"job {job_id} is already at attempt {old_attempt}; refusing attempt {attempt_id}") + record["last_seq"] = int(old.get("last_seq", 0) or 0) # seqs stay monotonic across attempts _atomic_write_record(job_id, registry_dir, record) # Seed the persistent audit log (meta.json + status.json + a "registered" # event). Best-effort inside init_job_log — never blocks registration. diff --git a/.agents/skills/lib_py/delegation/remote.py b/.agents/skills/lib_py/delegation/remote.py index ebbc352..cdbd0eb 100644 --- a/.agents/skills/lib_py/delegation/remote.py +++ b/.agents/skills/lib_py/delegation/remote.py @@ -33,6 +33,7 @@ import logging import os import queue import shlex +import signal import socket import subprocess import sys @@ -153,15 +154,39 @@ def clear_dispatch(config, topic: str) -> None: # remote side # --------------------------------------------------------------------------- +def _attempt_of(record: Dict[str, Any]) -> int: + try: + return int(record.get("attempt_id", record.get("iteration", 1)) or 1) + except (TypeError, ValueError): + return 1 + + def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> Optional[str]: - """Create the local job record a remote agent publishes against. + """Create the local job record a remote agent publishes against, or refuse + the dispatch (None). 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. + orchestrator's subscriber accepts. Admission (review round 2, codex R1): a + valid signature proves where a dispatch came from, not that it is current, + so + + * the same dispatch_token is idempotent while its attempt is open, and is + never run again once that attempt reached a terminal; + * a lower attempt, or the same attempt under another token, is a stale or + conflicting copy: refused before anything is written, so the record and + its events.log stay as they are; + * only a strictly newer attempt replaces the record, and that check and + the write are one step under the registry lock + (``register_job(supersede_older_attempt=True)``). The events.log is kept: + the history of earlier attempts is not erased. """ jid = d["job_id"] + try: + incoming = int(d.get("attempt_id") or 1) + except (TypeError, ValueError): + logger.warning("drop dispatch for %s: attempt_id %r is not a number", jid, d.get("attempt_id")) + return None path = mqtt_common._job_path(jid, registry_dir) if path.exists(): old = load_job(jid, registry_dir) @@ -169,28 +194,32 @@ def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> Optional[s if old.get("status") in ("completed", "error", "cancelled"): return None # this attempt already ran to its end: a re-sent dispatch is not run again 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", - ) + if incoming <= _attempt_of(old): + logger.warning("refusing dispatch for %s attempt %s: this runner already holds attempt %s " + "(a stale or conflicting copy)", jid, incoming, _attempt_of(old)) + return None + try: + 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=incoming, + transport="remote", + supersede_older_attempt=True, + ) + except FileExistsError as exc: # lost the race to a newer (or equal) attempt under the lock + logger.warning("refusing dispatch for %s: %s", jid, exc) + return None 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: @@ -212,13 +241,68 @@ def _publish(jid: str, registry_dir: str, event: str, detail: str, d: Dict[str, return publish_event.main(argv) +def _group_alive(pgid: int) -> bool: + try: + os.killpg(pgid, 0) + except OSError: + return False + return True + + +def _terminate_group(proc: "subprocess.Popen", grace: float = 5.0) -> None: + """TERM, then KILL, the agent command's own process group (it runs in its + own session) and wait until the group is gone. Only that owned group is + signalled (review round 2, codex R2).""" + for sig, wait in ((signal.SIGTERM, grace), (signal.SIGKILL, 2.0)): + try: + os.killpg(proc.pid, sig) + except OSError: + return # the group is already gone + deadline = time.monotonic() + wait + while time.monotonic() < deadline: + proc.poll() # reap the shell itself + if not _group_alive(proc.pid): + return + time.sleep(0.1) + + +def _run_agent_command(exec_cmd: str, env: Dict[str, str], timeout: float): + """Run the agent command in its own session; returns (rc, stdout, stderr, + timed_out). At the deadline - or when the runner itself is interrupted - + the whole process group is terminated before this returns, so no + descendant keeps working past the job's timeout or overlaps the next + dispatch (review round 2, codex R2: `subprocess.run(shell=True, + timeout=...)` killed only the shell).""" + proc = subprocess.Popen(exec_cmd, shell=True, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, cwd=os.getcwd(), start_new_session=True) + try: + out, err = proc.communicate(timeout=timeout) + return proc.returncode, out, err, False + except subprocess.TimeoutExpired: + _terminate_group(proc) + try: + out, err = proc.communicate(timeout=5) + except subprocess.TimeoutExpired: + # a descendant that left the group still holds the pipes: stop reading + for stream in (proc.stdout, proc.stderr): + try: + stream.close() + except Exception: + pass + out, err = "", "" + return 124, out, err, True + except BaseException: + _terminate_group(proc) + raise + + 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) if jid is None: - logger.info("dispatch for %s attempt %s already finished here; ignoring the re-sent copy", - d.get("job_id"), d.get("attempt_id")) + logger.info("dispatch for %s attempt %s not admitted (already finished here, or a stale or " + "conflicting copy); not running it", d.get("job_id"), d.get("attempt_id")) return 0 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), @@ -245,17 +329,15 @@ def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str, + (["--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() + rc, out, err, timed_out = _run_agent_command(exec_cmd, env, timeout) + if timed_out: + detail = "agent command exceeded timeout_sec {}".format(int(timeout)) + else: + tail = (out 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)) + errs = (err or "").strip().splitlines() + detail = "exit {}: {}".format(rc, (errs[-1] if errs else detail)[:200]) 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) @@ -316,6 +398,16 @@ def run_runner(session: str, registry_dir: str, exec_cmd: Optional[str] = None, 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)() + # Review round 2 (codex R2): SIGTERM unwinds the runner like an interrupt, + # so a command still running is terminated together with its process group + # (_run_agent_command) instead of being left behind as an orphan. + def _on_term(signum, _frame): + raise SystemExit(128 + signum) + + try: + previous_term = signal.signal(signal.SIGTERM, _on_term) + except ValueError: # not the main thread (a caller driving the runner in-process) + previous_term = None client.loop_start() deadline = time.monotonic() + timeout if timeout else None handled = 0 @@ -339,6 +431,8 @@ def run_runner(session: str, registry_dir: str, exec_cmd: Optional[str] = None, if once: return 0 finally: + if previous_term is not None: + signal.signal(signal.SIGTERM, previous_term) client.loop_stop() try: client.disconnect() diff --git a/.agents/skills/multi-agent-mux-loop/SKILL.md b/.agents/skills/multi-agent-mux-loop/SKILL.md index c32b41e..7e7fa26 100644 --- a/.agents/skills/multi-agent-mux-loop/SKILL.md +++ b/.agents/skills/multi-agent-mux-loop/SKILL.md @@ -55,7 +55,10 @@ A session row in `.mam/agent-sessions.yaml` with `transport: remote` is delegated as `remote:` (the brief goes over the broker to a `remote.py run --session ` runner on another host; no herdr, pane or process probing happens for it). The row still needs `role` and -`status: running` to be picked up by `--all-reviewer`. +`status: running` to be picked up by `--all-reviewer`. Remote rows need the +default Python loop runner: the legacy shell runner (`MAM_LOOP_PY=0`) +addresses every session as `herdr:` and does not support +`transport: remote`. ## Roles & Responsibilities diff --git a/MESSAGING.md b/MESSAGING.md index 7dae79c..f64704e 100644 --- a/MESSAGING.md +++ b/MESSAGING.md @@ -196,11 +196,11 @@ orchestrator's disk, take part on equal terms. | Rule | Contract | Where | |---|---|---| -| **Heartbeat obligation** | While working, the agent publishes `progress` at least every **N** s (`heartbeat_sec`, default 30). `publish_event.py --event started --heartbeat N` starts a detached timer that publishes `progress` (`detail: heartbeat`) every N s with the same `dispatch_token`/`attempt_id`, so the agent does not have to remember. The timer stops itself at the terminal event (which also kills it through `/.heartbeat.pid`), on re-dispatch, after `timeout_sec`, and within one period of the agent process it watches going away (`--watch-pid`, default the nearest agent-CLI ancestor of the publishing command; a remote runner watches itself) - a crashed agent stops heartbeating, an alive-but-stuck one does not. | `lib_py/delegation/heartbeat.py`, `publish_event.py --heartbeat` | -| **Unresponsive** | The orchestrator treats a job as unresponsive after **K x N** s without any event (`heartbeat_misses`, default 4 → 120 s; `--idle-timeout` overrides). The subscriber's idle clock counts every accepted message (`progress` and retained state included); the wrapper's post-idle grace re-reads only the job's events.log (`_recent_disk_heartbeat`, written before every wire publish). herdr status and the pane are logged once as a *diagnostic* (`liveness: ... herdr/pane says busy=yes/no (diagnostic only)`) and never decide. | `multi-agent-mux-delegate-job` (`finalize_unfinished`), `job_subscriber.py` | +| **Heartbeat obligation** | While working, the agent publishes `progress` at least every **N** s (`heartbeat_sec`, default 30). `publish_event.py --event started --heartbeat N` starts a detached timer that publishes `progress` (`detail: heartbeat`) every N s with the same `dispatch_token`/`attempt_id`, so the agent does not have to remember. No heartbeat can follow a terminal event: the terminal publish first raises an attempt-scoped barrier (`/.heartbeat.barrier`, written under the registry lock that also allocates seqs, so a beat either took its seq before it or is refused), then stops the timer through `/.heartbeat.pid` and waits for it to drain its in-flight beats (killing it after `MAM_HEARTBEAT_DRAIN_S`, default 1 s - a beat still on the wire is cancelled, never allowed to trail the terminal), and only then takes its own seq. The timer also stops itself on a terminal record or barrier, on re-dispatch, after `timeout_sec`, and within one period of the agent process it watches going away (`--watch-pid`, default the nearest agent-CLI ancestor of the publishing command; a remote runner watches itself) - a crashed agent stops heartbeating, an alive-but-stuck one does not. | `lib_py/delegation/heartbeat.py`, `publish_event.py --heartbeat` | +| **Unresponsive** | The orchestrator treats a job as unresponsive after **K x N** s without any event (`heartbeat_misses`, default 4 → 120 s; `--idle-timeout` overrides). K x N is when silence is *detected*; the wrapper's post-idle grace below can stretch the final close to about 2 x K x N. The subscriber's idle clock counts every accepted event, and a retained state only when its seq is newer than every event and state already accepted - the same snapshot re-delivered on a reconnect, or an older one, is not liveness; the wrapper's post-idle grace re-reads only the job's events.log (`_recent_disk_heartbeat`, written before every wire publish). herdr status and the pane are logged once as a *diagnostic* (`liveness: ... herdr/pane says busy=yes/no (diagnostic only)`) and never decide. | `multi-agent-mux-delegate-job` (`finalize_unfinished`), `job_subscriber.py` | | **Delivery** | After the prompt is handed over (herdr accepted the submission, or the remote dispatch was published), the agent's `started` must arrive within **T** s (`ack_timeout_sec`, default 180; a real agent reads the brief before it runs the start command): the registry leaves `pending`, a `started` line lands in events.log, or the subscriber prints one. No `started` within T → the inject fails closed: rc 1, registry `error` (`agent_blocked`), and the prompt is **never re-injected**. `send_keys_safe` confirms an accepted-but-unconfirmed submission only by `agent_status == working` or the job's `started`; pane changes and busy tokens are no longer signals. | `_wait_started`, `lib.sh: _sks_job_started` | | **Termination** | Only `completed` / `error` (and an operator `cancelled`) end a job. Terminal immutability (Issue #6 CAS) refuses anything that arrives after `deadline_timeout` closed the job. | `engine.py` | -| **Wire durability** | Every event (QoS 1) is followed, on the same connection, by a **retained** `/state` message (`event: state`, `status`, `last_event`, `seq`, signed like an event). A subscriber that connects or reconnects receives it at once: a terminal status ends the wait, any state resets the idle window, and a `seq` beyond the last one seen is recorded as `seq_gap` in the audit log (also for gaps between events). paho reconnects on its own; `on_connect` re-subscribes both topics and logs `reconnected`. | `publish_event.py: build_state_payload`, `job_subscriber.py: _on_state` | +| **Wire durability** | Every event (QoS 1) is followed, on the same connection, by a **retained** `/state` message (`event: state`, `status`, `last_event`, `seq`, signed like an event). A subscriber that connects or reconnects receives it at once: a terminal status ends the wait, a state whose seq is newer than every event and state already accepted resets the idle window (a re-delivered or older snapshot still recovers a terminal status but is not liveness), and a `seq` beyond the last one seen is recorded as `seq_gap` in the audit log (also for gaps between events). paho reconnects on its own; `on_connect` re-subscribes both topics and logs `reconnected`. | `publish_event.py: build_state_payload`, `job_subscriber.py: _on_state` | | **Credentials** | Each host connects with its own `.mam.env` (Issue #12: the password is never in a record). Every event and state message carries the job's HMAC (`auth_token`). A remote dispatch is signed with the pre-shared `MAM_DISPATCH_SECRET`. | `mqtt_common.py`, `remote.py` | | **What screen scraping is still for** | (a) operations that need the screen: prompt injection into a pane, trust/permission dialogs, the pre-injection idle handshake; (b) diagnostics on the same host when events stop. Neither is a liveness or delivery decision. | `lib.sh` | diff --git a/VERSIONS.md b/VERSIONS.md index a5fc109..6414876 100644 --- a/VERSIONS.md +++ b/VERSIONS.md @@ -43,15 +43,15 @@ > **마이너 마일스톤 (MINOR Release)**: 에이전트의 생존·전달 판정을 TUI 스크래핑(herdr `agent_status`, pane 문자열, 화면 변화)에서 떼어내어, 에이전트가 스스로 발행하는 이벤트(`started`/`progress`/`completed`/`error`)에 **의무·주기·타임아웃 규칙이 있는 계약**을 부여했다(Issue #18, #17 항목 1). 새 이벤트 종류는 없다. 원격 호스트의 에이전트(pane·디스크 공유 없음)를 같은 계약으로 다루는 `remote:` 전송을 추가했다. 래퍼 옵션 3개(`--heartbeat`, `--heartbeat-misses`, `--ack-timeout`)와 `publish_event --heartbeat`, `job_subscriber --remote`, `remote.py` CLI 가 추가되어 MINOR 로 올린다. 기본값(N=30, K=4 → idle 120 s, T=180)은 종전 idle 기본값과 같으므로 기존 호출은 동작이 유지된다. #### 🌟 주요 개선 사항 -1. **heartbeat 의무의 자동화 (`lib_py/delegation/heartbeat.py`, `publish_event.py --heartbeat N`)**: `started --heartbeat N` 이 분리 세션의 타이머를 띄워 N초마다 `progress`(detail `heartbeat`)를 같은 `dispatch_token`/`attempt_id` 로 서명·발행한다. 터미널 이벤트가 pidfile(`/.heartbeat.pid`)로 타이머를 먼저 멈추고(pid 재사용 대비 cmdline 확인), 타이머는 레코드 터미널·재dispatch·`timeout_sec` 초과·**감시 대상 에이전트 프로세스 소멸**(`--watch-pid`, 기본은 시작 명령을 실행한 가장 가까운 에이전트 CLI 조상; 원격 러너는 자기 자신) 시 스스로 끝난다. 죽은 에이전트는 heartbeat 를 멈추고 K×N 뒤 닫히며, 살아 있지만 멈춘 에이전트는 계속 heartbeat 를 낸다(프로세스 생존 계약). 각 beat 는 와이어 이전에 events.log 에 기록되므로 디스크 생존 판정에도 그대로 쓰인다. 래퍼가 주입하는 시작 명령이 `--heartbeat N` 을 포함하고, 브리프에 계약(N/K/T)이 명시된다. +1. **heartbeat 의무의 자동화 (`lib_py/delegation/heartbeat.py`, `publish_event.py --heartbeat N`)**: `started --heartbeat N` 이 분리 세션의 타이머를 띄워 N초마다 `progress`(detail `heartbeat`)를 같은 `dispatch_token`/`attempt_id` 로 서명·발행한다. 터미널 이벤트는 먼저 attempt 범위의 barrier(`/.heartbeat.barrier`, seq 를 떼는 레지스트리 lock 안에서 기록)를 세우고, pidfile(`/.heartbeat.pid`)로 타이머를 멈춘 뒤 진행 중인 beat 가 끝날 때까지 기다린 다음(`MAM_HEARTBEAT_DRAIN_S`, 기본 1 s 뒤 강제 종료 — 와이어에 남은 beat 는 취소되며 터미널을 붙잡아 두지 않는다; pid 재사용 대비 cmdline 확인) 자기 seq 를 뗀다. beat 는 같은 lock 안에서 barrier·터미널 레코드를 확인해 거부되므로 터미널 뒤에는 heartbeat 가 디스크에도 와이어에도 없고, 늦은 비터미널 이벤트의 retained state 가 터미널 스냅샷을 `running` 으로 덮지 않는다(리뷰 2회차 codex R3). 타이머는 레코드 터미널·재dispatch·`timeout_sec` 초과·**감시 대상 에이전트 프로세스 소멸**(`--watch-pid`, 기본은 시작 명령을 실행한 가장 가까운 에이전트 CLI 조상; 원격 러너는 자기 자신) 시 스스로 끝난다. 죽은 에이전트는 heartbeat 를 멈추고 K×N 뒤 닫히며, 살아 있지만 멈춘 에이전트는 계속 heartbeat 를 낸다(프로세스 생존 계약). 각 beat 는 와이어 이전에 events.log 에 기록되므로 디스크 생존 판정에도 그대로 쓰인다. 래퍼가 주입하는 시작 명령이 `--heartbeat N` 을 포함하고, 브리프에 계약(N/K/T)이 명시된다. 2. **무응답 판정 = 마지막 이벤트로부터 K×N (`registry.py`, 래퍼 `finalize_unfinished`)**: 레지스트리 레코드에 `heartbeat_sec`·`heartbeat_misses`·`ack_timeout_sec`·`transport` 가 추가되고 `idle_timeout_sec` 은 명시하지 않으면 K×N 으로 유도된다. 래퍼의 R1 유예는 events.log 의 최근 이벤트(K×N 이내)만을 증거로 삼고, herdr 상태·pane busy 토큰은 진입 시 한 번 `liveness: ... (diagnostic only)` 로 기록만 한다. 헤드리스 에이전트(항상 idle, 빈 pane)도 heartbeat 만으로 K×N 을 넘겨 살아 있고, heartbeat 가 끊기면 `deadline_timeout` 으로 닫히며 이후 `completed` 는 CAS 가 거부한다. 3. **전달 확인 = T 내 `started` (`_wait_started`, `lib.sh: _sks_job_started`)**: 주입 후 T초(`--ack-timeout`, 기본 180 — 실제 에이전트는 브리프를 읽고 나서 시작 명령을 실행하므로 여유 있게; 벽시계 상한) 안에 레지스트리가 `pending` 을 벗어나거나 events.log(이번 attempt 의 dispatch_token 을 가진 `started` 만)/구독자 로그에 `started` 가 없으면 rc 1·`agent_blocked` 로 fail-closed 하고 재주입하지 않는다. `send_keys_safe` 는 accepted-but-unconfirmed 제출을 `agent_status working` 또는 job 의 `started` 로만 확인하며 pane 변화·busy 토큰 신호는 제거됐다(무신호 rc 4 는 유지). `MAM_INJECT_ACK_GRACE_S` 는 T 의 override 로 남는다. -4. **와이어 내구성 (`publish_event.py: build_state_payload`, `job_subscriber.py`)**: 모든 이벤트 뒤에 같은 연결로 retained `/state`(status·last_event·seq, HMAC 서명)를 발행한다. 구독자는 state 토픽도 구독해 (재)접속 즉시 터미널 상태를 복구하고, 모든 메시지(progress·state 포함)로 idle 창을 리셋하며, seq 점프를 감사 로그 `seq_gap` 으로 남기고, 재접속을 `reconnected` 로 기록한다. `--no-state` 로 끌 수 있다. -5. **원격 전송 (`lib_py/delegation/remote.py`, `scripts/remote.py`, 래퍼 `remote:`)**: `--agent-session remote:` 이면 preflight·idle 핸드셰이크·pane 주입을 건너뛰고 `MAM_DISPATCH_SECRET` 로 서명한 retained `/dispatch`(브리프·auth_token·dispatch_token·attempt_id·계약)를 발행한 뒤 T 내 `started` 를 기다린다. 다른 호스트의 `remote.py run --session [--exec CMD] [--once]` 러너가 자기 레지스트리에 같은 id·identity 로 잡을 미러링하고, dispatch 를 retire 하고, `started --heartbeat N` 후 명령을 실행해 터미널 이벤트를 발행한다. 구독자의 `--remote`(transport=remote 잡은 자동)가 `started`·터미널을 오케스트레이터 레지스트리에 반영한다. 서명 없는 dispatch 는 거부된다. 루프 러너는 YAML 행의 `transport: remote` 를 `remote:` 으로 넘기고, `MAM_HEARTBEAT_SEC`/`MAM_HEARTBEAT_MISSES`/`MAM_ACK_TIMEOUT_S` 를 계약 옵션으로 전달한다. +4. **와이어 내구성 (`publish_event.py: build_state_payload`, `job_subscriber.py`)**: 모든 이벤트 뒤에 같은 연결로 retained `/state`(status·last_event·seq, HMAC 서명)를 발행한다. 구독자는 state 토픽도 구독해 (재)접속 즉시 터미널 상태를 복구하고, 이벤트, 그리고 지금까지 받은 어떤 이벤트·state 보다 seq 가 앞선 state 로만 idle 창을 리셋하며(재접속 때 다시 오는 스냅샷이나 더 오래된 state 는 터미널 복구에만 쓰고 생존으로 치지 않는다 — 리뷰 2회차 codex R4), seq 점프를 감사 로그 `seq_gap` 으로 남기고, 재접속을 `reconnected` 로 기록한다. `--no-state` 로 끌 수 있다. +5. **원격 전송 (`lib_py/delegation/remote.py`, `scripts/remote.py`, 래퍼 `remote:`)**: `--agent-session remote:` 이면 preflight·idle 핸드셰이크·pane 주입을 건너뛰고 `MAM_DISPATCH_SECRET` 로 서명한 retained `/dispatch`(브리프·auth_token·dispatch_token·attempt_id·계약)를 발행한 뒤 T 내 `started` 를 기다린다. 다른 호스트의 `remote.py run --session [--exec CMD] [--once]` 러너가 자기 레지스트리에 같은 id·identity 로 잡을 미러링하고, dispatch 를 retire 하고, `started --heartbeat N` 후 명령을 실행해 터미널 이벤트를 발행한다. 구독자의 `--remote`(transport=remote 잡은 자동)가 `started`·터미널을 오케스트레이터 레지스트리에 반영한다. 서명 없는 dispatch 는 거부된다. 러너는 dispatch 를 레지스트리 lock 안에서 받아들인다: 같은 토큰은 멱등(끝난 attempt 는 다시 돌리지 않음), 더 낮은 attempt 나 같은 attempt 의 다른 토큰은 레코드·events.log 를 건드리지 않고 거부, 더 새 attempt 만 교체한다(`register_job(supersede_older_attempt=True)`, events.log 는 보존). `--exec` 명령은 자기 세션(프로세스 그룹)에서 돌고, `timeout_sec` 이나 러너 중단(SIGTERM/SIGINT) 때 그룹 전체를 TERM→KILL 로 끝낸 뒤 오류를 발행한다(리뷰 2회차 codex R1·R2). 원격 행은 기본 Python 루프 러너에서만 지원된다(레거시 `MAM_LOOP_PY=0` 은 `herdr:` 전용). 루프 러너는 YAML 행의 `transport: remote` 를 `remote:` 으로 넘기고, `MAM_HEARTBEAT_SEC`/`MAM_HEARTBEAT_MISSES`/`MAM_ACK_TIMEOUT_S` 를 계약 옵션으로 전달한다. 6. **문서**: MESSAGING.md §2.5 계약 표·원격 전송, delegate-job SKILL.md(옵션·종료 코드·프로토콜), registry.md(계약 필드), loop SKILL.md(환경변수·원격 행). #### ✅ 품질 검증 -- **신규 테스트 16개** (`tests/test_issue18_liveness_contract.py`): 레지스트리 계약 필드·idle 유도(2), 자동 heartbeat 디스크 기록·터미널 정지·재dispatch 정지·터미널 이벤트와의 조합 거부(3), retained state 서명·seq 갭·state 로 터미널 복구·위조/stale state 드롭(3), `started` 기반 전달 확인과 pane 신호 무시(2), 수용 기준 1·2·3(헤드리스 heartbeat 생존, T 내 미`started` fail-closed·단일 주입, heartbeat 소실 후 deadline_timeout·늦은 completed 거부), 수용 기준 5(브로커만 공유하는 원격 러너 시뮬레이션), 서명 없는 dispatch 거부, 루프 러너 계약·`remote:` 라벨. +- **신규 테스트 25개** (`tests/test_issue18_liveness_contract.py`): 레지스트리 계약 필드·idle 유도(2), 자동 heartbeat 디스크 기록·터미널 정지·재dispatch 정지·터미널 이벤트와의 조합 거부(3), retained state 서명·seq 갭·state 로 터미널 복구·위조/stale state 드롭(3), `started` 기반 전달 확인과 pane 신호 무시(2), 수용 기준 1·2·3(헤드리스 heartbeat 생존, T 내 미`started` fail-closed·단일 주입, heartbeat 소실 후 deadline_timeout·늦은 completed 거부), 수용 기준 5(브로커만 공유하는 원격 러너 시뮬레이션), 서명 없는 dispatch 거부, 루프 러너 계약·`remote:` 라벨. 리뷰 교정분 — 1회차: 감시 에이전트 프로세스 소멸 시 타이머 정지, 재사용된 pid 미신호, 이번 attempt 의 `started` 만 인정, 끝난 attempt 의 재전송 dispatch 무시; 2회차: 더 낮은·충돌 attempt 의 dispatch 거부와 레코드·로그 무롤백, 명령 타임아웃 시 프로세스 그룹 전체 종료, in-flight beat 가 터미널 뒤에 발행되지 않음, state 는 seq 가 앞설 때만 생존 갱신(단위·재전달 종단 2건). - **기존 테스트 조정** (`tests/test_issue14_delivery_ack.py`): busy 토큰·pane 변화 확인 테스트를 `started` 기반으로 뒤집고(수용 기준 6), R1 유예의 pane busy 토큰 테스트를 "pane 은 생존 증거가 아니다"로 전환. registry CLI 스냅샷·help 캡처(publish_event/job_subscriber)를 의도된 확장으로 갱신. ### 🚀 `v4.4.5` — Issue #19 구독 데몬 수명 계약: 샌드박스 옵트아웃·회수, 워크스페이스 소실 시 자기 종료, SIGTERM 전달 (2026-09-13) diff --git a/tests/test_issue18_liveness_contract.py b/tests/test_issue18_liveness_contract.py index 6a8f5a9..59f6d5a 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/tests/test_issue18_liveness_contract.py @@ -30,6 +30,7 @@ import subprocess import sys import threading import time +import uuid from pathlib import Path import pytest @@ -505,6 +506,265 @@ def test_remote_runner_ignores_a_resent_dispatch_of_a_finished_attempt(tmp_path, assert _job(reg, "feed1234")["dispatch_token"] == "tok-2" and _job(reg, "feed1234")["status"] == "pending" +def test_remote_runner_never_rolls_a_newer_attempt_back(tmp_path, monkeypatch): + """Review round 2 (codex R1): a validly signed dispatch for attempt 1 that + arrives after attempt 2 is stale. It must neither replace the record nor + touch its events.log, and it is never run; the same attempt under another + token is refused as a conflict.""" + from lib_py.delegation import remote + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent")) + monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs")) + reg = tmp_path / "jobs" + jid = uuid.uuid4().hex[:8] + a1 = {"job_id": jid, "brief": "attempt one", "dispatch_token": "tok-" + uuid.uuid4().hex, "attempt_id": 1, + "auth_token": None, "timeout_sec": 30, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5} + b2 = dict(a1, brief="attempt two", dispatch_token="tok-" + uuid.uuid4().hex, attempt_id=2) + assert remote.mirror_job(a1, str(reg), "s") == jid + assert remote.mirror_job(b2, str(reg), "s") == jid, "a newer attempt is admitted" + log = reg / f"{jid}.events.log" + log.write_text(json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": b2["dispatch_token"]}}) + "\n") + record_before, log_before = (reg / f"{jid}.json").read_text(), log.read_text() + + assert remote.mirror_job(a1, str(reg), "s") is None, "the replayed attempt-1 dispatch is stale" + assert remote.mirror_job(dict(b2, dispatch_token="tok-" + uuid.uuid4().hex), str(reg), "s") is None, \ + "the same attempt under another token is a conflict" + assert (reg / f"{jid}.json").read_text() == record_before, "no record rollback" + assert log.read_text() == log_before, "no events.log rollback" + ran = tmp_path / "ran" + assert remote.handle_dispatch(a1, str(reg), "s", f"touch {ran}") == 0 + assert not ran.exists(), "a stale dispatch is never executed" + rec = _job(reg, jid) + assert (rec["attempt_id"], rec["dispatch_token"]) == (2, b2["dispatch_token"]) + assert (reg / jid / "brief.md").read_text() == "attempt two" + + +def test_remote_command_timeout_ends_the_whole_process_group(tmp_path, monkeypatch): + """Review round 2 (codex R2): at timeout_sec the runner stops the whole + command - not only its shell - before it reports the error, so no + descendant keeps working past the deadline.""" + from lib_py.delegation import remote + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent")) + monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs")) + published = [] + monkeypatch.setattr(remote, "_publish", lambda jid, reg, event, detail, d, heartbeat=None: + published.append((event, detail)) or 0) + reg = tmp_path / "jobs" + pid_file, marker = tmp_path / "child.pid", tmp_path / "late.marker" + child = tmp_path / "child.py" + child.write_text("import os, time\n" + f"open({str(pid_file)!r}, 'w').write(str(os.getpid()))\n" + "time.sleep(2)\n" + f"open({str(marker)!r}, 'w').write('too late')\n") + d = {"job_id": uuid.uuid4().hex[:8], "brief": "b", "dispatch_token": "tok-" + uuid.uuid4().hex, "attempt_id": 1, + "auth_token": None, "timeout_sec": 1, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5} + pid = None + try: + # `; :` keeps the shell alive as the python child's parent (no exec) + rc = remote.handle_dispatch(d, str(reg), "s", f'"{sys.executable}" "{child}"; :') + assert rc == 124 + assert published[-1][0] == "error" and "timeout_sec 1" in published[-1][1], published + pid = int(pid_file.read_text()) + time.sleep(2.5) + assert not marker.exists(), "a descendant kept working past the deadline" + assert not _alive(pid) + finally: + if pid and _alive(pid): + os.kill(pid, signal.SIGKILL) + + +_PAUSING_TIMER = r''' +import json, os, sys, time +from lib_py.delegation import heartbeat as hb, publish_event as pe + +capture, paused, go = sys.argv[1], sys.argv[2], sys.argv[3] + + +def fake_publish(config, topic, body, retain, extra=()): + with open(capture, "a") as fh: + fh.write(json.dumps({"who": "timer", "event": json.loads(body)["event"], + "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n") + + +pe._publish_once = fake_publish +real_prepare = pe.prepare_event + + +def paused_prepare(*a, **k): + open(paused, "w").close() + deadline = time.time() + 20 + while not os.path.exists(go) and time.time() < deadline: + time.sleep(0.05) + return real_prepare(*a, **k) + + +pe.prepare_event = paused_prepare +argv = sys.argv[5:] # argv[4] names the module so stop_timer recognises this process as the job's timer +job, reg = argv[argv.index("--job") + 1], argv[argv.index("--registry-dir") + 1] +with open(hb.pidfile_path(job, reg), "w") as fh: + fh.write(str(os.getpid())) +sys.exit(hb.main(argv)) +''' + + +def test_no_heartbeat_follows_the_terminal_event_even_when_one_is_in_flight(tmp_path, monkeypatch): + """Review round 2 (codex R3): a beat already running when the agent + publishes `completed` must not publish after it. The timer's first beat is + held just before it asks for a seq; the terminal publish raises the + attempt's barrier, and the beat - released only then - is refused under the + seq lock. Nothing follows `completed` on disk or on the (stubbed) wire, and + the last retained state stays `completed`.""" + from lib_py.delegation import publish_event as pe, registry + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent")) + monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs")) + monkeypatch.setenv("MAM_HEARTBEAT_DRAIN_S", "8") + reg = tmp_path / "jobs" + jid = registry.register_job("barrier", registry_dir=str(reg), auth_token="tok-" + uuid.uuid4().hex) + tok = _job(reg, jid)["dispatch_token"] + capture, paused, go = tmp_path / "wire.ndjson", tmp_path / "beat.paused", tmp_path / "beat.go" + script = tmp_path / "pausing_timer.py" + script.write_text(_PAUSING_TIMER) + env = dict(os.environ, PYTHONPATH=str(SKILLS), MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"), + DELEGATE_JOB_LOGS_DIR=str(tmp_path / "logs")) + timer = subprocess.Popen([sys.executable, str(script), str(capture), str(paused), str(go), + "lib_py.delegation.heartbeat", "--job", jid, "--registry-dir", str(reg), + "--interval", "1", "--expect-dispatch-token", tok], + cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + deadline = time.time() + 15 + while time.time() < deadline and not paused.exists(): + time.sleep(0.05) + assert paused.exists(), "the timer's first beat never reached its publish" + + def fake_publish(config, topic, body, retain, extra=()): + with open(capture, "a") as fh: + fh.write(json.dumps({"who": "terminal", "event": json.loads(body)["event"], + "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n") + + monkeypatch.setattr(pe, "_publish_once", fake_publish) + result = {} + terminal = threading.Thread(target=lambda: result.update(rc=pe.main( + ["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS", + "--expect-dispatch-token", tok, "--attempts", "1"]))) + terminal.start() + barrier = reg / f"{jid}.heartbeat.barrier" + deadline = time.time() + 10 + while time.time() < deadline and terminal.is_alive() and not barrier.exists(): + time.sleep(0.02) + go.touch() # release the in-flight beat + terminal.join(timeout=30) + assert result.get("rc") == 0 + assert timer.wait(timeout=30) is not None + wire = [json.loads(l) for l in capture.read_text().splitlines() if l.strip()] + assert wire and wire[-1]["event"] == "completed", f"something was published after completed: {wire}" + assert wire[-1]["state"] == ["completed"], wire + local = [e["event"] for e in _events(reg, jid)] + assert local and local[-1] == "completed", local + assert _job(reg, jid)["status"] == "completed" + finally: + if timer.poll() is None: + timer.kill() + + +def _signed_state(job_id, seq, token, dispatch_token, status="running", attempt_id=1): + import hashlib, hmac + from lib_py.delegation import publish_event as pe + p = pe.build_payload(job_id, seq, "state", "progress", {"dispatch_token": dispatch_token, "attempt_id": attempt_id}, + token) + p.update(status=status, last_event="progress") + p["data"].pop("hmac_sig", None) + sp = {k: v for k, v in p.items() if k != "data"} + sp["data"] = dict(p["data"]) + p["data"]["hmac_sig"] = hmac.new(token.encode(), json.dumps(sp, sort_keys=True, separators=(",", ":")).encode(), + hashlib.sha256).hexdigest() + return p + + +def test_subscriber_state_renews_liveness_only_when_it_advances(tmp_path, monkeypatch): + """Review round 2 (codex R4): the same retained snapshot re-delivered (as on + every reconnect) or an older one is still queued - a terminal status must + be recoverable - but does not renew the idle clock; only a state newer than + every event and state already accepted does.""" + from lib_py.delegation import job_subscriber as js + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs")) + token = "t" * 24 + w = js._Watcher({"j1"}, {"j1": token}, {"j1": 0}, {"j1": "d1"}, {"j1": 1}) + w.on_message(None, None, _Msg("x/events", _signed("j1", 1, "started", token))) + w.on_message(None, None, _Msg("x/events", _signed("j1", 10, "progress", token))) + for _ in range(3): # duplicate delivery of an old snapshot + w.on_message(None, None, _Msg("x/state", _signed("j1", 1, "state", token, status="running", last_event="started"))) + w.on_message(None, None, _Msg("x/state", _signed("j1", 10, "state", token, status="running", last_event="progress"))) + w.on_message(None, None, _Msg("x/state", _signed("j1", 11, "state", token, status="running", last_event="progress"))) + states = [p for _t, p in list(w.events.queue) if p["event"] == "state"] + assert [p["_renews_liveness"] for p in states] == [False, False, False, False, True] + + +def test_a_replayed_retained_state_does_not_keep_a_silent_job_alive(tmp_path, monkeypatch): + """Review round 2 (codex R4), end to end: the agent went silent after seq 10 + and the broker keeps re-delivering an old retained snapshot, as it does on + every reconnect. The subscriber must still reach its idle deadline rather + than run into the wall clock.""" + from lib_py.delegation import job_subscriber as js, registry + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent")) + monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs")) + reg = tmp_path / "jobs" + token = "tok-" + uuid.uuid4().hex + jid = registry.register_job("replay", registry_dir=str(reg), auth_token=token) + rec = _job(reg, jid) + rec["last_seq"] = 10 + (reg / f"{jid}.json").write_text(json.dumps(rec)) + state = _signed_state(jid, 1, token, rec["dispatch_token"]) + topic = rec["topic_prefix"] + "/state" + + class FakeClient: + def __init__(self): + self.on_message = self.on_connect = self.on_subscribe = self.on_disconnect = None + self.stopped = threading.Event() + + def reconnect_delay_set(self, **_kw): + pass + + def connect(self, *_a, **_kw): + return 0 + + def subscribe(self, *_a, **_kw): + return (0, 1) + + def disconnect(self): + pass + + def loop_stop(self): + self.stopped.set() + + def loop_start(self): + def run(): + self.on_connect(self, None, None, 0, None) + self.on_subscribe(self, None, 1, [1], None) + while not self.stopped.is_set(): + self.on_message(self, None, _Msg(topic, state)) + time.sleep(0.3) + threading.Thread(target=run, daemon=True).start() + + fake = FakeClient() + monkeypatch.setattr(js, "make_client", lambda *_a, **_kw: fake) + started = time.monotonic() + rc = js.main(["--job", jid, "--registry-dir", str(reg), "--idle-timeout", "2", "--ack-timeout", "0", + "--timeout", "20"]) + elapsed = time.monotonic() - started + fake.stopped.set() + assert rc == 2 + assert elapsed < 8, f"the replayed state kept the silent job alive for {elapsed:.1f}s" + + # --------------------------------------------------------------------------- # 5. wrapper acceptance criteria (broker) # --------------------------------------------------------------------------- diff --git a/tests/test_issue6_fail_closed.py b/tests/test_issue6_fail_closed.py index 2ca1fea..3d42f1e 100644 --- a/tests/test_issue6_fail_closed.py +++ b/tests/test_issue6_fail_closed.py @@ -615,7 +615,7 @@ def test_t1_14_working_agent_reobserve_completed(tmp_path, mock_herdr): "--prompt", "test T1-14", "--registry-dir", str(reg_dir), "--workdir", str(tmp_path), - "--timeout", "15", + "--timeout", "30", "--idle-timeout", "3", ], cwd=tmp_path,