Real-time Architecture

Four Interlocking Systems

Real-time in Slop is built from four components that each do exactly one job. Together they form a pipeline from daemon action to React re-render with no polling, no WebSocket server, and no external message broker.

The Event Bus

The event bus is a lightweight in-process pub-sub built on a Set<Subscriber>. It is created once at daemon boot, registered on globalThis under a well-known Symbol.for("slop.eventBus") key, and injected into every component that needs to publish or subscribe via the DaemonDeps object. Nothing imports the singleton directly; everything receives it as a constructor argument so tests can swap in a fake.

Factory and registration

// Created once at daemon boot (src/server/daemon/boot.ts)
const bus = createEventBus({ eventRepo });
setEventBus(bus);        // registers on globalThis

// The SSE route retrieves it at request time
const bus = getEventBus(); // throws if not initialized

What publish() does

Calling bus.publish(event) triggers two actions in sequence:

1. Persist to SQLite. If the event carries a workerId or runId, it is written to the Event table via eventRepo.appendEvent. The human-readable message comes from formatEvent in src/lib/event-format.ts, which is the single source of truth for event-to-string conversion.
2. Fan out to subscribers. Every registered in-memory subscriber (including the SSE stream) receives the event synchronously. If a subscriber throws, the error is logged and the remaining subscribers still receive the event.

Event type reference

Every event is typed by the BusEvent discriminated union in src/types/bus-event.ts.

SSE Stream

Server-Sent Events (SSE) is a simple HTTP protocol where the server holds the connection open and pushes newline-delimited frames. The browser handles reconnection automatically via the EventSource API - no WebSocket handshake, no custom framing protocol.

Endpoint

GET /api/events
Response headers:
Content-Type:  text/event-stream
Cache-Control: no-cache
Connection:    keep-alive

The route handler in src/app/api/events/route.ts is marked export const dynamic = "force-dynamic". This prevents Next.js from ever trying to statically render or cache it. The handler calls createEventStream(getEventBus(), { signal: request.signal }) and returns the ReadableStream directly as the response body.

Event wire format

Each bus event is encoded as a standard SSE data frame:

// A state transition event on the wire
data: {"type":"worker.state_changed","workerId":"wk_abc123","from":"implementing","to":"merging","at":"2026-06-13T14:22:00.000Z"}
                                                                                                      ^^
                                                                                              double newline terminates frame

// Heartbeat comment (every 25 seconds)
: ping

The double newline (\n\n) at the end of each frame is the SSE protocol delimiter. The browser's EventSource parser splits on it automatically. The heartbeat comment (lines beginning with :) is ignored by the browser but keeps proxies and load balancers from treating a silent connection as dead.

Stream lifecycle

The createEventStream function returns a ReadableStream<Uint8Array>. Inside its start() callback it:

1. Subscribes to the event bus. Every published event is immediately encoded and enqueued to the stream.
2. Starts a setInterval that enqueues a : ping\n\n comment every 25 seconds.
3. Attaches an abort listener to request.signal. When the browser disconnects, the signal fires, cleanup() runs: the bus subscription is removed, the interval is cleared.

Typed event shape

The browser hook validates incoming data with isBusEvent() before processing it. This function checks the type field against the full BUS_EVENT_TYPES set, so malformed or future-unknown event types are silently discarded rather than crashing the state update.

// src/types/bus-event.ts
export function isBusEvent(value: unknown): value is BusEvent {
  if (typeof value !== "object" || value === null) return false;
  const type = (value as Record<string, unknown>).type;
  return typeof type === "string" && BUS_EVENT_TYPES.has(type as BusEvent["type"]);
}

useWorkerEvents Hook

This is the single consumer of the SSE stream in the browser. All board columns and the worker detail page call it. It owns the live worker list for the duration of the component's mount.

Signature

function useWorkerEvents(
  initialWorkers: WorkerRow[],           // server-rendered snapshot from page props
  initialResources: ResourceSnapshot | null,
  repoId: string | null,
): UseWorkerEventsResult

// Returns:
type UseWorkerEventsResult = {
  workers:   WorkerRow[];           // live-updating worker list
  resources: ResourceSnapshot | null; // latest RSS/CPU sample
  connected: boolean;               // false while reconnecting
  refresh:   () => Promise<void>;   // force a full snapshot fetch
  subscribe: (cb: LiveEventSubscriber) => () => void; // raw event tap
};

What SSE drives vs. what stays in initialWorkers

This is the most important design decision in the hook. Not all worker fields arrive via SSE. Some are fetched once at page load and remain stable; others are updated continuously via events.

The rationale: SSE events carry the minimum data needed for live status display. Loading all PR and CI fields into bus events would bloat every worker.state_changed message and duplicate data already correctly served by the REST snapshot. On certain transitions that are known to have updated cost or timing data - such as waiting_merge to merging - the hook automatically re-fetches the full snapshot from /api/workers.

The applyEvent reducer

State is updated by a pure function exported for testing. The board never mutates worker rows directly.

// src/app/_components/use-worker-events.ts
export function applyEvent(state: WorkerRow[], event: BusEvent): WorkerRow[] {
  switch (event.type) {
    case "worker.claimed":
      // prepend a new row if not already present (idempotent)
      if (state.some((w) => w.id === event.workerId)) return state;
      return [buildRow(event), ...state];
    case "worker.state_changed":
      return state.map((w) =>
        w.id !== event.workerId ? w : {
          ...w,
          status: event.to,
          ...(event.to === "paused" ? { statusBeforePause: event.from } : {}),
        }
      );
    // ... worker.failed, worker.completed, worker.github_state_changed ...
    default:
      return state; // repo, batch, resource events do not affect worker rows
  }
}

Reconnection with exponential backoff

const RECONNECT_INITIAL_MS = 1_000;
const RECONNECT_MAX_MS     = 30_000;

// On onerror:
source.close();
setConnected(false);
scheduleReconnect(); // delay doubles each attempt, caps at 30 s

// On onopen:
reconnectDelay.current = RECONNECT_INITIAL_MS; // reset
setConnected(true);
void fetchSnapshot();  // re-sync missed events

The connected boolean lets the UI show a subtle indicator when disconnected. The reconnection itself is invisible to the user - the snapshot fetch on reconnect catches up any state changes that arrived while the connection was down.

Event Summarization

The Claude agent SDK emits a stream of structured SDKMessage objects: assistant turns with content blocks, tool call frames, tool results, system init messages, and streaming deltas. These objects are not suitable for display. The summarizer converts them into compact, human-readable log lines that appear on the worker detail page.

Factory: one summarizer per session

// src/server/runs/summarize.ts
export function createEventSummarizer(): (event: DriverEvent) => SummarizedLine | null {
  const toolNames = new Map<string, string>(); // tool_use_id -> tool_name
  return (event) => {
    if (event.type === "message" && event.payload.type === "assistant") {
      extractToolNames(event.payload.message?.content, toolNames);
    }
    return summarizeEvent(event, toolNames);
  };
}

Create one instance per driver session and never share it across sessions. The toolNames map is session-scoped: tool_use_id values are only unique within a single agent session, not globally.

Why tool attribution matters

Tool results arrive in a separate SDK message from the tool call itself, identified only by tool_use_id. Without the map, a tool_result event cannot know which tool produced it. This matters because the summarizer intentionally drops most tool results as noise - a Read or Grep result is usually a file dump that adds nothing to the log. Signal is extracted only from Bash results (test tallies, exit codes). Without tool attribution, a file that happens to contain the text "54 errors" would be logged as a phantom failure.

// Only Bash output carries test/build tallies worth a log line.
// A Read/Grep file dump matching COUNT_RE is a false positive, so skip it.
const toolName = toolNames.get(b.tool_use_id);
if (toolName !== undefined && toolName !== "Bash") return null;

Noise filtering

The summarizer returns null for events that carry no actionable information:

• Streaming deltas and status pings (emitted frequently by the SDK; never complete)
• Empty text blocks in assistant messages
• Read, Grep, and other non-Bash tool results without test/build signal
• system messages with subtypes other than init

SummarizedLine shape

type SummarizedLine = {
  message: string;    // human-readable, max 200 chars (condensed)
  level:   LogLevel;  // "info" | "warn" | "error"
};

Each SummarizedLine is published as a worker.event_logged or run.event_logged bus event, which is then persisted to the Event table and forwarded to all connected browsers via SSE.

Complete Event Flow

Here is the exact path a status change takes from the daemon to the browser. On localhost with no network round trip, total latency from state change to visible re-render is typically under 50 ms.

Step-by-step: implementing to merging