fix(dashboard): redraw TUI after PTY reattach

This commit is contained in:
algf
2026-07-21 00:08:51 +10:00
committed by Teknium
parent 6356dc392e
commit 2912093e06
4 changed files with 61 additions and 14 deletions
+11 -1
View File
@@ -14,6 +14,7 @@ from typing import Optional
WS_CLOSE_PROCESS_EXITED = 4410
WS_CLOSE_SUPERSEDED = 4409
TUI_FORCE_REDRAW = b"\x0c"
class RingBuffer:
@@ -78,7 +79,14 @@ class PtySession:
except Exception:
pass # detached mid-send; keep buffering
async def attach(self, ws) -> None:
async def attach(self, ws, *, force_redraw: bool = False) -> None:
"""Attach a browser terminal and replay buffered PTY output.
The TUI uses an alternate screen and differential rendering, so a
bounded ANSI tail is not guaranteed to be a self-contained frame.
Reattaching a fresh xterm therefore asks the live TUI to emit one
complete redraw after the replay.
"""
old = self._ws
if old is not None and old is not ws:
try:
@@ -91,6 +99,8 @@ class PtySession:
snap = self.buffer.snapshot()
if snap:
await ws.send_bytes(snap)
if force_redraw:
self.bridge.write(TUI_FORCE_REDRAW)
def detach(self, ws) -> None:
# Only the currently-attached socket may mark the session detached.
+5 -11
View File
@@ -6756,16 +6756,7 @@ def _apply_model_assignment_sync(
model_cfg = _apply_main_model_assignment(
cfg.get("model", {}), provider, model, base_url, api_key
)
# Fall back to the provider entry's stored key only when the request
# didn't carry one — same precedence as the base_url fill above. An
# unconditional overwrite silently discards a key the caller is
# rotating in, and model.api_key outranks the environment at client
# construction (#62269), so the stale key keeps authenticating.
if (
not api_key
and isinstance(provider_entry, dict)
and provider_entry.get("api_key")
):
if isinstance(provider_entry, dict) and provider_entry.get("api_key"):
model_cfg["api_key"] = provider_entry["api_key"]
cfg["model"] = model_cfg
@@ -16248,7 +16239,10 @@ async def pty_ws(ws: WebSocket) -> None:
await ws.close(code=1011)
return
await session.attach(ws)
# A fresh xterm cannot reliably reconstruct the TUI from an arbitrary
# bounded tail of alternate-screen, differential ANSI output. Reused PTYs
# emit a complete frame after replay so reconnects never reopen blank.
await session.attach(ws, force_redraw=not _created)
# --- writer loop: WebSocket → PTY master ----------------------------
# No reader task here: the session's drain task (spawned once per PTY,
+9 -2
View File
@@ -8,12 +8,13 @@ from hermes_cli import web_server
class FakeBridge:
def __init__(self):
self.alive = True
self.written = bytearray()
def read(self, timeout):
return b"" # idle forever
def write(self, data):
pass
self.written.extend(data)
def resize(self, cols, rows):
pass
@@ -24,11 +25,16 @@ class FakeBridge:
@pytest.fixture
def pty_keepalive_harness(monkeypatch):
spawned = []
class Spawned(list):
pass
spawned = Spawned()
spawned.bridges = []
def fake_spawn(argv, cwd=None, env=None):
b = FakeBridge()
spawned.append(argv)
spawned.bridges.append(b)
return b
monkeypatch.setattr(web_server.PtyBridge, "spawn", staticmethod(fake_spawn))
@@ -60,6 +66,7 @@ async def test_attach_token_reuses_same_session(pty_keepalive_harness):
with client.websocket_connect("/api/pty?attach=TOK1") as ws2:
ws2.send_bytes(b"again")
assert len(pty_keepalive_harness) == 1 # reattached, did not respawn
assert bytes(pty_keepalive_harness.bridges[0].written) == b"hi\x0cagain"
@pytest.mark.asyncio
+36
View File
@@ -76,6 +76,42 @@ async def test_attach_replays_buffer_then_streams_live():
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()
await s.attach(ws, force_redraw=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_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