diff --git a/packages/server/src/server/agent/providers/claude/agent.image-rendering.test.ts b/packages/server/src/server/agent/providers/claude/agent.image-rendering.test.ts index 5a06fea7fd..a772e544d1 100644 --- a/packages/server/src/server/agent/providers/claude/agent.image-rendering.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.image-rendering.test.ts @@ -13,8 +13,8 @@ const MATERIALIZED_PNG_PATH_PATTERN = /paseo-attachments(?:-[^\\/]+)?[\\/](?:[^\\/]+[\\/])?[0-9a-f]{64}\.png$/; interface ClaudeImageTestSession { - translateMessageToEvents(message: SDKMessage): AgentStreamEvent[]; - convertHistoryEntry(entry: unknown): AgentTimelineItem[]; + translateMessageToEvents(message: SDKMessage): Promise; + convertHistoryEntry(entry: unknown): Promise; } async function createSession(): Promise { @@ -157,7 +157,7 @@ describe("Claude tool_result image rendering", () => { test("emits the image as assistant markdown and keeps base64 out of the live tool output", async () => { const session = await createSession(); - const events = session.translateMessageToEvents(imageToolResultUserMessage()); + const events = await session.translateMessageToEvents(imageToolResultUserMessage()); const timelineItems = events .filter((event) => event.type === "timeline") @@ -181,7 +181,7 @@ describe("Claude tool_result image rendering", () => { test("replays the image as assistant markdown through history conversion", async () => { const session = await createSession(); - const items = session.convertHistoryEntry(imageToolResultHistoryEntry()); + const items = await session.convertHistoryEntry(imageToolResultHistoryEntry()); const [imageMessage, ...extraImages] = imageMessages(items); expect(extraImages).toEqual([]); @@ -202,7 +202,7 @@ describe("Claude tool_result image rendering", () => { test("keeps base64 out of an errored tool_result that carries an image", async () => { const session = await createSession(); - const events = session.translateMessageToEvents(erroredImageToolResultUserMessage()); + const events = await session.translateMessageToEvents(erroredImageToolResultUserMessage()); const timelineItems = events .filter((event) => event.type === "timeline") @@ -226,7 +226,7 @@ describe("Claude tool_result image rendering", () => { test("emits one image message per image block in a multi-image tool_result", async () => { const session = await createSession(); - const events = session.translateMessageToEvents(multiImageToolResultUserMessage()); + const events = await session.translateMessageToEvents(multiImageToolResultUserMessage()); const timelineItems = events .filter((event) => event.type === "timeline") diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index 2b2c725269..52b70c25bc 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.test.ts @@ -32,7 +32,7 @@ import type { AgentAttachment } from "@getpaseo/protocol/messages"; import { buildAgentPrompt, renderPromptAttachmentAsText } from "../../prompt-attachments.js"; interface TestClaudeSession { - translateMessageToEvents(message: SDKMessage): AgentStreamEvent[]; + translateMessageToEvents(message: SDKMessage): Promise; close(): Promise; } @@ -53,7 +53,7 @@ afterEach(() => { }); describe("convertClaudeHistoryEntry", () => { - test("maps user tool results to timeline items", () => { + test("maps user tool results to timeline items", async () => { const toolUseId = "toolu_test"; const entry = { type: "user", @@ -79,14 +79,14 @@ describe("convertClaudeHistoryEntry", () => { ]; const mapBlocks = vi.fn().mockReturnValue(stubTimeline); - const result = convertClaudeHistoryEntry(entry, mapBlocks); + const result = await convertClaudeHistoryEntry(entry, mapBlocks); expect(result).toEqual(stubTimeline); expect(mapBlocks).toHaveBeenCalledTimes(1); expect(Array.isArray(mapBlocks.mock.calls[0][0])).toBe(true); }); - test("replays persisted Claude tool results as completed tool calls", () => { + test("replays persisted Claude tool results as completed tool calls", async () => { const entry = { type: "user", message: { @@ -124,12 +124,12 @@ describe("convertClaudeHistoryEntry", () => { const mapPersistedToolResultBlocks = (): AgentTimelineItem[] => completedToolCall; - expect(convertClaudeHistoryEntry(entry, mapPersistedToolResultBlocks)).toEqual( + expect(await convertClaudeHistoryEntry(entry, mapPersistedToolResultBlocks)).toEqual( completedToolCall, ); }); - test("returns user messages when no tool blocks exist", () => { + test("returns user messages when no tool blocks exist", async () => { const entry = { type: "user", message: { @@ -138,7 +138,7 @@ describe("convertClaudeHistoryEntry", () => { }, }; - expect(convertClaudeHistoryEntry(entry, () => [])).toEqual([ + expect(await convertClaudeHistoryEntry(entry, () => [])).toEqual([ { type: "user_message", text: "Run npm test", @@ -146,7 +146,7 @@ describe("convertClaudeHistoryEntry", () => { ]); }); - test("converts compact boundary metadata variants", () => { + test("converts compact boundary metadata variants", async () => { const fixtures = [ { entry: { @@ -175,7 +175,7 @@ describe("convertClaudeHistoryEntry", () => { ] as const; for (const fixture of fixtures) { - expect(convertClaudeHistoryEntry(fixture.entry, () => [])).toEqual([ + expect(await convertClaudeHistoryEntry(fixture.entry, () => [])).toEqual([ { type: "compaction", status: "completed", @@ -186,7 +186,7 @@ describe("convertClaudeHistoryEntry", () => { } }); - test("skips synthetic user entries", () => { + test("skips synthetic user entries", async () => { const entry = { type: "user", isSynthetic: true, @@ -197,13 +197,13 @@ describe("convertClaudeHistoryEntry", () => { }; const mapBlocks = vi.fn().mockReturnValue([]); - const result = convertClaudeHistoryEntry(entry, mapBlocks); + const result = await convertClaudeHistoryEntry(entry, mapBlocks); expect(result).toEqual([]); expect(mapBlocks).not.toHaveBeenCalled(); }); - test("skips meta user entries from Claude skill loading", () => { + test("skips meta user entries from Claude skill loading", async () => { const entry = { type: "user", isMeta: true, @@ -219,13 +219,13 @@ describe("convertClaudeHistoryEntry", () => { }; const mapBlocks = vi.fn().mockReturnValue([]); - const result = convertClaudeHistoryEntry(entry, mapBlocks); + const result = await convertClaudeHistoryEntry(entry, mapBlocks); expect(result).toEqual([]); expect(mapBlocks).not.toHaveBeenCalled(); }); - test("skips interrupt placeholder transcript noise", () => { + test("skips interrupt placeholder transcript noise", async () => { const interruptEntry = { type: "user", message: { @@ -246,11 +246,11 @@ describe("convertClaudeHistoryEntry", () => { .fn() .mockReturnValue([{ type: "assistant_message", text: "No response requested." }]); - expect(convertClaudeHistoryEntry(interruptEntry, mapBlocks)).toEqual([]); - expect(convertClaudeHistoryEntry(assistantNoiseEntry, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(interruptEntry, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(assistantNoiseEntry, mapBlocks)).toEqual([]); }); - test("skips messages (model switch, /context, etc.)", () => { + test("skips messages (model switch, /context, etc.)", async () => { // Real entries from Claude Code JSONL history files const modelSwitch = { type: "user", @@ -305,24 +305,24 @@ describe("convertClaudeHistoryEntry", () => { const mapBlocks = vi.fn().mockReturnValue([]); - expect(convertClaudeHistoryEntry(modelSwitch, mapBlocks)).toEqual([]); - expect(convertClaudeHistoryEntry(modelSwitchWithAnsi, mapBlocks)).toEqual([]); - expect(convertClaudeHistoryEntry(contextDump, mapBlocks)).toEqual([]); - expect(convertClaudeHistoryEntry(planMode, mapBlocks)).toEqual([]); - expect(convertClaudeHistoryEntry(goodbye, mapBlocks)).toEqual([]); - expect(convertClaudeHistoryEntry(empty, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(modelSwitch, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(modelSwitchWithAnsi, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(contextDump, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(planMode, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(goodbye, mapBlocks)).toEqual([]); + expect(await convertClaudeHistoryEntry(empty, mapBlocks)).toEqual([]); // Real user messages must NOT be filtered const realMessage = { type: "user", message: { role: "user", content: "fix the bug in auth.ts" }, }; - expect(convertClaudeHistoryEntry(realMessage, mapBlocks)).toEqual([ + expect(await convertClaudeHistoryEntry(realMessage, mapBlocks)).toEqual([ { type: "user_message", text: "fix the bug in auth.ts" }, ]); }); - test("maps task notifications to synthetic tool calls", () => { + test("maps task notifications to synthetic tool calls", async () => { const entry = { type: "system", subtype: "task_notification", @@ -333,7 +333,7 @@ describe("convertClaudeHistoryEntry", () => { output_file: "/tmp/bg-fail-1.txt", }; - expect(convertClaudeHistoryEntry(entry, () => [])).toEqual([ + expect(await convertClaudeHistoryEntry(entry, () => [])).toEqual([ { type: "tool_call", callId: "task_notification_task-note-system-1", @@ -357,7 +357,7 @@ describe("convertClaudeHistoryEntry", () => { ]); }); - test("maps queue-operation task notifications to synthetic tool calls", () => { + test("maps queue-operation task notifications to synthetic tool calls", async () => { const entry = { type: "queue-operation", operation: "enqueue", @@ -372,7 +372,7 @@ describe("convertClaudeHistoryEntry", () => { ].join("\n"), }; - expect(convertClaudeHistoryEntry(entry, () => [])).toEqual([ + expect(await convertClaudeHistoryEntry(entry, () => [])).toEqual([ { type: "tool_call", callId: "task_notification_task-note-queue-1", @@ -396,7 +396,7 @@ describe("convertClaudeHistoryEntry", () => { ]); }); - test("passes assistant content blocks through to the mapper", () => { + test("passes assistant content blocks through to the mapper", async () => { const entry = { type: "assistant", message: { @@ -414,7 +414,7 @@ describe("convertClaudeHistoryEntry", () => { ]; const mapBlocks = vi.fn().mockReturnValue(mappedTimeline); - expect(convertClaudeHistoryEntry(entry, mapBlocks)).toEqual(mappedTimeline); + expect(await convertClaudeHistoryEntry(entry, mapBlocks)).toEqual(mappedTimeline); expect(mapBlocks).toHaveBeenCalledWith(entry.message.content); }); }); @@ -1208,7 +1208,7 @@ describe("ClaudeAgentSession features", () => { input: Record, options: Record, ): Promise; - translateMessageToEvents(message: SDKMessage): AgentStreamEvent[]; + translateMessageToEvents(message: SDKMessage): Promise; }; try { @@ -1248,7 +1248,7 @@ describe("ClaudeAgentSession features", () => { ), ).resolves.toMatchObject({ behavior: "deny", interrupt: undefined }); - internal.translateMessageToEvents({ + await internal.translateMessageToEvents({ type: "command_lifecycle", command_uuid: steerUuid, state: "started", @@ -2395,7 +2395,7 @@ describe("ClaudeAgentSession context window usage", () => { test("emits canonical task snapshots from Claude TaskCreate results", async () => { const session = await createSessionForTest(); try { - session.translateMessageToEvents({ + await session.translateMessageToEvents({ type: "assistant", message: { content: [ @@ -2409,7 +2409,7 @@ describe("ClaudeAgentSession context window usage", () => { }, } as unknown as SDKMessage); - const events = session.translateMessageToEvents({ + const events = await session.translateMessageToEvents({ type: "user", message: { content: [{ type: "tool_result", tool_use_id: "task-create-1", content: "created" }], @@ -3611,9 +3611,9 @@ describe("ClaudeAgentSession context window usage", () => { ]); try { - const compactEvents = (session as unknown as TestClaudeSession).translateMessageToEvents( - createCompactBoundary(), - ); + const compactEvents = await ( + session as unknown as TestClaudeSession + ).translateMessageToEvents(createCompactBoundary()); expect(compactEvents).toContainEqual( expect.objectContaining({ type: "usage_updated", @@ -3799,7 +3799,7 @@ describe("ClaudeAgentSession context window usage", () => { test("result.result is surfaced as an assistant message when no model output was produced", async () => { const session = await createSessionForTest(); - const events = session.translateMessageToEvents({ + const events = await session.translateMessageToEvents({ type: "result", subtype: "success", result: "Unknown command: /foo-doesnt-exist", @@ -3834,7 +3834,7 @@ describe("ClaudeAgentSession context window usage", () => { test("result.result is not duplicated when the model produced output during the turn", async () => { const session = await createSessionForTest(); - const events = session.translateMessageToEvents({ + const events = await session.translateMessageToEvents({ type: "result", subtype: "success", result: "Here is the answer.", diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index f1b259b5d5..f166bb5dee 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -2546,7 +2546,11 @@ class ClaudeAgentSession implements AgentSession { this.input.push(sdkMessage); setTimeout(() => { if (this.activeForegroundTurnId === turnId) { - this.emitSubmittedUserMessage(sdkMessage, turnId, options?.clientMessageId); + void this.emitSubmittedUserMessage(sdkMessage, turnId, options?.clientMessageId).catch( + (error: unknown) => { + this.logger.warn({ err: error }, "Failed to emit submitted Claude message"); + }, + ); } }, 0); } catch (error) { @@ -4133,7 +4137,7 @@ class ClaudeAgentSession implements AgentSession { consecutiveInterruptAbortRecoveries = 0; if (await this.handleMissingResumedConversation(message, activeQuery)) return true; if (!this.ownsQuery(activeQuery)) return true; - this.routeSdkMessageFromPump(message); + await this.routeSdkMessageFromPump(message); return false; } finally { this.processingSdkMessages--; @@ -4257,7 +4261,7 @@ class ClaudeAgentSession implements AgentSession { return !this.closed && this.query === query; } - private routeSdkMessageFromPump(message: SDKMessage): void { + private async routeSdkMessageFromPump(message: SDKMessage): Promise { if (this.consumeSessionStateChange(message)) { return; } @@ -4291,7 +4295,7 @@ class ClaudeAgentSession implements AgentSession { "provider.claude.parsed_event", ); - const events = this.buildPumpedMessageEvents(message, identifiers.messageId, turnId); + const events = await this.buildPumpedMessageEvents(message, identifiers.messageId, turnId); if (events.length === 0) { return; @@ -4326,15 +4330,22 @@ class ClaudeAgentSession implements AgentSession { this.dispatchEvents(events); } - private buildPumpedMessageEvents( + private async buildPumpedMessageEvents( message: SDKMessage, messageIdHint: string | null, turnId: string | null, - ): AgentStreamEvent[] { - const messageEvents = this.translateMessageToEvents(message, { + ): Promise { + const query = this.query; + const messageEvents = await this.translateMessageToEvents(message, { suppressAssistantText: true, suppressReasoning: true, }); + if ( + this.closed || + this.query !== query || + (this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? null) !== turnId + ) + return []; const assistantTimelineEvents = readClaudeParentToolUseId(message) ? [] : this.timelineAssembler @@ -4479,13 +4490,13 @@ class ClaudeAgentSession implements AgentSession { return withdrewAll; } - private translateMessageToEvents( + private async translateMessageToEvents( message: SDKMessage, options?: { suppressAssistantText?: boolean; suppressReasoning?: boolean; }, - ): AgentStreamEvent[] { + ): Promise { const parentToolUseId = readClaudeParentToolUseId(message); if (parentToolUseId) { return this.translateSidechainFrameToEvents(message, parentToolUseId); @@ -4533,11 +4544,11 @@ class ClaudeAgentSession implements AgentSession { this.appendSystemMessageEvents(message, events); break; case "user": - this.appendUserMessageEvents(message, events); + await this.appendUserMessageEvents(message, events); this.appendSidechainResultEvents(message, events); break; case "assistant": { - const timelineItems = this.mapBlocksToTimeline(message.message.content, { + const timelineItems = await this.mapBlocksToTimeline(message.message.content, { suppressAssistantText: options?.suppressAssistantText ?? false, suppressReasoning: options?.suppressReasoning ?? false, }); @@ -4548,7 +4559,7 @@ class ClaudeAgentSession implements AgentSession { break; } case "stream_event": - this.appendStreamEventEvents(message, events, options); + await this.appendStreamEventEvents(message, events, options); break; case "result": this.appendResultEvents(message, events); @@ -4650,13 +4661,14 @@ class ClaudeAgentSession implements AgentSession { } } - private emitSubmittedUserMessage( + private async emitSubmittedUserMessage( message: Extract, turnId: string, clientMessageId?: string, - ): void { + ): Promise { const events: AgentStreamEvent[] = []; - this.appendUserMessageEvents(message, events); + await this.appendUserMessageEvents(message, events); + if (this.activeForegroundTurnId !== turnId) return; if (events.length === 0) { return; } @@ -4786,10 +4798,10 @@ class ClaudeAgentSession implements AgentSession { } } - private appendUserMessageEvents( + private async appendUserMessageEvents( message: Extract, events: AgentStreamEvent[], - ): void { + ): Promise { if (isSyntheticUserEntry(message)) { return; } @@ -4828,7 +4840,7 @@ class ClaudeAgentSession implements AgentSession { return; } if (Array.isArray(content)) { - this.appendUserContentArrayEvents(content, messageId, events); + await this.appendUserContentArrayEvents(content, messageId, events); } } @@ -4854,12 +4866,12 @@ class ClaudeAgentSession implements AgentSession { events.push({ type: "timeline", item, provider: "claude" }); } - private appendUserContentArrayEvents( + private async appendUserContentArrayEvents( content: ReadonlyArray, messageId: string | undefined, events: AgentStreamEvent[], - ): void { - const timelineItems = this.mapBlocksToTimeline(content, { + ): Promise { + const timelineItems = await this.mapBlocksToTimeline(content, { textMessageType: "user_message", }); for (const item of timelineItems) { @@ -4875,16 +4887,16 @@ class ClaudeAgentSession implements AgentSession { } } - private appendStreamEventEvents( + private async appendStreamEventEvents( message: Extract, events: AgentStreamEvent[], options: { suppressAssistantText?: boolean; suppressReasoning?: boolean } | undefined, - ): void { + ): Promise { const usageUpdatedEvent = this.contextUsage.buildStreamUsageEvent(message.event); if (usageUpdatedEvent) { events.push(usageUpdatedEvent); } - const timelineItems = this.mapPartialEvent(message.event, { + const timelineItems = await this.mapPartialEvent(message.event, { suppressAssistantText: options?.suppressAssistantText ?? false, suppressReasoning: options?.suppressReasoning ?? false, }); @@ -5350,7 +5362,7 @@ class ClaudeAgentSession implements AgentSession { ): Promise { const timeline: PersistedTimelineEntry[] = []; for await (const entry of entries) { - this.ingestPersistedHistoryEntry(entry, timeline, replay); + await this.ingestPersistedHistoryEntry(entry, timeline, replay); } if (timeline.length > 0) { @@ -5410,11 +5422,11 @@ class ClaudeAgentSession implements AgentSession { return replay; } - private ingestPersistedHistoryEntry( + private async ingestPersistedHistoryEntry( entry: ClaudeHistoryEntry, timeline: PersistedTimelineEntry[], replay: ClaudeReplayOwnership, - ): void { + ): Promise { if (entry.isSidechain) { return; } @@ -5428,7 +5440,7 @@ class ClaudeAgentSession implements AgentSession { ? replay.toolOwners.get(notificationToolUseId) : undefined; if (notificationOwner) { - for (const item of this.convertHistoryEntry(entry)) { + for (const item of await this.convertHistoryEntry(entry)) { this.persistedProviderSubagentEvents.push({ type: "provider_subagent", provider: "claude", @@ -5445,7 +5457,7 @@ class ClaudeAgentSession implements AgentSession { const taskSnapshot = this.taskState.observe(entry); const items = [ ...(taskSnapshot ? [taskSnapshot] : []), - ...this.convertHistoryEntry(entry).map((item) => + ...(await this.convertHistoryEntry(entry)).map((item) => labelReplayedSubagentCard(item, replay.subagentToolCalls), ), ]; @@ -5482,10 +5494,10 @@ class ClaudeAgentSession implements AgentSession { }); } - private convertHistoryEntry(entry: ClaudeHistoryEntry): AgentTimelineItem[] { - return convertClaudeHistoryEntry(entry, (content) => this.mapBlocksToTimeline(content)).map( - limitAgentTimelineItemContent, - ); + private async convertHistoryEntry(entry: ClaudeHistoryEntry): Promise { + return ( + await convertClaudeHistoryEntry(entry, (content) => this.mapBlocksToTimeline(content)) + ).map(limitAgentTimelineItemContent); } // Maps Claude content blocks into AgentTimelineItems. @@ -5500,14 +5512,14 @@ class ClaudeAgentSession implements AgentSession { // // NOTE: convertClaudeHistoryEntry uses extractUserMessageText directly instead of this function // for user entries. Both paths must produce equivalent user_message items. - private mapBlocksToTimeline( + private async mapBlocksToTimeline( content: string | ReadonlyArray, options?: { textMessageType?: "assistant_message" | "user_message"; suppressAssistantText?: boolean; suppressReasoning?: boolean; }, - ): AgentTimelineItem[] { + ): Promise { const textMessageType = options?.textMessageType ?? "assistant_message"; const suppressText = textMessageType === "assistant_message" && (options?.suppressAssistantText ?? false); @@ -5534,7 +5546,7 @@ class ClaudeAgentSession implements AgentSession { if (!isClaudeContentChunk(block) || this.mapSubagentHandbackBlock(block, items)) { continue; } - this.mapBlockToTimeline(block, { + await this.mapBlockToTimeline(block, { items, userTextParts, textMessageType, @@ -5579,7 +5591,7 @@ class ClaudeAgentSession implements AgentSession { } } - private mapBlockToTimeline( + private async mapBlockToTimeline( block: ClaudeContentChunk, context: { items: AgentTimelineItem[]; @@ -5588,7 +5600,7 @@ class ClaudeAgentSession implements AgentSession { suppressText: boolean; suppressReasoning: boolean; }, - ): void { + ): Promise { switch (block.type) { case "text": case "text_delta": @@ -5612,7 +5624,7 @@ class ClaudeAgentSession implements AgentSession { case "code_execution_tool_result": case "bash_code_execution_tool_result": case "text_editor_code_execution_tool_result": - this.handleToolResult(block, context.items); + await this.handleToolResult(block, context.items); break; default: break; @@ -5658,7 +5670,10 @@ class ClaudeAgentSession implements AgentSession { ); } - private handleToolResult(block: ClaudeContentChunk, items: AgentTimelineItem[]): void { + private async handleToolResult( + block: ClaudeContentChunk, + items: AgentTimelineItem[], + ): Promise { const entry = typeof block.tool_use_id === "string" ? this.toolUseCache.get(block.tool_use_id) : undefined; const blockToolName = typeof block.tool_name === "string" ? block.tool_name : undefined; @@ -5691,7 +5706,7 @@ class ClaudeAgentSession implements AgentSession { this.rememberSettledSubagentToolCall(settledToolCall, entry); for (const image of images) { - const imageItem = renderProviderImageOutputAsAssistantMarkdown(image, { + const imageItem = await renderProviderImageOutputAsAssistantMarkdown(image, { materialize: materializeProviderImage, }); if (imageItem) { @@ -5915,13 +5930,13 @@ class ClaudeAgentSession implements AgentSession { return false; } - private mapPartialEvent( + private async mapPartialEvent( event: SDKPartialAssistantMessage["event"], options?: { suppressAssistantText?: boolean; suppressReasoning?: boolean; }, - ): AgentTimelineItem[] { + ): Promise { if (this.updatePartialEventToolState(event)) { return []; } @@ -5929,14 +5944,14 @@ class ClaudeAgentSession implements AgentSession { switch (event.type) { case "content_block_start": return isClaudeContentChunk(event.content_block) - ? this.mapBlocksToTimeline([event.content_block], { + ? await this.mapBlocksToTimeline([event.content_block], { suppressAssistantText: options?.suppressAssistantText, suppressReasoning: options?.suppressReasoning, }) : []; case "content_block_delta": return isClaudeContentChunk(event.delta) - ? this.mapBlocksToTimeline([event.delta], { + ? await this.mapBlocksToTimeline([event.delta], { suppressAssistantText: options?.suppressAssistantText, suppressReasoning: options?.suppressReasoning, }) @@ -6299,12 +6314,14 @@ interface ClaudeHistoryEntry { [key: string]: unknown; } -function mapAssistantHistoryBlocksWithMessageId( +async function mapAssistantHistoryBlocksWithMessageId( entry: ClaudeHistoryEntry, content: string | ClaudeContentChunk[], - mapBlocks: (content: string | ClaudeContentChunk[]) => AgentTimelineItem[], -): AgentTimelineItem[] { - const items = mapBlocks(content); + mapBlocks: ( + content: string | ClaudeContentChunk[], + ) => AgentTimelineItem[] | Promise, +): Promise { + const items = await mapBlocks(content); const assistantMessageId = typeof entry.uuid === "string" && entry.uuid.length > 0 ? entry.uuid : null; if (!assistantMessageId) { @@ -6372,10 +6389,12 @@ function isProviderImageMessage(item: AgentTimelineItem): boolean { return item.type === "assistant_message" && isProviderImageMarkdown(item.text); } -export function convertClaudeHistoryEntry( +export async function convertClaudeHistoryEntry( entry: ClaudeHistoryEntry, - mapBlocks: (content: string | ClaudeContentChunk[]) => AgentTimelineItem[], -): AgentTimelineItem[] { + mapBlocks: ( + content: string | ClaudeContentChunk[], + ) => AgentTimelineItem[] | Promise, +): Promise { const preamble = convertClaudeHistoryEntryPreamble(entry); if ("shortCircuit" in preamble) { return preamble.shortCircuit; @@ -6414,7 +6433,7 @@ export function convertClaudeHistoryEntry( } if (hasToolBlock && normalizedBlocks) { - const mapped = mapBlocks(normalizedBlocks); + const mapped = await mapBlocks(normalizedBlocks); if (entry.type === "user") { // tool_result handling (handleToolResult) emits image markdown as an assistant_message // alongside the tool_call. User-entry text blocks also map to assistant_message in this path @@ -6428,7 +6447,7 @@ export function convertClaudeHistoryEntry( } if (entry.type === "assistant" && contentValue) { - return mapAssistantHistoryBlocksWithMessageId(entry, contentValue, mapBlocks); + return await mapAssistantHistoryBlocksWithMessageId(entry, contentValue, mapBlocks); } return timeline; diff --git a/packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts b/packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts index 413f39d427..f5bbf52422 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts @@ -1,3 +1,9 @@ +import { readFile } from "node:fs/promises"; +import { fileURLToPath } from "node:url"; +import { + materializeProviderImage, + renderProviderImageOutputAsAssistantMarkdown, +} from "../../provider-image-output.js"; import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; import { describe, expect, it } from "vitest"; @@ -660,3 +666,34 @@ describe("live and replay agree", () => { }); }); }); + +it("awaits image materialization before publishing replay observations", async () => { + const bytes = Buffer.from("async replay image fixture"); + const convertEntry = async () => { + const item = await renderProviderImageOutputAsAssistantMarkdown( + { data: bytes.toString("base64"), mimeType: "image/png" }, + { materialize: materializeProviderImage }, + ); + return item ? [item] : []; + }; + const { observations } = await observeReplaySubagents({ + parent: parentWithTaskCall(), + subagents: [ + { + agentId: AGENT_ID, + meta: { toolUseId: TOOL_USE_ID }, + entries: [{ type: "assistant", message: { content: [] } }], + }, + ], + convertEntry, + }); + const image = observations.find( + (observation) => + observation.kind === "timeline" && observation.item.type === "assistant_message", + ); + if (!image || image.kind !== "timeline" || image.item.type !== "assistant_message") + throw new Error("Expected replay image"); + const source = image.item.text.match(/!\[Image\]\(([^)]+)\)/)?.[1]; + expect(source).toBeTruthy(); + expect(await readFile(fileURLToPath(source!))).toEqual(bytes); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts b/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts index 567a98dfe5..2a3ca74846 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts @@ -253,7 +253,7 @@ function observeSubtitle( async function observeSubagent( subagent: ClaudeReplaySubagentInput, parent: ClaudeReplayParentFacts, - convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[], + convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise, toolOwners: Map, parentSubagentId?: string, ): Promise { @@ -280,7 +280,7 @@ async function observeSubagent( childTerminalStatus = entry.message?.stop_reason === "end_turn" ? "completed" : null; } recordReplayToolOwners(toolOwners, entry, link.id); - for (const item of convertEntry(entry)) { + for (const item of await convertEntry(entry)) { timeline.push({ kind: "timeline", id: link.id, item, ...(timestamp ? { timestamp } : {}) }); } } @@ -323,7 +323,7 @@ function recordReplayToolOwners( export async function observeReplaySubagents(input: { subagents: readonly ClaudeReplaySubagentInput[]; parent: ClaudeReplayParentFacts; - convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; + convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise; }): Promise<{ observations: SubagentObservation[]; toolOwners: ReadonlyMap }> { const observations: SubagentObservation[] = []; const toolOwners = new Map(); diff --git a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts index af03a515b7..edb673c664 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.test.ts @@ -1,3 +1,9 @@ +import { readFile } from "node:fs/promises"; +import { fileURLToPath } from "node:url"; +import { + materializeProviderImage, + renderProviderImageOutputAsAssistantMarkdown, +} from "../../provider-image-output.js"; import { describe, expect, it } from "vitest"; import { ProviderSubagentStore } from "../../../provider-subagents/store.js"; @@ -195,3 +201,29 @@ describe("Claude workflow replay", () => { }, ); }); + +it("awaits image materialization before publishing replay observations", async () => { + const bytes = Buffer.from("async replay image fixture"); + const convertEntry = async () => { + const item = await renderProviderImageOutputAsAssistantMarkdown( + { data: bytes.toString("base64"), mimeType: "image/png" }, + { materialize: materializeProviderImage }, + ); + return item ? [item] : []; + }; + const observations = await observeReplayWorkflows({ + workflows: [{ runId: RUN_ID, status: "completed" }], + parentEntries: parentEntries(), + entriesByRunId: new Map([[RUN_ID, [{ type: "assistant", message: { content: [] } }]]]), + convertEntry, + }); + const image = observations.find( + (observation) => + observation.kind === "timeline" && observation.item.type === "assistant_message", + ); + if (!image || image.kind !== "timeline" || image.item.type !== "assistant_message") + throw new Error("Expected replay image"); + const source = image.item.text.match(/!\[Image\]\(([^)]+)\)/)?.[1]; + expect(source).toBeTruthy(); + expect(await readFile(fileURLToPath(source!))).toEqual(bytes); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts index 7496ce7c8d..17a278d00f 100644 --- a/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts +++ b/packages/server/src/server/agent/providers/claude/subagents/workflow-replay-source.ts @@ -53,7 +53,7 @@ export async function observeReplayWorkflows(input: { string, Iterable | AsyncIterable >; - convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; + convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise; }): Promise { const toolCallIdByRunId = await readWorkflowLinks(input.parentEntries); const observations: SubagentObservation[] = []; @@ -122,7 +122,7 @@ async function observeWorkflowTimeline(input: { id: string; result: unknown; entries: Iterable | AsyncIterable; - convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; + convertEntry?: (entry: ClaudeReplayEntry) => AgentTimelineItem[] | Promise; finishedAt?: string; }): Promise { const observations: SubagentObservation[] = []; @@ -130,7 +130,7 @@ async function observeWorkflowTimeline(input: { for await (const entry of input.entries) { const timestamp = normalizeProviderReplayTimestamp(entry.timestamp); - const items = input.convertEntry?.(entry) ?? []; + const items = (await input.convertEntry?.(entry)) ?? []; for (const item of items) { if (item.type === "assistant_message") replayedAssistantText.add(item.text.trim()); observations.push({ diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 8c61635d6f..1d113fb27a 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -113,7 +113,7 @@ interface CodexSessionTestAccess { }; ensureThreadLoaded(): Promise; handleToolApprovalRequest(params: unknown): Promise; - handleNotification(method: string, params: unknown): void; + handleNotification(method: string, params: unknown): Promise; loadPersistedHistory(client: CodexClientLike | null): Promise; refreshResolvedCollaborationMode(): void; serviceTier: "fast" | null; @@ -212,6 +212,7 @@ async function startPublicSteeringSession( const started = await session.startTurn("first"); await appServer.waitForTurnStart(); appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" }); + await new Promise((resolve) => setImmediate(resolve)); await new Promise((resolve) => setTimeout(resolve, 0)); return { session, paseoTurnId: started.turnId }; } @@ -354,9 +355,11 @@ describe("Codex active-turn steering admission", () => { }); await resolverEntered.promise; appServer.completeTurn({ threadId: "thread-1" }); + await new Promise((resolve) => setImmediate(resolve)); await new Promise((resolve) => setTimeout(resolve, 0)); const startedB = await session.startTurn("second"); appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" }); + await new Promise((resolve) => setImmediate(resolve)); await new Promise((resolve) => setTimeout(resolve, 0)); commandResolution.resolve(null); @@ -478,6 +481,7 @@ async function startCompactionTurnTest(): Promise<{ await session.startTurn("exercise compaction lifecycle"); appServer.startsTurn({ threadId: "thread-1", turnId: "codex-turn-1" }); + await new Promise((resolve) => setImmediate(resolve)); return { appServer, session, events, terminalEvent }; } @@ -1460,7 +1464,9 @@ describe("Codex app-server provider", () => { command: "gh pr merge 2030 --squash", cwd: "/workspace/project", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "thread-1", text: "Merged." }); + await new Promise((resolve) => setImmediate(resolve)); await expect(nextTimelineItem).resolves.toEqual({ type: "timeline", @@ -1504,6 +1510,7 @@ describe("Codex app-server provider", () => { command: "gh pr merge 2030 --squash", cwd: "/workspace/project", }); + await new Promise((resolve) => setImmediate(resolve)); await expect(nextTimelineItem).resolves.toEqual({ type: "timeline", @@ -1548,6 +1555,7 @@ describe("Codex app-server provider", () => { processId: "4242", text: "gh pr merge 2030 --squash\n", }); + await new Promise((resolve) => setImmediate(resolve)); await expect(nextTimelineItem).resolves.toEqual({ type: "timeline", @@ -1576,6 +1584,7 @@ describe("Codex app-server provider", () => { command: "sleep 30", output: "Process running with session id 4242", }); + await new Promise((resolve) => setImmediate(resolve)); await expect(relabeledTerminal).resolves.toEqual({ type: "timeline", @@ -1623,6 +1632,7 @@ describe("Codex app-server provider", () => { processId: "73", command, }); + await new Promise((resolve) => setImmediate(resolve)); const terminalWrite = waitForTimelineToolCall(session, "terminal-session-73-1"); appServer.typesIntoTerminal({ @@ -1632,6 +1642,7 @@ describe("Codex app-server provider", () => { processId: "73", text: "print(1)\n", }); + await new Promise((resolve) => setImmediate(resolve)); await expect(terminalWrite).resolves.toMatchObject({ item: { @@ -1668,7 +1679,9 @@ describe("Codex app-server provider", () => { processId: "4242", text: "git status\n", }); + await new Promise((resolve) => setImmediate(resolve)); + await firstTimelineItem; const secondTimelineItem = waitForNextTimelineItem(session); appServer.typesIntoTerminal({ threadId: "thread-1", @@ -1677,6 +1690,7 @@ describe("Codex app-server provider", () => { processId: "4242", text: "git push\n", }); + await new Promise((resolve) => setImmediate(resolve)); const [first, second] = await Promise.all([firstTimelineItem, secondTimelineItem]); expect(first.item).toMatchObject({ @@ -2075,9 +2089,13 @@ describe("Codex app-server provider", () => { await session.startTurn("remember first"); emitCodexUserMessage(appServer, { id: "codex-first", text: "remember first" }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.startTurn("remember second"); emitCodexUserMessage(appServer, { id: "codex-second", text: "remember second" }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.revertConversation({ messageId: "codex-first" }); @@ -2116,6 +2134,8 @@ describe("Codex app-server provider", () => { turnId: "turn-first", }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.startTurn("remember second"); emitCodexUserMessage(appServer, { id: "codex-second", @@ -2123,6 +2143,8 @@ describe("Codex app-server provider", () => { turnId: "turn-second", }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.revertConversation({ messageId: "codex-second" }); @@ -2172,6 +2194,8 @@ describe("Codex app-server provider", () => { turnId: "turn-first", }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.startTurn("remember second"); emitCodexUserMessage(appServer, { id: "codex-second", @@ -2179,6 +2203,8 @@ describe("Codex app-server provider", () => { turnId: "turn-second", }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.revertConversation({ messageId: "codex-first" }); @@ -2249,6 +2275,8 @@ describe("Codex app-server provider", () => { turnId: "turn-first", }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.startTurn("remember second"); emitCodexUserMessage(appServer, { id: "codex-second", @@ -2256,6 +2284,8 @@ describe("Codex app-server provider", () => { turnId: "turn-second", }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.revertConversation({ messageId: "codex-first" }); @@ -2290,7 +2320,7 @@ describe("Codex app-server provider", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); await session.startTurn("remember this", { clientMessageId: "client-message" }); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { threadId: "thread-1", turn: { id: "turn-1" }, }); @@ -2306,6 +2336,8 @@ describe("Codex app-server provider", () => { }); expect(events.slice(0, 2).map((event) => event.type)).toEqual(["turn_started", "timeline"]); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); await session.close(); }); @@ -3181,12 +3213,12 @@ describe("Codex app-server provider", () => { ]); }); - test("converts Codex collab agent notifications through the normal timeline path", () => { + test("converts Codex collab agent notifications through the normal timeline path", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/started", { + await asInternals(session).handleNotification("item/started", { threadId: "test-thread", item: { type: "collabAgentToolCall", @@ -3222,12 +3254,12 @@ describe("Codex app-server provider", () => { ]); }); - test("folds child-thread Codex activity into the parent sub-agent tool call", () => { + test("folds child-thread Codex activity into the parent sub-agent tool call", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "collabAgentToolCall", @@ -3241,12 +3273,12 @@ describe("Codex app-server provider", () => { }, }, }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { threadId: "child-thread-1", itemId: "child-message-1", delta: "Found", }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "child-thread-1", item: { type: "agentMessage", @@ -3254,7 +3286,7 @@ describe("Codex app-server provider", () => { text: "Found the path.", }, }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { threadId: "child-thread-1", turn: { status: "completed" }, }); @@ -3332,12 +3364,15 @@ describe("Codex app-server provider", () => { threadId: "settled-child-thread", agentPath: "/root/settled-child", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn({ threadId: "settled-child-thread" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "settled-child-thread", itemId: "late-child-message", text: "Late trailing output.", }); + await new Promise((resolve) => setImmediate(resolve)); const providerUpserts = events.flatMap((event) => event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [], @@ -3361,6 +3396,7 @@ describe("Codex app-server provider", () => { agentPath: "/root/settled-child", kind: "interacted", }); + await new Promise((resolve) => setImmediate(resolve)); expect( events.findLast( (event) => event.type === "provider_subagent" && event.event.type === "upsert", @@ -3371,6 +3407,7 @@ describe("Codex app-server provider", () => { threadId: "settled-child-thread", turnId: "next-child-turn", }); + await new Promise((resolve) => setImmediate(resolve)); expect( events.findLast( (event) => event.type === "provider_subagent" && event.event.type === "upsert", @@ -3378,7 +3415,9 @@ describe("Codex app-server provider", () => { ).toMatchObject({ event: { id: "settled-child-thread", status: "running" } }); appServer.completeTurn({ threadId: "settled-child-thread" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await resultPromise; appServer.assertNoErrors(); } finally { @@ -3386,12 +3425,12 @@ describe("Codex app-server provider", () => { } }); - test("updates a registered child with its later native activity name", () => { + test("updates a registered child with its later native activity name", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "collabAgentToolCall", @@ -3405,7 +3444,7 @@ describe("Codex app-server provider", () => { }, }, }); - asInternals(session).handleNotification("item/started", { + await asInternals(session).handleNotification("item/started", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -3440,11 +3479,11 @@ describe("Codex app-server provider", () => { }); }); - test("renders child MCP image results in the provider subagent timeline", () => { + test("renders child MCP image results in the provider subagent timeline", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -3455,7 +3494,7 @@ describe("Codex app-server provider", () => { }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "image-child-thread", item: { id: "child-mcp-image", @@ -3488,11 +3527,11 @@ describe("Codex app-server provider", () => { rmSync(source, { force: true }); }); - test("renders a child user message once across lifecycle notifications", () => { + test("renders a child user message once across lifecycle notifications", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -3508,11 +3547,11 @@ describe("Codex app-server provider", () => { content: [{ type: "text", text: "Inspect this path." }], }; - asInternals(session).handleNotification("item/started", { + await asInternals(session).handleNotification("item/started", { threadId: "user-child-thread", item: childUserMessage, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "user-child-thread", item: childUserMessage, }); @@ -3546,19 +3585,24 @@ describe("Codex app-server provider", () => { threadId: "child-thread-1", agentPath: "/root/child", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "child-thread-1", itemId: "child-message-1", text: "Child findings.", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn({ threadId: "child-thread-1" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "thread-1", itemId: "parent-message-1", text: "Parent report.", chunks: ["Parent ", "report."], }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; expect(result.finalText).toBe("Parent report."); @@ -3592,13 +3636,16 @@ describe("Codex app-server provider", () => { itemId: "first-parent-message", text: "First report.", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "thread-1", itemId: "second-parent-message", text: "Second report.", chunks: ["", "Second report."], }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; expect(result.finalText).toBe("Second report."); @@ -3623,8 +3670,11 @@ describe("Codex app-server provider", () => { await appServer.waitForTurnStart(); appServer.says({ threadId: "thread-1", text: "First report." }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "thread-1", text: "Second report." }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; expect(result.finalText).toBe("Second report."); @@ -3648,23 +3698,29 @@ describe("Codex app-server provider", () => { await appServer.waitForTurnStart(); appServer.startsTurn({ threadId: "child-thread-early" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "child-thread-early", itemId: "child-message-early", text: "Early child findings.", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn({ threadId: "child-thread-early" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.startsSubAgent({ callId: "spawn-child-early", threadId: "child-thread-early", agentPath: "/root/early-child", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "thread-1", itemId: "parent-message-after-early-child", text: "Parent report after replay.", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; expect(result.finalText).toBe("Parent report after replay."); @@ -3708,36 +3764,43 @@ describe("Codex app-server provider", () => { threadId: "child-thread-stable", agentPath: "/root/stable-child", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.beginsSubAgentActivity({ callId: "message-child-stable", threadId: "child-thread-stable", agentPath: "/root/stable-child", kind: "interacted", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completesSubAgentActivity({ callId: "message-child-stable", threadId: "child-thread-stable", agentPath: "/root/stable-child", kind: "interacted", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.says({ threadId: "child-thread-stable", itemId: "stable-child-message", text: "Still on the same card.", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.beginsSubAgentActivity({ callId: "interrupt-child-stable", threadId: "child-thread-stable", agentPath: "/root/stable-child", kind: "interrupted", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completesSubAgentActivity({ callId: "interrupt-child-stable", threadId: "child-thread-stable", agentPath: "/root/stable-child", kind: "interrupted", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); @@ -3772,6 +3835,7 @@ describe("Codex app-server provider", () => { await appServer.waitForTurnStart(); appServer.completeTurn({ threadId: "child-thread-fast" }); + await new Promise((resolve) => setImmediate(resolve)); const activity = { callId: "spawn-child-fast", threadId: "child-thread-fast", @@ -3779,8 +3843,11 @@ describe("Codex app-server provider", () => { kind: "started" as const, }; appServer.beginsSubAgentActivity(activity); + await new Promise((resolve) => setImmediate(resolve)); appServer.completesSubAgentActivity(activity); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); @@ -3809,16 +3876,20 @@ describe("Codex app-server provider", () => { await appServer.waitForTurnStart(); appServer.completeTurn({ threadId: "child-late-compaction" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completesCompaction({ threadId: "child-late-compaction", itemId: "late-child-compaction", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.startsSubAgent({ callId: "spawn-child-late-compaction", threadId: "child-late-compaction", agentPath: "/root/late-compaction", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); @@ -3852,6 +3923,7 @@ describe("Codex app-server provider", () => { threadId: "legacy-tool-child", agentPath: "/root/legacy-tool-child", }); + await new Promise((resolve) => setImmediate(resolve)); const command = { threadId: "legacy-tool-child", callId: "legacy-child-command", @@ -3859,15 +3931,20 @@ describe("Codex app-server provider", () => { output: "child output", }; appServer.runsLegacyCommand(command); + await new Promise((resolve) => setImmediate(resolve)); appServer.completesCommand(command); + await new Promise((resolve) => setImmediate(resolve)); appServer.appliesLegacyPatch({ threadId: "legacy-tool-child", callId: "legacy-child-patch", path: "/workspace/project/src/child.ts", diff: "@@\n-old\n+new\n", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn({ threadId: "legacy-tool-child" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); const result = await resultPromise; const toolCalls = result.timeline.filter((item) => item.type === "tool_call"); @@ -3890,12 +3967,12 @@ describe("Codex app-server provider", () => { } }); - test("keeps nested MultiAgentV2 output inside the root sub-agent card", () => { + test("keeps nested MultiAgentV2 output inside the root sub-agent card", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -3905,7 +3982,7 @@ describe("Codex app-server provider", () => { agentPath: "/root/child", }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "child-thread-root", item: { type: "subAgentActivity", @@ -3915,12 +3992,12 @@ describe("Codex app-server provider", () => { agentPath: "/root/child/grandchild", }, }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { threadId: "grandchild-thread", itemId: "grandchild-message", delta: "Grandchild findings.", }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { threadId: "grandchild-thread", turn: { status: "completed" }, }); @@ -3953,7 +4030,7 @@ describe("Codex app-server provider", () => { detail: { type: "sub_agent", log: expect.stringContaining("Grandchild findings.") }, }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { threadId: "child-thread-root", turn: { status: "completed" }, }); @@ -3963,12 +4040,12 @@ describe("Codex app-server provider", () => { }); }); - test("keeps a nested child under its spawning parent after a root interaction", () => { + test("keeps a nested child under its spawning parent after a root interaction", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -3978,7 +4055,7 @@ describe("Codex app-server provider", () => { agentPath: "/root/child", }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "child-thread-root", item: { type: "subAgentActivity", @@ -3988,7 +4065,7 @@ describe("Codex app-server provider", () => { agentPath: "/root/child/grandchild", }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "collabAgentToolCall", @@ -4001,7 +4078,7 @@ describe("Codex app-server provider", () => { }, }, }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { threadId: "grandchild-thread", itemId: "grandchild-after-wait", delta: "Still nested.", @@ -4027,28 +4104,28 @@ describe("Codex app-server provider", () => { }); }); - test("never treats an unmapped foreign terminal as the root terminal", () => { + test("never treats an unmapped foreign terminal as the root terminal", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { threadId: "unmapped-child-thread", turn: { status: "completed" }, }); expect(events).toEqual([]); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { threadId: "test-thread", turn: { status: "completed" }, }); expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(1); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { threadId: "test-thread", turn: { id: "next-root-turn" }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -4064,12 +4141,12 @@ describe("Codex app-server provider", () => { }); }); - test("routes msg-scoped legacy Codex events to their child thread", () => { + test("routes msg-scoped legacy Codex events to their child thread", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -4079,7 +4156,7 @@ describe("Codex app-server provider", () => { agentPath: "/root/legacy-envelope-child", }, }); - asInternals(session).handleNotification("codex/event/exec_command_begin", { + await asInternals(session).handleNotification("codex/event/exec_command_begin", { msg: { type: "exec_command_begin", threadId: "legacy-envelope-child", @@ -4087,7 +4164,7 @@ describe("Codex app-server provider", () => { command: "pwd", }, }); - asInternals(session).handleNotification("codex/event/task_complete", { + await asInternals(session).handleNotification("codex/event/task_complete", { msg: { type: "task_complete", thread_id: "legacy-envelope-child", @@ -4125,7 +4202,7 @@ describe("Codex app-server provider", () => { }, }); - asInternals(session).handleNotification("codex/event/task_complete", { + await asInternals(session).handleNotification("codex/event/task_complete", { msg: { type: "task_complete" }, }); expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(1); @@ -4147,11 +4224,13 @@ describe("Codex app-server provider", () => { const spawn = waitForTimelineToolCall(session, "spawn-legacy-only-child"); appServer.startsTurn({ threadId: "thread-1", turnId: "turn-with-legacy-only-child" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.startsLegacyOnlySubAgent({ callId: "spawn-legacy-only-child", threadId: "legacy-only-child-thread", agentPath: "/root/legacy-only-child", }); + await new Promise((resolve) => setImmediate(resolve)); await expect(child).resolves.toMatchObject({ type: "provider_subagent", @@ -4179,6 +4258,7 @@ describe("Codex app-server provider", () => { }); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await resultPromise; appServer.assertNoErrors(); } finally { @@ -4203,10 +4283,12 @@ describe("Codex app-server provider", () => { const resultPromise = session.run("Wait for the child."); await appServer.waitForTurnStart(); appServer.startsTurn({ threadId: "thread-1", turnId: "turn-waiting-for-child" }); + await new Promise((resolve) => setImmediate(resolve)); await expect(session.interrupt()).rejects.toThrow("A foreground turn is already active"); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await resultPromise; appServer.assertNoErrors(); } finally { @@ -4231,10 +4313,12 @@ describe("Codex app-server provider", () => { const resultPromise = session.run("Wait for the child."); await appServer.waitForTurnStart(); appServer.startsTurn({ threadId: "thread-1", turnId: "turn-already-idle" }); + await new Promise((resolve) => setImmediate(resolve)); await expect(session.interrupt()).resolves.toBeUndefined(); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await resultPromise; appServer.assertNoErrors(); } finally { @@ -4263,10 +4347,12 @@ describe("Codex app-server provider", () => { const interruptPromise = session.interrupt(); appServer.startsTurn({ threadId: "thread-1", turnId: "turn-identified-late" }); + await new Promise((resolve) => setImmediate(resolve)); await interruptPromise; expect(interruptedTurns).toEqual([{ threadId: "thread-1", turnId: "turn-identified-late" }]); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await resultPromise; appServer.assertNoErrors(); } finally { @@ -4294,6 +4380,7 @@ describe("Codex app-server provider", () => { await appServer.waitForTurnStart(); const interruptPromise = session.interrupt(); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await expect(interruptPromise).resolves.toBeUndefined(); await resultPromise; @@ -4353,7 +4440,9 @@ describe("Codex app-server provider", () => { const nativeStart = await appServer.waitForTurnStart().catch(() => null); if (nativeStart) { appServer.startsTurn({ threadId: "thread-1", turnId: "cleanup-turn" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); } await resultPromise.catch(() => undefined); await session.close(); @@ -4371,7 +4460,7 @@ describe("Codex app-server provider", () => { }, }; - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { threadId: "test-thread", turn: { id: "autonomous-turn" }, }); @@ -4387,13 +4476,13 @@ describe("Codex app-server provider", () => { }); }); - test("never replaces the root identity with an early child thread start", () => { + test("never replaces the root identity with an early child thread start", async () => { const session = createSession(); - asInternals(session).handleNotification("thread/started", { + await asInternals(session).handleNotification("thread/started", { thread: { id: "child-thread-started-early" }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -4407,12 +4496,12 @@ describe("Codex app-server provider", () => { expect(session.currentThreadId).toBe("test-thread"); }); - test("does not leak aggregate child telemetry into the root timeline", () => { + test("does not leak aggregate child telemetry into the root timeline", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -4424,7 +4513,7 @@ describe("Codex app-server provider", () => { }); const eventCountAfterSpawn = events.length; - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { threadId: "child-thread-telemetry", plan: [{ step: "Child-only plan", status: "inProgress" }], }); @@ -4432,12 +4521,12 @@ describe("Codex app-server provider", () => { expect(events).toHaveLength(eventCountAfterSpawn); }); - test("keeps child context compaction inside the child card", () => { + test("keeps child context compaction inside the child card", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -4447,11 +4536,11 @@ describe("Codex app-server provider", () => { agentPath: "/root/compacting-child", }, }); - asInternals(session).handleNotification("item/started", { + await asInternals(session).handleNotification("item/started", { threadId: "child-thread-compaction", item: { type: "contextCompaction", id: "child-compaction" }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "child-thread-compaction", item: { type: "contextCompaction", id: "child-compaction" }, }); @@ -4471,12 +4560,12 @@ describe("Codex app-server provider", () => { }); }); - test("keeps the parent sub-agent running when a child command fails during the child turn", () => { + test("keeps the parent sub-agent running when a child command fails during the child turn", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "collabAgentToolCall", @@ -4490,7 +4579,7 @@ describe("Codex app-server provider", () => { }, }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "child-thread-1", item: { type: "commandExecution", @@ -4517,12 +4606,12 @@ describe("Codex app-server provider", () => { }); }); - test("does not synthesize a parent sub-agent failure from child error state alone", () => { + test("does not synthesize a parent sub-agent failure from child error state alone", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "collabAgentToolCall", @@ -4773,11 +4862,11 @@ describe("Codex app-server provider", () => { const liveEvents: AgentStreamEvent[] = []; session.subscribe((event) => liveEvents.push(event)); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { threadId: "v2-child-thread", turn: { id: "v2-child-turn-after-resume" }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "subAgentActivity", @@ -4787,7 +4876,7 @@ describe("Codex app-server provider", () => { agentPath: "/root/v2-child", }, }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { threadId: "v2-child-thread", itemId: "v2-child-message-after-resume", delta: "More findings after resume.", @@ -4805,11 +4894,11 @@ describe("Codex app-server provider", () => { }); liveEvents.length = 0; - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { threadId: "legacy-child-thread", turn: { id: "legacy-child-turn-after-resume" }, }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { threadId: "legacy-child-thread", itemId: "legacy-child-message-after-resume", delta: "Legacy findings after resume.", @@ -5058,6 +5147,7 @@ describe("Codex app-server provider", () => { threadId: "child-thread", agentPath: "/root/child", }); + await new Promise((resolve) => setImmediate(resolve)); await child; appServer.beginsSubAgentActivity({ callId: "parent-interacted", @@ -5066,7 +5156,9 @@ describe("Codex app-server provider", () => { kind: "interacted", agentPath: "/root", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await resultPromise; expect(providerSubagentIds).toEqual(new Set(["child-thread"])); @@ -5198,7 +5290,7 @@ describe("Codex app-server provider", () => { ]); }); - test("captures live Codex user message ids from item events", () => { + test("captures live Codex user message ids from item events", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); @@ -5209,11 +5301,11 @@ describe("Codex app-server provider", () => { content: [{ type: "text", text: "Use the native Codex id." }], }; - asInternals(session).handleNotification("item/started", { + await asInternals(session).handleNotification("item/started", { threadId: "test-thread", item: userMessageItem, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: userMessageItem, }); @@ -5232,19 +5324,19 @@ describe("Codex app-server provider", () => { ]); }); - test("emits Codex context compaction markers from live thread items", () => { + test("emits Codex context compaction markers from live thread items", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/started", { + await asInternals(session).handleNotification("item/started", { threadId: "test-thread", item: { type: "contextCompaction", id: "compact-live", }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "contextCompaction", @@ -5282,7 +5374,9 @@ describe("Codex app-server provider", () => { threadId: "thread-1", itemId: "compact-without-completion", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await terminalEvent; expect( @@ -5304,11 +5398,14 @@ describe("Codex app-server provider", () => { threadId: "thread-1", itemId: "compact-completed-normally", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completesCompaction({ threadId: "thread-1", itemId: "compact-completed-normally", }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await terminalEvent; expect( @@ -5327,8 +5424,11 @@ describe("Codex app-server provider", () => { try { appServer.startsCompaction({ threadId: "thread-1", itemId: "current-compaction" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completesCompaction({ threadId: "thread-1", itemId: "older-compaction" }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); await terminalEvent; expect( @@ -5350,10 +5450,12 @@ describe("Codex app-server provider", () => { try { appServer.startsCompaction({ threadId: "thread-1", itemId: `compact-${status}` }); + await new Promise((resolve) => setImmediate(resolve)); appServer.completeTurn({ status, error: status === "failed" ? { message: "Compaction failed" } : null, }); + await new Promise((resolve) => setImmediate(resolve)); await terminalEvent; expect( @@ -5367,17 +5469,17 @@ describe("Codex app-server provider", () => { } }); - test("emits and dedupes Codex thread/compacted notifications", () => { + test("emits and dedupes Codex thread/compacted notifications", async () => { const session = createSession(); session.activeForegroundTurnId = null; const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("thread/compacted", { + await asInternals(session).handleNotification("thread/compacted", { threadId: "test-thread", turnId: "legacy-compact-turn", }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "contextCompaction", @@ -5398,17 +5500,17 @@ describe("Codex app-server provider", () => { ]); }); - test("emits consecutive Codex thread/compacted notifications", () => { + test("emits consecutive Codex thread/compacted notifications", async () => { const session = createSession(); session.activeForegroundTurnId = null; const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("thread/compacted", { + await asInternals(session).handleNotification("thread/compacted", { threadId: "test-thread", turnId: "legacy-compact-turn-1", }); - asInternals(session).handleNotification("thread/compacted", { + await asInternals(session).handleNotification("thread/compacted", { threadId: "test-thread", turnId: "legacy-compact-turn-2", }); @@ -5534,14 +5636,14 @@ describe("Codex app-server provider", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); await handler?.run({ emit: (event) => events.push(event) }); - asInternals(session).handleNotification("item/started", { + await asInternals(session).handleNotification("item/started", { threadId: "test-thread", item: { type: "contextCompaction", id: "manual-compact", }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { threadId: "test-thread", item: { type: "contextCompaction", @@ -5656,23 +5758,23 @@ describe("Codex app-server provider", () => { }); }); - test("emits a synthetic plan approval permission after a successful Codex plan turn", () => { + test("emits a synthetic plan approval permission after a successful Codex plan turn", async () => { const session = createSession({ featureValues: { plan_mode: true, fast_mode: true }, }); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-1" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [ { step: "Inspect the existing auth flow", status: "completed" }, { step: "Implement the button behavior", status: "pending" }, ], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -5726,24 +5828,24 @@ describe("Codex app-server provider", () => { }); }); - test("does not complete Codex plan timeline cards while plan approval is pending", () => { + test("does not complete Codex plan timeline cards while plan approval is pending", async () => { const session = createSession({ featureValues: { plan_mode: true, fast_mode: true }, }); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-thread-item" }, }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "plan-item-1", type: "plan", text: "- Inspect README\n- Add a short note", }, }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -5772,28 +5874,28 @@ describe("Codex app-server provider", () => { }); }); - test("replaces a pending synthetic plan approval when a later plan completes", () => { + test("replaces a pending synthetic plan approval when a later plan completes", async () => { const session = createSession({ featureValues: { plan_mode: true }, }); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-first" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the first plan", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-second" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the revised plan", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -5812,13 +5914,13 @@ describe("Codex app-server provider", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-pending" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the original plan", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); const pendingPlan = session.getPendingPermissions()[0]; @@ -5852,13 +5954,13 @@ describe("Codex app-server provider", () => { featureValues: { plan_mode: true }, }); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-pending" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the original plan", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); const pendingPlan = session.getPendingPermissions()[0]; @@ -5906,13 +6008,13 @@ describe("Codex app-server provider", () => { featureValues: { plan_mode: true }, }); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-pending" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the original plan", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -5937,10 +6039,10 @@ describe("Codex app-server provider", () => { const startTurn = session.startTurn("Revise the plan"); await promptRequested; - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the newer plan", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); acceptPrompt?.(); @@ -5956,13 +6058,13 @@ describe("Codex app-server provider", () => { featureValues: { plan_mode: true }, }); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-pending" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the original plan", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); const pendingPlan = session.getPendingPermissions()[0]; @@ -5982,12 +6084,12 @@ describe("Codex app-server provider", () => { expect(session.getPendingPermissions()).toEqual([]); }); - test("emits imageView paths with spaces as valid assistant markdown images", () => { + test("emits imageView paths with spaces as valid assistant markdown images", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "image-view-1", type: "imageView", @@ -6013,12 +6115,12 @@ describe("Codex app-server provider", () => { ["saved_path", { saved_path: "/tmp/generated-snake.png" }, "file:///tmp/generated-snake.png"], ])( "emits imageGeneration thread items with %s as assistant markdown images", - (_fieldName, imageFields, expectedPath) => { + async (_fieldName, imageFields, expectedPath) => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: `image-generation-${_fieldName}`, type: "imageGeneration", @@ -6041,12 +6143,12 @@ describe("Codex app-server provider", () => { }, ); - test("materializes imageGeneration base64 results before rendering markdown", () => { + test("materializes imageGeneration base64 results before rendering markdown", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "image-generation-base64", type: "imageGeneration", @@ -6194,21 +6296,22 @@ describe("Codex app-server provider", () => { } }); - test("ignores incomplete imageGeneration thread items without failing the turn", () => { + test("ignores incomplete imageGeneration thread items without failing the turn", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - expect(() => - asInternals(session).handleNotification("item/completed", { - item: { - id: "image-generation-incomplete", - type: "imageGeneration", - status: "in_progress", - }, - }), + expect( + async () => + await asInternals(session).handleNotification("item/completed", { + item: { + id: "image-generation-incomplete", + type: "imageGeneration", + status: "in_progress", + }, + }), ).not.toThrow(); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -6231,12 +6334,12 @@ describe("Codex app-server provider", () => { ["no failure details", null, { message: "Image generation failed" }], ])( "emits failed imageGeneration items with %s as failed tool calls", - (_label, failure, expectedError) => { + async (_label, failure, expectedError) => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "image-generation-failed", type: "imageGeneration", @@ -6323,12 +6426,12 @@ describe("Codex app-server provider", () => { ]); }); - test("emits usage_updated on token usage updates and keeps usage on turn completion", () => { + test("emits usage_updated on token usage updates and keeps usage on turn completion", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("thread/tokenUsage/updated", { + await asInternals(session).handleNotification("thread/tokenUsage/updated", { tokenUsage: { model_context_window: 200000, last: { @@ -6339,7 +6442,7 @@ describe("Codex app-server provider", () => { }, }, }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -6369,20 +6472,20 @@ describe("Codex app-server provider", () => { }); }); - test("streams Codex assistant message deltas and does not replay completed text", () => { + test("streams Codex assistant message deltas and does not replay completed text", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { itemId: "assistant-item-1", delta: "Hel", }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { itemId: "assistant-item-1", delta: "lo", }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "assistant-item-1", type: "agentMessage", @@ -6406,20 +6509,20 @@ describe("Codex app-server provider", () => { ]); }); - test("emits only the missing assistant suffix when completed text extends streamed deltas", () => { + test("emits only the missing assistant suffix when completed text extends streamed deltas", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { itemId: "assistant-item-2", delta: "Hel", }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { itemId: "assistant-item-2", delta: "lo", }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "assistant-item-2", type: "agentMessage", @@ -6449,24 +6552,24 @@ describe("Codex app-server provider", () => { ]); }); - test("emits a markdown divider when a new Codex assistant item starts after the previous one completed", () => { + test("emits a markdown divider when a new Codex assistant item starts after the previous one completed", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { itemId: "assistant-item-3", delta: "I’m in the waiting phase now. The next read is intentionally delayed so we get meaningful CI state instead of churn.", }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "assistant-item-3", type: "agentMessage", text: "I’m in the waiting phase now. The next read is intentionally delayed so we get meaningful CI state instead of churn.", }, }); - asInternals(session).handleNotification("item/agentMessage/delta", { + await asInternals(session).handleNotification("item/agentMessage/delta", { itemId: "assistant-item-4", delta: "CI is still cooking. I’m staying on the current run rather than jumping around, because the first red job will tell us exactly whether anything else needs work.", @@ -6496,20 +6599,20 @@ describe("Codex app-server provider", () => { ]); }); - test("streams Codex reasoning deltas and does not replay completed reasoning", () => { + test("streams Codex reasoning deltas and does not replay completed reasoning", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { + await asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { itemId: "reasoning-item-1", delta: "Think", }); - asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { + await asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { itemId: "reasoning-item-1", delta: "ing", }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "reasoning-item-1", type: "reasoning", @@ -6533,20 +6636,20 @@ describe("Codex app-server provider", () => { ]); }); - test("emits only the missing reasoning suffix when completed reasoning extends streamed deltas", () => { + test("emits only the missing reasoning suffix when completed reasoning extends streamed deltas", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { + await asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { itemId: "reasoning-item-2", delta: "Think", }); - asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { + await asInternals(session).handleNotification("item/reasoning/summaryTextDelta", { itemId: "reasoning-item-2", delta: "ing", }); - asInternals(session).handleNotification("item/completed", { + await asInternals(session).handleNotification("item/completed", { item: { id: "reasoning-item-2", type: "reasoning", @@ -6583,13 +6686,13 @@ describe("Codex app-server provider", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-2" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the new flow", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -6637,13 +6740,13 @@ describe("Codex app-server provider", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-3" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the safe flow", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -6705,13 +6808,13 @@ describe("Codex app-server provider", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); - asInternals(session).handleNotification("turn/started", { + await asInternals(session).handleNotification("turn/started", { turn: { id: "turn-plan-4" }, }); - asInternals(session).handleNotification("turn/plan/updated", { + await asInternals(session).handleNotification("turn/plan/updated", { plan: [{ step: "Implement the fast flow", status: "pending" }], }); - asInternals(session).handleNotification("turn/completed", { + await asInternals(session).handleNotification("turn/completed", { turn: { status: "completed", error: null }, }); @@ -7182,6 +7285,7 @@ test("Codex runtime safety queries live terminals and retains unsupported app se await appServer.waitForTurnStart(); expect(await session.canStopRuntime()).toBe(false); appServer.completeTurn({ threadId: "thread-1" }); + await new Promise((resolve) => setImmediate(resolve)); await expect.poll(() => session.canStopRuntime()).toBe(true); response = { data: [{ processId: "shell-1" }], nextCursor: null }; expect(await session.canStopRuntime()).toBe(false); @@ -7199,3 +7303,60 @@ test("Codex runtime safety queries live terminals and retains unsupported app se await session.close(); } }); + +test("publishes a materialized image before dispatching the following turn completion", async () => { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig(), + null, + createTestLogger(), + async () => appServer.child, + ); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + try { + await session.startTurn("generate an image"); + const data = Buffer.alloc(1024 * 1024, 47).toString("base64"); + const imageFrame = + JSON.stringify({ + method: "item/completed", + params: { + item: { + id: "ordered-image", + type: "imageGeneration", + status: "completed", + result: { data, mimeType: "image/png" }, + }, + }, + }) + "\n"; + const completed = new Promise((resolve) => + session.subscribe((event) => { + if (event.type === "turn_completed") resolve(); + }), + ); + appServer.child.stdout.write(imageFrame); + appServer.child.stdout.write( + JSON.stringify({ + method: "turn/completed", + params: { turn: { status: "completed", error: null } }, + }) + "\n", + ); + await completed; + await new Promise((resolve) => setTimeout(resolve, 5)); + const imageIndex = events.findIndex( + (event) => + event.type === "timeline" && + event.item.type === "assistant_message" && + event.item.text.includes("![Image]"), + ); + const completedIndex = events.findIndex((event) => event.type === "turn_completed"); + expect(imageIndex).toBeGreaterThanOrEqual(0); + expect(imageIndex).toBeLessThan(completedIndex); + const imageEvent = events[imageIndex]; + if (imageEvent.type !== "timeline" || imageEvent.item.type !== "assistant_message") + throw new Error("Expected image"); + expect(existsSync(markdownImageSource(imageEvent.item.text))).toBe(true); + } finally { + await session.close(); + } +}); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index f438c98ec4..291841c730 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -1939,12 +1939,12 @@ function codexImageOutputFromResult(result: unknown): ProviderImageOutput | null }; } -function mapCodexThreadImageItem( +async function mapCodexThreadImageItem( normalizedType: string, normalizedItem: Record, -): AgentTimelineItem | null { +): Promise { if (normalizedType === "imageView") { - return renderProviderImageOutputAsAssistantMarkdown({ + return await renderProviderImageOutputAsAssistantMarkdown({ path: firstStringField(normalizedItem, ["path"]), }); } @@ -1960,7 +1960,7 @@ function mapCodexThreadImageItem( const savedPath = firstStringField(normalizedItem, ["savedPath", "saved_path"]); const result = codexImageOutputFromResult(normalizedItem.result); - return renderProviderImageOutputAsAssistantMarkdown( + return await renderProviderImageOutputAsAssistantMarkdown( { path: savedPath ?? result?.path ?? null, url: result?.url ?? null, @@ -1982,10 +1982,10 @@ function mapCodexAgentMessage(item: Record): AgentTimelineItem }; } -export function threadItemToTimeline( +export async function threadItemToTimeline( item: unknown, options?: { includeUserMessage?: boolean; cwd?: string | null }, -): AgentTimelineItem | null { +): Promise { const itemRecord = toObjectRecord(item); if (!itemRecord) return null; const includeUserMessage = options?.includeUserMessage ?? true; @@ -1999,7 +1999,7 @@ export function threadItemToTimeline( : itemRecord; if (normalizedType === "imageView" || normalizedType === "imageGeneration") { - return mapCodexThreadImageItem(normalizedType, normalizedItem); + return await mapCodexThreadImageItem(normalizedType, normalizedItem); } if (normalizedType && CODEX_TOOL_THREAD_ITEM_TYPES.has(normalizedType)) { return mapCodexToolCallFromThreadItem(normalizedItem, { cwd }); @@ -2024,7 +2024,7 @@ export function threadItemToTimeline( } } -function mcpToolResultImagesToTimeline(item: unknown): AgentTimelineItem[] { +async function mcpToolResultImagesToTimeline(item: unknown): Promise { const itemRecord = toObjectRecord(item); if (!itemRecord) { return []; @@ -2037,24 +2037,25 @@ function mcpToolResultImagesToTimeline(item: unknown): AgentTimelineItem[] { } const { images } = splitCodexMcpToolResultImages(itemRecord.result); - return images - .map((image) => - renderProviderImageOutputAsAssistantMarkdown(image, { - materialize: materializeProviderImage, - }), - ) - .filter((timelineItem): timelineItem is AgentTimelineItem => timelineItem !== null); + const items: AgentTimelineItem[] = []; + for (const image of images) { + const imageItem = await renderProviderImageOutputAsAssistantMarkdown(image, { + materialize: materializeProviderImage, + }); + if (imageItem) items.push(imageItem); + } + return items; } -function threadItemToTimelineEntries( +async function threadItemToTimelineEntries( item: unknown, options?: { includeUserMessage?: boolean; cwd?: string | null }, -): AgentTimelineItem[] { - const timelineItem = threadItemToTimeline(item, options); +): Promise { + const timelineItem = await threadItemToTimeline(item, options); if (!timelineItem) { return []; } - return [timelineItem, ...mcpToolResultImagesToTimeline(item)]; + return [timelineItem, ...(await mcpToolResultImagesToTimeline(item))]; } const CodexThreadReadResponseSchema = z @@ -2103,7 +2104,7 @@ async function loadCodexThreadHistoryTimeline(params: { historicalSubAgentActivity.agentThreadId, ); if (existingIndex !== undefined) { - const activityTimelineItem = threadItemToTimeline(item, { cwd: params.cwd }); + const activityTimelineItem = await threadItemToTimeline(item, { cwd: params.cwd }); updateHistoricalSubAgentActivity( timeline, existingIndex, @@ -2116,7 +2117,7 @@ async function loadCodexThreadHistoryTimeline(params: { continue; } } - for (const timelineItem of threadItemToTimelineEntries(item, { cwd: params.cwd })) { + for (const timelineItem of await threadItemToTimelineEntries(item, { cwd: params.cwd })) { const timestamp = readCodexHistoryTimestamp(item) ?? readCodexTurnHistoryTimestamp(turn, timelineItem); const settledTimelineItem = @@ -3290,10 +3291,12 @@ export async function codexAppServerTurnInputFromPrompt( } if (block.type === "image") { try { - const filePath = materializeProviderImage({ - data: block.data, - mimeType: block.mimeType, - }).path; + const filePath = ( + await materializeProviderImage({ + data: block.data, + mimeType: block.mimeType, + }) + ).path; output.push({ type: "localImage", path: filePath }); } catch (error) { const message = error instanceof Error ? error.message : String(error); @@ -4370,6 +4373,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.dismissPendingPlanApprovals("Dismissed by a new prompt"); try { + await this.pendingNotifications; await this.connect(); if (!this.client) { throw new Error("Codex client not initialized"); @@ -5437,7 +5441,21 @@ export class CodexAppServerAgentSession implements AgentSession { return `codex-turn-${this.nextTurnOrdinal++}`; } - private handleNotification(method: string, params: unknown): void { + private pendingNotifications: Promise = Promise.resolve(); + + private handleNotification(method: string, params: unknown): Promise { + // An image must exist before its event, and completion must follow that event. + // Transport callbacks cannot await: the session owns their ordered drain. + const next = this.pendingNotifications.then(() => { + return this.closed ? undefined : this.processNotification(method, params); + }); + this.pendingNotifications = next.catch((error: unknown) => { + this.handleUnexpectedTermination(error instanceof Error ? error : new Error(String(error))); + }); + return this.pendingNotifications; + } + + private async processNotification(method: string, params: unknown): Promise { const notificationParams = toObjectRecord(params); if (method === "serverRequest/resolved" && typeof notificationParams?.requestId === "number") { const requestId = this.mcpElicitationPermissionIds.get(notificationParams.requestId); @@ -5467,13 +5485,16 @@ export class CodexAppServerAgentSession implements AgentSession { return; } if (route.kind === "sub_agent") { - this.dispatchSubAgentNotification(parsed, route.callId); + await this.dispatchSubAgentNotification(parsed, route.callId); return; } - this.dispatchParsedNotification(parsed); + await this.dispatchParsedNotification(parsed); } - private dispatchSubAgentNotification(parsed: ParsedCodexNotification, callId: string): void { + private async dispatchSubAgentNotification( + parsed: ParsedCodexNotification, + callId: string, + ): Promise { switch (parsed.kind) { case "thread_started": this.emitSubAgentActivityUpdate(callId, "running", { reopen: true }); @@ -5484,7 +5505,7 @@ export class CodexAppServerAgentSession implements AgentSession { case "reasoning_delta": case "item_started": case "item_completed": - this.dispatchParsedNotification(parsed); + await this.dispatchParsedNotification(parsed); return; case "exec_command_output_delta": case "file_change_output_delta": @@ -5510,7 +5531,7 @@ export class CodexAppServerAgentSession implements AgentSession { } } - private dispatchParsedNotification(parsed: ParsedCodexNotification): void { + private async dispatchParsedNotification(parsed: ParsedCodexNotification): Promise { if (isCodexDeltaNotification(parsed)) { this.handleCodexDeltaNotification(parsed); return; @@ -5555,10 +5576,10 @@ export class CodexAppServerAgentSession implements AgentSession { this.handlePatchApplyCompletedNotification(parsed); return; case "item_completed": - this.handleItemCompletedNotification(parsed); + await this.handleItemCompletedNotification(parsed); return; case "item_started": - this.handleItemStartedNotification(parsed); + await this.handleItemStartedNotification(parsed); return; case "invalid_payload": this.warnInvalidNotificationPayload(parsed.method, parsed.params); @@ -5600,7 +5621,7 @@ export class CodexAppServerAgentSession implements AgentSession { pending.push(parsed); } - private replayPendingSubAgentNotifications(threadIds: readonly string[]): void { + private async replayPendingSubAgentNotifications(threadIds: readonly string[]): Promise { for (const threadId of threadIds) { const pending = this.pendingSubAgentNotificationsByThreadId.get(threadId); if (!pending) { @@ -5612,7 +5633,7 @@ export class CodexAppServerAgentSession implements AgentSession { continue; } for (const parsed of pending) { - this.dispatchSubAgentNotification(parsed, callId); + await this.dispatchSubAgentNotification(parsed, callId); } } } @@ -6578,9 +6599,13 @@ export class CodexAppServerAgentSession implements AgentSession { } } - private handleItemCompletedNotification( + private ownsNotificationTurn(turnId: string | null): boolean { + return !this.closed && this.activeForegroundTurnId === turnId; + } + + private async handleItemCompletedNotification( parsed: Extract, - ): void { + ): Promise { // Codex emits mirrored lifecycle notifications via both `codex/event/item_*` // and canonical `item/*`. Render ordinary items only from the canonical // channel, but accept a legacy-only child announcement so it can establish @@ -6590,20 +6615,23 @@ export class CodexAppServerAgentSession implements AgentSession { } this.receiveAsyncQuestion(parsed.threadId, parsed.item); if (this.isUserMessageItem(parsed.item)) { - this.handleUserMessageItem(parsed); + await this.handleUserMessageItem(parsed); return; } const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); if (this.handleCompletedSpecialItem(parsed, childSubAgentCallId)) { return; } - const timelineItem = threadItemToTimeline(parsed.item, { + const foregroundTurnId = this.activeForegroundTurnId; + const timelineItem = await threadItemToTimeline(parsed.item, { includeUserMessage: false, cwd: this.config.cwd ?? null, }); if (!timelineItem) { return; } + const imageItems = await mcpToolResultImagesToTimeline(parsed.item); + if (!this.ownsNotificationTurn(foregroundTurnId)) return; const registeredChildThreadIds = timelineItem.type === "tool_call" ? this.registerSubAgentToolCall({ @@ -6613,12 +6641,11 @@ export class CodexAppServerAgentSession implements AgentSession { parentSubagentId: resolveCodexParentSubagentId(childSubAgentCallId, parsed.threadId), }) : []; - const imageItems = mcpToolResultImagesToTimeline(parsed.item); if (childSubAgentCallId) { this.emitCompletedProviderSubagentItem(parsed, timelineItem); this.handleSubAgentChildItemCompleted(childSubAgentCallId, parsed.item.id, timelineItem); this.emitProviderSubagentTimelineItems(parsed.threadId, imageItems); - this.replayPendingSubAgentNotifications(registeredChildThreadIds); + await this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } const normalizedItemType = normalizeCodexThreadItemType( @@ -6626,7 +6653,7 @@ export class CodexAppServerAgentSession implements AgentSession { ); const itemId = parsed.item.id; if (this.shouldSkipCompletedThreadItem(timelineItem, normalizedItemType, itemId)) { - this.replayPendingSubAgentNotifications(registeredChildThreadIds); + await this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } if (this.consumeStreamedTextCompletion(timelineItem, itemId)) { @@ -6637,7 +6664,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.emittedItemCompletedIds.add(itemId); this.emittedItemStartedIds.delete(itemId); } - this.replayPendingSubAgentNotifications(registeredChildThreadIds); + await this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } this.applyBufferedDeltaTextToTimelineItem(timelineItem, itemId); @@ -6666,7 +6693,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.pendingCommandOutputDeltas.delete(itemId); this.pendingFileChangeOutputDeltas.delete(itemId); } - this.replayPendingSubAgentNotifications(registeredChildThreadIds); + await this.replayPendingSubAgentNotifications(registeredChildThreadIds); } private consumeStreamedTextCompletion( @@ -6738,14 +6765,14 @@ export class CodexAppServerAgentSession implements AgentSession { } } - private handleItemStartedNotification( + private async handleItemStartedNotification( parsed: Extract, - ): void { + ): Promise { if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) { return; } if (this.isUserMessageItem(parsed.item)) { - this.handleUserMessageItem(parsed); + await this.handleUserMessageItem(parsed); return; } const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId); @@ -6767,7 +6794,7 @@ export class CodexAppServerAgentSession implements AgentSession { if (this.handleRegisteredSubAgentActivity(parsed.item)) { return; } - const timelineItem = threadItemToTimeline(parsed.item, { + const timelineItem = await threadItemToTimeline(parsed.item, { includeUserMessage: false, cwd: this.config.cwd ?? null, }); @@ -6786,7 +6813,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.upsertSubAgentChildItem(childSubAgentCallId, parsed.item.id, timelineItem); } this.emitSubAgentActivityUpdate(childSubAgentCallId, "running"); - this.replayPendingSubAgentNotifications(registeredChildThreadIds); + await this.replayPendingSubAgentNotifications(registeredChildThreadIds); return; } const normalizedItemType = normalizeCodexThreadItemType( @@ -6810,14 +6837,14 @@ export class CodexAppServerAgentSession implements AgentSession { this.pendingCommandOutputDeltas.delete(itemId); this.pendingFileChangeOutputDeltas.delete(itemId); } - this.replayPendingSubAgentNotifications(registeredChildThreadIds); + await this.replayPendingSubAgentNotifications(registeredChildThreadIds); } - private handleUserMessageItem( + private async handleUserMessageItem( parsed: Extract, - ): void { + ): Promise { const itemId = parsed.item.id; - const timelineItem = threadItemToTimeline(parsed.item, { + const timelineItem = await threadItemToTimeline(parsed.item, { includeUserMessage: true, cwd: this.config.cwd ?? null, }); diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index 983f9cf55f..a741514305 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -1627,3 +1627,40 @@ test.each([ expect(await session.canStopRuntime!()).toBe(expected); }, ); + +test.each([false, true])( + "image prompt keeps OMP's vision capability contract (vision: %s)", + async (vision) => { + const runtime = new FakeOmp(); + runtime.queueSessionSetup((session) => { + session.state = { + ...session.state, + model: { + provider: "test", + id: "image-model", + input: vision ? ["text", "image"] : ["text"], + }, + }; + }); + const client = new OmpAgentClient({ logger: createTestLogger(), runtime }); + const session = await client.createSession({ provider: "omp", cwd: process.cwd() }); + onTestFinished(() => session.close()); + const bytes = Buffer.from("OMP attachment fixture"); + const first = session.startTurn([ + { type: "text", text: "Describe this image." }, + { type: "image", data: bytes.toString("base64"), mimeType: "image/png" }, + ]); + await expect(session.startTurn("overlapping prompt")).rejects.toThrow("already active"); + await first; + const [prompt] = runtime.latestSession().prompts; + expect(prompt.imageCount).toBe(vision ? 1 : 0); + if (vision) expect(prompt.message).toBe("Describe this image."); + else { + const imagePath = prompt.message.match(/\[Image available at: (.+)\]/)?.[1]; + expect(imagePath).toBeTruthy(); + const { readFile, rm } = await import("node:fs/promises"); + expect(await readFile(imagePath!)).toEqual(bytes); + await rm(imagePath!); + } + }, +); diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 434a49d89e..9e01b2ac46 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -317,9 +317,9 @@ function ompModelSupportsImageInput(model: OmpModel | null | undefined): boolean return model?.input?.includes("image") === true; } -function renderTextOnlyImageHint(image: { data: string; mimeType: string }): string { +async function renderTextOnlyImageHint(image: { data: string; mimeType: string }): Promise { try { - const materialized = materializeProviderImage({ + const materialized = await materializeProviderImage({ data: image.data, mimeType: image.mimeType, }); @@ -329,10 +329,10 @@ function renderTextOnlyImageHint(image: { data: string; mimeType: string }): str } } -function convertPromptInput( +async function convertPromptInput( prompt: AgentPromptInput, options: { model: OmpModel | null | undefined }, -): OmpPromptPayload { +): Promise { if (typeof prompt === "string") { return { text: prompt }; } @@ -355,7 +355,7 @@ function convertPromptInput( mimeType: block.mimeType, }); } else { - textParts.push(renderTextOnlyImageHint(block)); + textParts.push(await renderTextOnlyImageHint(block)); } continue; } @@ -868,11 +868,9 @@ export class OmpAgentSession implements AgentSession { throw new Error("An OMP turn is already active"); } - const payload = convertPromptInput(prompt, { model: this.state.model }); const turnId = randomUUID(); this.live = true; this.activeTurnId = turnId; - this.rememberClientMessage(options?.clientMessageId ?? null, payload.text); this.activeAssistantMessageId = null; this.activeTurnTerminalAssistantMessage = null; this.activeTurnStarted = false; @@ -880,6 +878,9 @@ export class OmpAgentSession implements AgentSession { this.activeTurnHasUserMessage = false; this.activePromptRequestId = null; this.clearNoTurnBuffers(); + const payload = await convertPromptInput(prompt, { model: this.state.model }); + if (this.closed || this.activeTurnId !== turnId) return { turnId }; + this.rememberClientMessage(options?.clientMessageId ?? null, payload.text); this.activeNoTurnPromptText = payload.text; this.usagePoller.startTurn(); if (announceTurn) { @@ -946,7 +947,9 @@ export class OmpAgentSession implements AgentSession { ): Promise { if (this.closed || this.activeTurnId !== options.expectedTurnId) return { status: "unavailable" }; - const payload = convertPromptInput(prompt, { model: this.state.model }); + const payload = await convertPromptInput(prompt, { model: this.state.model }); + if (this.closed || this.activeTurnId !== options.expectedTurnId) + return { status: "unavailable" }; if (this.parseSlashCommandInput(payload.text)) return { status: "unavailable" }; const submission = this.rememberClientMessage(options.clientMessageId ?? null, payload.text); try { diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts b/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts index d9d254d9de..e855a8b595 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/v2-streamed-text-memory.ts @@ -21,16 +21,18 @@ for (let index = 0; index < chunks; index++) { const event = mode === "delta" ? timeline.delta({ ...part, delta }) - : timeline.messages([ - { - id: "answer", - type: "assistant", - agent: "build", - model: { providerID: "test", id: "model" }, - time: { created: 2 }, - content: [{ type: "reasoning", text }], - }, - ])[0]; + : ( + await timeline.messages([ + { + id: "answer", + type: "assistant", + agent: "build", + model: { providerID: "test", id: "model" }, + time: { created: 2 }, + content: [{ type: "reasoning", text }], + }, + ]) + )[0]; if (event?.type !== "timeline") throw new Error("Expected a timeline event"); projection.append({ seq: index + 1, diff --git a/packages/server/src/server/agent/providers/opencode/v2/children.ts b/packages/server/src/server/agent/providers/opencode/v2/children.ts index 8c2e0ec312..1ec3bf01f8 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/children.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/children.ts @@ -92,7 +92,7 @@ export class SessionChildren { this.childStates.set(info.id, signature); this.options.emit({ type: "provider_subagent", provider: "opencode", event: presentation }); } - for (const event of timeline.messages(await messages(this.options.client(), info.id))) { + for (const event of await timeline.messages(await messages(this.options.client(), info.id))) { if (event.type === "timeline") this.options.emit({ type: "provider_subagent", diff --git a/packages/server/src/server/agent/providers/opencode/v2/session.ts b/packages/server/src/server/agent/providers/opencode/v2/session.ts index 9358a14040..865c576e52 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/session.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/session.ts @@ -131,7 +131,7 @@ export class OpenCodeV2Session implements AgentSession { await this.configureConnection(); const location = { directory: this.config.cwd }; this.modes = modesFromV2((await this.client.agent.list({ location })).data); - this.timeline.messages(await messages(this.client, this.id)); + await this.timeline.messages(await messages(this.client, this.id)); await this.startStream(); await this.children.reconcile(this.id); } @@ -261,7 +261,7 @@ export class OpenCodeV2Session implements AgentSession { async *streamHistory() { await this.reconnectIfExited(); const history = new V2Timeline(false); - yield* history.messages(await messages(this.client, this.id)); + yield* await history.messages(await messages(this.client, this.id)); } async getRuntimeInfo() { await this.reconnectIfExited(); @@ -417,7 +417,7 @@ export class OpenCodeV2Session implements AgentSession { ]); this.info = info; this.history = history; - for (const event of this.timeline.messages(history)) this.emitTimeline(event); + for (const event of await this.timeline.messages(history)) this.emitTimeline(event); await this.permissions.reconcile(this.id); } private async reconcile() { diff --git a/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts b/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts index aa74faa6cf..ef30689255 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/streaming.test.ts @@ -80,7 +80,7 @@ describe("OpenCode v2 token streaming", () => { } }); - test("deduplicates snapshots before and after deltas using per-type ordinals", () => { + test("deduplicates snapshots before and after deltas using per-type ordinals", async () => { const timeline = new V2Timeline(); const text = { assistantMessageID: "answer", type: "text", ordinal: 0 } as const; const reasoning = { ...text, type: "reasoning" } as const; @@ -99,24 +99,24 @@ describe("OpenCode v2 token streaming", () => { }, { type: "text", text: "Hello" }, ]); - expect(timeline.messages([snapshot])).toMatchObject([{ item: { text: "lo" } }]); + expect(await timeline.messages([snapshot])).toMatchObject([{ item: { text: "lo" } }]); expect(timeline.delta({ ...text, delta: "lo" })).toBeNull(); - expect(timeline.messages([snapshot])).toEqual([]); + expect(await timeline.messages([snapshot])).toEqual([]); expect(timeline.delta({ ...text, delta: "!" })).toMatchObject({ item: { text: "!" } }); - expect(timeline.messages([snapshot])).toEqual([]); + expect(await timeline.messages([snapshot])).toEqual([]); }); - test("recovers missed fragments from snapshots after reconnect", () => { + test("recovers missed fragments from snapshots after reconnect", async () => { const timeline = new V2Timeline(); const part = { assistantMessageID: "answer", type: "text", ordinal: 0 } as const; timeline.startPart(part); expect(timeline.delta({ ...part, delta: "Hel" })).toMatchObject({ item: { text: "Hel" } }); timeline.resetStreams(); expect(timeline.delta({ ...part, delta: "!" })).toBeNull(); - expect(timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toMatchObject([ + expect(await timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toMatchObject([ { item: { text: "lo!" } }, ]); - expect(timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toEqual([]); + expect(await timeline.messages([assistant([{ type: "text", text: "Hello!" }])])).toEqual([]); }); test("withholds streamed prose during a structured-output turn", async () => { diff --git a/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts b/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts index d86b9a7dcb..e5c23bb25a 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/timeline.test.ts @@ -15,7 +15,7 @@ function assistant(content: SessionMessageAssistant["content"]): SessionMessageA }; } -test("a completed edit carries the replaced and replacement text", () => { +test("a completed edit carries the replaced and replacement text", async () => { // Captured from OpenCode 2.0.18, whose edit tool takes `path` instead of `filePath`. const edit = assistant([ { @@ -45,7 +45,7 @@ test("a completed edit carries the replaced and replacement text", () => { }, ]); - expect(new V2Timeline().messages([edit])).toMatchObject([ + expect(await new V2Timeline().messages([edit])).toMatchObject([ { item: { type: "tool_call", @@ -57,7 +57,7 @@ test("a completed edit carries the replaced and replacement text", () => { ]); }); -test("a tool's image output is shown as an image, not as base64 in the tool output", () => { +test("a tool's image output is shown as an image, not as base64 in the tool output", async () => { const png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="; const screenshot = assistant([ @@ -80,7 +80,7 @@ test("a tool's image output is shown as an image, not as base64 in the tool outp ]); const timeline = new V2Timeline(); - const events = timeline.messages([screenshot]); + const events = await timeline.messages([screenshot]); expect(JSON.stringify(events)).not.toContain(png); expect(events.map((event) => event.type === "timeline" && event.item)).toMatchObject([ @@ -95,7 +95,7 @@ test("a tool's image output is shown as an image, not as base64 in the tool outp text: expect.stringMatching(/^\n\n!\[Image\]\(file:\/\/.*paseo-attachments.*\.png\)\n\n$/), }, ]); - expect(timeline.messages([screenshot])).toEqual([]); + expect(await timeline.messages([screenshot])).toEqual([]); }); test("a surrogate pair split across deltas streams back intact", () => { diff --git a/packages/server/src/server/agent/providers/opencode/v2/timeline.ts b/packages/server/src/server/agent/providers/opencode/v2/timeline.ts index 0fdd6f9f40..944699e22e 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/timeline.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/timeline.ts @@ -55,7 +55,15 @@ export class V2Timeline { return { type: "timeline", provider: "opencode", item, timestamp: new Date().toISOString() }; } - messages(messages: SessionMessageInfo[]): AgentStreamEvent[] { + private pendingMessages: Promise = Promise.resolve(); + + messages(messages: SessionMessageInfo[]): Promise { + const result = this.pendingMessages.then(() => this.projectMessages(messages)); + this.pendingMessages = result.catch(() => {}); + return result; + } + + private async projectMessages(messages: SessionMessageInfo[]): Promise { const events: AgentStreamEvent[] = []; const state: TimelineState = { structured: false, accepted: false }; for (const message of messages) { @@ -63,7 +71,7 @@ export class V2Timeline { const push = (item: AgentTimelineItem) => events.push({ type: "timeline", provider: "opencode", item, timestamp }); if (message.type === "user") this.userMessage(message, state, push); - else if (message.type === "assistant") this.assistantMessage(message, state, push); + else if (message.type === "assistant") await this.assistantMessage(message, state, push); else if (message.type === "compaction") this.compactionMessage(message, push); } return events; @@ -89,7 +97,7 @@ export class V2Timeline { }); } - private assistantMessage( + private async assistantMessage( message: Extract, state: TimelineState, push: (item: AgentTimelineItem) => void, @@ -99,16 +107,16 @@ export class V2Timeline { // paths address the same cursor entry. let textOrdinal = 0; let reasoningOrdinal = 0; - message.content.forEach((part, partIndex) => { + for (const [partIndex, part] of message.content.entries()) { if (part.type === "tool" && part.name === STRUCTURED_OUTPUT_TOOL) { this.structuredOutputPart(message.id, partIndex, part, state, push); } else if (part.type === "text" || part.type === "reasoning") { const ordinal = part.type === "text" ? textOrdinal++ : reasoningOrdinal++; this.textPart(message.id, part, ordinal, state, push); } else { - this.toolPart(message.id, partIndex, part, push); + await this.toolPart(message.id, partIndex, part, push); } - }); + } } private structuredOutputPart( @@ -156,7 +164,7 @@ export class V2Timeline { else push({ type: "reasoning", text: suffix }); } - private toolPart( + private async toolPart( messageID: string, partIndex: number, part: Extract< @@ -174,16 +182,16 @@ export class V2Timeline { if (item) push(item); // Images follow their tool call as markdown, as Claude and Codex emit them, // so base64 never reaches the tool output. - images.forEach((image, index) => { + for (const [index, image] of images.entries()) { const imageKey = `${key}:image:${index}`; - if (this.content.has(imageKey)) return; + if (this.content.has(imageKey)) continue; this.content.set(imageKey, ""); - const imageItem = renderProviderImageOutputAsAssistantMarkdown( + const imageItem = await renderProviderImageOutputAsAssistantMarkdown( { url: image.uri, mimeType: image.mime }, { materialize: materializeProviderImage }, ); if (imageItem) push(imageItem); - }); + } } private compactionMessage( diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index b9f31b61b2..ed2ac45cbd 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -3583,3 +3583,14 @@ test("pi runtime safety protects queued replies and runtime-owned activity", asy native.getStateError = new Error("Unable to prove idle"); await expect(session.canStopRuntime!()).rejects.toThrow("Unable to prove idle"); }); + +test("reserves the Pi turn while an image prompt is materialized", async () => { + const { pi, session } = await createSession(); + onTestFinished(() => session.close()); + const first = session.startTurn([ + { type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" }, + ]); + await expect(session.startTurn("overlapping prompt")).rejects.toThrow("already active"); + await first; + expect(pi.latestSession().prompts).toHaveLength(1); +}); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 881f8cbb29..fc369f9f16 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -367,9 +367,9 @@ function piModelSupportsImageInput(model: PiModel | null | undefined): boolean { return model?.input?.includes("image") === true; } -function renderTextOnlyImageHint(image: { data: string; mimeType: string }): string { +async function renderTextOnlyImageHint(image: { data: string; mimeType: string }): Promise { try { - const materialized = materializeProviderImage({ + const materialized = await materializeProviderImage({ data: image.data, mimeType: image.mimeType, }); @@ -379,10 +379,10 @@ function renderTextOnlyImageHint(image: { data: string; mimeType: string }): str } } -function convertPromptInput( +async function convertPromptInput( prompt: AgentPromptInput, options: { model: PiModel | null | undefined }, -): PiPromptPayload { +): Promise { if (typeof prompt === "string") { return { text: prompt }; } @@ -405,7 +405,7 @@ function convertPromptInput( mimeType: block.mimeType, }); } else { - textParts.push(renderTextOnlyImageHint(block)); + textParts.push(await renderTextOnlyImageHint(block)); } continue; } @@ -1280,7 +1280,6 @@ export class PiRpcAgentSession implements AgentSession { throw new Error("A Pi turn is already active"); } - const payload = convertPromptInput(prompt, { model: this.state.model }); const turnId = randomUUID(); this.activeTurnId = turnId; this.usagePoller.startTurn(); @@ -1292,6 +1291,8 @@ export class PiRpcAgentSession implements AgentSession { this.activePromptRequestId = null; this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); + const payload = await convertPromptInput(prompt, { model: this.state.model }); + if (this.closed || this.activeTurnId !== turnId) return { turnId }; this.activeNoTurnPromptText = payload.text; const shouldProbeForNoTurnPrompt = this.parseSlashCommandInput(payload.text) !== null; @@ -1363,7 +1364,9 @@ export class PiRpcAgentSession implements AgentSession { if (this.closed || this.activeTurnId !== options.expectedTurnId) { return { status: "unavailable" }; } - const payload = convertPromptInput(prompt, { model: this.state.model }); + const payload = await convertPromptInput(prompt, { model: this.state.model }); + if (this.closed || this.activeTurnId !== options.expectedTurnId) + return { status: "unavailable" }; // Pi rejects steer RPCs that are extension commands, so slash inputs keep the // interrupt-and-replace fallback where they can run directly. if (this.parseSlashCommandInput(payload.text)) { diff --git a/packages/server/src/server/agent/providers/provider-image-output.posix.test.ts b/packages/server/src/server/agent/providers/provider-image-output.posix.test.ts index 1f4da6e04a..5e71bad0bd 100644 --- a/packages/server/src/server/agent/providers/provider-image-output.posix.test.ts +++ b/packages/server/src/server/agent/providers/provider-image-output.posix.test.ts @@ -5,8 +5,8 @@ import { describe, expect, test } from "vitest"; import { materializeProviderImage } from "./provider-image-output.js"; describe.skipIf(process.platform === "win32")("materializeProviderImage", () => { - test("writes image attachments under a private temp directory", () => { - const materialized = materializeProviderImage({ + test("writes image attachments under a private temp directory", async () => { + const materialized = await materializeProviderImage({ data: "YWJjMTIz", mimeType: "image/png", }); diff --git a/packages/server/src/server/agent/providers/provider-image-output.test.ts b/packages/server/src/server/agent/providers/provider-image-output.test.ts index b4f4784a66..0606c59de3 100644 --- a/packages/server/src/server/agent/providers/provider-image-output.test.ts +++ b/packages/server/src/server/agent/providers/provider-image-output.test.ts @@ -1,10 +1,11 @@ -import { existsSync, rmSync } from "node:fs"; +import { existsSync, rmSync, statSync, readFileSync } from "node:fs"; import path from "node:path"; import MarkdownIt from "markdown-it"; import { projectTimelineRows } from "../timeline-projection.js"; import { describe, expect, test } from "vitest"; import { + MAX_PROVIDER_IMAGE_BYTES, isProviderImageMarkdown, materializeProviderImage, renderProviderImageOutputAsAssistantMarkdown, @@ -12,8 +13,8 @@ import { const HASH = "a".repeat(64); -function renderImageMarkdown(imagePath: string): string { - const item = renderProviderImageOutputAsAssistantMarkdown({ path: imagePath }); +async function renderImageMarkdown(imagePath: string): Promise { + const item = await renderProviderImageOutputAsAssistantMarkdown({ path: imagePath }); if (!item || item.type !== "assistant_message") { throw new Error("Expected provider image output to render as assistant markdown."); } @@ -40,8 +41,8 @@ describe("isProviderImageMarkdown", () => { ).toBe(true); }); - test("emits Windows file paths as file URIs", () => { - const markdown = renderImageMarkdown( + test("emits Windows file paths as file URIs", async () => { + const markdown = await renderImageMarkdown( `C:\\Users\\me\\AppData\\Local\\Temp\\paseo-attachments\\${HASH}.png`, ); @@ -51,22 +52,24 @@ describe("isProviderImageMarkdown", () => { expect(isProviderImageMarkdown(markdown)).toBe(true); }); - test("emits POSIX file paths with spaces as valid file URI markdown", () => { - const markdown = renderImageMarkdown("/home/user/Projects/Project With Spaces/screenshot.png"); + test("emits POSIX file paths with spaces as valid file URI markdown", async () => { + const markdown = await renderImageMarkdown( + "/home/user/Projects/Project With Spaces/screenshot.png", + ); expect(markdown).toBe( "\n\n![Image](file:///home/user/Projects/Project%20With%20Spaces/screenshot.png)\n\n", ); }); - test("encodes URI-significant characters in POSIX file paths", () => { - const markdown = renderImageMarkdown("/tmp/screenshot#1?draft.png"); + test("encodes URI-significant characters in POSIX file paths", async () => { + const markdown = await renderImageMarkdown("/tmp/screenshot#1?draft.png"); expect(markdown).toBe("\n\n![Image](file:///tmp/screenshot%231%3Fdraft.png)\n\n"); }); - test("preserves double-leading slashes in POSIX file paths", () => { - const markdown = renderImageMarkdown("//tmp/screenshot#1.png"); + test("preserves double-leading slashes in POSIX file paths", async () => { + const markdown = await renderImageMarkdown("//tmp/screenshot#1.png"); expect(markdown).toBe("\n\n![Image](file:////tmp/screenshot%231.png)\n\n"); }); @@ -78,8 +81,8 @@ describe("isProviderImageMarkdown", () => { "\\\\?\\UNC\\server\\share\\shot?draft.png", "file://server/share/shot%3Fdraft.png", ], - ])("encodes %s image paths as file URIs", (_label, imagePath, expectedSource) => { - expect(renderImageMarkdown(imagePath)).toBe(`\n\n![Image](${expectedSource})\n\n`); + ])("encodes %s image paths as file URIs", async (_label, imagePath, expectedSource) => { + expect(await renderImageMarkdown(imagePath)).toBe(`\n\n![Image](${expectedSource})\n\n`); }); test("rejects user-authored markdown that is not a materialized attachment", () => { @@ -92,8 +95,8 @@ describe("isProviderImageMarkdown", () => { }); describe("materializeProviderImage", () => { - test("recreates the private temp directory if the cached directory is removed", () => { - const first = materializeProviderImage({ + test("recreates the private temp directory if the cached directory is removed", async () => { + const first = await materializeProviderImage({ data: "YWJjMTIz", mimeType: "image/png", }); @@ -102,7 +105,7 @@ describe("materializeProviderImage", () => { rmSync(firstDir, { recursive: true, force: true }); - const second = materializeProviderImage({ + const second = await materializeProviderImage({ data: "ZGVmNDU2", mimeType: "image/png", }); @@ -117,8 +120,8 @@ describe("materializeProviderImage", () => { }); describe("provider image paragraph boundaries", () => { - test.each(["", "\n"])("keeps image outputs separate from prose ending in %j", (ending) => { - const image = renderProviderImageOutputAsAssistantMarkdown({ + test.each(["", "\n"])("keeps image outputs separate from prose ending in %j", async (ending) => { + const image = await renderProviderImageOutputAsAssistantMarkdown({ url: "https://example.com/chart.png", }); if (!image) throw new Error("Expected image output"); @@ -152,10 +155,10 @@ describe("provider image paragraph boundaries", () => { }); }); -test("materialized image output retains paragraph boundaries and attachment recognition", () => { - const item = renderProviderImageOutputAsAssistantMarkdown( +test("materialized image output retains paragraph boundaries and attachment recognition", async () => { + const item = await renderProviderImageOutputAsAssistantMarkdown( { data: "YWJjMTIz", mimeType: "image/png" }, - { materialize: () => ({ path: `/tmp/paseo-attachments/${HASH}.png` }) }, + { materialize: async () => ({ path: `/tmp/paseo-attachments/${HASH}.png` }) }, ); expect(item).toEqual({ type: "assistant_message", @@ -164,3 +167,76 @@ test("materialized image output retains paragraph boundaries and attachment reco if (!item || item.type !== "assistant_message") throw new Error("Expected image output"); expect(isProviderImageMarkdown(item.text)).toBe(true); }); + +test("repeated content is reused without rewriting the attachment", async () => { + const image = { + data: Buffer.from("dedupe regression").toString("base64"), + mimeType: "image/png", + }; + const first = await materializeProviderImage(image); + const before = statSync(first.path, { bigint: true }); + await new Promise((resolve) => setTimeout(resolve, 20)); + const second = await materializeProviderImage(image); + expect(second.path).toBe(first.path); + expect(statSync(second.path, { bigint: true }).mtimeNs).toBe(before.mtimeNs); +}); + +test("64 MiB materialization leaves the event loop responsive and preserves bytes", async () => { + const bytes = Buffer.alloc(64 * 1024 * 1024, 37); + const data = bytes.toString("base64"); + let previous = performance.now(); + let longest = 0; + const probe = setInterval(() => { + const now = performance.now(); + longest = Math.max(longest, now - previous); + previous = now; + }, 1); + try { + const result = await materializeProviderImage({ data, mimeType: "image/png" }); + await new Promise((resolve) => setTimeout(resolve, 5)); + console.log(`64 MiB attachment longest loop gap: ${longest.toFixed(2)} ms`); + expect(longest).toBeLessThan(50); + expect(readFileSync(result.path).equals(bytes)).toBe(true); + rmSync(result.path, { force: true }); + } finally { + clearInterval(probe); + } +}); + +test("preserves base64 decoding across whitespace and chunk boundaries", async () => { + const bytes = Buffer.alloc(512 * 1024, 123); + const base64 = bytes.toString("base64"); + const data = base64.match(/.{1,77}/g)!.join("\n"); + const result = await materializeProviderImage({ data, mimeType: "image/png" }); + expect(readFileSync(result.path).equals(bytes)).toBe(true); +}); + +test("rejects oversized materialization before work and leaves the queue usable", async () => { + const data = "A".repeat(Math.ceil(MAX_PROVIDER_IMAGE_BYTES / 3) * 4 + 260); + await expect(materializeProviderImage({ data, mimeType: "image/png" })).rejects.toThrow( + "exceeds", + ); + const omitted = await renderProviderImageOutputAsAssistantMarkdown( + { data, mimeType: "image/png" }, + { materialize: materializeProviderImage }, + ); + expect(omitted).toEqual({ + type: "assistant_message", + text: "Image output was omitted because it was not available as a file path or URL.", + }); + const image = await materializeProviderImage({ data: "YWJjMTIz", mimeType: "image/png" }); + expect(readFileSync(image.path).toString()).toBe("abc123"); +}); + +test("simultaneous requests for the same content share the published file", async () => { + const image = { + data: Buffer.from("concurrent dedupe fixture").toString("base64"), + mimeType: "image/png", + }; + const results = await Promise.all([ + materializeProviderImage(image), + materializeProviderImage(image), + ]); + expect(results[0]).toEqual(results[1]); + expect(readFileSync(results[0].path).toString()).toBe("concurrent dedupe fixture"); +}); diff --git a/packages/server/src/server/agent/providers/provider-image-output.ts b/packages/server/src/server/agent/providers/provider-image-output.ts index ba8843918b..0cd0787c1e 100644 --- a/packages/server/src/server/agent/providers/provider-image-output.ts +++ b/packages/server/src/server/agent/providers/provider-image-output.ts @@ -1,5 +1,6 @@ import { createHash } from "node:crypto"; -import * as fsSync from "node:fs"; +import * as fs from "node:fs/promises"; +import { setImmediate } from "node:timers/promises"; import os from "node:os"; import path from "node:path"; @@ -23,32 +24,26 @@ const PRIVATE_ATTACHMENT_DIR_MODE = 0o700; const MATERIALIZED_IMAGE_FILE_MODE = 0o600; let materializedImageAttachmentDir: string | null = null; +// Serialize materializations: one decoder and one file operation at a time, including +// concurrent history replay and live turns. A rejected operation must not poison the queue. +let pendingMaterialization: Promise = Promise.resolve(); -function canReuseMaterializedImageAttachmentDir(dir: string): boolean { - try { - const stats = fsSync.lstatSync(dir); - if (!stats.isDirectory()) { - return false; +async function getMaterializedImageAttachmentDir(): Promise { + if (materializedImageAttachmentDir) { + try { + const stats = await fs.lstat(materializedImageAttachmentDir); + if (stats.isDirectory()) { + await fs.chmod(materializedImageAttachmentDir, PRIVATE_ATTACHMENT_DIR_MODE); + return materializedImageAttachmentDir; + } + } catch { + /* Temp storage may have been removed between turns. */ } - fsSync.chmodSync(dir, PRIVATE_ATTACHMENT_DIR_MODE); - return true; - } catch { - return false; } -} - -function getMaterializedImageAttachmentDir(): string { - if ( - materializedImageAttachmentDir && - canReuseMaterializedImageAttachmentDir(materializedImageAttachmentDir) - ) { - return materializedImageAttachmentDir; - } - - materializedImageAttachmentDir = fsSync.mkdtempSync( + materializedImageAttachmentDir = await fs.mkdtemp( path.join(os.tmpdir(), PROVIDER_IMAGE_ATTACHMENT_DIR_PREFIX), ); - fsSync.chmodSync(materializedImageAttachmentDir, PRIVATE_ATTACHMENT_DIR_MODE); + await fs.chmod(materializedImageAttachmentDir, PRIVATE_ATTACHMENT_DIR_MODE); return materializedImageAttachmentDir; } @@ -71,32 +66,92 @@ function getImageExtension(mimeType: string): string { } } -function normalizeImageData(mimeType: string, data: string): { mimeType: string; data: string } { - if (data.startsWith("data:")) { - const match = data.match(/^data:([^;]+);base64,(.*)$/); - if (match) { - return { mimeType: match[1], data: match[2] }; - } - } - return { mimeType, data }; -} +// 128 MiB admits the measured 64 MiB fixture with headroom while bounding disk +// and decoding per image. This is materialization admission, not a wire schema limit. +export const MAX_PROVIDER_IMAGE_BYTES = 128 * 1024 * 1024; +const BASE64_CHUNK_CHARACTERS = 256 * 1024; // Multiple of four; decode units stay aligned. -// Filenames are a content hash of the bytes so re-materializing the same image -// within a process reuses the existing temp file instead of leaking a fresh one -// for repeated image blocks or history replay. export function materializeProviderImage(image: { data: string; mimeType: string | null; -}): MaterializedProviderImage { - const attachmentsDir = getMaterializedImageAttachmentDir(); - const normalized = normalizeImageData(image.mimeType ?? "image/png", image.data); - const bytes = Buffer.from(normalized.data, "base64"); - const extension = getImageExtension(normalized.mimeType); - const hash = createHash("sha256").update(bytes).digest("hex"); - const filePath = path.join(attachmentsDir, `${hash}.${extension}`); - fsSync.writeFileSync(filePath, bytes, { mode: MATERIALIZED_IMAGE_FILE_MODE }); - fsSync.chmodSync(filePath, MATERIALIZED_IMAGE_FILE_MODE); - return { path: filePath }; +}): Promise { + const result = pendingMaterialization.then(() => materializeImage(image)); + pendingMaterialization = result.catch(() => {}); + return result; +} + +async function materializeImage(image: { + data: string; + mimeType: string | null; +}): Promise { + const maxEncodedCharacters = Math.ceil(MAX_PROVIDER_IMAGE_BYTES / 3) * 4; + // Check length before inspecting/copying the payload, including a bounded data URI header. + if (image.data.length > maxEncodedCharacters + 256) { + throw new Error(`Image attachment exceeds ${MAX_PROVIDER_IMAGE_BYTES} bytes`); + } + let data = image.data; + let mimeType = image.mimeType ?? "image/png"; + if (data.startsWith("data:")) { + const separator = data.indexOf(";base64,"); + if (separator >= 0 && separator < 256) { + mimeType = data.slice(5, separator); + data = data.slice(separator + 8); + } + } + if (data.length > maxEncodedCharacters) { + throw new Error(`Image attachment exceeds ${MAX_PROVIDER_IMAGE_BYTES} bytes`); + } + const hash = createHash("sha256"); + for await (const bytes of decodeImage(data)) hash.update(bytes); + const dir = await getMaterializedImageAttachmentDir(); + const filePath = path.join(dir, `${hash.digest("hex")}.${getImageExtension(mimeType)}`); + try { + await fs.stat(filePath); + return { path: filePath }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + + const staging = await fs.mkdtemp(path.join(dir, ".pending-")); + const stagingPath = path.join(staging, "image"); + try { + await fs.chmod(staging, PRIVATE_ATTACHMENT_DIR_MODE); + const file = await fs.open(stagingPath, "wx", MATERIALIZED_IMAGE_FILE_MODE); + try { + await file.chmod(MATERIALIZED_IMAGE_FILE_MODE); + // Decode again only for a new hash. This bounds retained memory to one chunk + // and performs no writes at all for repeated content, including history replay. + for await (const bytes of decodeImage(data)) await file.writeFile(bytes); + } finally { + await file.close(); + } + await fs.rename(stagingPath, filePath); + return { path: filePath }; + } finally { + await fs.rm(staging, { recursive: true, force: true }); + } +} + +async function* decodeImage(data: string): AsyncGenerator { + let carry = ""; + let decodedBytes = 0; + for (let offset = 0; offset < data.length; offset += BASE64_CHUNK_CHARACTERS) { + // Buffer.from ignores non-base64 characters. Carry incomplete quartets so + // line-wrapped payloads decode identically across chunk boundaries. + const chunk = + carry + + data.slice(offset, offset + BASE64_CHUNK_CHARACTERS).replace(/[^A-Za-z0-9+/_=-]/g, ""); + const finalChunk = offset + BASE64_CHUNK_CHARACTERS >= data.length || chunk.includes("="); + const end = finalChunk ? chunk.length : chunk.length - (chunk.length % 4); + const bytes = Buffer.from(chunk.slice(0, end), "base64"); + carry = chunk.slice(end); + decodedBytes += bytes.length; + if (decodedBytes > MAX_PROVIDER_IMAGE_BYTES) + throw new Error("Image attachment exceeds size limit"); + yield bytes; + await setImmediate(); + if (finalChunk) break; + } } // Recognizes markdown rendered for a materialized provider image: its source is a content-hashed @@ -112,7 +167,10 @@ export function isProviderImageMarkdown(text: string): boolean { } interface RenderProviderImageOutputOptions { - materialize?: (image: { data: string; mimeType: string | null }) => MaterializedProviderImage; + materialize?: (image: { + data: string; + mimeType: string | null; + }) => Promise; } function nonEmptyString(value: string | null | undefined): string | null { @@ -169,10 +227,10 @@ function escapeMarkdownImageSource(value: string): string { return markdownImageSource(value).replace(/\\/g, "\\\\").replace(/\)/g, "\\)"); } -export function renderProviderImageOutputAsAssistantMarkdown( +export async function renderProviderImageOutputAsAssistantMarkdown( image: ProviderImageOutput, options: RenderProviderImageOutputOptions = {}, -): AgentTimelineItem | null { +): Promise { const source = nonEmptyString(image.path) ?? nonEmptyString(image.url); if (source && !isDataImageSource(source)) { const altText = escapeMarkdownImageAlt(nonEmptyString(image.altText) ?? "Image"); @@ -191,7 +249,7 @@ export function renderProviderImageOutputAsAssistantMarkdown( let materialized: MaterializedProviderImage | null = null; try { materialized = options.materialize - ? options.materialize({ + ? await options.materialize({ data, mimeType: nonEmptyString(image.mimeType), })