From d80f0e958d4b7aad7bd432332fcd5985431825f8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 26 Oct 2026 16:38:23 +0100 Subject: [PATCH] fix(server): keep image materialization responsive Decode and hash attachments in bounded yielding units, deduplicate before writing, and publish complete private files before provider timeline events. Await mapping through existing replay boundaries and serialize notification dispatch so turn completion cannot overtake an image. --- .../claude/agent.image-rendering.test.ts | 12 +- .../agent/providers/claude/agent.test.ts | 78 +-- .../server/agent/providers/claude/agent.ts | 127 +++-- .../claude/subagents/replay-source.test.ts | 37 ++ .../claude/subagents/replay-source.ts | 6 +- .../subagents/workflow-replay-source.test.ts | 32 ++ .../subagents/workflow-replay-source.ts | 6 +- .../providers/codex-app-server-agent.test.ts | 469 ++++++++++++------ .../agent/providers/codex-app-server-agent.ts | 133 +++-- .../server/agent/providers/omp/agent.test.ts | 37 ++ .../src/server/agent/providers/omp/agent.ts | 19 +- .../test-utils/v2-streamed-text-memory.ts | 22 +- .../agent/providers/opencode/v2/children.ts | 2 +- .../agent/providers/opencode/v2/session.ts | 6 +- .../providers/opencode/v2/streaming.test.ts | 14 +- .../providers/opencode/v2/timeline.test.ts | 10 +- .../agent/providers/opencode/v2/timeline.ts | 30 +- .../server/agent/providers/pi/agent.test.ts | 11 + .../src/server/agent/providers/pi/agent.ts | 17 +- .../provider-image-output.posix.test.ts | 4 +- .../providers/provider-image-output.test.ts | 118 ++++- .../agent/providers/provider-image-output.ts | 154 ++++-- 22 files changed, 909 insertions(+), 435 deletions(-) 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), })