fix(issue-14): keep a job open on fresh disk heartbeats after the subscriber idles out
The R1 grace after a subscriber idle timeout only continued while herdr reported `working` or the pane showed a busy token, and gave up the moment that evidence vanished. Two live jobs in the v4.4.2 review loop were closed as deadline_timeout that way: a grok review whose heartbeat gap was 147 s, and the creator running the full suite in a background shell (idle to herdr, no busy token) while publishing progress every 45 s. Both later published `completed`, which the terminal record refused. - `_recent_disk_heartbeat`: publish_event appends every event to the job's events.log before the wire publish, so a heartbeat the subscriber never saw is still on disk; treat one at most IDLE_TIMEOUT old as liveness - R1 loop: any evidence (herdr working, pane busy token, disk heartbeat) opens a new IDLE_TIMEOUT window; a job with no evidence for a whole window is closed as before, still inside the wall clock - test: a human-agent job whose subscriber idles out stays open across five disk heartbeats and picks up the out-of-band completion (rc 0) - VERSIONS.md v4.4.2 item 6 and test count updated Refs #14 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
This commit is contained in:
1 parent
7c40bf4930
commit
d5a191c47c
3 files changed
+126
-37
No files matched your search
@@ -599,6 +599,32 @@ _agent_still_working() {
|
|||||||
_pane_tail "$sess" 8 2>/dev/null | grep -Fq "esc to interrupt"
|
_pane_tail "$sess" 8 2>/dev/null | grep -Fq "esc to interrupt"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# ISSUE-14: agent-agnostic liveness. publish_event appends every event to the
|
||||||
|
# job's on-disk events.log before the wire publish, so a `progress` heartbeat
|
||||||
|
# that the subscriber never saw (it had already idled out, or the broker
|
||||||
|
# dropped it) is still visible here. An agent running its tests in a
|
||||||
|
# background shell looks idle to herdr and shows no busy token, yet keeps
|
||||||
|
# heartbeating - this is the only evidence that covers that case.
|
||||||
|
_recent_disk_heartbeat() {
|
||||||
|
local job_id="$1" window="$2" f="$REGISTRY_DIR/$job_id.events.log"
|
||||||
|
[ -f "$f" ] || return 1
|
||||||
|
"$PY" - "$f" "$window" <<'PYEOF'
|
||||||
|
import calendar, json, sys, time
|
||||||
|
path, window = sys.argv[1], float(sys.argv[2])
|
||||||
|
last = ""
|
||||||
|
with open(path) as fh:
|
||||||
|
for line in fh:
|
||||||
|
if line.strip():
|
||||||
|
last = line
|
||||||
|
try:
|
||||||
|
stamp = json.loads(last)["timestamp"]
|
||||||
|
seen = calendar.timegm(time.strptime(stamp, "%Y-%m-%dT%H:%M:%SZ"))
|
||||||
|
except Exception:
|
||||||
|
sys.exit(1)
|
||||||
|
sys.exit(0 if time.time() - seen <= window else 1)
|
||||||
|
PYEOF
|
||||||
|
}
|
||||||
|
|
||||||
_map_sub_rc() {
|
_map_sub_rc() {
|
||||||
local sub_rc="$1" job_id="$2"
|
local sub_rc="$1" job_id="$2"
|
||||||
local job_status="running"
|
local job_status="running"
|
||||||
@@ -675,42 +701,41 @@ finalize_unfinished() {
|
|||||||
local sess="${target_sess#herdr:}"
|
local sess="${target_sess#herdr:}"
|
||||||
local target_agent
|
local target_agent
|
||||||
target_agent=$(_sanitize_herdr_agent_name "$sess")
|
target_agent=$(_sanitize_herdr_agent_name "$sess")
|
||||||
local still_working=0
|
|
||||||
if [ "$sess" != "human" ] && [ -n "$sess" ]; then
|
|
||||||
_agent_still_working "$sess" "$target_agent" && still_working=1
|
|
||||||
fi
|
|
||||||
|
|
||||||
if [ "$still_working" = "1" ]; then
|
# ISSUE-14: the subscriber's idle clock only counts events it received over
|
||||||
local now elapsed remaining
|
# the wire. Keep re-observing the disk while ANY liveness evidence is at
|
||||||
now=$(date +%s)
|
# most IDLE_TIMEOUT old - herdr `working`, the pane busy token, or a fresh
|
||||||
elapsed=$(( now - ${START_TIME:-now} ))
|
# heartbeat in events.log (see _recent_disk_heartbeat). Each sighting opens
|
||||||
remaining=$(( TIMEOUT - elapsed ))
|
# a new idle window; an agent that leaves no evidence for a whole window is
|
||||||
[ $remaining -lt 0 ] && remaining=0
|
# closed, still inside the wall clock. Before this, one heartbeat gap
|
||||||
while [ $remaining -gt 0 ]; do
|
# longer than the idle timeout closed a live review as deadline_timeout
|
||||||
local disk_st
|
# and its later `completed` was refused.
|
||||||
disk_st="$(_disk_status "$job_id")"
|
local now elapsed remaining evidence_until
|
||||||
if [ "$disk_st" = "completed" ] || [ "$disk_st" = "error" ] || [ "$disk_st" = "cancelled" ]; then
|
now=$(date +%s)
|
||||||
job_status="$disk_st"
|
elapsed=$(( now - ${START_TIME:-now} ))
|
||||||
FINAL_JOB_STATUS="$disk_st"
|
remaining=$(( TIMEOUT - elapsed ))
|
||||||
[ "$disk_st" = "completed" ] && return 0
|
[ $remaining -lt 0 ] && remaining=0
|
||||||
break
|
evidence_until=$(( now + IDLE_TIMEOUT ))
|
||||||
fi
|
while [ $remaining -gt 0 ]; do
|
||||||
if ! _agent_still_working "$sess" "$target_agent"; then
|
local disk_st
|
||||||
sleep 1
|
disk_st="$(_disk_status "$job_id")"
|
||||||
disk_st="$(_disk_status "$job_id")"
|
if [ "$disk_st" = "completed" ] || [ "$disk_st" = "error" ] || [ "$disk_st" = "cancelled" ]; then
|
||||||
if [ "$disk_st" = "completed" ] || [ "$disk_st" = "error" ] || [ "$disk_st" = "cancelled" ]; then
|
job_status="$disk_st"
|
||||||
job_status="$disk_st"
|
FINAL_JOB_STATUS="$disk_st"
|
||||||
FINAL_JOB_STATUS="$disk_st"
|
[ "$disk_st" = "completed" ] && return 0
|
||||||
[ "$disk_st" = "completed" ] && return 0
|
break
|
||||||
fi
|
fi
|
||||||
break
|
if { [ "$sess" != "human" ] && [ -n "$sess" ] && _agent_still_working "$sess" "$target_agent"; } \
|
||||||
fi
|
|| _recent_disk_heartbeat "$job_id" "$IDLE_TIMEOUT"; then
|
||||||
local wait_interval=1
|
evidence_until=$(( $(date +%s) + IDLE_TIMEOUT ))
|
||||||
[ $remaining -lt $wait_interval ] && wait_interval=$remaining
|
elif [ "$(date +%s)" -ge "$evidence_until" ]; then
|
||||||
sleep "$wait_interval"
|
break
|
||||||
remaining=$(( remaining - wait_interval ))
|
fi
|
||||||
done
|
local wait_interval=1
|
||||||
fi
|
[ $remaining -lt $wait_interval ] && wait_interval=$remaining
|
||||||
|
sleep "$wait_interval"
|
||||||
|
remaining=$(( remaining - wait_interval ))
|
||||||
|
done
|
||||||
fi
|
fi
|
||||||
|
|
||||||
local close_req="deadline_timeout"
|
local close_req="deadline_timeout"
|
||||||
|
|||||||
+2
-2
@@ -47,11 +47,11 @@
|
|||||||
3. **쉼 `ls` 의 "no server running" 번역 (Issue #9)**: 서버가 내려가 있으면 쉼의 `ls` 가 `no server running (herdr session <name>)` 으로 보고하고, `reconcile.sh` 는 이를 빈 서버로 판정해 `herdr_confirmed` 를 유지한다(파싱 실패로 모든 서버의 drift 감지가 멈추는 회귀 방지). 번역은 `agent list` 가 **실패**하고 그 오류 봉투의 `error.code` 가 정확히 `server_not_running` 일 때만 적용한다. 성공한 목록에 같은 문자열이 들어 있거나(예: 에이전트 이름 `server_not_running-codex`) 다른 실패인 경우는 번역하지 않으므로, 살아 있는 서버가 빈 서버로 오판되어 drift A 로 전 세션이 종료 처리되지 않고 그 밖의 실패는 종전대로 unknown(`herdr_confirmed=False`)으로 남는다.
|
3. **쉼 `ls` 의 "no server running" 번역 (Issue #9)**: 서버가 내려가 있으면 쉼의 `ls` 가 `no server running (herdr session <name>)` 으로 보고하고, `reconcile.sh` 는 이를 빈 서버로 판정해 `herdr_confirmed` 를 유지한다(파싱 실패로 모든 서버의 drift 감지가 멈추는 회귀 방지). 번역은 `agent list` 가 **실패**하고 그 오류 봉투의 `error.code` 가 정확히 `server_not_running` 일 때만 적용한다. 성공한 목록에 같은 문자열이 들어 있거나(예: 에이전트 이름 `server_not_running-codex`) 다른 실패인 경우는 번역하지 않으므로, 살아 있는 서버가 빈 서버로 오판되어 drift A 로 전 세션이 종료 처리되지 않고 그 밖의 실패는 종전대로 unknown(`herdr_confirmed=False`)으로 남는다.
|
||||||
4. **accepted 제출의 전달 확인 `SKS_DELIVERY_GRACE_S` (Issue #14, `send_keys_safe`)**: `agent prompt --wait` 결과가 `timeout`/`agent_prompt_stalled` 이면 herdr 가 입력을 이미 수락한 것으로 보고, `SKS_DELIVERY_GRACE_S`(기본 15초) 동안 `agent_status` working, 제출 전 화면에 없던 busy 토큰(`esc to interrupt`), 제출 전 스냅샷 대비 pane 변화 중 하나를 전달 확인 신호로 인정한다. 신호가 없으면 종전대로 rc 4 로 fail-closed 하고, 그 밖의 오류(`transport_error` 등)는 accepted 로 취급하지 않는다. 단일 주입 계약(재전송 없음)은 유지된다.
|
4. **accepted 제출의 전달 확인 `SKS_DELIVERY_GRACE_S` (Issue #14, `send_keys_safe`)**: `agent prompt --wait` 결과가 `timeout`/`agent_prompt_stalled` 이면 herdr 가 입력을 이미 수락한 것으로 보고, `SKS_DELIVERY_GRACE_S`(기본 15초) 동안 `agent_status` working, 제출 전 화면에 없던 busy 토큰(`esc to interrupt`), 제출 전 스냅샷 대비 pane 변화 중 하나를 전달 확인 신호로 인정한다. 신호가 없으면 종전대로 rc 4 로 fail-closed 하고, 그 밖의 오류(`transport_error` 등)는 accepted 로 취급하지 않는다. 단일 주입 계약(재전송 없음)은 유지된다.
|
||||||
5. **래퍼 레지스트리 ack 유예 `MAM_INJECT_ACK_GRACE_S` (Issue #14, `multi-agent-mux-delegate-job`)**: `send_keys_safe` 가 전달을 확인하지 못해도 잡을 `agent_blocked` 로 닫기 전에 `MAM_INJECT_ACK_GRACE_S`(기본 10초) 동안 레지스트리를 관측하고, `pending` 이 아닌 상태(에이전트의 `started` 등)가 보이면 전달된 것으로 보고 진행한다(`_inject_acknowledged`).
|
5. **래퍼 레지스트리 ack 유예 `MAM_INJECT_ACK_GRACE_S` (Issue #14, `multi-agent-mux-delegate-job`)**: `send_keys_safe` 가 전달을 확인하지 못해도 잡을 `agent_blocked` 로 닫기 전에 `MAM_INJECT_ACK_GRACE_S`(기본 10초) 동안 레지스트리를 관측하고, `pending` 이 아닌 상태(에이전트의 `started` 등)가 보이면 전달된 것으로 보고 진행한다(`_inject_acknowledged`).
|
||||||
6. **R1 유예의 pane busy 토큰 인정 (Issue #14, `_agent_still_working`)**: 구독자 idle 타임아웃 후 R1 유예가 herdr `working` 외에 pane 최근 8줄의 `esc to interrupt` 토큰도 작업 중 근거로 받아들인다. idle 타임아웃보다 긴 codex 리뷰가 턴 도중 `deadline_timeout` 으로 닫히고 이후 `completed` 가 거부되던 문제를 방지.
|
6. **R1 유예의 pane busy 토큰·디스크 heartbeat 인정 (Issue #14, `_agent_still_working`, `_recent_disk_heartbeat`)**: 구독자 idle 타임아웃 후 R1 유예가 herdr `working` 외에 pane 최근 8줄의 `esc to interrupt` 토큰, 그리고 잡 `events.log` 의 최근 이벤트(`publish_event` 가 와이어 발행 전에 로컬에 기록하는 `progress` heartbeat)도 작업 중 근거로 받아들인다. 근거가 관측될 때마다 `IDLE_TIMEOUT` 만큼의 새 창이 열리고, 한 창 내내 아무 근거도 없으면 종전대로 닫는다(벽시계 `TIMEOUT` 내). idle 타임아웃보다 긴 codex 리뷰, heartbeat 간격이 한 번 idle 창을 넘긴 리뷰, 백그라운드 셸로 테스트를 돌려 herdr 에 idle 로 보이는 creator 가 턴 도중 `deadline_timeout` 으로 닫히고 이후 `completed` 가 거부되던 문제를 방지.
|
||||||
|
|
||||||
#### ✅ 품질 검증
|
#### ✅ 품질 검증
|
||||||
- **T1-7 트리거 변경** (`tests/test_issue6_fail_closed.py`): 주입 실패 유발값을 `prompt_wait_result=timeout` → `transport_error` 로 변경(Issue #14 이후 `timeout` 은 accepted 제출이므로 더 이상 확정 실패가 아님)하고 `MAM_INJECT_ACK_GRACE_S=1` 로 대기를 단축.
|
- **T1-7 트리거 변경** (`tests/test_issue6_fail_closed.py`): 주입 실패 유발값을 `prompt_wait_result=timeout` → `transport_error` 로 변경(Issue #14 이후 `timeout` 은 accepted 제출이므로 더 이상 확정 실패가 아님)하고 `MAM_INJECT_ACK_GRACE_S=1` 로 대기를 단축.
|
||||||
- **신규 테스트 16개**: `tests/test_issue9_server_env.py` 8개(마커 제거, `MAM_NO_AUTOSTART` 부트스트랩 생략·호출 전달, status 조회 무기동, 정적 opt-out, `ls` 번역(최소·herdr 0.9 실제 오류 봉투), 성공 목록 속 `server_not_running` 문자열 비오판, 그 밖의 실패 unknown 유지, reconcile drift A 비발생), `tests/test_issue14_delivery_ack.py` 8개(busy 토큰·pane 변화 확인, 무신호 fail-closed, 선행 busy 토큰 배제, `transport_error` 비수용, mock herdr timeout 확인, 래퍼 레지스트리 ack, idle 타임아웃 pane busy 유예).
|
- **신규 테스트 17개**: `tests/test_issue9_server_env.py` 8개(마커 제거, `MAM_NO_AUTOSTART` 부트스트랩 생략·호출 전달, status 조회 무기동, 정적 opt-out, `ls` 번역(최소·herdr 0.9 실제 오류 봉투), 성공 목록 속 `server_not_running` 문자열 비오판, 그 밖의 실패 unknown 유지, reconcile drift A 비발생), `tests/test_issue14_delivery_ack.py` 9개(busy 토큰·pane 변화 확인, 무신호 fail-closed, 선행 busy 토큰 배제, `transport_error` 비수용, mock herdr timeout 확인, 래퍼 레지스트리 ack, idle 타임아웃 pane busy 유예, idle 타임아웃 디스크 heartbeat 유예).
|
||||||
- **테스트 기대값 정정**: `tests/test_a4_adapter_contract.py` 가 `4d444ce` 의 agy `exit_key=/exit` 변경을 반영하도록 수정(그동안 전체 스위트에서 1건 실패하던 문제 해소).
|
- **테스트 기대값 정정**: `tests/test_a4_adapter_contract.py` 가 `4d444ce` 의 agy `exit_key=/exit` 변경을 반영하도록 수정(그동안 전체 스위트에서 1건 실패하던 문제 해소).
|
||||||
- **실측 (커밋 `497633f` 기록)**: 새로 부트스트랩한 서버에 마커가 없고 claude pane 의 transcript 저장이 복구됨, codex 잡 `d04ebcb5` 의 `started`/`completed` 가 `agent_blocked` 없이 수락됨.
|
- **실측 (커밋 `497633f` 기록)**: 새로 부트스트랩한 서버에 마커가 없고 claude pane 의 transcript 저장이 복구됨, codex 잡 `d04ebcb5` 의 `started`/`completed` 가 `agent_blocked` 없이 수락됨.
|
||||||
|
|
||||||
|
|||||||
@@ -302,3 +302,67 @@ def test_wrapper_idle_timeout_grace_accepts_pane_busy_token(tmp_path, mock_herdr
|
|||||||
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
||||||
assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
|
assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
|
||||||
assert proc.returncode == 0, f"expected rc=0 after the working grace, got {proc.returncode}\n{stdout}\n{stderr}"
|
assert proc.returncode == 0, f"expected rc=0 after the working grace, got {proc.returncode}\n{stdout}\n{stderr}"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# wrapper: idle-timeout grace (R1) must honour on-disk progress heartbeats
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
|
||||||
|
|
||||||
|
|
||||||
|
def test_wrapper_idle_timeout_grace_accepts_disk_heartbeat(tmp_path):
|
||||||
|
"""No herdr evidence at all (a human agent, or an agent running its tests
|
||||||
|
in a background shell): once the subscriber has idled out, the wrapper
|
||||||
|
must keep the job open as long as fresh `progress` heartbeats keep landing
|
||||||
|
in events.log, and pick up the eventual completion. Each heartbeat here
|
||||||
|
arrives after the previous idle window would have expired."""
|
||||||
|
reg_dir = tmp_path / "jobs"
|
||||||
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
env = _base_env(tmp_path)
|
||||||
|
proc = subprocess.Popen(
|
||||||
|
[
|
||||||
|
"bash", str(WRAPPER), "submit",
|
||||||
|
"--agent", "human",
|
||||||
|
"--prompt", "issue 14 disk heartbeat",
|
||||||
|
"--registry-dir", str(reg_dir),
|
||||||
|
"--workdir", str(tmp_path),
|
||||||
|
"--timeout", "60",
|
||||||
|
"--idle-timeout", "6",
|
||||||
|
],
|
||||||
|
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
jid = ""
|
||||||
|
for _ in range(100):
|
||||||
|
job_files = list(reg_dir.glob("*.json"))
|
||||||
|
if job_files:
|
||||||
|
jid = job_files[0].stem
|
||||||
|
break
|
||||||
|
time.sleep(0.1)
|
||||||
|
assert jid, "job was not registered"
|
||||||
|
|
||||||
|
def heartbeat(detail: str):
|
||||||
|
subprocess.run(
|
||||||
|
[sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir),
|
||||||
|
"--job", jid, "--event", "progress", "--detail", detail],
|
||||||
|
capture_output=True, text=True, cwd=str(tmp_path), env=env,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Subscriber idles out at ~6 s (plus broker connect); heartbeats then keep
|
||||||
|
# arriving 2 s apart, well inside the 6 s window each one re-opens (the
|
||||||
|
# wrapper's poll iteration costs ~2 s of herdr RPCs, so the ratio matters).
|
||||||
|
time.sleep(9)
|
||||||
|
for i in range(5):
|
||||||
|
heartbeat(f"still working {i}")
|
||||||
|
time.sleep(2)
|
||||||
|
done = subprocess.run(
|
||||||
|
[sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
|
||||||
|
"update", "--job", jid, "--status", "completed"],
|
||||||
|
capture_output=True, text=True, cwd=str(tmp_path), env=env,
|
||||||
|
)
|
||||||
|
assert done.returncode == 0, done.stderr
|
||||||
|
|
||||||
|
stdout, stderr = proc.communicate(timeout=90)
|
||||||
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
||||||
|
assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
|
||||||
|
assert proc.returncode == 0, f"expected rc=0 while heartbeats kept landing, got {proc.returncode}\n{stdout}\n{stderr}"
|
||||||
Reference in new issue
Block a user