From ab842f1ee021f66e494827c286ad009271b33a46 Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 13 Sep 2026 23:43:48 +0900 Subject: [PATCH] fix(issue-18): drain the previous heartbeat timer before handing over its pidfile Review round 3 (planner-reviewer-codex-01, NOT PASS R3 at 7e1d3f0): start_timer stopped a running timer with a bare TERM and overwrote the only pidfile. A beat the old timer had already admitted - seq taken, recorded on disk, held at the wire - stayed alive but undiscoverable, so a later terminal publish drained only the replacement and the old beat could still put a retained `running` on the broker after `completed`. - heartbeat.start_timer: stop_timer(drain=MAM_HEARTBEAT_DRAIN_S). The old timer is drained or killed, and confirmed gone, before the pidfile names the new one. A terminal publish racing the replacement either drains the same old pid or meets a new timer whose beats the barrier refuses. - test: an old timer held at the wire after admission is replaced, then the job completes; the old timer is gone before the handover and nothing it held follows the terminal (fails on 7e1d3f0). - docs: MESSAGING 4.3 no longer says any state resets the idle clock; MESSAGING 2.5 and VERSIONS state the replacement rule; test count 26. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe --- .agents/skills/lib_py/delegation/heartbeat.py | 6 +- MESSAGING.md | 4 +- VERSIONS.md | 4 +- tests/test_issue18_liveness_contract.py | 89 +++++++++++++++++++ 4 files changed, 98 insertions(+), 5 deletions(-) diff --git a/.agents/skills/lib_py/delegation/heartbeat.py b/.agents/skills/lib_py/delegation/heartbeat.py index 338cf1e..c371cb3 100644 --- a/.agents/skills/lib_py/delegation/heartbeat.py +++ b/.agents/skills/lib_py/delegation/heartbeat.py @@ -258,10 +258,14 @@ def start_timer(job_id: str, registry_dir: str, interval: int, 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. + It is drained as at a terminal event - and killed if it does not exit - + before its pidfile is handed over (review round 3, codex R3): a beat it had + already admitted would otherwise still reach the wire after a terminal + publish that can find only the new timer through the pidfile. ``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) + stop_timer(job_id, registry_dir, drain=drain_seconds()) if watch_pid is None: watch_pid = nearest_agent_cli_ancestor() cmd: List[str] = [python or sys.executable, "-m", "lib_py.delegation.heartbeat", diff --git a/MESSAGING.md b/MESSAGING.md index e81367a..7f931bb 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. 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` | +| **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; replacing the timer (another `started --heartbeat` for the job) drains or kills the previous one the same way before it hands over the pidfile. 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` | @@ -340,7 +340,7 @@ The subscriber acts as the central execution watchdog: 1. **Wall-Clock Timeout**: Calculated relative to absolute startup time (`wall_deadline = start + wall_timeout`). It acts as a hard budget limit, guarding against an agent hanging indefinitely. 2. **Activity Idle Timeout**: Measured as the difference between the current monotonic time and the last packet arrival time (`idle_left = idle_timeout - (now - last_event)`). If the agent fails to publish any event (heartbeats and retained state included) for the duration of the idle window - K x N under the Issue #18 contract - the subscriber aborts and exits with code 2. * **Idle budget before the first event (Issue #18)**: a standalone subscriber allows `ack_timeout_sec + idle_timeout_sec` before the first accepted event (the injection has to produce `started`). The delegate wrapper passes `--ack-timeout ` instead: it judges delivery itself with `_wait_started` (T from the injection, which happens after the subscriber is up) and reaps the subscriber when that fails, so the subscriber's own idle clock only starts with the first event. -* **Retained state and seq gaps (Issue #18)**: the subscriber also subscribes to `/state`; a retained terminal state resolves the job at (re)connect, any state resets the idle clock, and a seq jump on either topic is written to the audit log as `seq_gap`. With `--remote` (or `transport: remote`) accepted `started`/terminal events are mirrored into the local job record. +* **Retained state and seq gaps (Issue #18)**: the subscriber also subscribes to `/state`; a retained terminal state resolves the job at (re)connect, a state renews the idle clock only when its seq is newer than every event and state already accepted (a re-delivered or older snapshot does not), and a seq jump on either topic is written to the audit log as `seq_gap`. With `--remote` (or `transport: remote`) accepted `started`/terminal events are mirrored into the local job record. --- diff --git a/VERSIONS.md b/VERSIONS.md index 6414876..c175523 100644 --- a/VERSIONS.md +++ b/VERSIONS.md @@ -43,7 +43,7 @@ > **마이너 마일스톤 (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` 로 서명·발행한다. 터미널 이벤트는 먼저 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)이 명시된다. +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). 같은 잡에 `started --heartbeat` 를 다시 보내 타이머를 교체할 때도 이전 타이머를 같은 방식으로 drain·강제 종료한 뒤에야 pidfile 을 넘기므로, 이미 admission 을 통과해 와이어 직전에 있던 beat 가 터미널 뒤에 나가지 않는다(리뷰 3회차 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 토픽도 구독해 (재)접속 즉시 터미널 상태를 복구하고, 이벤트, 그리고 지금까지 받은 어떤 이벤트·state 보다 seq 가 앞선 state 로만 idle 창을 리셋하며(재접속 때 다시 오는 스냅샷이나 더 오래된 state 는 터미널 복구에만 쓰고 생존으로 치지 않는다 — 리뷰 2회차 codex R4), seq 점프를 감사 로그 `seq_gap` 으로 남기고, 재접속을 `reconnected` 로 기록한다. `--no-state` 로 끌 수 있다. @@ -51,7 +51,7 @@ 6. **문서**: MESSAGING.md §2.5 계약 표·원격 전송, delegate-job SKILL.md(옵션·종료 코드·프로토콜), registry.md(계약 필드), loop SKILL.md(환경변수·원격 행). #### ✅ 품질 검증 -- **신규 테스트 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건). +- **신규 테스트 26개** (`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건); 3회차: 타이머 교체 시 이전 타이머가 터미널 뒤에 발행하지 않음. - **기존 테스트 조정** (`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 59f6d5a..62d8e26 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/tests/test_issue18_liveness_contract.py @@ -671,6 +671,95 @@ def test_no_heartbeat_follows_the_terminal_event_even_when_one_is_in_flight(tmp_ timer.kill() +_WIRE_HELD_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 held_publish(config, topic, body, retain, extra=()): + # the beat was admitted and recorded on disk already: hold it at the wire + open(paused, "w").close() + deadline = time.time() + 20 + while not os.path.exists(go) and time.time() < deadline: + time.sleep(0.05) + with open(capture, "a") as fh: + fh.write(json.dumps({"who": "old-timer", "event": json.loads(body)["event"], + "state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n") + + +pe._publish_once = held_publish +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_replacing_the_timer_drains_the_old_one_before_the_terminal(tmp_path, monkeypatch): + """Review round 3 (codex R3): a timer replaced by a second + `started --heartbeat` must be gone before it loses the pidfile. Its beat - + already admitted and on disk, held at the wire boundary - may otherwise be + published after `completed`, because the terminal publish can only find + and drain the new timer. After the fix the old timer is dead by the time + start_timer returns and nothing it held follows the terminal.""" + from lib_py.delegation import heartbeat, 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")) + reg = tmp_path / "jobs" + jid = registry.register_job("replace", 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" + + def fake_publish(config, topic, body, retain, extra=()): + with open(capture, "a") as fh: + fh.write(json.dumps({"who": "parent", "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) + assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "started", + "--expect-dispatch-token", tok, "--attempts", "1"]) == 0 + script = tmp_path / "wire_held_timer.py" + script.write_text(_WIRE_HELD_TIMER) + env = dict(os.environ, PYTHONPATH=str(SKILLS)) + old = 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) + new_pid = None + try: + deadline = time.time() + 15 + while time.time() < deadline and not paused.exists(): + time.sleep(0.05) + assert paused.exists(), "the old timer's beat never reached the wire boundary" + + new_pid = heartbeat.start_timer(jid, str(reg), 60, dispatch_token=tok, python=sys.executable, watch_pid=0) + assert old.poll() is not None, "the replaced timer must be gone before its pidfile is handed over" + assert _timer_pid(reg, jid) == new_pid + + assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS", + "--expect-dispatch-token", tok, "--attempts", "1"]) == 0 + go.touch() # would release the held beat if its timer were still alive + time.sleep(0.5) + wire = [json.loads(l) for l in capture.read_text().splitlines() if l.strip()] + assert all(w["who"] != "old-timer" for w in wire), f"the replaced timer published: {wire}" + assert wire[-1]["event"] == "completed" and wire[-1]["state"] == ["completed"], wire + assert [e["event"] for e in _events(reg, jid)][-1] == "completed" + deadline = time.time() + 5 + while time.time() < deadline and _alive(new_pid): + time.sleep(0.1) + assert not _alive(new_pid), "the terminal publish stops the replacement timer" + finally: + if old.poll() is None: + old.kill() + if new_pid and _alive(new_pid): + os.kill(new_pid, signal.SIGKILL) + + 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