mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
unsupportedpastels
e843e0427f
After a trimmed replay the browser sends a resize one row away and then the real size, so the foreground program redraws. The host terminal pump applied both at once, and a program that reads the size late can handle both SIGWINCHes after the size is back where it was. Node reports a stdout 'resize' only when the size it reads has changed, and Hermes' TUI redraws on that event, so nothing redrew: on a real PTY a Node 26 child got the signals and reported no resize. A resize back to the size before the previous one now waits until the program has written output since that one (a program that redraws on a resize does), or one second for a program that does not. Other resizes, including a drag's, are applied at once as before. The POSIX bridge reads the PTY size with TIOCGWINSZ; a bridge that cannot report it (ConPTY) never waits. Known limit: output the program wrote before the step but read after it counts as an answer, so a program printing continuously can still miss the step.
612 lines
28 KiB
Python
612 lines
28 KiB
Python
"""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.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import threading
|
|
import time
|
|
from concurrent.futures import Executor
|
|
from pathlib import Path
|
|
from typing import Callable, Dict, Optional, Tuple
|
|
|
|
from fastapi import WebSocketDisconnect
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
WS_CLOSE_PROCESS_EXITED = 4410
|
|
# 4410 also ends a session the server closed (expiry, a deleted profile, shutdown). Only a child
|
|
# that exited by itself carries this reason, so the Webapp terminal can close its tab on `exit`.
|
|
WS_CLOSE_PROCESS_EXITED_REASON = "Process exited"
|
|
WS_CLOSE_SUPERSEDED = 4409
|
|
# Going Away: the viewer stopped taking output, but its session lives on and the same
|
|
# token reattaches. The dashboard chat tab and the Webapp terminal both redial on 1001;
|
|
# the chat tab reads 1011 as a failed start and offers a new session instead.
|
|
WS_CLOSE_VIEWER_STALLED = 1001
|
|
WS_CLOSE_VIEWER_STALLED_REASON = "Terminal viewer stopped reading; reconnect"
|
|
# How long one frame may wait on the viewer before it counts as stalled.
|
|
_VIEWER_SEND_TIMEOUT_SECONDS = 5.0
|
|
# A reattach replay is a single frame of up to the 1 MiB buffer (host-terminal clients take exactly
|
|
# the next binary frame as the replay, so it is not chunked). It gets the stall limit plus the time
|
|
# to cross a link this slow (about 256 kbit/s; 37 s for 1 MiB). A replay that misses that deadline is
|
|
# halved for the session's next attaches and never grows back: a send returns once the socket's buffers
|
|
# take the frame, so a quick one does not show that the link carries more, and growing back would
|
|
# replay a size that already timed out. A viewer on a slower link redials onto a replay it carries.
|
|
_REPLAY_MIN_BYTES_PER_SECOND = 32 * 1024
|
|
# The smallest replay a timed-out one shrinks to, so a reattach always shows the newest output. Its
|
|
# deadline needs about 3 KB/s, less than a link needs to keep up with live output (one frame per
|
|
# stall limit).
|
|
_REPLAY_FLOOR_BYTES = 16 * 1024
|
|
# A viewer send that fails because the browser is gone: Starlette raises WebSocketDisconnect for a
|
|
# dead transport and RuntimeError for a socket already closed; OSError covers the TimeoutError of
|
|
# a viewer that stopped reading.
|
|
_VIEWER_SEND_ERRORS = (WebSocketDisconnect, RuntimeError, OSError)
|
|
TUI_FORCE_REDRAW = b"\x0c"
|
|
|
|
|
|
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)
|
|
|
|
|
|
async def _close_ws(ws, code: int, reason: Optional[str] = None) -> None:
|
|
try:
|
|
if ws is not None:
|
|
await asyncio.wait_for(ws.close(code=code, reason=reason), timeout=5.0)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _process_ancestors(pid: int) -> set[int]:
|
|
"""Pids of ``pid``'s live ancestors; empty when the process is gone or unreadable."""
|
|
import psutil # type: ignore
|
|
|
|
try:
|
|
return {parent.pid for parent in psutil.Process(pid).parents()}
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess, ValueError):
|
|
return set()
|
|
|
|
|
|
def _key_segments(key: str) -> tuple[str, str]:
|
|
"""``(profile, resume)`` a keep-alive key was registered under.
|
|
|
|
Registry keys read ``token\\0profile\\0resume``; a key without those
|
|
segments — a chat that was never resumed from — yields ``("", "")``.
|
|
Session ids are only unique within a profile's store, so a resume target
|
|
is identified by both.
|
|
"""
|
|
parts = key.split("\0")
|
|
return (parts[1], parts[2]) if len(parts) >= 3 else ("", "")
|
|
|
|
|
|
class PtySession:
|
|
def __init__(
|
|
self, key: str, bridge, *, buffer_cap: int, read_timeout: float,
|
|
active_session_file: Optional[Path] = None, drain_executor: Optional[Executor] = None,
|
|
) -> None:
|
|
self.key = key
|
|
self.bridge = bridge
|
|
self.active_session_file = active_session_file
|
|
self.active_session_cleanup: Optional[Callable[[], None]] = None
|
|
self.buffer = RingBuffer(buffer_cap)
|
|
# How many of the buffer's newest bytes a reattach replays: a replay that times out halves it,
|
|
# down to _REPLAY_FLOOR_BYTES, for the rest of the session.
|
|
self._replay_limit = buffer_cap
|
|
self.alive = True
|
|
self.attached = False
|
|
# Admission can fail before the first attach (e.g. lost metadata).
|
|
self.last_detached_at: Optional[float] = time.monotonic()
|
|
self._read_timeout = read_timeout
|
|
# Runs the blocking reads; None is the loop's default executor.
|
|
self._drain_executor = drain_executor
|
|
self._ws = None
|
|
self._attach_generation = 0
|
|
# Only final sink writes and viewer claims hold this lock. Profile
|
|
# admission/backpressure waits happen outside it.
|
|
self._viewer_lock = threading.Lock()
|
|
self._drain_task: Optional[asyncio.Task] = None
|
|
self._write_lock = asyncio.Lock()
|
|
self._output_lock = asyncio.Lock()
|
|
self._input_task: Optional[asyncio.Task] = None
|
|
self._close_task: Optional[asyncio.Task] = None
|
|
# Chunks of output read so far, and set when one arrives (``wait_for_output``).
|
|
self._outputs = 0
|
|
self._output_arrived = asyncio.Event()
|
|
|
|
async def start(self) -> None:
|
|
self._drain_task = asyncio.create_task(self._drain())
|
|
|
|
async def _drain(self) -> None:
|
|
loop = asyncio.get_running_loop()
|
|
while self.alive:
|
|
try:
|
|
chunk = await loop.run_in_executor(self._drain_executor, self.bridge.read, self._read_timeout)
|
|
except OSError:
|
|
chunk = None
|
|
if chunk is None: # EOF — the agent process exited
|
|
self.alive = False
|
|
ws = self._ws
|
|
async with self._output_lock:
|
|
if self._ws is ws:
|
|
await _close_ws(ws, WS_CLOSE_PROCESS_EXITED, WS_CLOSE_PROCESS_EXITED_REASON)
|
|
return
|
|
if not chunk: # idle tick
|
|
await asyncio.sleep(0)
|
|
continue
|
|
self.buffer.append(chunk)
|
|
self._outputs += 1
|
|
self._output_arrived.set()
|
|
ws = self._ws
|
|
output_lock = self._output_lock
|
|
try:
|
|
if ws is not None:
|
|
async with output_lock:
|
|
if self._ws is ws:
|
|
await asyncio.wait_for(ws.send_bytes(chunk), timeout=_VIEWER_SEND_TIMEOUT_SECONDS)
|
|
except Exception:
|
|
# The viewer is gone; nothing else observes this failure (the handler's finally
|
|
# only runs once ws.receive() sees the disconnect). detach() is a no-op when a
|
|
# replacement socket attached during the send, so the new viewer keeps its session.
|
|
if self._ws is ws:
|
|
self.detach(ws)
|
|
await _close_ws(ws, WS_CLOSE_VIEWER_STALLED, WS_CLOSE_VIEWER_STALLED_REASON)
|
|
|
|
async def write(self, ws, data: bytes, *, fence=None, **bridge_options) -> bool:
|
|
"""Serialize input and discard bytes from a superseded socket.
|
|
|
|
``fence`` is handed to the bridge (see ``PtyBridge.write``); what it raises propagates.
|
|
``bridge_options`` (``timeout``) go to the bridge's ``write`` unchanged.
|
|
"""
|
|
async with self._write_lock:
|
|
if self._ws is not ws or not self.alive:
|
|
return True
|
|
generation = self._attach_generation
|
|
|
|
def viewer_fence(write):
|
|
def admitted_write():
|
|
with self._viewer_lock:
|
|
if self._ws is not ws or self._attach_generation != generation or not self.alive:
|
|
return None
|
|
return write()
|
|
|
|
return fence(admitted_write)
|
|
|
|
if fence is not None:
|
|
bridge_options["fence"] = viewer_fence
|
|
task = self._input_task = asyncio.create_task(self.bridge.write(data, **bridge_options))
|
|
try:
|
|
delivered = await task
|
|
except asyncio.CancelledError:
|
|
if self._attach_generation != generation:
|
|
return False
|
|
raise
|
|
finally:
|
|
if self._input_task is task:
|
|
self._input_task = None
|
|
# A replacement socket can attach while the bridge write is
|
|
# suspended on backpressure. A late failure from the superseded
|
|
# socket must not poison the replacement's shared PTY session.
|
|
if (
|
|
not delivered
|
|
and self._ws is ws
|
|
and self._attach_generation == generation
|
|
):
|
|
self.alive = False
|
|
return delivered
|
|
|
|
def resize(self, ws, *, cols: int, rows: int) -> None:
|
|
if self._ws is ws and self.alive:
|
|
self.bridge.resize(cols=cols, rows=rows)
|
|
|
|
def size(self) -> Optional[tuple[int, int]]:
|
|
"""The PTY's ``(cols, rows)``, or None where the bridge cannot tell (ConPTY, stubs)."""
|
|
return self.bridge.size() if hasattr(self.bridge, "size") else None
|
|
|
|
@property
|
|
def outputs(self) -> int:
|
|
"""How many chunks of output have been read from the program so far."""
|
|
return self._outputs
|
|
|
|
async def wait_for_output(self, since: int, timeout: float) -> None:
|
|
"""Return once output past ``since`` (an ``outputs`` value) has been read, the session has
|
|
ended, or ``timeout`` seconds have passed."""
|
|
loop = asyncio.get_running_loop()
|
|
deadline = loop.time() + timeout
|
|
while self.alive and self._outputs == since:
|
|
self._output_arrived.clear()
|
|
remaining = deadline - loop.time()
|
|
if remaining <= 0:
|
|
return
|
|
try:
|
|
await asyncio.wait_for(self._output_arrived.wait(), remaining)
|
|
except TimeoutError:
|
|
return
|
|
|
|
async def attach(
|
|
self, ws, *, force_redraw: bool = False,
|
|
initial_text: Optional[str | Callable[[bool], str]] = None,
|
|
) -> bool:
|
|
"""Attach a browser terminal and replay buffered PTY output.
|
|
|
|
The TUI renders differentially on an alternate screen, so a bounded ANSI tail is not a
|
|
self-contained frame; ``force_redraw`` asks the live TUI for one full redraw after replay.
|
|
``initial_text`` is sent before the replay; given as a function, it is called with whether
|
|
the replay leaves out older output (dropped from the buffer or past the replay limit).
|
|
"""
|
|
self._attach_generation += 1
|
|
generation = self._attach_generation
|
|
if self._input_task is not None:
|
|
self._input_task.cancel()
|
|
# Revoke queued workers first, then serialize the claim after any
|
|
# already-admitted sink write. ConPTY writes can block: keep the loop
|
|
# free to run their timeout/cancellation cleanup while we wait.
|
|
while not self._viewer_lock.acquire(blocking=False):
|
|
await asyncio.sleep(0.001)
|
|
if self._attach_generation != generation:
|
|
return False
|
|
try:
|
|
if self._attach_generation != generation:
|
|
return False
|
|
old_ws = self._ws
|
|
self._ws = ws
|
|
self.attached = True
|
|
self.last_detached_at = None
|
|
# Per-viewer lock: a stalled old send must not block its replacement.
|
|
output_lock = self._output_lock = asyncio.Lock()
|
|
finally:
|
|
self._viewer_lock.release()
|
|
try:
|
|
async with output_lock:
|
|
held = self.buffer.snapshot()
|
|
snap = held[len(held) - min(len(held), self._replay_limit):]
|
|
trimmed = self.buffer.truncated or len(snap) < len(held)
|
|
if old_ws is not ws:
|
|
await _close_ws(old_ws, WS_CLOSE_SUPERSEDED)
|
|
if self._ws is not ws:
|
|
return False
|
|
if initial_text is not None:
|
|
text = initial_text(trimmed) if callable(initial_text) else initial_text
|
|
await asyncio.wait_for(ws.send_text(text), timeout=_VIEWER_SEND_TIMEOUT_SECONDS)
|
|
if self._ws is not ws:
|
|
return False
|
|
# Metadata-bearing host clients treat exactly the next binary
|
|
# frame as replay. Send an empty snapshot too, so the first live
|
|
# frame never loses terminal-query replies on an empty history.
|
|
if snap or initial_text is not None:
|
|
replay_timeout = _VIEWER_SEND_TIMEOUT_SECONDS + len(snap) / _REPLAY_MIN_BYTES_PER_SECOND
|
|
try:
|
|
await asyncio.wait_for(ws.send_bytes(snap), timeout=replay_timeout)
|
|
except TimeoutError:
|
|
self._replay_limit = max(_REPLAY_FLOOR_BYTES, len(snap) // 2)
|
|
raise
|
|
except _VIEWER_SEND_ERRORS:
|
|
self.detach(ws)
|
|
return False
|
|
except BaseException: # cancellation or a bug: never leave this socket attached
|
|
self.detach(ws)
|
|
raise
|
|
if self._ws is not ws:
|
|
return False
|
|
if not self.alive:
|
|
self.detach(ws)
|
|
await _close_ws(ws, WS_CLOSE_PROCESS_EXITED, WS_CLOSE_PROCESS_EXITED_REASON)
|
|
return False
|
|
if force_redraw:
|
|
return await self.write(ws, TUI_FORCE_REDRAW)
|
|
return True
|
|
|
|
def detach(self, ws) -> None:
|
|
# Only the currently-attached socket may mark the session detached: a superseded socket's
|
|
# handler also calls detach on its way out (after the new tab attached), and flipping
|
|
# ``attached`` then would make a session with a live viewer look idle and reapable.
|
|
if self._ws is not ws:
|
|
return
|
|
self._ws = None
|
|
self.attached = False
|
|
self.last_detached_at = time.monotonic()
|
|
|
|
def hosts_pid(self, pid: Optional[int]) -> bool:
|
|
"""Whether ``pid`` runs inside this PTY: the leader itself or one of its descendants.
|
|
|
|
The dashboard's PTY leader is ``node entry.js``; the process that takes the
|
|
session lease is its ``tui_gateway`` child, so the lease pid is never the
|
|
bridge's own pid and only ancestry identifies the terminal. A bridge that
|
|
cannot report a pid (already reaped, or a stub) is simply not a match.
|
|
"""
|
|
if pid is None:
|
|
return False
|
|
try:
|
|
leader = int(self.bridge.pid)
|
|
except (AttributeError, TypeError, ValueError, OSError):
|
|
return False
|
|
return leader == pid or leader in _process_ancestors(pid)
|
|
|
|
async def close(self) -> None:
|
|
# Disconnect cancellation must not strand a child after its registry
|
|
# entry was removed. Cleanup has independent, idempotent ownership.
|
|
if self._close_task is None:
|
|
self.alive = False
|
|
self._attach_generation += 1
|
|
if self._input_task is not None:
|
|
self._input_task.cancel()
|
|
ws = self._ws
|
|
self.detach(ws)
|
|
self._close_task = asyncio.create_task(self._close(ws))
|
|
await asyncio.shield(self._close_task)
|
|
|
|
async def _close(self, ws) -> None:
|
|
await _close_ws(ws, WS_CLOSE_PROCESS_EXITED)
|
|
if self._drain_task is not None:
|
|
self._drain_task.cancel()
|
|
try:
|
|
await self._drain_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except Exception:
|
|
# The drain handles its own read and viewer errors, so this is a bug; the
|
|
# child below must still be closed.
|
|
logger.warning("PTY output drain failed", exc_info=True)
|
|
try:
|
|
# bridge.close() joins the child — blocking; keep it off the event loop.
|
|
# See #53227.
|
|
await asyncio.to_thread(self.bridge.close)
|
|
except Exception: # health: allow BLE001 S110 -- teardown of an already-dead PTY must not mask the caller's error path
|
|
pass
|
|
try:
|
|
if self.active_session_file is not None:
|
|
self.active_session_file.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
if self.active_session_cleanup is not None:
|
|
try:
|
|
self.active_session_cleanup()
|
|
except Exception: # health: allow BLE001 S110 -- cleanup callback must not mask the close path; the PTY is dead either way
|
|
pass
|
|
self.active_session_cleanup = None
|
|
|
|
|
|
class RegistryFull(Exception):
|
|
"""Every keep-alive slot holds a PTY that some tab is still attached to."""
|
|
|
|
def __init__(self, message: str = "Too many chat terminals are open in other tabs; close one and try again.") -> None:
|
|
super().__init__(message)
|
|
|
|
|
|
async def run_reaper(registry: PtySessionRegistry, *, interval: float = 60.0) -> None:
|
|
"""Periodically reap idle/dead keep-alive sessions. Cancelled on shutdown."""
|
|
while True:
|
|
await asyncio.sleep(interval)
|
|
try:
|
|
await registry.reap_idle()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class PtySessionRegistry:
|
|
def __init__(
|
|
self, *, ttl: float, max_sessions: int, buffer_cap: int, read_timeout: float,
|
|
drain_executor: Optional[Executor] = None,
|
|
) -> None:
|
|
self._ttl = ttl
|
|
self._max = max_sessions
|
|
self._buffer_cap = buffer_cap
|
|
self._read_timeout = read_timeout
|
|
self._drain_executor = drain_executor
|
|
self._sessions: dict[str, PtySession] = {}
|
|
# One registry-wide reservation spans lookup, spawn, and registration:
|
|
# racing connections with one attach token must share one tracked PTY.
|
|
self._spawn_lock = asyncio.Lock()
|
|
self._spawn_cleanup: set[asyncio.Task] = set()
|
|
self._closed = False
|
|
# Retain ownership of removed sessions until helper-process shutdown completes.
|
|
self._background_closes: set[asyncio.Task] = set()
|
|
|
|
async def attach_or_spawn(
|
|
self, key: str, *, spawn: Callable[[], object], active_session_file: Optional[Path] = None,
|
|
) -> tuple[PtySession, bool]:
|
|
# Reserve capacity and the key across blocking fork/exec. On cancellation
|
|
# the registry, not the request's cancellation scope, owns that reservation
|
|
# until the admission finishes and any unclaimed child has been closed.
|
|
await self._spawn_lock.acquire()
|
|
admission = asyncio.create_task(
|
|
self._attach_or_spawn(key, spawn=spawn, active_session_file=active_session_file))
|
|
try:
|
|
# wait() leaves admission running on caller cancellation without
|
|
# shield() reporting its late exception (Python 3.14). The registry's
|
|
# discard task owns that result once the caller disconnects.
|
|
await asyncio.wait({admission})
|
|
result = admission.result()
|
|
except asyncio.CancelledError:
|
|
cleanup = asyncio.create_task(self._discard_admission(admission))
|
|
self._spawn_cleanup.add(cleanup)
|
|
cleanup.add_done_callback(self._spawn_cleanup.discard)
|
|
raise
|
|
except BaseException:
|
|
self._spawn_lock.release()
|
|
raise
|
|
self._spawn_lock.release()
|
|
return result
|
|
|
|
async def _discard_admission(self, admission: asyncio.Task) -> None:
|
|
try:
|
|
try:
|
|
session, created = await admission
|
|
except Exception:
|
|
# The disconnected caller cannot observe a late fork/exec
|
|
# failure: retrieve it here and keep the traceback.
|
|
logger.debug("PTY spawn failed after its caller disconnected", exc_info=True)
|
|
return
|
|
if created:
|
|
if self._sessions.get(session.key) is session:
|
|
self._sessions.pop(session.key)
|
|
await session.close()
|
|
finally:
|
|
self._spawn_lock.release()
|
|
|
|
async def _attach_or_spawn(
|
|
self, key: str, *, spawn: Callable[[], object], active_session_file: Optional[Path] = None,
|
|
) -> tuple[PtySession, bool]:
|
|
if self._closed:
|
|
raise RegistryFull("Terminal service is shutting down.")
|
|
await self.reap_idle()
|
|
existing = self._sessions.get(key)
|
|
if existing is not None and existing.alive:
|
|
return existing, False
|
|
if existing is not None: # dead remnant
|
|
# Close in the background: ending a dead leader's helpers can take the helper
|
|
# grace, and this reservation serializes every new chat.
|
|
self._sessions.pop(key, None)
|
|
self._close_in_background(existing)
|
|
if len(self._sessions) >= self._max:
|
|
self._reap_one_idle_or_raise()
|
|
# PTY spawn does blocking fork/exec work — keep it off the event loop.
|
|
# See #53227.
|
|
bridge = await asyncio.to_thread(spawn)
|
|
session = PtySession(
|
|
key,
|
|
bridge,
|
|
buffer_cap=self._buffer_cap,
|
|
read_timeout=self._read_timeout,
|
|
active_session_file=active_session_file,
|
|
drain_executor=self._drain_executor,
|
|
)
|
|
await session.start()
|
|
self._sessions[key] = session
|
|
return session, True
|
|
|
|
async def close_other_sessions(self, prefix: str, *, keep_key: str) -> None:
|
|
"""Close sessions belonging to the same logical client except ``keep_key``.
|
|
|
|
Dashboard profile changes keep the browser's attach token but change the
|
|
canonical session key. The previous profile's detached PTY must not
|
|
remain alive long enough to hold the TUI session lease and reject a
|
|
later return to that chat.
|
|
"""
|
|
async with self._spawn_lock:
|
|
keys = [
|
|
key for key in self._sessions
|
|
if key != keep_key and (key == prefix or key.startswith(prefix + "\0"))
|
|
]
|
|
for key in keys:
|
|
session = self._sessions.pop(key, None)
|
|
if session is not None:
|
|
# A sibling tab sharing the attach token may still be viewing this
|
|
# PTY: supersede it explicitly (4409) instead of leaving it silent
|
|
# until its next keystroke fails with 1013. Detach first so close()
|
|
# does not overwrite that code with WS_CLOSE_PROCESS_EXITED.
|
|
ws = session._ws
|
|
session.detach(ws)
|
|
await _close_ws(ws, WS_CLOSE_SUPERSEDED)
|
|
await session.close()
|
|
|
|
async def close_orphaned_sessions(
|
|
self, resume: Optional[str], *, keep_key: str, holder_pid: Optional[int] = None,
|
|
) -> None:
|
|
"""Close a keep-alive PTY stranded under a superseded attach token.
|
|
|
|
Rotating the token — the dashboard's *New chat* — moves the tab to a
|
|
fresh PTY while the previous one keeps the TUI's single-writer lease on
|
|
its session. That terminal is out of the user's reach, so the next
|
|
return to the same chat is refused as a session held elsewhere. A PTY
|
|
with a live viewer is never a candidate: somebody is still using it.
|
|
|
|
A chat that was never resumed from carries no resume target in its key,
|
|
so ``holder_pid`` — the process holding the lease for the session being
|
|
resumed — identifies that terminal instead.
|
|
"""
|
|
if not resume:
|
|
return
|
|
# The requested chat is (profile, session): the same session id in
|
|
# another profile's store is a different chat whose terminal stays.
|
|
target = (_key_segments(keep_key)[0], resume)
|
|
async with self._spawn_lock:
|
|
doomed = [
|
|
key for key, session in self._sessions.items()
|
|
if key != keep_key and not session.attached
|
|
and (_key_segments(key) == target or session.hosts_pid(holder_pid))
|
|
]
|
|
sessions = [self._sessions.pop(key) for key in doomed]
|
|
# Close outside the registry lock — a close can wait out its helpers' SIGHUP grace and
|
|
# this lock serializes every new chat — but still before the caller spawns: the child's
|
|
# session lease is only released once its close finishes.
|
|
for session in sessions:
|
|
await session.close()
|
|
|
|
def detach(self, key: str, ws) -> None:
|
|
s = self._sessions.get(key)
|
|
if s is not None:
|
|
s.detach(ws)
|
|
|
|
def expired(self, s: PtySession, now: Optional[float] = None) -> bool:
|
|
"""Whether ``s`` has had no viewer for longer than the retention TTL."""
|
|
return (not s.attached and s.last_detached_at is not None
|
|
and (time.monotonic() if now is None else now) - s.last_detached_at > self._ttl)
|
|
|
|
def _reapable(self, s: PtySession, now: Optional[float] = None) -> bool:
|
|
# EOF never arrives if a helper still holds the PTY slave after the child died (#76759);
|
|
# ask the process itself (a WNOHANG waitpid).
|
|
return not s.alive or not s.bridge.is_alive() or self.expired(s, now)
|
|
|
|
async def reap_idle(self, now: Optional[float] = None) -> None:
|
|
now = time.monotonic() if now is None else now
|
|
doomed = [(key, s) for key, s in self._sessions.items() if self._reapable(s, now)]
|
|
for key, expected in doomed:
|
|
# Reaps overlap (attach_or_spawn and the background reaper) and close()
|
|
# awaits, so a concurrent reap can have popped this key already — skip
|
|
# it instead of raising KeyError into the websocket handler. A viewer can
|
|
# also re-attach during an earlier close, so re-check before closing.
|
|
if self._sessions.get(key) is not expected or not self._reapable(expected, now):
|
|
continue
|
|
session = self._sessions.pop(key, None)
|
|
if session is not None:
|
|
await session.close()
|
|
|
|
def _reap_one_idle_or_raise(self) -> None:
|
|
idle = [s for s in self._sessions.values() if not s.attached and s.last_detached_at is not None]
|
|
if not idle:
|
|
raise RegistryFull()
|
|
oldest = min(idle, key=lambda s: s.last_detached_at or 0.0)
|
|
self._sessions.pop(oldest.key, None)
|
|
self._close_in_background(oldest)
|
|
|
|
def _close_in_background(self, session: PtySession) -> asyncio.Task:
|
|
task = asyncio.create_task(session.close())
|
|
self._background_closes.add(task)
|
|
task.add_done_callback(self._background_closes.discard)
|
|
return task
|
|
|
|
def reopen(self) -> None:
|
|
"""Take spawns again after :meth:`close_all`: the registry is process-wide, and a lifespan can
|
|
start again in one process (the gateway falling back from the Webapp's listener to its
|
|
private one)."""
|
|
self._closed = False
|
|
|
|
async def close_all(self) -> None:
|
|
self._closed = True
|
|
# Wait for any admitted spawn before snapshotting shutdown ownership.
|
|
async with self._spawn_lock:
|
|
pass
|
|
if self._spawn_cleanup:
|
|
await asyncio.shield(asyncio.gather(*self._spawn_cleanup))
|
|
# Helper-process grace must run concurrently inside the backend shutdown budget.
|
|
sessions = [self._sessions.pop(key) for key in list(self._sessions)]
|
|
await asyncio.gather(*(s.close() for s in sessions), *self._background_closes, return_exceptions=True)
|