From b8a30d5b93e56f8b272e3dd362b5e02faa2ef63e Mon Sep 17 00:00:00 2001 From: Godopu Date: Wed, 21 Oct 2026 15:31:42 +0000 Subject: [PATCH] feat(agents): session_ids - the stop/purge/resume id decision (Issue #63) --- .agents/skills/lib.sh | 13 + .agents/skills/lib_py/agents/session_ids.py | 329 ++++++++++++++++++++ tests/test_issue63_same_cwd_ids.py | 103 ++++++ 3 files changed, 445 insertions(+) create mode 100644 .agents/skills/lib_py/agents/session_ids.py diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index daa38b1..5aee8c4 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2032,12 +2032,25 @@ PYEOF # resolver (per-row own id -> workspace-scoped disk scan), # so recording its result into the row before kill guarantees tier-1 on the next # resume. Pass session_name to prefer that specific row's recorded id. Always exits 0. +# Note (Issue #63): stop_session.sh no longer calls this directly (it uses mam_session_ids). # --------------------------------------------------------------------------- capture_conversation_id() { local agent="$1" workdir="$2" session_name="${3:-}" find_workspace_uuid "$workdir" "$agent" "$session_name" } +# --------------------------------------------------------------------------- +# mam_session_ids [args...] +# Issue #63: stop/purge/resume conversation-id decision (lib_py.agents.session_ids). +# --------------------------------------------------------------------------- +mam_session_ids() { + env_python "$AGENT_SESSIONS_YAML" "$@" <<'PYEOF' +import sys +from lib_py.agents.session_ids 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/session_ids.py b/.agents/skills/lib_py/agents/session_ids.py new file mode 100644 index 0000000..44e5335 --- /dev/null +++ b/.agents/skills/lib_py/agents/session_ids.py @@ -0,0 +1,329 @@ +"""Stop, purge, and resume conversation-id decisions (Issue #63). + +Post-discovered agents (agy, codex, hermes, opencode) write their conversation +id to disk only after their first turn, and agy's workspace cache holds one id +per cwd. Stop, purge, and resume therefore use the row's own recorded id first +(never falling through to a workspace guess when a recorded id is present), and +otherwise use the #40 active-session capture when herdr is alive. Pre-assigned +agents (claude, grok) keep find_workspace_uuid with other rows' claimed ids +excluded. +""" +from __future__ import annotations + +import argparse +import os +import shlex +import sys +from typing import Any, Dict, List, Optional + +from lib_py.agents.autopin import OWN_KEYS, capture, claimed_elsewhere +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.workspace_uuid import find_workspace_uuid + + +def own_id(row: Optional[Dict[str, Any]], agent: str) -> str: + adapter = get_adapter(agent) + if not adapter or not isinstance(row, dict): + return "" + val = row.get(adapter.own_key) + return str(val).strip() if val else "" + + +def claiming_rows(row: Dict[str, Any], uuid: str, state: Dict[str, Any]) -> List[str]: + if not uuid: + return [] + me = row.get("name") + holders: List[str] = [] + for other in state.get("herdr_sessions") or []: + if other is row or (me and other.get("name") == me): + continue + if any(other.get(k) == uuid for k in OWN_KEYS): + other_name = str(other.get("name") or "") + if other_name not in holders: + holders.append(other_name) + return holders + + +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) + 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: + out.append(s) + return out + + +def row_for_capture(row: Optional[Dict[str, Any]], pane_pid: Optional[int], + cwd: str = "", name: str = "", agent: str = "") -> Dict[str, Any]: + copy = dict(row) if isinstance(row, dict) else {} + if name and not copy.get("name"): + copy["name"] = name + if agent and not copy.get("agent"): + copy["agent"] = agent + pane = dict(copy.get("pane") or {}) + if cwd and not pane.get("cwd"): + pane["cwd"] = cwd + if isinstance(pane_pid, int) and pane_pid > 0: + pane["pid"] = pane_pid + copy["pane"] = pane + return copy + + +def _find_row(state: Dict[str, Any], name: str) -> Optional[Dict[str, Any]]: + if not name: + return None + for s in state.get("herdr_sessions") or []: + if isinstance(s, dict) and s.get("name") == name: + return s + return None + + +def _parse_pid(raw: Any) -> Optional[int]: + if isinstance(raw, int): + return raw if raw > 0 else None + s = str(raw or "").strip() + return int(s) if s.isdigit() and int(s) > 0 else None + + +def decide_stop(state: Dict[str, Any], name: str, agent: str, cwd: str, + herdr_alive: bool, pane_pid: Optional[int] = None, + purge: bool = False, home: Optional[str] = None, + claude_dir: Optional[str] = None) -> Dict[str, 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) + 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: + if own in claimed: + others = ", ".join(claiming_rows(row_ref, own, state)) + return { + "captured": "", + "status": "sibling-claimed", + "purge_uuid": "", + "note": f"{own} is also recorded by {others}; registry needs fixing", + } + if adapter and adapter.id_lifecycle == POST_DISCOVERED: + if verify_session_uuid(cwd, agent, own, row, home_dir=home_resolved, + claude_dir=claude_resolved, mode="revalidate"): + return { + "captured": own, + "status": "already-set", + "purge_uuid": own if purge else "", + "note": "", + } + ctx = DiscoveryContext(workspace=cwd, agent_name=agent, home_dir=home_resolved, + claude_dir=claude_resolved, row=row, mode="revalidate") + can_purge = bool(purge and adapter.has_artifacts(own, ctx)) + return { + "captured": "", + "status": "unverified-own", + "purge_uuid": own if can_purge else "", + "note": f"recorded id {own} no longer verifies; kept as is", + } + + if adapter is None or adapter.id_lifecycle != POST_DISCOVERED: + found = find_workspace_uuid(cwd, agent, name, state, home_resolved, claude_resolved, + exclude_ids=claimed) if cwd else "" + if found and found in claimed: + others = ", ".join(claiming_rows(row_ref, found, state)) + return { + "captured": "", + "status": "sibling-claimed", + "purge_uuid": "", + "note": f"{found} is also recorded by {others}; registry needs fixing", + } + return { + "captured": found, + "status": "resolved" if found else "no-candidate", + "purge_uuid": found if purge else "", + "note": "" if found else "nothing on disk yet", + } + + if herdr_alive: + cap_row = row_for_capture(row, _parse_pid(pane_pid), cwd=cwd, name=name, agent=agent) + res = capture(cap_row, agent, state, home_dir=home_resolved) + if res.get("status") == "captured" and res.get("session_id"): + u = str(res["session_id"]) + if u in claimed: + others = ", ".join(claiming_rows(row_ref, u, state)) + return { + "captured": "", + "status": "sibling-claimed", + "purge_uuid": "", + "note": f"{u} is also recorded by {others}; registry needs fixing", + } + return { + "captured": u, + "status": "captured", + "purge_uuid": u if purge else "", + "note": "", + } + st = str(res.get("status") or "no-candidate") + reason = str(res.get("reason") or "") + return { + "captured": "", + "status": st, + "purge_uuid": "", + "note": f"{st}: {reason}" if reason else st, + } + + return { + "captured": "", + "status": "herdr-dead", + "purge_uuid": "", + "note": "herdr session was already gone, so the pane that owns the conversation could not be checked; no id recorded", + } + + +def decide_resume(state: Dict[str, Any], name: str, agent: str, workspace: str, + herdr_alive: bool = False, pane_pid: Optional[int] = None, + home: Optional[str] = None, + claude_dir: Optional[str] = None) -> Dict[str, Any]: + 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) + 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: + if own in claimed: + others = ", ".join(claiming_rows(row_ref, own, state)) + return { + "uuid": "", + "drop_stale_id": False, + "notes": [], + "error": (f"ERROR: {name}'s recorded id {own} is also recorded by {others}; " + "refusing to resume on a conversation another session claims (Issue #63). " + "One of these rows holds the wrong id - fix the registry before resuming."), + } + if adapter and adapter.id_lifecycle == POST_DISCOVERED: + if verify_session_uuid(workspace, agent, own, row, home_dir=home_resolved, + claude_dir=claude_resolved, mode="revalidate"): + return {"uuid": own, "drop_stale_id": False, "notes": [], "error": None} + if herdr_alive: + return {"uuid": own, "drop_stale_id": False, "notes": [], "error": None} + note = (f"NOTE: {name}'s recorded {agent} id {own} no longer verifies " + "(its conversation is missing or empty); starting a new conversation (Issue #63)") + return {"uuid": "", "drop_stale_id": True, "notes": [note], "error": None} + + if adapter is None or adapter.id_lifecycle != POST_DISCOVERED: + u = find_workspace_uuid(workspace, agent, name, state, home_resolved, claude_resolved, + exclude_ids=claimed) + if u and u in claimed: + others = ", ".join(claiming_rows(row_ref, u, state)) + return { + "uuid": "", + "drop_stale_id": False, + "notes": [], + "error": (f"ERROR: {name or agent}'s resolved id {u} is also recorded by {others}; " + "refusing to resume on a conversation another session claims (Issue #63)."), + } + return {"uuid": u, "drop_stale_id": False, "notes": [], "error": None} + + if herdr_alive: + cap_row = row_for_capture(row, _parse_pid(pane_pid), cwd=workspace, name=name, agent=agent) + res = capture(cap_row, agent, state, home_dir=home_resolved) + u = str(res["session_id"]) if res.get("status") == "captured" and res.get("session_id") else "" + else: + same = same_type_rows(state, workspace, agent) + n = len(same) + if name and all(r.get("name") != name for r in same): + n += 1 + if n >= 2: + label = name or agent + note = (f"NOTE: {label} has no recorded {agent} id and {n} {agent} sessions share " + f"{workspace}; the cache cannot tell them apart, so it starts a new conversation (Issue #63)") + return {"uuid": "", "drop_stale_id": False, "notes": [note], "error": None} + u = find_workspace_uuid(workspace, agent, name, state, home_resolved, claude_resolved, + exclude_ids=claimed) + + if u and u in claimed: + others = ", ".join(claiming_rows(row_ref, u, state)) + return { + "uuid": "", + "drop_stale_id": False, + "notes": [], + "error": (f"ERROR: {name or agent}'s resolved id {u} is also recorded by {others}; " + "refusing to resume on a conversation another session claims (Issue #63)."), + } + return {"uuid": u, "drop_stale_id": False, "notes": [], "error": None} + + +def main(argv: Optional[List[str]] = None) -> int: + p = argparse.ArgumentParser(prog="python -m lib_py.agents.session_ids") + sub = p.add_subparsers(dest="cmd", required=True) + + s_stop = sub.add_parser("stop") + s_stop.add_argument("--session", required=True) + s_stop.add_argument("--agent", required=True) + s_stop.add_argument("--cwd", default="") + s_stop.add_argument("--herdr-alive", default="0") + s_stop.add_argument("--pane-pid", default="") + s_stop.add_argument("--purge", default="0") + + s_res = sub.add_parser("resume") + s_res.add_argument("--session", default="") + s_res.add_argument("--agent", required=True) + s_res.add_argument("--workspace", required=True) + s_res.add_argument("--herdr-alive", default="0") + s_res.add_argument("--pane-pid", default="") + s_res.add_argument("--shell", action="store_true") + + a = p.parse_args(argv) + state = load_state_json() + q = shlex.quote + + if a.cmd == "stop": + res = decide_stop( + state=state, + name=a.session, + agent=a.agent, + cwd=a.cwd, + herdr_alive=(a.herdr_alive == "1"), + pane_pid=_parse_pid(a.pane_pid), + purge=(a.purge == "1"), + ) + print(f"CAPTURED_UUID={q(res['captured'])}") + print(f"CAPTURE_STATUS={q(res['status'])}") + print(f"PURGE_UUID={q(res['purge_uuid'])}") + print(f"STOP_ID_NOTE={q(res['note'])}") + return 0 + + res = decide_resume( + state=state, + name=a.session, + agent=a.agent, + workspace=a.workspace, + herdr_alive=(a.herdr_alive == "1"), + pane_pid=_parse_pid(a.pane_pid), + ) + for note in res.get("notes") or []: + print(note, file=sys.stderr) + if res.get("error"): + print(res["error"], file=sys.stderr) + return 1 + if a.shell: + print(f"RESUME_UUID={q(res['uuid'])}") + print(f"DROP_STALE_ID={'1' if res.get('drop_stale_id') else '0'}") + else: + print(res["uuid"]) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_issue63_same_cwd_ids.py b/tests/test_issue63_same_cwd_ids.py index 3e54ac0..9dbdfc0 100644 --- a/tests/test_issue63_same_cwd_ids.py +++ b/tests/test_issue63_same_cwd_ids.py @@ -145,3 +145,106 @@ def test_find_workspace_uuid_uses_explicit_home_and_claude_dir(tmp_path, monkeyp home=str(home), claude_dir=str(claude_dir)) == UA assert find_workspace_uuid(str(ws), "claude", "ws-creator-claude", state, home=str(home), claude_dir=str(claude_dir)) == UB + + +# --------------------------------------------------------------------------- 2.2 session_ids unit tests + +@pytest.fixture +def owners(monkeypatch): + """agy presence ownership without lsof/ps: {uuid: [holder pids]} and a + pid -> ppid table the test fills in.""" + from lib_py.agents.adapters import agy + held, parents = {}, {} + + def holders(paths): + if held.get("__lsof__") is None and "__lsof__" in held: + return None + return sorted({p for path in paths for u, pids in held.items() if u in path for p in pids}) + + monkeypatch.setattr(agy, "_holder_pids", holders) + monkeypatch.setattr(agy, "_process_parents", lambda: dict(parents) or None) + return held, parents + + +def test_session_ids_decide_stop_and_resume_unit(tmp_path, owners): + from lib_py.agents.session_ids import decide_resume, decide_stop + + home = tmp_path / "home" + ws = tmp_path / "ws" + ws.mkdir(parents=True) + now = time.time() + epoch = int(now) - 100 + _agy_conv(home, UA, mtime=now - 5) + _agy_conv(home, UB, mtime=now - 2) + _agy_cache(home, {ws: UA}) + held, parents = owners + held[UA] = [5001] + held[UB] = [6001] + parents.update({5001: 1, 6001: 1}) + + state = { + "herdr_sessions": [ + _row("ws-creator-agy", "agy", ws, epoch, pid=5001), + _row("ws-reviewer-agy", "agy", ws, epoch, pid=6001), + ] + } + res_a = decide_stop(state, "ws-creator-agy", "agy", str(ws), True, pane_pid=5001, purge=False, home=str(home)) + assert res_a["captured"] == UA and res_a["status"] == "captured" + + res_b = decide_stop(state, "ws-reviewer-agy", "agy", str(ws), True, pane_pid=6001, purge=False, home=str(home)) + assert res_b["captured"] == "" and res_b["status"] == "owner-mismatch" + + res_dead = decide_stop(state, "ws-reviewer-agy", "agy", str(ws), False, pane_pid=None, purge=True, home=str(home)) + assert res_dead["captured"] == "" and res_dead["purge_uuid"] == "" + assert "herdr session was already gone" in res_dead["note"] + + # Two same-type rows in one cwd and B has no own id -> fresh spawn with NOTE + state["herdr_sessions"][0]["status"] = "stopped" + state["herdr_sessions"][0]["agy_conversation_id_own"] = UA + state["herdr_sessions"][1]["status"] = "stopped" + r_b = decide_resume(state, "ws-reviewer-agy", "agy", str(ws), herdr_alive=False, home=str(home)) + assert r_b["uuid"] == "" and r_b["error"] is None and not r_b["drop_stale_id"] + assert any("2 agy sessions share" in n for n in r_b["notes"]) + + +def test_session_ids_own_id_priority_rules_unit(tmp_path): + """rev 2 (A): own id never falls through to tier-2; broken registry refuses even when DB is missing.""" + from lib_py.agents.session_ids import decide_resume, decide_stop + + home = tmp_path / "home" + ws = tmp_path / "ws" + ws.mkdir(parents=True) + epoch = int(time.time()) - 100 + # Missing UA.db on disk, but both rows claim UA + state_dup = { + "herdr_sessions": [ + _row("ws-creator-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA, + session_id_source="discovered", session_id_verified=True), + _row("ws-reviewer-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UA, + session_id_source="discovered", session_id_verified=True), + ] + } + r_dup = decide_resume(state_dup, "ws-reviewer-agy", "agy", str(ws), herdr_alive=False, home=str(home)) + assert r_dup["error"] and "ws-creator-agy" in r_dup["error"] + assert r_dup["notes"] == [] + + s_dup = decide_stop(state_dup, "ws-reviewer-agy", "agy", str(ws), False, purge=True, home=str(home)) + assert s_dup["captured"] == "" and s_dup["purge_uuid"] == "" + assert "also recorded by ws-creator-agy" in s_dup["note"] + + # Single row with stale own id UX (missing UX.db) while cache has UC + _agy_conv(home, UC) + _agy_cache(home, {ws: UC}) + state_stale = { + "herdr_sessions": [ + _row("ws-creator-agy", "agy", ws, epoch, status="stopped", agy_conversation_id_own=UX, + session_id_source="discovered", session_id_verified=True), + ] + } + r_stale = decide_resume(state_stale, "ws-creator-agy", "agy", str(ws), herdr_alive=False, home=str(home)) + assert r_stale["uuid"] == "" and r_stale["drop_stale_id"] is True and r_stale["error"] is None + assert any(f"recorded agy id {UX} no longer verifies" in n for n in r_stale["notes"]) + + s_stale = decide_stop(state_stale, "ws-creator-agy", "agy", str(ws), False, purge=False, home=str(home)) + assert s_stale["captured"] == "" and "no longer verifies; kept as is" in s_stale["note"] +