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

@@ -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):
+            continue
-            status = row.get("status")
+        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)
-            if status == "running" or end_epoch <= m_at:
+        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")
-    fd, tmp = tempfile.mkstemp(dir=dpath, prefix=".marker.", suffix=".tmp")
+    payload: Dict[str, Any] = {
-    with os.fdopen(fd, "w", encoding="utf-8") as f:
-        json.dump({
         "session": session,
         "owner_pid": int(owner_pid),
-            "cwd": os.path.realpath(cwd),
+        "cwd": canonical_cwd(cwd),
         "agent": agent,
         "at": int(time.time()),
-        }, f)
+    }
+    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(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)

.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:

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"] == []