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 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-23 07:39:43 +09:00
1 parent a907cc5294
commit ede9c62be3
3 files changed
+216 -14

No files matched your search

+2 -1
View File
@@ -2043,7 +2043,8 @@ mam_id_order_wait() {
local workspace="$1" agent="$2" session_name="$3" uuid="${4:-}" local workspace="$1" agent="$2" session_name="$3" uuid="${4:-}"
local abs; abs="$(mam_abs_workspace "$workspace")" local abs; abs="$(mam_abs_workspace "$workspace")"
env_python "$AGENT_SESSIONS_YAML" wait --cwd "$abs" --agent "$agent" \ 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 import sys
from lib_py.agents.id_order import main from lib_py.agents.id_order import main
sys.exit(main(sys.argv[1:])) sys.exit(main(sys.argv[1:]))
+74 -12
View File
@@ -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 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 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 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 from __future__ import annotations
@@ -13,6 +15,7 @@ import fcntl
import hashlib import hashlib
import json import json
import os import os
import subprocess
import sys import sys
import tempfile import tempfile
import time import time
@@ -50,6 +53,24 @@ def pid_alive(pid: int) -> bool:
return False 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, def order_dir(cwd: str, agent: str, env: Mapping[str, str] = os.environ,
home: Optional[str] = None) -> str: home: Optional[str] = None) -> str:
h = home or env.get("HOME") or env.get("HOME_DIR") or os.path.expanduser("~") 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): if not isinstance(m, dict):
continue continue
m_sess = str(m.get("session") or "") 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 continue
m_cwd = str(m.get("cwd") or "") m_cwd = str(m.get("cwd") or "")
if m_cwd and canonical_cwd(m_cwd) != target_cwd: 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 "") m_agent = str(m.get("agent") or "")
if m_agent and m_agent != agent: if m_agent and m_agent != agent:
continue 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) row = rows_by_name.get(m_sess)
if row is None: if row is None:
found.append(m_sess) found.append(m_sess)
@@ -141,8 +159,24 @@ def blockers(state: Dict[str, Any], cwd: str, agent: str, me: str,
return found return found
def _read_and_prune_markers(dpath: str, me: str, def _registered_for_attempt(m: Dict[str, Any], row: Optional[Dict[str, Any]]) -> bool:
is_alive: Callable[[int], bool] = pid_alive) -> List[Dict[str, Any]]: 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]] = [] out: List[Dict[str, Any]] = []
try: try:
names = sorted(os.listdir(dpath)) names = sorted(os.listdir(dpath))
@@ -165,7 +199,18 @@ def _read_and_prune_markers(dpath: str, me: str,
continue continue
owner_pid = _parse_pos_int(data.get("owner_pid")) owner_pid = _parse_pos_int(data.get("owner_pid"))
sess = str(data.get("session") or "") 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: try:
os.remove(fpath) os.remove(fpath)
except OSError: 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, 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) os.makedirs(dpath, exist_ok=True)
marker_path = os.path.join(dpath, f"{session}.json") marker_path = os.path.join(dpath, f"{session}.json")
payload: Dict[str, Any] = { 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), "owner_pid": int(owner_pid),
"cwd": canonical_cwd(cwd), "cwd": canonical_cwd(cwd),
"agent": agent, "agent": agent,
"herdr_server": herdr_server,
"at": int(time.time()), "at": int(time.time()),
} }
if isinstance(prev_row, dict): 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, uuid: str = "", yaml_path: Optional[str] = None,
home: Optional[str] = None, wait_s_val: Optional[int] = None, home: Optional[str] = None, wait_s_val: Optional[int] = None,
poll: float = 0.25, 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) adapter = get_adapter(agent)
if adapter is None or adapter.id_lifecycle != POST_DISCOVERED: if adapter is None or adapter.id_lifecycle != POST_DISCOVERED:
return 0 return 0
@@ -215,6 +263,7 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int,
deadline = time.monotonic() + cap deadline = time.monotonic() + cap
announced = False announced = False
row_ref = {"name": session} row_ref = {"name": session}
probe = pane_alive or herdr_session_alive
while True: while True:
with open(lock_file, "a", encoding="utf-8") as lf: 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, file=sys.stderr, flush=True,
) )
return 1 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) blks = blockers(state, cwd, agent, session, markers, is_alive=is_alive)
if not blks: if not blks:
prev_row = next( 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 already_recorded = bool(uuid) and own_id(prev_row, agent) == uuid
if not already_recorded and owner_pid and owner_pid > 0: 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 return 0
names = ", ".join(blks) 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.", f"or stop it with multi-agent-mux-stop.",
file=sys.stderr, flush=True, 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 return 1
if not announced: if not announced:
print( print(
@@ -275,6 +335,7 @@ def main(argv: Optional[List[str]] = None) -> int:
s_wait.add_argument("--session", required=True) s_wait.add_argument("--session", required=True)
s_wait.add_argument("--owner-pid", type=int, default=0) s_wait.add_argument("--owner-pid", type=int, default=0)
s_wait.add_argument("--uuid", default="") s_wait.add_argument("--uuid", default="")
s_wait.add_argument("--herdr-server", default="")
a = p.parse_args(argv) a = p.parse_args(argv)
return wait( return wait(
cwd=a.cwd, cwd=a.cwd,
@@ -282,6 +343,7 @@ def main(argv: Optional[List[str]] = None) -> int:
session=a.session, session=a.session,
owner_pid=a.owner_pid, owner_pid=a.owner_pid,
uuid=a.uuid or "", uuid=a.uuid or "",
herdr_server=a.herdr_server or "",
) )
+140 -1
View File
@@ -25,7 +25,15 @@ UB = "b6300002-0000-4000-8000-000000000002"
UC = "c6300003-0000-4000-8000-000000000003" UC = "c6300003-0000-4000-8000-000000000003"
UX = "d6300004-0000-4000-8000-000000000004" 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 # --------------------------------------------------------------------------- helpers
@@ -1034,3 +1042,134 @@ def test_id_order_resolved_but_unrecorded_resume_uuid_reserves_startup_until_rec
proc_b.kill() proc_b.kill()
proc_a.wait() proc_a.wait()
proc_b.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()