import asyncio import time import pytest from hermes_cli.pty_session import RingBuffer def test_ringbuffer_keeps_everything_under_capacity(): rb = RingBuffer(10) rb.append(b"abc") rb.append(b"def") assert rb.snapshot() == b"abcdef" assert rb.truncated is False def test_ringbuffer_drops_oldest_over_capacity(): rb = RingBuffer(4) rb.append(b"abcdef") # 6 bytes into a 4-byte buffer assert rb.snapshot() == b"cdef" assert rb.truncated is True class FakeBridge: """Implements the bridge contract PtySession depends on.""" def __init__(self, chunks, *, write_result=True): self._chunks = list(chunks) # bytes; b"" = idle tick; None = EOF self.written = bytearray() self.write_result = write_result self.closed = False self.resized = None self.dead = False def read(self, timeout): if not self._chunks: return b"" # idle return self._chunks.pop(0) async def write(self, data): if self.write_result: self.written.extend(data) return self.write_result def resize(self, cols, rows): self.resized = (cols, rows) def close(self): self.closed = True def is_alive(self): return not self.dead class FakeWS: def __init__(self): self.sent = [] # list of ("bytes"|"text", payload) self.close_code = None async def send_bytes(self, data): self.sent.append(("bytes", bytes(data))) async def send_text(self, text): self.sent.append(("text", text)) async def close(self, code=1000, reason=""): self.close_code = code class FailingWS(FakeWS): """A socket whose client already dropped: every send raises, optionally after a hold.""" def __init__(self, *, hold=False): super().__init__() self.send_started = asyncio.Event() self.release = asyncio.Event() self._hold = hold async def send_bytes(self, data): self.send_started.set() if self._hold: await self.release.wait() raise RuntimeError("socket closed") @pytest.mark.asyncio async def test_attach_replays_buffer_then_streams_live(): from hermes_cli.pty_session import PtySession bridge = FakeBridge([b"hello ", b"world", None]) s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01) await s.start() await asyncio.sleep(0.05) # drain consumes "hello world" ws = FakeWS() await s.attach(ws) replay = b"".join(p for kind, p in ws.sent if kind == "bytes") assert replay == b"hello world" await s.close() @pytest.mark.asyncio async def test_failed_replay_detaches_session_so_reaper_reclaims_it(): """A client dropping mid-replay must not pin the PTY as attached forever (#110849).""" from hermes_cli.pty_session import PtySessionRegistry reg = PtySessionRegistry(ttl=0.0, max_sessions=2, buffer_cap=1024, read_timeout=0.01) bridge = FakeBridge([b""]) s, _ = await reg.attach_or_spawn("k", spawn=lambda: bridge) s.buffer.append(b"replay") assert await s.attach(FailingWS()) is False assert s.attached is False assert s.last_detached_at is not None await reg.reap_idle(now=time.monotonic() + 1.0) assert "k" not in reg._sessions assert bridge.closed @pytest.mark.asyncio async def test_drain_send_failure_detaches_current_socket_but_not_a_replacement(): from hermes_cli.pty_session import PtySession s = PtySession("k", FakeBridge([b"live"]), buffer_cap=1024, read_timeout=0.01) await s.start() ws = FailingWS() await s.attach(ws) await asyncio.wait_for(ws.send_started.wait(), timeout=1) await asyncio.sleep(0) assert s.attached is False assert s.last_detached_at is not None await s.close() # The failure of a socket superseded mid-send must leave the new viewer attached. s = PtySession("k", FakeBridge([b"live"]), buffer_cap=1024, read_timeout=0.01) await s.start() stale = FailingWS(hold=True) await s.attach(stale) await asyncio.wait_for(stale.send_started.wait(), timeout=1) replacement = FakeWS() assert await s.attach(replacement) is True stale.release.set() await asyncio.sleep(0.05) assert s.attached is True assert s._ws is replacement assert s.last_detached_at is None await s.close() @pytest.mark.asyncio async def test_reattach_can_force_complete_tui_redraw_after_replay(): """A fresh terminal cannot reconstruct a differential ANSI tail alone.""" from hermes_cli.pty_session import PtySession bridge = FakeBridge([b"partial differential frame", b""]) s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01) await s.start() await asyncio.sleep(0.05) ws = FakeWS() assert await s.attach(ws, force_redraw=True) is True replay = b"".join(p for kind, p in ws.sent if kind == "bytes") assert replay == b"partial differential frame" assert bytes(bridge.written) == b"\x0c" await s.close() @pytest.mark.asyncio async def test_failed_redraw_marks_session_dead_for_replacement(): from hermes_cli.pty_session import PtySession bridge = FakeBridge([b""], write_result=False) s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01) await s.start() ws = FakeWS() assert await s.attach(ws, force_redraw=True) is False assert s.alive is False await s.close() @pytest.mark.asyncio async def test_session_serializes_input_across_socket_tasks(): from hermes_cli.pty_session import PtySession class OrderedBridge(FakeBridge): def __init__(self): super().__init__([b""]) self.first_started = asyncio.Event() self.release_first = asyncio.Event() async def write(self, data): if not self.written: self.first_started.set() await self.release_first.wait() self.written.extend(data) return True bridge = OrderedBridge() s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01) await s.start() ws = FakeWS() await s.attach(ws) first = asyncio.create_task(s.write(ws, b"first")) await bridge.first_started.wait() second = asyncio.create_task(s.write(ws, b"second")) await asyncio.sleep(0) assert bytes(bridge.written) == b"" bridge.release_first.set() assert await first is True assert await second is True assert bytes(bridge.written) == b"firstsecond" await s.close() @pytest.mark.asyncio async def test_superseded_failed_write_does_not_kill_replacement_session(): from hermes_cli.pty_session import PtySession class SupersededBridge(FakeBridge): def __init__(self): super().__init__([b""]) self.old_write_started = asyncio.Event() self.release_old_write = asyncio.Event() self.calls = 0 async def write(self, data): self.calls += 1 if self.calls == 1: self.old_write_started.set() await self.release_old_write.wait() return False self.written.extend(data) return True bridge = SupersededBridge() s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01) await s.start() old_ws = FakeWS() new_ws = FakeWS() await s.attach(old_ws) old_write = asyncio.create_task(s.write(old_ws, b"old input")) await bridge.old_write_started.wait() new_attach = asyncio.create_task(s.attach(new_ws, force_redraw=True)) for _ in range(10): if s._ws is new_ws: break await asyncio.sleep(0) assert s._ws is new_ws bridge.release_old_write.set() assert await old_write is False assert await new_attach is True assert s.alive is True assert await s.write(new_ws, b"new input") is True assert bytes(bridge.written) == b"\x0cnew input" await s.close() @pytest.mark.asyncio async def test_detach_keeps_draining_into_buffer(): from hermes_cli.pty_session import PtySession bridge = FakeBridge([b"one", b"", b"two"]) s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01) await s.start() ws = FakeWS() await s.attach(ws) s.detach(ws) assert s.attached is False assert s.last_detached_at is not None await asyncio.sleep(0.05) # "two" drains while detached ws2 = FakeWS() await s.attach(ws2) replay = b"".join(p for kind, p in ws2.sent if kind == "bytes") assert replay == b"onetwo" await s.close() @pytest.mark.asyncio async def test_eof_marks_dead_and_closes_socket_4410(): from hermes_cli.pty_session import PtySession bridge = FakeBridge([b"bye", None]) s = PtySession("k", bridge, buffer_cap=1024, read_timeout=0.01) await s.start() ws = FakeWS() await s.attach(ws) await asyncio.sleep(0.05) # drain hits None (EOF) assert s.alive is False assert ws.close_code == 4410 await s.close() from hermes_cli.pty_session import PtySessionRegistry, RegistryFull def make_registry(ttl=1800.0, max_sessions=16): return PtySessionRegistry(ttl=ttl, max_sessions=max_sessions, buffer_cap=1024, read_timeout=0.01) @pytest.mark.asyncio async def test_same_key_reattaches_same_session(): reg = make_registry() b1 = FakeBridge([b"", b"", b""]) s1, created1 = await reg.attach_or_spawn("tok", spawn=lambda: b1) s2, created2 = await reg.attach_or_spawn("tok", spawn=lambda: FakeBridge([])) assert created1 is True and created2 is False assert s1 is s2 assert s2.bridge is b1 # second spawn callable was NOT used await reg.close_all() @pytest.mark.asyncio async def test_new_key_at_capacity_raises_when_none_reapable(): reg = make_registry(max_sessions=1) b = FakeBridge([b"", b""]) s, _ = await reg.attach_or_spawn("a", spawn=lambda: b) await s.attach(FakeWS()) # attached → not reapable with pytest.raises(RegistryFull): await reg.attach_or_spawn("b", spawn=lambda: FakeBridge([])) await reg.close_all() @pytest.mark.asyncio async def test_concurrent_attach_on_one_token_forks_one_pty(): """Two connections racing one attach token must share ONE registered PTY. The get-or-spawn decision spans awaits (reap, the spawn thread, start()), so both racing callers used to see "no session" and fork their own: the token then mapped to whichever registered last, the other tab's live session fell out of the registry (never reaped) and a reattach landed on the wrong terminal (#115304). """ from hermes_cli.pty_session import WS_CLOSE_SUPERSEDED reg = make_registry() spawned = [] def spawn(): bridge = FakeBridge([b"", b""]) spawned.append(bridge) return bridge (s1, created1), (s2, created2) = await asyncio.gather( reg.attach_or_spawn("tok", spawn=spawn), reg.attach_or_spawn("tok", spawn=spawn), ) assert len(spawned) == 1 # one token, one PTY assert (s1, created1) == (s2, True) assert created2 is False assert s1.bridge is spawned[0] assert list(reg._sessions.values()) == [s1] # every handed-out session is tracked # Whichever socket attached last owns the terminal; the loser is superseded # by contract, so no viewer is left writing into an untracked PTY. ws_a, ws_b = FakeWS(), FakeWS() await s1.attach(ws_a) await s2.attach(ws_b) assert reg._sessions["tok"] is s1 assert s1._ws is ws_b and ws_b.close_code is None assert ws_a.close_code == WS_CLOSE_SUPERSEDED await reg.close_all() async def _two_idle_sessions_first_close_gated(reg): """Two detached (idle) sessions; k0's close() parks until ``release`` is set.""" from hermes_cli.pty_session import PtySession bridges = [] for i in range(2): bridge = FakeBridge([b""]) s = PtySession("k%d" % i, bridge, buffer_cap=1024, read_timeout=0.01) await s.start() s.detach(None) # unattached, last_detached_at set reg._sessions[s.key] = s bridges.append(bridge) entered, release = asyncio.Event(), asyncio.Event() k0 = reg._sessions["k0"] original_close = k0.close async def gated_close(): entered.set() await release.wait() await original_close() k0.close = gated_close return bridges, entered, release @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, entered, release = await _two_idle_sessions_first_close_gated(reg) 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 takes k1 release.set() await first # first reap reaches the taken k1 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, entered, release = await _two_idle_sessions_first_close_gated(reg) 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_close_all_does_not_wait_on_one_slow_close_before_the_next(): """Each close() can wait out its helpers' SIGHUP grace, and the backend's teardown runs under a SIGKILL budget, so sessions close concurrently rather than one after another.""" reg = make_registry(ttl=60.0) bridges, entered, release = await _two_idle_sessions_first_close_gated(reg) closer = asyncio.create_task(reg.close_all()) await entered.wait() # k0's close() is parked for _ in range(50): if bridges[1].closed: break await asyncio.sleep(0.01) assert bridges[1].closed # k1 closed while k0 still waits release.set() await closer assert all(b.closed for b in bridges) @pytest.mark.asyncio async def test_close_all_waits_for_closes_already_running_in_the_background(): """A dead remnant or evicted session leaves the registry before its close() finishes; the backend's teardown must still wait for it, or its helpers outlive the backend.""" reg = make_registry(ttl=60.0) bridges, entered, release = await _two_idle_sessions_first_close_gated(reg) reg._close_in_background(reg._sessions.pop("k0")) await entered.wait() # k0 is closing outside the registry closer = asyncio.create_task(reg.close_all()) await asyncio.sleep(0.05) assert not closer.done() # still waiting on k0 release.set() await closer assert all(b.closed for b in bridges) @pytest.mark.asyncio async def test_close_other_sessions_removes_old_profile_session(): from hermes_cli.pty_session import WS_CLOSE_SUPERSEDED, PtySession reg = make_registry() old_bridge = FakeBridge([b""]) current_bridge = FakeBridge([b""]) old = PtySession("token\0alpha\0session-a", old_bridge, buffer_cap=1024, read_timeout=0.01) current = PtySession("token\0beta\0session-b", current_bridge, buffer_cap=1024, read_timeout=0.01) await old.start() await current.start() # A sibling tab (same attach token, other profile) is still viewing the old PTY. old_ws = FakeWS() assert await old.attach(old_ws) reg._sessions[old.key] = old reg._sessions[current.key] = current await reg.close_other_sessions("token", keep_key=current.key) assert old_bridge.closed assert old.key not in reg._sessions assert reg._sessions[current.key] is current # The displaced viewer gets the documented supersede code rather than going silent. assert old_ws.close_code == WS_CLOSE_SUPERSEDED await reg.close_all() @pytest.mark.asyncio async def test_reap_reaps_dead_process_even_when_attached(): # Child killed externally (OOM killer / cgroup SIGKILL) while a grandchild # holds the PTY slave, so the drain never sees EOF: the session must still # be reaped, otherwise its bridge and registry slot leak forever (#76759). reg = make_registry(ttl=3600.0) b = FakeBridge([b"", b"", b""]) s, _ = await reg.attach_or_spawn("tok", spawn=lambda: b) await s.attach(FakeWS()) b.dead = True await reg.reap_idle() assert "tok" not in reg._sessions assert b.closed is True await reg.close_all() @pytest.mark.asyncio async def test_reap_keeps_live_attached_session(): # A session with a live child and an attached viewer is never reaped, # even past the detached TTL. reg = make_registry(ttl=1.0) b = FakeBridge([b"", b""]) s, _ = await reg.attach_or_spawn("tok", spawn=lambda: b) await s.attach(FakeWS()) await reg.reap_idle(now=time.monotonic() + 10) assert "tok" in reg._sessions assert b.closed is False await reg.close_all()