7. Realtime — SSE Consumer Architecture

The Server-Sent Events (SSE) event stream, the invalidation router, and how realtime cache fan-out works. Part of the Frontend Architecture reference.

Read this if you need to understand how a backend change reaches the screen without a manual refresh — or you are adding a new event type.

SSE is a one-way channel: the browser opens a long-lived HTTP connection and the server pushes named events down it. The web app never sends messages back on this channel — it only listens, then refreshes the affected parts of the query cache. At a glance:

The worker records an event, the endpoint streams it, the router turns it into the right cache invalidations, and the views re-fetch. §7.6 shows this same path as a detailed sequence diagram.

The backend records JobEvent rows in job_events (workers/automation's EventPublisher + apps/api/src/projections.ts) and streams them over GET /v1/events/stream. The frontend consumer is implemented: SseEventStreamAdapter (behind EventStreamPort) opens the connection, EventStreamProvider (in contexts/operations/providers/) manages the subscription lifecycle, and the invalidation router fans each event out to the query cache. <ConnectionStatusPill> in the Topbar shows liveness.

This section defines that realtime architecture, including the apps/api/ SSE endpoint contract it depends on.

7.1 The Endpoint — GET /v1/events/stream

Decision (resolves §6 question 9 transport): Server-Sent Events (SSE) on a new dedicated endpoint.

Why SSE, not WebSocket / polling:

• Unidirectional fits the use case. The frontend only consumes events; it does not need to send messages on the channel. SSE is exactly this.
• Native EventSource API. No library, automatic reconnect with Last-Event-ID, plays nicely with HTTP/2 multiplexing, no framing-protocol custom handling.
• Fastify SSE support. Fastify can stream text/event-stream responses with backpressure; no extra runtime.
• CDN / proxy friendliness. Plain HTTP; one long-lived response; tracable; debuggable in the network panel.
• Auth simplicity. EventSource sends cookies (or Authorization via a small eventsource polyfill) — same auth path as REST.
• Polling rejected as the durable-change transport: event arrival is sparse but bursty, and a polling-only design has poor latency for "apply run completed." Narrow polling remains appropriate for ephemeral runtime facts that do not emit domain events (see §7.5).
• WebSocket rejected for now: bidirectional, framing overhead, harder to cache-debug, harder to terminate at edge proxies. Named as evolution path (§9) if event volume or duplex requirements emerge.

Endpoint contract:

GET /v1/events/stream?tenantId=<tenantId>&since=<lastEventId>
Accept: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
Last-Event-ID: <lastEventId>      # set by EventSource auto-reconnect

(server)
HTTP/1.1 200 OK
Content-Type: text/event-stream
X-Accel-Buffering: no

retry: 5000

id: 12345
event: JobScored
data: {"tenantId":"local","jobId":"job-...","fitScore":8,"version":1,"scoredAt":"..."}

id: 12346
event: ResumeApproved
data: {"tenantId":"local","jobId":"job-...","artifactId":"...","generation":2,"approvedAt":"..."}

: keepalive (every 15s)

Resume-position precedence: the server prefers the Last-Event-IDheader when present (this is what the browser's native EventSource auto-reconnect sends — the application code does not populate it). The ?since=<lastEventId> query string is the first-connect fallback for cases where the client wants to resume from a known watermark without relying on header-based reconnect — primarily the IndexedDB cache-hydration evolution path (§9.7), where the client knows the watermark of its persisted cache before opening the connection. If both are present, Last-Event-ID wins. If neither is present, the server streams from the current tail (no backfill).

Server-side responsibilities:

• Tail job_events for new rows where tenant_id = :tenantId AND event_id > :resumeFrom, where resumeFrom is taken from Last-Event-ID (preferred) or ?since (fallback) or current_max(event_id) (default if neither is supplied).
• Map event_type to the SSE event: field; serialize payload_json as data: (already JSON; pass through).
• Set id: to event_id (so EventSource automatically reconnects with Last-Event-ID).
• Send a comment line : keepalive every 15s (overridable) to keep intermediaries from idling the connection out.
• Set retry: 5000 (5s reconnect baseline).
• Tenant scope is mandatory: the server enforces that returned events match :tenantId. In local mode, this is LOCAL_TENANT; in hosted mode, the server resolves tenantId from the JWT and rejects mismatched query-string values.
• Heartbeat with current watermark id every 30s in a separate event: heartbeat so the client can verify liveness even when no domain events fire.

Client-side responsibilities:

• Open new EventSource("/v1/events/stream?tenantId=" + tenantId) once per tab when the application mounts (after <TenantProvider /> resolves). Do not pass ?since on first connect; the server defaults to current tail.
• The browser's auto-reconnect sends Last-Event-ID automatically; no application code needed for the common case.
• For IndexedDB-hydrated cold start (§9.7), the application explicitly passes ?since=<persistedWatermark> on first connect.
• Parse each event + data frame with parseDomainEvent — it validates the eventType against the runtime DOMAIN_EVENT_TYPES set, JSON.parses the payload, and object-checks it (no Zod, no payload-shape schema; see §7.2).
• Dispatch to the invalidation router (§7.4).
• Expose a status indicator (connecting | open | closed) consumed by the AppShell to render a small "live"/"reconnecting" badge.

7.2 Typed Event Schemas

The event taxonomy lives in @jobctrl/domain-types at packages/domain-types/src/events/. It is a plain TypeScript discriminated union — there is no Zod. DomainEvent<T, P> is the generic base interface (its eventType field is the discriminant); the union of all registered concrete events is DomainEventUnion, with DomainEventType = DomainEventUnion["eventType"] and a runtime companion array DOMAIN_EVENT_TYPES (kept exhaustive against DomainEventType by a compile-time assertion). @jobctrl/domain-types has no zod dependency.

The frontend's parseDomainEvent(rawFrame) (in shared/ports/lib/parseDomainEvent.ts) validates only that the SSE frame's eventType is a member of DOMAIN_EVENT_TYPES, then JSON.parses the data payload and object-checks it — it does not schema-validate the payload shape. An unknown event: type is dropped (forward-compat: the backend can introduce SomethingNew events without breaking the client; the client routes them once the union and a handler are added).

7.3 The EventStreamProvider

// contexts/operations/providers/EventStreamProvider.tsx
export function EventStreamProvider({ children }: { children: ReactNode }) {
  const tenantId = useTenantId();
  const { eventStream } = usePorts();
  const router = useInvalidationRouter();

  useEffect(() => {
    const sub = eventStream.subscribe({ tenantId });
    const off = sub.on((event) => router(event));
    return () => { off(); sub.close(); };
  }, [tenantId, eventStream, router]);

  return <>{children}</>;
}

It lives in contexts/operations/providers/ (not shared/providers/) and is mounted in the main.tsx provider stack below <QueryClientProvider /> and above the theme/density providers. It also exposes useEventStreamStatus (consumed by <ConnectionStatusPill>). It renders no UI of its own — it manages the subscription lifecycle.

7.4 The Invalidation Router

A pure function maps each DomainEvent to tenant-scoped invalidation or exact patch instructions. The router lives in contexts/operations/invalidation-router.ts; each backend event type has a registered aggregate-owned handler:

export const jobActiveStateChangedHandler = (event) => [
  patchQuery(jobsKeys.detail(event.tenantId, event.payload.jobId),
    (current) => patchJobActiveState(current, event.payload)),
  invalidate(jobsKeys.lists(event.tenantId)),
  invalidate(dashboardKeys.summary(event.tenantId)),
];

export const resumeApprovedHandler = (event) => [
  patchQuery(jobsKeys.detail(event.tenantId, event.payload.jobId),
    (current) => patchResumeApproved(current, event.payload)),
  patchQuery(artifactsKeys.detail(event.tenantId, event.payload.artifactId),
    (current) => patchResumeApproved(current, event.payload)),
  invalidate(jobsKeys.lists(event.tenantId)),
  invalidate(artifactsKeys.lists(event.tenantId)),
];

// Workflow lifecycle handlers patch the existing detail, then invalidate the
// list/dashboard reads whose membership or aggregation may have changed.
// ApplyRunEventRecorded remains a direct ordered append.

In practice the per-event handler functions are authored in each aggregate context's handlers.ts (seven files: discovery, enrichment, profile, scoring, materials, apply, pipeline) and registered centrally in invalidation-router.ts, which exports invalidate, patchQuery, patchApplyRunEvent, and useInvalidationRouter. The illustration above inlines representative handlers for clarity; Operations itself has no handlers.ts.

Why a router and not per-context subscriptions:

• Single point to reason about cross-context invalidation. A new event type means one PR touching one file (the router) plus the schema.
• Testable in isolation. The router is a pure function; tests assert that a specific event triggers the expected invalidation set without touching the network or React.
• The handlers can use the registry of keys (§4.1) so contexts do not need to know about each other.

Fitness function — every backend DomainEvent has a router handler. Two layers, both required:

1. Compile-time: the handlers map is typed Record<DomainEventType, InvalidationHandler>. Adding a new variant to the discriminated union in @jobctrl/domain-types/events/ (mirroring a new backend event type) is a TypeScript compile error in apps/web until a handler is wired. This is the primary guard.
2. Runtime parity test:contexts/operations/every-event-has-handler.test.ts iterates the runtime DOMAIN_EVENT_TYPES array (from @jobctrl/domain-types; there is no Zod schema to read .options from) and asserts a handler is registered for each. This is the backstop that catches the case where a developer adds a stub handler () => [] (TS-passing, behaviorally wrong). The web Vitest suite runs in TypeScript CI; the compile-time check and runtime parity test are both required.

The pattern mirrors the backend's scripts/check-domain-type-parity.py (per architecture.md's verification-commands section). A new event on the backend triggers a TypeScript compile error (in CI, via pnpm -r check) AND a runtime parity-test failure (locally) on the frontend — silent invalidation gaps are prevented by construction.

7.5 Strategy: invalidate vs setQueryData (resolves §6 question 9)

Two patterns exist; both have a place:

Pattern	When to use	Example event
queryClient.invalidateQueries({ queryKey })	Use when the payload is incomplete, list/filter membership can change, a dashboard aggregate changed, or no exact registered row is known. Keep the key tenant-scoped and no broader than the affected resource family.	JobScored → invalidate job list/detail and dashboard projections.
queryClient.setQueriesData / setQueryData	Use immediately when the event contains the canonical fields needed for a truthful patch of an existing cache row. Pair it with bounded invalidation when another view needs membership/refilter/aggregate reconciliation.	Active job detail, registered artifact approval, workflow detail, and ordered Apply-run append.

Why invalidation remains necessary:

• Single source of truth. The projection on the server is canonical; the cache always reconciles to it.
• No hand-rolled merge bugs. Patching cache shape by hand introduces mismatch between the patched value and what a fresh fetch would return.
• Membership safety. Events cannot insert rows into filtered or paginated lists without the complete projection and filter context.

Exact patch rules:

• JobActiveStateChanged patches an open job detail immediately; job lists and Dashboard invalidate because active-state filters and aggregates can change.
• ResumeApproved changes status only on an artifact already present in an open job/artifact detail. It never creates a missing artifact; list pages invalidate so persisted registration and filtering remain authoritative.
• Workflow* lifecycle events patch the matching run detail by exact workflow identity and append a deduplicated timeline entry. Run lists, Dashboard, and operations invalidate because status membership and aggregation can change.
• ApplyRunEventRecorded directly appends its complete ordered event. This high-frequency path avoids a refetch per event; a later read still provides reconciliation.

The router never resets component-owned filters, selection, pagination, or scroll position. A query update changes data under the existing view state.

Runtime snapshot polling complements SSE

usePipelineOperationsQuery reads GET /v1/pipeline/operations under pipelineKeys.operations(tenantId). Stage*, PreparationWorkItem*, PipelineStep*, and Workflow* handlers invalidate that key, so durable changes still take the fast SSE path.

Worker heartbeats, exact active-slot counts, bounded active-item inventory, and Temporal task-queue observations are ephemeral runtime telemetry and do not append domain events. The same query therefore has a 10-second stale time and polls every 15 seconds while the selected execution is discovering or draining, every 60 seconds otherwise, and never while the page is in the background. This narrow polling path refreshes only the operations snapshot; it does not replace the event stream or trigger broad cache invalidation.

7.6 Realtime Data Flow

7.7 Reconnect / Backoff

EventSource's built-in reconnect is sufficient for the MVP:

• Server sends retry: 5000 (5s baseline).
• On disconnect, browser auto-reconnects, sending Last-Event-ID header so the server resumes from the last delivered event.

The EventStreamProvider exposes status to the AppShell. When status === "closed" for more than 30s, the shell renders a banner "Connection lost — events paused; data will refresh when reconnected." On reconnection, the provider triggers a one-shot queryClient.invalidateQueries() (full cache invalidation) to recover from any events lost during the gap. (Last-Event-ID covers the common case; the full invalidation is a backstop.)

7.8 Tenant Scoping in Realtime

The connection is parameterized by tenantId. In local mode, the value is LOCAL_TENANT. In hosted mode:

• The server validates :tenantId against the JWT. Mismatch → 403.
• The connection is per-tenant; if a user switches tenants (cloud-only feature), the EventStreamProvider closes the old connection and opens a new one (the useEffect dependency on tenantId does this naturally).
• Invalidation routing already includes tenantId in every query key, so there is zero cross-tenant cache leak even if events were mis-delivered.

7.9 What If SSE Is Not Enough Later

Named-not-built evolution paths (also see §9):

• WebSocket adapter — if duplex (e.g., the frontend driving an interactive worker session) becomes a requirement, swap to WebSocketEventStreamAdapter behind the same EventStreamPort.
• Push notifications — for "your apply run completed" while the tab is closed, integrate Web Push via a NotificationsPort.
• Per-resource subscriptions — today, every event reaches every client. If event volume grows so large that per-tenant filtering at the server is insufficient, introduce subscribe(resource: "job", id) semantics in the port, with the SSE endpoint accepting filter params.

---