From c747374aa6b685b6d42e62fed2c96a16e298ef90 Mon Sep 17 00:00:00 2001 From: Godopu Date: Wed, 21 Oct 2026 18:24:14 +0000 Subject: [PATCH] fix(session_ids,id_order): optional --session resolution, fresh-marker retention, and canonical cwd matching (Issue #63) --- .agents/skills/lib_py/agents/id_order.py | 75 ++++++-- .agents/skills/lib_py/agents/session_ids.py | 30 ++- tests/test_issue63_same_cwd_ids.py | 191 ++++++++++++++++++++ 3 files changed, 273 insertions(+), 23 deletions(-) diff --git a/.agents/skills/lib_py/agents/id_order.py b/.agents/skills/lib_py/agents/id_order.py index 32360d6..97f5074 100644 --- a/.agents/skills/lib_py/agents/id_order.py +++ b/.agents/skills/lib_py/agents/id_order.py @@ -22,9 +22,8 @@ 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.agents.session_ids import canonical_cwd, claiming_rows, own_id from lib_py.state import load_state_json -from lib_py.verify_session import workspace_key DEFAULT_WAIT_S = 600 @@ -54,7 +53,7 @@ def pid_alive(pid: int) -> bool: 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 ".") + real_cwd = canonical_cwd(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}") @@ -71,7 +70,7 @@ def _parse_pos_int(val: Any) -> Optional[int]: 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) + target_cwd = canonical_cwd(cwd) found: List[str] = [] rows_by_name: Dict[str, Dict[str, Any]] = {} @@ -84,7 +83,7 @@ def blockers(state: Dict[str, Any], cwd: str, agent: str, me: str, 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: + if not s_cwd or canonical_cwd(s_cwd) != target_cwd: continue if agent_of_row(s) != agent or s.get("status") != "running": continue @@ -101,17 +100,40 @@ def blockers(state: Dict[str, Any], cwd: str, agent: str, me: str, m_sess = str(m.get("session") or "") if not m_sess or m_sess == me or m_sess in found: continue + m_cwd = str(m.get("cwd") or "") + if m_cwd and canonical_cwd(m_cwd) != target_cwd: + continue + 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) - 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: + continue + status = str(row.get("status") or "") + cur_own = own_id(row, agent) + row_epoch = int(row.get("herdr_session_epoch") or 0) + end_epoch = int(row.get("stopped_at_epoch") or row.get("terminated_at_epoch") or 0) + m_at = int(m.get("at") or 0) + prev_own = str(m.get("prev_own_id") or "") + prev_epoch = int(m.get("prev_epoch") or 0) + prev_end_epoch = int(m.get("prev_end_epoch") or 0) + + if status == "running": + pinned_for_attempt = bool(cur_own) and ( + cur_own != prev_own or row_epoch > prev_epoch + ) + if not pinned_for_attempt: + found.append(m_sess) + else: + ended_after_attempt = ( + end_epoch > max(m_at, prev_end_epoch) + and (prev_epoch == 0 or row_epoch > prev_epoch) + ) + if not ended_after_attempt: found.append(m_sess) return found @@ -151,18 +173,26 @@ def _read_and_prune_markers(dpath: str, me: str, return out -def write_marker(dpath: str, session: str, owner_pid: int, cwd: str, agent: str) -> str: +def write_marker(dpath: str, session: str, owner_pid: int, cwd: str, agent: str, + prev_row: Optional[Dict[str, Any]] = None) -> str: os.makedirs(dpath, exist_ok=True) marker_path = os.path.join(dpath, f"{session}.json") + payload: Dict[str, Any] = { + "session": session, + "owner_pid": int(owner_pid), + "cwd": canonical_cwd(cwd), + "agent": agent, + "at": int(time.time()), + } + if isinstance(prev_row, dict): + payload["prev_own_id"] = own_id(prev_row, agent) + payload["prev_epoch"] = int(prev_row.get("herdr_session_epoch") or 0) + payload["prev_end_epoch"] = int( + prev_row.get("stopped_at_epoch") or prev_row.get("terminated_at_epoch") or 0 + ) 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) + json.dump(payload, f) os.replace(tmp, marker_path) return marker_path @@ -202,7 +232,14 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int, 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) + prev_row = next( + ( + s for s in (state.get("herdr_sessions") or []) + if isinstance(s, dict) and s.get("name") == session + ), + None, + ) + write_marker(dpath, session, owner_pid, cwd, agent, prev_row=prev_row) return 0 names = ", ".join(blks) diff --git a/.agents/skills/lib_py/agents/session_ids.py b/.agents/skills/lib_py/agents/session_ids.py index 44e5335..3d1b33e 100644 --- a/.agents/skills/lib_py/agents/session_ids.py +++ b/.agents/skills/lib_py/agents/session_ids.py @@ -21,10 +21,19 @@ from lib_py.agents.base import DiscoveryContext, POST_DISCOVERED from lib_py.agents.registry import agent_of_row, get_adapter from lib_py.paths import resolve_home from lib_py.state import load_state_json -from lib_py.verify_session import verify_session_uuid, workspace_key +from lib_py.verify_session import verify_session_uuid from lib_py.workspace_uuid import find_workspace_uuid +def canonical_cwd(path: str) -> str: + if not path: + return "" + try: + return os.path.realpath(path) + except Exception: + return os.path.abspath(path) + + def own_id(row: Optional[Dict[str, Any]], agent: str) -> str: adapter = get_adapter(agent) if not adapter or not isinstance(row, dict): @@ -51,13 +60,13 @@ def claiming_rows(row: Dict[str, Any], uuid: str, state: Dict[str, Any]) -> List def same_type_rows(state: Dict[str, Any], cwd: str, agent: str) -> List[Dict[str, Any]]: if not cwd or not agent: return [] - target_key = workspace_key(cwd) + target_cwd = canonical_cwd(cwd) out: List[Dict[str, Any]] = [] for s in state.get("herdr_sessions") or []: if not isinstance(s, dict): continue s_cwd = (s.get("pane") or {}).get("cwd") or "" - if s_cwd and workspace_key(s_cwd) == target_key and agent_of_row(s) == agent: + if s_cwd and canonical_cwd(s_cwd) == target_cwd and agent_of_row(s) == agent: out.append(s) return out @@ -195,9 +204,22 @@ def decide_resume(state: Dict[str, Any], name: str, agent: str, workspace: str, home_resolved = resolve_home(home) claude_resolved = claude_dir or os.environ.get("CLAUDE_PROJECT_DIR") or f"{home_resolved}/.claude/projects" row = _find_row(state, name) + adapter = get_adapter(agent) + if row is None and not name and workspace: + same = same_type_rows(state, workspace, agent) + if len(same) == 1: + row = same[0] + name = str(row.get("name") or "") + elif adapter is None or adapter.id_lifecycle != POST_DISCOVERED: + creator = next( + (s for s in same if str(s.get("name") or "").endswith(f"-creator-{agent}")), + None, + ) + if creator is not None: + row = creator + name = str(row.get("name") or "") row_ref = row if row is not None else ({"name": name} if name else {}) claimed = set(claimed_elsewhere(row_ref, state)) - adapter = get_adapter(agent) own = own_id(row, agent) if own: diff --git a/tests/test_issue63_same_cwd_ids.py b/tests/test_issue63_same_cwd_ids.py index c6e0aa7..d7fdd42 100644 --- a/tests/test_issue63_same_cwd_ids.py +++ b/tests/test_issue63_same_cwd_ids.py @@ -785,5 +785,196 @@ def test_portability_python39_ast_and_bash_syntax(): assert res.returncode == 0, f"{rel}: {res.stderr}" +def test_resolve_session_id_without_session_arg_returns_workspace_id(mam_sandbox): + """resolve_session_id.sh and decide_resume with name='' return the workspace's stopped row id.""" + from lib_py.agents.session_ids import decide_resume + + ws = mam_sandbox / "opt-sess-ws" + ws.mkdir(parents=True, exist_ok=True) + epoch = int(time.time()) - 100 + _agy_full_artifacts(mam_sandbox, UA) + _agy_cache(mam_sandbox, {ws: UA}) + _claude_transcript(mam_sandbox / ".claude" / "projects", ws, UB) + + rows = [ + _row("opt-sess-ws-creator-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA), + _row("opt-sess-ws-creator-claude", "claude", ws, epoch, status="stopped", claude_session_id_own=UB), + ] + _write_rows(mam_sandbox, rows) + state = {"herdr_sessions": rows} + + assert decide_resume(state, "", "agy", str(ws), home=str(mam_sandbox))["uuid"] == UA + assert decide_resume( + state, "", "claude", str(ws), home=str(mam_sandbox), + claude_dir=str(mam_sandbox / ".claude" / "projects"), + )["uuid"] == UB + + script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resolve_session_id.sh" + res_agy = subprocess.run( + ["bash", str(script), "--workspace", str(ws), "--agent", "agy"], + cwd=str(mam_sandbox), capture_output=True, text=True, timeout=30, + ) + assert res_agy.returncode == 0 + assert res_agy.stdout.strip() == UA + + res_claude = subprocess.run( + ["bash", str(script), "--workspace", str(ws), "--agent", "claude"], + cwd=str(mam_sandbox), capture_output=True, text=True, timeout=30, + ) + assert res_claude.returncode == 0 + assert res_claude.stdout.strip() == UB +def test_id_order_old_stopped_row_with_own_id_does_not_release_fresh_startup_marker(tmp_path, monkeypatch): + """A pre-admission stopped row's own_id must not release a live startup marker + for a fresh resume (stale own_id) or a create reusing an old stopped session name.""" + import threading + from lib_py.agents import id_order + from lib_py.agents.session_ids import decide_resume + + home = tmp_path / "home" + ws = tmp_path / "ws" + ws.mkdir(parents=True) + yaml_path = tmp_path / "agent-sessions.yaml" + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path)) + + for mode, old_own in (("stale_resume", UX), ("recreate_stopped", UA)): + if mode == "recreate_stopped": + _agy_conv(home, UA, time.time() - 50) + old_epoch = int(time.time()) - 100 + old_stop = int(time.time()) - 10 + row_a = _row( + "same-ws-creator-agy", "agy", ws, old_epoch, status="stopped", + agy_conversation_id_own=old_own, stopped_at_epoch=old_stop, + ) + state = {"herdr_sessions": [row_a]} + yaml_path.write_text(yaml.safe_dump(state)) + + if mode == "stale_resume": + dec = decide_resume(state, "same-ws-creator-agy", "agy", str(ws), home=str(home)) + assert dec["uuid"] == "" and dec["drop_stale_id"] is True + + proc_a = subprocess.Popen(["sleep", "120"]) + proc_b = subprocess.Popen(["sleep", "120"]) + try: + # Admit A (fresh startup, uuid=''): writes live marker while old stopped row is in state + assert id_order.wait( + str(ws), "agy", "same-ws-creator-agy", proc_a.pid, uuid="", + yaml_path=str(yaml_path), home=str(home), wait_s_val=0, + ) == 0 + + # Immediate check at cap 0: B must be blocked even before A updates its YAML row + assert id_order.wait( + str(ws), "agy", "same-ws-reviewer-agy", proc_b.pid, uuid="", + yaml_path=str(yaml_path), home=str(home), wait_s_val=0, + ) == 1 + + # Concurrent check: B waits while A transitions old-stopped -> running-unpinned -> running-pinned + b_rc: list = [] + + def _run_b(): + rc = id_order.wait( + str(ws), "agy", "same-ws-reviewer-agy", proc_b.pid, uuid="", + yaml_path=str(yaml_path), home=str(home), wait_s_val=10, poll=0.1, + ) + b_rc.append(rc) + + t = threading.Thread(target=_run_b) + t.start() + time.sleep(0.35) + assert b_rc == [], f"B unblocked prematurely while A was still old stopped row ({mode})" + + # A registers its fresh running row without an own_id + running_unpinned = _row( + "same-ws-creator-agy", "agy", ws, old_epoch + 100, + pid=proc_a.pid, status="running", agy_conversation_id_own=None, + ) + yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [running_unpinned]})) + time.sleep(0.35) + assert b_rc == [], f"B unblocked prematurely while A was running unpinned ({mode})" + + # A records its new conversation id UB -> B unblocks + running_pinned = _row( + "same-ws-creator-agy", "agy", ws, old_epoch + 100, + pid=proc_a.pid, status="running", agy_conversation_id_own=UB, + ) + yaml_path.write_text(yaml.safe_dump({"herdr_sessions": [running_pinned]})) + t.join(timeout=5) + assert b_rc == [0], f"B did not unblock after A recorded its new id ({mode})" + finally: + proc_a.kill() + proc_b.kill() + proc_a.wait() + proc_b.wait() + + +def test_canonical_cwd_distinguishes_underscore_hyphen_and_nested_paths_with_symlink_control(tmp_path, monkeypatch): + """Distinct directories whose lossy workspace_key collides (project_a vs project-a, + nested/sub vs nested-sub) do not block or merge rows; a symlink alias to the same + directory does match.""" + from lib_py.agents import id_order + from lib_py.agents.session_ids import decide_resume, same_type_rows + + home = tmp_path / "home" + ws_under = tmp_path / "project_a" + ws_hyphen = tmp_path / "project-a" + ws_nested = tmp_path / "nested" / "sub" + ws_flat = tmp_path / "nested-sub" + for p in (ws_under, ws_hyphen, ws_nested, ws_flat): + p.mkdir(parents=True) + ws_alias = tmp_path / "project_a_link" + ws_alias.symlink_to(ws_under) + + yaml_path = tmp_path / "agent-sessions.yaml" + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path)) + + _agy_conv(home, UA, time.time() - 10) + _agy_conv(home, UB, time.time() - 5) + _agy_cache(home, {ws_under: UA, ws_hyphen: UB}) + + proc = subprocess.Popen(["sleep", "120"]) + try: + state_live = { + "herdr_sessions": [ + _row("under-agy", "agy", ws_under, 100, pid=proc.pid, status="running"), + _row("nested-agy", "agy", ws_nested, 100, pid=proc.pid, status="running"), + ] + } + yaml_path.write_text(yaml.safe_dump(state_live)) + + # Colliding workspace_key directories do NOT block each other + assert id_order.wait( + str(ws_hyphen), "agy", "hyphen-agy", os.getpid(), + yaml_path=str(yaml_path), home=str(home), wait_s_val=0, + ) == 0 + assert id_order.wait( + str(ws_flat), "agy", "flat-agy", os.getpid(), + yaml_path=str(yaml_path), home=str(home), wait_s_val=0, + ) == 0 + # Symlink alias to ws_under DOES see live unpinned under-agy and blocks + assert id_order.wait( + str(ws_alias), "agy", "alias-agy", os.getpid(), + yaml_path=str(yaml_path), home=str(home), wait_s_val=0, + ) == 1 + finally: + proc.kill() + proc.wait() + + state_stopped = { + "herdr_sessions": [ + _row("under-agy", "agy", ws_under, 100, status="stopped"), + _row("hyphen-agy", "agy", ws_hyphen, 100, status="stopped"), + _row("nested-agy", "agy", ws_nested, 100, status="stopped"), + _row("flat-agy", "agy", ws_flat, 100, status="stopped"), + ] + } + assert [r["name"] for r in same_type_rows(state_stopped, str(ws_hyphen), "agy")] == ["hyphen-agy"] + assert [r["name"] for r in same_type_rows(state_stopped, str(ws_flat), "agy")] == ["flat-agy"] + assert [r["name"] for r in same_type_rows(state_stopped, str(ws_alias), "agy")] == ["under-agy"] + + # Single stopped row in ws_hyphen uses single-row workspace discovery (UB) instead of 2-row fresh spawn + res_hyphen = decide_resume(state_stopped, "hyphen-agy", "agy", str(ws_hyphen), home=str(home)) + assert res_hyphen["uuid"] == UB + assert res_hyphen["notes"] == []