paseo/docs/design/agent-event-stream-redesign.md

Agent Event Stream Redesign

Status: Implemented (2026-03-24)

Problem

The Claude provider had three event paths delivering the same events to the agent-manager:

1. Foreground stream (stream() → activeForegroundTurn.queue)
2. Live event pump (streamLiveEvents() → liveEventQueue) fed by the query pump
3. JSONL history poller (startLiveHistoryPolling() → routeSdkMessageFromPump())

Routing between paths was timing-based (Boolean(activeForegroundTurn), pendingRun). This caused:

• Duplicate user messages: trailing SDK events routed to the live queue after activeForegroundTurn cleared
• Stuck running state: stale turn_started from the live path flipped lifecycle back to running after finalize set it to terminal
• Fragile dedup: shouldSuppressLiveUserMessageEcho checked pendingRun (already null) and messageId (Claude assigns its own UUID)

Codex and OpenCode were stable because they had ONE event path with no routing decision.

Design

Core principle

One event source per provider session. Identity-based turn ownership, not timing-based routing.

Provider contract (AgentSession)

interface AgentSession {
  readonly provider: AgentProvider;
  readonly id: string | null;
  readonly capabilities: AgentCapabilityFlags;

  // Turn lifecycle
  startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>;
  interrupt(): Promise<void>;

  // Event delivery (push-based)
  subscribe(callback: (event: AgentStreamEvent) => void): () => void;

  // History (hydration only — never live dispatch)
  streamHistory(): AsyncGenerator<AgentStreamEvent>;

  // Run (uses startTurn + subscribe internally)
  run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult>;

  // Session metadata (unchanged)
  getRuntimeInfo(): Promise<AgentRuntimeInfo>;
  getAvailableModes(): Promise<AgentMode[]>;
  getCurrentMode(): Promise<string | null>;
  setMode(modeId: string): Promise<void>;
  getPendingPermissions(): AgentPermissionRequest[];
  respondToPermission(requestId: string, response: AgentPermissionResponse): Promise<void>;
  describePersistence(): AgentPersistenceHandle | null;
  close(): Promise<void>;
  listCommands?(): Promise<AgentSlashCommand[]>;
  setModel?(modelId: string | null): Promise<void>;
  setThinkingOption?(thinkingOptionId: string | null): Promise<void>;
}

Method contracts

startTurn(prompt, options?): Promise<{ turnId: string }>

Initiates a foreground turn. The provider validates readiness, generates a unique turnId, submits the prompt to the runtime, and resolves once accepted. Resolving means the prompt was accepted — not that the turn has started processing.

Rejects if: session not connected, foreground turn already active, runtime rejects prompt.

subscribe(callback): () => void

Registers a callback that receives ALL provider events — foreground and autonomous — in provider order. Returns an unsubscribe function. Events carry turnId when they belong to a turn.

streamHistory(): AsyncGenerator<AgentStreamEvent>

Yields persisted timeline items from prior sessions. Hydration only. Does NOT yield live events.

interrupt(): Promise<void>

Cancels the active foreground turn. The resulting turn_canceled event arrives via subscribe().

Provider-side guarantees

1. Per-session ordering: callbacks invoked in provider event order
2. No concurrent callback execution: serialized delivery per session
3. Subscribe-before-start safety: manager subscribes at session creation, before any startTurn() call — no events missed
4. Callback error isolation: subscriber throws → provider logs and continues
5. Deterministic cleanup: close() stops all callbacks; unsubscribe() stops that specific callback

Event tagging

All turn-scoped events carry turnId: string. Providers stamp turnId in notifySubscribers() from the active turn state (activeForegroundTurnId or autonomousTurn.id). The manager derives turn kind (foreground vs autonomous) by comparing against its own activeForegroundTurnId.

User message dedup

Claude SDK assigns its own UUID to user messages (does not preserve ours). The provider deduplicates user_message echoes by text content against the most recent foreground prompt.

Manager

Single subscription per session

When a session is loaded, the manager subscribes once via session.subscribe(). This is the only live input path. Events flow through a single dispatcher that handles lifecycle projection, foreground turn waiters, and UI updates.

Lifecycle projection from turn identity

• After startTurn() resolves: foreground turn is active
• On turn_started for active foreground turnId: lifecycle = running
• On terminal for active foreground turnId: lifecycle = idle or error, clear foreground turn
• On autonomous turn_started: lifecycle = running
• On autonomous terminal: lifecycle = idle or error

streamAgent() as filtered view

async *streamAgent(agentId, prompt, options) {
  const { turnId } = await session.startTurn(prompt, options);
  agent.activeForegroundTurnId = turnId;

  // Foreground turn waiter yields events matching this turnId
  // Ends when terminal event for turnId arrives
}

State model

Concept	Implementation
Foreground turn tracking	activeForegroundTurnId: string | null
Lifecycle projection	From turn events via turnId matching
Cancellation	session.interrupt() + await waiter settlement

What was deleted

• stream() from AgentSession interface and all providers
• Pushable<T> async queue from all providers
• streamLiveEvents() capability
• activeForegroundTurn + foreground queue in Claude provider
• liveEventQueue in Claude provider
• routeSdkMessageFromPump() timing-based routing (simplified to direct dispatch)
• startLiveEventPump() in manager
• liveEventBacklog + flushLiveEventBacklog() in manager
• shouldSuppressLiveUserMessageEcho() in manager
• startLiveHistoryPolling() for live dispatch
• snapHistoryOffsetToEnd()
• pendingRun as iterator reference

Integration tests

All tests run against real Claude sessions with credentials from .env.test. No mocks.

File: packages/server/src/server/agent/providers/__tests__/claude-agent.event-stream.integration.test.ts

Test	What it verifies
Basic foreground turn	startTurn → events via subscribe → terminal with matching turnId
No duplicate user_messages	Exactly ONE user_message per prompt, even after terminal
Lifecycle doesn't get stuck	No stale turn_started after terminal for same turnId
Autonomous run	sleep 5 in bg → idle → autonomous wake → idle (distinct turnIds)
Interruption	Start long task → interrupt → turn_canceled arrives
Sequential turns	Two turns produce distinct turnIds, no cross-contamination
Fast-fail	Quick error produces clean terminal, no stale events
User message dedup	Exactly one user_message with matching text in event log

Invariants (asserted on every test)

1. For each foreground turnId, exactly ONE user_message event
2. Every turn_started has exactly one matching terminal
3. After terminal for a foreground turnId, no later event with that turnId gets projected as autonomous
4. Autonomous turns between foreground turns are visible with distinct turnIds