Multi-Client Realtime and Cursor Replay

Fan out persisted events to multiple browser tabs, bound slow consumers, and reconnect from a durable SQLite cursor.

Chapter 03 streamed one run to its initiating browser. A production-shaped session can also be open in another tab or device, and the run should continue when the initiating socket disappears. This chapter adds per-session fan-out, bounded observer queues, and replay from SQLite so connection state never replaces durable session state.

Persist first, publish second

SessionBroker carries dictionaries that already contain event ids and database cursors. AutocodeApplication appends through SessionRepository before calling publish, so a frame cannot reach a browser before the corresponding event is recoverable. The broker is deliberately ephemeral: after a process restart, reconnect reads the missing suffix from SQLite rather than asking an in-memory queue to remember history.

from autocode.realtime import SessionBroker

broker = SessionBroker(max_queue=2)
left_id, left = broker.subscribe("session-1")
right_id, right = broker.subscribe("session-1")

first = {"event_id": "e-1", "cursor": 1, "kind": "text_delta", "payload": {"content": "a"}}
broker.publish("session-1", first)

assert left.get_nowait() == first
assert right.get_nowait() == first
assert broker.observer_count("session-1") == 2
broker.unsubscribe("session-1", left_id)
broker.unsubscribe("session-1", right_id)
print("fan-out observers received cursor", first["cursor"])

fan-out observers received cursor 1

Both observers receive the same event dictionary, including the stable id and cursor created by persistence. The broker does not assign a second realtime cursor. One ordering domain across database reads and live frames keeps the browser’s deduplication rule simple.

Two browser observers converge on one run

The FastAPI WebSocket endpoint subscribes each accepted connection to the session broker. Sending a user_message command starts one background task stored by session id. That task is not owned by the socket lifecycle, so closing the initiating browser does not erase the run. Any connected observer receives the published events, and a later observer replays them by cursor.

from tempfile import TemporaryDirectory

from fastapi.testclient import TestClient

from autocode.runner import DemoAgentRunner
from autocode_service.api import create_app

with TemporaryDirectory() as directory:
    app = create_app(
        database_path=f"{directory}/sessions.db",
        runner=DemoAgentRunner(),
    )
    with TestClient(app) as client:
        session_id = client.post("/api/sessions", json={"title": "two observers"}).json()["session_id"]
        with (
            client.websocket_connect(f"/ws/sessions/{session_id}") as left,
            client.websocket_connect(f"/ws/sessions/{session_id}") as right,
        ):
            left.send_json({"type": "user_message", "content": "fan this out"})
            left_events, right_events = [], []
            while not left_events or left_events[-1]["kind"] != "run_finished":
                left_events.append(left.receive_json())
            while not right_events or right_events[-1]["kind"] != "run_finished":
                right_events.append(right.receive_json())

assert [event["event_id"] for event in left_events] == [
    event["event_id"] for event in right_events
]
print("observers converged on", len(left_events), "events")

observers converged on 10 events

The comparison uses event identifiers rather than visible final text. Two browsers can show the same answer while one silently missed a tool event. Exact event-set equality proves both projections observed the same durable run.

Reconnect uses the last applied cursor

The browser tracks the largest durable cursor it has applied and reconnects with ?after=<cursor>. The server loads that suffix before joining live publication. A race can repeat an event around the replay/live boundary, so event-id deduplication remains required. Missing an event is not acceptable; receiving one twice is harmless when projection is idempotent.

Every observer queue is bounded. If a client cannot keep up, the broker evicts it and the client recovers through SQLite replay. This protects the agent run from an unbounded memory queue.

from tempfile import TemporaryDirectory

from fastapi.testclient import TestClient

from autocode.runner import DemoAgentRunner
from autocode_service.api import create_app

with TemporaryDirectory() as directory:
    app = create_app(database_path=f"{directory}/sessions.db", runner=DemoAgentRunner())
    with TestClient(app) as client:
        session_id = client.post("/api/sessions", json={"title": "reconnect"}).json()["session_id"]
        with client.websocket_connect(f"/ws/sessions/{session_id}") as socket:
            socket.send_json({"type": "user_message", "content": "finish while connected"})
            original = []
            while not original or original[-1]["kind"] != "run_finished":
                original.append(socket.receive_json())

        after = original[-3]["cursor"]
        expected = original[-2:]
        with client.websocket_connect(f"/ws/sessions/{session_id}?after={after}") as reconnected:
            replayed = [reconnected.receive_json() for _ in expected]

assert [event["event_id"] for event in replayed] == [event["event_id"] for event in expected]
print("replayed cursors:", [event["cursor"] for event in replayed])

replayed cursors: [10, 11]

The reconnect socket receives only the suffix after the known cursor. In a retained-log system, a cursor older than available history must return an explicit cursor-too-old response and force a full session reload. An apparently complete partial timeline is worse than a visible resync.

Exercises

Design a slow-consumer test for the browser broker and durable replay path. Include queue size, eviction signal, last applied cursor, reconnect request, duplicate handling, and the assertion that proves the observer recovered every event.

[P08.1] Recover an evicted observer

Fill one observer’s bounded queue until it is evicted, let the run finish, and specify how the browser detects the close, reconnects, and proves its final projection is complete and duplicate-free.