fix(issue-17): review R1 round 2 - an exited, unreaped wrapper is not a live owner

kill(pid, 0) succeeds for a zombie. A caller that captures the wrapper's
output and calls communicate() after TERM leaves the wrapper unreaped while
the finalizer holds its pipes, so the owner check passed until the deadline
and the job was still closed (codex, round 2; reproduced with the real
registry). _alive now asks `ps -o stat=` (POSIX, no /proc) once kill(0)
succeeds and treats state Z as gone, as tests/conftest.py::_pid_alive does;
no ps answer keeps kill(0)'s verdict. The owner is looked at every 0.5 s.

Tests: TERM + communicate() with no wait() for both graces (stand-ins) and
for the timeout grace with the real registry, old vs new; a zombie is not
alive. All four fail on c5cebe6.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-06 08:09:13 +00:00
1 parent c5cebe6c32
commit 518bedac5d
2 files changed
+90 -3

No files matched your search

+11 -3
View File
@@ -120,13 +120,21 @@ class OwnerGone(Exception):
def _alive(pid: int) -> bool: def _alive(pid: int) -> bool:
"""A running process. An exited one its caller has not reaped yet is a
zombie that still answers kill(0) (review R1, round 2): `ps -o stat=`
(POSIX, no /proc) tells it apart, as tests/conftest.py::_pid_alive does."""
try: try:
os.kill(pid, 0) os.kill(pid, 0)
except ProcessLookupError: except ProcessLookupError:
return False return False
except PermissionError: except PermissionError:
return True return True
return True try:
state = subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True,
timeout=10).stdout.strip()
except (OSError, subprocess.SubprocessError):
return True # no ps answer: kill(0) said it exists
return bool(state) and not state.startswith("Z")
def _check_owner(a: argparse.Namespace) -> None: def _check_owner(a: argparse.Namespace) -> None:
@@ -135,14 +143,14 @@ def _check_owner(a: argparse.Namespace) -> None:
def _sleep(a: argparse.Namespace, seconds: float) -> None: def _sleep(a: argparse.Namespace, seconds: float) -> None:
"""time.sleep that looks at the wrapper every 0.2 s.""" """time.sleep that looks at the wrapper every 0.5 s."""
end = time.time() + seconds end = time.time() + seconds
while True: while True:
_check_owner(a) _check_owner(a)
left = end - time.time() left = end - time.time()
if left <= 0: if left <= 0:
return return
time.sleep(min(left, 0.2)) time.sleep(min(left, 0.5))
def _tail(path: str, n: int) -> List[str]: def _tail(path: str, n: int) -> List[str]:
+79
View File
@@ -512,3 +512,82 @@ def test_u_finalize_end_leaves_without_side_effects_once_its_wrapper_is_gone(tmp
finally: finally:
del os.environ["CALL_LOG"] del os.environ["CALL_LOG"]
assert "registry.py" in log.read_text() assert "registry.py" in log.read_text()
def test_u_an_exited_unreaped_owner_is_not_alive():
"""Review R1 round 2: a zombie answers kill(0); it is not a live wrapper."""
z = subprocess.Popen(["true"])
try:
assert _until_true(lambda: subprocess.run(["ps", "-o", "stat=", "-p", str(z.pid)], capture_output=True,
text=True).stdout.strip().startswith("Z"), 10)
os.kill(z.pid, 0) # still answers ...
assert decide._alive(z.pid) is False # ... but is gone
finally:
z.wait()
assert decide._alive(z.pid) is False
assert decide._alive(os.getpid()) is True
def _terminate_unreaped(script: str, cwd: Path, env: dict, ready: Path):
"""TERM to the wrapper, then communicate() with no wait() first - the
captured-output caller's pattern, where the wrapper stays a zombie while a
survivor holds its pipes (review R1 round 2). Returns (rc, seconds)."""
proc = subprocess.Popen(["bash", "-c", script], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
cwd=str(cwd), env=env, start_new_session=True)
try:
assert _until_true(lambda: ready.exists(), 30), "the wrapper never reached the grace"
time.sleep(2) # inside the grace
t0 = time.monotonic()
proc.terminate()
proc.communicate(timeout=60)
return proc.returncode, time.monotonic() - t0
finally:
if not _gone(proc.pid):
os.killpg(proc.pid, 9)
@pytest.mark.parametrize("status,sub_rc,idle", [("broker_unavailable", 3, 0), ("timeout", 2, 10)])
def test_c_finalize_ends_with_an_unreaped_wrapper(tmp_path, status, sub_rc, idle):
results = []
for which in ("old", "new"):
work = tmp_path / which
reg = work / "reg"
reg.mkdir(parents=True)
record = json.dumps({"job_id": "j1", "status": "running"})
(reg / "j1.json").write_text(record)
log = work / "calls.log"
log.write_text("")
shim = _shim(work)
script = (f'source "{WRAPPER}"\n' + (f'source "{OLD}"\n' if which == "old" else "")
+ f'REGISTRY_DIR="{reg}"; PY="{shim}"\ntouch "{work}/ready"\n'
+ _finalize_body(status, sub_rc, "direct", start_offset=0, timeout=6, idle=idle))
rc, took = _terminate_unreaped(script, work, _env(work, CALL_LOG=log, DELEGATE_JOB_PYTHON=shim), work / "ready")
time.sleep(6) # past the deadline
results.append((rc, took < 4.5, log.read_text(), (reg / "j1.json").read_text() == record))
assert results[0] == results[1] == (-15, True, "", True), results
def test_c_an_unreaped_wrapper_leaves_the_real_record_alone(tmp_path):
"""The same with the real registry CLI and a real dispatch token (loop
type: no publication): the record is still pending after the deadline."""
registry = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
states = []
for which in ("old", "new"):
work = tmp_path / which
reg = work / "reg"
reg.mkdir(parents=True)
env = _env(work, DELEGATE_JOB_PYTHON=sys.executable)
res = subprocess.run([sys.executable, str(registry), "--registry-dir", str(reg), "register", "--prompt", "p",
"--agent", "claude", "--agent-session", "herdr:x", "--job-type", "loop"],
capture_output=True, text=True, env=env, timeout=60)
assert res.returncode == 0, res.stderr
jid = res.stdout.strip().splitlines()[-1]
tok = json.loads((reg / f"{jid}.json").read_text())["dispatch_token"]
script = (f'source "{WRAPPER}"\n' + (f'source "{OLD}"\n' if which == "old" else "")
+ f'REGISTRY_DIR="{reg}"; PY="{sys.executable}"; TYPE=loop; AGENT=human; AGENT_SESSION=herdr:x\n'
+ f'TIMEOUT=6; IDLE_TIMEOUT=10; START_TIME=$(date +%s); _disp_tok="{tok}"\n'
+ f'touch "{work}/ready"; finalize_unfinished {jid} timeout 2 worker\n')
rc, took = _terminate_unreaped(script, work, env, work / "ready")
time.sleep(6)
states.append((rc, took < 4.5, json.loads((reg / f"{jid}.json").read_text())["status"]))
assert states[0] == states[1] == (-15, True, "pending"), states