Merge origin/main into local-streaming-speech-tts

This commit is contained in:
Mohamed Boudra
2026-02-06 13:45:31 +07:00
52 changed files with 2977 additions and 1059 deletions
@@ -494,4 +494,125 @@ describe("AgentManager", () => {
const updatedAgent = manager.getAgent(snapshot.id);
expect(updatedAgent?.currentModeId).toBe("acceptEdits");
});
test("close during in-flight stream does not clear persistence sessionId", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
class CloseRaceSession implements AgentSession {
readonly provider = "codex" as const;
readonly capabilities = TEST_CAPABILITIES;
readonly id = randomUUID();
private threadId: string | null = this.id;
private releaseStream: (() => void) | null = null;
private closed = false;
async run(): Promise<AgentRunResult> {
return { sessionId: this.id, finalText: "", timeline: [] };
}
async *stream(): AsyncGenerator<AgentStreamEvent> {
yield { type: "turn_started", provider: this.provider };
if (!this.closed) {
await new Promise<void>((resolve) => {
this.releaseStream = resolve;
});
}
yield { type: "turn_canceled", provider: this.provider, reason: "closed" };
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {}
async getRuntimeInfo() {
return {
provider: this.provider,
sessionId: this.threadId,
model: null,
modeId: null,
};
}
async getAvailableModes() {
return [];
}
async getCurrentMode() {
return null;
}
async setMode(): Promise<void> {}
getPendingPermissions() {
return [];
}
async respondToPermission(): Promise<void> {}
describePersistence() {
if (!this.threadId) {
return null;
}
return { provider: this.provider, sessionId: this.threadId };
}
async interrupt(): Promise<void> {}
async close(): Promise<void> {
this.closed = true;
this.threadId = null;
this.releaseStream?.();
}
}
class CloseRaceClient implements AgentClient {
readonly provider = "codex" as const;
readonly capabilities = TEST_CAPABILITIES;
async isAvailable(): Promise<boolean> {
return true;
}
async createSession(): Promise<AgentSession> {
return new CloseRaceSession();
}
async resumeSession(): Promise<AgentSession> {
return new CloseRaceSession();
}
}
const manager = new AgentManager({
clients: {
codex: new CloseRaceClient(),
},
registry: storage,
logger,
idFactory: () => "close-race-agent",
});
const snapshot = await manager.createAgent({
provider: "codex",
cwd: workdir,
});
const stream = manager.streamAgent(snapshot.id, "hello");
await stream.next();
await manager.closeAgent(snapshot.id);
// Drain stream finalizer path after close().
while (true) {
const next = await stream.next();
if (next.done) {
break;
}
}
await manager.flush();
await storage.flush();
const persisted = await storage.get(snapshot.id);
expect(persisted?.persistence?.sessionId).toBe(snapshot.persistence?.sessionId);
});
});
@@ -601,13 +601,17 @@ export class AgentManager {
mutableAgent.pendingRun = null;
mutableAgent.lifecycle = error ? "error" : "idle";
mutableAgent.lastError = error;
mutableAgent.persistence = attachPersistenceCwd(
const persistenceHandle =
mutableAgent.session.describePersistence() ??
(mutableAgent.runtimeInfo?.sessionId
? { provider: mutableAgent.provider, sessionId: mutableAgent.runtimeInfo.sessionId }
: null),
mutableAgent.cwd
);
(mutableAgent.runtimeInfo?.sessionId
? { provider: mutableAgent.provider, sessionId: mutableAgent.runtimeInfo.sessionId }
: null);
if (persistenceHandle) {
mutableAgent.persistence = attachPersistenceCwd(
persistenceHandle,
mutableAgent.cwd
);
}
this.emitState(mutableAgent);
};
@@ -1113,7 +1117,12 @@ export class AgentManager {
case "thread_started":
// Update persistence with the new session ID from the provider.
// persistence.sessionId is the single source of truth for session identity.
agent.persistence = attachPersistenceCwd(agent.session.describePersistence(), agent.cwd);
{
const handle = agent.session.describePersistence();
if (handle) {
agent.persistence = attachPersistenceCwd(handle, agent.cwd);
}
}
break;
case "timeline":
this.recordTimeline(agent, event.item);
@@ -113,6 +113,7 @@ type ClaudeAgentClientOptions = {
type ClaudeAgentSessionOptions = {
defaults?: { agents?: Record<string, AgentDefinition> };
claudePath: string | null;
handle?: AgentPersistenceHandle;
logger: Logger;
};
@@ -315,16 +316,23 @@ export class ClaudeAgentClient implements AgentClient {
private readonly defaults?: { agents?: Record<string, AgentDefinition> };
private readonly logger: Logger;
private readonly claudePath: string | null;
constructor(options: ClaudeAgentClientOptions) {
this.defaults = options.defaults;
this.logger = options.logger.child({ module: "agent", provider: "claude" });
try {
this.claudePath = execSync("which claude", { encoding: "utf8" }).trim() || null;
} catch {
this.claudePath = null;
}
}
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
const claudeConfig = this.assertConfig(config);
return new ClaudeAgentSession(claudeConfig, {
defaults: this.defaults,
claudePath: this.claudePath,
logger: this.logger,
});
}
@@ -342,6 +350,7 @@ export class ClaudeAgentClient implements AgentClient {
const claudeConfig = this.assertConfig(mergedConfig);
return new ClaudeAgentSession(claudeConfig, {
defaults: this.defaults,
claudePath: this.claudePath,
handle,
logger: this.logger,
});
@@ -353,6 +362,7 @@ export class ClaudeAgentClient implements AgentClient {
cwd: options?.cwd ?? process.cwd(),
permissionMode: "plan",
includePartialMessages: false,
...(this.claudePath ? { pathToClaudeCodeExecutable: this.claudePath } : {}),
};
const claudeQuery = query({ prompt, options: claudeOptions });
@@ -407,12 +417,7 @@ export class ClaudeAgentClient implements AgentClient {
}
async isAvailable(): Promise<boolean> {
try {
const claudePath = execSync("which claude", { encoding: "utf8" }).trim();
return Boolean(claudePath);
} catch {
return false;
}
return this.claudePath !== null;
}
private assertConfig(config: AgentSessionConfig): ClaudeAgentConfig {
@@ -429,6 +434,7 @@ class ClaudeAgentSession implements AgentSession {
private readonly config: ClaudeAgentConfig;
private readonly defaults?: { agents?: Record<string, AgentDefinition> };
private readonly claudePath: string | null;
private readonly logger: Logger;
private query: Query | null = null;
private input: Pushable<SDKUserMessage> | null = null;
@@ -464,6 +470,7 @@ class ClaudeAgentSession implements AgentSession {
) {
this.config = config;
this.defaults = options.defaults;
this.claudePath = options.claudePath;
this.logger = options.logger;
const handle = options.handle;
@@ -880,6 +887,7 @@ class ClaudeAgentSession implements AgentSession {
permissionMode: this.currentMode,
agents: this.defaults?.agents,
canUseTool: this.handlePermissionRequest,
...(this.claudePath ? { pathToClaudeCodeExecutable: this.claudePath } : {}),
// Use Claude Code preset system prompt and load CLAUDE.md files
// Append orchestrator mode instructions for agents
systemPrompt: {
@@ -158,7 +158,6 @@ describe("Codex app-server provider (integration)", () => {
thinkingOptionId: CODEX_TEST_THINKING_OPTION_ID,
});
await session.run("Reply with OK.");
const info = await session.getRuntimeInfo();
await session.close();
@@ -730,9 +729,20 @@ describe("Codex app-server provider (integration)", () => {
if (captured) {
expect(sawPermissionResolved).toBe(true);
}
const text = await waitForFileToContainText(targetPath, "ok", { timeoutMs: 5000 });
expect(timelineItems.some((item) => hasApplyPatchFile(item, "approval-test.txt"))).toBe(true);
expect(text?.trim()).toBe("ok");
const sawPatch = timelineItems.some((item) => hasApplyPatchFile(item, "approval-test.txt"));
expect(sawPatch).toBe(true);
const text = await waitForFileToContainText(targetPath, "ok", { timeoutMs: 10000 });
if (!text) {
const toolNames = timelineItems
.filter((item) => item.type === "tool_call")
.map((item) => item.name)
.join(", ");
throw new Error(
`approval-test.txt was not written after file change approval flow (saw tools: ${toolNames || "none"})`
);
}
expect(text.trim()).toBe("ok");
} finally {
cleanup();
rmSync(cwd, { recursive: true, force: true });
@@ -33,9 +33,12 @@ import { Dirent } from "node:fs";
import os from "node:os";
import path from "node:path";
import readline from "node:readline";
import { z } from "zod";
import { loadCodexPersistedTimeline } from "./codex-mcp-agent.js";
const DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000;
const TURN_START_TIMEOUT_MS = 90 * 1000;
const CODEX_PROVIDER = "codex" as const;
const CODEX_IMAGE_ATTACHMENT_DIR = "paseo-attachments";
@@ -676,12 +679,26 @@ function planStepsToTodoItems(steps: Array<{ step: string; status: string }>): {
}));
}
function normalizeCodexFilePath(filePath: unknown, cwd: string | null | undefined): string | null {
if (typeof filePath !== "string") return null;
const trimmed = filePath.trim();
if (!trimmed) return null;
if (typeof cwd === "string" && cwd.trim().length > 0) {
const normalizedCwd = cwd.endsWith(path.sep) ? cwd : `${cwd}${path.sep}`;
if (trimmed.startsWith(normalizedCwd)) {
return trimmed.slice(normalizedCwd.length);
}
}
return trimmed;
}
function threadItemToTimeline(
item: any,
options?: { includeUserMessage?: boolean }
options?: { includeUserMessage?: boolean; cwd?: string | null }
): AgentTimelineItem | null {
if (!item || typeof item !== "object") return null;
const includeUserMessage = options?.includeUserMessage ?? true;
const cwd = options?.cwd ?? null;
switch (item.type) {
case "userMessage": {
if (!includeUserMessage) {
@@ -722,14 +739,17 @@ function threadItemToTimeline(
case "fileChange": {
const files = Array.isArray(item.changes)
? item.changes.map((change: any) => ({
path: change.path,
path: normalizeCodexFilePath(change.path, cwd) ?? change.path,
kind: change.kind,
}))
: [];
const outputFiles = Array.isArray(item.changes)
? item.changes.map((change: any) => ({
path: change.path,
patch: change.diff,
path: normalizeCodexFilePath(change.path, cwd) ?? change.path,
patch:
typeof change.diff === "string"
? truncateUtf8Bytes(change.diff, MAX_FILE_PATCH_BYTES).text
: change.diff,
kind: change.kind,
}))
: [];
@@ -823,6 +843,223 @@ function normalizeImageData(mimeType: string, data: string): ImageDataPayload {
return { mimeType, data };
}
function truncateUtf8Bytes(text: string, maxBytes: number): { text: string; truncated: boolean } {
if (maxBytes <= 0) {
return { text: "", truncated: text.length > 0 };
}
const bytes = Buffer.byteLength(text, "utf8");
if (bytes <= maxBytes) {
return { text, truncated: false };
}
const buffer = Buffer.from(text, "utf8");
const sliced = buffer.subarray(0, maxBytes);
return { text: sliced.toString("utf8"), truncated: true };
}
const MAX_FILE_PATCH_BYTES = 128 * 1024;
const ThreadStartedNotificationSchema = z.object({
thread: z.object({ id: z.string() }).passthrough(),
}).passthrough();
const TurnStartedNotificationSchema = z.object({
turn: z.object({ id: z.string() }).passthrough(),
}).passthrough();
const TurnCompletedNotificationSchema = z.object({
turn: z
.object({
status: z.string(),
error: z
.object({
message: z.string().optional(),
})
.passthrough()
.optional(),
})
.passthrough(),
}).passthrough();
const TurnPlanUpdatedNotificationSchema = z.object({
plan: z.array(
z
.object({
step: z.string().optional(),
status: z.string().optional(),
})
.passthrough()
),
}).passthrough();
const TurnDiffUpdatedNotificationSchema = z.object({
diff: z.string(),
}).passthrough();
const ThreadTokenUsageUpdatedNotificationSchema = z.object({
tokenUsage: z.unknown(),
}).passthrough();
const ItemTextDeltaNotificationSchema = z.object({
itemId: z.string(),
delta: z.string(),
}).passthrough();
const ItemLifecycleNotificationSchema = z.object({
item: z
.object({
id: z.string().optional(),
type: z.string().optional(),
})
.passthrough(),
}).passthrough();
const CodexEventTurnAbortedNotificationSchema = z.object({
msg: z
.object({
type: z.literal("turn_aborted"),
reason: z.string().optional(),
})
.passthrough(),
}).passthrough();
const CodexEventTaskCompleteNotificationSchema = z.object({
msg: z
.object({
type: z.literal("task_complete"),
})
.passthrough(),
}).passthrough();
type ParsedCodexNotification =
| { kind: "thread_started"; threadId: string }
| { kind: "turn_started"; turnId: string }
| { kind: "turn_completed"; status: string; errorMessage: string | null }
| { kind: "plan_updated"; plan: Array<{ step: string | null; status: string | null }> }
| { kind: "diff_updated"; diff: string }
| { kind: "token_usage_updated"; tokenUsage: unknown }
| { kind: "agent_message_delta"; itemId: string; delta: string }
| { kind: "reasoning_delta"; itemId: string; delta: string }
| { kind: "item_completed"; item: { id?: string; type?: string; [key: string]: unknown } }
| { kind: "item_started"; item: { id?: string; type?: string; [key: string]: unknown } }
| { kind: "invalid_payload"; method: string; params: unknown }
| { kind: "unknown_method"; method: string; params: unknown };
const CodexNotificationSchema = z.union([
z.object({ method: z.literal("thread/started"), params: ThreadStartedNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({ kind: "thread_started", threadId: params.thread.id })
),
z.object({ method: z.literal("thread/started"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.literal("turn/started"), params: TurnStartedNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({ kind: "turn_started", turnId: params.turn.id })
),
z.object({ method: z.literal("turn/started"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.literal("turn/completed"), params: TurnCompletedNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({
kind: "turn_completed",
status: params.turn.status,
errorMessage: params.turn.error?.message ?? null,
})
),
z.object({ method: z.literal("turn/completed"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.literal("turn/plan/updated"), params: TurnPlanUpdatedNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({
kind: "plan_updated",
plan: params.plan.map((entry) => ({
step: entry.step ?? null,
status: entry.status ?? null,
})),
})
),
z.object({ method: z.literal("turn/plan/updated"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.literal("turn/diff/updated"), params: TurnDiffUpdatedNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({ kind: "diff_updated", diff: params.diff })
),
z.object({ method: z.literal("turn/diff/updated"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({
method: z.literal("thread/tokenUsage/updated"),
params: ThreadTokenUsageUpdatedNotificationSchema,
}).transform(
({ params }): ParsedCodexNotification => ({ kind: "token_usage_updated", tokenUsage: params.tokenUsage })
),
z.object({ method: z.literal("thread/tokenUsage/updated"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.literal("item/agentMessage/delta"), params: ItemTextDeltaNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({
kind: "agent_message_delta",
itemId: params.itemId,
delta: params.delta,
})
),
z.object({ method: z.literal("item/agentMessage/delta"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({
method: z.literal("item/reasoning/summaryTextDelta"),
params: ItemTextDeltaNotificationSchema,
}).transform(
({ params }): ParsedCodexNotification => ({
kind: "reasoning_delta",
itemId: params.itemId,
delta: params.delta,
})
),
z.object({ method: z.literal("item/reasoning/summaryTextDelta"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.literal("item/completed"), params: ItemLifecycleNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({ kind: "item_completed", item: params.item })
),
z.object({ method: z.literal("item/completed"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.literal("item/started"), params: ItemLifecycleNotificationSchema }).transform(
({ params }): ParsedCodexNotification => ({ kind: "item_started", item: params.item })
),
z.object({ method: z.literal("item/started"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({
method: z.literal("codex/event/turn_aborted"),
params: CodexEventTurnAbortedNotificationSchema,
}).transform(
(): ParsedCodexNotification => ({
kind: "turn_completed",
status: "interrupted",
errorMessage: null,
})
),
z.object({ method: z.literal("codex/event/turn_aborted"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({
method: z.literal("codex/event/task_complete"),
params: CodexEventTaskCompleteNotificationSchema,
}).transform(
(): ParsedCodexNotification => ({
kind: "turn_completed",
status: "completed",
errorMessage: null,
})
),
z.object({ method: z.literal("codex/event/task_complete"), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "invalid_payload", method, params })
),
z.object({ method: z.string(), params: z.unknown() }).transform(
({ method, params }): ParsedCodexNotification => ({ kind: "unknown_method", method, params })
),
]);
async function writeImageAttachment(mimeType: string, data: string): Promise<string> {
const attachmentsDir = path.join(os.tmpdir(), CODEX_IMAGE_ATTACHMENT_DIR);
await fs.mkdir(attachmentsDir, { recursive: true });
@@ -899,6 +1136,8 @@ class CodexAppServerAgentSession implements AgentSession {
private resolvedPermissionRequests = new Set<string>();
private pendingAgentMessages = new Map<string, string>();
private pendingReasoning = new Map<string, string[]>();
private warnedUnknownNotificationMethods = new Set<string>();
private warnedInvalidNotificationPayloads = new Set<string>();
private latestUsage: AgentUsage | undefined;
private connected = false;
private collaborationModes: Array<{
@@ -1054,22 +1293,40 @@ class CodexAppServerAgentSession implements AgentSession {
private async loadPersistedHistory(): Promise<void> {
if (!this.client || !this.currentThreadId) return;
try {
let rolloutTimeline: AgentTimelineItem[] = [];
try {
rolloutTimeline = await loadCodexPersistedTimeline(
this.currentThreadId,
undefined,
this.logger
);
} catch {
rolloutTimeline = [];
}
const response = (await this.client.request("thread/read", {
threadId: this.currentThreadId,
includeTurns: true,
})) as { thread?: { turns?: Array<{ items?: any[] }> } };
const thread = response?.thread;
if (!thread || !Array.isArray(thread.turns)) return;
const timeline: AgentTimelineItem[] = [];
for (const turn of thread.turns) {
const items = Array.isArray(turn.items) ? turn.items : [];
for (const item of items) {
const timelineItem = threadItemToTimeline(item);
if (timelineItem) {
timeline.push(timelineItem);
const threadTimeline: AgentTimelineItem[] = [];
if (thread && Array.isArray(thread.turns)) {
for (const turn of thread.turns) {
const items = Array.isArray(turn.items) ? turn.items : [];
for (const item of items) {
const timelineItem = threadItemToTimeline(item, {
cwd: this.config.cwd ?? null,
});
if (timelineItem) {
threadTimeline.push(timelineItem);
}
}
}
}
const timeline =
rolloutTimeline.length > 0 ? rolloutTimeline : threadTimeline;
if (timeline.length > 0) {
this.persistedHistory = timeline;
this.historyPending = true;
@@ -1176,7 +1433,7 @@ class CodexAppServerAgentSession implements AgentSession {
params.outputSchema = options.outputSchema;
}
await this.client.request("turn/start", params);
await this.client.request("turn/start", params, TURN_START_TIMEOUT_MS);
for await (const event of queue) {
yield event;
@@ -1209,6 +1466,12 @@ class CodexAppServerAgentSession implements AgentSession {
async getRuntimeInfo(): Promise<AgentRuntimeInfo> {
if (this.cachedRuntimeInfo) return { ...this.cachedRuntimeInfo };
if (!this.connected) {
await this.connect();
}
if (!this.currentThreadId) {
await this.ensureThread();
}
const info: AgentRuntimeInfo = {
provider: CODEX_PROVIDER,
sessionId: this.currentThreadId,
@@ -1479,125 +1742,141 @@ class CodexAppServerAgentSession implements AgentSession {
private handleNotification(method: string, params: unknown): void {
this.emitEvent({ type: "provider_event", provider: CODEX_PROVIDER, raw: { method, params } });
const parsed = CodexNotificationSchema.parse({ method, params });
switch (method) {
case "thread/started": {
const threadId = (params as any)?.thread?.id;
if (threadId) {
this.currentThreadId = threadId;
this.emitEvent({ type: "thread_started", provider: CODEX_PROVIDER, sessionId: threadId });
}
break;
}
case "turn/started": {
const turnId = (params as any)?.turn?.id;
if (turnId) {
this.currentTurnId = turnId;
}
this.emitEvent({ type: "turn_started", provider: CODEX_PROVIDER });
break;
}
case "turn/completed": {
const turn = (params as any)?.turn;
const status = turn?.status;
if (status === "failed") {
const message = turn?.error?.message ?? "Codex turn failed";
this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, error: message });
} else if (status === "interrupted") {
this.emitEvent({ type: "turn_canceled", provider: CODEX_PROVIDER, reason: "interrupted" });
} else {
this.emitEvent({ type: "turn_completed", provider: CODEX_PROVIDER, usage: this.latestUsage });
}
this.eventQueue?.end();
break;
}
case "turn/plan/updated": {
const steps = (params as any)?.plan;
if (Array.isArray(steps)) {
const items = planStepsToTodoItems(
steps.map((entry: any) => ({
step: entry.step ?? "",
status: entry.status ?? "pending",
}))
);
this.emitEvent({
type: "timeline",
provider: CODEX_PROVIDER,
item: { type: "todo", items },
});
}
break;
}
case "turn/diff/updated": {
const diff = (params as any)?.diff;
if (typeof diff === "string" && diff.trim().length > 0) {
this.emitEvent({
type: "timeline",
provider: CODEX_PROVIDER,
item: createToolCallTimelineItem({
name: "apply_patch",
status: "running",
output: { diff },
}),
});
}
break;
}
case "thread/tokenUsage/updated": {
const tokenUsage = (params as any)?.tokenUsage;
this.latestUsage = toAgentUsage(tokenUsage);
break;
}
case "item/agentMessage/delta": {
const itemId = (params as any)?.itemId;
const delta = (params as any)?.delta;
if (itemId && typeof delta === "string") {
const prev = this.pendingAgentMessages.get(itemId) ?? "";
this.pendingAgentMessages.set(itemId, prev + delta);
}
break;
}
case "item/reasoning/summaryTextDelta": {
const itemId = (params as any)?.itemId;
const delta = (params as any)?.delta;
if (itemId && typeof delta === "string") {
const prev = this.pendingReasoning.get(itemId) ?? [];
prev.push(delta);
this.pendingReasoning.set(itemId, prev);
}
break;
}
case "item/completed": {
const item = (params as any)?.item;
const timelineItem = threadItemToTimeline(item, { includeUserMessage: false });
if (timelineItem) {
if (timelineItem.type === "assistant_message" && item?.id) {
const buffered = this.pendingAgentMessages.get(item.id);
if (buffered && buffered.length > 0) {
timelineItem.text = buffered;
}
}
if (timelineItem.type === "reasoning" && item?.id) {
const buffered = this.pendingReasoning.get(item.id);
if (buffered && buffered.length > 0) {
timelineItem.text = buffered.join("");
}
}
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
}
break;
}
case "item/started": {
const item = (params as any)?.item;
const timelineItem = threadItemToTimeline(item, { includeUserMessage: false });
if (timelineItem && timelineItem.type === "tool_call") {
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
}
break;
}
default:
break;
if (parsed.kind === "thread_started") {
this.currentThreadId = parsed.threadId;
this.emitEvent({ type: "thread_started", provider: CODEX_PROVIDER, sessionId: parsed.threadId });
return;
}
if (parsed.kind === "turn_started") {
this.currentTurnId = parsed.turnId;
this.emitEvent({ type: "turn_started", provider: CODEX_PROVIDER });
return;
}
if (parsed.kind === "turn_completed") {
if (parsed.status === "failed") {
this.emitEvent({
type: "turn_failed",
provider: CODEX_PROVIDER,
error: parsed.errorMessage ?? "Codex turn failed",
});
} else if (parsed.status === "interrupted") {
this.emitEvent({ type: "turn_canceled", provider: CODEX_PROVIDER, reason: "interrupted" });
} else {
this.emitEvent({ type: "turn_completed", provider: CODEX_PROVIDER, usage: this.latestUsage });
}
this.eventQueue?.end();
return;
}
if (parsed.kind === "plan_updated") {
const items = planStepsToTodoItems(
parsed.plan.map((entry) => ({
step: entry.step ?? "",
status: entry.status ?? "pending",
}))
);
this.emitEvent({
type: "timeline",
provider: CODEX_PROVIDER,
item: { type: "todo", items },
});
return;
}
if (parsed.kind === "diff_updated") {
if (parsed.diff.trim().length > 0) {
// NOTE: Codex app-server emits frequent `turn/diff/updated` notifications
// containing a full accumulated unified diff for the *entire turn*.
// This is not a concrete file-change tool call; it is progress telemetry.
// We intentionally do NOT store it in the agent timeline to avoid
// snapshot bloat and relay/WebSocket size limits.
}
return;
}
if (parsed.kind === "token_usage_updated") {
this.latestUsage = toAgentUsage(parsed.tokenUsage);
return;
}
if (parsed.kind === "agent_message_delta") {
const prev = this.pendingAgentMessages.get(parsed.itemId) ?? "";
this.pendingAgentMessages.set(parsed.itemId, prev + parsed.delta);
return;
}
if (parsed.kind === "reasoning_delta") {
const prev = this.pendingReasoning.get(parsed.itemId) ?? [];
prev.push(parsed.delta);
this.pendingReasoning.set(parsed.itemId, prev);
return;
}
if (parsed.kind === "item_completed") {
const timelineItem = threadItemToTimeline(parsed.item, {
includeUserMessage: false,
cwd: this.config.cwd ?? null,
});
if (timelineItem) {
const itemId = parsed.item.id;
if (timelineItem.type === "assistant_message" && itemId) {
const buffered = this.pendingAgentMessages.get(itemId);
if (buffered && buffered.length > 0) {
timelineItem.text = buffered;
}
}
if (timelineItem.type === "reasoning" && itemId) {
const buffered = this.pendingReasoning.get(itemId);
if (buffered && buffered.length > 0) {
timelineItem.text = buffered.join("");
}
}
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
}
return;
}
if (parsed.kind === "item_started") {
const timelineItem = threadItemToTimeline(parsed.item, {
includeUserMessage: false,
cwd: this.config.cwd ?? null,
});
if (timelineItem && timelineItem.type === "tool_call") {
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
}
return;
}
if (parsed.kind === "invalid_payload") {
this.warnInvalidNotificationPayload(parsed.method, parsed.params);
return;
}
this.warnUnknownNotificationMethod(parsed.method, parsed.params);
}
private warnUnknownNotificationMethod(method: string, params: unknown): void {
if (this.warnedUnknownNotificationMethods.has(method)) {
return;
}
this.warnedUnknownNotificationMethods.add(method);
this.logger.warn({ method, params }, "Unhandled Codex app-server notification method");
}
private warnInvalidNotificationPayload(method: string, params: unknown): void {
const key = method;
if (this.warnedInvalidNotificationPayloads.has(key)) {
return;
}
this.warnedInvalidNotificationPayloads.add(key);
this.logger.warn(
{ method, params },
"Invalid Codex app-server notification payload"
);
}
private handleCommandApprovalRequest(params: unknown): Promise<unknown> {
@@ -1741,19 +2020,27 @@ export class CodexAppServerAgentClient implements AgentClient {
const title = thread.preview ?? null;
let timeline: AgentTimelineItem[] = [];
try {
const rolloutTimeline = await loadCodexPersistedTimeline(
threadId,
undefined,
this.logger
);
const read = (await client.request("thread/read", {
threadId,
includeTurns: true,
})) as { thread?: { turns?: Array<{ items?: any[] }> } };
const turns = read.thread?.turns ?? [];
const items: AgentTimelineItem[] = [];
const itemsFromThreadRead: AgentTimelineItem[] = [];
for (const turn of turns) {
for (const item of turn.items ?? []) {
const timelineItem = threadItemToTimeline(item);
if (timelineItem) items.push(timelineItem);
const timelineItem = threadItemToTimeline(item, { cwd });
if (timelineItem) itemsFromThreadRead.push(timelineItem);
}
}
timeline = items;
timeline =
rolloutTimeline.length > 0
? rolloutTimeline
: itemsFromThreadRead;
} catch {
timeline = [];
}
@@ -5097,10 +5097,20 @@ async function findRolloutFile(
return null;
}
const RolloutContentItemSchema = z.object({
type: z.string(),
text: z.string(),
}).passthrough();
const RolloutContentItemSchema = z
.union([
z.object({ type: z.literal("input_text"), text: z.string() }),
z.object({ type: z.literal("output_text"), text: z.string() }),
z.object({ type: z.literal("reasoning_text"), text: z.string() }),
z.object({ type: z.literal("text"), text: z.string() }),
z
.object({
type: z.string(),
text: z.string().optional(),
message: z.string().optional(),
})
.passthrough(),
]);
const RolloutContentArraySchema = z.array(RolloutContentItemSchema);
@@ -5116,77 +5126,131 @@ function extractContentTextByType(content: unknown, itemType: string): string {
.trim();
}
const RolloutResponsePayloadSchema = z.object({
type: z.string().optional(),
role: z.string().optional(),
const RolloutResponseMessagePayloadSchema = z.object({
type: z.literal("message"),
role: z.enum(["user", "assistant"]).optional(),
content: z.unknown().optional(),
});
const RolloutResponseReasoningPayloadSchema = z.object({
type: z.literal("reasoning"),
content: z.unknown().optional(),
name: z.string().optional(),
call_id: z.string().optional(),
arguments: z.string().optional(),
output: z.string().optional(),
summary: z.array(z.object({ text: z.string().optional() })).optional(),
text: z.string().optional(),
});
const RolloutEventPayloadSchema = z.object({
type: z.string().optional(),
const RolloutResponseFunctionCallPayloadSchema = z.object({
type: z.literal("function_call"),
name: z.string().optional(),
call_id: z.string().optional(),
arguments: z.string().optional(),
});
const RolloutResponseCustomToolCallPayloadSchema = z.object({
type: z.literal("custom_tool_call"),
name: z.string().optional(),
call_id: z.string().optional(),
arguments: z.string().optional(),
});
const RolloutResponseFunctionCallOutputPayloadSchema = z.object({
type: z.literal("function_call_output"),
call_id: z.string().optional(),
output: z.string().optional(),
});
const RolloutEventAgentReasoningPayloadSchema = z.object({
type: z.literal("agent_reasoning"),
text: z.string().optional(),
message: z.string().optional(),
});
const RolloutEntrySchema = z.object({
type: z.enum(["response_item", "event_msg"]),
payload: z.unknown().optional(),
const RolloutEventAgentMessagePayloadSchema = z.object({
type: z.literal("agent_message"),
message: z
.union([
z.string(),
z
.object({
role: z.string().optional(),
message: z.string().optional(),
text: z.string().optional(),
})
.passthrough(),
])
.optional(),
});
type RolloutEntry = z.infer<typeof RolloutEntrySchema>;
type RolloutResponsePayload = z.infer<typeof RolloutResponsePayloadSchema>;
const RolloutEventUserMessagePayloadSchema = z.object({
type: z.literal("user_message"),
message: z
.union([
z.string(),
z
.object({
role: z.string().optional(),
message: z.string().optional(),
text: z.string().optional(),
})
.passthrough(),
])
.optional(),
});
function parseRolloutEntryFromLine(line: string): RolloutEntry | null {
if (!line) {
return null;
}
try {
const parsed = JSON.parse(line);
const result = RolloutEntrySchema.safeParse(parsed);
if (result.success) {
return result.data;
}
if (
parsed &&
typeof parsed === "object" &&
typeof (parsed as { output?: unknown }).output === "string"
) {
return parseRolloutEntryFromLine((parsed as { output: string }).output);
}
} catch {
return null;
}
return null;
}
type RolloutResponseReasoningPayload = z.infer<
typeof RolloutResponseReasoningPayloadSchema
>;
type ParsedRolloutRecord =
| { kind: "timeline"; item: AgentTimelineItem }
| { kind: "call"; name: string; callId?: string; input?: unknown }
| { kind: "output"; callId: string; output: string }
| { kind: "ignore" };
const RolloutMessageContentSchema = z
.union([
z.string().transform((content) => content.trim()),
z.array(
z
.object({
text: z.string().optional(),
message: z.string().optional(),
})
.passthrough()
).transform((content) =>
content
.map((block) => block.text ?? block.message ?? "")
.map((text) => text.trim())
.filter(Boolean)
.join("\n")
.trim()
),
]);
function extractMessageText(content: unknown): string {
if (!Array.isArray(content)) {
const parsed = RolloutMessageContentSchema.safeParse(content);
if (!parsed.success) {
return "";
}
const parts: string[] = [];
for (const block of content) {
if (!block || typeof block !== "object") {
continue;
}
const record = block as Record<string, unknown>;
const text = typeof record.text === "string" ? record.text : undefined;
if (text && text.trim()) {
parts.push(text.trim());
continue;
}
const message =
typeof record.message === "string" ? record.message : undefined;
if (message && message.trim()) {
parts.push(message.trim());
}
return parsed.data;
}
const RolloutEventMessageTextSchema = z
.union([
z.string(),
z
.object({
message: z.string().optional(),
text: z.string().optional(),
})
.passthrough()
.transform((message) => message.message ?? message.text ?? ""),
]);
function extractEventMessageText(message: unknown): string {
const parsed = RolloutEventMessageTextSchema.safeParse(message);
if (!parsed.success) {
return "";
}
return parts.join("\n").trim();
return parsed.data;
}
function isSyntheticRolloutUserMessage(text: string): boolean {
@@ -5207,10 +5271,10 @@ function isSyntheticRolloutUserMessage(text: string): boolean {
return false;
}
function extractReasoningText(payload: RolloutResponsePayload): string {
if (Array.isArray(payload?.summary)) {
function extractReasoningText(payload: RolloutResponseReasoningPayload): string {
if (Array.isArray(payload.summary)) {
const text = payload.summary
.map((item) => (item && typeof item.text === "string" ? item.text : ""))
.map((item) => item.text ?? "")
.filter(Boolean)
.join("\n")
.trim();
@@ -5218,17 +5282,117 @@ function extractReasoningText(payload: RolloutResponsePayload): string {
return text;
}
}
// Handle content array with reasoning_text items
const contentText = extractContentTextByType(payload.content, "reasoning_text");
if (contentText) {
return contentText;
}
if (typeof payload?.text === "string") {
if (typeof payload.text === "string") {
return payload.text;
}
return "";
}
function parseJsonLikeString(value: string): unknown {
try {
return JSON.parse(value);
} catch {
return value;
}
}
const FunctionCallInputNormalizationSchema = z
.union([
z.object({ cmd: z.string() }).transform((input) => ({ name: "Bash", input: { command: input.cmd } })),
z
.object({ command: z.array(z.string()) })
.transform((input) => ({ name: "Bash", input: { command: input.command[2] ?? "" } })),
z.unknown().transform((input) => ({ name: "unknown", input })),
]);
const RolloutResponseRecordSchema = z
.union([
RolloutResponseMessagePayloadSchema.transform((payload): ParsedRolloutRecord => {
const text = extractMessageText(payload.content);
const itemType = payload.role === "assistant" ? "assistant_message" : "user_message";
const shouldEmit = text.length > 0 && (itemType !== "user_message" || !isSyntheticRolloutUserMessage(text));
return shouldEmit
? { kind: "timeline", item: { type: itemType, text } }
: { kind: "ignore" };
}),
RolloutResponseReasoningPayloadSchema.transform((payload): ParsedRolloutRecord => {
const text = extractReasoningText(payload);
return text.length > 0
? { kind: "timeline", item: { type: "reasoning", text } }
: { kind: "ignore" };
}),
z
.union([
RolloutResponseFunctionCallPayloadSchema,
RolloutResponseCustomToolCallPayloadSchema,
])
.transform((payload): ParsedRolloutRecord => {
const rawName = payload.name ?? "unknown";
const parsedArguments = payload.arguments ? parseJsonLikeString(payload.arguments) : undefined;
const normalized =
rawName === "exec_command" || rawName === "shell"
? FunctionCallInputNormalizationSchema.parse(parsedArguments)
: { name: rawName, input: parsedArguments };
const skip = rawName === "write_stdin";
return skip
? { kind: "ignore" }
: {
kind: "call",
name: normalized.name,
callId: payload.call_id,
input: normalized.input,
};
}),
RolloutResponseFunctionCallOutputPayloadSchema.transform(
(payload): ParsedRolloutRecord =>
payload.call_id && payload.output
? { kind: "output", callId: payload.call_id, output: payload.output }
: { kind: "ignore" }
),
z.unknown().transform((): ParsedRolloutRecord => ({ kind: "ignore" })),
]);
const RolloutEventRecordSchema = z
.union([
RolloutEventAgentReasoningPayloadSchema.transform(
(payload): ParsedRolloutRecord =>
payload.text
? { kind: "timeline", item: { type: "reasoning", text: payload.text } }
: { kind: "ignore" }
),
RolloutEventAgentMessagePayloadSchema.transform((payload): ParsedRolloutRecord => {
const text = extractEventMessageText(payload.message);
return text.length > 0
? { kind: "timeline", item: { type: "assistant_message", text } }
: { kind: "ignore" };
}),
RolloutEventUserMessagePayloadSchema.transform((payload): ParsedRolloutRecord => {
const text = extractEventMessageText(payload.message);
const shouldEmit = text.length > 0 && !isSyntheticRolloutUserMessage(text);
return shouldEmit
? { kind: "timeline", item: { type: "user_message", text } }
: { kind: "ignore" };
}),
z.unknown().transform((): ParsedRolloutRecord => ({ kind: "ignore" })),
]);
const RolloutRecordSchema = z
.object({
type: z.enum(["response_item", "event_msg"]),
payload: z.unknown().optional(),
item: z.unknown().optional(),
msg: z.unknown().optional(),
})
.transform((entry) =>
entry.type === "response_item"
? RolloutResponseRecordSchema.parse(entry.payload ?? entry.item)
: RolloutEventRecordSchema.parse(entry.payload ?? entry.msg)
);
function parseJsonRolloutTimeline(
parsed: unknown
): AgentTimelineItem[] | null {
@@ -5244,25 +5408,28 @@ function parseJsonRolloutTimeline(
if (!entry || typeof entry !== "object") {
continue;
}
const record = entry as Record<string, unknown>;
const type = record.type;
if (type === "message") {
const role = record.role;
const text = extractMessageText(record.content);
if (!text || typeof role !== "string") {
const messagePayloadResult =
RolloutResponseMessagePayloadSchema.safeParse(entry);
if (messagePayloadResult.success) {
const payload = messagePayloadResult.data;
const text = extractMessageText(payload.content);
if (!text) {
continue;
}
if (role === "assistant") {
if (payload.role === "assistant") {
timeline.push({ type: "assistant_message", text });
} else if (role === "user") {
} else if (payload.role === "user") {
if (!isSyntheticRolloutUserMessage(text)) {
timeline.push({ type: "user_message", text });
}
}
continue;
}
if (type === "reasoning") {
const text = extractReasoningText(record as RolloutResponsePayload);
const reasoningPayloadResult =
RolloutResponseReasoningPayloadSchema.safeParse(entry);
if (reasoningPayloadResult.success) {
const text = extractReasoningText(reasoningPayloadResult.data);
if (text) {
timeline.push({ type: "reasoning", text });
}
@@ -5294,126 +5461,39 @@ export async function parseRolloutFile(
.map((line) => line.trim())
.filter(Boolean);
// First pass: collect function_call_output entries by call_id
const outputsByCallId = new Map<string, string>();
for (const line of lines) {
const entry = parseRolloutEntryFromLine(line);
if (!entry || entry.type !== "response_item") continue;
const payloadResult = RolloutResponsePayloadSchema.safeParse(entry.payload);
if (!payloadResult.success) continue;
const payload = payloadResult.data;
if (payload.type === "function_call_output" && payload.call_id && payload.output) {
outputsByCallId.set(payload.call_id, payload.output);
}
}
// Second pass: build timeline
const timeline: AgentTimelineItem[] = [];
for (const line of lines) {
const entry = parseRolloutEntryFromLine(line);
if (!entry) continue;
if (entry.type === "response_item") {
const payloadResult = RolloutResponsePayloadSchema.safeParse(entry.payload);
if (!payloadResult.success) continue;
const payload = payloadResult.data;
switch (payload.type) {
case "message": {
const text = extractMessageText(payload.content);
if (text) {
if (payload.role === "assistant") {
timeline.push({ type: "assistant_message", text });
} else if (payload.role === "user") {
if (!isSyntheticRolloutUserMessage(text)) {
timeline.push({ type: "user_message", text });
}
}
}
break;
}
case "reasoning": {
const text = extractReasoningText(payload);
if (text) {
timeline.push({ type: "reasoning", text });
}
break;
}
case "function_call":
case "custom_tool_call": {
const rawName = payload.name ?? "unknown";
const callId = payload.call_id;
// Skip internal polling calls
if (rawName === "write_stdin") {
break;
}
let input: unknown;
if (payload.arguments) {
try {
input = JSON.parse(payload.arguments);
} catch {
input = payload.arguments;
}
}
// Map exec_command and shell to Bash with normalized input
let name = rawName;
if (rawName === "exec_command" && input && typeof input === "object") {
const execInput = input as { cmd?: string };
if (execInput.cmd) {
name = "Bash";
input = { command: execInput.cmd };
}
} else if (rawName === "shell" && input && typeof input === "object") {
// Older format: { command: ["bash", "-lc", "actual cmd"], workdir: "..." }
const shellInput = input as { command?: string[] };
if (Array.isArray(shellInput.command) && shellInput.command.length >= 3) {
name = "Bash";
// command[2] is the actual shell command after "bash -lc"
input = { command: shellInput.command[2] };
}
}
// Attach output if available
const output = callId ? outputsByCallId.get(callId) : undefined;
timeline.push({
type: "tool_call",
name,
callId,
status: "completed",
input,
...(output ? { output } : {}),
});
break;
}
case "function_call_output":
// Already processed in first pass
break;
default:
break;
const parsedRecords = lines
.map((line) => {
try {
return JSON.parse(line);
} catch {
return null;
}
} else if (entry.type === "event_msg") {
const payloadResult = RolloutEventPayloadSchema.safeParse(entry.payload);
if (!payloadResult.success) continue;
const payload = payloadResult.data;
})
.filter((record): record is unknown => record !== null)
.map((record) => RolloutRecordSchema.safeParse(record))
.filter((result): result is { success: true; data: ParsedRolloutRecord } => result.success)
.map((result) => result.data);
if (payload.type === "agent_reasoning" && payload.text) {
timeline.push({ type: "reasoning", text: payload.text });
} else if (payload.type === "agent_message" && payload.message) {
timeline.push({ type: "assistant_message", text: payload.message });
} else if (payload.type === "user_message" && payload.message) {
if (!isSyntheticRolloutUserMessage(payload.message)) {
timeline.push({ type: "user_message", text: payload.message });
}
}
}
}
const outputsByCallId = parsedRecords
.filter((record): record is Extract<ParsedRolloutRecord, { kind: "output" }> => record.kind === "output")
.reduce((map, record) => map.set(record.callId, record.output), new Map<string, string>());
return timeline;
return parsedRecords.flatMap((record): AgentTimelineItem[] =>
record.kind === "timeline"
? [record.item]
: record.kind === "call"
? [
{
type: "tool_call",
name: record.name,
callId: record.callId,
status: "completed",
input: record.input,
output: record.callId ? outputsByCallId.get(record.callId) : undefined,
},
]
: []
);
}
type CodexPersistedTimelineOptions = {
@@ -216,6 +216,79 @@ describe("codex rollout parsing", () => {
text: "Let me think about this.",
});
});
test("parses legacy response_item shape using item", async () => {
const rolloutPath = join(tmpDir, "rollout.jsonl");
const lines = [
JSON.stringify({
timestamp: "2026-01-22T07:08:54.378Z",
type: "response_item",
item: {
type: "function_call",
name: "exec_command",
arguments: '{"cmd":"echo hello"}',
call_id: "call_legacy_1",
},
}),
JSON.stringify({
timestamp: "2026-01-22T07:09:01.785Z",
type: "response_item",
item: {
type: "function_call_output",
call_id: "call_legacy_1",
output: "hello",
},
}),
];
writeFileSync(rolloutPath, lines.join("\n") + "\n");
const timeline = await parseRolloutFile(rolloutPath);
const toolCalls = timeline.filter((i) => i.type === "tool_call");
expect(toolCalls.length).toBe(1);
expect(toolCalls[0]).toMatchObject({
type: "tool_call",
name: "Bash",
callId: "call_legacy_1",
input: { command: "echo hello" },
output: "hello",
});
});
test("parses legacy event_msg shape using msg", async () => {
const rolloutPath = join(tmpDir, "rollout.jsonl");
const lines = [
JSON.stringify({
timestamp: "2026-01-22T07:08:54.378Z",
type: "event_msg",
msg: {
type: "agent_reasoning",
text: "thinking",
},
}),
JSON.stringify({
timestamp: "2026-01-22T07:08:54.378Z",
type: "event_msg",
msg: {
type: "agent_message",
message: { role: "assistant", message: "done" },
},
}),
JSON.stringify({
timestamp: "2026-01-22T07:08:54.378Z",
type: "event_msg",
msg: {
type: "user_message",
message: { role: "user", message: "question" },
},
}),
];
writeFileSync(rolloutPath, lines.join("\n") + "\n");
const timeline = await parseRolloutFile(rolloutPath);
expect(timeline).toContainEqual({ type: "reasoning", text: "thinking" });
expect(timeline).toContainEqual({ type: "assistant_message", text: "done" });
expect(timeline).toContainEqual({ type: "user_message", text: "question" });
});
});
describe("complex conversation", () => {
+1
View File
@@ -33,6 +33,7 @@ async function main() {
if (process.argv.includes("--no-mcp")) {
config.mcpEnabled = false;
}
const daemon = await createPaseoDaemon(config, logger);
try {
+18
View File
@@ -585,6 +585,24 @@ export class Session {
return;
}
// Reduce bandwidth/CPU on mobile: only forward high-frequency agent stream events
// while the app is visible and the user is focused on that agent.
//
// History catch-up is handled via explicit `initialize_agent_request` which emits a
// batched `agent_stream_snapshot`.
const activity = this.clientActivity;
if (activity?.deviceType === "mobile") {
if (!activity.appVisible) {
return;
}
if (!activity.focusedAgentId) {
return;
}
if (activity.focusedAgentId !== event.agentId) {
return;
}
}
const payload = {
agentId: event.agentId,
event: serializeAgentStreamEvent(event.event),
@@ -90,6 +90,25 @@ function buildPersistence(
function buildToolCallForPrompt(provider: string, prompt: string) {
const text = prompt.toLowerCase();
const createFileMatch =
/create a file named\s+"([^"]+)"\s+with the content\s+"([^"]*)"/i.exec(prompt) ??
/create a file named\s+"([^"]+)"\s+with the content\s+'([^']*)'/i.exec(prompt);
if (createFileMatch) {
const fileName = createFileMatch[1] ?? "test.txt";
const content = createFileMatch[2] ?? "";
if (provider === "codex") {
return {
name: "shell",
input: { command: `printf "%s" "${content}" > ${fileName}` },
output: { ok: true },
};
}
return {
name: "Bash",
input: { command: `printf "%s" "${content}" > ${fileName}` },
output: { ok: true },
};
}
if (provider === "claude") {
if (text.includes("read") && text.includes("/etc/hosts")) {
return { name: "Read", input: { path: "/etc/hosts" }, output: undefined };
@@ -374,6 +393,7 @@ class FakeAgentSession implements AgentSession {
{ id: "default", label: "Default", description: "Ask for permissions" },
{ id: "full-access", label: "Full access", description: "No prompts" },
{ id: "auto", label: "Auto", description: "Ask/allow based on policy" },
{ id: "always-ask", label: "Always Ask", description: "Always prompt" },
];
}
@@ -475,6 +495,20 @@ class FakeAgentSession implements AgentSession {
private buildAssistantText(prompt: string): string {
const lower = prompt.toLowerCase();
// Special-case for tests that ask the agent to run pwd but use a placeholder in the
// "respond with exactly" instruction.
if (lower.includes("run `pwd`") && lower.includes("respond with exactly: cwd:")) {
const cwd = this.config.cwd ?? process.cwd();
return `CWD: ${cwd}`;
}
const respondExactlyMatch =
/respond with exactly:\s*([^\n\r]+)\s*$/i.exec(prompt) ??
/respond with exactly:\s*([^\n\r]+)/i.exec(prompt);
if (respondExactlyMatch) {
return (respondExactlyMatch[1] ?? "").trim();
}
if (lower.includes("state saved")) return "state saved";
if (lower.includes("timeline test")) return "timeline test";
if (lower.includes("quick brown fox") && lower.includes("lazy dog")) {
@@ -499,6 +533,9 @@ class FakeAgentSession implements AgentSession {
prompt: string
): Promise<void> {
const lower = prompt.toLowerCase();
const createFileMatch =
/create a file named\s+"([^"]+)"\s+with the content\s+"([^"]*)"/i.exec(prompt) ??
/create a file named\s+"([^"]+)"\s+with the content\s+'([^']*)'/i.exec(prompt);
if (toolName === "Read" || toolName === "read_file") {
const p = typeof toolInput.path === "string" ? toolInput.path : "/etc/hosts";
@@ -512,6 +549,17 @@ class FakeAgentSession implements AgentSession {
if (toolName === "Bash" || toolName === "shell") {
const command = typeof toolInput.command === "string" ? toolInput.command : "";
// Deterministic file-create behavior for permission prompt tests:
// Prompt: Create a file named "X" with the content "Y"
if (createFileMatch) {
const fileName = createFileMatch[1] ?? "test.txt";
const content = createFileMatch[2] ?? "";
const dest = path.join(this.config.cwd ?? process.cwd(), fileName);
writeFileSync(dest, content);
return;
}
if (lower.includes("rm -f permission.txt") || command.includes("rm -f permission.txt")) {
const dest = path.join(this.config.cwd ?? process.cwd(), "permission.txt");
try {
@@ -660,9 +708,18 @@ class FakeAgentClient implements AgentClient {
}
async listModels(_options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
return [
{ provider: this.provider, id: "test-model", label: "Test Model", isDefault: true },
];
if (this.provider === "claude") {
return [
{ provider: this.provider, id: "haiku", label: "Haiku", isDefault: true },
{ provider: this.provider, id: "sonnet", label: "Sonnet", isDefault: false },
];
}
if (this.provider === "codex") {
return [
{ provider: this.provider, id: "gpt-5.1-codex-mini", label: "gpt-5.1-codex-mini", isDefault: true },
];
}
return [{ provider: this.provider, id: "test-model", label: "Test Model", isDefault: true }];
}
async isAvailable(): Promise<boolean> {