fix(id_order): a dead-owner marker on the caller's own session follows the orphan rules (Issue #63)

A marker left on the invoking session name by a killed create/resume was
pruned before herdr was asked, so a retry under the same name took it over
while the orphaned pane was still alive (and create's failure cleanup could
then kill it). It is now kept and blocks like any other orphan until its
row is registered for the attempt or herdr reports the session gone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.agents/skills/lib_py/agents/id_order.py

@@ -180,15 +180,16 @@ def _registered_for_attempt(m: Dict[str, Any], row: Optional[Dict[str, Any]]) ->
     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],
+def _read_and_prune_markers(dpath: str, my_pid: int, state: Dict[str, Any],
                             is_alive: Callable[[int], bool] = pid_alive,
                             pane_state: Callable[[str, str], Optional[bool]] = (
                                 lambda server, session: None),
                             ) -> 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 herdr answers
-    that its session is gone; until then it is kept as an orphan."""
+    attempt on the caller's session included). A dead owner's marker goes once
+    its row is registered for the attempt (the row decides from then on) or
+    herdr answers that its session is gone; until then it is kept as an orphan,
+    one an earlier attempt left on the caller's session included."""
     rows_by_name = {
         str(s.get("name") or ""): s
         for s in (state.get("herdr_sessions") or []) if isinstance(s, dict)
@@ -219,7 +220,7 @@ def _read_and_prune_markers(dpath: str, me: str, my_pid: int, state: Dict[str, A
             drop = True
         elif is_alive(owner_pid):
             drop = False
-        elif sess == me or not sess:
+        elif not sess:
             drop = True
         elif _registered_for_attempt(data, rows_by_name.get(sess)):
             drop = True
@@ -304,7 +305,7 @@ def wait(cwd: str, agent: str, session: str, owner_pid: int,
                         file=sys.stderr, flush=True,
                     )
                     return 1
-            markers = _read_and_prune_markers(dpath, session, owner_pid, state,
+            markers = _read_and_prune_markers(dpath, owner_pid, state,
                                               is_alive=is_alive, pane_state=bounded_probe)
             blks = blockers(state, cwd, agent, session, markers, is_alive=is_alive)
             if not blks:

tests/test_issue63_same_cwd_ids.py

@@ -1050,8 +1050,9 @@ def test_id_order_dead_owner_marker_blocks_while_its_unregistered_herdr_session_
     """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)."""
+    decides) or the session is gone; on the same session name a live attempt is
+    not pruned and a dead one follows the orphan rules (PR #69 review r3); a
+    running row without a pane pid blocks (fail-closed)."""
     from lib_py.agents import id_order
 
     home = tmp_path / "home"
@@ -1116,16 +1117,33 @@ def test_id_order_dead_owner_marker_blocks_while_its_unregistered_herdr_session_
         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 session name: a live attempt is kept and blocks; a dead one is kept and blocks
+        # while its herdr session is alive or herdr cannot tell, and is taken over once gone
         same = Path(dpath) / "same-agy.json"
+
+        def wait_same():
+            return id_order.wait(str(ws), "agy", "same-agy", os.getpid(), yaml_path=str(yaml_path),
+                                 home=str(home), wait_s_val=0, pane_state=pane_state)
+
         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_state=pane_state) == 1
+        assert wait_same() == 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_state=pane_state) == 0
+        alive_sessions.add("same-agy")
+        probes.clear()
+        capsys.readouterr()
+        assert wait_same() == 1
+        assert probes == [("", "same-agy")]
+        assert json.loads(same.read_text())["owner_pid"] == live.pid
+        err = capsys.readouterr().err
+        assert "its herdr session is alive" in err and "kill-session -t same-agy" in err
+        unknown_sessions.add("same-agy")
+        assert wait_same() == 1
+        assert json.loads(same.read_text())["owner_pid"] == live.pid
+        unknown_sessions.clear()
+        alive_sessions.clear()
+        assert wait_same() == 0
         assert json.loads(same.read_text())["owner_pid"] == os.getpid()
         same.unlink()
 
@@ -1233,7 +1251,8 @@ def test_id_order_real_probe_against_mock_herdr_fails_closed(mam_sandbox, mock_h
     """PR #69 review r2 P1: lib.sh mam_id_order_wait with the real probe (sandbox
     copy, mock herdr). A listed orphan blocks; an unlisted one is pruned; a failed
     `agent list` (server_not_running included) keeps the marker and refuses;
-    with MAM_ID_ORDER_WAIT_S=0 herdr is not asked and the marker is kept."""
+    with MAM_ID_ORDER_WAIT_S=0 herdr is not asked and the marker is kept. A
+    retry on the orphan's own session name is held the same way (review r3)."""
     from lib_py.agents import id_order
 
     ws = mam_sandbox / "probe-ws"
@@ -1253,12 +1272,12 @@ def test_id_order_real_probe_against_mock_herdr_fails_closed(mam_sandbox, mock_h
         state.update(kw)
         Path(mock_herdr).write_text(json.dumps(state))
 
-    def wait_b(cap="2"):
+    def wait_b(cap="2", who=name_b):
         for f in Path(dpath).glob(f"{name_b}.json"):
             f.unlink()
         id_order.write_marker(dpath, name_a, dead.pid, str(ws), "agy", herdr_server="probe-ws")
         return subprocess.run(
-            ["bash", "-c", 'source "$1" && mam_id_order_wait "$2" agy "$3"', "_", str(lib), str(ws), name_b],
+            ["bash", "-c", 'source "$1" && mam_id_order_wait "$2" agy "$3"', "_", str(lib), str(ws), who],
             capture_output=True, text=True, cwd=str(mam_sandbox), timeout=120,
             env=dict(os.environ, MAM_ID_ORDER_WAIT_S=cap, HERDR_SESSION_NAME="probe-ws"),
         )
@@ -1268,6 +1287,9 @@ def test_id_order_real_probe_against_mock_herdr_fails_closed(mam_sandbox, mock_h
         res = wait_b()
         assert res.returncode == 1 and "its herdr session is alive" in res.stderr, res.stderr
         assert marker_a.exists()
+        res = wait_b(who=name_a)
+        assert res.returncode == 1 and "its herdr session is alive" in res.stderr, res.stderr
+        assert json.loads(marker_a.read_text())["owner_pid"] == dead.pid
 
         set_mock(force_error_code="protocol_mismatch")
         res = wait_b()
@@ -1290,6 +1312,9 @@ def test_id_order_real_probe_against_mock_herdr_fails_closed(mam_sandbox, mock_h
         res = wait_b()
         assert res.returncode == 0, res.stderr
         assert not marker_a.exists()
+        res = wait_b(who=name_a)
+        assert res.returncode == 0, res.stderr
+        assert json.loads(marker_a.read_text())["owner_pid"] != dead.pid
     finally:
         holder.kill()
         holder.wait()