From 0ecfbc989005cb3eccdbb8351a2510ac7001450d Mon Sep 17 00:00:00 2001
From: Benn Denton <80194363+TinkerOfThings@users.noreply.github.com>
Date: Sun, 21 Jun 2026 08:48:21 +0000
Subject: [PATCH] feat(pty): RingBuffer for keep-alive output capture

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 hermes_cli/pty_session.py | 32 ++++++++++++++++++++++++++++++++
 tests/test_pty_session.py | 24 ++++++++++++++++++++++++
 2 files changed, 56 insertions(+)
 create mode 100644 hermes_cli/pty_session.py
 create mode 100644 tests/test_pty_session.py

diff --git a/hermes_cli/pty_session.py b/hermes_cli/pty_session.py
new file mode 100644
index 00000000000..59cede640a0
--- /dev/null
+++ b/hermes_cli/pty_session.py
@@ -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
diff --git a/tests/test_pty_session.py b/tests/test_pty_session.py
new file mode 100644
index 00000000000..1ab5bacdbaa
--- /dev/null
+++ b/tests/test_pty_session.py
@@ -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