fix(issue-17): review R1 - the finalizer ends with its wrapper

finalize-end ran the disk grace and the broker re-observation as a child
of the wrapper; TERM to the wrapper left it running, and at the deadline it
still closed the job and published the error (codex R1, reproduced with the
real registry). The bash loops it replaced ended with the wrapper.

The wrapper now passes its own pid (--watch-pid "$$", as for the stall
watcher). finalize-end checks it every 0.2 s through both waits and before
the close and the event, and leaves without either once it is gone.
A comparison test sends TERM to the old and the new wrapper inside each
grace: both exit at once, leave no process behind and make no call.

Also: no blank line at the end of the old-decisions fixture, and the #42
inventory follows main() one line down.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.agents/skills/lib_py/delegation/wrapper.py

@@ -15,7 +15,11 @@ what a subcommand prints or returns:
                                              herdr diagnostic probe bash still runs
 
 finalize-end answers on fd 3 (stdout when fd 3 is closed): its stdout carries
-registry.py's close output to the wrapper's stdout, as before.
+registry.py's close output to the wrapper's stdout, as before. It runs as a
+child of the wrapper and must not outlive it (review R1): with --watch-pid it
+checks that the wrapper is still alive throughout both waits and before each
+side effect, and leaves without closing or publishing once it is gone - as the
+same loops in bash ended with the wrapper.
 
 Stdlib only and Python 3.9 (macOS /usr/bin/python3); registry.py and
 publish_event.py run as the processes the wrapper ran, with its interpreter.
@@ -111,6 +115,36 @@ def recent_disk_heartbeat(registry_dir: str, job_id: str, window: float) -> bool
     return time.time() - seen <= window
 
 
+class OwnerGone(Exception):
+    """The wrapper that ran finalize-end is gone."""
+
+
+def _alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    return True
+
+
+def _check_owner(a: argparse.Namespace) -> None:
+    if a.watch_pid and not _alive(a.watch_pid):
+        raise OwnerGone(a.watch_pid)
+
+
+def _sleep(a: argparse.Namespace, seconds: float) -> None:
+    """time.sleep that looks at the wrapper every 0.2 s."""
+    end = time.time() + seconds
+    while True:
+        _check_owner(a)
+        left = end - time.time()
+        if left <= 0:
+            return
+        time.sleep(min(left, 0.2))
+
+
 def _tail(path: str, n: int) -> List[str]:
     with open(path, errors="replace") as fh:
         return fh.read().splitlines()[-n:]
@@ -163,7 +197,7 @@ def finalize_end(a: argparse.Namespace) -> str:
             left = deadline - _now()
             if left <= 0:
                 break
-            time.sleep(min(left, 3))
+            _sleep(a, min(left, 3))
             disk = disk_status(reg, job)
             if disk in TERMINAL:
                 status = final = disk
@@ -191,7 +225,7 @@ def finalize_end(a: argparse.Namespace) -> str:
                 break
             if deadline - _now() <= 0:
                 break
-            time.sleep(1)
+            _sleep(a, 1)
         if _now() >= deadline:
             clock = "wall"
 
@@ -211,6 +245,7 @@ def finalize_end(a: argparse.Namespace) -> str:
         print(f"finalize: {job} closed by {'wall_clock' if clock == 'wall' else 'no_event'} clock "
               f"(budget {a.timeout}s / idle {a.idle_timeout}s)", file=sys.stderr)
     tok = ["--expect-dispatch-token", a.dispatch_token] if a.dispatch_token else []
+    _check_owner(a)
     close_args = ["close", "--job", job, "--request", close_req] + (["--detail", detail] if detail else []) + tok
     if _run(a, "registry.py", close_args, merge=True) == 4:
         # a dispatch token mismatch (stale attempt): no secondary event (CAS/ABA)
@@ -221,6 +256,7 @@ def finalize_end(a: argparse.Namespace) -> str:
         return "completed"
     # direct path only: publish the error/cancelled event (loop/discuss forbidden: O-1)
     if a.type == "direct":
+        _check_owner(a)
         if close_req == "cancelled":
             _run(a, "publish_event.py", ["--job", job, "--event", "cancelled", "--detail", detail] + tok, merge=False)
         elif a.sub_rc != SUB_RC_ERROR or a.phase == "inject":
@@ -270,6 +306,7 @@ def _parser() -> argparse.ArgumentParser:
             s.add_argument("--preflight-reason", default="")
             s.add_argument("--python", default=sys.executable)
             s.add_argument("--scripts-dir", required=True)
+            s.add_argument("--watch-pid", type=int, default=None, help="the wrapper's pid; gone -> no close, no event")
     return p
 
 
@@ -288,7 +325,10 @@ def main(argv: Optional[List[str]] = None) -> int:
     elif a.cmd == "finalize-begin":
         print(finalize_begin(a))
     else:
-        _answer(finalize_end(a))
+        try:
+            _answer(finalize_end(a))
+        except OwnerGone:
+            print(f"finalize: wrapper pid {a.watch_pid} is gone; job {a.job} left unclosed", file=sys.stderr)
     return 0
 
 

.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -821,11 +821,12 @@ finalize_unfinished() {
     echo "liveness: no wire event for ${IDLE_TIMEOUT}s on job $job_id; herdr/pane says busy=$diag (diagnostic only, not evidence)" >&2
   fi
   # The answer comes back on fd 3: stdout carries registry.py's close output.
+  # --watch-pid: the waits end, unclosed, with this wrapper (review R1).
   { out="$(_decide finalize-end --job "$job_id" --registry-dir "$REGISTRY_DIR" --status "$job_status" \
       ${START_TIME:+--start-time "$START_TIME"} --timeout "$TIMEOUT" --idle-timeout "$IDLE_TIMEOUT" \
       --sub-rc "$sub_rc" --phase "$phase" --type "$TYPE" --final "$final" --evidence-until "$evidence_until" \
       --dispatch-token "${_disp_tok:-}" --preflight-reason "${PREFLIGHT_REASON:-}" \
-      --python "$PY" --scripts-dir "$SCRIPT_DIR/scripts" 3>&1 1>&4 4>&-)"; } 4>&1
+      --python "$PY" --scripts-dir "$SCRIPT_DIR/scripts" --watch-pid "$$" 3>&1 1>&4 4>&-)"; } 4>&1
   FINAL_JOB_STATUS="$out"
 }
 

tests/data/issue17_item4_old_decisions.sh

@@ -261,4 +261,3 @@ cmd_wait() {
   [ "$rc" -eq "$SUB_RC_WALL_CLOCK" ] && rc=$SUB_RC_NO_EVENT
   return "$rc"
 }
-

tests/data/skill_code_blocks.json

@@ -122,7 +122,7 @@
    "class": "c",
    "kind": "invocation",
    "evidence": [
-    "multi-agent-mux-delegate-job/multi-agent-mux-delegate-job:1045-1056"
+    "multi-agent-mux-delegate-job/multi-agent-mux-delegate-job:1046-1057"
    ],
    "mismatch": null,
    "note": "CLI reference; subcommands submit/status/list/verify/wait/logs match main()"

tests/test_issue17_item4_decisions.py

@@ -343,6 +343,64 @@ def test_c_finalize_grace_sees_a_late_completion(tmp_path):
     assert "FINAL=completed" in new[1] and new[3] == ""
 
 
+def _gone(pgid: int) -> bool:
+    try:
+        os.killpg(pgid, 0)
+    except ProcessLookupError:
+        return True
+    except PermissionError:
+        return False
+    return False
+
+
+@pytest.mark.parametrize("status,sub_rc,idle", [("broker_unavailable", 3, 0), ("timeout", 2, 10)])
+def test_c_finalize_ends_with_its_wrapper(tmp_path, status, sub_rc, idle):
+    """Review R1: TERM to the wrapper during either grace ends the grace too -
+    the wrapper exits at once, nothing it started outlives it for long, and no
+    close or event follows after the deadline (6 s)."""
+    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)
+        out = work / "out.txt"
+        script = (f'source "{WRAPPER}"\n' + (f'source "{OLD}"\n' if which == "old" else "")
+                  + f'REGISTRY_DIR="{reg}"; PY="{shim}"\n'
+                  + 'echo READY\n' + _finalize_body(status, sub_rc, "direct", start_offset=0, timeout=6, idle=idle))
+        with open(out, "w") as fh:
+            proc = subprocess.Popen(["bash", "-c", script], stdout=fh, stderr=subprocess.STDOUT, cwd=str(work),
+                                    env=_env(work, CALL_LOG=log, DELEGATE_JOB_PYTHON=shim), start_new_session=True)
+        try:
+            assert _until_true(lambda: "READY" in out.read_text(), 30), out.read_text()
+            time.sleep(2)  # inside the grace
+            assert proc.poll() is None, out.read_text()
+            t0 = time.monotonic()
+            proc.terminate()  # the wrapper only
+            proc.wait(timeout=10)
+            exit_s = time.monotonic() - t0
+            group_gone = _until_true(lambda: _gone(proc.pid), 6)
+            time.sleep(6)  # past the deadline
+            results.append((exit_s < 2, group_gone, log.read_text(), (reg / "j1.json").read_text() == record))
+        finally:
+            if not _gone(proc.pid):
+                os.killpg(proc.pid, 9)
+    assert results[0] == results[1] == (True, True, "", True), results
+
+
+def _until_true(pred, limit):
+    end = time.monotonic() + limit
+    while time.monotonic() < end:
+        if pred():
+            return True
+        time.sleep(0.1)
+    return pred()
+
+
 @pytest.mark.parametrize("rc", [0, 1, 2, 3, 4, 5])
 def test_c_sub_status_exit_code_and_subscriber_up(tmp_path, rc):
     body = (f'st=$(_map_sub_rc {rc} j1 2>/dev/null || _decide sub-status {rc} --job j1 --registry-dir "$REGISTRY_DIR")\n'
@@ -422,3 +480,35 @@ def test_s_the_wrapper_carries_no_moved_decision():
     for label in ("wall_clock_timeout", "deadline_timeout", "agent_blocked", "session_busy"):
         assert label not in text, label
     assert "BASH_REMATCH" not in LIB_INT.read_text()
+
+
+def test_u_finalize_end_leaves_without_side_effects_once_its_wrapper_is_gone(tmp_path):
+    """Review R1, in process: a dead --watch-pid stops the wait and every side effect."""
+    import argparse
+    dead = subprocess.Popen(["true"])
+    dead.wait()
+    (tmp_path / "j.json").write_text(json.dumps({"status": "running"}))
+    log = tmp_path / "calls.log"
+    a = argparse.Namespace(job="j", registry_dir=str(tmp_path), status="broker_unavailable", start_time=int(time.time()),
+                           timeout=30, idle_timeout=0, sub_rc=3, phase="direct", type="direct", final="broker_unavailable",
+                           evidence_until=0, dispatch_token="t", preflight_reason="", python=str(_shim(tmp_path)),
+                           scripts_dir=str(tmp_path / "scripts"), watch_pid=dead.pid)
+    os.environ["CALL_LOG"] = str(log)
+    try:
+        t0 = time.monotonic()
+        with pytest.raises(decide.OwnerGone):
+            decide.finalize_end(a)
+        assert time.monotonic() - t0 < 2
+        a.status, a.start_time = "error", 0  # no wait at all: the close is still withheld
+        with pytest.raises(decide.OwnerGone):
+            decide.finalize_end(a)
+    finally:
+        del os.environ["CALL_LOG"]
+    assert not log.exists() or log.read_text() == ""
+    a.watch_pid = os.getpid()  # alive: the close goes ahead
+    os.environ["CALL_LOG"] = str(log)
+    try:
+        assert decide.finalize_end(a) == "broker_unavailable"
+    finally:
+        del os.environ["CALL_LOG"]
+    assert "registry.py" in log.read_text()