From 518bedac5d8ab832347bfc6682b146f3201021f1 Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 6 Oct 2026 08:09:13 +0000 Subject: [PATCH] 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 --- .agents/skills/lib_py/delegation/wrapper.py | 14 +++- tests/test_issue17_item4_decisions.py | 79 +++++++++++++++++++++ 2 files changed, 90 insertions(+), 3 deletions(-) diff --git a/.agents/skills/lib_py/delegation/wrapper.py b/.agents/skills/lib_py/delegation/wrapper.py index 78bc332..7051e46 100644 --- a/.agents/skills/lib_py/delegation/wrapper.py +++ b/.agents/skills/lib_py/delegation/wrapper.py @@ -120,13 +120,21 @@ class OwnerGone(Exception): 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: os.kill(pid, 0) except ProcessLookupError: return False except PermissionError: 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: @@ -135,14 +143,14 @@ def _check_owner(a: argparse.Namespace) -> 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 while True: _check_owner(a) left = end - time.time() if left <= 0: return - time.sleep(min(left, 0.2)) + time.sleep(min(left, 0.5)) def _tail(path: str, n: int) -> List[str]: diff --git a/tests/test_issue17_item4_decisions.py b/tests/test_issue17_item4_decisions.py index 33de11b..fdd68e2 100644 --- a/tests/test_issue17_item4_decisions.py +++ b/tests/test_issue17_item4_decisions.py @@ -512,3 +512,82 @@ def test_u_finalize_end_leaves_without_side_effects_once_its_wrapper_is_gone(tmp finally: del os.environ["CALL_LOG"] 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