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>

.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]:

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