Durable stream agent with background execution and live UI

[KyleAMathews]

Screen.Recording.2026-03-18.at.4.51.26.PM.mp4

Demo of a Vercel AI SDK agent running as a BullMQ background job, with all events captured into a durable state stream. A React client live-tails the stream via StreamDB and renders an incremental timeline — text streaming in token by token, tool calls appearing with expandable args/results, follow-up messages continuing the conversation.

Built as a prototype for Runable, who need to move long-running AI agents (2-5 hours, 50k concurrent streams) out of the HTTP request and into background workers with durable, resumable event streams.

Approach

The stream is the source of truth. Every agent lifecycle event (run start/end, step start/end, text deltas, tool calls) is written to a durable stream as typed state events following the State Protocol. The client materializes these into TanStack DB collections and renders reactively.

Key design decisions:

• OutboundBridge adapter — maps AI SDK `fullStream` events to typed `ChangeEvent` writes using schema helpers (`schema.runs.insert()`, etc.).

• Multi-collection schema — 7 entity types (runs, steps, texts, textDeltas, toolCalls, errors, inbox) with insert/update lifecycle semantics.

• Per-collection live queries — `useLiveQuery` with direct collection references (not `createLiveQueryCollection` derived queries) for reliable incremental updates. Collections are merged into `TimelineRow[]` in JS and sorted by `_seq`.

• Stream as conversation memory — follow-up messages are written to the same stream. The worker reconstructs conversation history by reading the stream's inbox and text_delta events, then passes the full message array to the agent.

Key invariants

• Every event value includes `_seq` for deterministic cross-collection ordering
• Text keys are scoped by step (`step-N-text-M`) to prevent duplicate key collisions across agent loop iterations
• Tool call updates include `args` (StreamDB replaces rows on update, doesn't merge)
• The API creates the durable stream before returning `runId` (eliminates client preload race)
• Input is disabled while the agent is working (prevents `_seq` collision between API and worker writes)

Verification

# Unit tests (OutboundBridge + timeline builder)
bun test src/lib/outbound-bridge.test.ts

# E2e tests (StreamDB incremental updates) — requires no other server on port 4437
bun test src/lib/e2e.test.ts

# Manual test
bun run dev     # terminal 1: API + stream server
bun run worker  # terminal 2: BullMQ worker (needs Redis + ANTHROPIC_API_KEY)
# Open http://localhost:4000, submit a prompt, watch live streaming
# Send a follow-up after completion, verify conversation continues

Files changed

File	What changed
`src/lib/schema.ts`	State schema: 7 entity types with passthrough validators
`src/lib/outbound-bridge.ts`	Maps AI SDK fullStream → typed ChangeEvents with `_seq` stamping
`src/lib/run-agent.ts`	Wires agent.stream() → bridge → handle.append(), reconstructs conversation from stream
`src/api.ts`	POST /api/agent (new run) + POST /api/agent/:runId/message (follow-up), input validation
`src/worker.ts`	BullMQ agent-run handler
`src/lib/durable.ts`	DurableStreamTestServer config
`src/lib/agent.ts`	Switch to @ai-sdk/anthropic provider, fix eval expression/statement handling
`src/web/AgentChat.tsx`	Chat UI: sidebar, per-collection live queries, timeline rendering, follow-up input
`src/web/timeline.ts`	buildTimeline: materialized-state-aware section builder with ensureAgentSection pattern
`src/web/use-agent-db.ts`	StreamDB lifecycle hook with preload error handling
`src/web/styles.css`	Sidebar layout, tool call cards, status indicators
`src/lib/outbound-bridge.test.ts`	11 unit tests: lifecycle, key scoping, args preservation, seq continuation
`src/lib/e2e.test.ts`	4 e2e tests: incremental updates, sequential writes, subscribeChanges reactivity
`docs/`	Design spec + implementation plan

🤖 Generated with Claude Code