diff --git a/CHANGELOG.md b/CHANGELOG.md index 010a3db1..66c29ad9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,8 @@ ### Fixed +- **Watching the same terminal from two tabs no longer splits the output stream.** Terminal output is now broadcast to every subscribed viewer instead of draining from one shared queue, so a second tab (or a reconnect) sees the full stream rather than a random half. SSE responses carry event `id:`s and a reconnecting viewer replays only events after its `Last-Event-ID` cursor (fresh viewers still get the full bounded backlog). Thanks @ai-ag2026. (#5836) + - **A gateway command-approval card with no `approval_id` is now respondable.** When a gateway `approval.request` omits both `approval_id` and the legacy `id`, WebUI now synthesizes a non-empty `uuid4().hex` id instead of leaving it blank, so the approval card can be answered (matching the `#527` stable-uuid design). The downstream respond path resolves by `run_id` independently, so the synthetic id causes no mismatch; a provided or legacy id is preserved verbatim. Thanks @webtecnica. (#6168, #6008) - **The update check no longer lets an unexpected exception escape the route.** `POST /api/updates/check` now wraps `check_for_updates()` in a defensive guard that returns a clean `500` (and logs) instead of propagating, giving the endpoint the same exception protection as its siblings. Defensive hardening only — it does not change the success path. Thanks @webtecnica. (#6180) diff --git a/api/routes.py b/api/routes.py index 4494619d..699289ef 100644 --- a/api/routes.py +++ b/api/routes.py @@ -17726,22 +17726,36 @@ def _handle_terminal_output(handler, parsed): handler.send_header("Connection", "close") end_sse_headers(handler) _sse_set_write_deadline(handler) # Defect A: slow tab can't pin this thread + # EventSource automatically returns the last received SSE id on transport + # reconnect. Seed only newer backlog entries in that case so already-rendered + # terminal bytes (including ANSI cursor controls) are not written twice. A + # genuinely new viewer has no cursor and receives the full bounded backlog. + after_seq = None + last_event_id = str(handler.headers.get("Last-Event-ID", "") or "").strip() + if last_event_id: + try: + after_seq = max(0, int(last_event_id)) + except ValueError: + pass + output = term.subscribe(after_seq=after_seq) try: while True: try: - event, data = term.output.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS) + event_seq, event, data = output.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS) except queue.Empty: handler.wfile.write(b": terminal heartbeat\n\n") handler.wfile.flush() - if term.closed.is_set() and term.output.empty(): + if term.closed.is_set() and output.empty(): _sse(handler, "terminal_closed", {"exit_code": term.proc.poll()}) break continue - _sse(handler, event, data) + _sse_with_id(handler, event, data, event_id=event_seq) if event in ("terminal_closed", "terminal_error"): break except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): pass + finally: + term.unsubscribe(output) return True diff --git a/api/terminal.py b/api/terminal.py index 44e4a12a..2e6000c2 100644 --- a/api/terminal.py +++ b/api/terminal.py @@ -11,6 +11,7 @@ from __future__ import annotations import errno import atexit import codecs +import collections import os import queue import shutil @@ -53,6 +54,12 @@ def _safe_close_fd(fd: int) -> None: pass +# Bounds both the replay backlog and each subscriber's live queue: how many +# output chunks are buffered before the oldest is dropped. One value so the +# catch-up backlog and the per-viewer queue stay in lockstep. +_OUTPUT_BUFFER_MAXLEN = 2000 + + @dataclass class TerminalSession: session_id: str @@ -61,7 +68,18 @@ class TerminalSession: master_fd: int rows: int = 24 cols: int = 80 - output: queue.Queue = field(default_factory=lambda: queue.Queue(maxsize=2000)) + # Output is fanned out to every attached viewer. Each SSE consumer + # subscribe()s its own queue; put_output broadcasts to all of them plus a + # bounded backlog that a newly attaching consumer replays. Two tabs/windows + # on the same session therefore each get the FULL byte stream — previously a + # single shared queue was read destructively, so two consumers split the + # output between them and only one saw terminal_closed. + _subscribers: list = field(default_factory=list) + _backlog: collections.deque = field( + default_factory=lambda: collections.deque(maxlen=_OUTPUT_BUFFER_MAXLEN) + ) + _sub_lock: threading.Lock = field(default_factory=threading.Lock) + _next_output_seq: int = 1 closed: threading.Event = field(default_factory=threading.Event) reader: threading.Thread | None = None # Serializes fd-touching ops (os.write, resize ioctl) against os.close, so a @@ -79,20 +97,53 @@ class TerminalSession: def is_alive(self) -> bool: return not self.closed.is_set() and self.proc.poll() is None + def subscribe(self, after_seq: int | None = None) -> queue.Queue: + """Attach a viewer: return a queue seeded with the current backlog and + registered to receive all subsequent output. + + A reconnecting EventSource supplies its last received sequence so only + newer backlog entries are replayed. A new viewer leaves ``after_seq`` + unset and receives the full bounded backlog. + """ + q: queue.Queue = queue.Queue(maxsize=_OUTPUT_BUFFER_MAXLEN) + with self._sub_lock: + for item in self._backlog: + if after_seq is None or item[0] > after_seq: + q.put_nowait(item) + self._subscribers.append(q) + return q + + def unsubscribe(self, q: queue.Queue) -> None: + with self._sub_lock: + try: + self._subscribers.remove(q) + except ValueError: + pass + def put_output(self, event: str, payload: dict) -> None: self.last_activity = time.time() - try: - self.output.put_nowait((event, payload)) - except queue.Full: - # Keep the terminal responsive by dropping the oldest queued chunk. - try: - self.output.get_nowait() - except queue.Empty: - pass - try: - self.output.put_nowait((event, payload)) - except queue.Full: - pass + with self._sub_lock: + item = (self._next_output_seq, event, payload) + self._next_output_seq += 1 + self._backlog.append(item) + # Keep sequence assignment, backlog append, and non-blocking fanout in + # one publication order. Releasing this lock before fanout lets two + # producers enqueue seq N+1 before seq N to a live subscriber. + for q in self._subscribers: + try: + q.put_nowait(item) + except queue.Full: + # Slow viewer: drop its oldest chunk to stay responsive. + # Isolated per subscriber, so one lagging tab can't starve + # another; all queue operations remain non-blocking. + try: + q.get_nowait() + except queue.Empty: + pass + try: + q.put_nowait(item) + except queue.Full: + pass _TERMINALS: dict[str, TerminalSession] = {} diff --git a/tests/test_issue1623_sse_heartbeat_alignment.py b/tests/test_issue1623_sse_heartbeat_alignment.py index 90109a6c..db775d9f 100644 --- a/tests/test_issue1623_sse_heartbeat_alignment.py +++ b/tests/test_issue1623_sse_heartbeat_alignment.py @@ -72,7 +72,7 @@ def test_each_named_sse_handler_uses_constant(): expected_callers = [ "subscriber.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS)", # main agent SSE - "term.output.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS)", # terminal SSE + "output.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS)", # terminal SSE (per-consumer queue) ] for caller in expected_callers: assert caller in src, ( diff --git a/tests/test_terminal_linux_lifecycle.py b/tests/test_terminal_linux_lifecycle.py index 01495225..581f6286 100644 --- a/tests/test_terminal_linux_lifecycle.py +++ b/tests/test_terminal_linux_lifecycle.py @@ -50,12 +50,13 @@ def test_terminal_survives_short_lived_request_thread(tmp_path): assert term.is_alive() marker = f"lifecycle-ok-{os.getpid()}" + out = term.subscribe() write_terminal(sid, f"printf '{marker}\\n'\n") deadline = time.monotonic() + 1.0 seen = "" while time.monotonic() < deadline: try: - event, payload = term.output.get(timeout=0.1) + _seq, event, payload = out.get(timeout=0.1) except queue.Empty: continue if event == "output": @@ -63,6 +64,7 @@ def test_terminal_survives_short_lived_request_thread(tmp_path): if f"{marker}\r\n" in seen or f"{marker}\n" in seen: break assert f"{marker}\r\n" in seen or f"{marker}\n" in seen + term.unsubscribe(out) finally: close_terminal(sid) diff --git a/tests/test_terminal_output_broadcast.py b/tests/test_terminal_output_broadcast.py new file mode 100644 index 00000000..4d357e64 --- /dev/null +++ b/tests/test_terminal_output_broadcast.py @@ -0,0 +1,379 @@ +"""Regression tests — terminal output is broadcast to every attached viewer. + +Previously `TerminalSession.output` was a single `queue.Queue` read +destructively by the SSE handler. Two tabs/windows viewing the SAME session +each opened their own EventSource -> two handlers competed on that one queue, so +every PTY chunk went to exactly one of them: each tab saw a disjoint half of the +stream and only one saw `terminal_closed`. + +Output now fans out: each consumer `subscribe()`s its own queue (seeded with a +bounded backlog so a late/first attach still catches up), and `put_output` +broadcasts to all of them. These pin that two subscribers each receive the full +stream, that a late subscriber replays the backlog, and that a slow subscriber +can't starve another. +""" +import io +import os +import queue +import threading +from types import SimpleNamespace + +import pytest + +if os.name != "posix": + pytest.skip("terminal tests require POSIX terminal support", allow_module_level=True) + +import api.terminal as terminal +from api import routes + + +def _make_term(sid="bcast"): + class _Proc: + pid = 4242 + + def poll(self): + return None + + return terminal.TerminalSession( + session_id=sid, workspace="/tmp", proc=_Proc(), master_fd=-1 + ) + + +def _drain(q): + out = [] + while True: + try: + out.append(q.get_nowait()) + except queue.Empty: + return out + + +def test_two_subscribers_each_get_the_full_stream(): + term = _make_term() + a = term.subscribe() + b = term.subscribe() + + for i in range(5): + term.put_output("output", {"text": f"chunk-{i}"}) + + got_a = [p["text"] for _seq, _e, p in _drain(a)] + got_b = [p["text"] for _seq, _e, p in _drain(b)] + expected = [f"chunk-{i}" for i in range(5)] + assert got_a == expected, "subscriber A missed chunks (stream was split)" + assert got_b == expected, "subscriber B missed chunks (stream was split)" + + +def test_terminal_closed_reaches_all_subscribers(): + term = _make_term() + a = term.subscribe() + b = term.subscribe() + + term.put_output("terminal_closed", {"exit_code": 0}) + + assert [e for _seq, e, _p in _drain(a)] == ["terminal_closed"] + assert [e for _seq, e, _p in _drain(b)] == ["terminal_closed"], ( + "second tab did not receive terminal_closed exactly once" + ) + + +def test_late_subscriber_replays_backlog(): + term = _make_term() + # Output produced before any viewer attaches (e.g. the initial shell prompt). + for i in range(3): + term.put_output("output", {"text": f"pre-{i}"}) + + late = term.subscribe() + term.put_output("output", {"text": "live"}) + + got = [p["text"] for _seq, _e, p in _drain(late)] + assert got == ["pre-0", "pre-1", "pre-2", "live"], ( + "late subscriber did not replay the backlog then receive live output" + ) + + +def test_reconnecting_subscriber_replays_only_events_after_cursor(): + term = _make_term() + first = term.subscribe() + term.put_output("output", {"text": "A"}) + first_items = _drain(first) + assert [payload["text"] for _seq, _event, payload in first_items] == ["A"] + cursor = first_items[-1][0] + term.unsubscribe(first) + + term.put_output("output", {"text": "B"}) + term.put_output("output", {"text": "C"}) + reconnect = term.subscribe(after_seq=cursor) + + replayed = _drain(reconnect) + assert [seq for seq, _event, _payload in replayed] == [cursor + 1, cursor + 2] + assert [payload["text"] for _seq, _event, payload in replayed] == ["B", "C"] + + new_viewer = term.subscribe() + assert [payload["text"] for _seq, _event, payload in _drain(new_viewer)] == [ + "A", + "B", + "C", + ] + + +def test_sse_reconnect_honors_last_event_id_and_emits_ids(monkeypatch): + term = _make_term("sse-cursor") + term.put_output("output", {"text": "A"}) + term.put_output("output", {"text": "B"}) + term.put_output("terminal_closed", {"exit_code": 0}) + + class _Handler: + headers = {"Last-Event-ID": "1"} + + def __init__(self): + self.wfile = io.BytesIO() + self.status = None + + def send_response(self, status): + self.status = status + + def send_header(self, _name, _value): + pass + + def end_headers(self): + pass + + monkeypatch.setattr(routes, "_embedded_terminal_gate_allows", lambda _handler: True) + monkeypatch.setattr(routes, "_sse_set_write_deadline", lambda _handler: None) + monkeypatch.setitem(terminal._TERMINALS, term.session_id, term) + handler = _Handler() + + routes._handle_terminal_output( + handler, + SimpleNamespace(query=f"session_id={term.session_id}"), + ) + + body = handler.wfile.getvalue().decode("utf-8") + assert handler.status == 200 + assert '"text": "A"' not in body + assert '"text": "B"' in body + assert "id: 2\n" in body + assert "id: 3\n" in body + assert term._subscribers == [] + + +def test_sse_heartbeat_is_a_valid_comment_and_cleans_up(monkeypatch): + term = _make_term("sse-heartbeat") + term.closed.set() + + class _Handler: + headers = {} + + def __init__(self): + self.wfile = io.BytesIO() + self.status = None + + def send_response(self, status): + self.status = status + + def send_header(self, _name, _value): + pass + + def end_headers(self): + pass + + monkeypatch.setattr(routes, "_embedded_terminal_gate_allows", lambda _handler: True) + monkeypatch.setattr(routes, "_sse_set_write_deadline", lambda _handler: None) + monkeypatch.setattr(routes, "_SSE_HEARTBEAT_INTERVAL_SECONDS", 0) + monkeypatch.setitem(terminal._TERMINALS, term.session_id, term) + handler = _Handler() + + routes._handle_terminal_output( + handler, + SimpleNamespace(query=f"session_id={term.session_id}"), + ) + + body = handler.wfile.getvalue().decode("utf-8") + assert handler.status == 200 + assert body.startswith(": terminal heartbeat\n\n") + assert "event: terminal_closed\n" in body + assert "id: None" not in body + assert term._subscribers == [] + + +def test_unsubscribe_stops_delivery_and_shrinks_list(): + term = _make_term() + a = term.subscribe() + b = term.subscribe() + assert len(term._subscribers) == 2 + + term.unsubscribe(a) + assert len(term._subscribers) == 1 + + term.put_output("output", {"text": "after-unsub"}) + assert _drain(a) == [] # a got nothing after unsubscribing + assert [p["text"] for _seq, _e, p in _drain(b)] == ["after-unsub"] + + +def test_slow_subscriber_drops_oldest_without_starving_others(): + term = _make_term() + slow = term.subscribe() + fast = term.subscribe() + + # Overflow the slow subscriber's queue while draining fast. + total = terminal._OUTPUT_BUFFER_MAXLEN + 50 + fast_items = [] + for i in range(total): + term.put_output("output", {"text": f"c{i}"}) + # Keep 'fast' drained so it never overflows. + fast_items.append(fast.get_nowait()) + + # Slow subscriber is capped and kept the newest chunks (drop-oldest). + slow_items = _drain(slow) + assert len(slow_items) == terminal._OUTPUT_BUFFER_MAXLEN + assert slow_items[-1][2]["text"] == f"c{total - 1}", "slow queue didn't keep newest" + assert [seq for seq, _event, _payload in fast_items] == list(range(1, total + 1)) + assert [payload["text"] for _seq, _event, payload in fast_items] == [ + f"c{i}" for i in range(total) + ] + + +def test_unsubscribe_unknown_queue_is_safe(): + term = _make_term() + known = term.subscribe() + term.unsubscribe(known) + term.unsubscribe(known) # double-unsubscribe must be idempotent + stray: queue.Queue = queue.Queue() + term.unsubscribe(stray) # must not raise + assert term._subscribers == [] + + +def test_concurrent_producers_keep_monotonic_sequences_through_backlog_rollover(): + term = _make_term() + live = term.subscribe() + producer_count = 4 + events_per_producer = 600 + barrier = threading.Barrier(producer_count) + + def produce(producer_id): + barrier.wait(timeout=1.0) + for index in range(events_per_producer): + term.put_output( + "output", + {"text": f"producer-{producer_id}-{index}"}, + ) + + threads = [ + threading.Thread(target=produce, args=(producer_id,)) + for producer_id in range(producer_count) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=3.0) + assert not thread.is_alive() + + total = producer_count * events_per_producer + backlog = list(term._backlog) + assert len(backlog) == terminal._OUTPUT_BUFFER_MAXLEN + assert [seq for seq, _event, _payload in backlog] == list( + range(total - terminal._OUTPUT_BUFFER_MAXLEN + 1, total + 1) + ) + assert [seq for seq, _event, _payload in _drain(live)] == list( + range(total - terminal._OUTPUT_BUFFER_MAXLEN + 1, total + 1) + ) + assert term._next_output_seq == total + 1 + + cursor = backlog[0][0] + replay = _drain(term.subscribe(after_seq=cursor)) + assert [seq for seq, _event, _payload in replay] == list( + range(cursor + 1, total + 1) + ) + + +def test_concurrent_producers_keep_live_subscriber_delivery_order(): + term = _make_term() + first_delivery_started = threading.Event() + second_delivery_completed = threading.Event() + release_first_delivery = threading.Event() + + class _PausingQueue(queue.Queue): + def put_nowait(self, item): + if item[0] == 1: + first_delivery_started.set() + assert release_first_delivery.wait(timeout=1.0) + result = super().put_nowait(item) + if item[0] == 2: + second_delivery_completed.set() + return result + + subscriber = _PausingQueue(maxsize=terminal._OUTPUT_BUFFER_MAXLEN) + with term._sub_lock: + term._subscribers.append(subscriber) + + first = threading.Thread( + target=term.put_output, + args=("output", {"text": "first"}), + ) + second = threading.Thread( + target=term.put_output, + args=("output", {"text": "second"}), + ) + first.start() + assert first_delivery_started.wait(timeout=1.0) + second.start() + second_delivered_before_release = second_delivery_completed.wait(timeout=0.1) + release_first_delivery.set() + first.join(timeout=1.0) + second.join(timeout=1.0) + + assert not first.is_alive() + assert not second.is_alive() + assert not second_delivered_before_release + assert [seq for seq, _event, _payload in _drain(subscriber)] == [1, 2] + + +def test_unsubscribe_waits_for_inflight_delivery_then_stops_future_output(): + term = _make_term() + delivery_started = threading.Event() + release_delivery = threading.Event() + unsubscribe_started = threading.Event() + unsubscribe_completed = threading.Event() + + class _PausingQueue(queue.Queue): + def put_nowait(self, item): + if item[0] == 1: + delivery_started.set() + assert release_delivery.wait(timeout=1.0) + return super().put_nowait(item) + + subscriber = _PausingQueue(maxsize=terminal._OUTPUT_BUFFER_MAXLEN) + with term._sub_lock: + term._subscribers.append(subscriber) + + publisher = threading.Thread( + target=term.put_output, + args=("output", {"text": "in-flight"}), + ) + + def unsubscribe(): + unsubscribe_started.set() + term.unsubscribe(subscriber) + unsubscribe_completed.set() + + remover = threading.Thread(target=unsubscribe) + publisher.start() + assert delivery_started.wait(timeout=1.0) + remover.start() + assert unsubscribe_started.wait(timeout=1.0) + assert not unsubscribe_completed.wait(timeout=0.1) + + release_delivery.set() + publisher.join(timeout=1.0) + remover.join(timeout=1.0) + + assert not publisher.is_alive() + assert not remover.is_alive() + assert unsubscribe_completed.is_set() + assert term._subscribers == [] + + term.put_output("output", {"text": "after-unsubscribe"}) + assert [ + (seq, payload["text"]) + for seq, _event, payload in _drain(subscriber) + ] == [(1, "in-flight")]