From e594d2b6503747b3b55241c46be5efd2b8dfb4d0 Mon Sep 17 00:00:00 2001 From: Godopu Date: Wed, 21 Oct 2026 15:55:37 +0000 Subject: [PATCH] feat(create,resume): same-cwd ordering of post-discovered agents (Issue #63) --- .agents/skills/lib.sh | 15 ++ .agents/skills/lib_py/agents/id_order.py | 249 ++++++++++++++++++ .../scripts/create_session.sh | 4 +- .../scripts/resume_session.sh | 4 + tests/test_issue63_same_cwd_ids.py | 174 ++++++++++++ 5 files changed, 444 insertions(+), 2 deletions(-) create mode 100644 .agents/skills/lib_py/agents/id_order.py diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index 5aee8c4..67ff805 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2051,6 +2051,21 @@ sys.exit(main(sys.argv[1:])) PYEOF } +# --------------------------------------------------------------------------- +# mam_id_order_wait [uuid] +# Issue #63: same-cwd creation/resume ordering for post-discovered agents. +# --------------------------------------------------------------------------- +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' +import sys +from lib_py.agents.id_order import main +sys.exit(main(sys.argv[1:])) +PYEOF +} + # --------------------------------------------------------------------------- # Session isolation — Universal Global Config (Option B, P3-1) # diff --git a/.agents/skills/lib_py/agents/id_order.py b/.agents/skills/lib_py/agents/id_order.py new file mode 100644 index 0000000..32360d6 --- /dev/null +++ b/.agents/skills/lib_py/agents/id_order.py @@ -0,0 +1,249 @@ +"""Same-cwd creation/resume ordering for post-discovered agents (Issue #63). + +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). +""" +from __future__ import annotations + +import argparse +import fcntl +import hashlib +import json +import os +import sys +import tempfile +import time +from typing import Any, Callable, Dict, List, Mapping, Optional + +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.session_ids import claiming_rows, own_id +from lib_py.state import load_state_json +from lib_py.verify_session import workspace_key + +DEFAULT_WAIT_S = 600 + + +def wait_s(env: Mapping[str, str] = os.environ) -> int: + raw = env.get("MAM_ID_ORDER_WAIT_S") + if raw is None: + return DEFAULT_WAIT_S + n = ints.positive_int(raw, 0) + return DEFAULT_WAIT_S if n is None else n + + +def pid_alive(pid: int) -> bool: + if not isinstance(pid, int) or pid <= 0: + return False + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + + +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("~") + real_cwd = os.path.realpath(cwd or ".") + digest = hashlib.sha256(real_cwd.encode("utf-8")).hexdigest()[:16] + return os.path.join(h, ".local", "state", "mam", "id-order", f"{digest}-{agent}") + + +def _parse_pos_int(val: Any) -> Optional[int]: + if isinstance(val, bool): + return None + if isinstance(val, int): + return val if val > 0 else None + s = str(val or "").strip() + return int(s) if s.isdigit() and int(s) > 0 else None + + +def blockers(state: Dict[str, Any], cwd: str, agent: str, me: str, + markers: List[Dict[str, Any]], + is_alive: Callable[[int], bool] = pid_alive) -> List[str]: + target_key = workspace_key(cwd) + found: List[str] = [] + rows_by_name: Dict[str, Dict[str, Any]] = {} + + for s in state.get("herdr_sessions") or []: + if not isinstance(s, dict): + continue + rname = str(s.get("name") or "") + if rname: + rows_by_name[rname] = s + if not rname or rname == me: + continue + s_cwd = (s.get("pane") or {}).get("cwd") or "" + if not s_cwd or workspace_key(s_cwd) != target_key: + continue + if agent_of_row(s) != agent or s.get("status") != "running": + continue + if own_id(s, agent): + continue + pane_pid = _parse_pos_int((s.get("pane") or {}).get("pid")) + if pane_pid is None or is_alive(pane_pid): + if rname not in found: + found.append(rname) + + for m in markers or []: + 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: + 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) + elif not own_id(row, agent): + status = row.get("status") + end_epoch = int(row.get("stopped_at_epoch") or row.get("terminated_at_epoch") or 0) + m_at = int(m.get("at") or 0) + if status == "running" or end_epoch <= m_at: + found.append(m_sess) + + return found + + +def _read_and_prune_markers(dpath: str, me: str, + is_alive: Callable[[int], bool] = pid_alive) -> List[Dict[str, Any]]: + out: List[Dict[str, Any]] = [] + try: + names = sorted(os.listdir(dpath)) + except OSError: + return out + for fname in names: + if not fname.endswith(".json") or fname.startswith("."): + continue + fpath = os.path.join(dpath, fname) + try: + with open(fpath, encoding="utf-8") as f: + data = json.load(f) + except (OSError, ValueError): + try: + os.remove(fpath) + except OSError: + pass + continue + if not isinstance(data, dict): + 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): + try: + os.remove(fpath) + except OSError: + pass + continue + out.append(data) + return out + + +def write_marker(dpath: str, session: str, owner_pid: int, cwd: str, agent: str) -> str: + os.makedirs(dpath, exist_ok=True) + marker_path = os.path.join(dpath, f"{session}.json") + fd, tmp = tempfile.mkstemp(dir=dpath, prefix=".marker.", suffix=".tmp") + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump({ + "session": session, + "owner_pid": int(owner_pid), + "cwd": os.path.realpath(cwd), + "agent": agent, + "at": int(time.time()), + }, f) + os.replace(tmp, marker_path) + return marker_path + + +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: + adapter = get_adapter(agent) + if adapter is None or adapter.id_lifecycle != POST_DISCOVERED: + return 0 + + cap = wait_s(os.environ) if wait_s_val is None else int(wait_s_val) + dpath = order_dir(cwd, agent, os.environ, home=home) + os.makedirs(dpath, exist_ok=True) + lock_file = os.path.join(dpath, "lock") + deadline = time.monotonic() + cap + announced = False + row_ref = {"name": session} + + while True: + with open(lock_file, "a", encoding="utf-8") as lf: + fcntl.flock(lf, fcntl.LOCK_EX) + state = load_state_json(yaml_path) + if uuid: + claimed = set(claimed_elsewhere(row_ref, state)) + if uuid in claimed: + others = ", ".join(claiming_rows(row_ref, uuid, state)) + print( + f"ERROR: {session}'s recorded id {uuid} is also recorded by {others}; " + f"refusing to resume on a conversation another session claims (Issue #63).", + file=sys.stderr, flush=True, + ) + return 1 + markers = _read_and_prune_markers(dpath, session, is_alive=is_alive) + blks = blockers(state, cwd, agent, session, markers, is_alive=is_alive) + if not blks: + if not uuid and owner_pid and owner_pid > 0: + write_marker(dpath, session, owner_pid, cwd, agent) + return 0 + + names = ", ".join(blks) + if time.monotonic() >= deadline: + print( + f"ERROR: {names} in {cwd} still has no recorded {agent} id after {cap}s; " + f"not starting {session} (Issue #63). Wait for its first turn to finish, " + f"or stop it with multi-agent-mux-stop.", + file=sys.stderr, flush=True, + ) + return 1 + if not announced: + print( + f"id-order: waiting for {names} ({agent}, no recorded id yet) in {cwd} " + f"- up to {cap}s (MAM_ID_ORDER_WAIT_S)", + file=sys.stderr, flush=True, + ) + announced = True + remaining = deadline - time.monotonic() + if remaining > 0: + time.sleep(min(poll, remaining)) + + +def main(argv: Optional[List[str]] = None) -> int: + p = argparse.ArgumentParser(prog="python -m lib_py.agents.id_order") + sub = p.add_subparsers(dest="cmd", required=True) + s_wait = sub.add_parser("wait") + s_wait.add_argument("--cwd", required=True) + s_wait.add_argument("--agent", required=True) + s_wait.add_argument("--session", required=True) + s_wait.add_argument("--owner-pid", type=int, default=0) + s_wait.add_argument("--uuid", default="") + a = p.parse_args(argv) + return wait( + cwd=a.cwd, + agent=a.agent, + session=a.session, + owner_pid=a.owner_pid, + uuid=a.uuid or "", + ) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh index 210e2ad..365598a 100755 --- a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh +++ b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh @@ -283,10 +283,10 @@ if [ "$DRY_RUN" = "1" ]; then echo "[dry-run] would spawn: herdr session '$SESSION_NAME' in $WORKSPACE (agent=$AGENT, herdr_session=${HERDR_SESSION_NAME:-default}, herdr_workspace=${MAM_WS_LABEL})" exit 0 fi - +# Issue #63: a same-cwd session of this post-discovered agent records its id first (lib.sh mam_id_order_wait). +mam_id_order_wait "$WORKSPACE" "$AGENT" "$SESSION_NAME" || exit 1 HERDR_EPOCH=$(date +%s) NOW_ISO=$(date -u +'%Y-%m-%dT%H:%M:%SZ') - spawn # _mark_row_rolled_back diff --git a/.agents/skills/multi-agent-mux-resume/scripts/resume_session.sh b/.agents/skills/multi-agent-mux-resume/scripts/resume_session.sh index 6988234..400e827 100755 --- a/.agents/skills/multi-agent-mux-resume/scripts/resume_session.sh +++ b/.agents/skills/multi-agent-mux-resume/scripts/resume_session.sh @@ -238,6 +238,10 @@ if [ "${DRY_RUN:-0}" = "1" ]; then exit 0 fi +# Issue #63: wait if another same-cwd session of this post-discovered agent +# is still being created or has not recorded its id yet (lib.sh mam_id_order_wait). +mam_id_order_wait "$WORKSPACE" "$AGENT" "$SESSION_NAME" "$UUID" || exit 1 + if [ "$AGENT" = "opencode" ]; then if [ -z "${OPENCODE_PERMISSION:-}" ]; then export OPENCODE_PERMISSION='{"*":"allow"}' diff --git a/tests/test_issue63_same_cwd_ids.py b/tests/test_issue63_same_cwd_ids.py index 1d6ece5..c6e0aa7 100644 --- a/tests/test_issue63_same_cwd_ids.py +++ b/tests/test_issue63_same_cwd_ids.py @@ -612,4 +612,178 @@ def test_rev2_claude_unpinned_sibling_row_skips_creator_claimed_id(mam_sandbox, assert f"-r {UB}" in dry_after.stdout +# --------------------------------------------------------------------------- Done Criterion 6 (id_order & create/resume wait) + +def test_c6_create_waits_or_refuses_while_same_cwd_post_discovered_row_has_no_id( + mam_sandbox, mock_herdr, mock_agents +): + """Done Criterion 6 (a) & (b): A is running without an id and has a live pane pid. + (a) With MAM_ID_ORDER_WAIT_S=0, create B exits 1 without spawning or adding a row + and names A on stderr. + (b) With MAM_ID_ORDER_WAIT_S=30, once A records its id in a background thread, + create B proceeds (rc 0) and records herdr_session_epoch >= the pin time.""" + import threading + + ws = mam_sandbox / "c6-ws" + ws.mkdir(parents=True, exist_ok=True) + epoch_a = int(time.time()) - 100 + name_a, name_b = "c6-ws-creator-agy", "c6-ws-reviewer-agy" + create_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" + + holder = subprocess.Popen(["sleep", "120"]) + try: + _write_rows(mam_sandbox, [_row(name_a, "agy", ws, epoch_a, pid=holder.pid)]) + _mock_herdr_agent(mock_herdr, name_a, ws, holder.pid, epoch_a, pane_id="w1:p_a") + + # (a) Wait cap 0 -> immediate refusal + env0 = dict(os.environ, MAM_ID_ORDER_WAIT_S="0") + res0 = 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=env0, timeout=120, + ) + assert res0.returncode == 1, res0.stdout + res0.stderr + assert name_a in res0.stderr + assert name_b not in _read_rows(mam_sandbox) + herdr_state = json.loads(Path(mock_herdr).read_text()) + assert name_b not in herdr_state.get("agents", {}) + + # Also check resume --dry-run does not wait even with MAM_ID_ORDER_WAIT_S=0 + _write_rows(mam_sandbox, [ + _row(name_a, "agy", ws, epoch_a, pid=holder.pid), + _row(name_b, "agy", ws, epoch_a, status="stopped", agy_conversation_id_own=UB), + ]) + _agy_full_artifacts(mam_sandbox, UB) + dry_res = subprocess.run( + ["bash", str(mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh"), + "--workspace", str(ws), "--agent", "agy", "--session", name_b, "--dry-run"], + capture_output=True, text=True, cwd=str(mam_sandbox), env=env0, timeout=60, + ) + assert dry_res.returncode == 0, dry_res.stdout + dry_res.stderr + + # Reset to only A running without id for (b) + _write_rows(mam_sandbox, [_row(name_a, "agy", ws, epoch_a, pid=holder.pid)]) + pinned_at = [] + + def _pin_a_later(): + time.sleep(1.5) + pinned_at.append(int(time.time())) + _write_rows(mam_sandbox, [ + _row(name_a, "agy", ws, epoch_a, pid=holder.pid, agy_conversation_id_own=UA, + session_id_source="discovered", session_id_verified=True), + ]) + + t = threading.Thread(target=_pin_a_later) + t.start() + try: + env30 = dict(os.environ, MAM_ID_ORDER_WAIT_S="30") + res_ok = 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=env30, timeout=120, + ) + finally: + t.join(timeout=10) + + assert res_ok.returncode == 0, res_ok.stdout + res_ok.stderr + assert "id-order: waiting for" in res_ok.stderr and name_a in res_ok.stderr + rows = _read_rows(mam_sandbox) + assert name_b in rows and rows[name_b]["status"] == "running" + assert pinned_at and rows[name_b]["herdr_session_epoch"] >= pinned_at[0] + finally: + holder.kill() + holder.wait() + + +def test_c6_id_order_markers_non_blocking_cases_and_uuid_recheck(tmp_path, monkeypatch): + """Done Criterion 6 (c), (d), and Criterion 7 uuid recheck during wait + knob parsing.""" + from lib_py.agents import id_order + + home = tmp_path / "home" + ws = tmp_path / "ws" + ws_other = tmp_path / "ws-other" + ws.mkdir(parents=True) + ws_other.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)) + + # Knob parsing rules + assert id_order.wait_s({}) == 600 + assert id_order.wait_s({"MAM_ID_ORDER_WAIT_S": "0"}) == 0 + assert id_order.wait_s({"MAM_ID_ORDER_WAIT_S": " 05 "}) == 5 + for bad in ("abc", "-1", "9" * 10, ""): + assert id_order.wait_s({"MAM_ID_ORDER_WAIT_S": bad}) == 600, bad + + # (c) Marker-only "being created": blocks while owner_pid is alive, passes once dead + dpath = id_order.order_dir(str(ws), "agy", home=str(home)) + proc = subprocess.Popen(["sleep", "120"]) + try: + id_order.write_marker(dpath, "creating-a-agy", proc.pid, str(ws), "agy") + assert id_order.wait(str(ws), "agy", "creating-b-agy", os.getpid(), + yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 1 + finally: + proc.kill() + proc.wait() + + # Dead owner_pid marker is pruned and wait succeeds + assert id_order.wait(str(ws), "agy", "creating-b-agy", os.getpid(), + yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 0 + # Clean up creating-b-agy's own marker for next sub-checks + for f in Path(dpath).glob("*.json"): + f.unlink() + + # (d) Non-blocking cases: claude, different folder, different agent, dead pane pid, pinned row + dead_proc = subprocess.Popen(["true"]) + dead_proc.wait() + live_proc = subprocess.Popen(["sleep", "120"]) + try: + state_nonblock = { + "herdr_sessions": [ + _row("other-cwd-agy", "agy", ws_other, 100, pid=live_proc.pid), + _row("same-cwd-codex", "codex", ws, 100, pid=live_proc.pid), + _row("dead-pane-agy", "agy", ws, 100, pid=dead_proc.pid), + _row("pinned-agy", "agy", ws, 100, pid=live_proc.pid, agy_conversation_id_own=UA), + ] + } + yaml_path.write_text(yaml.safe_dump(state_nonblock)) + assert id_order.wait(str(ws), "claude", "new-claude", os.getpid(), + yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 0 + assert id_order.wait(str(ws), "agy", "new-agy", os.getpid(), + yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 0 + + # Criterion 7: wait --uuid refuses when another row already claimed that uuid + assert id_order.wait(str(ws), "agy", "resuming-agy", os.getpid(), uuid=UA, + yaml_path=str(yaml_path), home=str(home), wait_s_val=0) == 1 + finally: + live_proc.kill() + live_proc.wait() + + +def test_portability_python39_ast_and_bash_syntax(): + """Static guard: new/updated Python modules parse under Python 3.9 and shell scripts pass bash -n.""" + import ast + + for rel in ( + "lib_py/workspace_uuid.py", + "lib_py/agents/session_ids.py", + "lib_py/agents/id_order.py", + ): + src = (SKILLS / rel).read_text(encoding="utf-8") + ast.parse(src, filename=rel, feature_version=(3, 9)) + + for rel in ( + "lib.sh", + "multi-agent-mux-stop/scripts/stop_session.sh", + "multi-agent-mux-resume/scripts/resolve_session_id.sh", + "multi-agent-mux-resume/scripts/resume_session.sh", + "multi-agent-mux-resume/scripts/update_yaml_resumed.sh", + "multi-agent-mux-create/scripts/create_session.sh", + ): + res = subprocess.run(["bash", "-n", str(SKILLS / rel)], capture_output=True, text=True, timeout=30) + assert res.returncode == 0, f"{rel}: {res.stderr}" + + +