feat(#51): add SSE endpoint GET /api/swarms/{id}/events/stream

Real-time event stream for the cockpit, reverse-proxied by HM to the client
EventSource (orchestrator SSE → HM → client). The SSE overlay over the existing
GET /events?after=<sequence>:

- Replays history after `after`, then holds the connection and pushes new events.
- Reuses the existing swarm_events:{swarm_id} store + per-swarm INCR sequence — no
  new storage, no schema change.
- Each frame: `id: <sequence>` / `event: message` / `data: <envelope JSON>`, where
  data is byte-identical to /events so SSE and polling share one cursor space (a
  dropped SSE can fall back to /events?after=<last id> with no gap/dup).
- Supports Last-Event-ID header (takes precedence over the `after` query) for
  end-to-end resume through HM.
- Heartbeat `: ping` every ~15s + X-Accel-Buffering:no to survive nginx ingress /
  HM reverse-proxy buffering.
- Closes after a terminal event (swarm.completed/failed/stopped); releases on client
  disconnect (request.is_disconnected).
- Auth: require_runtime_auth (service token) — caller is HM, never the client direct.

Adds TERMINAL_CLIENT_EVENT_TYPES to swarm_runtime. Events are already redacted at
emit time, so frames are streamed as-is. No change to the callback POST path,
event schema, sequence, or terminal definitions (event-schema v1 frozen).

Verified: contract-freeze / runtime-contract / merge-smoke / workflow-e2e all pass;
plus a dedicated SSE check (history replay, id ordering, Last-Event-ID resume,
byte-identical envelope, terminal close).

Part of #40 (the SSE half). HM reverse-proxy side = heicode-mananger#46.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

orchestrator/main.py

@@ -9,7 +9,7 @@ from contextlib import asynccontextmanager
 from typing import Any, Dict, List, Optional
 
 from fastapi import FastAPI, Header, Request, WebSocket, WebSocketDisconnect, HTTPException
-from fastapi.responses import JSONResponse, PlainTextResponse
+from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
 from pydantic import BaseModel, Field
 from prometheus_client import Counter, Gauge, Histogram, generate_latest, CONTENT_TYPE_LATEST
 from dotenv import load_dotenv
@@ -22,7 +22,7 @@ from .redis_client import redis_client
 from .agent_registry import agent_registry, AgentStatus, AgentMetadata
 from .handoff_manager import handoff_manager, HandoffRequest
 from .task_queue import task_queue, TaskStatus
-from .swarm_runtime import RuntimeValidationError, swarm_runtime
+from .swarm_runtime import RuntimeValidationError, swarm_runtime, TERMINAL_CLIENT_EVENT_TYPES
 from .planner import planner
 from .master_agent import master_agent
 from .quality import evaluate_run_quality
@@ -2128,6 +2128,93 @@ async def get_swarm_events(
     return {"success": True, "data": events}
 
 
+@app.get("/api/agent/swarm/deployments/{deployment_id}/events/stream")
+@app.get("/api/swarms/{deployment_id}/events/stream")
+@app.get("/api/agnet/deployments/{deployment_id}/events/stream")
+async def stream_swarm_events(
+    deployment_id: str,
+    request: Request,
+    after: int = 0,
+):
+    """SSE real-time event stream for a swarm run (agent_swarm#51).
+
+    The realtime overlay over `GET /events?after=<sequence>`: on connect it replays the events
+    after `after`, then holds the connection and pushes new events as they are emitted. Reuses the
+    existing `swarm_events:{swarm_id}` store and per-swarm `sequence` (no new storage, no schema
+    change). Each frame's `id:` is the event `sequence`, so SSE and polling share one cursor space
+    — a dropped SSE connection can fall back to `/events?after=<last id>` with no gap/dup.
+
+    Auth is `require_runtime_auth` (service token): the caller is **HM** (which reverse-proxies to
+    the client EventSource, heicode-mananger#46), never the client directly. Events are already
+    redacted at emit time, so frames are streamed as-is. The stream closes after a terminal event
+    (`swarm.completed` / `swarm.failed` / `swarm.stopped`).
+    """
+    run, auth_error = await get_runtime_run_or_404(request, deployment_id)
+    if auth_error:
+        return auth_error
+
+    # Last-Event-ID (sent by the browser/HM on reconnect) takes precedence over the `after` query
+    # (#51 point 4): identical cursor semantics — resume from the event after the last one seen.
+    start_index = after
+    last_event_id = request.headers.get("Last-Event-ID")
+    if last_event_id:
+        try:
+            start_index = int(last_event_id)
+        except (TypeError, ValueError):
+            pass
+    start_index = max(0, start_index)
+    swarm_id = run.swarm_id
+
+    async def event_generator():
+        # `cursor` is the Redis list index to read from. Event with sequence S sits at index S-1,
+        # so events *after* sequence N start at index N — i.e. cursor == "last sequence seen".
+        cursor = start_index
+        # Hint the EventSource how long to wait before reconnecting after a drop.
+        yield "retry: 3000\n\n"
+        heartbeat_seconds = 15
+        poll_seconds = 1.0
+        last_ping = time.monotonic()
+        while True:
+            if await request.is_disconnected():
+                return
+            page = await swarm_runtime.list_events(swarm_id, limit=500, cursor=str(cursor))
+            events = page.get("events", [])
+            if events:
+                for event in events:
+                    sequence = event.get("sequence")
+                    data = json.dumps(event, ensure_ascii=False, separators=(",", ":"))
+                    # `data:` is byte-identical to the /events envelope; `event: message` matches the
+                    # client EventSource default channel (Mem0ried points 1–2).
+                    yield f"id: {sequence}\nevent: message\ndata: {data}\n\n"
+                    cursor = sequence if isinstance(sequence, int) else cursor + 1
+                    if event.get("event_type") in TERMINAL_CLIENT_EVENT_TYPES:
+                        # Terminal frame is flushed above; close normally so the client stops
+                        # reconnecting (#51 point 6 / Mem0ried point 3).
+                        return
+                last_ping = time.monotonic()
+                # Drain any remaining backlog immediately (no sleep) before holding the connection.
+                continue
+            # No new events: heartbeat to keep the long-lived connection alive through the nginx
+            # ingress + HM reverse-proxy (#51 point 5), then poll again. EventSource ignores
+            # comment frames, so no `data:` is needed.
+            now = time.monotonic()
+            if now - last_ping >= heartbeat_seconds:
+                yield ": ping\n\n"
+                last_ping = now
+            await asyncio.sleep(poll_seconds)
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            # Disable nginx proxy buffering so frames flush immediately (#51 point 5).
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
 @app.get("/api/agent/swarm/deployments/{deployment_id}/metrics")
 @app.get("/api/swarms/{deployment_id}/metrics")
 @app.get("/api/agnet/deployments/{deployment_id}/metrics")

orchestrator/swarm_runtime.py

@@ -107,6 +107,10 @@ FROZEN_CLIENT_EVENT_TYPES = (
     "rework.completed",
 )
 
+# Terminal client events (agent_swarm#40): once one of these is streamed, the run has reached a
+# final state and the SSE stream can close — the client cockpit keys its terminal UI on these.
+TERMINAL_CLIENT_EVENT_TYPES = frozenset({"swarm.completed", "swarm.failed", "swarm.stopped"})
+
 
 class SwarmRuntime:
     """Stores swarm runs and emits Agent Manager callback events."""