refactor(server): replace stream() with subscribe() + startTurn() across all providers

Replace three competing event paths (foreground stream, live event pump,
JSONL history poller) with a single push-based subscribe() + startTurn()
contract. This fixes duplicate user messages and stuck running state caused
by timing-based routing between concurrent event sources.

Key changes:
- AgentSession interface: remove stream(), add subscribe() and startTurn()
- All providers (Claude, Codex, OpenCode): single subscribers Set with
  notifySubscribers() for push-based event delivery and turnId stamping
- Agent manager: identity-based turn ownership via activeForegroundTurnId
  replacing pendingRun async generator
- Delete: dual queues, routeSdkMessageFromPump, startLiveHistoryPolling,
  snapHistoryOffsetToEnd, liveEventBacklog, Pushable
- Fix Codex provider not clearing activeForegroundTurnId on turn completion
- Add real-provider integration tests for event stream invariants
This commit is contained in:
Mohamed Boudra
2026-03-24 21:22:00 +07:00
parent 6cc81e3ab4
commit a54687d3ef
23 changed files with 2501 additions and 1708 deletions
+170
View File
@@ -0,0 +1,170 @@
# 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
+2 -1
View File
@@ -1069,7 +1069,8 @@ export class HostRuntimeController {
}
const REGISTRY_STORAGE_KEY = "@paseo:daemon-registry";
const DEFAULT_LOCALHOST_ENDPOINT = "localhost:6767";
const DEFAULT_LOCALHOST_ENDPOINT =
process.env.EXPO_PUBLIC_LOCAL_DAEMON?.trim() || "localhost:6767";
const DEFAULT_LOCALHOST_BOOTSTRAP_KEY = "@paseo:default-localhost-bootstrap-v1";
const DEFAULT_LOCALHOST_BOOTSTRAP_TIMEOUT_MS = 2500;
const E2E_STORAGE_KEY = "@paseo:e2e";
@@ -137,10 +137,7 @@ function startAgentRun(
logger: Logger,
options?: { replaceRunning?: boolean },
): void {
const snapshot = agentManager.getAgent(agentId);
const shouldReplace =
options?.replaceRunning &&
Boolean(snapshot && (snapshot.lifecycle === "running" || snapshot.pendingRun));
const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
const iterator = shouldReplace
? agentManager.replaceAgentRun(agentId, prompt)
: agentManager.streamAgent(agentId, prompt);
@@ -527,7 +524,7 @@ export async function createAgentManagementMcpServer(
throw new Error(`Agent ${agentId} not found`);
}
if (snapshot.lifecycle === "running" || snapshot.pendingRun) {
if (agentManager.hasInFlightRun(agentId)) {
waitTracker.cancel(agentId, "Agent run interrupted by new prompt");
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -40,8 +40,8 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
} = overrides;
const sessionValue = lifecycle === "closed" ? null : (restOverrides.session ?? ({} as any));
const pendingRunValue =
restOverrides.pendingRun ?? (lifecycle === "running" ? (async function* noop() {})() : null);
const activeForegroundTurnIdValue =
restOverrides.activeForegroundTurnId ?? (lifecycle === "running" ? "test-turn-id" : null);
const lastErrorValue =
restOverrides.lastError ?? (lifecycle === "error" ? "encountered error" : undefined);
@@ -69,7 +69,9 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
],
currentModeId: "plan",
pendingPermissions: pendingPermissionsOverride ?? new Map<string, AgentPermissionRequest>(),
pendingRun: pendingRunValue as ManagedAgent["pendingRun"],
activeForegroundTurnId: activeForegroundTurnIdValue,
foregroundTurnWaiters: new Set(),
unsubscribeSession: null,
timeline: [],
runtimeInfo: {
provider: "claude",
@@ -259,23 +259,25 @@ export type AgentTimelineItem =
export type AgentStreamEvent =
| { type: "thread_started"; sessionId: string; provider: AgentProvider }
| { type: "turn_started"; provider: AgentProvider }
| { type: "turn_completed"; provider: AgentProvider; usage?: AgentUsage }
| { type: "turn_started"; provider: AgentProvider; turnId?: string }
| { type: "turn_completed"; provider: AgentProvider; usage?: AgentUsage; turnId?: string }
| {
type: "turn_failed";
provider: AgentProvider;
error: string;
code?: string;
diagnostic?: string;
turnId?: string;
}
| { type: "turn_canceled"; provider: AgentProvider; reason: string }
| { type: "timeline"; item: AgentTimelineItem; provider: AgentProvider }
| { type: "permission_requested"; provider: AgentProvider; request: AgentPermissionRequest }
| { type: "turn_canceled"; provider: AgentProvider; reason: string; turnId?: string }
| { type: "timeline"; item: AgentTimelineItem; provider: AgentProvider; turnId?: string }
| { type: "permission_requested"; provider: AgentProvider; request: AgentPermissionRequest; turnId?: string }
| {
type: "permission_resolved";
provider: AgentProvider;
requestId: string;
resolution: AgentPermissionResponse;
turnId?: string;
}
| {
type: "attention_required";
@@ -391,7 +393,8 @@ export interface AgentSession {
readonly id: string | null;
readonly capabilities: AgentCapabilityFlags;
run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult>;
stream(prompt: AgentPromptInput, options?: AgentRunOptions): AsyncGenerator<AgentStreamEvent>;
startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>;
subscribe(callback: (event: AgentStreamEvent) => void): () => void;
streamHistory(): AsyncGenerator<AgentStreamEvent>;
getRuntimeInfo(): Promise<AgentRuntimeInfo>;
getAvailableModes(): Promise<AgentMode[]>;
@@ -402,19 +405,8 @@ export interface AgentSession {
describePersistence(): AgentPersistenceHandle | null;
interrupt(): Promise<void>;
close(): Promise<void>;
/**
* List available slash commands for this session.
* Commands are provider-specific - Claude supports skills and built-in commands.
*/
listCommands?(): Promise<AgentSlashCommand[]>;
/**
* Update the model used for subsequent turns (if supported by provider).
*/
setModel?(modelId: string | null): Promise<void>;
/**
* Update the thinking/effort setting used for subsequent turns (if supported).
* Normalized to a string option id (provider-specific interpretation).
*/
setThinkingOption?(thinkingOptionId: string | null): Promise<void>;
}
@@ -15,12 +15,12 @@ import type {
type ManagedAgentOverrides = Omit<
Partial<ManagedAgent>,
"config" | "pendingPermissions" | "session" | "pendingRun"
"config" | "pendingPermissions" | "session" | "activeForegroundTurnId"
> & {
config?: Partial<AgentSessionConfig>;
pendingPermissions?: Map<string, AgentPermissionRequest>;
session?: AgentSession | null;
pendingRun?: ManagedAgent["pendingRun"];
activeForegroundTurnId?: string | null;
runtimeInfo?: ManagedAgent["runtimeInfo"];
attention?: ManagedAgent["attention"];
};
@@ -42,8 +42,8 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
mcpServers: configOverrides.mcpServers,
};
const session = lifecycle === "closed" ? null : (overrides.session ?? ({} as AgentSession));
const pendingRun =
overrides.pendingRun ?? (lifecycle === "running" ? (async function* noop() {})() : null);
const activeForegroundTurnId =
overrides.activeForegroundTurnId ?? (lifecycle === "running" ? "test-turn-id" : null);
const agent: ManagedAgent = {
id: overrides.id ?? "agent-test",
@@ -65,7 +65,9 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
availableModes: overrides.availableModes ?? [],
currentModeId: overrides.currentModeId ?? config.modeId ?? null,
pendingPermissions: overrides.pendingPermissions ?? new Map<string, AgentPermissionRequest>(),
pendingRun,
activeForegroundTurnId,
foregroundTurnWaiters: new Set(),
unsubscribeSession: null,
timeline: overrides.timeline ?? [],
attention: overrides.attention ?? { requiresAttention: false },
runtimeInfo: overrides.runtimeInfo ?? {
@@ -199,10 +199,7 @@ function startAgentRun(
logger: Logger,
options?: { replaceRunning?: boolean },
): void {
const snapshot = agentManager.getAgent(agentId);
const shouldReplace =
options?.replaceRunning &&
Boolean(snapshot && (snapshot.lifecycle === "running" || snapshot.pendingRun));
const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
const iterator = shouldReplace
? agentManager.replaceAgentRun(agentId, prompt)
: agentManager.streamAgent(agentId, prompt);
@@ -696,7 +693,7 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
throw new Error(`Agent ${agentId} not found`);
}
if (snapshot.lifecycle === "running" || snapshot.pendingRun) {
if (agentManager.hasInFlightRun(agentId)) {
waitTracker.cancel(agentId, "Agent run interrupted by new prompt");
}
@@ -0,0 +1,477 @@
/**
* Integration tests for the agent event stream redesign (Unit 3).
*
* These tests verify the behavioral guarantees of the new provider contract
* (`startTurn` + `subscribe`) as specified in docs/design/agent-event-stream-redesign.md.
*
* All tests use REAL Claude SDK sessions — no mocks.
*
* CREDENTIALS: These tests require a running `claude` CLI and either
* CLAUDE_CODE_OAUTH_TOKEN or ANTHROPIC_API_KEY in the environment.
* They are skipped automatically when credentials are unavailable.
*/
import { describe, expect, test } from "vitest";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import pino from "pino";
import type { AgentSession, AgentStreamEvent } from "../../agent-sdk-types.js";
import { isCommandAvailable } from "../../provider-launch-config.js";
import { ClaudeAgentClient } from "../claude-agent.js";
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
const logger = pino({ level: "silent" });
const client = new ClaudeAgentClient({ logger });
const hasClaudeCredentials =
!!process.env.CLAUDE_CODE_OAUTH_TOKEN || !!process.env.ANTHROPIC_API_KEY;
const canRun = isCommandAvailable("claude") && hasClaudeCredentials;
function tmpCwd(prefix: string): string {
return mkdtempSync(path.join(tmpdir(), prefix));
}
function isTerminalEvent(event: AgentStreamEvent): boolean {
return (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
);
}
// turnId is optional on AgentStreamEvent — this narrows to events where it's present.
type EventWithTurnId = AgentStreamEvent & { turnId: string };
function hasTurnId(event: AgentStreamEvent): event is EventWithTurnId {
return "turnId" in event && typeof (event as Record<string, unknown>).turnId === "string";
}
function eventsForTurn(events: AgentStreamEvent[], turnId: string): AgentStreamEvent[] {
return events.filter((e) => hasTurnId(e) && e.turnId === turnId);
}
function userMessagesWithText(events: AgentStreamEvent[], text: string): AgentStreamEvent[] {
return events.filter(
(e) =>
e.type === "timeline" &&
e.item.type === "user_message" &&
e.item.text === text,
);
}
async function createSession(params?: {
cwdPrefix?: string;
}): Promise<{ cwd: string; session: AgentSession }> {
const cwd = tmpCwd(params?.cwdPrefix ?? "event-stream-integration-");
const session = await client.createSession({
provider: "claude",
cwd,
title: "event-stream integration",
modeId: "acceptEdits",
model: "haiku",
});
return { cwd, session };
}
async function cleanupSession(handle: { cwd: string; session: AgentSession }): Promise<void> {
await handle.session.close().catch(() => undefined);
rmSync(handle.cwd, { recursive: true, force: true });
}
async function startTurnAndCollectEvents(
session: AgentSession,
prompt: string,
options?: { extraMs?: number; timeoutMs?: number },
): Promise<{ turnId: string; events: AgentStreamEvent[] }> {
const { extraMs = 0, timeoutMs = 45_000 } = options ?? {};
return await new Promise((resolve, reject) => {
const events: AgentStreamEvent[] = [];
let turnId: string | null = null;
let settled = false;
const timeout = setTimeout(() => {
unsubscribe();
reject(new Error(`Timed out after ${timeoutMs}ms waiting for terminal event`));
}, timeoutMs);
const finish = () => {
if (settled || !turnId) {
return;
}
settled = true;
clearTimeout(timeout);
unsubscribe();
resolve({ turnId, events });
};
const unsubscribe = session.subscribe((event) => {
events.push(event);
if (!turnId) {
return;
}
if (!isTerminalEvent(event) || !hasTurnId(event) || event.turnId !== turnId) {
return;
}
if (extraMs > 0) {
setTimeout(finish, extraMs);
return;
}
finish();
});
void session
.startTurn(prompt)
.then((result) => {
turnId = result.turnId;
})
.catch((error) => {
clearTimeout(timeout);
unsubscribe();
reject(error);
});
});
}
// ---------------------------------------------------------------------------
// Invariant assertions — run after every test
// ---------------------------------------------------------------------------
function assertInvariants(events: AgentStreamEvent[], foregroundTurnIds: string[]): void {
// Invariant 1: For each foreground turnId, at most ONE user_message event.
// The manager records foreground prompts separately, so provider echoes may be suppressed.
for (const turnId of foregroundTurnIds) {
const turnEvents = eventsForTurn(events, turnId);
const userMsgs = turnEvents.filter(
(e) => e.type === "timeline" && e.item.type === "user_message",
);
expect(
userMsgs.length,
`Expected at most 1 user_message for turnId ${turnId}, got ${userMsgs.length}`,
).toBeLessThanOrEqual(1);
}
// Invariant 2: Every turn_started has exactly one matching terminal
const turnStartedIds = events
.filter((e) => e.type === "turn_started" && hasTurnId(e))
.map((e) => (e as EventWithTurnId).turnId);
for (const turnId of turnStartedIds) {
const terminals = eventsForTurn(events, turnId).filter(isTerminalEvent);
expect(
terminals.length,
`Expected exactly 1 terminal for turnId ${turnId}, got ${terminals.length}`,
).toBe(1);
}
// Invariant 3: After terminal for a foreground turnId, no later event with
// that turnId appears (would indicate stale routing to autonomous)
for (const turnId of foregroundTurnIds) {
const allWithTurn = events
.map((e, i) => ({ event: e, index: i }))
.filter(({ event }) => hasTurnId(event) && event.turnId === turnId);
const terminalEntry = allWithTurn.find(({ event }) => isTerminalEvent(event));
if (!terminalEntry) continue;
const afterTerminal = allWithTurn.filter(({ index }) => index > terminalEntry.index);
expect(
afterTerminal.length,
`No events should appear for foreground turnId ${turnId} after terminal`,
).toBe(0);
}
// Invariant 4: Autonomous turns have distinct turnIds from foreground turns
const allTurnIds = new Set(
events.filter(hasTurnId).map((e) => e.turnId),
);
const autonomousTurnIds = [...allTurnIds].filter(
(id) => !foregroundTurnIds.includes(id),
);
for (const autoId of autonomousTurnIds) {
expect(foregroundTurnIds).not.toContain(autoId);
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
describe("Agent event stream redesign — integration", () => {
test.skipIf(!canRun)("Test 1: Basic foreground turn", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-basic-" });
try {
const { turnId, events } = await startTurnAndCollectEvents(
handle.session,
"respond with just the word hello",
);
const turnStarted = events.find(
(e) => e.type === "turn_started" && hasTurnId(e) && e.turnId === turnId,
);
expect(turnStarted).toBeDefined();
const terminal = events.find(
(e) => isTerminalEvent(e) && hasTurnId(e) && e.turnId === turnId,
);
expect(terminal).toBeDefined();
assertInvariants(events, [turnId]);
} finally {
await cleanupSession(handle);
}
}, 60_000);
test.skipIf(!canRun)("Test 2: No duplicate user_messages — THE BUG", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-dedup-" });
try {
const { turnId, events } = await startTurnAndCollectEvents(handle.session, "say hi", {
extraMs: 3_000,
});
expect(userMessagesWithText(events, "say hi").length).toBeLessThanOrEqual(1);
// No turn_started after terminal for the same turnId
const terminalIdx = events.findIndex(
(e) => isTerminalEvent(e) && hasTurnId(e) && e.turnId === turnId,
);
const staleTurnStarted = events.slice(terminalIdx + 1).filter(
(e) => e.type === "turn_started" && hasTurnId(e) && e.turnId === turnId,
);
expect(staleTurnStarted.length).toBe(0);
assertInvariants(events, [turnId]);
} finally {
await cleanupSession(handle);
}
}, 60_000);
test.skipIf(!canRun)("Test 3: Lifecycle doesn't get stuck in running", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-lifecycle-" });
try {
const { turnId, events } = await startTurnAndCollectEvents(handle.session, "say hi", {
extraMs: 3_000,
});
const terminalIdx = events.findIndex(
(e) => isTerminalEvent(e) && hasTurnId(e) && e.turnId === turnId,
);
const afterTerminal = events.slice(terminalIdx + 1);
// No subsequent turn_started for same turnId
expect(
afterTerminal.filter(
(e) => e.type === "turn_started" && hasTurnId(e) && e.turnId === turnId,
).length,
).toBe(0);
// Any turn_started after terminal must have a different turnId
for (const ts of afterTerminal.filter((e) => e.type === "turn_started" && hasTurnId(e))) {
expect((ts as EventWithTurnId).turnId).not.toBe(turnId);
}
assertInvariants(events, [turnId]);
} finally {
await cleanupSession(handle);
}
}, 60_000);
test.skipIf(!canRun)("Test 4: Autonomous run", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-autonomous-" });
const autonomousWakeToken = `AUTONOMOUS_WAKE_${Date.now().toString(36)}`;
try {
const { turnId: fgTurnId, events } = await startTurnAndCollectEvents(
handle.session,
[
"Use the Task tool to start a background sub-agent.",
"In that task, run the Bash command exactly: sleep 3 && echo BACKGROUND_DONE",
"Do not wait for task completion.",
"Reply immediately with exactly: SPAWNED",
`When the background task completes later, reply with exactly: ${autonomousWakeToken}`,
].join(" "),
{
extraMs: 10_000,
timeoutMs: 60_000,
},
);
const fgTerminalIdx = events.findIndex(
(e) => isTerminalEvent(e) && hasTurnId(e) && e.turnId === fgTurnId,
);
const afterForeground = events.slice(fgTerminalIdx + 1);
// Autonomous turn_started with a different turnId
const autoStarts = afterForeground.filter(
(e) => e.type === "turn_started" && hasTurnId(e) && e.turnId !== fgTurnId,
) as EventWithTurnId[];
if (autoStarts.length === 0) {
assertInvariants(events, [fgTurnId]);
return;
}
const autoTurnId = autoStarts[0]!.turnId;
expect(fgTurnId).not.toBe(autoTurnId);
// Autonomous turn reaches terminal
expect(
afterForeground.find(
(e) => isTerminalEvent(e) && hasTurnId(e) && e.turnId === autoTurnId,
),
).toBeDefined();
assertInvariants(events, [fgTurnId]);
} finally {
await cleanupSession(handle);
}
}, 90_000);
test.skipIf(!canRun)("Test 5: Interruption", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-interrupt-" });
try {
const { turnId } = await handle.session.startTurn(
"write a very long essay about the history of computing",
);
// Use a single subscription to avoid missing events between unsubscribe/resubscribe
const events = await new Promise<AgentStreamEvent[]>((resolve, reject) => {
const collected: AgentStreamEvent[] = [];
let interrupted = false;
const timeout = setTimeout(() => {
unsubscribe();
reject(new Error("Timed out after 45000ms waiting for terminal event"));
}, 45_000);
const unsubscribe = handle.session.subscribe((event) => {
collected.push(event);
// Once we see turn_started, fire the interrupt
if (
!interrupted &&
event.type === "turn_started" &&
hasTurnId(event) &&
event.turnId === turnId
) {
interrupted = true;
handle.session.interrupt().catch(() => undefined);
}
// Resolve when we get a terminal event for this turn
if (isTerminalEvent(event) && hasTurnId(event) && event.turnId === turnId) {
clearTimeout(timeout);
unsubscribe();
resolve(collected);
}
});
});
// turn_canceled or turn_failed arrives for that turnId
const terminal = events.find(
(e) =>
(e.type === "turn_canceled" || e.type === "turn_failed") &&
hasTurnId(e) &&
e.turnId === turnId,
);
expect(terminal).toBeDefined();
// No further events for that turnId after terminal
const terminalIdx = events.indexOf(terminal!);
expect(
events.slice(terminalIdx + 1).filter((e) => hasTurnId(e) && e.turnId === turnId).length,
).toBe(0);
assertInvariants(events, [turnId]);
} finally {
await cleanupSession(handle);
}
}, 60_000);
test.skipIf(!canRun)("Test 6: Sequential foreground turns", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-sequential-" });
try {
const { turnId: turnId1, events: events1 } = await startTurnAndCollectEvents(
handle.session,
"say first",
);
const { turnId: turnId2, events: events2 } = await startTurnAndCollectEvents(
handle.session,
"say second",
);
const allEvents = [...events1, ...events2];
expect(turnId1).not.toBe(turnId2);
// No events from turn 1 after turn 2 starts
const turn2StartIdx = allEvents.findIndex(
(e) => e.type === "turn_started" && hasTurnId(e) && e.turnId === turnId2,
);
expect(
allEvents.slice(turn2StartIdx + 1).filter((e) => hasTurnId(e) && e.turnId === turnId1)
.length,
).toBe(0);
assertInvariants(allEvents, [turnId1, turnId2]);
} finally {
await cleanupSession(handle);
}
}, 90_000);
test.skipIf(!canRun)("Test 7: Fast-fail", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-fast-fail-" });
try {
const { turnId, events } = await startTurnAndCollectEvents(handle.session, "", {
extraMs: 3_000,
});
// At most one turn_started
expect(
events.filter((e) => e.type === "turn_started" && hasTurnId(e) && e.turnId === turnId)
.length,
).toBeLessThanOrEqual(1);
// Terminal present
const terminal = events.find(
(e) => isTerminalEvent(e) && hasTurnId(e) && e.turnId === turnId,
);
expect(terminal).toBeDefined();
// No stale turn_started after terminal
const terminalIdx = events.indexOf(terminal!);
expect(
events.slice(terminalIdx + 1).filter((e) => e.type === "turn_started").length,
).toBe(0);
assertInvariants(events, [turnId]);
} finally {
await cleanupSession(handle);
}
}, 60_000);
test.skipIf(!canRun)("Test 8: User message dedup by text", async () => {
const handle = await createSession({ cwdPrefix: "event-stream-user-dedup-" });
try {
const { turnId, events } = await startTurnAndCollectEvents(handle.session, "hello world", {
extraMs: 3_000,
});
expect(userMessagesWithText(events, "hello world").length).toBeLessThanOrEqual(1);
assertInvariants(events, [turnId]);
} finally {
await cleanupSession(handle);
}
}, 60_000);
});
@@ -7,6 +7,7 @@ import pino from "pino";
import type { AgentSession, AgentStreamEvent, ToolCallTimelineItem } from "../agent-sdk-types.js";
import { isCommandAvailable } from "../provider-launch-config.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
const logger = pino({ level: "silent" });
const client = new ClaudeAgentClient({ logger });
@@ -92,6 +93,30 @@ async function collectUntil(
}
}
function collectSubscribedUntil(
session: AgentSession,
predicate: (event: AgentStreamEvent) => boolean,
timeoutMs = 45_000,
): Promise<AgentStreamEvent[]> {
return new Promise((resolve, reject) => {
const events: AgentStreamEvent[] = [];
const timeout = setTimeout(() => {
unsubscribe();
reject(new Error(`Timed out after ${timeoutMs}ms waiting for subscribed event`));
}, timeoutMs);
const unsubscribe = session.subscribe((event) => {
events.push(event);
if (!predicate(event)) {
return;
}
clearTimeout(timeout);
unsubscribe();
resolve(events);
});
});
}
function getAssistantText(events: AgentStreamEvent[]): string {
return events
.flatMap((event) => {
@@ -159,7 +184,7 @@ describe("ClaudeAgentSession integration", () => {
try {
const events = await collectUntilTerminal(
handle.session.stream("Respond with exactly: HELLO_WORLD"),
streamSession(handle.session, "Respond with exactly: HELLO_WORLD"),
);
expect(events[0]).toMatchObject({
@@ -190,7 +215,8 @@ describe("ClaudeAgentSession integration", () => {
try {
const events = await collectUntilTerminal(
handle.session.stream(
streamSession(
handle.session,
[
"Use the Bash tool.",
"Run exactly: echo TOOL_TEST_OUTPUT",
@@ -227,7 +253,8 @@ describe("ClaudeAgentSession integration", () => {
});
try {
const firstStream = handle.session.stream(
const firstStream = streamSession(
handle.session,
[
"Use the Bash tool.",
"Run exactly: sleep 10",
@@ -262,7 +289,7 @@ describe("ClaudeAgentSession integration", () => {
).toBe(true);
const followUpEvents = await collectUntilTerminal(
handle.session.stream("Respond with exactly: AFTER_INTERRUPT_OK"),
streamSession(handle.session, "Respond with exactly: AFTER_INTERRUPT_OK"),
);
const secondQuery = getInternalQuery(handle.session);
@@ -288,9 +315,9 @@ describe("ClaudeAgentSession integration", () => {
const autonomousWakeToken = `AUTONOMOUS_WAKE_${Date.now().toString(36)}`;
try {
const liveEventsStream = handle.session.streamLiveEvents();
const foregroundEvents = await collectUntilTerminal(
handle.session.stream(
streamSession(
handle.session,
[
"Use the Task tool to start a background sub-agent.",
"In that task, run the Bash command exactly: sleep 3 && echo BACKGROUND_DONE",
@@ -304,9 +331,11 @@ describe("ClaudeAgentSession integration", () => {
expect(compactText(getAssistantText(foregroundEvents))).toContain("spawned");
const liveEvents = await collectUntilTerminal(liveEventsStream, {
timeoutMs: 45_000,
});
const liveEvents = await collectSubscribedUntil(
handle.session,
(event) => isTerminalEvent(event),
45_000,
);
expect(
liveEvents.some((event) => event.type === "turn_started" && event.provider === "claude"),
@@ -334,7 +363,8 @@ describe("ClaudeAgentSession integration", () => {
try {
const events = await collectUntilTerminal(
handle.session.stream(
streamSession(
handle.session,
[
"Use the Bash tool to run exactly: printf 'PERM_TEST' > permission.txt",
"If approval is required, wait for approval.",
@@ -2,6 +2,7 @@ import { afterEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
import type { AgentStreamEvent } from "../agent-sdk-types.js";
type QueryMock = {
@@ -211,6 +212,21 @@ function collectAssistantText(events: AgentStreamEvent[]): string {
.join("");
}
function subscribeToEvents(session: { subscribe: (callback: (event: AgentStreamEvent) => void) => () => void }) {
const queue = createAsyncQueue<AgentStreamEvent>();
const unsubscribe = session.subscribe((event) => {
queue.push(event);
});
return {
next: () => queue.next(),
close: () => {
unsubscribe();
queue.end();
},
};
}
async function waitFor(
predicate: () => boolean,
options?: { timeoutMs?: number; intervalMs?: number },
@@ -250,7 +266,7 @@ describe("ClaudeAgentSession interrupt regression", () => {
cwd: process.cwd(),
});
const firstTurn = session.stream("first prompt");
const firstTurn = streamSession(session, "first prompt");
await firstTurn.next();
await waitFor(() => queries[0]?.prompts.length === 1);
@@ -270,7 +286,7 @@ describe("ClaudeAgentSession interrupt regression", () => {
await session.close();
});
test("pushes the next prompt into the existing query instead of rebuilding it", async () => {
test("reuses the existing query after interrupt before starting the next prompt", async () => {
const logger = createTestLogger();
const queries: ScriptedQuery[] = [];
@@ -300,11 +316,14 @@ describe("ClaudeAgentSession interrupt regression", () => {
cwd: process.cwd(),
});
const firstTurn = session.stream("first prompt");
const firstTurn = streamSession(session, "first prompt");
await firstTurn.next();
await waitFor(() => queries[0]?.prompts.length === 1);
const secondTurnEvents = await collectUntilTerminal(session.stream("second prompt"));
await session.interrupt();
await collectUntilTerminal(firstTurn);
const secondTurnEvents = await collectUntilTerminal(streamSession(session, "second prompt"));
expect(sdkMocks.query).toHaveBeenCalledTimes(1);
expect(queries[0]?.prompts.map((prompt) => prompt.text)).toEqual([
@@ -315,7 +334,6 @@ describe("ClaudeAgentSession interrupt regression", () => {
expect(queries[0]?.return).not.toHaveBeenCalled();
expect(collectAssistantText(secondTurnEvents)).toContain("SECOND_PROMPT_RESPONSE");
await firstTurn.return?.();
await session.close();
});
@@ -396,12 +414,12 @@ describe("ClaudeAgentSession interrupt regression", () => {
cwd: process.cwd(),
});
const firstTurn = session.stream("first prompt");
const firstTurn = streamSession(session, "first prompt");
await firstTurn.next();
await session.interrupt();
await collectUntilTerminal(firstTurn);
const secondTurnEvents = await collectUntilTerminal(session.stream("second prompt"));
const secondTurnEvents = await collectUntilTerminal(streamSession(session, "second prompt"));
expect(sdkMocks.query).toHaveBeenCalledTimes(1);
expect(prompts.map((prompt) => prompt.text)).toEqual(["first prompt", "second prompt"]);
@@ -442,9 +460,9 @@ describe("ClaudeAgentSession autonomous turns", () => {
cwd: process.cwd(),
});
await collectUntilTerminal(session.stream("seed prompt"));
await collectUntilTerminal(streamSession(session, "seed prompt"));
const liveIterator = session.streamLiveEvents();
const subscribedEvents = subscribeToEvents(session);
queryRef?.emit({
type: "assistant",
message: { content: "AUTONOMOUS_WAKE_RESPONSE" },
@@ -452,9 +470,9 @@ describe("ClaudeAgentSession autonomous turns", () => {
});
queryRef?.emit(buildSuccessResult("autonomous-live-session"));
const started = await liveIterator.next();
const timeline = await liveIterator.next();
const completed = await liveIterator.next();
const started = await subscribedEvents.next();
const timeline = await subscribedEvents.next();
const completed = await subscribedEvents.next();
expect(started.value).toMatchObject({ type: "turn_started", provider: "claude" });
expect(timeline.value).toMatchObject({
@@ -470,7 +488,7 @@ describe("ClaudeAgentSession autonomous turns", () => {
provider: "claude",
});
await liveIterator.return?.();
subscribedEvents.close();
await session.close();
});
@@ -512,19 +530,21 @@ describe("ClaudeAgentSession autonomous turns", () => {
cwd: process.cwd(),
});
await collectUntilTerminal(session.stream("seed prompt"));
await collectUntilTerminal(streamSession(session, "seed prompt"));
const liveIterator = session.streamLiveEvents();
const subscribedEvents = subscribeToEvents(session);
queryRef?.emit({
type: "assistant",
message: { content: "BACKGROUND_ONLY_RESPONSE" },
session_id: "autonomous-handoff-session",
});
const autonomousStart = await liveIterator.next();
const autonomousTimeline = await liveIterator.next();
const foregroundEvents = await collectUntilTerminal(session.stream("foreground prompt"));
const autonomousComplete = await liveIterator.next();
const autonomousStart = await subscribedEvents.next();
const autonomousTimeline = await subscribedEvents.next();
const foregroundEvents = await collectUntilTerminal(
streamSession(session, "foreground prompt"),
);
const autonomousComplete = await subscribedEvents.next();
expect(autonomousStart.value).toMatchObject({
type: "turn_started",
@@ -555,7 +575,7 @@ describe("ClaudeAgentSession autonomous turns", () => {
"foreground prompt",
]);
await liveIterator.return?.();
subscribedEvents.close();
await session.close();
});
});
@@ -3,6 +3,7 @@ import type { Logger } from "pino";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { ClaudeAgentClient, readEventIdentifiers } from "./claude-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
import type { AgentStreamEvent, AgentTimelineItem } from "../agent-sdk-types.js";
type QueryMock = {
@@ -693,7 +694,7 @@ describe("ClaudeAgentSession redesign invariants", () => {
const session = await createSession();
try {
const events = await Promise.race([
collectUntilTerminal(session.stream("metadata helper prompt")),
collectUntilTerminal(streamSession(session, "metadata helper prompt")),
new Promise<never>((_, reject) => {
setTimeout(
() => reject(new Error("Timed out waiting for foreground terminal event")),
@@ -855,18 +856,18 @@ describe("ClaudeAgentSession redesign invariants", () => {
});
streamCase = "success";
const successEvents = await collectUntilTerminal(session.stream("success prompt"));
const successEvents = await collectUntilTerminal(streamSession(session, "success prompt"));
expect(successEvents.some((event) => event.type === "turn_completed")).toBe(true);
expect(successEvents.some((event) => event.type === "turn_failed")).toBe(false);
expect(successEvents.some((event) => event.type === "turn_canceled")).toBe(false);
streamCase = "error";
const errorEvents = await collectUntilTerminal(session.stream("error prompt"));
const errorEvents = await collectUntilTerminal(streamSession(session, "error prompt"));
expect(errorEvents.some((event) => event.type === "turn_failed")).toBe(true);
expect(errorEvents.some((event) => event.type === "turn_completed")).toBe(false);
streamCase = "interrupt";
const interruptStream = session.stream("interrupt prompt");
const interruptStream = streamSession(session, "interrupt prompt");
const interruptEvents: AgentStreamEvent[] = [];
for await (const event of interruptStream) {
interruptEvents.push(event);
@@ -985,7 +986,7 @@ describe("ClaudeAgentSession redesign invariants", () => {
});
const session = await createSession();
const events = await collectUntilTerminal(session.stream("timeline prompt"));
const events = await collectUntilTerminal(streamSession(session, "timeline prompt"));
const assistantText = events
.filter(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
@@ -1104,7 +1105,7 @@ describe("ClaudeAgentSession redesign invariants", () => {
});
const session = await createSession();
const events = await collectUntilTerminal(session.stream("uuid fallback prompt"));
const events = await collectUntilTerminal(streamSession(session, "uuid fallback prompt"));
const assistantText = events
.filter(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
@@ -5,6 +5,7 @@ import type { AgentStreamEvent } from "../agent-sdk-types.js";
import type { AgentTimelineRow } from "../agent-manager.js";
import { projectTimelineRows } from "../timeline-projection.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
const sdkMocks = vi.hoisted(() => ({
query: vi.fn(),
@@ -255,7 +256,7 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
cwd: process.cwd(),
});
const events = await collectUntilTerminal(session.stream("delegate work"));
const events = await collectUntilTerminal(streamSession(session, "delegate work"));
await session.close();
const timelineToolCalls = events
@@ -316,7 +317,7 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
cwd: process.cwd(),
});
const events = await collectUntilTerminal(session.stream("delegate work"));
const events = await collectUntilTerminal(streamSession(session, "delegate work"));
await session.close();
const timelineToolCalls = events
@@ -29,7 +29,6 @@ import {
mapClaudeRunningToolCall,
} from "./claude/tool-call-mapper.js";
import {
coerceTaskNotificationHistoryRecordToSystemMessage,
mapTaskNotificationSystemRecordToToolCall,
mapTaskNotificationUserContentToToolCall,
} from "./claude/task-notification-tool-call.js";
@@ -83,16 +82,16 @@ type EventIdentifiers = {
messageId: string | null;
};
type ForegroundTurnState = {
id: string;
queue: Pushable<AgentStreamEvent>;
hasVisibleActivity: boolean;
};
type AutonomousTurnState = {
id: string;
};
type AsyncMessageInput<T> = {
push: (item: T) => void;
end: () => void;
iterable: AsyncIterable<T>;
};
type NormalizeClaudeRuntimeModelIdOptions = {
runtimeModelId: string;
supportedModelIds: ReadonlySet<string> | null;
@@ -1201,7 +1200,7 @@ class ClaudeAgentSession implements AgentSession {
private readonly logger: Logger;
private readonly queryFactory: typeof query;
private query: Query | null = null;
private input: Pushable<SDKUserMessage> | null = null;
private input: AsyncMessageInput<SDKUserMessage> | null = null;
private claudeSessionId: string | null;
private persistence: AgentPersistenceHandle | null;
private currentMode: PermissionMode;
@@ -1210,22 +1209,18 @@ class ClaudeAgentSession implements AgentSession {
private toolUseIndexToId = new Map<number, string>();
private toolUseInputBuffers = new Map<string, string>();
private pendingPermissions = new Map<string, PendingPermission>();
private activeForegroundTurn: ForegroundTurnState | null = null;
private activeForegroundTurnId: string | null = null;
private autonomousTurn: AutonomousTurnState | null = null;
private liveEventQueue = new Pushable<AgentStreamEvent>();
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private readonly timelineAssembler = new TimelineAssembler();
private readonly sidechainTracker = new ClaudeSidechainTracker({
getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null,
});
private persistedHistory: AgentTimelineItem[] = [];
private historyPending = false;
private historyOffsetSessionId: string | null = null;
private historyReadOffsetBytes = 0;
private historyLineFragment = "";
private turnState: TurnState = "idle";
private nextTurnOrdinal = 1;
private cancelCurrentTurn: (() => void) | null = null;
private activeTurnPromise: Promise<void> | null = null;
private cachedRuntimeInfo: AgentRuntimeInfo | null = null;
private lastOptionsModel: string | null = null;
private selectableModelIds: Set<string> | null = buildClaudeSelectableModelIds();
@@ -1235,8 +1230,8 @@ class ClaudeAgentSession implements AgentSession {
private queryPumpPromise: Promise<void> | null = null;
private queryRestartNeeded = false;
private pendingInterruptAbort = false;
private liveEventSubscriberCount = 0;
private liveHistoryPollTimer: NodeJS.Timeout | null = null;
private lastForegroundPromptText: string | null = null;
private foregroundHasVisibleActivity = false;
private userMessageIds: string[] = [];
private recentStderr = "";
private closed = false;
@@ -1292,12 +1287,23 @@ class ClaudeAgentSession implements AgentSession {
}
async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
const events = this.stream(prompt, options);
const timeline: AgentTimelineItem[] = [];
let finalText = "";
let usage: AgentUsage | undefined;
let turnId: string | null = null;
const bufferedEvents: AgentStreamEvent[] = [];
let settled = false;
let resolveCompletion!: () => void;
let rejectCompletion!: (error: Error) => void;
for await (const event of events) {
const processEvent = (event: AgentStreamEvent) => {
if (settled) {
return;
}
const eventTurnId = (event as { turnId?: string }).turnId;
if (turnId && eventTurnId && eventTurnId !== turnId) {
return;
}
if (event.type === "timeline") {
timeline.push(event.item);
if (event.item.type === "assistant_message") {
@@ -1309,11 +1315,48 @@ class ClaudeAgentSession implements AgentSession {
finalText += event.item.text;
}
}
} else if (event.type === "turn_completed") {
usage = event.usage;
} else if (event.type === "turn_failed") {
throw new Error(event.error);
return;
}
if (event.type === "turn_completed") {
usage = event.usage;
settled = true;
resolveCompletion();
return;
}
if (event.type === "turn_failed") {
settled = true;
rejectCompletion(new Error(event.error));
return;
}
if (event.type === "turn_canceled") {
settled = true;
resolveCompletion();
}
};
const completion = new Promise<void>((resolve, reject) => {
resolveCompletion = resolve;
rejectCompletion = reject;
});
const unsubscribe = this.subscribe((event) => {
if (!turnId) {
bufferedEvents.push(event);
return;
}
processEvent(event);
});
try {
const result = await this.startTurn(prompt, options);
turnId = result.turnId;
for (const event of bufferedEvents) {
processEvent(event);
}
if (!settled) {
await completion;
}
} finally {
unsubscribe();
}
this.cachedRuntimeInfo = {
@@ -1335,19 +1378,24 @@ class ClaudeAgentSession implements AgentSession {
};
}
async *stream(
async startTurn(
prompt: AgentPromptInput,
options?: AgentRunOptions,
): AsyncGenerator<AgentStreamEvent> {
void options;
if (this.cancelCurrentTurn) {
this.cancelCurrentTurn();
_options?: AgentRunOptions,
): Promise<{ turnId: string }> {
if (this.closed) {
throw new Error("Claude session is closed");
}
if (this.activeForegroundTurnId) {
throw new Error("A foreground turn is already active");
}
const slashCommand = this.resolveSlashCommandInvocation(prompt);
if (slashCommand?.commandName === REWIND_COMMAND_NAME) {
yield* this.streamRewindCommand(slashCommand);
return;
const turnId = this.createTurnId("foreground");
this.activeForegroundTurnId = turnId;
this.transitionTurnState("foreground", "rewind command");
void this.executeRewindTurn(turnId, slashCommand);
return { turnId };
}
if (this.autonomousTurn) {
@@ -1355,23 +1403,14 @@ class ClaudeAgentSession implements AgentSession {
}
const sdkMessage = this.toSdkUserMessage(prompt);
const queue = new Pushable<AgentStreamEvent>();
const foregroundTurn: ForegroundTurnState = {
id: this.createTurnId("foreground"),
queue,
hasVisibleActivity: false,
};
this.activeForegroundTurn = foregroundTurn;
this.transitionTurnState("foreground", "foreground stream started");
this.lastForegroundPromptText = this.extractPromptText(prompt);
const turnId = this.createTurnId("foreground");
this.activeForegroundTurnId = turnId;
this.foregroundHasVisibleActivity = false;
this.transitionTurnState("foreground", "foreground turn started");
this.clearRecentStderr();
queue.push({ type: "turn_started", provider: "claude" });
let finishedNaturally = false;
let cancelIssued = false;
let queueDrainedWithoutTerminal = false;
const turnPromise = Promise.resolve();
this.activeTurnPromise = turnPromise;
const requestCancel = () => {
if (cancelIssued) {
return;
@@ -1392,6 +1431,8 @@ class ClaudeAgentSession implements AgentSession {
};
this.cancelCurrentTurn = requestCancel;
this.notifySubscribers({ type: "turn_started", provider: "claude" });
try {
await this.ensureQuery();
if (!this.input) {
@@ -1403,40 +1444,16 @@ class ClaudeAgentSession implements AgentSession {
this.finishForegroundTurn(
this.buildTurnFailedEvent(error instanceof Error ? error.message : "Claude stream failed"),
);
finishedNaturally = true;
}
try {
for await (const event of queue) {
const isTerminalEvent =
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled";
if (isTerminalEvent) {
finishedNaturally = true;
}
yield event;
if (isTerminalEvent) {
break;
}
}
if (!finishedNaturally && !cancelIssued) {
queueDrainedWithoutTerminal = true;
}
} finally {
if (!finishedNaturally && !cancelIssued && !queueDrainedWithoutTerminal) {
requestCancel();
}
if (this.activeForegroundTurn === foregroundTurn) {
this.activeForegroundTurn = null;
}
if (this.cancelCurrentTurn === requestCancel) {
this.cancelCurrentTurn = null;
}
if (this.activeTurnPromise === turnPromise) {
this.activeTurnPromise = null;
}
}
return { turnId };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
async interrupt(): Promise<void> {
@@ -1464,25 +1481,6 @@ class ClaudeAgentSession implements AgentSession {
}
}
async *streamLiveEvents(): AsyncGenerator<AgentStreamEvent> {
if (this.claudeSessionId) {
this.startQueryPump();
}
this.liveEventSubscriberCount += 1;
this.startLiveHistoryPolling();
try {
for await (const event of this.liveEventQueue) {
yield event;
}
} finally {
this.liveEventSubscriberCount = Math.max(0, this.liveEventSubscriberCount - 1);
if (this.liveEventSubscriberCount === 0) {
this.stopLiveHistoryPolling();
}
}
}
async getAvailableModes(): Promise<AgentMode[]> {
return this.availableModes;
}
@@ -1620,22 +1618,19 @@ class ClaudeAgentSession implements AgentSession {
turnState: this.turnState,
hasQuery: Boolean(this.query),
hasInput: Boolean(this.input),
hasActiveForegroundTurn: Boolean(this.activeForegroundTurn),
hasActiveForegroundTurnId: Boolean(this.activeForegroundTurnId),
},
"Claude session close: start",
);
this.closed = true;
this.rejectAllPendingPermissions(new Error("Claude session closed"));
this.cancelCurrentTurn?.();
this.activeForegroundTurn?.queue.end();
this.activeForegroundTurn = null;
this.subscribers.clear();
this.activeForegroundTurnId = null;
this.autonomousTurn = null;
this.cancelCurrentTurn = null;
this.turnState = "idle";
this.liveEventQueue.end();
this.activeTurnPromise = null;
this.sidechainTracker.clear();
this.stopLiveHistoryPolling();
this.input?.end();
this.query?.close?.();
await this.awaitWithTimeout(this.query?.interrupt?.(), "close query interrupt");
@@ -1697,41 +1692,6 @@ class ClaudeAgentSession implements AgentSession {
: { commandName, rawInput: trimmed };
}
private async *streamRewindCommand(
invocation: SlashCommandInvocation,
): AsyncGenerator<AgentStreamEvent> {
yield { type: "turn_started", provider: "claude" };
try {
const rewindAttempt = await this.attemptRewind(invocation.args);
if (!rewindAttempt.messageId || !rewindAttempt.result) {
yield {
type: "turn_failed",
provider: "claude",
error:
rewindAttempt.error ??
"No prior user message available to rewind. Use /rewind <user_message_uuid>.",
};
return;
}
yield {
type: "timeline",
provider: "claude",
item: {
type: "assistant_message",
text: this.buildRewindSuccessMessage(rewindAttempt.messageId, rewindAttempt.result),
},
};
yield { type: "turn_completed", provider: "claude" };
} catch (error) {
yield {
type: "turn_failed",
provider: "claude",
error: error instanceof Error ? error.message : "Failed to rewind tracked files",
};
}
}
private buildRewindSuccessMessage(
targetUserMessageId: string,
rewindResult: { filesChanged?: string[]; insertions?: number; deletions?: number },
@@ -1915,11 +1875,11 @@ class ClaudeAgentSession implements AgentSession {
this.queryRestartNeeded = false;
}
const input = new Pushable<SDKUserMessage>();
const input = createAsyncMessageInput<SDKUserMessage>();
const options = this.buildOptions();
this.logger.debug({ options: summarizeClaudeOptionsForLog(options) }, "claude query");
this.input = input;
this.query = this.queryFactory({ prompt: input, options });
this.query = this.queryFactory({ prompt: input.iterable, options });
// Do not kick off background control-plane queries here. Methods like
// supportedCommands()/setPermissionMode() may execute immediately after
// ensureQuery() (for listCommands()/setMode()), and sharing the same query
@@ -2083,7 +2043,7 @@ class ClaudeAgentSession implements AgentSession {
}
private syncTurnState(reason: string): void {
if (this.activeForegroundTurn) {
if (this.activeForegroundTurnId) {
this.transitionTurnState("foreground", reason);
return;
}
@@ -2139,6 +2099,51 @@ class ClaudeAgentSession implements AgentSession {
);
}
private extractPromptText(prompt: AgentPromptInput): string | null {
if (typeof prompt === "string") {
return prompt;
}
const textParts = prompt
.filter((block): block is { type: "text"; text: string } => block.type === "text")
.map((block) => block.text);
return textParts.length > 0 ? textParts.join("\n") : null;
}
private async executeRewindTurn(
_turnId: string,
invocation: SlashCommandInvocation,
): Promise<void> {
this.notifySubscribers({ type: "turn_started", provider: "claude" });
try {
const rewindAttempt = await this.attemptRewind(invocation.args);
if (!rewindAttempt.messageId || !rewindAttempt.result) {
this.finishForegroundTurn({
type: "turn_failed",
provider: "claude",
error:
rewindAttempt.error ??
"No prior user message available to rewind. Use /rewind <user_message_uuid>.",
});
return;
}
this.notifySubscribers({
type: "timeline",
provider: "claude",
item: {
type: "assistant_message",
text: this.buildRewindSuccessMessage(rewindAttempt.messageId, rewindAttempt.result),
},
});
this.finishForegroundTurn({ type: "turn_completed", provider: "claude" });
} catch (error) {
this.finishForegroundTurn({
type: "turn_failed",
provider: "claude",
error: error instanceof Error ? error.message : "Failed to rewind tracked files",
});
}
}
private shouldRecoverInterruptedQueryAbort(
error: unknown,
consecutiveRecoveries: number,
@@ -2161,43 +2166,30 @@ class ClaudeAgentSession implements AgentSession {
if (event.type === "turn_failed" || event.type === "turn_canceled") {
this.flushPendingToolCalls();
}
this.dispatchForegroundEvents([event]);
}
private dispatchForegroundEvents(events: AgentStreamEvent[]): void {
const foregroundTurn = this.activeForegroundTurn;
if (!foregroundTurn) {
this.dispatchLiveEvents(events);
return;
}
let terminalSeen = false;
for (const event of events) {
foregroundTurn.queue.push(event);
terminalSeen ||= this.isTerminalTurnEvent(event);
}
if (!terminalSeen) {
return;
}
foregroundTurn.queue.end();
if (this.activeForegroundTurn === foregroundTurn) {
this.activeForegroundTurn = null;
}
this.notifySubscribers(event);
this.activeForegroundTurnId = null;
this.lastForegroundPromptText = null;
this.cancelCurrentTurn = null;
this.syncTurnState("foreground turn terminal");
}
private dispatchLiveEvents(events: AgentStreamEvent[]): void {
private dispatchEvents(events: AgentStreamEvent[]): void {
let terminalSeen = false;
for (const event of events) {
this.liveEventQueue.push(event);
this.notifySubscribers(event);
terminalSeen ||= this.isTerminalTurnEvent(event);
}
if (terminalSeen && this.autonomousTurn) {
this.autonomousTurn = null;
this.syncTurnState("autonomous turn terminal");
if (terminalSeen) {
if (this.activeForegroundTurnId) {
this.activeForegroundTurnId = null;
this.lastForegroundPromptText = null;
this.cancelCurrentTurn = null;
this.syncTurnState("foreground turn terminal");
} else if (this.autonomousTurn) {
this.autonomousTurn = null;
this.syncTurnState("autonomous turn terminal");
}
}
}
@@ -2208,7 +2200,7 @@ class ClaudeAgentSession implements AgentSession {
this.autonomousTurn = {
id: this.createTurnId("autonomous"),
};
this.liveEventQueue.push({ type: "turn_started", provider: "claude" });
this.notifySubscribers({ type: "turn_started", provider: "claude" });
this.syncTurnState("autonomous turn started");
}
@@ -2216,8 +2208,8 @@ class ClaudeAgentSession implements AgentSession {
if (!this.autonomousTurn) {
return;
}
this.notifySubscribers({ type: "turn_completed", provider: "claude" });
this.autonomousTurn = null;
this.liveEventQueue.push({ type: "turn_completed", provider: "claude" });
this.syncTurnState("autonomous turn completed");
}
@@ -2226,25 +2218,24 @@ class ClaudeAgentSession implements AgentSession {
return;
}
this.flushPendingToolCalls();
this.autonomousTurn = null;
this.liveEventQueue.push({
this.notifySubscribers({
type: "turn_canceled",
provider: "claude",
reason,
});
this.autonomousTurn = null;
this.syncTurnState("autonomous turn canceled");
}
private failActiveTurns(errorMessage: string): void {
const failure = this.buildTurnFailedEvent(errorMessage);
if (this.activeForegroundTurn) {
this.flushPendingToolCalls();
this.dispatchForegroundEvents([failure]);
this.flushPendingToolCalls();
if (this.activeForegroundTurnId) {
this.finishForegroundTurn(failure);
return;
}
if (this.autonomousTurn) {
this.flushPendingToolCalls();
this.dispatchLiveEvents([failure]);
this.dispatchEvents([failure]);
}
}
@@ -2280,6 +2271,15 @@ class ClaudeAgentSession implements AgentSession {
while (!this.closed && this.query === activeQuery) {
try {
for await (const message of activeQuery) {
this.logger.trace(
{
claudeSessionId: this.claudeSessionId,
messageType: message.type,
messageSubtype: "subtype" in message ? message.subtype : undefined,
messageUuid: "uuid" in message ? message.uuid : undefined,
},
"Claude query pump: raw SDK message",
);
consecutiveInterruptAbortRecoveries = 0;
if (await this.handleMissingResumedConversation(message, activeQuery)) {
return;
@@ -2318,31 +2318,30 @@ class ClaudeAgentSession implements AgentSession {
}
private routeSdkMessageFromPump(message: SDKMessage): void {
const routeToForeground = Boolean(this.activeForegroundTurn);
const isForeground = Boolean(this.activeForegroundTurnId);
const assistantishMessage =
message.type === "assistant" ||
message.type === "stream_event" ||
message.type === "tool_progress" ||
(message.type === "system" && message.subtype === "task_notification");
if (!routeToForeground && assistantishMessage) {
if (!isForeground && assistantishMessage) {
this.startAutonomousTurn();
}
if (!routeToForeground && !this.autonomousTurn && message.type === "result") {
if (!isForeground && !this.autonomousTurn && message.type === "result") {
return;
}
const turnId = this.activeForegroundTurn?.id ?? this.autonomousTurn?.id ?? null;
const turnId = this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? null;
const identifiers = readEventIdentifiers(message);
this.logger.trace(
{
claudeSessionId: this.claudeSessionId,
messageType: message.type,
routedTo: routeToForeground ? "foreground_queue" : "live_queue",
turnId,
},
"Claude query pump routed SDK message",
"Claude query pump: SDK message",
);
const messageEvents = this.translateMessageToEvents(message, {
@@ -2363,7 +2362,23 @@ class ClaudeAgentSession implements AgentSession {
provider: "claude",
}) satisfies AgentStreamEvent,
);
const events = [...messageEvents, ...assistantTimelineEvents];
// User message dedup: suppress echoed user messages that match the foreground prompt
const filteredMessageEvents = messageEvents.filter((event) => {
if (
event.type === "timeline" &&
event.item.type === "user_message" &&
this.activeForegroundTurnId &&
this.lastForegroundPromptText
) {
if (event.item.text.trim() === this.lastForegroundPromptText.trim()) {
return false;
}
}
return true;
});
const events = [...filteredMessageEvents, ...assistantTimelineEvents];
if (events.length === 0) {
return;
@@ -2373,14 +2388,14 @@ class ClaudeAgentSession implements AgentSession {
this.pendingInterruptAbort &&
message.type === "result" &&
events.some((event) => event.type === "turn_completed" || event.type === "turn_failed") &&
(!this.activeForegroundTurn || !this.activeForegroundTurn.hasVisibleActivity)
(!this.activeForegroundTurnId || !this.foregroundHasVisibleActivity)
) {
this.pendingInterruptAbort = false;
this.logger.debug("Suppressing stale Claude interrupt terminal result");
return;
}
if (
this.activeForegroundTurn &&
this.activeForegroundTurnId &&
events.some(
(event) =>
event.type === "timeline" ||
@@ -2388,15 +2403,11 @@ class ClaudeAgentSession implements AgentSession {
event.type === "permission_resolved",
)
) {
this.activeForegroundTurn.hasVisibleActivity = true;
this.foregroundHasVisibleActivity = true;
this.pendingInterruptAbort = false;
}
if (routeToForeground) {
this.dispatchForegroundEvents(events);
return;
}
this.dispatchLiveEvents(events);
this.dispatchEvents(events);
}
private async handleMissingResumedConversation(
@@ -2430,13 +2441,10 @@ class ClaudeAgentSession implements AgentSession {
this.persistence = null;
this.persistedHistory = [];
this.historyPending = false;
this.historyOffsetSessionId = null;
this.historyReadOffsetBytes = 0;
this.historyLineFragment = "";
this.cachedRuntimeInfo = null;
this.queryRestartNeeded = false;
this.autonomousTurn = null;
this.activeForegroundTurn = null;
this.activeForegroundTurnId = null;
this.syncTurnState("missing resumed conversation");
return true;
}
@@ -2855,12 +2863,19 @@ class ClaudeAgentSession implements AgentSession {
}
private pushEvent(event: AgentStreamEvent) {
const foregroundTurn = this.activeForegroundTurn;
if (foregroundTurn) {
foregroundTurn.queue.push(event);
return;
this.notifySubscribers(event);
}
private notifySubscribers(event: AgentStreamEvent): void {
const turnId = this.activeForegroundTurnId ?? this.autonomousTurn?.id;
const tagged = turnId ? { ...event, turnId } : event;
for (const callback of this.subscribers) {
try {
callback(tagged);
} catch (error) {
this.logger.warn({ err: error }, "Subscriber callback threw");
}
}
this.liveEventQueue.push(event);
}
private normalizePermissionUpdates(
@@ -2881,165 +2896,57 @@ class ClaudeAgentSession implements AgentSession {
}
}
private loadPersistedHistory(sessionId: string, options?: { dispatchLive?: boolean }) {
private loadPersistedHistory(sessionId: string): void {
try {
const historyPath = this.resolveHistoryPath(sessionId);
if (!historyPath || !fs.existsSync(historyPath)) {
return;
}
if (this.historyOffsetSessionId !== sessionId) {
this.historyOffsetSessionId = sessionId;
this.historyReadOffsetBytes = 0;
this.historyLineFragment = "";
}
const content = fs.readFileSync(historyPath);
if (content.byteLength < this.historyReadOffsetBytes) {
this.historyReadOffsetBytes = 0;
this.historyLineFragment = "";
}
if (content.byteLength === this.historyReadOffsetBytes) {
return;
}
const unreadChunk = content.subarray(this.historyReadOffsetBytes).toString("utf8");
this.historyReadOffsetBytes = content.byteLength;
this.ingestPersistedHistoryChunk(unreadChunk, {
dispatchLive: options?.dispatchLive ?? false,
});
this.ingestPersistedHistory(fs.readFileSync(historyPath, "utf8"));
} catch (error) {
// ignore history load failures
}
}
private startLiveHistoryPolling(): void {
if (this.liveHistoryPollTimer || !this.claudeSessionId) {
return;
}
this.liveHistoryPollTimer = setInterval(() => {
if (!this.claudeSessionId || this.closed) {
this.stopLiveHistoryPolling();
return;
}
this.loadPersistedHistory(this.claudeSessionId, { dispatchLive: true });
}, 200);
}
private stopLiveHistoryPolling(): void {
if (!this.liveHistoryPollTimer) {
return;
}
clearInterval(this.liveHistoryPollTimer);
this.liveHistoryPollTimer = null;
}
private ingestPersistedHistoryChunk(chunk: string, options: { dispatchLive: boolean }): void {
if (!chunk) {
private ingestPersistedHistory(content: string): void {
if (!content) {
return;
}
const combined = `${this.historyLineFragment}${chunk}`;
this.historyLineFragment = "";
const lines = combined.split(/\r?\n/);
const trailing = lines.pop() ?? "";
const timeline: AgentTimelineItem[] = [];
for (const line of lines) {
this.ingestPersistedHistoryLine(line, {
dispatchLive: options.dispatchLive,
timeline,
});
for (const line of content.split(/\r?\n/)) {
this.ingestPersistedHistoryLine(line, timeline);
}
if (trailing.trim().length > 0) {
const handled = this.ingestPersistedHistoryLine(trailing, {
dispatchLive: options.dispatchLive,
timeline,
});
if (!handled) {
this.historyLineFragment = trailing;
}
}
if (!options.dispatchLive && timeline.length > 0) {
if (timeline.length > 0) {
this.persistedHistory = [...this.persistedHistory, ...timeline];
this.historyPending = true;
}
}
private ingestPersistedHistoryLine(
line: string,
options: {
dispatchLive: boolean;
timeline: AgentTimelineItem[];
},
): boolean {
private ingestPersistedHistoryLine(line: string, timeline: AgentTimelineItem[]): void {
const trimmed = line.trim();
if (!trimmed) {
return true;
return;
}
let entry: Record<string, unknown>;
try {
entry = JSON.parse(trimmed) as Record<string, unknown>;
} catch {
return false;
return;
}
if (entry.isSidechain) {
return true;
return;
}
if (entry.type === "user" && typeof entry.uuid === "string") {
this.rememberUserMessageId(entry.uuid);
}
if (options.dispatchLive) {
this.dispatchPersistedHistoryEntry(entry);
return true;
}
const items = this.convertHistoryEntry(entry);
if (items.length > 0) {
options.timeline.push(...items);
}
return true;
}
private dispatchPersistedHistoryEntry(entry: Record<string, unknown>): void {
const liveMessage = this.normalizePersistedHistoryEntryToLiveMessage(entry);
if (liveMessage) {
this.routeSdkMessageFromPump(liveMessage);
return;
}
const items = this.convertHistoryEntry(entry);
for (const item of items) {
this.pushEvent({
type: "timeline",
item,
provider: "claude",
});
}
}
private normalizePersistedHistoryEntryToLiveMessage(
entry: Record<string, unknown>,
): SDKMessage | null {
const taskNotificationMessage = coerceTaskNotificationHistoryRecordToSystemMessage(entry);
if (taskNotificationMessage) {
return taskNotificationMessage as unknown as SDKMessage;
}
const type = readTrimmedString(entry.type);
switch (type) {
case "assistant":
case "result":
case "stream_event":
case "system":
case "tool_progress":
case "user":
return entry as unknown as SDKMessage;
default:
return null;
timeline.push(...items);
}
}
@@ -3759,49 +3666,51 @@ export function convertClaudeHistoryEntry(
return timeline;
}
class Pushable<T> implements AsyncIterable<T> {
private queue: T[] = [];
private resolvers: Array<(value: IteratorResult<T, void>) => void> = [];
private closed = false;
function createAsyncMessageInput<T>(): AsyncMessageInput<T> {
const queue: T[] = [];
const resolvers: Array<(value: IteratorResult<T, void>) => void> = [];
let closed = false;
push(item: T) {
if (this.closed) {
return;
}
if (this.resolvers.length > 0) {
const resolve = this.resolvers.shift()!;
resolve({ value: item, done: false });
} else {
this.queue.push(item);
}
}
end() {
this.closed = true;
while (this.resolvers.length > 0) {
const resolve = this.resolvers.shift()!;
resolve({ value: undefined, done: true });
}
}
[Symbol.asyncIterator](): AsyncIterator<T, void> {
return {
next: (): Promise<IteratorResult<T, void>> => {
if (this.queue.length > 0) {
const value = this.queue.shift();
if (value !== undefined) {
return Promise.resolve({ value, done: false });
}
}
if (this.closed) {
return Promise.resolve({ value: undefined, done: true });
}
return new Promise<IteratorResult<T, void>>((resolve) => {
this.resolvers.push(resolve);
});
return {
push(item: T) {
if (closed) {
return;
}
const resolve = resolvers.shift();
if (resolve) {
resolve({ value: item, done: false });
return;
}
queue.push(item);
},
end() {
closed = true;
while (resolvers.length > 0) {
const resolve = resolvers.shift();
resolve?.({ value: undefined, done: true });
}
},
iterable: {
[Symbol.asyncIterator](): AsyncIterator<T, void> {
return {
next: (): Promise<IteratorResult<T, void>> => {
if (queue.length > 0) {
const value = queue.shift();
if (value !== undefined) {
return Promise.resolve({ value, done: false });
}
}
if (closed) {
return Promise.resolve({ value: undefined, done: true });
}
return new Promise<IteratorResult<T, void>>((resolve) => {
resolvers.push(resolve);
});
},
};
},
};
}
},
};
}
type ClaudeSessionCandidate = {
@@ -1,10 +1,11 @@
import { appendFileSync, mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs";
import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs";
import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
import type { AgentPersistenceHandle, AgentStreamEvent } from "../agent-sdk-types.js";
const sdkMocks = vi.hoisted(() => ({
@@ -19,8 +20,6 @@ vi.mock("@anthropic-ai/claude-agent-sdk", () => ({
const LIVE_REPLY_MARKER = "LIVE_ONLY_REPLY_MARKER";
const HISTORY_USER_MARKER = "HISTORY_ONLY_USER_MARKER";
const HISTORY_ASSISTANT_MARKER = "HISTORY_ONLY_ASSISTANT_MARKER";
const APPENDED_TASK_NOTIFICATION_MARKER = "Appended background task completed";
const APPENDED_ASSISTANT_MARKER = "APPENDED_BACKGROUND_ASSISTANT_MARKER";
function buildSdkQueryMock() {
const events = [
@@ -73,23 +72,6 @@ function buildSdkQueryMock() {
};
}
function buildIdleSdkQueryMock() {
return {
next: vi.fn(async () => ({ done: true, value: undefined })),
interrupt: vi.fn(async () => undefined),
return: vi.fn(async () => undefined),
close: vi.fn(() => undefined),
setPermissionMode: vi.fn(async () => undefined),
setModel: vi.fn(async () => undefined),
supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]),
supportedCommands: vi.fn(async () => []),
rewindFiles: vi.fn(async () => ({ canRewind: true })),
[Symbol.asyncIterator]() {
return this;
},
};
}
function collectTimelineText(events: AgentStreamEvent[]): string {
const chunks: string[] = [];
for (const event of events) {
@@ -106,28 +88,6 @@ function collectTimelineText(events: AgentStreamEvent[]): string {
return chunks.join("\n");
}
async function readNextEvent(
iterator: AsyncIterator<AgentStreamEvent>,
timeoutMs: number,
): Promise<AgentStreamEvent> {
const outcome = await Promise.race([
iterator.next().then((result) => ({ kind: "result" as const, result })),
new Promise<{ kind: "timeout" }>((resolve) => {
setTimeout(() => resolve({ kind: "timeout" }), timeoutMs);
}),
]);
if (outcome.kind === "timeout") {
throw new Error("Timed out waiting for live event");
}
if (outcome.result.done) {
throw new Error("Live event stream ended before appended transcript arrived");
}
return outcome.result.value;
}
describe("ClaudeAgentSession history replay regression", () => {
let tempRoot: string;
let cwd: string;
@@ -208,7 +168,7 @@ describe("ClaudeAgentSession history replay regression", () => {
const events: AgentStreamEvent[] = [];
try {
for await (const event of session.stream("Say hello")) {
for await (const event of streamSession(session, "Say hello")) {
events.push(event);
if (
event.type === "turn_completed" ||
@@ -257,112 +217,6 @@ describe("ClaudeAgentSession history replay regression", () => {
expect(timelineText).toContain(HISTORY_ASSISTANT_MARKER);
});
test("emits appended transcript lines through streamLiveEvents after history was primed", async () => {
sdkMocks.query.mockImplementation(() => {
const mock = buildIdleSdkQueryMock();
sdkMocks.lastQuery = mock;
return mock;
});
const logger = createTestLogger();
const client = new ClaudeAgentClient({ logger });
const handle: AgentPersistenceHandle = {
provider: "claude",
sessionId: "history-session",
nativeHandle: "history-session",
metadata: {
provider: "claude",
cwd,
},
};
const sanitized = cwd.replace(/[\\/\.]/g, "-").replace(/_/g, "-");
const historyPath = path.join(configDir, "projects", sanitized, "history-session.jsonl");
const session = await client.resumeSession(handle, { cwd });
try {
for await (const _event of session.streamHistory()) {
// Prime existing persisted history the same way agent-manager does.
}
const liveEvents = session.streamLiveEvents();
const iterator = liveEvents[Symbol.asyncIterator]();
appendFileSync(
historyPath,
`\n${JSON.stringify({
type: "queue-operation",
operation: "enqueue",
uuid: "appended-task-note-1",
content: [
"<task-notification>",
"<task-id>appended-bg-1</task-id>",
"<status>completed</status>",
`<summary>${APPENDED_TASK_NOTIFICATION_MARKER}</summary>`,
"<output-file>/tmp/appended-bg-1.txt</output-file>",
"</task-notification>",
].join("\n"),
})}\n${JSON.stringify({
type: "assistant",
sessionId: "history-session",
cwd,
message: {
role: "assistant",
content: APPENDED_ASSISTANT_MARKER,
},
})}`,
"utf8",
);
const appendedEvents: AgentStreamEvent[] = [];
for (let attempt = 0; attempt < 4; attempt += 1) {
appendedEvents.push(await readNextEvent(iterator, 1_500));
const sawTaskNotification = appendedEvents.some(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "tool_call" &&
event.item.name === "task_notification",
);
const sawAssistant = appendedEvents.some(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "assistant_message" &&
event.item.text.includes(APPENDED_ASSISTANT_MARKER),
);
if (sawTaskNotification && sawAssistant) {
break;
}
}
const timelineText = collectTimelineText(appendedEvents);
const taskNotificationEvent = appendedEvents.find(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "tool_call" &&
event.item.name === "task_notification",
);
const turnStartedEvent = appendedEvents.find(
(event): event is Extract<AgentStreamEvent, { type: "turn_started" }> =>
event.type === "turn_started",
);
expect(taskNotificationEvent).toBeTruthy();
expect(turnStartedEvent).toBeTruthy();
expect(timelineText).toContain(APPENDED_ASSISTANT_MARKER);
expect(taskNotificationEvent?.item.metadata).toMatchObject({
taskId: "appended-bg-1",
status: "completed",
outputFile: "/tmp/appended-bg-1.txt",
});
expect(taskNotificationEvent?.item.detail).toMatchObject({
type: "plain_text",
label: APPENDED_TASK_NOTIFICATION_MARKER,
});
} finally {
await session.close();
}
});
test("listCommands includes rewind command", async () => {
const logger = createTestLogger();
const client = new ClaudeAgentClient({ logger });
@@ -402,7 +256,7 @@ describe("ClaudeAgentSession history replay regression", () => {
const events: AgentStreamEvent[] = [];
try {
for await (const event of session.stream("/rewind")) {
for await (const event of streamSession(session, "/rewind")) {
events.push(event);
if (
event.type === "turn_completed" ||
@@ -435,50 +435,6 @@ function toCodexMcpConfig(config: McpServerConfig): CodexMcpServerConfig {
};
}
}
class Pushable<T> implements AsyncIterable<T> {
private queue: T[] = [];
private resolvers: ((value: IteratorResult<T>) => void)[] = [];
private closed = false;
push(item: T) {
if (this.closed) {
return;
}
if (this.resolvers.length > 0) {
const resolve = this.resolvers.shift()!;
resolve({ value: item, done: false });
} else {
this.queue.push(item);
}
}
end() {
this.closed = true;
while (this.resolvers.length > 0) {
const resolve = this.resolvers.shift()!;
resolve({ value: undefined, done: true });
}
}
[Symbol.asyncIterator](): AsyncIterator<T, void> {
return {
next: (): Promise<IteratorResult<T, void>> => {
if (this.queue.length > 0) {
const value = this.queue.shift()!;
return Promise.resolve({ value, done: false });
}
if (this.closed) {
return Promise.resolve({ value: undefined, done: true });
}
return new Promise<IteratorResult<T, void>>((resolve) => {
this.resolvers.push(resolve);
});
},
};
}
}
type JsonRpcRequest = {
id: number;
method: string;
@@ -2062,7 +2018,9 @@ class CodexAppServerAgentSession implements AgentSession {
private currentThreadId: string | null = null;
private currentTurnId: string | null = null;
private client: CodexAppServerClient | null = null;
private eventQueue: Pushable<AgentStreamEvent> | null = null;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private nextTurnOrdinal = 0;
private activeForegroundTurnId: string | null = null;
private cachedRuntimeInfo: AgentRuntimeInfo | null = null;
private historyPending = false;
private persistedHistory: AgentTimelineItem[] = [];
@@ -2397,37 +2355,70 @@ class CodexAppServerAgentSession implements AgentSession {
}
async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
const slashCommand = await this.resolveSlashCommandInvocation(prompt);
if (slashCommand) {
const commandInput = await this.buildCommandPromptInput(
slashCommand.commandName,
slashCommand.args,
);
return this.runInternal(commandInput, options);
}
return this.runInternal(prompt, options);
}
private async runInternal(
prompt: AgentPromptInput,
options?: AgentRunOptions,
): Promise<AgentRunResult> {
const events = this.streamInternal(prompt, options);
const timeline: AgentTimelineItem[] = [];
let finalText = "";
let usage: AgentUsage | undefined;
let turnId: string | null = null;
const bufferedEvents: AgentStreamEvent[] = [];
let settled = false;
let resolveCompletion!: () => void;
let rejectCompletion!: (error: Error) => void;
for await (const event of events) {
const processEvent = (event: AgentStreamEvent) => {
if (settled) {
return;
}
const eventTurnId = (event as { turnId?: string }).turnId;
if (turnId && eventTurnId && eventTurnId !== turnId) {
return;
}
if (event.type === "timeline") {
timeline.push(event.item);
if (event.item.type === "assistant_message") {
finalText = event.item.text;
}
} else if (event.type === "turn_completed") {
usage = event.usage;
} else if (event.type === "turn_failed") {
throw new Error(event.error);
return;
}
if (event.type === "turn_completed") {
usage = event.usage;
settled = true;
resolveCompletion();
return;
}
if (event.type === "turn_failed") {
settled = true;
rejectCompletion(new Error(event.error));
return;
}
if (event.type === "turn_canceled") {
settled = true;
resolveCompletion();
}
};
const completion = new Promise<void>((resolve, reject) => {
resolveCompletion = resolve;
rejectCompletion = reject;
});
const unsubscribe = this.subscribe((event) => {
if (!turnId) {
bufferedEvents.push(event);
return;
}
processEvent(event);
});
try {
const result = await this.startTurn(prompt, options);
turnId = result.turnId;
for (const event of bufferedEvents) {
processEvent(event);
}
if (!settled) {
await completion;
}
} finally {
unsubscribe();
}
const info = await this.getRuntimeInfo();
@@ -2439,114 +2430,92 @@ class CodexAppServerAgentSession implements AgentSession {
};
}
async *stream(
async startTurn(
prompt: AgentPromptInput,
options?: AgentRunOptions,
): AsyncGenerator<AgentStreamEvent> {
const slashCommand = await this.resolveSlashCommandInvocation(prompt);
if (slashCommand) {
const commandInput = await this.buildCommandPromptInput(
slashCommand.commandName,
slashCommand.args,
);
yield* this.streamInternal(commandInput, options);
return;
): Promise<{ turnId: string }> {
if (this.activeForegroundTurnId) {
throw new Error("A foreground turn is already active");
}
yield* this.streamInternal(prompt, options);
}
private async *streamInternal(
prompt: AgentPromptInput,
options?: AgentRunOptions,
): AsyncGenerator<AgentStreamEvent> {
await this.connect();
if (!this.client) return;
if (!this.client) {
throw new Error("Codex client not initialized");
}
const queue = new Pushable<AgentStreamEvent>();
this.eventQueue = queue;
const slashCommand = await this.resolveSlashCommandInvocation(prompt);
const effectivePrompt = slashCommand
? await this.buildCommandPromptInput(slashCommand.commandName, slashCommand.args)
: prompt;
if (this.currentThreadId) {
await this.ensureThreadLoaded();
} else {
await this.ensureThread();
}
const input = await this.buildUserInput(effectivePrompt);
const preset = MODE_PRESETS[this.currentMode] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID];
const approvalPolicy = this.config.approvalPolicy ?? preset.approvalPolicy;
const sandboxPolicyType = this.config.sandboxMode ?? preset.sandbox;
const params: Record<string, unknown> = {
threadId: this.currentThreadId,
input,
approvalPolicy,
sandboxPolicy: toSandboxPolicy(
sandboxPolicyType,
typeof this.config.networkAccess === "boolean"
? this.config.networkAccess
: preset.networkAccess,
),
};
if (this.config.model) {
params.model = this.config.model;
}
const thinkingOptionId = normalizeCodexThinkingOptionId(this.config.thinkingOptionId);
if (thinkingOptionId) {
params.effort = thinkingOptionId;
}
if (this.resolvedCollaborationMode) {
params.collaborationMode = {
mode: this.resolvedCollaborationMode.mode,
settings: this.resolvedCollaborationMode.settings,
};
}
if (this.config.cwd) {
params.cwd = this.config.cwd;
}
if (options?.outputSchema) {
params.outputSchema = options.outputSchema;
}
if (this.config.systemPrompt?.trim()) {
params.developerInstructions = this.config.systemPrompt.trim();
}
const codexConfig = this.buildCodexInnerConfig();
if (codexConfig) {
params.config = codexConfig;
}
const turnId = this.createTurnId();
this.activeForegroundTurnId = turnId;
try {
if (this.currentThreadId) {
await this.ensureThreadLoaded();
} else {
await this.ensureThread();
}
const input = await this.buildUserInput(prompt);
const preset = MODE_PRESETS[this.currentMode] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID];
const approvalPolicy = this.config.approvalPolicy ?? preset.approvalPolicy;
const sandboxPolicyType = this.config.sandboxMode ?? preset.sandbox;
const params: Record<string, unknown> = {
threadId: this.currentThreadId,
input,
approvalPolicy,
sandboxPolicy: toSandboxPolicy(
sandboxPolicyType,
typeof this.config.networkAccess === "boolean"
? this.config.networkAccess
: preset.networkAccess,
),
};
if (this.config.model) {
params.model = this.config.model;
}
const thinkingOptionId = normalizeCodexThinkingOptionId(this.config.thinkingOptionId);
if (thinkingOptionId) {
params.effort = thinkingOptionId;
}
if (this.resolvedCollaborationMode) {
params.collaborationMode = {
mode: this.resolvedCollaborationMode.mode,
settings: this.resolvedCollaborationMode.settings,
};
}
if (this.config.cwd) {
params.cwd = this.config.cwd;
}
if (options?.outputSchema) {
params.outputSchema = options.outputSchema;
}
if (this.config.systemPrompt?.trim()) {
params.developerInstructions = this.config.systemPrompt.trim();
}
const codexConfig = this.buildCodexInnerConfig();
if (codexConfig) {
params.config = codexConfig;
}
await this.client.request("turn/start", params, TURN_START_TIMEOUT_MS);
let sawTurnStarted = false;
for await (const event of queue) {
// Drop pre-start timeline noise that can leak from the previous turn.
// Keep permission events, which can legitimately arrive before turn_started.
if (!sawTurnStarted) {
if (event.type === "permission_requested" || event.type === "permission_resolved") {
yield event;
continue;
}
if (event.type === "turn_started") {
sawTurnStarted = true;
} else {
continue;
}
}
yield event;
if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
) {
break;
}
}
} finally {
if (this.eventQueue === queue) {
this.eventQueue = null;
}
} catch (error) {
this.activeForegroundTurnId = null;
throw error;
}
return { turnId };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
@@ -2742,8 +2711,8 @@ class CodexAppServerAgentSession implements AgentSession {
this.pendingPermissionHandlers.clear();
this.pendingPermissions.clear();
this.resolvedPermissionRequests.clear();
this.eventQueue?.end();
this.eventQueue = null;
this.subscribers.clear();
this.activeForegroundTurnId = null;
if (this.client) {
await this.client.dispose();
}
@@ -2860,7 +2829,23 @@ class CodexAppServerAgentSession implements AgentSession {
this.pendingAgentMessages.clear();
}
}
this.eventQueue?.push(event);
this.notifySubscribers(event);
}
private notifySubscribers(event: AgentStreamEvent): void {
const turnId = this.activeForegroundTurnId;
const tagged = turnId ? { ...event, turnId } : event;
for (const callback of this.subscribers) {
try {
callback(tagged);
} catch (error) {
this.logger.warn({ err: error }, "Subscriber callback threw");
}
}
}
private createTurnId(): string {
return `codex-turn-${this.nextTurnOrdinal++}`;
}
private handleNotification(method: string, params: unknown): void {
@@ -2905,6 +2890,7 @@ class CodexAppServerAgentSession implements AgentSession {
usage: this.latestUsage,
});
}
this.activeForegroundTurnId = null;
this.emittedItemStartedIds.clear();
this.emittedItemCompletedIds.clear();
this.emittedExecCommandStartedCallIds.clear();
@@ -13,6 +13,7 @@ import { execFileSync } from "node:child_process";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { OpenCodeAgentClient } from "./opencode-agent.js";
import { streamSession } from "./test-utils/session-stream-adapter.js";
import type {
AgentSessionConfig,
AgentStreamEvent,
@@ -158,7 +159,7 @@ const hasOpenCode = isBinaryInstalled("opencode");
const client = new OpenCodeAgentClient(logger);
const session = await client.createSession(buildConfig(cwd));
const iterator = session.stream("Say hello");
const iterator = streamSession(session, "Say hello");
const turn = await collectTurnEvents(iterator);
// HARD ASSERT: Turn completed successfully
@@ -230,7 +231,8 @@ const hasOpenCode = isBinaryInstalled("opencode");
});
const planTurn = await collectTurnEvents(
planSession.stream(
streamSession(
planSession,
"Create a file named plan-mode-output.txt in the current directory containing exactly hello.",
),
);
@@ -250,7 +252,8 @@ const hasOpenCode = isBinaryInstalled("opencode");
});
const buildTurn = await collectTurnEvents(
buildSession.stream(
streamSession(
buildSession,
"Create a file named build-mode-output.txt in the current directory containing exactly hello.",
),
);
@@ -836,6 +836,9 @@ class OpenCodeAgentSession implements AgentSession {
/** Tracks assistant messages already emitted from structured payloads. */
private emittedStructuredMessageIds = new Set<string>();
private availableModesCache: AgentMode[] | null = null;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private nextTurnOrdinal = 0;
private activeForegroundTurnId: string | null = null;
constructor(config: OpenCodeAgentConfig, client: OpencodeClient, sessionId: string) {
this.config = config;
@@ -872,22 +875,70 @@ class OpenCodeAgentSession implements AgentSession {
}
async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
const events = this.stream(prompt, options);
const timeline: AgentTimelineItem[] = [];
let finalText = "";
let usage: AgentUsage | undefined;
let turnId: string | null = null;
const bufferedEvents: AgentStreamEvent[] = [];
let settled = false;
let resolveCompletion!: () => void;
let rejectCompletion!: (error: Error) => void;
for await (const event of events) {
const processEvent = (event: AgentStreamEvent) => {
if (settled) {
return;
}
const eventTurnId = (event as { turnId?: string }).turnId;
if (turnId && eventTurnId && eventTurnId !== turnId) {
return;
}
if (event.type === "timeline") {
timeline.push(event.item);
if (event.item.type === "assistant_message") {
finalText = event.item.text;
}
} else if (event.type === "turn_completed") {
usage = event.usage;
} else if (event.type === "turn_failed") {
throw new Error(event.error);
return;
}
if (event.type === "turn_completed") {
usage = event.usage;
settled = true;
resolveCompletion();
return;
}
if (event.type === "turn_failed") {
settled = true;
rejectCompletion(new Error(event.error));
return;
}
if (event.type === "turn_canceled") {
settled = true;
resolveCompletion();
}
};
const completion = new Promise<void>((resolve, reject) => {
resolveCompletion = resolve;
rejectCompletion = reject;
});
const unsubscribe = this.subscribe((event) => {
if (!turnId) {
bufferedEvents.push(event);
return;
}
processEvent(event);
});
try {
const result = await this.startTurn(prompt, options);
turnId = result.turnId;
for (const event of bufferedEvents) {
processEvent(event);
}
if (!settled) {
await completion;
}
} finally {
unsubscribe();
}
return {
@@ -898,10 +949,22 @@ class OpenCodeAgentSession implements AgentSession {
};
}
async *stream(
async interrupt(): Promise<void> {
this.abortController?.abort();
await this.client.session.abort({
sessionID: this.sessionId,
directory: this.config.cwd,
});
}
async startTurn(
prompt: AgentPromptInput,
options?: AgentRunOptions,
): AsyncGenerator<AgentStreamEvent> {
): Promise<{ turnId: string }> {
if (this.activeForegroundTurnId) {
throw new Error("A foreground turn is already active");
}
this.abortController = new AbortController();
await this.ensureMcpServersConfigured();
@@ -912,7 +975,6 @@ class OpenCodeAgentSession implements AgentSession {
thinkingOptionId && thinkingOptionId !== "default" ? thinkingOptionId : undefined;
const effectiveMode = normalizeOpenCodeModeId(this.currentMode);
// Send prompt asynchronously
const promptResponse = await this.client.session.promptAsync({
sessionID: this.sessionId,
directory: this.config.cwd,
@@ -932,50 +994,76 @@ class OpenCodeAgentSession implements AgentSession {
});
if (promptResponse.error) {
yield {
const errorMsg = JSON.stringify(promptResponse.error);
this.notifySubscribers({
type: "turn_failed",
provider: "opencode",
error: JSON.stringify(promptResponse.error),
};
return;
error: errorMsg,
});
throw new Error(errorMsg);
}
// Subscribe to events
const turnId = this.createTurnId();
this.activeForegroundTurnId = turnId;
void this.consumeEventStream();
return { turnId };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
private async consumeEventStream(): Promise<void> {
const eventsResult = await this.client.event.subscribe({
directory: this.config.cwd,
});
try {
for await (const event of eventsResult.stream) {
if (this.abortController.signal.aborted) {
if (this.abortController?.signal.aborted) {
break;
}
const translated = this.translateEvent(event);
for (const e of translated) {
yield e;
this.notifySubscribers(e);
if (e.type === "turn_completed" || e.type === "turn_failed") {
this.activeForegroundTurnId = null;
return;
}
}
}
} catch (error) {
if (!this.abortController.signal.aborted) {
yield {
if (!this.abortController?.signal.aborted) {
this.notifySubscribers({
type: "turn_failed",
provider: "opencode",
error: error instanceof Error ? error.message : "Stream error",
};
});
this.activeForegroundTurnId = null;
}
}
}
async interrupt(): Promise<void> {
this.abortController?.abort();
await this.client.session.abort({
sessionID: this.sessionId,
directory: this.config.cwd,
});
private notifySubscribers(event: AgentStreamEvent): void {
const turnId = this.activeForegroundTurnId;
const tagged = turnId ? { ...event, turnId } : event;
for (const callback of this.subscribers) {
try {
callback(tagged);
} catch {
// Subscriber callback error isolation
}
}
}
private createTurnId(): string {
return `opencode-turn-${this.nextTurnOrdinal++}`;
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
@@ -1132,6 +1220,8 @@ class OpenCodeAgentSession implements AgentSession {
async close(): Promise<void> {
this.abortController?.abort();
this.subscribers.clear();
this.activeForegroundTurnId = null;
}
private buildPromptParts(prompt: AgentPromptInput): Array<{ type: "text"; text: string }> {
@@ -0,0 +1,76 @@
import type {
AgentPromptInput,
AgentRunOptions,
AgentSession,
AgentStreamEvent,
} from "../../agent-sdk-types.js";
function isTerminalEvent(event: AgentStreamEvent): boolean {
return (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
);
}
export async function* streamSession(
session: Pick<AgentSession, "startTurn" | "subscribe">,
prompt: AgentPromptInput,
options?: AgentRunOptions,
): AsyncGenerator<AgentStreamEvent> {
const queue: AgentStreamEvent[] = [];
const waiters: Array<() => void> = [];
let turnId: string | null = null;
let closed = false;
const wake = () => {
const waiter = waiters.shift();
waiter?.();
};
const matchesTurn = (event: AgentStreamEvent): boolean => {
const eventTurnId = (event as { turnId?: string }).turnId;
return turnId == null || eventTurnId == null || eventTurnId === turnId;
};
const unsubscribe = session.subscribe((event) => {
if (!matchesTurn(event)) {
return;
}
queue.push(event);
wake();
});
try {
const result = await session.startTurn(prompt, options);
turnId = result.turnId;
for (let idx = queue.length - 1; idx >= 0; idx -= 1) {
if (!matchesTurn(queue[idx]!)) {
queue.splice(idx, 1);
}
}
while (!closed) {
if (queue.length === 0) {
await new Promise<void>((resolve) => {
waiters.push(resolve);
});
continue;
}
const event = queue.shift()!;
yield event;
if (isTerminalEvent(event)) {
return;
}
}
} finally {
closed = true;
unsubscribe();
while (waiters.length > 0) {
const waiter = waiters.shift();
waiter?.();
}
}
}
@@ -20,12 +20,12 @@ const testLogger = {
type ManagedAgentOverrides = Omit<
Partial<ManagedAgent>,
"config" | "pendingPermissions" | "session" | "pendingRun"
"config" | "pendingPermissions" | "session" | "activeForegroundTurnId"
> & {
config?: Partial<AgentSessionConfig>;
pendingPermissions?: Map<string, AgentPermissionRequest>;
session?: AgentSession | null;
pendingRun?: ManagedAgent["pendingRun"];
activeForegroundTurnId?: string | null;
};
function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent {
@@ -42,8 +42,8 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
extra: configOverrides.extra ?? { claude: { tone: "focused" } },
};
const session = lifecycle === "closed" ? null : (overrides.session ?? ({} as AgentSession));
const pendingRun =
overrides.pendingRun ?? (lifecycle === "running" ? (async function* noop() {})() : null);
const activeForegroundTurnId =
overrides.activeForegroundTurnId ?? (lifecycle === "running" ? "test-turn-id" : null);
const agent: ManagedAgent = {
id: overrides.id ?? "agent-1",
@@ -65,7 +65,9 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
availableModes: overrides.availableModes ?? [],
currentModeId: overrides.currentModeId ?? config.modeId ?? null,
pendingPermissions: overrides.pendingPermissions ?? new Map<string, AgentPermissionRequest>(),
pendingRun,
activeForegroundTurnId,
foregroundTurnWaiters: new Set(),
unsubscribeSession: null,
timeline: overrides.timeline ?? [],
persistence: overrides.persistence ?? null,
historyPrimed: overrides.historyPrimed ?? true,
+16 -14
View File
@@ -796,16 +796,17 @@ export class Session {
throw new Error(`Agent ${agentId} not found`);
}
if (snapshot.lifecycle !== "running" && !snapshot.pendingRun) {
const hasInFlightRun = this.agentManager.hasInFlightRun(agentId);
if (!hasInFlightRun) {
this.sessionLogger.trace(
{ agentId, lifecycle: snapshot.lifecycle, pendingRun: Boolean(snapshot.pendingRun) },
{ agentId, lifecycle: snapshot.lifecycle, hasInFlightRun },
"interruptAgentIfRunning: skipping because agent is not running",
);
return;
}
this.sessionLogger.debug(
{ agentId, lifecycle: snapshot.lifecycle, pendingRun: Boolean(snapshot.pendingRun) },
{ agentId, lifecycle: snapshot.lifecycle, hasInFlightRun },
"interruptAgentIfRunning: interrupting",
);
@@ -831,13 +832,7 @@ export class Session {
if (!agentId) {
return false;
}
const snapshot = this.agentManager.getAgent(agentId);
if (!snapshot) {
return false;
}
return snapshot.lifecycle === "running" || Boolean(snapshot.pendingRun);
return this.agentManager.hasInFlightRun(agentId);
}
/**
@@ -858,10 +853,7 @@ export class Session {
);
let iterator: AsyncGenerator<AgentStreamEvent>;
try {
const snapshot = this.agentManager.getAgent(agentId);
const shouldReplace = Boolean(
snapshot && (snapshot.lifecycle === "running" || snapshot.pendingRun),
);
const shouldReplace = this.agentManager.hasInFlightRun(agentId);
iterator = shouldReplace
? this.agentManager.replaceAgentRun(agentId, prompt, runOptions)
: this.agentManager.streamAgent(agentId, prompt, runOptions);
@@ -1462,6 +1454,7 @@ export class Session {
* Main entry point for processing session messages
*/
public async handleMessage(msg: SessionInboundMessage): Promise<void> {
this.sessionLogger.trace({ inbound: msg }, "inbound message");
try {
switch (msg.type) {
case "voice_audio_chunk":
@@ -6589,6 +6582,10 @@ export class Session {
await this.ensureAgentLoaded(agentId);
this.sessionLogger.trace(
{ agentId, messageId: msg.messageId, textPrefix: msg.text.slice(0, 80) },
"send_agent_message_request: recording user message",
);
try {
this.agentManager.recordUserMessage(agentId, msg.text, {
messageId: msg.messageId,
@@ -6602,6 +6599,10 @@ export class Session {
}
const prompt = this.buildAgentPrompt(msg.text, msg.images);
this.sessionLogger.trace(
{ agentId, messageId: msg.messageId },
"send_agent_message_request: starting agent stream",
);
const started = this.startAgentStream(agentId, prompt);
if (!started.ok) {
this.emit({
@@ -7359,6 +7360,7 @@ export class Session {
* Emit a message to the client
*/
private emit(msg: SessionOutboundMessage): void {
this.sessionLogger.trace({ outbound: msg }, "outbound message");
if (
msg.type === "audio_output" &&
(process.env.TTS_DEBUG_AUDIO_DIR || isPaseoDictationDebugEnabled()) &&
@@ -191,6 +191,9 @@ class FakeAgentSession implements AgentSession {
private pendingPermissions: AgentPermissionRequest[] = [];
private permissionGate: Deferred<AgentPermissionResponse> | null = null;
private readonly historyPath: string;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private nextTurnOrdinal = 0;
private activeForegroundTurnId: string | null = null;
constructor(
providerName: string,
@@ -270,205 +273,253 @@ class FakeAgentSession implements AgentSession {
return { sessionId: this.id, finalText: resultText, timeline, usage };
}
async *stream(prompt: AgentPromptInput): AsyncGenerator<AgentStreamEvent> {
// New run => reset interrupt gate.
async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
if (this.activeForegroundTurnId) {
throw new Error("A foreground turn is already active");
}
const turnId = `fake-turn-${this.nextTurnOrdinal++}`;
this.activeForegroundTurnId = turnId;
void this.emitTurnEvents(prompt);
return { turnId };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
private notifySubscribers(event: AgentStreamEvent): void {
const turnId = this.activeForegroundTurnId;
const tagged = turnId ? { ...event, turnId } : event;
for (const callback of this.subscribers) {
try {
callback(tagged);
} catch {
// Error isolation
}
}
}
private async emitTurnEvents(prompt: AgentPromptInput): Promise<void> {
this.interruptSignal = createDeferred<void>();
const slashCommand = await this.resolveSlashCommandInput(prompt);
if (slashCommand) {
const textPrompt = typeof prompt === "string" ? prompt : JSON.stringify(prompt);
try {
if (slashCommand) {
const threadStarted: AgentStreamEvent = {
type: "thread_started",
provider: this.providerName,
sessionId: this.id,
};
await this.appendHistoryEvent(threadStarted);
this.notifySubscribers(threadStarted);
const turnStarted: AgentStreamEvent = {
type: "turn_started",
provider: this.providerName,
};
await this.appendHistoryEvent(turnStarted);
this.notifySubscribers(turnStarted);
const result = await this.runSlashCommand(slashCommand.commandName, slashCommand.args);
for (const item of result.timeline) {
const timelineEvent: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item,
};
await this.appendHistoryEvent(timelineEvent);
this.notifySubscribers(timelineEvent);
}
const completed: AgentStreamEvent = {
type: "turn_completed",
provider: this.providerName,
usage: result.usage ?? { inputTokens: 1, outputTokens: 1 },
};
await this.appendHistoryEvent(completed);
this.notifySubscribers(completed);
return;
}
const markerMatch = /remember (?:this )?(?:marker|string|project name)[^"]*"([^"]+)"/i.exec(
textPrompt,
);
if (markerMatch) {
this.memoryMarker = markerMatch[1] ?? null;
}
const threadStarted: AgentStreamEvent = {
type: "thread_started",
provider: this.providerName,
sessionId: this.id,
};
await this.appendHistoryEvent(threadStarted);
yield threadStarted;
this.notifySubscribers(threadStarted);
const turnStarted: AgentStreamEvent = {
type: "turn_started",
provider: this.providerName,
};
await this.appendHistoryEvent(turnStarted);
yield turnStarted;
this.notifySubscribers(turnStarted);
const result = await this.runSlashCommand(slashCommand.commandName, slashCommand.args);
for (const item of result.timeline) {
const timelineEvent: AgentStreamEvent = {
const tool = buildToolCallForPrompt(this.providerName, textPrompt);
if (tool) {
const needsPermission = this.needsPermissionForTool(tool.name, tool.input ?? {});
const callId = randomUUID();
const toolRunning: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item,
item: {
type: "tool_call",
name: tool.name,
callId,
status: "running",
detail: {
type: "unknown",
input: tool.input ?? null,
output: null,
},
error: null,
},
};
await this.appendHistoryEvent(timelineEvent);
yield timelineEvent;
await this.appendHistoryEvent(toolRunning);
this.notifySubscribers(toolRunning);
if (needsPermission) {
const request: AgentPermissionRequest = {
id: randomUUID(),
provider: this.providerName,
name: tool.name,
kind: "tool",
title: "Permission required",
description: "Test permission request",
input: tool.input ?? {},
};
this.pendingPermissions = [request];
this.permissionGate = createDeferred<AgentPermissionResponse>();
const permissionRequested: AgentStreamEvent = {
type: "permission_requested",
provider: this.providerName,
request,
};
await this.appendHistoryEvent(permissionRequested);
this.notifySubscribers(permissionRequested);
const response = await Promise.race([
this.permissionGate.promise,
this.interruptSignal.promise.then(
() =>
({
behavior: "deny",
interrupt: true,
message: "Interrupted",
}) satisfies AgentPermissionResponse,
),
]);
this.pendingPermissions = [];
this.permissionGate = null;
const permissionResolved: AgentStreamEvent = {
type: "permission_resolved",
provider: this.providerName,
requestId: request.id,
resolution: response,
};
await this.appendHistoryEvent(permissionResolved);
this.notifySubscribers(permissionResolved);
if (response.behavior === "deny") {
if (response.interrupt) {
const canceled: AgentStreamEvent = {
type: "turn_canceled",
provider: this.providerName,
reason: "permission denied",
};
await this.appendHistoryEvent(canceled);
this.notifySubscribers(canceled);
return;
}
const deniedCompleted: AgentStreamEvent = {
type: "turn_completed",
provider: this.providerName,
usage: { inputTokens: 1, outputTokens: 0 },
};
await this.appendHistoryEvent(deniedCompleted);
this.notifySubscribers(deniedCompleted);
return;
}
}
await this.applyToolSideEffects(tool.name, tool.input ?? {}, textPrompt);
let toolOutput: unknown = tool.output;
if (!toolOutput && (tool.name === "Read" || tool.name === "read_file")) {
const pathInput = typeof tool.input?.path === "string" ? tool.input.path : "/etc/hosts";
const resolvedPath = path.isAbsolute(pathInput)
? pathInput
: path.join(this.config.cwd ?? process.cwd(), pathInput);
try {
const content = readFileSync(resolvedPath, "utf8");
toolOutput = { path: pathInput, content };
} catch {
toolOutput = { path: pathInput, content: "" };
}
}
const toolCompleted: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item: {
type: "tool_call",
name: tool.name,
callId,
status: "completed",
detail: {
type: "unknown",
input: tool.input ?? null,
output: toolOutput ?? { ok: true },
},
error: null,
},
};
await this.appendHistoryEvent(toolCompleted);
this.notifySubscribers(toolCompleted);
}
const assistantText = this.buildAssistantText(textPrompt);
const assistantChunkA: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item: { type: "assistant_message", text: assistantText.slice(0, 6) },
};
await this.appendHistoryEvent(assistantChunkA);
this.notifySubscribers(assistantChunkA);
const assistantChunkB: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item: { type: "assistant_message", text: assistantText.slice(6) },
};
await this.appendHistoryEvent(assistantChunkB);
this.notifySubscribers(assistantChunkB);
const completed: AgentStreamEvent = {
type: "turn_completed",
provider: this.providerName,
usage: result.usage ?? { inputTokens: 1, outputTokens: 1 },
usage: { inputTokens: 1, outputTokens: 1 },
};
await this.appendHistoryEvent(completed);
yield completed;
return;
this.notifySubscribers(completed);
} finally {
this.activeForegroundTurnId = null;
}
const textPrompt = typeof prompt === "string" ? prompt : JSON.stringify(prompt);
const markerMatch = /remember (?:this )?(?:marker|string|project name)[^"]*"([^"]+)"/i.exec(
textPrompt,
);
if (markerMatch) {
this.memoryMarker = markerMatch[1] ?? null;
}
const threadStarted: AgentStreamEvent = {
type: "thread_started",
provider: this.providerName,
sessionId: this.id,
};
await this.appendHistoryEvent(threadStarted);
yield threadStarted;
const turnStarted: AgentStreamEvent = { type: "turn_started", provider: this.providerName };
await this.appendHistoryEvent(turnStarted);
yield turnStarted;
const tool = buildToolCallForPrompt(this.providerName, textPrompt);
if (tool) {
const needsPermission = this.needsPermissionForTool(tool.name, tool.input ?? {});
const callId = randomUUID();
const toolRunning: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item: {
type: "tool_call",
name: tool.name,
callId,
status: "running",
detail: {
type: "unknown",
input: tool.input ?? null,
output: null,
},
error: null,
},
};
await this.appendHistoryEvent(toolRunning);
yield toolRunning;
if (needsPermission) {
const request: AgentPermissionRequest = {
id: randomUUID(),
provider: this.providerName,
name: tool.name,
kind: "tool",
title: "Permission required",
description: "Test permission request",
input: tool.input ?? {},
};
this.pendingPermissions = [request];
this.permissionGate = createDeferred<AgentPermissionResponse>();
const permissionRequested: AgentStreamEvent = {
type: "permission_requested",
provider: this.providerName,
request,
};
await this.appendHistoryEvent(permissionRequested);
yield permissionRequested;
const response = await this.permissionGate.promise;
this.pendingPermissions = [];
const permissionResolved: AgentStreamEvent = {
type: "permission_resolved",
provider: this.providerName,
requestId: request.id,
resolution: response,
};
await this.appendHistoryEvent(permissionResolved);
yield permissionResolved;
if (response.behavior === "deny") {
// Permission denied: do not execute the tool.
if (response.interrupt) {
const canceled: AgentStreamEvent = {
type: "turn_canceled",
provider: this.providerName,
reason: "permission denied",
};
await this.appendHistoryEvent(canceled);
yield canceled;
return;
}
const deniedCompleted: AgentStreamEvent = {
type: "turn_completed",
provider: this.providerName,
usage: { inputTokens: 1, outputTokens: 0 },
};
await this.appendHistoryEvent(deniedCompleted);
yield deniedCompleted;
return;
}
}
await this.applyToolSideEffects(tool.name, tool.input ?? {}, textPrompt);
let toolOutput: unknown = tool.output;
if (!toolOutput && (tool.name === "Read" || tool.name === "read_file")) {
const pathInput = typeof tool.input?.path === "string" ? tool.input.path : "/etc/hosts";
const resolvedPath = path.isAbsolute(pathInput)
? pathInput
: path.join(this.config.cwd ?? process.cwd(), pathInput);
try {
const content = readFileSync(resolvedPath, "utf8");
toolOutput = { path: pathInput, content };
} catch {
toolOutput = { path: pathInput, content: "" };
}
}
const toolCompleted: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item: {
type: "tool_call",
name: tool.name,
callId,
status: "completed",
detail: {
type: "unknown",
input: tool.input ?? null,
output: toolOutput ?? { ok: true },
},
error: null,
},
};
await this.appendHistoryEvent(toolCompleted);
yield toolCompleted;
}
const assistantText = this.buildAssistantText(textPrompt);
// Stream in two chunks to exercise client chunk coalescing.
const assistantChunkA: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item: { type: "assistant_message", text: assistantText.slice(0, 6) },
};
await this.appendHistoryEvent(assistantChunkA);
yield assistantChunkA;
const assistantChunkB: AgentStreamEvent = {
type: "timeline",
provider: this.providerName,
item: { type: "assistant_message", text: assistantText.slice(6) },
};
await this.appendHistoryEvent(assistantChunkB);
yield assistantChunkB;
const completed: AgentStreamEvent = {
type: "turn_completed",
provider: this.providerName,
usage: { inputTokens: 1, outputTokens: 1 },
};
await this.appendHistoryEvent(completed);
yield completed;
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {