From 8f64174b6b8b5b64eaa06c0530f552b3440f46b5 Mon Sep 17 00:00:00 2001 From: Godopu Date: Fri, 23 Oct 2026 10:07:54 +0900 Subject: [PATCH] fix(id_order): prune an orphaned marker only on an authoritative "gone", within the wait cap (Issue #63) PR #69 review round 2. - The herdr probe is tri-state (present / absent / unknown) from a parsed `_herdr agent list`: absent only when the list parses without the name or herdr answers server_not_running. A failed, unparseable or timed-out lookup is unknown and keeps the marker (codex P1: has-session's exit 1 meant both "absent" and "lookup failed"). - Probes run with MAM_NO_AUTOSTART=1 so a check never starts a server (agy P2). - Probe time is capped by max(deadline, start + PROBE_FLOOR_S=2 s), shared by all markers, so a slow herdr cannot hold the wait or its lock past the cap (codex P2); MAM_ID_ORDER_WAIT_S=0 still gets one bounded look. - Tests: unknown probe keeps and refuses; delayed probe at caps 0/1/3 stays within the bound; real probe via lib.sh mam_id_order_wait against mock herdr (listed / unlisted / failing list / server_not_running). Co-Authored-By: Claude Opus 5.5 --- .agents/skills/lib_py/agents/id_order.py | 87 +++++++++++----- tests/test_issue63_same_cwd_ids.py | 122 +++++++++++++++++++++-- 2 files changed, 179 insertions(+), 30 deletions(-) diff --git a/.agents/skills/lib_py/agents/id_order.py b/.agents/skills/lib_py/agents/id_order.py index 37ca3c2..9419cff 100644 --- a/.agents/skills/lib_py/agents/id_order.py +++ b/.agents/skills/lib_py/agents/id_order.py @@ -5,8 +5,8 @@ While a post-discovered agent (agy, codex, hermes, opencode) in a workspace is without a recorded conversation id (live pane pid), a second create or resume of the same agent type in that workspace waits until the first records its id, up to MAM_ID_ORDER_WAIT_S (default 600 s; 0 = do not wait). A marker whose owner -died before the row was registered keeps blocking while its herdr session is -alive (the pane may still discover the workspace id). +died before the row was registered keeps blocking until herdr answers that its +session is gone (the pane may still discover the workspace id). """ from __future__ import annotations @@ -25,10 +25,15 @@ from lib_py import ints from lib_py.agents.autopin import claimed_elsewhere from lib_py.agents.base import POST_DISCOVERED from lib_py.agents.registry import agent_of_row, get_adapter +from lib_py.agents.sanitize import sanitize_herdr_agent_name from lib_py.agents.session_ids import canonical_cwd, claiming_rows, own_id from lib_py.state import load_state_json DEFAULT_WAIT_S = 600 +# herdr probes of dead-owner markers share this budget past the wait deadline, +# so MAM_ID_ORDER_WAIT_S=0 still gets one look at whether an orphan is gone. +PROBE_FLOOR_S = 2.0 +PROBE_TIMEOUT_S = 30.0 def wait_s(env: Mapping[str, str] = os.environ) -> int: @@ -53,22 +58,48 @@ def pid_alive(pid: int) -> bool: return False -def herdr_session_alive(server: str, session: str) -> bool: - """lib.sh `_herdr has-session` on the marker's herdr server; an unanswered - probe counts as alive (the wait stays bounded by MAM_ID_ORDER_WAIT_S).""" +def _server_not_running(*texts: str) -> bool: + """herdr's own error envelope for a server that is down (lib.sh ISSUE-9).""" + for text in texts: + for line in [text] + text.splitlines(): + try: + doc = json.loads(line) + except ValueError: + continue + err = doc.get("error") if isinstance(doc, dict) else None + if isinstance(err, dict) and err.get("code") == "server_not_running": + return True + return False + + +def herdr_session_state(server: str, session: str, timeout: float) -> Optional[bool]: + """Is `session` live on herdr server `server`: True, False, or None when + herdr gave no authoritative answer. False only from a parsed `agent list` + without the name or herdr's server_not_running error; a failed, timed-out + or unparseable lookup is None. Never starts a server (MAM_NO_AUTOSTART).""" + if timeout <= 0: + return None skills_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - env = dict(os.environ) - if server: - env["HERDR_SESSION_NAME"] = server - script = 'source "$1" && _herdr has-session -t "$2"' + env = dict(os.environ, MAM_NO_AUTOSTART="1", HERDR_SESSION_NAME=server or "default") + script = 'source "$1" && _herdr agent list' try: - return subprocess.run( - ["bash", "-c", script, "_", os.path.join(skills_dir, "lib.sh"), session], - env=env, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, timeout=30, - ).returncode == 0 + res = subprocess.run( + ["bash", "-c", script, "_", os.path.join(skills_dir, "lib.sh")], + env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, + timeout=min(timeout, PROBE_TIMEOUT_S), + ) except (OSError, subprocess.SubprocessError): - return True + return None + if res.returncode != 0: + return False if _server_not_running(res.stdout, res.stderr) else None + try: + agents = json.loads(res.stdout)["result"]["agents"] + except (ValueError, KeyError, TypeError): + return None + if not isinstance(agents, list): + return None + names = {session, sanitize_herdr_agent_name(session)} + return any(isinstance(a, dict) and a.get("name") in names for a in agents) def order_dir(cwd: str, agent: str, env: Mapping[str, str] = os.environ, @@ -167,12 +198,13 @@ def _registered_for_attempt(m: Dict[str, Any], row: Optional[Dict[str, Any]]) -> def _read_and_prune_markers(dpath: str, me: str, my_pid: int, state: Dict[str, Any], is_alive: Callable[[int], bool] = pid_alive, - pane_alive: Callable[[str, str], bool] = herdr_session_alive, + pane_state: Callable[[str, str], Optional[bool]] = ( + lambda server, session: None), ) -> List[Dict[str, Any]]: """Markers that still count. A live owner keeps its marker (another live attempt on `me` included). A dead owner's marker goes once its row is - registered for the attempt (the row decides from then on) or its herdr - session is gone; until then it is kept as an orphan.""" + registered for the attempt (the row decides from then on) or herdr answers + that its session is gone; until then it is kept as an orphan.""" rows_by_name = { str(s.get("name") or ""): s for s in (state.get("herdr_sessions") or []) if isinstance(s, dict) @@ -208,8 +240,9 @@ def _read_and_prune_markers(dpath: str, me: str, my_pid: int, state: Dict[str, A elif _registered_for_attempt(data, rows_by_name.get(sess)): drop = True else: - drop = not pane_alive(str(data.get("herdr_server") or ""), sess) - data["_orphan"] = True + live = pane_state(str(data.get("herdr_server") or ""), sess) + drop = live is False + data["_orphan"] = "alive" if live else "unknown" if drop: try: os.remove(fpath) @@ -251,7 +284,7 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, poll: float = 0.25, is_alive: Callable[[int], bool] = pid_alive, herdr_server: str = "", - pane_alive: Optional[Callable[[str, str], bool]] = None) -> int: + pane_state: Optional[Callable[[str, str, float], Optional[bool]]] = None) -> int: adapter = get_adapter(agent) if adapter is None or adapter.id_lifecycle != POST_DISCOVERED: return 0 @@ -261,9 +294,13 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, os.makedirs(dpath, exist_ok=True) lock_file = os.path.join(dpath, "lock") deadline = time.monotonic() + cap + probe_until = max(deadline, time.monotonic() + PROBE_FLOOR_S) announced = False row_ref = {"name": session} - probe = pane_alive or herdr_session_alive + probe = pane_state or herdr_session_state + + def bounded_probe(server: str, sess: str) -> Optional[bool]: + return probe(server, sess, probe_until - time.monotonic()) while True: with open(lock_file, "a", encoding="utf-8") as lf: @@ -280,7 +317,7 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, ) return 1 markers = _read_and_prune_markers(dpath, session, owner_pid, state, - is_alive=is_alive, pane_alive=probe) + is_alive=is_alive, pane_state=bounded_probe) blks = blockers(state, cwd, agent, session, markers, is_alive=is_alive) if not blks: prev_row = next( @@ -307,9 +344,11 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, for m in markers: if m.get("_orphan") and m.get("session") in blks: srv = m.get("herdr_server") or "default" + seen = ("its herdr session is alive" if m["_orphan"] == "alive" + else "herdr could not tell whether its session is gone") print( f"ERROR: {m['session']}'s create/resume exited before recording it, " - f"but its herdr session is alive; close it with: " + f"and {seen}; close it with: " f"HERDR_SESSION_NAME={srv} herdr kill-session -t {m['session']}", file=sys.stderr, flush=True, ) diff --git a/tests/test_issue63_same_cwd_ids.py b/tests/test_issue63_same_cwd_ids.py index 90d1b2a..9b891fe 100644 --- a/tests/test_issue63_same_cwd_ids.py +++ b/tests/test_issue63_same_cwd_ids.py @@ -31,9 +31,9 @@ pytestmark = pytest.mark.usefixtures("own_workspace_root", "no_herdr_probe") @pytest.fixture def no_herdr_probe(monkeypatch): """In-process id_order.wait never asks the real herdr about a dead-owner - marker's session: it reads as gone unless a test passes pane_alive.""" + marker's session: it reads as gone unless a test passes pane_state.""" from lib_py.agents import id_order - monkeypatch.setattr(id_order, "herdr_session_alive", lambda server, session: False) + monkeypatch.setattr(id_order, "herdr_session_state", lambda server, session, timeout: False) # --------------------------------------------------------------------------- helpers @@ -1065,15 +1065,18 @@ def test_id_order_dead_owner_marker_blocks_while_its_unregistered_herdr_session_ marker_a = Path(dpath) / "orphan-a-agy.json" alive_sessions = {"orphan-a-agy"} + unknown_sessions = set() probes = [] - def pane_alive(server, session): + def pane_state(server, session, timeout): probes.append((server, session)) + if session in unknown_sessions: + return None return session in alive_sessions def wait_b(**kw): return id_order.wait(str(ws), "agy", "new-b-agy", os.getpid(), yaml_path=str(yaml_path), - home=str(home), wait_s_val=0, pane_alive=pane_alive, **kw) + home=str(home), wait_s_val=0, pane_state=pane_state, **kw) dead = subprocess.Popen(["true"]) dead.wait() @@ -1088,6 +1091,13 @@ def test_id_order_dead_owner_marker_blocks_while_its_unregistered_herdr_session_ err = capsys.readouterr().err assert "HERDR_SESSION_NAME=srv-x herdr kill-session -t orphan-a-agy" in err + # No authoritative answer from herdr -> kept and blocks as well + unknown_sessions.add("orphan-a-agy") + assert wait_b() == 1 + assert marker_a.exists() + assert "herdr could not tell whether its session is gone" in capsys.readouterr().err + unknown_sessions.clear() + # Row registered for the attempt -> marker pruned without a probe; the row decides probes.clear() epoch = int(time.time()) @@ -1110,12 +1120,12 @@ def test_id_order_dead_owner_marker_blocks_while_its_unregistered_herdr_session_ same = Path(dpath) / "same-agy.json" id_order.write_marker(dpath, "same-agy", live.pid, str(ws), "agy") assert id_order.wait(str(ws), "agy", "same-agy", os.getpid(), yaml_path=str(yaml_path), - home=str(home), wait_s_val=0, pane_alive=pane_alive) == 1 + home=str(home), wait_s_val=0, pane_state=pane_state) == 1 assert json.loads(same.read_text())["owner_pid"] == live.pid live.kill() live.wait() assert id_order.wait(str(ws), "agy", "same-agy", os.getpid(), yaml_path=str(yaml_path), - home=str(home), wait_s_val=0, pane_alive=pane_alive) == 0 + home=str(home), wait_s_val=0, pane_state=pane_state) == 0 assert json.loads(same.read_text())["owner_pid"] == os.getpid() same.unlink() @@ -1173,3 +1183,103 @@ def test_id_order_create_waits_for_orphan_herdr_session_of_a_killed_create( finally: holder.kill() holder.wait() + + +def test_id_order_herdr_probes_stay_within_the_wait_cap(tmp_path, monkeypatch): + """PR #69 review r2 P2: probes of dead-owner markers get the time left until + max(deadline, start + PROBE_FLOOR_S), shared by all markers, so a slow herdr + cannot hold the wait (and its lock) past the cap; an expired probe is unknown.""" + from lib_py.agents import id_order + + home = tmp_path / "home" + ws = tmp_path / "ws" + ws.mkdir(parents=True) + yaml_path = tmp_path / "agent-sessions.yaml" + yaml_path.write_text(yaml.safe_dump({"herdr_sessions": []})) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path)) + dpath = id_order.order_dir(str(ws), "agy", home=str(home)) + dead = subprocess.Popen(["true"]) + dead.wait() + for name in ("orphan-1-agy", "orphan-2-agy"): + id_order.write_marker(dpath, name, dead.pid, str(ws), "agy", herdr_server="srv") + + budgets = [] + + def slow_probe(server, session, timeout): + budgets.append(timeout) + if timeout <= 0: + return None + time.sleep(min(timeout, 5.0)) + return None + + for cap in (0, 1, 3): + budgets.clear() + t0 = time.monotonic() + rc = id_order.wait(str(ws), "agy", "new-agy", os.getpid(), yaml_path=str(yaml_path), + home=str(home), wait_s_val=cap, poll=0.05, pane_state=slow_probe) + elapsed = time.monotonic() - t0 + bound = max(cap, id_order.PROBE_FLOOR_S) + assert rc == 1, cap + assert elapsed < bound + 0.6, (cap, elapsed) + assert budgets and max(budgets) <= bound + 0.01, (cap, budgets) + assert len(list(Path(dpath).glob("orphan-*.json"))) == 2, cap + + +def test_id_order_real_probe_against_mock_herdr_fails_closed(mam_sandbox, mock_herdr): + """PR #69 review r2 P1: lib.sh mam_id_order_wait with the real probe (sandbox + copy, mock herdr). A listed orphan blocks; an unlisted one is pruned; a failed + `agent list` keeps the marker and refuses; herdr's server_not_running prunes.""" + from lib_py.agents import id_order + + ws = mam_sandbox / "probe-ws" + ws.mkdir(parents=True, exist_ok=True) + name_a, name_b = "probe-ws-creator-agy", "probe-ws-reviewer-agy" + lib = mam_sandbox / ".agents" / "skills" / "lib.sh" + dpath = id_order.order_dir(str(ws), "agy", home=str(mam_sandbox)) + marker_a = Path(dpath) / f"{name_a}.json" + dead = subprocess.Popen(["true"]) + dead.wait() + holder = subprocess.Popen(["sleep", "120"]) + + def set_mock(**kw): + state = json.loads(Path(mock_herdr).read_text()) + state.pop("force_error_code", None) + state["agents"].pop(name_a, None) + state.update(kw) + Path(mock_herdr).write_text(json.dumps(state)) + + def wait_b(): + for f in Path(dpath).glob(f"{name_b}.json"): + f.unlink() + id_order.write_marker(dpath, name_a, dead.pid, str(ws), "agy", herdr_server="probe-ws") + return subprocess.run( + ["bash", "-c", 'source "$1" && mam_id_order_wait "$2" agy "$3"', "_", str(lib), str(ws), name_b], + capture_output=True, text=True, cwd=str(mam_sandbox), timeout=120, + env=dict(os.environ, MAM_ID_ORDER_WAIT_S="0", HERDR_SESSION_NAME="probe-ws"), + ) + + try: + _mock_herdr_agent(mock_herdr, name_a, ws, holder.pid, int(time.time()), pane_id="w1:p_a") + res = wait_b() + assert res.returncode == 1 and "its herdr session is alive" in res.stderr, res.stderr + assert marker_a.exists() + + set_mock(force_error_code="protocol_mismatch") + res = wait_b() + assert res.returncode == 1, res.stderr + assert "herdr could not tell whether its session is gone" in res.stderr + assert marker_a.exists() + + set_mock(force_error_code="server_not_running") + res = wait_b() + assert res.returncode == 0, res.stderr + assert not marker_a.exists() + + set_mock() + res = wait_b() + assert res.returncode == 0, res.stderr + assert not marker_a.exists() + finally: + holder.kill() + holder.wait()