feat(pty): RingBuffer for keep-alive output capture

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Benn Denton 2026-06-21 08:48:21 +00:00 committed by Teknium
parent d5a5ea8640
commit 0ecfbc9890
2 changed files with 56 additions and 0 deletions

32
hermes_cli/pty_session.py Normal file
View file

@ -0,0 +1,32 @@
"""Keep-alive PTY sessions for dashboard terminals.
A PTY process outlives the WebSocket that created it: a single drain task
always reads the PTY into a bounded RingBuffer and forwards to the attached
socket when present. Reconnecting with the same opaque token replays the
buffer and resumes live. See
docs/superpowers/specs/2026-06-20-pty-keepalive-reattach-design.md.
"""
from __future__ import annotations
class RingBuffer:
"""Keeps only the most recent ``capacity`` bytes appended to it."""
def __init__(self, capacity: int) -> None:
self._cap = capacity
self._buf = bytearray()
self._truncated = False
def append(self, data: bytes) -> None:
self._buf.extend(data)
overflow = len(self._buf) - self._cap
if overflow > 0:
del self._buf[:overflow]
self._truncated = True
def snapshot(self) -> bytes:
return bytes(self._buf)
@property
def truncated(self) -> bool:
return self._truncated

24
tests/test_pty_session.py Normal file
View file

@ -0,0 +1,24 @@
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
def test_ringbuffer_truncation_across_appends():
rb = RingBuffer(3)
rb.append(b"ab")
rb.append(b"cd") # now "abcd" -> keep "bcd"
assert rb.snapshot() == b"bcd"
assert rb.truncated is True