Release: broadcast terminal output to every viewer (#5836, @ai-ag2026) (#6340)

* feat(terminal): broadcast output to every viewer instead of one shared queue

TerminalSession.output was a single queue.Queue read destructively by the SSE
handler. Two tabs/windows viewing the SAME session each open their own
EventSource, so two _handle_terminal_output handlers competed on that one queue:
every PTY chunk was delivered to exactly one of them. Each tab saw a disjoint
half of the byte stream, and only one ever received terminal_closed.

Output now fans out, mirroring StreamChannel/SessionChannel: each SSE consumer
subscribe()s its own queue (seeded with a bounded backlog so a first/late attach
still replays the recent scrollback, preserving the old buffer-until-first-
consumer behaviour), and put_output broadcasts to all subscribers. A slow
viewer's queue drops its own oldest chunk (drop-oldest, isolated per subscriber)
so one lagging tab can't starve another. The handler unsubscribes in a finally
so the subscriber list can't grow.

Stacked on the terminal fd-leak fix (same file).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix: cover terminal broadcast lifecycle

* fix: serialize terminal subscriber fanout

* test: cover terminal unsubscribe publication race

* restore timing-flaky pytest.skip on test_terminal_survives_short_lived_request_thread (keep API-updated body); changelog #5836

---------

Co-authored-by: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: nesquena-hermes <agent@nesquena-hermes>

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)

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
 
 

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] = {}

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, (

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)
 

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")]