One SSE connection, N upstream agents — the stream relay we shipped

The pod that was tailing one of our long-running agents got rotated during a deploy. The client connection to the dashboard didn't disconnect — it sat there, EventSource happily believing the world was fine, with zero events coming through. The agent was still working. The events were still being emitted upstream. The middle box that was supposed to forward them to the browser was simply gone.

That moment is the reason this post exists. If you're going to put a relay between N independent event-emitting backends and one client-facing SSE connection, you have to answer a specific question: when the relay process dies, what guarantees that another one picks up where it left off, and that the client doesn't notice? A lot of "just proxy SSE" designs fall apart at exactly this seam.

I'm building a SaaS that touches this problem (Codens, an AI dev harness with a few specialized agents — happy to talk about it but it isn't the point of this post). The relay I'll walk through fans in live telemetry from N agent jobs running on independent nodes into a single per-run aggregated stream the dashboard subscribes to.

The shape of the problem

The setup, concretely:

• N agent jobs run, each on its own host. Each exposes GET /jobs/{job_id}/stream and emits events for one workflow run only.
• The dashboard wants a live feed for a specific run. It opens one SSE connection to our API, not N.
• We have multiple instances of the relay process. Any can die at any time. Deploys rotate them constantly.
• Reconnects are constant — laptops sleep, networks blip, tabs reload.

The naive shape that almost works: each browser opens an SSE connection through one of our API pods, and that pod opens an SSE connection upstream to the agent. Two-hop proxy, no state. Here's why it breaks:

1. Two browsers viewing the same run = two upstream connections. Hits the agent's per-job stream twice. Some agents fight back; some let you do it but dedupe is your problem.
2. Reconnects = a new upstream connection too. You either lose all events between disconnect and reconnect, or you require the upstream to support Last-Event-ID. Many homegrown SSE servers don't.
3. No durable buffer. A client offline for 10 seconds loses those events. No tape.
4. Per-pod independent tailers = N tailers per run with N pods. Wasted upstream connections, no single source of truth for "what events have we seen for run X."

The shape we ended up with: exactly one stream-relay pod owns the upstream connection for a given run at a time. That pod tails the agent's SSE, writes every event into a Redis Stream keyed by run, and the API layer reads from Redis Streams to serve clients. Many client SSE connections, one upstream connection, durable buffer in the middle.

┌──────────────┐      one connection per run     ┌─────────────────┐
│  Agent pod A │ ─────────────────────────────▶  │ stream-relay    │
└──────────────┘                                 │  pod (owner)    │
┌──────────────┐                                 │                 │
│  Agent pod B │ ─────────────────────────────▶  │ XADD events:{r} │
└──────────────┘                                 └────────┬────────┘
   ... N agent jobs ...                                   │
                                                          ▼
                                                 ┌─────────────────┐
                                                 │  Redis Streams  │
                                                 │  events:{run_id}│
                                                 │   (maxlen 10k)  │
                                                 └────────┬────────┘
                                                          │
                                  many client SSE conns   │
                              ◀───────────────────────────┘
                              (FastAPI sse_starlette pods)

The interesting design decisions are all about the middle box: who tails, how takeover works, how the API layer reads from the buffer, what counts as "alive."

Stateless relay, leased ownership

The relay process is stateless. It boots, looks at the database for active workflow runs with an active job, and tries to acquire a per-run ownership lock in Redis. If it gets the lock, it spawns an asyncio task that opens the upstream SSE and tails it. If another relay instance got the lock first, it doesn't spawn a tailer. Every 5 seconds, it scans again.

The main loop, abridged from stream_relay/main.py:

POLL_INTERVAL = 5.0  # seconds between active-run scans

while True:
    # 1. Refresh owned locks
    await ownership.refresh_all()

    # 2. Clean up finished tasks
    finished = [rid for rid, t in active_tasks.items() if t.done()]
    for rid in finished:
        try:
            active_tasks[rid].result()
        except asyncio.CancelledError:
            pass
        except Exception:
            logger.exception("stream-relay: tailer task error run_id={}", rid)
        del active_tasks[rid]

    # 3. Discover new running workflow_runs
    async with session_factory() as session:
        running_infos = await _find_running_runs(session)

    active_run_ids = {info.run_id for info in running_infos}

    # Cancel tasks for runs that are no longer active
    stale = [rid for rid in active_tasks if rid not in active_run_ids]
    for rid in stale:
        active_tasks[rid].cancel()

    # 4. Acquire + start new tasks
    for info in running_infos:
        if info.run_id in active_tasks:
            continue
        acquired = await ownership.acquire(info.run_id)
        if not acquired:
            continue
        task = asyncio.create_task(
            tail_stream(http, redis_client, info, ownership, streams_client),
            name=f"tail-{info.run_id}",
        )
        active_tasks[info.run_id] = task

    await asyncio.sleep(POLL_INTERVAL)

Three pieces are doing real work. (1) The lock refresh runs first so we don't lose ownership while we're busy. (2) The DB scan is the source of truth for "what runs should be tailed right now"; the relay never trusts its own in-memory state. (3) Acquisition is best-effort — if someone else got there first, we move on.

Stateless is deliberate. The relay doesn't keep a cursor file. There's no "what was I doing before I crashed" reconciliation. If a pod restarts, on its next 5-second tick it refreshes whatever locks it still holds (none, just booted), looks at active runs, and tries to acquire each. If the previous owner died, the lock has expired (60s TTL) and this pod takes over.

The instance ID is computed once at boot — ECS task ID, falling back to hostname, falling back to UUID:

_INSTANCE_ID = (
    settings.STREAM_RELAY_INSTANCE_ID
    or os.environ.get("ECS_TASK_ID")
    or socket.gethostname()
    or str(uuid.uuid4())
)

This matters because the ownership value stored in Redis is this instance ID. When checking "do I still own this lock," you compare the value, not just check existence.

The lock, in Redis, in nine lines

Every run_id gets one Redis key with a 60-second TTL. The owner refreshes it every 20 seconds. If the owner dies, the lock expires within 60s and another pod can grab it.

LOCK_TTL_SECS = 60
REFRESH_INTERVAL_SECS = 20
_KEY_PREFIX = "stream_owner"

async def acquire(self, run_id: str) -> bool:
    key = self._key(run_id)
    try:
        acquired = await self._redis.set(
            key,
            self._instance_id,
            nx=True,
            ex=LOCK_TTL_SECS,
        )
        if acquired:
            self._owned[run_id] = time.monotonic()
            return True
        # Check if we already own it (e.g. after a crash + restart)
        owner = await self._redis.get(key)
        if owner == self._instance_id:
            self._owned[run_id] = time.monotonic()
            return True
        return False
    except Exception:
        logger.exception("OwnershipManager.acquire failed run_id={}", run_id)
        return False

SET key value NX EX 60 is the entire trick. NX = only if not exists, EX = TTL in seconds. Atomic. The "already own it after a crash + restart" branch is the case where the hostname is stable across restarts and we want to recognize the lock as ours rather than wait out the TTL.

Refresh is defensive too. It checks the value before extending TTL, so if someone else has taken over we don't blindly re-extend their lock:

async def refresh(self, run_id: str) -> bool:
    key = self._key(run_id)
    try:
        owner = await self._redis.get(key)
        if owner != self._instance_id:
            logger.warning(
                "stream_owner: lock lost for run_id={} (owner={} expected={})",
                run_id, owner, self._instance_id,
            )
            self._owned.pop(run_id, None)
            return False
        await self._redis.expire(key, LOCK_TTL_SECS)
        self._owned[run_id] = time.monotonic()
        return True
    except Exception:
        logger.exception("OwnershipManager.refresh failed run_id={}", run_id)
        return True  # conservative: don't stop on Redis error

That return True on Redis error is deliberate. If Redis is briefly unreachable, we keep tailing rather than abandon a working stream. Redis comes back, next refresh confirms or denies. The downside: a brief window where two pods could both think they own a run during a Redis flap, but they're both writing to the same events:{run_id} stream — duplicates, not data loss. Duplicates the consumer can dedupe; data loss it can't recover.

The 60s TTL / 20s refresh ratio gives a 3x safety factor. The pod refreshes three times before TTL would expire; even one missed refresh is fine.

Pod rotation, in practice

The flow when a relay pod is rotated mid-tail:

1. Pod A is tailing run X. Its stream_owner:X Redis key has its instance ID and a 60s TTL, refreshed every 20s.
2. Pod A receives SIGTERM (deploy, scale-down, ECS task draining).
3. Inside relay_main, the shutdown handler cancels all active tasks, awaits them, closes Redis.
4. tail_stream's finally block calls ownership.release(run_id), which deletes the Redis key — but only if it still belongs to this instance.
5. Pod B, on its next 5-second poll, finds run X in the DB, tries to acquire, gets it, spawns a tailer.

async def release(self, run_id: str) -> None:
    key = self._key(run_id)
    try:
        owner = await self._redis.get(key)
        if owner == self._instance_id:
            await self._redis.delete(key)
    except Exception:
        logger.exception("OwnershipManager.release failed run_id={}", run_id)
    self._owned.pop(run_id, None)

The bad case — the one that brought me to this design — is a pod dying without running its shutdown handler. SIGKILL after a stuck SIGTERM. OOM. Network partition. The key still exists with its TTL ticking down; within 60 seconds it expires and Pod B's next poll picks it up.

Worst-case takeover: ~60s for TTL + up to 5s for the next poll = ~65s of no events relayed. The dashboard sees a gap. The events emitted during that gap aren't lost — they're still buffered in the upstream agent's own per-job queue, and the new tailer reads them on reconnect. (Whether that holds depends on your upstream; ours has a per-job in-memory replay buffer.)

We picked the boring 60/20 numbers because they're easy to reason about and a one-minute gap during deploys is tolerable. Shorter TTLs halve worst-case latency at the cost of more Redis traffic. The knobs are there.

The tailer: one upstream connection per run

The tailer is the asyncio task spawned for each owned run. It opens an httpx streaming GET to the agent's SSE endpoint and parses events:

async def tail_stream(
    http: httpx.AsyncClient,
    redis_client: aioredis.Redis,
    run_info: RunInfo,
    ownership: OwnershipManager,
    streams_client: RedisStreamsClient,
) -> None:
    run_id = run_info.run_id
    url = f"{run_info.endpoint_url.rstrip('/')}/jobs/{run_info.job_id}/stream"
    headers = _build_headers(run_info.api_key_hash)
    consecutive_errors = 0

    try:
        while True:
            # --- ownership check (before reconnect) ---
            if not await ownership.is_owned_by_us(run_id):
                return

            try:
                timeout = httpx.Timeout(connect=15.0, read=1800.0, write=15.0, pool=15.0)
                async with http.stream("GET", url, headers=headers, timeout=timeout) as response:
                    response.raise_for_status()
                    consecutive_errors = 0

                    check_deadline = time.monotonic() + CHECK_INTERVAL_SECS

                    async for event in _parse_sse_stream(response):
                        await streams_client.xadd(run_id, event)
                        # ... heartbeats + terminal-event handling ...

Two details that matter.

The httpx client is shared across all tailers in the same pod. We have a single httpx.AsyncClient and pass it into every tailer task. If you create one client per tailer, you blow your file descriptor budget at ~100 concurrent runs and lose connection pooling per host:

http_limits = httpx.Limits(
    max_keepalive_connections=200,
    max_connections=200,
    keepalive_expiry=None,
)
async with httpx.AsyncClient(
    timeout=None,
    verify=False,
    limits=http_limits,
) as http:
    # all tailers share `http`

The read timeout is 30 minutes. SSE streams can be quiet for a long time and that's normal. Set the read timeout to something sensible-sounding like 30 seconds and you get spurious ReadTimeout exceptions during legitimate slow phases of the agent's work. Learned this the hard way.

Reconnect logic is unsurprising — exponential backoff on connection/read errors, give up after 5 consecutive failures, release the lock:

except (httpx.ReadTimeout, httpx.RemoteProtocolError, httpx.ReadError) as exc:
    consecutive_errors += 1
except httpx.ConnectError as exc:
    consecutive_errors += 1
# ...
if consecutive_errors >= MAX_CONSECUTIVE_ERRORS:
    return

backoff = min(5 * 2 ** (consecutive_errors - 1), 60)
await asyncio.sleep(backoff)

5s, 10s, 20s, 40s, capped at 60s. After 5 failures we exit; the main loop's next poll re-acquires and starts fresh.

Two heartbeats: liveness vs forward progress

You'd think one heartbeat is enough. It isn't. The bug: an upstream agent stops making real progress (stuck on a tool call, the underlying LLM silently retrying internally) but its SSE stream still emits keepalive / ping events every 15 seconds because the agent's web framework wants the connection warm. If your stall detector reads a single "is the SSE alive" heartbeat, it sees green forever. The job is stuck and you don't know.

The fix is two heartbeats, read by different consumers:

LIVENESS_TTL_SECS = 300        # 5 minutes
RUN_PROGRESS_TTL_SECS = 3600   # 60 minutes

# SSE event types that indicate Claude is making forward progress.
_FORWARD_PROGRESS_TYPES = {"output", "result", "assistant", "tool_result"}

async for event in _parse_sse_stream(response):
    await streams_client.xadd(run_id, event)
    now_ts = str(time.time())

    # Liveness heartbeat: any SSE frame proves the connection is alive.
    await redis_client.setex(
        f"liveness_heartbeat:{run_id}",
        LIVENESS_TTL_SECS,
        now_ts,
    )

    # Forward-progress heartbeat: only real work advances this.
    if _is_forward_progress(event):
        await redis_client.setex(
            f"forward_progress_heartbeat:{run_id}",
            RUN_PROGRESS_TTL_SECS,
            now_ts,
        )

liveness_heartbeat:{run_id} advances on any SSE event including pings. It answers "is the upstream connection alive at all?"

forward_progress_heartbeat:{run_id} advances only on event types that mean the agent emitted something it worked for — output, result, assistant, tool_result. It answers "is the agent doing useful work, or idling?" If this is stale but liveness is fresh, the connection is fine but the agent is stalled.

A separate orphan-detection job reads the forward-progress key, not the liveness key. Mix them up and you get false positives during legitimate slow-thinking phases or false negatives during real stalls.

The buffer: Redis Streams with a hard cap

Every event the tailer parses is appended to a Redis Stream keyed by run:

class RedisStreamsClient:
    def __init__(self, redis_client: aioredis.Redis, prefix: str = "events") -> None:
        self._redis = redis_client
        self._prefix = prefix

    def stream_key(self, run_id: str) -> str:
        return f"{self._prefix}:{run_id}"

    async def xadd(
        self,
        run_id: str,
        data: dict[str, Any],
        maxlen: int = _STREAM_MAXLEN,  # 10_000
    ) -> Optional[bytes]:
        key = self.stream_key(run_id)
        try:
            entry_id = await self._redis.xadd(
                key,
                {"data": json.dumps(data)},
                maxlen=maxlen,
                approximate=True,
            )
            return entry_id
        except Exception:
            logger.exception("RedisStreamsClient.xadd failed run_id={}", run_id)
            return None

maxlen=10_000, approximate=True is the load-bearing flag. Without approximate=True, Redis trims exactly to 10k on every XADD — O(n) walk. With it, Redis trims at radix-tree-node boundaries: slightly fuzzy but vastly faster. Streams stabilize around 10k–11k entries, which is fine.

10k events per run is roughly half an hour of busy work for our agents. A client reconnecting within that window gets the entire history replayed from Redis. Beyond that, old events fall off; persist them separately if you need them long-term.

We use Streams (XADD / XREAD / XREADGROUP) and not just Pub/Sub precisely for this durability. Pub/Sub is fire-and-forget — no subscriber, event gone. Streams keep the bytes around so a client reconnecting after a blip replays from where it left off.

There's a layer above this in our codebase that bridges the buffered events into the per-client SSE response, including a history replay on connect so a late joiner sees prior events:

async def _subscribe_to_channel(channel: str, request: Request) -> AsyncGenerator[dict, None]:
    r = aioredis.from_url(settings.REDIS_URL, encoding="utf-8", decode_responses=True)
    pubsub = r.pubsub()
    try:
        # Subscribe BEFORE fetching history so we don't miss events published
        # between the LRANGE and the first get_message() call.
        await pubsub.subscribe(channel)

        history_key = f"sse:history:{channel}"
        history_events: list[str] = await r.lrange(history_key, 0, -1)

        last_replayed_seq: int | None = None
        for event_json in history_events:
            event_data = json.loads(event_json)
            seq = event_data.get("event_seq")
            if seq is not None:
                last_replayed_seq = seq
            yield {"event": "message", "data": event_json}

The order matters: subscribe before reading history. Otherwise events published between LRANGE and the first get_message() disappear. Live messages are then deduplicated against last_replayed_seq — every event carries a monotonic event_seq and the loop drops anything whose seq was already replayed. That seq is what gives the client a continuous-looking stream across pod takeovers, even if a reconnect briefly lands on a different API pod.

Backpressure on slow clients

The trap with SSE fan-out is that a single slow client (hotel wifi, throttled background tab, dev who left a curl pipe running for two days) piles up unsent events in the per-connection buffer.

The choices that protect the rest:

1. The relay never blocks on a client. It only writes to Redis. A slow client is the API pod's problem, not the relay's. The relay keeps tailing at full speed.
2. The Redis Stream is bounded. 10k entries cap. A reconnecting client gets at most 10k events; beyond that, the stream has aged out.
3. Per-connection sse-starlette buffers are small. Connections that fall too far behind get closed. We don't extend their grace period.

The single most important rule is the first: the upstream-tailing path is independent of any single client. Many pods serve SSE to many clients all reading from the same Redis Streams buffer. A slow consumer only slows itself.

When this is overkill

If you have one service emitting events, one client per run, no deploys, a short session bound, and reconnects are rare — a single-pod direct SSE proxy with no Redis in the middle is fine. Ship that.

The relay is justified by N agents × M clients × frequent deploys × long session lengths. Take any of the four out and the cost-benefit shifts. The specific moment where this design pays off is the rotated-pod-mid-stream case. If that never happens for you, most of the ownership-lock complexity is theatre. Keep the Redis Streams buffer for reconnect-replay; drop the leased ownership.

What I'd build differently

A real Last-Event-ID header path. Reconnect today uses a history-list replay keyed off the channel. That's fine for our common case but it's not the SSE-spec way. Cleaner: client sends Last-Event-ID on reconnect, the API seeks the Redis Stream to that ID. We will get there.

Fewer Redis round-trips per event. Every event is one XADD plus a SETEX for liveness, sometimes two more SETEXes for forward progress. Three to four Redis calls per event. A pipelined batch would cut latency. We haven't needed it yet, but it'll show up the first time someone points a debugger at the inner loop.

If you've shipped a multi-instance SSE relay, I'd love to hear which corner you cut. Especially: did you go with a leased single-tailer-per-run like this, or did you let every pod tail every run and dedupe at the consumer layer? I'm half-convinced the second design is simpler and I keep talking myself out of it.