Force terminal redraw on reconnect

- Track last known terminal size in TerminalSession
- Add force_redraw() method that re-sends SIGWINCH to trigger redraw
- Call force_redraw() when WebSocket reconnects to existing session
- Helps tmux and similar apps restore proper display after disconnect
This commit is contained in:
GitHub Copilot
2026-01-24 16:28:50 +00:00
parent 2f61bd7747
commit 728681a195
2 changed files with 23 additions and 0 deletions
+20
View File
@@ -56,6 +56,9 @@ class TerminalSession(Session):
self._screen = pyte.Screen(DEFAULT_SCREEN_WIDTH, DEFAULT_SCREEN_HEIGHT)
self._stream = pyte.Stream(self._screen)
self._screen_lock = asyncio.Lock()
# Track last known terminal size for reconnection
self._last_width = DEFAULT_SCREEN_WIDTH
self._last_height = DEFAULT_SCREEN_HEIGHT
super().__init__()
def __rich_repr__(self) -> rich.repr.Result:
@@ -64,6 +67,9 @@ class TerminalSession(Session):
async def open(self, width: int = 80, height: int = 24) -> None:
log.info("Opening terminal session %s with command: %s", self.session_id, self.command)
# Track the initial size
self._last_width = width
self._last_height = height
# Initialize pyte screen with the requested size (under lock to prevent races)
async with self._screen_lock:
self._screen = pyte.Screen(width, height)
@@ -100,6 +106,9 @@ class TerminalSession(Session):
fcntl.ioctl(self.master_fd, termios.TIOCSWINSZ, buf)
async def set_terminal_size(self, width: int, height: int) -> None:
# Track the size for reconnection
self._last_width = width
self._last_height = height
# First resize the PTY (blocking call in executor)
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._set_terminal_size, width, height)
@@ -107,6 +116,17 @@ class TerminalSession(Session):
async with self._screen_lock:
self._screen.resize(height, width)
async def force_redraw(self) -> None:
"""Force a terminal redraw by re-sending the current size.
This triggers a SIGWINCH to the child process, causing applications
like tmux to redraw their display.
"""
loop = asyncio.get_running_loop()
await loop.run_in_executor(
None, self._set_terminal_size, self._last_width, self._last_height
)
async def _add_to_replay_buffer(self, data: bytes) -> None:
"""Add data to replay buffer, maintaining size limit."""
async with self._replay_lock: