diff --git a/docs/design/agent-event-stream-redesign.md b/docs/design/agent-event-stream-redesign.md deleted file mode 100644 index 06c1cd69b..000000000 --- a/docs/design/agent-event-stream-redesign.md +++ /dev/null @@ -1,170 +0,0 @@ -# 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`) - -```typescript -interface AgentSession { - readonly provider: AgentProvider; - readonly id: string | null; - readonly capabilities: AgentCapabilityFlags; - - // Turn lifecycle - startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>; - interrupt(): Promise; - - // Event delivery (push-based) - subscribe(callback: (event: AgentStreamEvent) => void): () => void; - - // History (hydration only — never live dispatch) - streamHistory(): AsyncGenerator; - - // Run (uses startTurn + subscribe internally) - run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise; - - // Session metadata (unchanged) - getRuntimeInfo(): Promise; - getAvailableModes(): Promise; - getCurrentMode(): Promise; - setMode(modeId: string): Promise; - getPendingPermissions(): AgentPermissionRequest[]; - respondToPermission(requestId: string, response: AgentPermissionResponse): Promise; - describePersistence(): AgentPersistenceHandle | null; - close(): Promise; - listCommands?(): Promise; - setModel?(modelId: string | null): Promise; - setThinkingOption?(thinkingOptionId: string | null): Promise; -} -``` - -### 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` - -Yields persisted timeline items from prior sessions. Hydration only. Does NOT yield live events. - -#### `interrupt(): Promise` - -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 - -```typescript -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` 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