From ede9c62be39a3c92662e7a0cd90c999c49c1d38f Mon Sep 17 00:00:00 2001 From: Godopu Date: Fri, 23 Oct 2026 07:39:43 +0900 Subject: [PATCH] fix(id_order): keep an orphaned marker blocking while its herdr session is alive (Issue #63) A create/resume killed after spawning its pane but before registering its row left a marker whose owner pid was dead; the next same-cwd waiter pruned it on the pid alone and could start a second session that discovers the same workspace conversation id (PR #69 review P1). - Markers record the herdr server. A dead owner's marker is pruned only once its row is registered for the attempt (running, newer epoch: the row decides) or `_herdr has-session` reports the session gone; an unanswered probe counts as alive. On timeout the kill-session command is printed. - A live marker of another process for the waiter's own session name is kept and blocks instead of being deleted; the waiter's own pid's markers go. - Tests: orphan / registered / gone / same-session / missing pane pid cases, and a create-level reproduction against mock herdr (fails before the fix). Co-Authored-By: Claude Opus 5.5 --- .agents/skills/lib.sh | 3 +- .agents/skills/lib_py/agents/id_order.py | 86 ++++++++++++-- tests/test_issue63_same_cwd_ids.py | 141 ++++++++++++++++++++++- 3 files changed, 216 insertions(+), 14 deletions(-) diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index 06d9a7b..d107fbe 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2043,7 +2043,8 @@ mam_id_order_wait() { local workspace="$1" agent="$2" session_name="$3" uuid="${4:-}" local abs; abs="$(mam_abs_workspace "$workspace")" env_python "$AGENT_SESSIONS_YAML" wait --cwd "$abs" --agent "$agent" \ - --session "$session_name" --owner-pid "$$" --uuid "$uuid" <<'PYEOF' + --session "$session_name" --owner-pid "$$" --uuid "$uuid" \ + --herdr-server "${HERDR_SESSION_NAME:-}" <<'PYEOF' import sys from lib_py.agents.id_order import main sys.exit(main(sys.argv[1:])) diff --git a/.agents/skills/lib_py/agents/id_order.py b/.agents/skills/lib_py/agents/id_order.py index d31fc92..37ca3c2 100644 --- a/.agents/skills/lib_py/agents/id_order.py +++ b/.agents/skills/lib_py/agents/id_order.py @@ -4,7 +4,9 @@ While a post-discovered agent (agy, codex, hermes, opencode) in a workspace is still being created (marker held by a live create/resume process) or is running 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). +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). """ from __future__ import annotations @@ -13,6 +15,7 @@ import fcntl import hashlib import json import os +import subprocess import sys import tempfile import time @@ -50,6 +53,24 @@ 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).""" + 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"' + 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 + except (OSError, subprocess.SubprocessError): + return True + + def order_dir(cwd: str, agent: str, env: Mapping[str, str] = os.environ, home: Optional[str] = None) -> str: h = home or env.get("HOME") or env.get("HOME_DIR") or os.path.expanduser("~") @@ -98,7 +119,7 @@ def blockers(state: Dict[str, Any], cwd: str, agent: str, me: str, if not isinstance(m, dict): continue m_sess = str(m.get("session") or "") - if not m_sess or m_sess == me or m_sess in found: + if not m_sess or m_sess in found: continue m_cwd = str(m.get("cwd") or "") if m_cwd and canonical_cwd(m_cwd) != target_cwd: @@ -106,9 +127,6 @@ def blockers(state: Dict[str, Any], cwd: str, agent: str, me: str, m_agent = str(m.get("agent") or "") if m_agent and m_agent != agent: continue - owner_pid = _parse_pos_int(m.get("owner_pid")) - if owner_pid is None or not is_alive(owner_pid): - continue row = rows_by_name.get(m_sess) if row is None: found.append(m_sess) @@ -141,8 +159,24 @@ def blockers(state: Dict[str, Any], cwd: str, agent: str, me: str, return found -def _read_and_prune_markers(dpath: str, me: str, - is_alive: Callable[[int], bool] = pid_alive) -> List[Dict[str, Any]]: +def _registered_for_attempt(m: Dict[str, Any], row: Optional[Dict[str, Any]]) -> bool: + if not isinstance(row, dict) or row.get("status") != "running": + return False + return int(row.get("herdr_session_epoch") or 0) > int(m.get("prev_epoch") or 0) + + +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, + ) -> 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.""" + rows_by_name = { + str(s.get("name") or ""): s + for s in (state.get("herdr_sessions") or []) if isinstance(s, dict) + } out: List[Dict[str, Any]] = [] try: names = sorted(os.listdir(dpath)) @@ -165,7 +199,18 @@ def _read_and_prune_markers(dpath: str, me: str, continue owner_pid = _parse_pos_int(data.get("owner_pid")) sess = str(data.get("session") or "") - if sess == me or owner_pid is None or not is_alive(owner_pid): + if owner_pid is None or owner_pid == my_pid: + drop = True + elif is_alive(owner_pid): + drop = False + elif sess == me or not sess: + drop = True + 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 + if drop: try: os.remove(fpath) except OSError: @@ -176,7 +221,7 @@ def _read_and_prune_markers(dpath: str, me: str, def write_marker(dpath: str, session: str, owner_pid: int, cwd: str, agent: str, - prev_row: Optional[Dict[str, Any]] = None) -> str: + prev_row: Optional[Dict[str, Any]] = None, herdr_server: str = "") -> str: os.makedirs(dpath, exist_ok=True) marker_path = os.path.join(dpath, f"{session}.json") payload: Dict[str, Any] = { @@ -184,6 +229,7 @@ def write_marker(dpath: str, session: str, owner_pid: int, cwd: str, agent: str, "owner_pid": int(owner_pid), "cwd": canonical_cwd(cwd), "agent": agent, + "herdr_server": herdr_server, "at": int(time.time()), } if isinstance(prev_row, dict): @@ -203,7 +249,9 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, uuid: str = "", yaml_path: Optional[str] = None, home: Optional[str] = None, wait_s_val: Optional[int] = None, poll: float = 0.25, - is_alive: Callable[[int], bool] = pid_alive) -> int: + is_alive: Callable[[int], bool] = pid_alive, + herdr_server: str = "", + pane_alive: Optional[Callable[[str, str], bool]] = None) -> int: adapter = get_adapter(agent) if adapter is None or adapter.id_lifecycle != POST_DISCOVERED: return 0 @@ -215,6 +263,7 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, deadline = time.monotonic() + cap announced = False row_ref = {"name": session} + probe = pane_alive or herdr_session_alive while True: with open(lock_file, "a", encoding="utf-8") as lf: @@ -230,7 +279,8 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, file=sys.stderr, flush=True, ) return 1 - markers = _read_and_prune_markers(dpath, session, is_alive=is_alive) + markers = _read_and_prune_markers(dpath, session, owner_pid, state, + is_alive=is_alive, pane_alive=probe) blks = blockers(state, cwd, agent, session, markers, is_alive=is_alive) if not blks: prev_row = next( @@ -242,7 +292,8 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, ) already_recorded = bool(uuid) and own_id(prev_row, agent) == uuid if not already_recorded and owner_pid and owner_pid > 0: - write_marker(dpath, session, owner_pid, cwd, agent, prev_row=prev_row) + write_marker(dpath, session, owner_pid, cwd, agent, prev_row=prev_row, + herdr_server=herdr_server) return 0 names = ", ".join(blks) @@ -253,6 +304,15 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, f"or stop it with multi-agent-mux-stop.", file=sys.stderr, flush=True, ) + for m in markers: + if m.get("_orphan") and m.get("session") in blks: + srv = m.get("herdr_server") or "default" + print( + f"ERROR: {m['session']}'s create/resume exited before recording it, " + f"but its herdr session is alive; close it with: " + f"HERDR_SESSION_NAME={srv} herdr kill-session -t {m['session']}", + file=sys.stderr, flush=True, + ) return 1 if not announced: print( @@ -275,6 +335,7 @@ def main(argv: Optional[List[str]] = None) -> int: s_wait.add_argument("--session", required=True) s_wait.add_argument("--owner-pid", type=int, default=0) s_wait.add_argument("--uuid", default="") + s_wait.add_argument("--herdr-server", default="") a = p.parse_args(argv) return wait( cwd=a.cwd, @@ -282,6 +343,7 @@ def main(argv: Optional[List[str]] = None) -> int: session=a.session, owner_pid=a.owner_pid, uuid=a.uuid or "", + herdr_server=a.herdr_server or "", ) diff --git a/tests/test_issue63_same_cwd_ids.py b/tests/test_issue63_same_cwd_ids.py index 0e95446..90d1b2a 100644 --- a/tests/test_issue63_same_cwd_ids.py +++ b/tests/test_issue63_same_cwd_ids.py @@ -25,7 +25,15 @@ UB = "b6300002-0000-4000-8000-000000000002" UC = "c6300003-0000-4000-8000-000000000003" UX = "d6300004-0000-4000-8000-000000000004" -pytestmark = pytest.mark.usefixtures("own_workspace_root") +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.""" + from lib_py.agents import id_order + monkeypatch.setattr(id_order, "herdr_session_alive", lambda server, session: False) # --------------------------------------------------------------------------- helpers @@ -1034,3 +1042,134 @@ def test_id_order_resolved_but_unrecorded_resume_uuid_reserves_startup_until_rec proc_b.kill() proc_a.wait() proc_b.wait() + + +def test_id_order_dead_owner_marker_blocks_while_its_unregistered_herdr_session_is_alive( + tmp_path, monkeypatch, capsys +): + """PR #69 review: a create/resume killed after spawning its pane but before + registering its row leaves a dead-owner marker. It keeps blocking while the + marker's herdr session is alive, goes once the row is registered (the row + decides) or the session is gone; a live attempt on the same session name is + not pruned; a running row without a pane pid blocks (fail-closed).""" + 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)) + marker_a = Path(dpath) / "orphan-a-agy.json" + + alive_sessions = {"orphan-a-agy"} + probes = [] + + def pane_alive(server, session): + probes.append((server, session)) + 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) + + dead = subprocess.Popen(["true"]) + dead.wait() + live = subprocess.Popen(["sleep", "120"]) + try: + # Orphan: owner dead, no row, herdr session alive -> blocks, marker kept, kill hint printed + id_order.write_marker(dpath, "orphan-a-agy", dead.pid, str(ws), "agy", herdr_server="srv-x") + capsys.readouterr() + assert wait_b() == 1 + assert marker_a.exists() + assert ("srv-x", "orphan-a-agy") in probes + err = capsys.readouterr().err + assert "HERDR_SESSION_NAME=srv-x herdr kill-session -t orphan-a-agy" in err + + # Row registered for the attempt -> marker pruned without a probe; the row decides + probes.clear() + epoch = int(time.time()) + yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [ + _row("orphan-a-agy", "agy", ws, epoch, pid=live.pid)]})) + assert wait_b() == 1 + assert not marker_a.exists() and probes == [] + yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [ + _row("orphan-a-agy", "agy", ws, epoch, pid=live.pid, agy_conversation_id_own=UA)]})) + assert wait_b() == 0 + + # Session gone -> pruned, B admitted + yaml_path.write_text(yaml.safe_dump({"herdr_sessions": []})) + id_order.write_marker(dpath, "orphan-a-agy", dead.pid, str(ws), "agy", herdr_server="srv-x") + alive_sessions.clear() + assert wait_b() == 0 + assert not marker_a.exists() + + # A live attempt on the same session name is kept and blocks; a dead one is taken over + 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 + 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 + assert json.loads(same.read_text())["owner_pid"] == os.getpid() + same.unlink() + + # Non-blocking review note: a running same-cwd row without a pane pid blocks + no_pid = _row("no-pid-agy", "agy", ws, epoch) + no_pid["pane"].pop("pid") + yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [no_pid]})) + assert wait_b() == 1 + finally: + live.kill() + live.wait() + + +def test_id_order_create_waits_for_orphan_herdr_session_of_a_killed_create( + mam_sandbox, mock_herdr, mock_agents +): + """PR #69 review reproduction: A's create spawned its pane and was killed + before appending its row (dead-owner marker, live herdr agent, no row). + create B refuses while A's herdr session is alive and proceeds once it is gone.""" + from lib_py.agents import id_order + + ws = mam_sandbox / "orphan-ws" + ws.mkdir(parents=True, exist_ok=True) + name_a, name_b = "orphan-ws-creator-agy", "orphan-ws-reviewer-agy" + create_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" + + dead = subprocess.Popen(["true"]) + dead.wait() + holder = subprocess.Popen(["sleep", "120"]) + try: + dpath = id_order.order_dir(str(ws), "agy", home=str(mam_sandbox)) + id_order.write_marker(dpath, name_a, dead.pid, str(ws), "agy", herdr_server="orphan-ws") + _mock_herdr_agent(mock_herdr, name_a, ws, holder.pid, int(time.time()), pane_id="w1:p_a") + + def create_b(): + return subprocess.run( + ["bash", str(create_sh), "--workspace", str(ws), "--agent", "agy", + "--role", "reviewer", "--session", name_b, "--no-onboard"], + capture_output=True, text=True, cwd=str(mam_sandbox), + env=dict(os.environ, MAM_ID_ORDER_WAIT_S="0"), timeout=120, + ) + + res = create_b() + assert res.returncode == 1, res.stdout + res.stderr + assert name_a in res.stderr and "kill-session -t " + name_a in res.stderr + assert name_b not in _read_rows(mam_sandbox) + assert name_b not in json.loads(Path(mock_herdr).read_text()).get("agents", {}) + + state = json.loads(Path(mock_herdr).read_text()) + state["agents"].pop(name_a) + Path(mock_herdr).write_text(json.dumps(state)) + res = create_b() + assert res.returncode == 0, res.stdout + res.stderr + assert _read_rows(mam_sandbox)[name_b]["status"] == "running" + finally: + holder.kill() + holder.wait()