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