diff --git a/.agents/MULTI_AGENT_RULES.md b/.agents/MULTI_AGENT_RULES.md index 797a197..19527d5 100644 --- a/.agents/MULTI_AGENT_RULES.md +++ b/.agents/MULTI_AGENT_RULES.md @@ -168,7 +168,8 @@ To ensure that agents running in TMUX environments do not lose debug logs or pre - **Identity Verification**: The guard validates process liveness via `pid` + `lstart` to prevent livelocks on PID rollover. ### ⏱️ Timeout Configuration & Alignment Rules -- **Job Execution Limits (`timeout_sec` & `idle_timeout_sec`)**: Each job independently manages its overall execution timeout (`timeout_sec`, default 3600s) and idle timeout without receiving messages (`idle_timeout_sec`, default 120s). +- **Job Execution Limits (`timeout_sec` & `idle_timeout_sec`)**: Each job independently manages its overall execution timeout (`timeout_sec`, default 3600s) and idle timeout without receiving messages (`idle_timeout_sec`). Since v4.5.0 (Issue #18) the idle timeout is the liveness contract's K x N: the agent heartbeats every `heartbeat_sec` (N, default 30 — automatic once it publishes `started --heartbeat N`), and `heartbeat_misses` (K, default 4) missed heartbeats close the job as `deadline_timeout` (default 120 s, `--idle-timeout` overrides). `started` must arrive within `ack_timeout_sec` (T, default 60) of the injection or the job fails as `agent_blocked` without re-injection. herdr status and pane text are never used for these decisions. +- **Heartbeat timer and agent life**: the automatic timer is bound to the agent process that ran the start command (the nearest agent-CLI ancestor, or `--watch-pid`) and stops within one period once that process is gone; a crashed agent is therefore closed after K x N like any silent one, while an agent that is alive but stuck keeps heartbeating (process liveness, not work liveness). - **Monitor Idle Waiting (`SUB_IDLE_TIMEOUT`)**: The idle timeout for the monitor script (`reconcile.sh`), `SUB_IDLE_TIMEOUT`, must always be set generously to `3600s` (1 hour) or more to align with the maximum job budget. This prevents the monitor from terminating early due to idle detection, which would lose control over background tasks before they finish. --- diff --git a/.agents/skills/lib_py/delegation/heartbeat.py b/.agents/skills/lib_py/delegation/heartbeat.py index 5586dfc..3988fde 100644 --- a/.agents/skills/lib_py/delegation/heartbeat.py +++ b/.agents/skills/lib_py/delegation/heartbeat.py @@ -18,6 +18,15 @@ timer stops on its own when A terminal publish also stops it directly through the pidfile (`/.heartbeat.pid`) so no heartbeat can trail a `completed`. +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 +that is an agent CLI (claude, codex, agy, ...; the agent runs the start +command from its shell tool, so that ancestor is the agent itself) - and stops +within one period once that process is gone. A crashed or killed agent +therefore stops heartbeating and is closed after K x N like any silent one; +an agent that is alive but stuck keeps heartbeating, which is the contract's +deliberate trade-off (liveness of the process, not of the work). + 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 @@ -46,6 +55,64 @@ HEARTBEAT_DETAIL = "heartbeat" MAX_IN_FLIGHT = 8 +# The agent CLIs, as lib_py.agents.base names them. Repeated here rather than +# imported: lib_py.delegation may not import a sibling lib_py package (a +# frozen loop binds delegation to a different tree, see __init__). +AGENT_CLI_NAMES = ("claude", "agy", "hermes", "grok", "opencode", "codex") + + +def _is_agent_cli(command: str) -> bool: + """Some token before the first option has an agent basename + (`claude ...`, `node .../claude ...`, `/bin/sh .../agy ...`).""" + for tok in command.split(): + if tok.startswith("-"): + break + if os.path.basename(tok) in AGENT_CLI_NAMES: + return True + return False + + +def _process_table(): + """{pid: (ppid, command)} from ps; empty when ps is unavailable.""" + try: + out = subprocess.run(["ps", "-axo", "pid=,ppid=,command="], capture_output=True, text=True).stdout + except OSError: + return {} + table = {} + for line in out.splitlines(): + parts = line.split(None, 2) + if len(parts) < 2: + continue + try: + pid, ppid = int(parts[0]), int(parts[1]) + except ValueError: + continue + table[pid] = (ppid, parts[2] if len(parts) > 2 else "") + return table + + +def nearest_agent_cli_ancestor(start_pid: Optional[int] = None) -> Optional[int]: + """PID of the closest ancestor (start_pid included) that is an agent CLI. + + The agent executes the injected start command through its shell tool, so + the chain is agent CLI -> shell -> python publish_event; the agent is the + first agent-CLI process met walking up. None when there is no such + ancestor (a test, a plain shell, a remote runner).""" + table = _process_table() + pid = start_pid if start_pid is not None else os.getpid() + seen = set() + while pid and pid > 1 and pid not in seen: + seen.add(pid) + entry = table.get(pid) + if entry is None: + return None + ppid, command = entry + if _is_agent_cli(command): + return pid + pid = ppid + return None + + def pidfile_path(job_id: str, registry_dir: str) -> str: return os.path.join(registry_dir, "{}.heartbeat.pid".format(job_id)) @@ -57,7 +124,12 @@ def _pid_alive(pid: int) -> bool: return False except PermissionError: return True - return True + try: + state = subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], + capture_output=True, text=True).stdout.strip() + except OSError: + return True + return bool(state) and not state.startswith("Z") def running_timer(job_id: str, registry_dir: str) -> Optional[int]: @@ -72,11 +144,21 @@ def running_timer(job_id: str, registry_dir: str) -> Optional[int]: return None +def _is_timer_process(pid: int, job_id: str) -> bool: + """The pid really is this job's timer (a stale pidfile after a crash may + name a recycled pid that now belongs to something else).""" + entry = _process_table().get(pid) + if entry is None: + return False + command = entry[1] + 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.""" pid = running_timer(job_id, registry_dir) path = pidfile_path(job_id, registry_dir) - if pid is None: + if pid is None or not _is_timer_process(pid, job_id): try: os.unlink(path) except OSError: @@ -95,13 +177,17 @@ def stop_timer(job_id: str, registry_dir: str) -> bool: 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: + python: Optional[str] = None, watch_pid: Optional[int] = 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. + ``watch_pid`` binds the timer to a process (default: the nearest agent-CLI + ancestor of this process, if any); 0 disables the watch explicitly. """ stop_timer(job_id, registry_dir) + if watch_pid is None: + watch_pid = nearest_agent_cli_ancestor() cmd: List[str] = [python or sys.executable, "-m", "lib_py.delegation.heartbeat", "--job", job_id, "--registry-dir", registry_dir, "--interval", str(int(interval))] @@ -109,6 +195,8 @@ def start_timer(job_id: str, registry_dir: str, interval: int, cmd += ["--expect-dispatch-token", dispatch_token] if attempt_id is not None: cmd += ["--attempt-id", str(int(attempt_id))] + if watch_pid: + cmd += ["--watch-pid", str(int(watch_pid))] 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__)))) @@ -119,11 +207,15 @@ def start_timer(job_id: str, registry_dir: str, interval: int, 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) + logger.info("heartbeat timer pid %d for job %s every %ss (watching pid %s)", + proc.pid, job_id, interval, watch_pid or "none") return proc.pid -def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id) -> Optional[str]: +def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id, + watch_pid: Optional[int] = None) -> Optional[str]: + if watch_pid and not _pid_alive(watch_pid): + return "watched agent process {} is gone".format(watch_pid) try: rec = load_job(job_id, registry_dir) except FileNotFoundError: @@ -138,7 +230,8 @@ def _should_stop(job_id: str, registry_dir: str, dispatch_token, attempt_id) -> def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None, - attempt_id=None, max_lifetime: Optional[float] = None) -> int: + attempt_id=None, max_lifetime: Optional[float] = None, + watch_pid: Optional[int] = None) -> int: """The timer loop. Runs until a stop condition holds.""" from lib_py.delegation import publish_event # late import: the CLI imports us @@ -173,7 +266,7 @@ def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None, 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) + reason = _should_stop(job_id, registry_dir, dispatch_token, attempt_id, watch_pid) if reason: logger.info("heartbeat timer for %s stopping: %s", job_id, reason) break @@ -227,12 +320,14 @@ def main(argv=None) -> int: 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("--watch-pid", type=int, default=None, + help="stop once this process (the agent) is gone") 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) + args.attempt_id, args.max_lifetime, args.watch_pid) if __name__ == "__main__": diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 9f6966d..3c1d747 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -139,6 +139,7 @@ class _Watcher: # 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() + self.state_seq: Dict[str, int] = {} 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.""" @@ -161,11 +162,14 @@ class _Watcher: 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) + # A state message never advances the event watermark: MQTT orders + # messages per connection, not across topics, so a state that overtakes + # its own event must not make the monotonic guard drop that event. + 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) - if seq > last: - self.last_seq[jid] = seq + if seq > self.state_seq.get(jid, 0): + self.state_seq[jid] = seq mqtt_common.append_event(jid, { "event": "received", "source_event": STATE_EVENT, diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index 677891a..b049f96 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -297,6 +297,9 @@ def main(argv=None) -> int: "(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("--watch-pid", type=int, default=None, + help="bind the heartbeat timer to this process (default: the nearest " + "agent-CLI ancestor; 0 = no watch)") parser.add_argument("-v", "--verbose", action="store_true") args = parser.parse_args(argv) @@ -456,7 +459,8 @@ def main(argv=None) -> int: 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) + dispatch_token=hb_token, attempt_id=hb_attempt, + watch_pid=args.watch_pid) 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) diff --git a/.agents/skills/lib_py/delegation/remote.py b/.agents/skills/lib_py/delegation/remote.py index c14560f..ebbc352 100644 --- a/.agents/skills/lib_py/delegation/remote.py +++ b/.agents/skills/lib_py/delegation/remote.py @@ -153,7 +153,7 @@ def clear_dispatch(config, topic: str) -> None: # remote side # --------------------------------------------------------------------------- -def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> str: +def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> Optional[str]: """Create the local job record a remote agent publishes against. The record carries the orchestrator's job_id, auth_token, dispatch_token @@ -165,8 +165,9 @@ def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> str: 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")): + if old.get("dispatch_token") == d.get("dispatch_token"): + 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() @@ -206,7 +207,8 @@ def _publish(jid: str, registry_dir: str, event: str, detail: str, d: Dict[str, if d.get("attempt_id") is not None: argv += ["--attempt-id", str(d["attempt_id"])] if heartbeat: - argv += ["--heartbeat", str(heartbeat)] + # The runner publishes in-process, so the timer must die with the runner. + argv += ["--heartbeat", str(heartbeat), "--watch-pid", str(os.getpid())] return publish_event.main(argv) @@ -214,6 +216,10 @@ 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")) + 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), d, heartbeat=heartbeat) 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 b040430..f9718df 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 @@ -604,6 +604,11 @@ _wait_started() { wall=$(( ${START_TIME:-$(date +%s)} + ${TIMEOUT:-3600} )) [ "$deadline" -gt "$wall" ] && deadline=$wall local ev_log="$REGISTRY_DIR/$job_id.events.log" + # A loop job keeps one events.log across iterations: only a `started` that + # carries THIS attempt's dispatch token counts (review R1). + local started_re='"event": *"started"' + [ -n "${_disp_tok:-}" ] && started_re='"event": *"started".*"dispatch_token": *"'"$_disp_tok"'"' + local budget=$(( deadline - $(date +%s) )) while :; do st="$(_disk_status "$job_id")" if [ -n "$st" ] && [ "$st" != "pending" ]; then @@ -612,7 +617,7 @@ _wait_started() { fi # publish_event records `started` here before its wire publish and before # the status sync, so a slow broker cannot delay the acknowledgement. - if [ -f "$ev_log" ] && grep -Eq '"event": *"started"' "$ev_log" 2>/dev/null; then + if [ -f "$ev_log" ] && grep -Eq "$started_re" "$ev_log" 2>/dev/null; then echo "inject: job $job_id recorded 'started' in events.log; treating the prompt as delivered and continuing." >&2 return 0 fi @@ -620,7 +625,10 @@ _wait_started() { echo "inject: subscriber received started for job $job_id; treating the prompt as delivered and continuing." >&2 return 0 fi - [ "$(date +%s)" -ge "$deadline" ] && return 1 + if [ "$(date +%s)" -ge "$deadline" ]; then + ACK_WAITED_S="$budget" + return 1 + fi sleep 1 done } @@ -951,7 +959,7 @@ inject_agent() { local ack_s="${MAM_INJECT_ACK_GRACE_S:-$ACK_TIMEOUT}" if ! _wait_started "$job_id" "$ack_s" "${4:-}"; then trap - EXIT - echo "ERROR: 프롬프트 주입 실패 — 세션 '$sess': no 'started' within ${ack_s}s (ack timeout); not re-injecting" >&2 + echo "ERROR: 프롬프트 주입 실패 — 세션 '$sess': no 'started' within ${ACK_WAITED_S:-$ack_s}s (ack timeout ${ack_s}s, bounded by the wall clock); not re-injecting" >&2 return 1 fi @@ -976,7 +984,7 @@ remote_dispatch() { fi local ack_s="${MAM_INJECT_ACK_GRACE_S:-$ACK_TIMEOUT}" if ! _wait_started "$job_id" "$ack_s" "$sub_log"; then - echo "ERROR: remote session '$sess' published no 'started' within ${ack_s}s (ack timeout); not re-dispatching" >&2 + echo "ERROR: remote session '$sess' published no 'started' within ${ACK_WAITED_S:-$ack_s}s (ack timeout ${ack_s}s, bounded by the wall clock); not re-dispatching" >&2 return 1 fi echo "작업이 원격 세션 '$sess'에 전달되었습니다." @@ -1069,4 +1077,7 @@ main() { esac } -main "$@" +# Tests source this file to exercise single helpers (MAM_DELEGATE_JOB_NO_MAIN=1). +if [ "${MAM_DELEGATE_JOB_NO_MAIN:-0}" != "1" ]; then + main "$@" +fi diff --git a/.agents/skills/multi-agent-mux-delegate-job/registry.md b/.agents/skills/multi-agent-mux-delegate-job/registry.md index 2d23559..9a4a8c1 100644 --- a/.agents/skills/multi-agent-mux-delegate-job/registry.md +++ b/.agents/skills/multi-agent-mux-delegate-job/registry.md @@ -205,4 +205,3 @@ $PY scripts/registry.py logs --list # every job, live status | `transport` | `herdr` (pane injection) or `remote` (signed retained dispatch over the broker) | `herdr` | `/.heartbeat.pid` / `.heartbeat.log` belong to the running heartbeat timer; a terminal publish removes the pidfile. - diff --git a/MESSAGING.md b/MESSAGING.md index 8944a4a..42be17d 100644 --- a/MESSAGING.md +++ b/MESSAGING.md @@ -196,7 +196,7 @@ 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, or after `timeout_sec`. | `lib_py/delegation/heartbeat.py`, `publish_event.py --heartbeat` | +| **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` | | **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 60): 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` | diff --git a/VERSIONS.md b/VERSIONS.md index a7bd2a3..7a4fe4e 100644 --- a/VERSIONS.md +++ b/VERSIONS.md @@ -43,9 +43,9 @@ > **마이너 마일스톤 (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=60)은 종전 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`)로 타이머를 먼저 멈추고, 타이머는 레코드 터미널·재dispatch·`timeout_sec` 초과 시 스스로 끝난다. 각 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` 로 서명·발행한다. 터미널 이벤트가 pidfile(`/.heartbeat.pid`)로 타이머를 먼저 멈추고(pid 재사용 대비 cmdline 확인), 타이머는 레코드 터미널·재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`, 기본 60; 벽시계 상한) 안에 레지스트리가 `pending` 을 벗어나거나 events.log/구독자 로그에 `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 로 남는다. +3. **전달 확인 = T 내 `started` (`_wait_started`, `lib.sh: _sks_job_started`)**: 주입 후 T초(`--ack-timeout`, 기본 60; 벽시계 상한) 안에 레지스트리가 `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` 를 계약 옵션으로 전달한다. 6. **문서**: MESSAGING.md §2.5 계약 표·원격 전송, delegate-job SKILL.md(옵션·종료 코드·프로토콜), registry.md(계약 필드), loop SKILL.md(환경변수·원격 행). diff --git a/tests/fixtures/issue_5/phase1/help-publish_event.txt b/tests/fixtures/issue_5/phase1/help-publish_event.txt index 851bd0e..887806d 100644 --- a/tests/fixtures/issue_5/phase1/help-publish_event.txt +++ b/tests/fixtures/issue_5/phase1/help-publish_event.txt @@ -7,7 +7,8 @@ usage: publish_event.py [-h] (--job JOB | --pick-pending) [--side-effect KIND=REFERENCE] [--attempt-id ATTEMPT_ID] [--expect-dispatch-token EXPECT_DISPATCH_TOKEN] - [--heartbeat N] [--no-state] [-v] + [--heartbeat N] [--no-state] [--watch-pid WATCH_PID] + [-v] Publish a Job event to MQTT @@ -39,4 +40,7 @@ optional arguments: the job ends (Issue #18; not with a terminal event) --no-state do not publish the retained /state message + --watch-pid WATCH_PID + bind the heartbeat timer to this process (default: the + nearest agent-CLI ancestor; 0 = no watch) -v, --verbose diff --git a/tests/test_issue18_liveness_contract.py b/tests/test_issue18_liveness_contract.py index 8582c84..aa9a003 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/tests/test_issue18_liveness_contract.py @@ -220,6 +220,61 @@ def test_heartbeat_timer_stops_on_its_own_when_the_job_is_redispatched(tmp_path) _kill_timer(reg, jid) +def test_heartbeat_stops_when_the_watched_agent_process_dies(tmp_path): + """Review B2: the timer follows the agent's life. The 'agent' here is a + shell that runs the start command with --watch-pid $$ and then idles; + killing that shell must stop the heartbeats within about one period.""" + reg = tmp_path / "jobs" + env = _env(tmp_path, UNREACHABLE) + jid = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "register", + "--prompt", "hb"], capture_output=True, text=True, env=env, cwd=str(tmp_path)).stdout.strip() + agent = subprocess.Popen( + ["bash", "-c", f'"{sys.executable}" "{PUBLISH_PY}" --registry-dir "{reg}" --job {jid} --event started ' + f'--heartbeat 1 --attempts 1 --watch-pid $$ >/dev/null 2>&1; sleep 120'], + cwd=str(tmp_path), env=env) + try: + deadline = time.time() + 15 + while time.time() < deadline and not _timer_pid(reg, jid): + time.sleep(0.1) + pid = _timer_pid(reg, jid) + assert pid and _alive(pid) + time.sleep(2.5) + before = len([e for e in _events(reg, jid) if e["event"] == "progress"]) + assert before >= 2 + agent.kill() # the agent crashes + agent.wait(timeout=10) + deadline = time.time() + 6 + while time.time() < deadline and _alive(pid): + time.sleep(0.1) + assert not _alive(pid), "timer must stop once the watched agent is gone" + time.sleep(1.5) + after = len([e for e in _events(reg, jid) if e["event"] == "progress"]) + assert after <= before + 2, f"heartbeats kept coming after the agent died: {before} -> {after}" + assert _job(reg, jid)["status"] == "running", "the timer stops; the orchestrator's K x N closes the job" + finally: + if agent.poll() is None: + agent.kill() + _kill_timer(reg, jid) + + +def test_stop_timer_never_signals_a_recycled_pid(tmp_path, monkeypatch): + """A stale pidfile (timer crashed) may name a pid that now belongs to + something else; stop_timer must only signal a heartbeat timer of this job.""" + from lib_py.delegation import heartbeat + + reg = tmp_path / "jobs" + reg.mkdir() + bystander = subprocess.Popen(["sleep", "30"]) + try: + (reg / "job1.heartbeat.pid").write_text(str(bystander.pid)) + assert heartbeat.running_timer("job1", str(reg)) == bystander.pid + assert heartbeat.stop_timer("job1", str(reg)) is False + assert bystander.poll() is None, "the bystander must not be killed" + assert not (reg / "job1.heartbeat.pid").exists(), "the stale pidfile is removed" + finally: + bystander.kill() + + def test_heartbeat_is_refused_on_a_terminal_event(tmp_path, monkeypatch): from lib_py.delegation import publish_event as pe, registry @@ -402,6 +457,54 @@ def test_send_keys_safe_ignores_pane_change_and_busy_token(tmp_path): assert "PASTE_PATH_USED" not in res.stderr +def test_wait_started_only_accepts_this_attempts_started(tmp_path): + """Review R1: a loop job keeps one events.log across iterations. A + `started` left by the previous attempt (other dispatch token) must not + acknowledge the new injection; one with the current token must.""" + reg = tmp_path / "jobs" + reg.mkdir() + jid = "abcd1234" + (reg / f"{jid}.json").write_text(json.dumps({"job_id": jid, "status": "pending", "dispatch_token": "new-token"})) + log = reg / f"{jid}.events.log" + log.write_text(json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": "old-token"}}) + "\n") + env = _env(tmp_path, {"MAM_DELEGATE_JOB_NO_MAIN": "1"}) + + def wait_started(extra: str) -> subprocess.CompletedProcess: + # the wrapper sets -e: capture the helper's rc instead of dying on it + script = (f'source "{WRAPPER}"; set +e; REGISTRY_DIR="{reg}"; PY="{sys.executable}"; TIMEOUT=60; START_TIME=$(date +%s); ' + f'_disp_tok="new-token"; {extra} _wait_started {jid} 3; echo "rc=$?"') + return subprocess.run(["bash", "-c", script], capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=60) + + res = wait_started("") + assert "rc=1" in res.stdout, f"old attempt's started must not count\n{res.stdout}\n{res.stderr}" + with open(log, "a") as fh: + fh.write(json.dumps({"event": "started", "seq": 2, "data": {"dispatch_token": "new-token"}}) + "\n") + res = wait_started("") + assert "rc=0" in res.stdout and "recorded 'started' in events.log" in res.stderr, res.stderr + + +def test_remote_runner_ignores_a_resent_dispatch_of_a_finished_attempt(tmp_path, monkeypatch): + """Review R2: a retained dispatch delivered again (runner restart, broker + replay) must not re-run an attempt that already reached its terminal.""" + from lib_py.delegation import remote, 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" + d = {"job_id": "feed1234", "brief": "b", "dispatch_token": "tok-1", "attempt_id": 1, "auth_token": None, + "timeout_sec": 30, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5} + assert remote.mirror_job(d, str(reg), "s") == "feed1234" + assert remote.mirror_job(d, str(reg), "s") == "feed1234", "same attempt, still open: idempotent" + from lib_py.delegation import mqtt_common + mqtt_common.update_job_status("feed1234", str(reg), _source="registry_cli", status="completed") + assert remote.mirror_job(d, str(reg), "s") is None, "finished attempt: the re-sent dispatch is ignored" + assert remote.handle_dispatch(d, str(reg), "s", "echo should-not-run") == 0 + d2 = dict(d, dispatch_token="tok-2", attempt_id=2) + assert remote.mirror_job(d2, str(reg), "s") == "feed1234", "a new attempt is mirrored afresh" + assert _job(reg, "feed1234")["dispatch_token"] == "tok-2" and _job(reg, "feed1234")["status"] == "pending" + + # --------------------------------------------------------------------------- # 5. wrapper acceptance criteria (broker) # --------------------------------------------------------------------------- @@ -490,14 +593,28 @@ def test_acceptance3_lost_heartbeats_close_the_job_and_a_late_completed_is_refus ["--heartbeat", "1", "--heartbeat-misses", "3", "--ack-timeout", "40"], timeout_s=60) jid = _wait_job_id(reg) t0 = None + agent = None try: time.sleep(1) - assert _publish(reg, jid, "started", env, "--heartbeat", "1").returncode == 0 + tok = _job(reg, jid)["dispatch_token"] + # the agent: a shell that runs the injected start command (timer bound + # to it via --watch-pid $$) and then keeps "working" silently + agent = subprocess.Popen( + ["bash", "-c", f'"{sys.executable}" "{PUBLISH_PY}" --registry-dir "{reg}" --job {jid} --event started ' + f'--expect-dispatch-token {tok} --heartbeat 1 --watch-pid $$; sleep 300'], + cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + deadline = time.time() + 30 + while time.time() < deadline and not _timer_pid(reg, jid): + time.sleep(0.2) + assert _timer_pid(reg, jid), "the start command must have armed the timer" time.sleep(4) - _kill_timer(reg, jid) # the agent dies: no more heartbeats + agent.kill() # the agent dies: its timer stops, no more heartbeats + agent.wait(timeout=10) t0 = time.time() stdout, stderr = proc.communicate(timeout=120) finally: + if agent is not None and agent.poll() is None: + agent.kill() _kill_timer(reg, jid) if proc.poll() is None: proc.kill()