fix(cli): concurrent PTY reaps skip already-popped keys

reap_idle() is reached concurrently from attach_or_spawn() and the
background run_reaper loop, and each pop is followed by an awaited
close(); while one reap is parked inside close(), a second reap reaching
the same doomed key hit dict.pop(key) without a default and raised
KeyError straight into the websocket handler. close_all() has the same
overlap window against an in-flight reap.

Pop with a default and skip keys a concurrent reap already removed, so
overlapping reaps are idempotent. Which sessions get reaped is
unchanged.

Fixes #106400

(cherry picked from commit 8ea63a845424263551487886a7526ca2f4bbd1ee)

hermes_cli/pty_session.py

@@ -191,7 +191,12 @@ class PtySessionRegistry:
             if not s.alive or (not s.attached and s.last_detached_at is not None and (now - s.last_detached_at) > self._ttl)
         ]
         for key in doomed:
-            await self._sessions.pop(key).close()
+            # Reaps overlap (attach_or_spawn and the background reaper) and close()
+            # awaits, so a concurrent reap can have popped this key already — skip
+            # it instead of raising KeyError into the websocket handler.
+            session = self._sessions.pop(key, None)
+            if session is not None:
+                await session.close()
 
     def _reap_one_idle_or_raise(self) -> None:
         idle = [s for s in self._sessions.values() if not s.attached and s.last_detached_at is not None]
@@ -203,4 +208,8 @@ class PtySessionRegistry:
 
     async def close_all(self) -> None:
         for key in list(self._sessions):
-            await self._sessions.pop(key).close()
+            # Same overlap window as reap_idle: an in-flight reap may have popped
+            # a snapshot key while we awaited an earlier close().
+            session = self._sessions.pop(key, None)
+            if session is not None:
+                await session.close()

tests/test_pty_session.py

@@ -277,3 +277,104 @@ async def test_reaper_loop_invokes_reap(monkeypatch):
     except asyncio.CancelledError:
         pass
     assert calls["n"] >= 2
+
+
+def _register_idle_session(reg, key):
+    from hermes_cli.pty_session import PtySession
+    bridge = FakeBridge([b""])
+    s = PtySession(key, bridge, buffer_cap=1024, read_timeout=0.01)
+    await_start = s.start()
+    return bridge, s, await_start
+
+
+@pytest.mark.asyncio
+async def test_concurrent_reap_idle_is_idempotent():
+    """reap_idle is reached from attach_or_spawn and the run_reaper loop; both
+    may doom the same keys, so one reap can pop a key the other already took
+    while awaiting its close(). The second pop must skip, not raise."""
+    reg = make_registry(ttl=60.0)
+    bridges = []
+    sessions = []
+    for i in range(2):
+        bridge, s, start = _register_idle_session(reg, "k%d" % i)
+        await start
+        s.detach(None)                        # unattached, last_detached_at set
+        reg._sessions[s.key] = s
+        bridges.append(bridge)
+        sessions.append(s)
+
+    entered, release = asyncio.Event(), asyncio.Event()
+    original_close = sessions[0].close
+
+    async def gated_close():
+        entered.set()
+        await release.wait()
+        await original_close()
+
+    sessions[0].close = gated_close
+
+    far_future = time.monotonic() + 10_000    # both idle past ttl → doomed
+    first = asyncio.create_task(reg.reap_idle(now=far_future))
+    await entered.wait()                      # k0 popped; first reap parked in close()
+    await reg.reap_idle(now=far_future)       # second reap hits the taken k0
+
+    release.set()
+    await first
+    assert not reg._sessions
+    assert all(b.closed for b in bridges)
+
+
+@pytest.mark.asyncio
+async def test_close_all_survives_key_popped_by_concurrent_reap():
+    """close_all snapshots keys, then awaits each close(); a reap that runs
+    during that await can remove a later key from the snapshot."""
+    reg = make_registry(ttl=60.0)
+    bridges = []
+    sessions = []
+    for i in range(2):
+        bridge, s, start = _register_idle_session(reg, "k%d" % i)
+        await start
+        s.detach(None)
+        reg._sessions[s.key] = s
+        bridges.append(bridge)
+        sessions.append(s)
+
+    entered, release = asyncio.Event(), asyncio.Event()
+    original_close = sessions[0].close
+
+    async def gated_close():
+        entered.set()
+        await release.wait()
+        await original_close()
+
+    sessions[0].close = gated_close
+
+    closer = asyncio.create_task(reg.close_all())
+    await entered.wait()                      # close_all popped k0, parked in close()
+    await reg.reap_idle(now=time.monotonic() + 10_000)   # pops k1 meanwhile
+    release.set()
+    await closer                              # k1 of the snapshot is already gone
+
+    assert not reg._sessions
+    assert all(b.closed for b in bridges)
+
+
+@pytest.mark.asyncio
+async def test_reap_idle_closes_doomed_and_keeps_attached():
+    reg = make_registry(ttl=60.0)
+    live_bridge = FakeBridge([b""])
+    s_live, _ = await reg.attach_or_spawn("live", spawn=lambda: live_bridge)
+    await s_live.attach(FakeWS())             # attached → survives the reap
+
+    dead_bridge = FakeBridge([b""])
+    s_dead, _ = await reg.attach_or_spawn("dead", spawn=lambda: dead_bridge)
+    s_dead.alive = False                      # dead remnant → reaped
+
+    idle_bridge = FakeBridge([b""])
+    s_idle, _ = await reg.attach_or_spawn("idle", spawn=lambda: idle_bridge)
+    s_idle.detach(None)                       # idle past ttl → reaped
+
+    await reg.reap_idle(now=time.monotonic() + 10_000)
+    assert list(reg._sessions) == ["live"]
+    assert dead_bridge.closed and idle_bridge.closed and not live_bridge.closed
+    await reg.close_all()