diff --git a/.agents/skills/lib_py/delegation/heartbeat.py b/.agents/skills/lib_py/delegation/heartbeat.py index b4be8ae..11a8371 100644 --- a/.agents/skills/lib_py/delegation/heartbeat.py +++ b/.agents/skills/lib_py/delegation/heartbeat.py @@ -184,7 +184,7 @@ def lifecycle_lock_path(job_id: str, registry_dir: str) -> str: @contextlib.contextmanager -def lifecycle_lock(job_id: str, registry_dir: str): +def lifecycle_lock(job_id: str, registry_dir: str, blocking: bool = True): """The per-job timer ownership lock (review round 4, codex R3). Observing/draining the previous owner, spawning the replacement and @@ -195,12 +195,17 @@ def lifecycle_lock(job_id: str, registry_dir: str): not arm at all. This is a separate flock from the registry/seq lock on purpose: an in-flight beat needs that lock to take its seq, and the drain waits for exactly those beats, so holding the seq lock across the drain - would deadlock.""" + would deadlock. With ``blocking=False`` it yields False instead of waiting + when the lock is busy - an exiting timer uses that (review round 5, codex F1).""" os.makedirs(registry_dir, exist_ok=True) with open(lifecycle_lock_path(job_id, registry_dir), "a+") as fh: - fcntl.flock(fh.fileno(), fcntl.LOCK_EX) try: - yield + fcntl.flock(fh.fileno(), fcntl.LOCK_EX if blocking else fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + yield False # busy and non-blocking: the caller must not touch owned state + return + try: + yield True finally: fcntl.flock(fh.fileno(), fcntl.LOCK_UN) @@ -335,8 +340,28 @@ def _spawn_timer_locked(job_id, registry_dir, interval, dispatch_token, attempt_ with open(log_path, "ab") as log: proc = subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, start_new_session=True, env=env, cwd=os.getcwd()) - with open(pidfile_path(job_id, registry_dir), "w") as fh: - fh.write(str(proc.pid)) + pidfile = pidfile_path(job_id, registry_dir) + try: + with open(pidfile, "w") as fh: + fh.write(str(proc.pid)) + except BaseException: + # Review round 5 (codex F2): a timer whose pid could not be published + # would be an owner nobody can find or stop. End and reap it here, while + # the lifecycle lock is still held, and drop only a partial pidfile of + # this attempted owner (the previous owner's was removed by the stop). + try: + proc.kill() + proc.wait(timeout=5) + except Exception: + pass + try: + with open(pidfile) as fh: + partial = fh.read().strip() + if not partial or str(proc.pid).startswith(partial): + os.unlink(pidfile) + except (OSError, ValueError): + pass + raise 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 @@ -439,13 +464,18 @@ def run(job_id: str, registry_dir: str, interval: int, dispatch_token=None, for t in in_flight: t.join(timeout=15) # Under the lifecycle lock: a replacement may be writing its own pid to - # this file right now, and that owner's pidfile must not be deleted. + # this file right now, and that owner's pidfile must not be deleted. Never + # *wait* for the lock here (review round 5, codex F1): a stopper holds it + # while it waits for this very exit, so waiting would only end in SIGKILL at + # its drain deadline. When the lock is busy its holder is an ownership + # operation, and that operation removes the pidfile itself. try: - with lifecycle_lock(job_id, registry_dir): - with open(own_pidfile) as fh: - mine = fh.read().strip() == str(os.getpid()) - if mine: - os.unlink(own_pidfile) + with lifecycle_lock(job_id, registry_dir, blocking=False) as owned: + if owned: + with open(own_pidfile) as fh: + mine = fh.read().strip() == str(os.getpid()) + if mine: + os.unlink(own_pidfile) except (OSError, ValueError): pass return 0 diff --git a/MESSAGING.md b/MESSAGING.md index 7f931bb..a2c53b0 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; 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` | +| **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. Starts, replacements and stops of a job's timer are serialized by a per-job lifecycle lock (`/.heartbeat.lock`, separate from the seq lock): a start that finds the attempt ended arms nothing, an exiting timer never waits for that lock, and a timer whose pid could not be recorded is killed at once. 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` | diff --git a/VERSIONS.md b/VERSIONS.md index c175523..42e8842 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). 같은 잡에 `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)이 명시된다. +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). 타이머의 생성·교체·정지는 잡별 lifecycle lock(`/.heartbeat.lock`, seq lock 과 별개)으로 직렬화되어 동시 `started --heartbeat` 도 추적되는 소유자 하나만 남기고, 끝난 attempt 에는 타이머를 arm 하지 않는다. 종료하는 타이머는 그 lock 을 기다리지 않으며(바쁘면 소유 연산이 pidfile 을 치운다), spawn 뒤 pidfile 기록이 실패하면 새 타이머를 곧바로 종료·회수한다(리뷰 4·5회차 codex). 타이머는 레코드 터미널·재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(환경변수·원격 행). #### ✅ 품질 검증 -- **신규 테스트 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회차: 타이머 교체 시 이전 타이머가 터미널 뒤에 발행하지 않음. +- **신규 테스트 31개** (`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회차: 타이머 교체 시 이전 타이머가 터미널 뒤에 발행하지 않음; 4·5회차: 동시 `start_timer` 는 추적되는 소유자 하나만 남김, barrier 뒤 start 는 arm 안 함, idle 타이머의 정상 종료(drain 기한 전·SIGKILL 없음), spawn 전·후 실패 시 추적되지 않는 타이머 없음. - **기존 테스트 조정** (`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 a0f68b4..0ef2738 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/tests/test_issue18_liveness_contract.py @@ -826,6 +826,119 @@ def test_replacing_the_timer_drains_the_old_one_before_the_terminal(tmp_path, mo os.kill(new_pid, signal.SIGKILL) +def test_an_idle_timer_stops_gracefully_well_before_the_drain_deadline(tmp_path, monkeypatch, caplog): + """Review round 5 (codex F1): the stopper holds the lifecycle lock while it + waits for the timer to exit, so the exiting timer must not wait for that + lock. An idle timer (nothing in flight) ends on TERM at once - well before + a generous drain deadline and without the SIGKILL fallback.""" + import logging + from lib_py.delegation import heartbeat, 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("idle-stop", registry_dir=str(reg)) + tok = _job(reg, jid)["dispatch_token"] + pid = heartbeat.start_timer(jid, str(reg), 60, dispatch_token=tok, python=sys.executable, watch_pid=0) + try: + assert pid and _alive(pid) + time.sleep(2.0) # past interpreter start-up: TERM handler installed, nothing in flight + caplog.set_level(logging.WARNING, logger="delegate_job.heartbeat") + t0 = time.monotonic() + assert heartbeat.stop_timer(jid, str(reg), drain=10) is True + elapsed = time.monotonic() - t0 + assert elapsed < 3.0, f"an idle timer took {elapsed:.1f}s to stop (drain deadline 10 s)" + assert "did not drain" not in caplog.text, caplog.text + assert not _alive(pid) + assert not (reg / f"{jid}.heartbeat.pid").exists() + finally: + if _alive(pid): + os.kill(pid, signal.SIGKILL) + + +def test_a_timer_whose_pid_cannot_be_recorded_is_killed_at_once(tmp_path, monkeypatch): + """Review round 5 (codex F2): if publishing the new pid fails after the + timer was spawned, that exact process is killed and reaped before the + lifecycle lock is released - no untracked owner survives, no partial + pidfile is left, and the lock is free for the next owner.""" + import builtins + import errno + import fcntl + from lib_py.delegation import heartbeat, 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("pidfile-fault", registry_dir=str(reg)) + pidfile = str(reg / f"{jid}.heartbeat.pid") + spawned = [] + real_popen, real_open = heartbeat.subprocess.Popen, builtins.open + + def recording_popen(*a, **k): + proc = real_popen(*a, **k) + args = a[0] if a else k.get("args", []) + if any("lib_py.delegation.heartbeat" in str(x) for x in args): + spawned.append(proc) + return proc + + def failing_open(file, mode="r", *a, **k): + if str(file) == pidfile and "w" in mode: + raise OSError(errno.ENOSPC, "No space left on device", pidfile) + return real_open(file, mode, *a, **k) + + monkeypatch.setattr(heartbeat.subprocess, "Popen", recording_popen) + monkeypatch.setattr(builtins, "open", failing_open) + try: + with pytest.raises(OSError) as exc: + heartbeat.start_timer(jid, str(reg), 60, python=sys.executable, watch_pid=0) + finally: + monkeypatch.setattr(builtins, "open", real_open) + try: + assert exc.value.errno == errno.ENOSPC + assert len(spawned) == 1, spawned + proc = spawned[0] + assert proc.poll() is not None, "the spawned timer must not survive untracked" + assert proc.returncode == -signal.SIGKILL + assert not os.path.exists(pidfile) + with open(reg / f"{jid}.heartbeat.lock", "a+") as fh: # the lifecycle lock is free again + fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + finally: + for proc in spawned: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + + +def test_a_timer_that_cannot_be_spawned_leaves_nothing_behind(tmp_path, monkeypatch): + """Review round 5 (codex F2), the other half: when the spawn itself fails + the error propagates, no pidfile is written, and the lifecycle lock is + released and serves the next owner at once.""" + import fcntl + from lib_py.delegation import heartbeat, 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("spawn-fault", registry_dir=str(reg)) + with pytest.raises(OSError): + heartbeat.start_timer(jid, str(reg), 60, python=str(tmp_path / "no-such-python"), watch_pid=0) + assert not (reg / f"{jid}.heartbeat.pid").exists() + with open(reg / f"{jid}.heartbeat.lock", "a+") as fh: + fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + pid = heartbeat.start_timer(jid, str(reg), 60, python=sys.executable, watch_pid=0) + try: + assert pid and _timer_pid(reg, jid) == pid and _alive(pid) + finally: + heartbeat.stop_timer(jid, str(reg), drain=5) + if _alive(pid): + os.kill(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