chore: remove stale agent-event-stream-redesign doc
This commit is contained in:
Mohamed Boudra
@@ -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<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
|
|
||||||
|
|
||||||
```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<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
|
|
||||||
Reference in New Issue
Block a user