diff --git a/.agents/skills/lib_py/delegation/wrapper.py b/.agents/skills/lib_py/delegation/wrapper.py index ca3c049..78bc332 100644 --- a/.agents/skills/lib_py/delegation/wrapper.py +++ b/.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 diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 665926b..93bff1a 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.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" } diff --git a/tests/data/issue17_item4_old_decisions.sh b/tests/data/issue17_item4_old_decisions.sh index 87dadf5..2c78f3e 100644 --- a/tests/data/issue17_item4_old_decisions.sh +++ b/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" } - diff --git a/tests/data/skill_code_blocks.json b/tests/data/skill_code_blocks.json index 633be69..e7c84e7 100644 --- a/tests/data/skill_code_blocks.json +++ b/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()" diff --git a/tests/test_issue17_item4_decisions.py b/tests/test_issue17_item4_decisions.py index 8370b4b..33de11b 100644 --- a/tests/test_issue17_item4_decisions.py +++ b/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()