feat: consume SSE event stream for live dashboard updates

[vredchenko]

Summary

Add frontend infrastructure to consume the backend SSE event stream (GET /frontend/events/stream) and display live updates in the SmartEM dashboard.

Depends on: DiamondLightSource/smartem-decisions#246 (backend SSE endpoint + agent log shipping)

Why the SSE endpoint needs a hand-written hook

The API client is auto-generated from the backend OpenAPI spec via Orval. Orval generates React Query hooks backed by Axios, which is request-response only -- it doesn't support the EventSource streaming protocol. The generated hook for the SSE endpoint would make a regular GET that either hangs or returns garbage. We need a custom useEventStream hook using the browser's native EventSource API instead.

Implementation steps

Step 1: Regenerate API client

npm run api:fetch:local   # Fetch OpenAPI spec from local backend (must be running with new endpoints)
npm run api:generate      # Regenerate hooks, types, MSW mocks

This picks up:

• AgentLog model type
• AgentLogBatchRequest / AgentLogBatchResponse types
• Updated Micrograph model with updated_at field
• The SSE endpoint (as a regular hook -- we'll ignore this generated hook and use our custom one)

Step 2: Create event type definitions

Create packages/api/src/events.ts:

• TypeScript types mirroring the backend FrontendEventType enum and event data models
• AgentStatusEvent, AcquisitionProgressEvent, InstructionLifecycleEvent, ProcessingMetricEvent, AgentLogEvent
• Hand-written (not generated) because SSE event payloads aren't modelled as response schemas in OpenAPI

Step 3: Create useEventStream hook

Create packages/api/src/useEventStream.ts:

• Uses browser EventSource API
• Connects to /frontend/events/stream with query params (acquisition_uuid, agent_id, event_types)
• Parses SSE event field to route to typed handlers
• Handles reconnection via Last-Event-ID (automatic with EventSource)
• Provides connection state (connected / reconnecting / disconnected)
• Exposes callbacks per event type
• Cleans up on unmount

Key API surface:

function useEventStream(options: {
  acquisitionUuid?: string
  agentId?: string
  eventTypes?: FrontendEventType[]
  enabled?: boolean
  onAgentStatus?: (data: AgentStatusEvent) => void
  onAcquisitionProgress?: (data: AcquisitionProgressEvent) => void
  onInstructionLifecycle?: (data: InstructionLifecycleEvent) => void
  onProcessingMetric?: (data: ProcessingMetricEvent) => void
  onAgentLog?: (data: AgentLogEvent) => void
}): {
  connectionState: 'connecting' | 'open' | 'closed'
  lastEventId: string | null
}

Step 4: React Query cache integration

SSE callbacks should invalidate relevant React Query caches so existing hooks automatically refetch:

• onAcquisitionProgress -> invalidate acquisition detail queries
• onProcessingMetric -> invalidate micrograph queries
• Or: directly update cache via queryClient.setQueryData for lower latency

Wrapper hook (e.g. useSmartEMEventStream) combines useEventStream with useQueryClient for cache wiring.

Step 5: UI components

• Connection indicator: Small badge in header showing SSE connection state
• Agent status badges: Online/offline/stale per agent
• Live acquisition counters: Grid/gridsquare/foilhole/micrograph counts
• Log viewer panel: Scrolling log view with level/agent filtering
• Processing metrics: Motion correction / CTF stats updating in real time

Step 6: Export from packages/api

Export useEventStream, event types, and useSmartEMEventStream from packages/api/src/index.ts.

Key files

All paths relative to repo root.

File	Action
packages/api/src/openapi.json	Regenerate
packages/api/src/generated/	Regenerate
packages/api/src/events.ts	Create -- SSE event type definitions
packages/api/src/useEventStream.ts	Create -- core SSE hook
packages/api/src/useSmartEMEventStream.ts	Create -- React Query integration wrapper
packages/api/src/index.ts	Modify -- export new hooks and types
apps/smartem/src/components/...	Create -- UI components per event type

Event types to consume

Event	Frontend use
agent.status	Microscope online/offline indicators, connection health
acquisition.progress	Live entity counts on acquisition detail views
instruction.lifecycle	Instruction status timeline/feed
processing.metric	Live motion correction, CTF, particle picking stats
agent.log	Remote log viewer panel (filterable by level, agent, logger)
heartbeat	Connection health indicator

Notes

• Backend filters events server-side via query params and DB queries -- ML/processing results from RabbitMQ only reach the SSE stream if they map to tracked entities (Micrograph processing fields). The frontend doesn't need additional filtering logic beyond choosing which event_types to subscribe to.
• The log ingestion endpoint (POST /agent/{id}/session/{id}/logs) is agent-to-backend only; the frontend consumes logs via the SSE stream, not by calling this endpoint directly.

Verification

1. Regenerate client, verify new types appear in generated/models/
2. Connect to local backend with SSE: open browser console, verify events arrive
3. UI components display live data when agent + EPUPlayer are running
4. Connection indicator reflects actual SSE state (reconnects on network interruption)
5. Log viewer shows agent logs in real time with level filtering

Related

• Backend implementation: feat: SSE event stream for frontend + agent log shipping smartem-decisions#246
• Backend PR: feat: SSE event stream for frontend + agent log shipping smartem-decisions#247

[vredchenko]

and when ML and data processing put things on the MQ - not all of those events should be streamed to the f/e - just those defined by filters

---

This is handled by the design. The SSE endpoint polls the DB, not RabbitMQ directly:

ML/processing plugins → RabbitMQ → consumer.py → writes specific fields to DB
                                                           ↓
Frontend SSE endpoint → polls DB with filters → streams only matching events

Filtering happens at two levels:

1. Consumer pre-filters: Only data the consumer explicitly persists to tracked entities (e.g. motion correction, CTF results written to Micrograph fields) reaches the DB. Raw MQ messages that don't map to tracked entities never enter the pipeline.
2. SSE query params post-filter: The client subscribes with event_types, acquisition_uuid, and agent_id params, so each frontend tab only receives the events it asked for.