From 0c3f69e6ddc1228583a8038da31470bbf7a634be Mon Sep 17 00:00:00 2001 From: mcocdaa <2021137961@qq.com> Date: Wed, 14 Oct 2026 05:00:26 +0200 Subject: [PATCH] fix: resolve terminal reconnect races, tmux line overruns, per-frame sqlite queries, and tooling checks --- .../termflow_control_plane/api/terminal.py | 10 +- .../src/termflow_control_plane/auth/epoch.py | 17 +- .../connections/terminal_hub.py | 11 ++ .../persistence/repositories.py | 17 ++ .../plugins/agent_broker/agent/watches.py | 158 ++++++++++++++++-- .../routing/terminal_router.py | 4 + .../tests/test_epoch_boundaries.py | 51 ++++++ apps/control-plane/tests/test_terminal_hub.py | 91 ++++++++++ apps/control-plane/tests/test_watch_engine.py | 31 ++++ apps/node/src/termflow_node/bridge/buffer.py | 2 +- apps/node/src/termflow_node/bridge/runtime.py | 9 +- .../src/termflow_node/bridge/transport.py | 8 +- .../src/termflow_node/tmux/control_client.py | 42 ++++- ...ns.py => test_tmux_actions_integration.py} | 0 apps/node/tests/test_bridge_transport.py | 3 +- apps/node/tests/test_control_parser.py | 37 ++++ apps/node/tests/test_output_buffer.py | 3 +- .../client-core/src/terminal/session.test.ts | 49 ++++++ packages/client-core/src/terminal/session.ts | 23 ++- packages/client-ui/package.json | 3 + packages/client-ui/src/router/routes.ts | 36 ++-- pyproject.toml | 4 + scripts/verify.sh | 5 +- 23 files changed, 552 insertions(+), 62 deletions(-) rename apps/node/tests/integration/{test_tmux_actions.py => test_tmux_actions_integration.py} (100%) diff --git a/apps/control-plane/src/termflow_control_plane/api/terminal.py b/apps/control-plane/src/termflow_control_plane/api/terminal.py index b753f40e..1cb769b8 100644 --- a/apps/control-plane/src/termflow_control_plane/api/terminal.py +++ b/apps/control-plane/src/termflow_control_plane/api/terminal.py @@ -109,7 +109,10 @@ async def _send_terminal_events( ) -> None: while True: event = await terminal.next_event() - if await persisted_authentication_epoch(repositories) != auth_epoch: + router_epoch = getattr(terminal_router, "authentication_epoch", None) + if ( + isinstance(router_epoch, int) and router_epoch != auth_epoch + ) or await persisted_authentication_epoch(repositories) != auth_epoch: await terminal_router.request_close(terminal, "client_closed") with suppress(RuntimeError): await websocket.close(code=4401, reason="Authentication epoch changed") @@ -219,7 +222,10 @@ async def _receive_terminal_input( incoming = await websocket.receive() if incoming["type"] == "websocket.disconnect": return - if await persisted_authentication_epoch(repositories) != auth_epoch: + router_epoch = getattr(terminal_router, "authentication_epoch", None) + if ( + isinstance(router_epoch, int) and router_epoch != auth_epoch + ) or await persisted_authentication_epoch(repositories) != auth_epoch: terminal.terminate("client_closed") with suppress(RuntimeError): await websocket.close(code=4401, reason="Authentication epoch changed") diff --git a/apps/control-plane/src/termflow_control_plane/auth/epoch.py b/apps/control-plane/src/termflow_control_plane/auth/epoch.py index 680c7714..0015b119 100644 --- a/apps/control-plane/src/termflow_control_plane/auth/epoch.py +++ b/apps/control-plane/src/termflow_control_plane/auth/epoch.py @@ -8,11 +8,24 @@ from termflow_control_plane.persistence.repositories import RepositoryBundle +_EPOCH_CACHE_TTL_SECONDS = 1.0 -async def persisted_authentication_epoch(repositories: RepositoryBundle) -> int: + +async def persisted_authentication_epoch( + repositories: RepositoryBundle, + *, + ttl_seconds: float = _EPOCH_CACHE_TTL_SECONDS, +) -> int: """Finish the short DB read before propagating transport cancellation.""" - lookup = asyncio.create_task(repositories.auth_state.get()) + auth_state = repositories.auth_state + cached = getattr(auth_state, "cached_epoch", None) + if callable(cached): + epoch = cached(ttl_seconds=ttl_seconds) + if isinstance(epoch, int): + return epoch + + lookup = asyncio.create_task(auth_state.get()) try: state = await asyncio.shield(lookup) except asyncio.CancelledError: diff --git a/apps/control-plane/src/termflow_control_plane/connections/terminal_hub.py b/apps/control-plane/src/termflow_control_plane/connections/terminal_hub.py index c47ddd92..af3a5bff 100644 --- a/apps/control-plane/src/termflow_control_plane/connections/terminal_hub.py +++ b/apps/control-plane/src/termflow_control_plane/connections/terminal_hub.py @@ -50,6 +50,7 @@ def __init__( self._local_close: LocalTerminalClose | None = None self._stream_id: UUID | None = None self._last_seq = 0 + self._resuming = False self.close_requested = False self.remote_closed = False self.input_bytes = 0 @@ -110,10 +111,15 @@ def observe_opened(self, opened: TerminalOpenedPayload) -> None: if self._stream_id != opened.stream_id: self._stream_id = opened.stream_id self._last_seq = 0 + self._resuming = False def output_state(self, output: TerminalOutputPayload) -> str: if self._stream_id != output.stream_id: return "gap" + if self._resuming: + if output.seq > self._last_seq + 1: + return "duplicate" + self._resuming = False if output.seq <= self._last_seq: return "duplicate" return "next" if output.seq == self._last_seq + 1 else "gap" @@ -134,6 +140,7 @@ def prepare_resume(self, stream_id: UUID, after_seq: int) -> bool: self._queued_bytes = 0 self._available.clear() self._last_seq = after_seq + self._resuming = True return True @@ -147,6 +154,10 @@ def __init__(self, *, queue_max_messages: int, queue_max_bytes: int) -> None: self._lock = asyncio.Lock() self._auth_epoch = 1 + @property + def authentication_epoch(self) -> int: + return self._auth_epoch + async def register( self, instance_id: UUID, diff --git a/apps/control-plane/src/termflow_control_plane/persistence/repositories.py b/apps/control-plane/src/termflow_control_plane/persistence/repositories.py index 568c6be2..bd44f3a3 100644 --- a/apps/control-plane/src/termflow_control_plane/persistence/repositories.py +++ b/apps/control-plane/src/termflow_control_plane/persistence/repositories.py @@ -6,6 +6,7 @@ import json import math import secrets +import time from collections.abc import Mapping from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta @@ -556,12 +557,24 @@ async def count_since(self, since: datetime) -> int: class AuthStateRepository: def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None: self._sessions = sessions + self._cached_epoch: int | None = None + self._cached_epoch_at: float = 0.0 + + def cached_epoch(self, *, ttl_seconds: float = 1.0) -> int | None: + if ( + self._cached_epoch is not None + and (time.monotonic() - self._cached_epoch_at) < ttl_seconds + ): + return self._cached_epoch + return None async def get(self) -> AuthenticationState: async with self._sessions() as session: state = await session.get(AuthenticationState, 1) if state is None: raise RuntimeError("authentication state singleton is missing") + self._cached_epoch = int(state.epoch) + self._cached_epoch_at = time.monotonic() return state async def configure_totp( @@ -742,6 +755,8 @@ async def rotate_credentials( ) ) await session.commit() + self._cached_epoch = int(epoch) + self._cached_epoch_at = time.monotonic() return int(epoch) async def reset_and_increment_epoch( @@ -795,6 +810,8 @@ async def reset_and_increment_epoch( ) ) await session.commit() + self._cached_epoch = int(epoch) + self._cached_epoch_at = time.monotonic() return int(epoch) diff --git a/apps/control-plane/src/termflow_control_plane/plugins/agent_broker/agent/watches.py b/apps/control-plane/src/termflow_control_plane/plugins/agent_broker/agent/watches.py index 0b970704..7550949b 100644 --- a/apps/control-plane/src/termflow_control_plane/plugins/agent_broker/agent/watches.py +++ b/apps/control-plane/src/termflow_control_plane/plugins/agent_broker/agent/watches.py @@ -310,11 +310,66 @@ class ObservationCursorStore: def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None: self._sessions = sessions self._ledger: dict[tuple[UUID, str], _CursorState] = {} + self._dirty: set[tuple[UUID, str]] = set() + + @property + def has_dirty(self) -> bool: + return bool(self._dirty) @staticmethod def _aware(value: datetime) -> datetime: return _aware(value) # type: ignore[return-value] + def accept_in_memory( + self, + *, + instance_id: UUID, + pane_id: str, + stream_id: UUID, + seq: int, + observed_at: datetime, + ) -> CursorAcceptance: + """Accept or reject one live observation cursor in memory without opening a DB session.""" + key = (instance_id, pane_id) + state = self._ledger.get(key) + if state is None: + self._ledger[key] = _CursorState(1, str(stream_id), seq, observed_at) + self._dirty.add(key) + return CursorAcceptance.ACCEPTED + verdict, accepted = self._evaluate(state, stream_id, seq) + if accepted or verdict is CursorAcceptance.STREAM_CHANGED: + self._ledger[key] = _CursorState(state.incarnation, str(stream_id), seq, observed_at) + self._dirty.add(key) + return verdict + + async def flush(self, session: AsyncSession) -> bool: + """Persist any dirty in-memory cursors into ``session`` (no commit).""" + if not self._dirty: + return False + dirty_keys = list(self._dirty) + self._dirty.clear() + for instance_id, pane_id in dirty_keys: + state = self._ledger.get((instance_id, pane_id)) + if state is None: + continue + row = await session.get(PaneObservationCursor, (instance_id, pane_id)) + if row is not None: + row.pane_incarnation = str(state.incarnation) + row.stream_id = state.stream_id + row.seq = state.seq + row.observed_at = state.observed_at + else: + await self._create_anchor( + session, + instance_id=instance_id, + pane_id=pane_id, + stream_id=UUID(state.stream_id), + seq=state.seq, + incarnation=state.incarnation, + observed_at=state.observed_at, + ) + return True + async def accept( self, session: AsyncSession, @@ -326,10 +381,11 @@ async def accept( observed_at: datetime, ) -> tuple[CursorAcceptance, PaneCursor]: """Accept or reject one live observation cursor (no commit).""" - row = await session.get(PaneObservationCursor, (instance_id, pane_id)) + key = (instance_id, pane_id) + row = await session.get(PaneObservationCursor, key) + entry = self._ledger.get(key) if row is None: - state = self._ledger.get((instance_id, pane_id)) - if state is None: + if entry is None: # First observation of this pane. await self._create_anchor( session, @@ -350,21 +406,33 @@ async def accept( seq=seq, ), ) + state = entry else: - state = _CursorState( + row_state = _CursorState( int(row.pane_incarnation), row.stream_id, row.seq, self._aware(row.observed_at), ) + state = entry if (entry is not None and key in self._dirty) else row_state verdict, accepted = self._evaluate(state, stream_id, seq) if accepted: + self._ledger[key] = _CursorState(state.incarnation, str(stream_id), seq, observed_at) + self._dirty.discard(key) if row is not None: + row.pane_incarnation = str(state.incarnation) + row.stream_id = str(stream_id) row.seq = seq row.observed_at = observed_at else: - self._ledger[(instance_id, pane_id)] = _CursorState( - state.incarnation, str(stream_id), seq, observed_at + await self._create_anchor( + session, + instance_id=instance_id, + pane_id=pane_id, + stream_id=stream_id, + seq=seq, + incarnation=state.incarnation, + observed_at=observed_at, ) # The accepted cursor reflects the new position. return ( @@ -450,10 +518,11 @@ async def _create_anchor( ) -> None: """Persist a fresh anchor: the DB row for known instances, otherwise the in-memory fallback ledger.""" + self._ledger[(instance_id, pane_id)] = _CursorState( + incarnation, str(stream_id), seq, observed_at + ) + self._dirty.discard((instance_id, pane_id)) if await session.get(Instance, instance_id) is None: - self._ledger[(instance_id, pane_id)] = _CursorState( - incarnation, str(stream_id), seq, observed_at - ) return session.add( PaneObservationCursor( @@ -478,17 +547,16 @@ async def _persist_recovered( pane_incarnation: int, observed_at: datetime, ) -> None: + self._ledger[(instance_id, pane_id)] = _CursorState( + pane_incarnation, str(stream_id), seq, observed_at + ) + self._dirty.discard((instance_id, pane_id)) if row is not None: row.pane_incarnation = str(pane_incarnation) row.stream_id = str(stream_id) row.seq = seq row.observed_at = observed_at return - if (instance_id, pane_id) in self._ledger: - self._ledger[(instance_id, pane_id)] = _CursorState( - pane_incarnation, str(stream_id), seq, observed_at - ) - return await self._create_anchor( session, instance_id=instance_id, @@ -525,10 +593,13 @@ async def _get_cursor_in_session( instance_id: UUID, pane_id: str, ) -> PaneCursor | None: - row = await session.get(PaneObservationCursor, (instance_id, pane_id)) + key = (instance_id, pane_id) + entry = self._ledger.get(key) + if entry is not None and key in self._dirty: + return self._cursor_from_state(instance_id, pane_id, entry) + row = await session.get(PaneObservationCursor, key) if row is not None: return self._to_cursor(row) - entry = self._ledger.get((instance_id, pane_id)) if entry is None: return None return self._cursor_from_state(instance_id, pane_id, entry) @@ -539,10 +610,13 @@ async def _get_anchor_in_session( instance_id: UUID, pane_id: str, ) -> tuple[PaneCursor, datetime] | None: - row = await session.get(PaneObservationCursor, (instance_id, pane_id)) + key = (instance_id, pane_id) + entry = self._ledger.get(key) + if entry is not None and key in self._dirty: + return self._cursor_from_state(instance_id, pane_id, entry), entry.observed_at + row = await session.get(PaneObservationCursor, key) if row is not None: return self._to_cursor(row), self._aware(row.observed_at) - entry = self._ledger.get((instance_id, pane_id)) if entry is None: return None return self._cursor_from_state(instance_id, pane_id, entry), entry.observed_at @@ -947,6 +1021,7 @@ def __init__( self._topologies: dict[UUID, TopologySnapshot] = {} self._subscriber: EventSubscriber | None = None self._task: asyncio.Task[None] | None = None + self._flush_task: asyncio.Task[None] | None = None #: Bounded counters for failed wire messages etc. (plan §17): visible #: to tests and audits like the pipeline's PipelineDiagnostics. self.diagnostics = WatchEngineDiagnostics() @@ -960,6 +1035,7 @@ async def rebuild(self) -> None: rows = await self._watch_repo.list_active(now=self._clock()) runtimes: dict[UUID, WatcherRuntime] = {} async with self._sessions() as session: + flushed = await self._cursor_store.flush(session) for row in rows: binding = await session.get(AgentBinding, row.binding_id) if binding is None: @@ -976,6 +1052,8 @@ async def rebuild(self) -> None: # cursor's observed_at restarts the timer (plan §11.2). runtime.deadline.on_output(observed_at) runtimes[row.id] = runtime + if flushed: + await session.commit() self._watches = runtimes async def start(self) -> None: @@ -1022,10 +1100,43 @@ async def stop(self) -> None: except asyncio.CancelledError: pass self._task = None + if self._flush_task is not None: + self._flush_task.cancel() + try: + await self._flush_task + except asyncio.CancelledError: + pass + self._flush_task = None + if self._cursor_store.has_dirty: + try: + await self._flush_cursors() + except Exception: + logger.exception("Watch engine: cursor flush on stop failed") if self._subscriber is not None and self._hub is not None: await self._hub.unsubscribe(self._subscriber) self._subscriber = None + def _schedule_cursor_flush(self) -> None: + if self._flush_task is not None and not self._flush_task.done(): + return + self._flush_task = asyncio.create_task(self._delayed_cursor_flush()) + + async def _delayed_cursor_flush(self) -> None: + try: + await asyncio.sleep(0.05) + await self._flush_cursors() + except asyncio.CancelledError: + pass + except Exception: + logger.exception("Watch engine: background cursor flush failed") + + async def _flush_cursors(self) -> None: + if not self._cursor_store.has_dirty: + return + async with self._sessions() as session: + if await self._cursor_store.flush(session): + await session.commit() + # ------------------------------------------------------------------ # Event intake # ------------------------------------------------------------------ @@ -1046,6 +1157,17 @@ async def evaluate_live_event( transaction share one session so a crash between inserts cannot orphan an inbox item. """ + if not self._runtimes_for(instance_id, pane_id): + self._cursor_store.accept_in_memory( + instance_id=instance_id, + pane_id=pane_id, + stream_id=stream_id, + seq=seq, + observed_at=observed_at, + ) + if self._task is not None: + self._schedule_cursor_flush() + return [] async with self._sessions() as session: acceptance, cursor = await self._cursor_store.accept( session, diff --git a/apps/control-plane/src/termflow_control_plane/routing/terminal_router.py b/apps/control-plane/src/termflow_control_plane/routing/terminal_router.py index 1b0d2ae4..798e0473 100644 --- a/apps/control-plane/src/termflow_control_plane/routing/terminal_router.py +++ b/apps/control-plane/src/termflow_control_plane/routing/terminal_router.py @@ -75,6 +75,10 @@ def __init__( self._resume_grace_seconds = resume_grace_seconds self._suspended: dict[UUID, asyncio.Task[None]] = {} + @property + def authentication_epoch(self) -> int: + return self._hub.authentication_epoch + def _cancel_suspension(self, terminal_id: UUID) -> None: task = self._suspended.pop(terminal_id, None) if task is not None and task is not asyncio.current_task(): diff --git a/apps/control-plane/tests/test_epoch_boundaries.py b/apps/control-plane/tests/test_epoch_boundaries.py index e6b9e76f..9972e659 100644 --- a/apps/control-plane/tests/test_epoch_boundaries.py +++ b/apps/control-plane/tests/test_epoch_boundaries.py @@ -221,3 +221,54 @@ async def test_epoch_watcher_retries_after_transient_database_error(monkeypatch) finally: stop.set() await watcher + + +@pytest.mark.asyncio +async def test_terminal_hot_path_caches_epoch_without_per_frame_sqlite_queries( + tmp_path, + monkeypatch, +) -> None: + database_url = f"sqlite+aiosqlite:///{tmp_path / 'terminal-cache.db'}" + database = Database(database_url) + await database.initialize() + repositories = RepositoryBundle(database.session_factory) + terminal_hub = TerminalHub(queue_max_messages=64, queue_max_bytes=65536) + instance_id = uuid4() + terminal = await terminal_hub.register(instance_id, session_key=None, auth_epoch=1) + websocket = _FakeWebSocket( + *[{"type": "websocket.receive", "bytes": b"k"} for _ in range(20)], + {"type": "websocket.disconnect"}, + ) + terminal_router = SimpleNamespace( + authentication_epoch=1, + input=AsyncMock(), + action=AsyncMock(), + request_close=AsyncMock(), + ) + settings = Settings( + admin_token="admin-token-that-is-long-enough-for-tests", + database_url=database_url, + allow_insecure_loopback=True, + ) + db_reads = 0 + original_get = repositories.auth_state.get + + async def counting_get(): + nonlocal db_reads + db_reads += 1 + return await original_get() + + monkeypatch.setattr(repositories.auth_state, "get", counting_get) + try: + await _receive_terminal_input( + websocket, + terminal, + terminal_router, + settings, + repositories, + 1, + ) + assert terminal_router.input.await_count == 20 + assert db_reads == 1 + finally: + await database.dispose() diff --git a/apps/control-plane/tests/test_terminal_hub.py b/apps/control-plane/tests/test_terminal_hub.py index adc291b0..89b807fd 100644 --- a/apps/control-plane/tests/test_terminal_hub.py +++ b/apps/control-plane/tests/test_terminal_hub.py @@ -156,3 +156,94 @@ async def test_epoch_synchronization_closes_current_and_rejects_stale_registrati await hub.register(uuid4(), session_key=None, auth_epoch=1) replacement = await hub.register(uuid4(), session_key=None, auth_epoch=2) assert replacement.auth_epoch == 2 + + +@pytest.mark.asyncio +async def test_resume_ignores_stale_inflight_output_before_replay_handshake() -> None: + hub = TerminalHub(queue_max_messages=8, queue_max_bytes=1024) + instance_id = uuid4() + terminal = await hub.register(instance_id, session_key="browser-session") + stream_id = uuid4() + assert await hub.forward( + _message( + instance_id, + MessageType.TERMINAL_OPENED, + TerminalOpenedPayload( + terminal_id=terminal.terminal_id, + stream_id=stream_id, + rows=24, + cols=80, + ), + ) + ) + assert await terminal.next_event() is not None + + for seq in (1, 2, 3): + assert await hub.forward( + _message( + instance_id, + MessageType.TERMINAL_OUTPUT, + TerminalOutputPayload.from_bytes( + terminal.terminal_id, + stream_id, + seq, + f"chunk-{seq}".encode(), + ), + ) + ) + + resumed = await hub.resume( + instance_id, + session_key="browser-session", + terminal_id=terminal.terminal_id, + stream_id=stream_id, + after_seq=1, + ) + assert resumed is terminal + + # In-flight pre-resume frame seq=4 arrives before Computer A processes TERMINAL_OPEN(resume) + stale_inflight = _message( + instance_id, + MessageType.TERMINAL_OUTPUT, + TerminalOutputPayload.from_bytes( + terminal.terminal_id, + stream_id, + 4, + b"stale-inflight", + ), + ) + assert await hub.forward(stale_inflight) + assert not terminal.terminated + + # Computer A replays from after_seq=1 -> seq=2, 3, 4 + for seq in (2, 3, 4): + replayed = _message( + instance_id, + MessageType.TERMINAL_OUTPUT, + TerminalOutputPayload.from_bytes( + terminal.terminal_id, + stream_id, + seq, + f"replayed-{seq}".encode(), + ), + ) + assert await hub.forward(replayed) + + assert not terminal.terminated + assert terminal.resume_cursor == (stream_id, 4) + + # Once resumed stream is synced, a real gap (seq=6 skipping 5) still terminates with stream_gap + real_gap = _message( + instance_id, + MessageType.TERMINAL_OUTPUT, + TerminalOutputPayload.from_bytes( + terminal.terminal_id, + stream_id, + 6, + b"real-gap", + ), + ) + assert not await hub.forward(real_gap) + closed = await terminal.next_event() + assert isinstance(closed, LocalTerminalClose) + assert closed.reason == "stream_gap" diff --git a/apps/control-plane/tests/test_watch_engine.py b/apps/control-plane/tests/test_watch_engine.py index a816a49b..f15e4d36 100644 --- a/apps/control-plane/tests/test_watch_engine.py +++ b/apps/control-plane/tests/test_watch_engine.py @@ -547,3 +547,34 @@ async def flaky_handler(message: WireMessage) -> list: assert engine.diagnostics.wire_messages_failed == 1 finally: await engine.stop() + + +@pytest.mark.asyncio +async def test_evaluate_live_event_skips_sqlite_session_when_no_active_watches( + repositories: RepositoryBundle, +) -> None: + session_calls = 0 + + def forbidden_sessions(): + nonlocal session_calls + session_calls += 1 + raise AssertionError("SQLite session must not be opened when no watches exist") + + clock = Clock() + engine = WatchEngine( + sessions=forbidden_sessions, # type: ignore[arg-type] + watches=repositories.watches, + deliveries=repositories.agent_watch_deliveries, + inbox=repositories.agent_inbox, + clock=clock, + ) + triggers = await engine.evaluate_live_event( + instance_id=uuid4(), + pane_id="%0", + stream_id=uuid4(), + seq=1, + data=b"unwatched output", + observed_at=clock(), + ) + assert triggers == [] + assert session_calls == 0 diff --git a/apps/node/src/termflow_node/bridge/buffer.py b/apps/node/src/termflow_node/bridge/buffer.py index 6204cb7f..3c8e8acf 100644 --- a/apps/node/src/termflow_node/bridge/buffer.py +++ b/apps/node/src/termflow_node/bridge/buffer.py @@ -73,7 +73,7 @@ def append(self, data: bytes) -> OutputChunk: def replay(self, stream_id: UUID, after_seq: int) -> list[OutputChunk] | ReplayGap: if stream_id != self.stream_id: return ReplayGap(reason="stream_changed") - if after_seq < self._overwritten_through: + if after_seq < self._overwritten_through or after_seq > self.last_seq: return ReplayGap(reason="overwritten") return [chunk for chunk in self._chunks if chunk.seq > after_seq] diff --git a/apps/node/src/termflow_node/bridge/runtime.py b/apps/node/src/termflow_node/bridge/runtime.py index 8c8dca2e..bac43458 100644 --- a/apps/node/src/termflow_node/bridge/runtime.py +++ b/apps/node/src/termflow_node/bridge/runtime.py @@ -556,8 +556,13 @@ async def _publish_gap( return True async def _control_loop(self, shutdown: asyncio.Event) -> None: - async for notification in self.control.notifications(): - await self.process_notification(notification) + while not shutdown.is_set(): + try: + async for notification in self.control.notifications(): + await self.process_notification(notification) + break + except (ValueError, asyncio.LimitOverrunError): + continue shutdown.set() async def run(self, shutdown: asyncio.Event) -> None: diff --git a/apps/node/src/termflow_node/bridge/transport.py b/apps/node/src/termflow_node/bridge/transport.py index 2169f116..e1b27df2 100644 --- a/apps/node/src/termflow_node/bridge/transport.py +++ b/apps/node/src/termflow_node/bridge/transport.py @@ -41,6 +41,7 @@ def __call__( *, additional_headers: Mapping[str, str], ping_interval: float | None, + ping_timeout: float | None = ..., ) -> AbstractAsyncContextManager[WebSocketLike]: ... @@ -54,12 +55,14 @@ def _connect_websocket( uri: str, *, additional_headers: Mapping[str, str], - ping_interval: float | None, + ping_interval: float | None = 20, + ping_timeout: float | None = 20, ) -> AbstractAsyncContextManager[ClientConnection]: return websocket_connect( uri, additional_headers=additional_headers, ping_interval=ping_interval, + ping_timeout=ping_timeout, ) @@ -230,7 +233,8 @@ async def _run_connected( additional_headers={ "Authorization": f"Bearer {token.get_secret_value()}", }, - ping_interval=None, + ping_interval=20, + ping_timeout=20, ) as websocket: if self._connection_listener is not None: self._connection_listener.bridge_connected() diff --git a/apps/node/src/termflow_node/tmux/control_client.py b/apps/node/src/termflow_node/tmux/control_client.py index 23058cb2..c9f26d7d 100644 --- a/apps/node/src/termflow_node/tmux/control_client.py +++ b/apps/node/src/termflow_node/tmux/control_client.py @@ -7,18 +7,31 @@ from pathlib import Path from .capture import RenderedCapture, capture_pane_bounded -from .control_parser import ControlNotification, parse_control_line +from .control_parser import ( + ControlNotification, + MalformedControlNotification, + parse_control_line, +) from .runner import tmux_subprocess_environment +CONTROL_STREAM_LIMIT = 4 * 1024 * 1024 + class ControlClientNotStarted(RuntimeError): pass class TmuxControlClient: - def __init__(self, socket_path: Path, session_name: str) -> None: + def __init__( + self, + socket_path: Path, + session_name: str, + *, + stream_limit: int = CONTROL_STREAM_LIMIT, + ) -> None: self._socket_path = socket_path self._session_name = session_name + self._stream_limit = stream_limit self._process: asyncio.subprocess.Process | None = None self._write_lock = asyncio.Lock() @@ -37,6 +50,7 @@ async def start(self) -> None: stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, env=tmux_subprocess_environment(), + limit=self._stream_limit, ) await self.write_command("refresh-client -f pause-after=5") @@ -57,8 +71,28 @@ async def notifications(self) -> AsyncIterator[ControlNotification]: process = self._running_process() if process.stdout is None: raise ControlClientNotStarted("tmux control stdout is unavailable") - while line := await process.stdout.readline(): - yield parse_control_line(line) + discard_until_newline = False + while True: + try: + line = await process.stdout.readline() + except asyncio.LimitOverrunError as exc: + if exc.consumed > 0: + await process.stdout.read(exc.consumed) + discard_until_newline = True + continue + except ValueError as exc: + discard_until_newline = "Separator is not found" in str(exc) + continue + if not line: + break + if discard_until_newline: + if line.endswith(b"\n"): + discard_until_newline = False + continue + try: + yield parse_control_line(line) + except MalformedControlNotification: + continue async def capture_pane(self, pane_id: str) -> bytes: process = await asyncio.create_subprocess_exec( diff --git a/apps/node/tests/integration/test_tmux_actions.py b/apps/node/tests/integration/test_tmux_actions_integration.py similarity index 100% rename from apps/node/tests/integration/test_tmux_actions.py rename to apps/node/tests/integration/test_tmux_actions_integration.py diff --git a/apps/node/tests/test_bridge_transport.py b/apps/node/tests/test_bridge_transport.py index 505d7337..cadfbe72 100644 --- a/apps/node/tests/test_bridge_transport.py +++ b/apps/node/tests/test_bridge_transport.py @@ -176,7 +176,8 @@ async def connect(uri: str, **kwargs): MessageType.BRIDGE_HELLO, MessageType.TOPOLOGY_SNAPSHOT, ] - assert connection_arguments["ping_interval"] is None + assert connection_arguments["ping_interval"] == 20 + assert connection_arguments["ping_timeout"] == 20 assert connection_arguments["additional_headers"] == { "Authorization": "Bearer instance-secret-token" } diff --git a/apps/node/tests/test_control_parser.py b/apps/node/tests/test_control_parser.py index acdab2c7..7f37ef2e 100644 --- a/apps/node/tests/test_control_parser.py +++ b/apps/node/tests/test_control_parser.py @@ -1,4 +1,9 @@ +import asyncio +from pathlib import Path +from types import SimpleNamespace + import pytest +from termflow_node.tmux.control_client import CONTROL_STREAM_LIMIT, TmuxControlClient from termflow_node.tmux.control_parser import ( GenericNotification, MalformedControlNotification, @@ -26,3 +31,35 @@ def test_extended_output_ignores_flow_control_metadata() -> None: def test_malformed_output_escape_is_rejected(line: bytes) -> None: with pytest.raises(MalformedControlNotification): parse_control_line(line) + + +@pytest.mark.asyncio +async def test_control_client_handles_lines_over_64kib_and_drains_overrun_chunks() -> None: + assert CONTROL_STREAM_LIMIT >= 4 * 1024 * 1024 + + # 1. Verify a >64 KiB line succeeds under the 4 MiB stream limit + reader_large = asyncio.StreamReader(limit=CONTROL_STREAM_LIMIT) + large_payload = b"x" * 100_000 + reader_large.feed_data(b"%output %1 " + large_payload + b"\n") + reader_large.feed_eof() + client_large = TmuxControlClient(Path("/tmp/fake.sock"), "main") + client_large._process = SimpleNamespace(stdout=reader_large) # type: ignore[assignment] + events_large = [item async for item in client_large.notifications()] + assert events_large == [OutputNotification("%1", large_payload)] + + # 2. Verify a chunk exceeding the stream limit is drained without crashing notifications() + reader_small = asyncio.StreamReader(limit=64) + # Chunk without newline exceeding limit, followed by tail with newline, followed by valid line + reader_small.feed_data(b"%output %1 " + b"a" * 100) + client_small = TmuxControlClient(Path("/tmp/fake.sock"), "main", stream_limit=64) + client_small._process = SimpleNamespace(stdout=reader_small) # type: ignore[assignment] + + async def feed_rest() -> None: + await asyncio.sleep(0.01) + reader_small.feed_data(b"tail\n%output %1 recovered\n") + reader_small.feed_eof() + + feeder = asyncio.create_task(feed_rest()) + events_small = [item async for item in client_small.notifications()] + await feeder + assert events_small == [OutputNotification("%1", b"recovered")] diff --git a/apps/node/tests/test_output_buffer.py b/apps/node/tests/test_output_buffer.py index e5248724..39629f8a 100644 --- a/apps/node/tests/test_output_buffer.py +++ b/apps/node/tests/test_output_buffer.py @@ -16,9 +16,10 @@ def test_buffer_replays_by_stream_and_sequence() -> None: def test_overwrite_reports_gap() -> None: buffer = PaneOutputBuffer(max_bytes=4) old = buffer.append(b"abc") - buffer.append(b"def") + latest = buffer.append(b"def") gap = buffer.replay(old.stream_id, 0) assert gap == ReplayGap(reason="overwritten") + assert buffer.replay(latest.stream_id, latest.seq + 1) == ReplayGap(reason="overwritten") def test_oversized_chunk_keeps_only_tail_and_marks_chunk_unreplayable() -> None: diff --git a/packages/client-core/src/terminal/session.test.ts b/packages/client-core/src/terminal/session.test.ts index f0b9670d..d6fbacc6 100644 --- a/packages/client-core/src/terminal/session.test.ts +++ b/packages/client-core/src/terminal/session.test.ts @@ -231,4 +231,53 @@ describe('TerminalSession', () => { await session.dispose() } }) + + it.each(['instance_offline', 'capability_unavailable', 'pane_not_found', 'resume_expired'])( + 'handles pre-ready terminal.error (%s) on zero terminal_id and suppresses auto-reconnect', + async (errorCode) => { + const optionError = vi.fn() + const { session, transport, scheduler, callbacks } = await setup({ onError: optionError }) + transport.emit({ type: 'open' }) + transport.emit({ + type: 'text', + data: JSON.stringify({ + type: 'terminal.error', + terminal_id: '00000000-0000-0000-0000-000000000000', + code: errorCode, + message: 'Open rejected.', + }), + }) + transport.emit({ type: 'close', code: 4409 }) + + expect(callbacks.onError).toHaveBeenCalledWith({ code: errorCode, message: 'Open rejected.' }) + expect(optionError).toHaveBeenCalledWith(errorCode) + expect(callbacks.onStatus).toHaveBeenLastCalledWith('closed') + expect(scheduler.pending).toHaveLength(0) + await session.dispose() + }, + ) + + it('handles pre-ready terminal.closed frames during reconnect when terminal_id matches', async () => { + const { session, transport, scheduler, callbacks } = await setup() + ready(transport) + transport.emit({ type: 'close', code: 1006 }) + scheduler.runNext() + await Promise.resolve() + + // While reconnecting (!ready), server sends terminal.closed for the known terminalId + transport.emit({ + type: 'text', + data: JSON.stringify({ + type: 'terminal.closed', + terminal_id: TERMINAL_1, + reason: 'replaced', + }), + }) + transport.emit({ type: 'close', code: 1000 }) + + expect(callbacks.onClosed).toHaveBeenCalledWith('replaced') + expect(callbacks.onStatus).toHaveBeenLastCalledWith('closed') + expect(scheduler.pending.filter((h) => !h.cancelled)).toHaveLength(0) + await session.dispose() + }) }) diff --git a/packages/client-core/src/terminal/session.ts b/packages/client-core/src/terminal/session.ts index e42590e4..b5550385 100644 --- a/packages/client-core/src/terminal/session.ts +++ b/packages/client-core/src/terminal/session.ts @@ -42,6 +42,7 @@ export interface TerminalSessionOptions { * Optional authentication probe used before a reconnect attempt. */ probeSession?: () => Promise + onError?: (code: string) => void } export interface TerminalSessionLike { @@ -52,6 +53,13 @@ export interface TerminalSessionLike { } const MAX_BINARY_FRAME = 65_536 +const ZERO_TERMINAL_ID = '00000000-0000-0000-0000-000000000000' +const NON_RECONNECTABLE_ERROR_CODES = new Set([ + 'instance_offline', + 'capability_unavailable', + 'pane_not_found', + 'resume_expired', +]) export class TerminalSession implements TerminalSessionLike { private connection: TerminalConnection | null = null @@ -131,7 +139,9 @@ export class TerminalSession implements TerminalSessionLike { private handleText(data: string): void { const control = parseTerminalControl(data) if (control === null) return - if (control.type !== 'terminal.ready' && (!this.ready || control.terminal_id !== this.terminalId)) return + const isTerminalLifecycle = control.type === 'terminal.error' || control.type === 'terminal.closed' + const matchesTerminal = control.terminal_id === this.terminalId || control.terminal_id === ZERO_TERMINAL_ID + if (control.type !== 'terminal.ready' && !(isTerminalLifecycle ? matchesTerminal : (this.ready && control.terminal_id === this.terminalId))) return switch (control.type) { case 'terminal.ready': @@ -159,7 +169,14 @@ export class TerminalSession implements TerminalSessionLike { this.callbacks.onBindings(control) break case 'terminal.error': + if (NON_RECONNECTABLE_ERROR_CODES.has(control.code)) { + this.stopHeartbeat() + this.ready = false + this.suppressReconnect = true + this.callbacks.onStatus('closed') + } this.callbacks.onError({ code: control.code, message: control.message }) + this.options.onError?.(control.code) break case 'terminal.action_result': this.callbacks.onActionResult(control) @@ -167,7 +184,9 @@ export class TerminalSession implements TerminalSessionLike { case 'terminal.closed': this.stopHeartbeat() this.ready = false - this.suppressReconnect = control.reason === 'replaced' || control.reason === 'instance_offline' || control.reason === 'client_closed' + if (control.reason === 'replaced' || control.reason === 'instance_offline' || control.reason === 'client_closed') { + this.suppressReconnect = true + } this.callbacks.onClosed(control.reason) this.callbacks.onStatus('closed') if (this.connection !== null) void this.connection.close(1000, control.reason) diff --git a/packages/client-ui/package.json b/packages/client-ui/package.json index ea677035..b3914a72 100644 --- a/packages/client-ui/package.json +++ b/packages/client-ui/package.json @@ -3,6 +3,9 @@ "version": "0.2.0", "private": true, "type": "module", + "sideEffects": [ + "*.css" + ], "engines": { "node": ">=22 <23" }, diff --git a/packages/client-ui/src/router/routes.ts b/packages/client-ui/src/router/routes.ts index 1538820d..8d4f716d 100644 --- a/packages/client-ui/src/router/routes.ts +++ b/packages/client-ui/src/router/routes.ts @@ -1,28 +1,16 @@ import type { RouteRecordRaw } from 'vue-router' -import AgentPreviewView from '../views/AgentPreviewView.vue' -import AgentChatView from '../views/AgentChatView.vue' -import AgentView from '../views/AgentView.vue' -import ComputersView from '../views/ComputersView.vue' -import DashboardView from '../views/DashboardView.vue' -import LoginView from '../views/LoginView.vue' -import NativeAuthorizeView from '../views/NativeAuthorizeView.vue' -import DeviceAuthorizeView from '../views/DeviceAuthorizeView.vue' -import NotFoundView from '../views/NotFoundView.vue' -import TerminalView from '../views/TerminalView.vue' -import SettingsView from '../views/SettingsView.vue' -import TotpActivationView from '../views/TotpActivationView.vue' export const clientRoutes: RouteRecordRaw[] = [ - { path: '/agent-preview', component: AgentPreviewView, meta: { bare: true } }, - { path: '/login', component: LoginView, meta: { bare: true } }, - { path: '/authorize', component: NativeAuthorizeView, meta: { bare: true } }, - { path: '/device', component: DeviceAuthorizeView, meta: { bare: true, requiresAuth: true } }, - { path: '/', component: DashboardView, meta: { requiresAuth: true } }, - { path: '/computers', component: ComputersView, meta: { requiresAuth: true } }, - { path: '/settings', component: SettingsView, meta: { requiresAuth: true } }, - { path: '/settings/two-factor-auth', component: TotpActivationView, meta: { requiresAuth: true, webOnly: true } }, - { path: '/agent', component: AgentView, meta: { requiresAuth: true } }, - { path: '/agent/:conversationId', component: AgentChatView, meta: { requiresAuth: true } }, - { path: '/terms/:termId', component: TerminalView, meta: { requiresAuth: true, terminal: true } }, - { path: '/:pathMatch(.*)*', component: NotFoundView }, + { path: '/agent-preview', component: () => import('../views/AgentPreviewView.vue'), meta: { bare: true } }, + { path: '/login', component: () => import('../views/LoginView.vue'), meta: { bare: true } }, + { path: '/authorize', component: () => import('../views/NativeAuthorizeView.vue'), meta: { bare: true } }, + { path: '/device', component: () => import('../views/DeviceAuthorizeView.vue'), meta: { bare: true, requiresAuth: true } }, + { path: '/', component: () => import('../views/DashboardView.vue'), meta: { requiresAuth: true } }, + { path: '/computers', component: () => import('../views/ComputersView.vue'), meta: { requiresAuth: true } }, + { path: '/settings', component: () => import('../views/SettingsView.vue'), meta: { requiresAuth: true } }, + { path: '/settings/two-factor-auth', component: () => import('../views/TotpActivationView.vue'), meta: { requiresAuth: true, webOnly: true } }, + { path: '/agent', component: () => import('../views/AgentView.vue'), meta: { requiresAuth: true } }, + { path: '/agent/:conversationId', component: () => import('../views/AgentChatView.vue'), meta: { requiresAuth: true } }, + { path: '/terms/:termId', component: () => import('../views/TerminalView.vue'), meta: { requiresAuth: true, terminal: true } }, + { path: '/:pathMatch(.*)*', component: () => import('../views/NotFoundView.vue') }, ] diff --git a/pyproject.toml b/pyproject.toml index ba6a538c..f1153780 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,6 +6,7 @@ index-url = "https://pypi.org/simple" [tool.pytest.ini_options] asyncio_mode = "auto" +pythonpath = ["."] addopts = ["--import-mode=importlib"] testpaths = ["packages/protocol/tests", "apps/control-plane/tests", "apps/node/tests", "tests"] markers = [ @@ -23,3 +24,6 @@ select = ["E", "F", "I", "UP", "B", "ASYNC"] [tool.mypy] python_version = "3.12" strict = true +explicit_package_bases = true +mypy_path = ["packages/protocol/src", "apps/control-plane/src", "apps/node/src"] +exclude = ["(^|/)tests/"] diff --git a/scripts/verify.sh b/scripts/verify.sh index 6c6f7a8c..287d63cf 100755 --- a/scripts/verify.sh +++ b/scripts/verify.sh @@ -5,9 +5,8 @@ SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" REPOSITORY_ROOT="$(cd -- "${SCRIPT_DIR}/.." && pwd)" cd "${REPOSITORY_ROOT}" -EXPECTED_NODE_VERSION="v22.23.2" -if [[ "$(node --version)" != "${EXPECTED_NODE_VERSION}" ]]; then - echo "TermFlow verification requires Node ${EXPECTED_NODE_VERSION}; found $(node --version)." >&2 +if [[ ! "$(node --version)" =~ ^v22\. ]]; then + echo "TermFlow verification requires Node 22.x; found $(node --version)." >&2 exit 1 fi