From 8bbd6a30d0a28ff50d9f091bb24c25eb9cffbcbd Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Thu, 22 Oct 2026 03:46:34 +0000 Subject: [PATCH 1/6] fix(server): Codex generated images show in chat Rebased onto current main as one commit (the original 18 commits conflicted with main's reshaped adapter and run-execution tests). Behavior is unchanged: saved imageGeneration results are appended to the final answer, late completions keep ingestion open through background_work.updated, and failed or unsaved results are dropped. Co-authored-by: Julius Marminge Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 317 +++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 119 +++++++ .../orchestration-v2/ProviderEventIngestor.ts | 2 + .../RunExecutionService.test.ts | 171 +++++++++- .../orchestration-v2/RunExecutionService.ts | 13 + .../src/server/ProviderAdapter.ts | 8 + packages/shared/package.json | 4 + packages/shared/src/generatedImageMarkdown.ts | 136 ++++++++ 8 files changed, 766 insertions(+), 4 deletions(-) create mode 100644 packages/shared/src/generatedImageMarkdown.ts diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index c20994423f3e9..4ab826a241693 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1,3 +1,5 @@ +import * as NodeModule from "node:module"; + import { historyResponseItems } from "@t3tools/provider-core/server/handoffBudget"; import { makeProviderTextDeltaCoalescer, @@ -29,7 +31,7 @@ import { ThreadId, TurnItemId, } from "@t3tools/contracts"; -import { assert, describe, it } from "@effect/vitest"; +import { assert, describe, expect, it } from "@effect/vitest"; import * as CodexClient from "effect-codex-app-server/client"; import * as CodexReplay from "effect-codex-app-server/replay"; import * as CodexError from "effect-codex-app-server/errors"; @@ -77,6 +79,7 @@ const encodeStringJson = Schema.encodeEffect(Schema.fromJsonString(Schema.String const isProviderAdapterEventStreamError = Schema.is( ProviderAdapter.ProviderAdapterEventStreamError, ); +const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); describe("CodexAdapterV2 context usage", () => { it("uses the current context rather than cumulative processed tokens", () => { @@ -1086,6 +1089,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { Effect.void, onRequest: (method: string, params: unknown) => Effect.Effect = () => Effect.void, readChildMetadata?: Parameters[2], + beforeEmitInbound?: CodexReplay.CodexAppServerReplayDriver["beforeEmitInbound"], ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1096,9 +1100,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { const changes = yield* makeChangeSignal; const continuationRequests: Array = []; + const driver = yield* CodexReplay.makeReplayDriver( + transcript, + beforeEmitInbound === undefined ? {} : { beforeEmitInbound }, + ); const clientFactory: CodexAdapterV2.CodexAppServerClientFactoryShape = { open: (openInput) => - Layer.build(CodexReplay.layerReplay(transcript)).pipe( + Layer.build(CodexReplay.layerReplayWithDriver(driver)).pipe( Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterOpenSessionError({ @@ -3237,6 +3245,311 @@ describe("CodexAdapterV2 post-settle continuation", () => { }); }; + it.effect.each([ + { + name: "multiple saved images appear once in a text-only reply", + reply: "Done.", + images: [ + { savedPath: "/tmp/a.png" }, + { savedPath: "/tmp/b.png" }, + { savedPath: "/tmp/a.png" }, + ], + expected: "Done.\n\n![]()\n\n![]()", + }, + { + name: "a link or quoted Markdown still needs a preview", + reply: + "Saved /tmp/a.png: [open](/tmp/a.png)\n`![](/tmp/a.png)`\n\n ![](/tmp/a.png)\n\n```md\n![](/tmp/a.png)\n```\n\\![](/tmp/a.png)", + images: [{ savedPath: "/tmp/a.png" }], + expected: + "Saved /tmp/a.png: [open](/tmp/a.png)\n`![](/tmp/a.png)`\n\n ![](/tmp/a.png)\n\n```md\n![](/tmp/a.png)\n```\n\\![](/tmp/a.png)\n\n![]()", + }, + { + name: "an unclosed trailing fence leaves the generated image visible", + reply: 'Created the image.\n\n~~~js\nconsole.log("done");', + images: [{ savedPath: "/tmp/generated.png" }], + expected: + 'Created the image.\n\n~~~js\nconsole.log("done");\n~~~\n\n![]()', + expectedImage: "/tmp/generated.png", + }, + { + name: "existing images are not duplicated", + reply: "![A]() ![B](/tmp/b.png) ![C][generated]\n\n[generated]: /tmp/c.png", + images: [ + { savedPath: "/tmp/a b.png" }, + { savedPath: "/tmp/b.png" }, + { savedPath: "/tmp/c.png" }, + { savedPath: "/tmp/B.png" }, + ], + expected: + "![A]() ![B](/tmp/b.png) ![C][generated]\n\n[generated]: /tmp/c.png\n\n![]()", + }, + { + name: "Windows drive paths survive mobile Markdown parsing", + reply: "Done.", + images: [{ savedPath: "C:\\Users\\me\\.codex\\a b.png" }], + expected: "Done.\n\n![]()", + }, + { + name: "existing Windows embeds are normalized without adding duplicates", + reply: + "![A](c:\\Users\\me\\.codex\\a.png) ![B](c:/users/me/.codex/b.png) ![C][unc]\n\n[unc]: <\\\\HOST\\Share\\.codex\\c.png>\n\n![D](file://host/share/.codex/d.png)", + images: [ + { savedPath: "C:\\Users\\Me\\.codex\\a.png" }, + { savedPath: "C:\\Users\\Me\\.codex\\b.png" }, + { savedPath: "\\\\host\\share\\.codex\\c.png" }, + { savedPath: "\\\\HOST\\Share\\.codex\\d.png" }, + ], + expected: + "![A]() ![B](c:/users/me/.codex/b.png) ![C][unc]\n\n[unc]: \n\n![D](file://host/share/.codex/d.png)", + }, + { + name: "reserved path characters are encoded once in new and existing embeds", + reply: + "![raw]() ![encoded](/tmp/a%23b.png) ![percent]() ![percent encoded](/tmp/100%2525.png) ![query][generated]\n\n[generated]: /tmp/q?x.png\n\n![unc](file://host/share/a#b%25.png) ![unc encoded](file://HOST/SHARE/a%23b%2525.png)", + images: [ + { savedPath: "/tmp/a#b.png" }, + { savedPath: "/tmp/100%25.png" }, + { savedPath: "/tmp/q?x.png" }, + { savedPath: "\\\\host\\share\\a#b%25.png" }, + { savedPath: "/tmp/new#image?100%25.png" }, + { savedPath: "/tmp/a©.png" }, + { savedPath: "C:\\Users\\Me#100%25\\new.png" }, + ], + expected: + "![raw]() ![encoded](/tmp/a%23b.png) ![percent]() ![percent encoded](/tmp/100%2525.png) ![query][generated]\n\n[generated]: \n\n![unc]() ![unc encoded](file://HOST/SHARE/a%23b%2525.png)\n\n![]()\n\n![]()\n\n![]()", + }, + { + name: "an image-only UNC result produces a filesystem media message", + reply: null, + images: [{ savedPath: "\\\\host\\share\\.codex\\a.png" }], + expected: "![]()", + }, + { + name: "failed and unsaved results never become media", + reply: "No image.", + images: [ + { savedPath: "/tmp/failed.png", status: "failed" }, + { + savedPath: "/tmp/limited.png", + failure: { type: "usageLimitExceeded", limitId: "image-generation" }, + }, + { savedPath: null }, + { savedPath: "" }, + ], + expected: "No image.", + }, + ])("projects generated images: $name", ({ reply, images, expected, expectedImage }) => + Effect.gen(function* () { + const scenario = "codex-generated-images"; + const result = "aW1hZ2UtYnl0ZXMtbXVzdC1uZXZlci1yZWFjaC1jbGllbnRz"; + const base = finalAnswerTranscript( + scenario, + reply === null ? [] : [{ id: "answer", text: reply }], + ); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario, + entries: [ + ...base.entries.slice(0, -1), + ...images.map((image, index) => ({ + type: "emit_inbound" as const, + frame: { + method: "item/completed", + params: { + threadId: `native-${scenario}-thread`, + turnId: `native-${scenario}-turn`, + item: { + type: "imageGeneration", + id: `image-${index}`, + status: "completed", + result, + ...image, + }, + }, + }, + })), + ...base.entries.slice(-1), + ], + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-generated-images"), + text: "Reply with the requested recovery marker.", + }), + ); + yield* harness.firstTerminal; + + const messages = assistantMessages(harness.events); + const final = messages.at(-1)?.message; + if (expectedImage !== undefined) { + // Resolve the shared package's real parser without adding server runtime dependencies. + const requireMarkdown = NodeModule.createRequire( + import.meta.resolve("@t3tools/shared/generatedImageMarkdown"), + ); + const parser: { parse(text: string): unknown } = requireMarkdown("unified") + .unified() + .use(requireMarkdown("remark-parse").default); + expect(parser.parse(final?.text ?? "")).toMatchObject({ + children: [ + { type: "paragraph" }, + { type: "code", value: 'console.log("done");' }, + { type: "paragraph", children: [{ type: "image", url: expectedImage }] }, + ], + }); + } + assert.strictEqual(final?.text, expected); + assert.strictEqual(final?.streaming, false); + assert.strictEqual(new Set(messages.map(({ message }) => message.id)).size, 1); + assert.notInclude(encodeUnknownJson(harness.events), result); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect.each(["complete", "stop"] as const)( + "retains a generating image after settlement until %s", + (outcome) => + Effect.gen(function* () { + const scenario = "codex-retained-image"; + const base = finalAnswerTranscript(scenario, [{ id: "answer", text: "Done." }]); + const completion = yield* Deferred.make(); + const secondCompletion = yield* Deferred.make(); + const projected = yield* Deferred.make(); + const drained = yield* Deferred.make(); + const expected = "Done.\n\n![]()"; + const imageEvent = (phase: "started" | "completed", second = false) => ({ + type: "emit_inbound" as const, + label: `image-${phase}${second ? "-second" : ""}`, + frame: { + method: `item/${phase}`, + params: { + threadId: `native-${scenario}-thread`, + turnId: `native-${scenario}-turn`, + item: { + type: "imageGeneration", + id: second ? "second-image" : "late-image", + status: phase === "started" ? "inProgress" : "completed", + result: "", + ...(phase === "completed" + ? { savedPath: second ? "/tmp/second.png" : "/tmp/late image.png" } + : {}), + }, + }, + }, + }); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario, + entries: [ + ...base.entries.slice(0, -2), + imageEvent("started"), + imageEvent("started", true), + ...base.entries.slice(-2), + imageEvent("completed"), + imageEvent("completed", true), + ], + }), + (event) => + event.type === "message.updated" && event.message.text === expected + ? Deferred.succeed(projected, undefined) + : event.type === "background_work.updated" && !event.pending + ? Deferred.succeed(drained, undefined) + : Effect.void, + undefined, + undefined, + (entry) => + entry.label === "image-completed" + ? Deferred.await(completion) + : entry.label === "image-completed-second" + ? Deferred.await(secondCompletion) + : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-retained-image"), + text: "Reply with the requested recovery marker.", + }), + ); + yield* harness.firstTerminal; + const threadPending = harness.runtime.hasPendingBackgroundWorkForThread!( + harness.providerThread, + ); + const pendingBefore = [yield* threadPending, yield* harness.hasPendingBackgroundWork]; + assert.strictEqual(assistantMessages(harness.events).at(-1)?.message.text, "Done."); + + if (outcome === "complete") { + yield* Deferred.succeed(completion, undefined); + yield* Deferred.await(projected); + assert.deepEqual( + [yield* threadPending, yield* harness.hasPendingBackgroundWork], + [true, true], + ); + assert.isFalse( + harness.events.some( + (event) => event.type === "background_work.updated" && !event.pending, + ), + ); + yield* Deferred.succeed(secondCompletion, undefined); + yield* Deferred.await(drained); + assert.strictEqual( + assistantMessages(harness.events).at(-1)?.message.text, + `${expected}\n\n![]()`, + ); + const messages = assistantMessages(harness.events); + assert.strictEqual(messages.at(-1)?.message.id, messages[0]?.message.id); + assert.strictEqual(messages.at(-1)?.message.streaming, false); + } else { + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: harness.terminalEvents()[0]!.providerTurnId, + requestRuntimeRestart: true, + }); + } + yield* Deferred.await(drained); + assert.deepEqual( + [yield* threadPending, yield* harness.hasPendingBackgroundWork], + [false, false], + ); + const started = harness.events.findIndex( + (event) => event.type === "background_work.updated" && event.pending, + ); + const terminal = harness.events.findIndex((event) => event.type === "turn.terminal"); + const finished = harness.events.findIndex( + (event) => event.type === "background_work.updated" && !event.pending, + ); + assert.isAtLeast(started, 0); + assert.isAbove(terminal, started); + assert.isAbove(finished, terminal); + if (outcome === "complete") { + assert.isAbove( + finished, + harness.events.findLastIndex( + (event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message", + ), + ); + } + assert.lengthOf(harness.terminalEvents(), 1); + assert.deepEqual(pendingBefore, [true, true]); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("suppresses a trailing empty final answer after a non-empty final answer", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 51a686e270389..6e933d74659e3 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -47,6 +47,7 @@ import { dynamicToolTitle, } from "@t3tools/shared/toolActivity"; import { getModelSelectionStringOptionValue, modelSelectionsEqual } from "@t3tools/shared/model"; +import { appendGeneratedImageEmbeds } from "@t3tools/shared/generatedImageMarkdown"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; import type { ChatAttachment, @@ -1842,6 +1843,23 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const offeredContinuationItemsByTurn = yield* Ref.make(new Map>()); const finalAnswerItemIdsByTurn = yield* Ref.make(new Map>()); const completedFinalAnswerTextsByTurn = yield* Ref.make(new Map>()); + const imageTurns = new Map< + string, + { + readonly running: Set; + readonly saved: Map; + pending: boolean; + answer?: { readonly id: string; readonly text: string }; + } + >(); + const imageTurn = (nativeTurnId: string) => { + let state = imageTurns.get(nativeTurnId); + if (state === undefined) { + state = { running: new Set(), saved: new Map(), pending: false }; + imageTurns.set(nativeTurnId, state); + } + return state; + }; // Serialize native completion and interrupt timeout so only one publishes terminal events. const turnTerminalizationPermit = yield* Semaphore.make(1); const goalsByNativeThread = new Map(); @@ -1868,6 +1886,26 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); + const emitImageWorkState = Effect.fnUntraced(function* ( + context: ActiveCodexTurnContext, + pending: boolean, + ) { + const state = imageTurns.get(context.nativeTurnId); + if (state === undefined || state.pending === pending) return; + state.pending = pending; + yield* emitProviderEvent({ + type: "background_work.updated", + driver: CODEX_PROVIDER, + providerTurnId: context.providerTurnId, + pending, + }); + }); + + const clearImageWork = Effect.fnUntraced(function* (context: ActiveCodexTurnContext) { + yield* emitImageWorkState(context, false); + imageTurns.delete(context.nativeTurnId); + }); + // Setting a goal passes through cleared and paused states; publish only the result. const goalUpdatesHeld = new Set(); const emitGoalUpdate = Effect.fnUntraced(function* (nativeThreadId: string) { @@ -2136,6 +2174,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if ((yield* Ref.get(pendingMcpAppCaptures)).has(nativeTurnId)) { return true; } + const images = imageTurns.get(nativeTurnId); + if (images !== undefined && (images.running.size > 0 || images.saved.size > 0)) { + return true; + } const tools = (yield* Ref.get(runningDynamicToolsByTurn)).get(nativeTurnId); if (tools === undefined) { return false; @@ -2178,6 +2220,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( updated.delete(nativeTurnId); return updated; }); + imageTurns.delete(nativeTurnId); yield* Ref.update(finalAnswerItemIdsByTurn, (current) => { if (!current.has(nativeTurnId)) { return current; @@ -3389,6 +3432,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( turnItem: artifacts.turnItem, }); if (finalAnswerItem && update.completed) { + imageTurn(update.turnId).answer = { id: update.itemId, text: update.text }; yield* Ref.update(completedFinalAnswerTextsByTurn, (current) => { const updated = new Map(current); const texts = new Set(updated.get(update.turnId) ?? []); @@ -4714,6 +4758,14 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } + if (payload.item.type === "imageGeneration") { + if (!(yield* Ref.get(interruptingNativeTurns)).has(payload.turnId)) { + imageTurn(payload.turnId).running.add(payload.item.id); + yield* emitImageWorkState(context, true); + } + return; + } + if (payload.item.type === "commandExecution") { if (!codexItemStatus(payload.item.status).completed) { yield* trackRunningCommandItem(payload.turnId, { @@ -4992,6 +5044,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } + if (payload.item.type === "imageGeneration") { + if ((yield* Ref.get(interruptingNativeTurns)).has(payload.turnId)) return; + const { id, savedPath, status, failure } = payload.item; + const state = imageTurn(payload.turnId); + if (savedPath?.trim() && status === "completed" && failure == null) { + state.saved.set(id, savedPath); + } + // Idle release runs concurrently, so keep this work pending through projection. + if (settled) { + yield* projectGeneratedImages(context, payload.turnId); + } + state.running.delete(id); + yield* emitImageWorkState(context, state.running.size > 0 || state.saved.size > 0); + if (settled) { + yield* releaseSettledTurnIfIdle(payload.turnId); + } + return; + } + if (payload.item.type !== "agentMessage") { return; } @@ -5846,6 +5917,40 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, ); + const projectGeneratedImages = Effect.fnUntraced(function* ( + context: ActiveCodexTurnContext, + nativeTurnId: string, + ) { + const state = imageTurns.get(nativeTurnId); + const firstImageItemId = state?.saved.keys().next().value; + if (state === undefined || firstImageItemId === undefined) return; + const finalAnswer = state.answer; + const text = appendGeneratedImageEmbeds(finalAnswer?.text ?? "", state.saved.values()); + if (text === (finalAnswer?.text ?? "")) { + state.saved.clear(); + return; + } + const message = { id: finalAnswer?.id ?? firstImageItemId, text }; + state.answer = message; + const artifacts = yield* buildAgentMessageArtifacts(context, message, true); + yield* emitProviderEvent({ + type: "node.updated", + driver: CODEX_PROVIDER, + node: artifacts.node, + }); + yield* emitProviderEvent({ + type: "message.updated", + driver: CODEX_PROVIDER, + message: artifacts.message, + }); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: artifacts.turnItem, + }); + state.saved.clear(); + }); + const finalizeCodexTurn = (input: { readonly context: ActiveCodexTurnContext; readonly nativeTurnId: string; @@ -5919,6 +6024,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( input.completedAt, ); } + yield* projectGeneratedImages(input.context, input.nativeTurnId); + if (input.status !== "completed") { + yield* clearImageWork(input.context); + } else { + const images = imageTurns.get(input.nativeTurnId); + if (images !== undefined) { + yield* emitImageWorkState(input.context, images.running.size > 0); + } + } const nativeThreadId = input.context.providerThread.nativeThreadRef?.nativeId; const activation = nativeThreadId == null ? undefined : goalActivations.get(nativeThreadId); @@ -6139,6 +6253,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( updated.delete(input.nativeTurnId); return updated; }); + imageTurns.delete(input.nativeTurnId); yield* Ref.update(finalAnswerItemIdsByTurn, (current) => { if (!current.has(input.nativeTurnId)) { return current; @@ -6748,6 +6863,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if ((yield* Ref.get(pendingMcpAppCaptures)).size > 0) { return true; } + for (const state of imageTurns.values()) { + if (state.running.size > 0 || state.saved.size > 0) return true; + } for (const subagent of (yield* Ref.get(subagentThreads)).values()) { if (subagent.task.status === "running") { return true; @@ -7418,6 +7536,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( updated.delete(context.nativeTurnId); return updated; }); + yield* clearImageWork(context); yield* releaseSettledTurnIfIdle(context.nativeTurnId); } }), diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 06f444776cb3c..5a2aa39faece4 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -700,6 +700,8 @@ export const layer: Layer.Layer< }), ]; } + case "background_work.updated": + return []; case "command_output.delta": // Live output is routed to the command output hub before ingestion. return []; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 394cc1527cb7d..b96a24fa1e47d 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1780,6 +1780,160 @@ it.effect("ends a nested native subagent and its thread when the root run is int }), ); +it.effect( + "ingests every queued generated-image update when pending work clears during root finalization", + () => + Effect.gen(function* () { + const key = "late-generated-image-chunks"; + const ids = backgroundScenarioIds(key); + const now = yield* DateTime.now; + const pending = yield* Ref.make(true); + const ingested = yield* Ref.make>([]); + const nodeId = NodeId.make(`node:${key}:assistant`); + const messageId = MessageId.make(`message:${key}:assistant`); + const text = "Here is the image."; + const imageText = `${text}\n\n![Generated image](file:///tmp/codex-08/a%23b%2525%3F.png)`; + const nodeEvent = { + type: "node.updated", + driver, + node: { + id: nodeId, + threadId: ids.threadId, + runId: ids.runId, + parentNodeId: ids.rootNodeId, + rootNodeId: ids.rootNodeId, + kind: "assistant_message", + status: "completed", + countsForRun: false, + providerThreadId: ids.providerThreadId, + providerTurnId: ids.rootProviderTurnId, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: now, + }, + } satisfies ProviderAdapter.ProviderAdapterV2Event; + const messageEvent = { + type: "message.updated", + driver, + message: { + id: messageId, + threadId: ids.threadId, + runId: ids.runId, + nodeId, + role: "assistant", + createdBy: "agent", + creationSource: "provider", + text, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + } satisfies ProviderAdapter.ProviderAdapterV2Event; + const itemEvent = { + type: "turn_item.updated", + driver, + turnItem: { + id: ids.itemId, + threadId: ids.threadId, + runId: ids.runId, + nodeId, + providerThreadId: ids.providerThreadId, + providerTurnId: ids.rootProviderTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "assistant_message", + messageId, + text, + streaming: false, + }, + } satisfies ProviderAdapter.ProviderAdapterV2Event; + const lateMessage = { + ...messageEvent, + message: { ...messageEvent.message, text: imageText }, + }; + const lateItem = { ...itemEvent, turnItem: { ...itemEvent.turnItem, text: imageText } }; + const bothImagesText = `${imageText}\n\n![Generated image](file:///tmp/codex-08/second.png)`; + const secondLateMessage = { + ...messageEvent, + message: { ...messageEvent.message, text: bothImagesText }, + }; + const secondLateItem = { + ...itemEvent, + turnItem: { ...itemEvent.turnItem, text: bothImagesText }, + }; + const terminal = rootTerminalEvent(ids, "completed"); + const backgroundStarted = { + type: "background_work.updated", + driver, + providerTurnId: ids.rootProviderTurnId, + pending: true, + } satisfies ProviderAdapter.ProviderAdapterV2Event; + const backgroundCompleted = { ...backgroundStarted, pending: false }; + + const observed = yield* runBackgroundItemScenario( + key, + () => + Stream.fromIterable([ + backgroundStarted, + nodeEvent, + messageEvent, + itemEvent, + terminal, + ]).pipe( + // Both completions are queued when pending clears during root finalization. + // Separate chunks expose stop checks before the first node and between completions. + Stream.concat(Stream.succeed(nodeEvent)), + Stream.concat(Stream.succeed(lateMessage)), + Stream.concat(Stream.succeed(lateItem)), + Stream.concat(Stream.succeed(nodeEvent)), + Stream.concat(Stream.succeed(secondLateMessage)), + Stream.concat(Stream.succeed(secondLateItem)), + Stream.concat(Stream.succeed(backgroundCompleted)), + ), + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: (providerThread) => { + assert.equal(providerThread.id, ids.providerThreadId); + return Ref.get(pending); + }, + onFinalized: () => Ref.set(pending, false), + onIngest: (event) => Ref.update(ingested, (current) => [...current, event]), + }, + ); + + assert.deepEqual(yield* Ref.get(ingested), [ + backgroundStarted, + nodeEvent, + messageEvent, + itemEvent, + terminal, + nodeEvent, + lateMessage, + lateItem, + nodeEvent, + secondLateMessage, + secondLateItem, + backgroundCompleted, + ]); + assert.deepEqual(observed, [ + "turn_item:completed", + "root-finalized", + "turn_item:completed", + "turn_item:completed", + ]); + assert.isFalse(yield* Ref.get(pending)); + }), +); + it.effect("keeps ingesting a child thread's late background item completion", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-child-item", (ids) => [ @@ -3525,9 +3679,17 @@ function rootTerminalEvent( function runBackgroundItemScenario( key: string, - makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray, + makeEvents: ( + ids: BackgroundScenarioIds, + ) => + | ReadonlyArray + | Stream.Stream, options?: { readonly keepEventStreamOpen?: boolean; + readonly hasPendingBackgroundWorkForThread?: ( + providerThread: OrchestrationV2ProviderThread, + ) => Effect.Effect; + readonly onIngest?: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect< ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }> >; @@ -3577,6 +3739,7 @@ function runBackgroundItemScenario( ingestNormalized: (input) => Effect.gen(function* () { const event = input.event; + yield* options?.onIngest?.(event) ?? Effect.void; if (event.type === "turn_item.updated") { yield* Ref.update(observed, (current) => [ ...current, @@ -3605,9 +3768,13 @@ function runBackgroundItemScenario( providerSessionId: ProviderSessionId.make(`session:${key}`), session: { events: Stream.empty, + ...(options?.hasPendingBackgroundWorkForThread === undefined + ? {} + : { hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread }), subscribeEvents: Effect.gen(function* () { yield* options?.onSubscribe ?? Effect.void; - const events = Stream.fromIterable(makeEvents(ids)).pipe( + const source = makeEvents(ids); + const events = (Stream.isStream(source) ? source : Stream.fromIterable(source)).pipe( Stream.concat( Stream.fromEffect( Effect.andThen( diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 3e3f37a15a936..e6587c308fb0a 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -507,6 +507,8 @@ export function routeProviderEvent( state.ownedProviderTurns.has(event.runtimeRequest.providerTurnId)), state, ]; + case "background_work.updated": + return [state.ownedProviderTurns.has(event.providerTurnId), state]; case "command_output.delta": // The session manager hands live output to the command output hub; runs never see it. return [false, state]; @@ -1030,6 +1032,7 @@ export const layer: Layer.Layer< const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); const providerThreadOwnerLost = yield* Ref.make(false); + const pendingBackgroundProviderTurns = new Set(); const activeChildProviderTurns = yield* Ref.make>(new Set()); const activeChildSubagents = yield* Ref.make>(new Map()); const activeBackgroundTurnItems = yield* Ref.make< @@ -1231,6 +1234,9 @@ export const layer: Layer.Layer< ) { return true; } + if (pendingBackgroundProviderTurns.size > 0) { + return false; + } const childProviderTurns = yield* Ref.get(activeChildProviderTurns); if (childProviderTurns.size > 0) { return false; @@ -1454,6 +1460,13 @@ export const layer: Layer.Layer< yield* Ref.set(rootTerminalSeen, true); yield* finalizeRootRun(event); } + if (event.type === "background_work.updated") { + if (event.pending) { + pendingBackgroundProviderTurns.add(event.providerTurnId); + } else { + pendingBackgroundProviderTurns.delete(event.providerTurnId); + } + } yield* trackChildLifecycle(event, deliveredEvent !== null); }), ), diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 0b5459f647f8d..cadf4173a60b1 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -119,6 +119,14 @@ export const ProviderAdapterV2Event = Schema.Union([ driver: ProviderDriverKind, turnItem: OrchestrationV2TurnItem, }), + // Start before the root terminal; clear only after the final covered projection. + // Consumer lifetime cannot depend on a producer probe that runs ahead of its queue. + Schema.Struct({ + type: Schema.Literal("background_work.updated"), + driver: ProviderDriverKind, + providerTurnId: ProviderTurnId, + pending: Schema.Boolean, + }), /** Unpersisted deltas go only to clients showing the row; final output arrives in turn_item.updated. */ Schema.Struct({ type: Schema.Literal("command_output.delta"), diff --git a/packages/shared/package.json b/packages/shared/package.json index 3171bea334dad..e3d0b2ce57cf7 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -463,6 +463,10 @@ "types": "./src/markdownWindowsPaths.ts", "import": "./src/markdownWindowsPaths.ts" }, + "./generatedImageMarkdown": { + "types": "./src/generatedImageMarkdown.ts", + "import": "./src/generatedImageMarkdown.ts" + }, "./markdownPipeline": { "types": "./src/markdownPipeline.ts", "import": "./src/markdownPipeline.ts" diff --git a/packages/shared/src/generatedImageMarkdown.ts b/packages/shared/src/generatedImageMarkdown.ts new file mode 100644 index 0000000000000..d10d2c74f686a --- /dev/null +++ b/packages/shared/src/generatedImageMarkdown.ts @@ -0,0 +1,136 @@ +import remarkParse from "remark-parse"; +import { unified } from "unified"; + +import { remarkKeepWindowsPaths } from "./markdownWindowsPaths.ts"; +import { parseFileUrlHref } from "./fileLinks.ts"; +import { safeDecodeURIComponent } from "./markdownLinks.ts"; +import { + isUncPath, + isWindowsDrivePath, + normalizeProjectPathForComparison, + stripSlashPrefixedWindowsDrive, +} from "./path.ts"; + +const parser = unified().use(remarkParse).use(remarkKeepWindowsPaths); +interface MarkdownNode { + readonly type: string; + readonly url?: string | undefined; + readonly identifier?: string | undefined; + readonly label?: string | null | undefined; + readonly alt?: string | null | undefined; + readonly title?: string | null | undefined; + readonly position?: + | { + readonly start: { readonly offset?: number | undefined }; + readonly end: { readonly offset?: number | undefined }; + } + | undefined; + readonly children?: ReadonlyArray | undefined; +} + +// Mobile Markdown parsers consume Windows backslashes; UNC paths also need file: to +// distinguish them from protocol-relative web URLs. +function imageDestination(path: string): string { + return isWindowsDrivePath(path) + ? path.replaceAll("\\", "/") + : isUncPath(path) + ? `file:${path.replaceAll("\\", "/")}` + : path; +} + +function angleDestination(path: string): string { + return `<${path.replace(/[%?#&]/g, encodeURIComponent).replace(/[<>]/g, "\\$&")}>`; +} + +export function appendGeneratedImageEmbeds(text: string, savedPaths: Iterable): string { + const destinations = new Map( + Array.from(savedPaths, (path) => [normalizeProjectPathForComparison(path), path]), + ); + const missing = new Map(destinations); + const definitions = new Map(); + const images: Array = []; + const visit = (node: MarkdownNode) => { + if ( + node.type === "definition" && + node.identifier !== undefined && + !definitions.has(node.identifier) + ) { + definitions.set(node.identifier, node); + } else if (node.type === "image" || node.type === "imageReference") { + images.push(node); + } + node.children?.forEach(visit); + }; + const nodes = parser.parse(text).children; + nodes.forEach(visit); + const edits = new Map(); + for (const image of images) { + const target = image.type === "image" ? image : definitions.get(image.identifier ?? ""); + if (target?.url === undefined) continue; + // Literal percent sequences may be file names, so raw paths take precedence over decoding. + // Protect file URL delimiters so parsing does not discard literal query/hash characters. + const file = /^file:/i.test(target.url) + ? parseFileUrlHref(target.url.replace(/[%?#]/g, encodeURIComponent)) + : null; + const rawPath = stripSlashPrefixedWindowsDrive( + file === null ? target.url : safeDecodeURIComponent(file.path), + ); + const path = [rawPath, safeDecodeURIComponent(rawPath)].find((candidate) => + destinations.has(normalizeProjectPathForComparison(candidate)), + ); + if (path === undefined) continue; + missing.delete(normalizeProjectPathForComparison(path)); + const destination = angleDestination(imageDestination(path)); + const start = target.position?.start.offset; + const end = target.position?.end.offset; + if ( + (imageDestination(target.url) !== target.url || + /[?#]/.test(target.url) || + (path === rawPath && path.includes("%"))) && + start !== undefined && + end !== undefined + ) { + const label = + target.type === "image" ? (target.alt ?? "") : (target.label ?? target.identifier ?? ""); + const escapedLabel = label.replace(/[\\[\]]/g, "\\$&"); + const title = target.title == null ? "" : ` "${target.title.replace(/["\\]/g, "\\$&")}"`; + edits.set(start, { + end, + text: + target.type === "image" + ? `![${escapedLabel}](${destination}${title})` + : `[${escapedLabel}]: ${destination}${title}`, + }); + } + } + let result = text; + for (const [start, edit] of Array.from(edits).sort(([left], [right]) => right - left)) { + result = result.slice(0, start) + edit.text + result.slice(edit.end); + } + if (missing.size > 0) { + // Only a root-level fence can swallow appended blocks; nested fences end with their container. + const last = nodes.at(-1); + const start = last?.position?.start.offset; + const source = + last?.type === "code" && start !== undefined + ? text.slice(start, last.position?.end.offset) + : ""; + const fence = /^(`{3,}|~{3,})/.exec(source)?.[0]; + if (fence !== undefined) { + const lines = source.split(/\r\n|\r|\n/); + const closing = /^ {0,3}(`{3,}|~{3,})[\t ]*$/.exec(lines.at(-1) ?? "")?.[1]; + if ( + lines.length === 1 || + closing === undefined || + closing[0] !== fence[0] || + closing.length < fence.length + ) { + result += `${/[\r\n]$/.test(result) ? "" : "\n"}${fence}`; + } + } + } + for (const path of missing.values()) { + result += `${result.length > 0 ? "\n\n" : ""}![](${angleDestination(imageDestination(path))})`; + } + return result; +} From b54548372401086f71898b25a3223410da36e656 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Thu, 22 Oct 2026 05:08:47 +0000 Subject: [PATCH 2/6] fix(server): a transferred child no longer pins ingestion with pending image work Also percent-encode backslashes and line breaks left in POSIX image paths, export the Windows-path remark plugin the way neighboring plugins are, and give the image lifecycle helpers spans or a tracing rationale. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 4 +- .../Adapters/CodexAdapterV2.ts | 92 +++++++++---------- .../RunExecutionService.test.ts | 58 ++++++++++++ .../orchestration-v2/RunExecutionService.ts | 5 + packages/shared/src/generatedImageMarkdown.ts | 4 +- 5 files changed, 115 insertions(+), 48 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 4ab826a241693..632d78a8d8bd0 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -3315,9 +3315,11 @@ describe("CodexAdapterV2 post-settle continuation", () => { { savedPath: "/tmp/new#image?100%25.png" }, { savedPath: "/tmp/a©.png" }, { savedPath: "C:\\Users\\Me#100%25\\new.png" }, + { savedPath: "/tmp/a\\!b.png" }, + { savedPath: "/tmp/line\nbreak.png" }, ], expected: - "![raw]() ![encoded](/tmp/a%23b.png) ![percent]() ![percent encoded](/tmp/100%2525.png) ![query][generated]\n\n[generated]: \n\n![unc]() ![unc encoded](file://HOST/SHARE/a%23b%2525.png)\n\n![]()\n\n![]()\n\n![]()", + "![raw]() ![encoded](/tmp/a%23b.png) ![percent]() ![percent encoded](/tmp/100%2525.png) ![query][generated]\n\n[generated]: \n\n![unc]() ![unc encoded](file://HOST/SHARE/a%23b%2525.png)\n\n![]()\n\n![]()\n\n![]()\n\n![]()\n\n![]()", }, { name: "an image-only UNC result produces a filesystem media message", diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 6e933d74659e3..59c26bd40b956 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1886,22 +1886,23 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); - const emitImageWorkState = Effect.fnUntraced(function* ( - context: ActiveCodexTurnContext, - pending: boolean, - ) { - const state = imageTurns.get(context.nativeTurnId); - if (state === undefined || state.pending === pending) return; - state.pending = pending; - yield* emitProviderEvent({ - type: "background_work.updated", - driver: CODEX_PROVIDER, - providerTurnId: context.providerTurnId, - pending, + const emitImageWorkState = + /* Provider events include streaming deltas; avoid a span for each update. */ + Effect.fnUntraced(function* (context: ActiveCodexTurnContext, pending: boolean) { + const state = imageTurns.get(context.nativeTurnId); + if (state === undefined || state.pending === pending) return; + state.pending = pending; + yield* emitProviderEvent({ + type: "background_work.updated", + driver: CODEX_PROVIDER, + providerTurnId: context.providerTurnId, + pending, + }); }); - }); - const clearImageWork = Effect.fnUntraced(function* (context: ActiveCodexTurnContext) { + const clearImageWork = Effect.fn("CodexAdapterV2.clearImageWork")(function* ( + context: ActiveCodexTurnContext, + ) { yield* emitImageWorkState(context, false); imageTurns.delete(context.nativeTurnId); }); @@ -5917,39 +5918,38 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, ); - const projectGeneratedImages = Effect.fnUntraced(function* ( - context: ActiveCodexTurnContext, - nativeTurnId: string, - ) { - const state = imageTurns.get(nativeTurnId); - const firstImageItemId = state?.saved.keys().next().value; - if (state === undefined || firstImageItemId === undefined) return; - const finalAnswer = state.answer; - const text = appendGeneratedImageEmbeds(finalAnswer?.text ?? "", state.saved.values()); - if (text === (finalAnswer?.text ?? "")) { + const projectGeneratedImages = Effect.fn("CodexAdapterV2.projectGeneratedImages")( + function* (context: ActiveCodexTurnContext, nativeTurnId: string) { + const state = imageTurns.get(nativeTurnId); + const firstImageItemId = state?.saved.keys().next().value; + if (state === undefined || firstImageItemId === undefined) return; + const finalAnswer = state.answer; + const text = appendGeneratedImageEmbeds(finalAnswer?.text ?? "", state.saved.values()); + if (text === (finalAnswer?.text ?? "")) { + state.saved.clear(); + return; + } + const message = { id: finalAnswer?.id ?? firstImageItemId, text }; + state.answer = message; + const artifacts = yield* buildAgentMessageArtifacts(context, message, true); + yield* emitProviderEvent({ + type: "node.updated", + driver: CODEX_PROVIDER, + node: artifacts.node, + }); + yield* emitProviderEvent({ + type: "message.updated", + driver: CODEX_PROVIDER, + message: artifacts.message, + }); + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CODEX_PROVIDER, + turnItem: artifacts.turnItem, + }); state.saved.clear(); - return; - } - const message = { id: finalAnswer?.id ?? firstImageItemId, text }; - state.answer = message; - const artifacts = yield* buildAgentMessageArtifacts(context, message, true); - yield* emitProviderEvent({ - type: "node.updated", - driver: CODEX_PROVIDER, - node: artifacts.node, - }); - yield* emitProviderEvent({ - type: "message.updated", - driver: CODEX_PROVIDER, - message: artifacts.message, - }); - yield* emitProviderEvent({ - type: "turn_item.updated", - driver: CODEX_PROVIDER, - turnItem: artifacts.turnItem, - }); - state.saved.clear(); - }); + }, + ); const finalizeCodexTurn = (input: { readonly context: ActiveCodexTurnContext; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index b96a24fa1e47d..99f89e902c054 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -2064,6 +2064,64 @@ it.effect("keeps ingesting a failed root's background item when its work continu }), ); +it.effect("releases a settled run when a transferred child still has pending background work", () => + Effect.gen(function* () { + const consumed = yield* Deferred.make(); + const result = yield* Effect.raceFirst( + runBackgroundItemScenario( + "child-transfer-pending-work", + (ids) => { + const childProviderThreadId = ProviderThreadId.make(`${ids.providerThreadId}:child`); + const childProviderTurnId = ProviderTurnId.make(`${ids.rootProviderTurnId}:child`); + const running = subagentEvent(ids, "running"); + if (running.type !== "subagent.updated") throw new Error("Missing subagent"); + const linked = { + ...running, + subagent: { + ...running.subagent, + origin: "provider_native" as const, + childThreadId: ids.childThreadId, + providerThreadId: childProviderThreadId, + }, + }; + return [ + childThreadCreatedEvent(ids), + linked, + { + type: "provider_turn.updated", + driver, + threadId: ids.childThreadId, + providerTurn: { + id: childProviderTurnId, + providerThreadId: childProviderThreadId, + runAttemptId: null, + status: "completed", + }, + } as ProviderAdapter.ProviderAdapterV2Event, + { + type: "background_work.updated", + driver, + providerTurnId: childProviderTurnId, + pending: true, + }, + rootTerminalEvent(ids, "completed"), + { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, + ]; + }, + { + keepEventStreamOpen: true, + onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), + }, + ).pipe(Effect.asSome), + Deferred.await(consumed).pipe(Effect.as(Option.none())), + ); + assert.isTrue( + Option.isSome(result), + "the transferred child's pending work must not pin ingestion", + ); + }), +); + it.effect("does not pin ingestion on background items when the root turn is interrupted", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-interrupted", (ids) => [ diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index e6587c308fb0a..e9713cfd2a654 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1356,6 +1356,11 @@ export const layer: Layer.Layer< (current) => new Set([...current].filter((id) => routing.ownedProviderTurns.has(id))), ); + for (const id of pendingBackgroundProviderTurns) { + if (!routing.ownedProviderTurns.has(id)) { + pendingBackgroundProviderTurns.delete(id); + } + } yield* Ref.update( activeBackgroundTurnItems, (current) => new Set([...current].filter((id) => !detached.itemIds.has(id))), diff --git a/packages/shared/src/generatedImageMarkdown.ts b/packages/shared/src/generatedImageMarkdown.ts index d10d2c74f686a..57a72796611a3 100644 --- a/packages/shared/src/generatedImageMarkdown.ts +++ b/packages/shared/src/generatedImageMarkdown.ts @@ -38,8 +38,10 @@ function imageDestination(path: string): string { : path; } +// Backslashes left after Windows conversion belong to POSIX names, where Markdown would read them +// as escapes; line breaks end an angle destination. function angleDestination(path: string): string { - return `<${path.replace(/[%?#&]/g, encodeURIComponent).replace(/[<>]/g, "\\$&")}>`; + return `<${path.replace(/[%?#&\\\r\n]/g, encodeURIComponent).replace(/[<>]/g, "\\$&")}>`; } export function appendGeneratedImageEmbeds(text: string, savedPaths: Iterable): string { From 15749cc4afed9a6c63559b04341752cde89a6cf4 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Fri, 30 Oct 2026 11:56:41 +0000 Subject: [PATCH 3/6] fix(shared): generated image embeds no longer rewrite the reply Detect embeds the reply already shows with the chat renderer's own image reader and only append missing ones, so escaped reference labels keep resolving. Clients now handle authored Windows and UNC destinations. Co-Authored-By: Claude Opus 5.5 --- .../shared/src/generatedImageMarkdown.test.ts | 66 ++++++++ packages/shared/src/generatedImageMarkdown.ts | 158 +++++------------- 2 files changed, 106 insertions(+), 118 deletions(-) create mode 100644 packages/shared/src/generatedImageMarkdown.test.ts diff --git a/packages/shared/src/generatedImageMarkdown.test.ts b/packages/shared/src/generatedImageMarkdown.test.ts new file mode 100644 index 0000000000000..8958cf52bb828 --- /dev/null +++ b/packages/shared/src/generatedImageMarkdown.test.ts @@ -0,0 +1,66 @@ +import { expect, it } from "vite-plus/test"; + +import { appendGeneratedImageEmbeds } from "./generatedImageMarkdown.ts"; + +it.each([ + { + name: "appends each new image once", + text: "Done.", + saved: ["/tmp/a.png", "/tmp/b.png", "/tmp/a.png"], + expected: "Done.\n\n![]()\n\n![]()", + }, + { + name: "skips images the reply already shows", + text: String.raw`![A]() ![B](c:\Users\me\.codex\b.png) ![C][a\*b] ![D][unc] + +[a\*b]: c:\Users\me\.codex\c.png +[unc]: <\\HOST\Share\d.png> + +![E](file://host/share/e.png)`, + saved: [ + "/tmp/a b.png", + String.raw`C:\Users\Me\.codex\b.png`, + String.raw`C:\Users\Me\.codex\c.png`, + String.raw`\\host\share\d.png`, + String.raw`\\HOST\SHARE\e.png`, + "/tmp/C.png", + ], + expected: String.raw`![A]() ![B](c:\Users\me\.codex\b.png) ![C][a\*b] ![D][unc] + +[a\*b]: c:\Users\me\.codex\c.png +[unc]: <\\HOST\Share\d.png> + +![E](file://host/share/e.png) + +![]()`, + }, + { + name: "links, code and escaped embeds still need a preview", + text: "[open](/tmp/a.png) `![](/tmp/a.png)`\n\n```md\n![](/tmp/a.png)\n```\n\\![](/tmp/a.png)", + saved: ["/tmp/a.png"], + expected: + "[open](/tmp/a.png) `![](/tmp/a.png)`\n\n```md\n![](/tmp/a.png)\n```\n\\![](/tmp/a.png)\n\n![]()", + }, + { + name: "an unclosed trailing fence is closed first", + text: 'Created.\n\n~~~js\nconsole.log("done");', + saved: ["/tmp/a.png"], + expected: 'Created.\n\n~~~js\nconsole.log("done");\n~~~\n\n![]()', + }, + { + name: "Windows, UNC and reserved characters keep their path", + text: "", + saved: [ + String.raw`C:\Users\me\a b#1.png`, + String.raw`\\host\share\a.png`, + "/tmp/a#b?100%25©\\!\n.png", + ], + expected: + "![]()\n\n![]()\n\n![](.png>)", + }, +])("generated image embeds: $name", ({ text, saved, expected }) => { + const result = appendGeneratedImageEmbeds(text, saved); + expect(result).toBe(expected); + // Clients read every appended embed back as its saved file. + expect(appendGeneratedImageEmbeds(result, saved)).toBe(result); +}); diff --git a/packages/shared/src/generatedImageMarkdown.ts b/packages/shared/src/generatedImageMarkdown.ts index 57a72796611a3..e2d117f196b3f 100644 --- a/packages/shared/src/generatedImageMarkdown.ts +++ b/packages/shared/src/generatedImageMarkdown.ts @@ -1,138 +1,60 @@ import remarkParse from "remark-parse"; import { unified } from "unified"; -import { remarkKeepWindowsPaths } from "./markdownWindowsPaths.ts"; -import { parseFileUrlHref } from "./fileLinks.ts"; -import { safeDecodeURIComponent } from "./markdownLinks.ts"; -import { - isUncPath, - isWindowsDrivePath, - normalizeProjectPathForComparison, - stripSlashPrefixedWindowsDrive, -} from "./path.ts"; +import { markdownDestinationHostPath } from "./markdownLinks.ts"; +import { chatMarkdownImageSources } from "./markdownPipeline.ts"; +import { isUncPath, isWindowsDrivePath, normalizeProjectPathForComparison } from "./path.ts"; -const parser = unified().use(remarkParse).use(remarkKeepWindowsPaths); -interface MarkdownNode { - readonly type: string; - readonly url?: string | undefined; - readonly identifier?: string | undefined; - readonly label?: string | null | undefined; - readonly alt?: string | null | undefined; - readonly title?: string | null | undefined; - readonly position?: - | { - readonly start: { readonly offset?: number | undefined }; - readonly end: { readonly offset?: number | undefined }; - } - | undefined; - readonly children?: ReadonlyArray | undefined; -} +const parser = unified().use(remarkParse); -// Mobile Markdown parsers consume Windows backslashes; UNC paths also need file: to -// distinguish them from protocol-relative web URLs. -function imageDestination(path: string): string { - return isWindowsDrivePath(path) +function imageEmbed(path: string): string { + // Forward slashes read the same in every client parser; UNC paths also need file: to + // distinguish them from protocol-relative web URLs. + const destination = isWindowsDrivePath(path) ? path.replaceAll("\\", "/") : isUncPath(path) ? `file:${path.replaceAll("\\", "/")}` : path; + // Clients split `?`/`#`, decode `%` and `&` entities, and read POSIX backslashes as escapes; + // line breaks end an angle destination. + const escaped = destination.replace(/[%?#&\\\r\n]/g, encodeURIComponent).replace(/[<>]/g, "\\$&"); + return `![](<${escaped}>)`; } -// Backslashes left after Windows conversion belong to POSIX names, where Markdown would read them -// as escapes; line breaks end an angle destination. -function angleDestination(path: string): string { - return `<${path.replace(/[%?#&\\\r\n]/g, encodeURIComponent).replace(/[<>]/g, "\\$&")}>`; +// Only a root-level fence can swallow appended blocks; nested fences end with their container. +function closingFence(text: string): string { + const last = parser.parse(text).children.at(-1); + if (last?.type !== "code") return ""; + const source = text.slice(last.position?.start.offset, last.position?.end.offset); + const fence = /^(`{3,}|~{3,})/.exec(source)?.[0]; + if (fence === undefined) return ""; + const lines = source.split(/\r\n|\r|\n/); + const closing = /^ {0,3}(`{3,}|~{3,})[\t ]*$/.exec(lines.at(-1) ?? "")?.[1]; + const closed = + lines.length > 1 && + closing !== undefined && + closing[0] === fence[0] && + closing.length >= fence.length; + return closed ? "" : `${/[\r\n]$/.test(text) ? "" : "\n"}${fence}`; } +/** Appends an embed for each saved image the reply does not already show. */ export function appendGeneratedImageEmbeds(text: string, savedPaths: Iterable): string { - const destinations = new Map( - Array.from(savedPaths, (path) => [normalizeProjectPathForComparison(path), path]), + const shown = new Set( + chatMarkdownImageSources(text).flatMap((source) => { + const path = markdownDestinationHostPath(source); + return path === null ? [] : [normalizeProjectPathForComparison(path)]; + }), ); - const missing = new Map(destinations); - const definitions = new Map(); - const images: Array = []; - const visit = (node: MarkdownNode) => { - if ( - node.type === "definition" && - node.identifier !== undefined && - !definitions.has(node.identifier) - ) { - definitions.set(node.identifier, node); - } else if (node.type === "image" || node.type === "imageReference") { - images.push(node); - } - node.children?.forEach(visit); - }; - const nodes = parser.parse(text).children; - nodes.forEach(visit); - const edits = new Map(); - for (const image of images) { - const target = image.type === "image" ? image : definitions.get(image.identifier ?? ""); - if (target?.url === undefined) continue; - // Literal percent sequences may be file names, so raw paths take precedence over decoding. - // Protect file URL delimiters so parsing does not discard literal query/hash characters. - const file = /^file:/i.test(target.url) - ? parseFileUrlHref(target.url.replace(/[%?#]/g, encodeURIComponent)) - : null; - const rawPath = stripSlashPrefixedWindowsDrive( - file === null ? target.url : safeDecodeURIComponent(file.path), - ); - const path = [rawPath, safeDecodeURIComponent(rawPath)].find((candidate) => - destinations.has(normalizeProjectPathForComparison(candidate)), - ); - if (path === undefined) continue; - missing.delete(normalizeProjectPathForComparison(path)); - const destination = angleDestination(imageDestination(path)); - const start = target.position?.start.offset; - const end = target.position?.end.offset; - if ( - (imageDestination(target.url) !== target.url || - /[?#]/.test(target.url) || - (path === rawPath && path.includes("%"))) && - start !== undefined && - end !== undefined - ) { - const label = - target.type === "image" ? (target.alt ?? "") : (target.label ?? target.identifier ?? ""); - const escapedLabel = label.replace(/[\\[\]]/g, "\\$&"); - const title = target.title == null ? "" : ` "${target.title.replace(/["\\]/g, "\\$&")}"`; - edits.set(start, { - end, - text: - target.type === "image" - ? `![${escapedLabel}](${destination}${title})` - : `[${escapedLabel}]: ${destination}${title}`, - }); - } - } - let result = text; - for (const [start, edit] of Array.from(edits).sort(([left], [right]) => right - left)) { - result = result.slice(0, start) + edit.text + result.slice(edit.end); - } - if (missing.size > 0) { - // Only a root-level fence can swallow appended blocks; nested fences end with their container. - const last = nodes.at(-1); - const start = last?.position?.start.offset; - const source = - last?.type === "code" && start !== undefined - ? text.slice(start, last.position?.end.offset) - : ""; - const fence = /^(`{3,}|~{3,})/.exec(source)?.[0]; - if (fence !== undefined) { - const lines = source.split(/\r\n|\r|\n/); - const closing = /^ {0,3}(`{3,}|~{3,})[\t ]*$/.exec(lines.at(-1) ?? "")?.[1]; - if ( - lines.length === 1 || - closing === undefined || - closing[0] !== fence[0] || - closing.length < fence.length - ) { - result += `${/[\r\n]$/.test(result) ? "" : "\n"}${fence}`; - } - } + const missing = new Map(); + for (const path of savedPaths) { + const key = normalizeProjectPathForComparison(path); + if (!shown.has(key) && !missing.has(key)) missing.set(key, path); } + if (missing.size === 0) return text; + let result = text + closingFence(text); for (const path of missing.values()) { - result += `${result.length > 0 ? "\n\n" : ""}![](${angleDestination(imageDestination(path))})`; + result += `${result.length > 0 ? "\n\n" : ""}${imageEmbed(path)}`; } return result; } From b9b4bc025e7d22b902398a45e327137ee0744aff Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Fri, 30 Oct 2026 11:59:35 +0000 Subject: [PATCH 4/6] test(server): trim generated image regressions to the cases that fail on main Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 211 +++------------ .../RunExecutionService.test.ts | 247 ++---------------- 2 files changed, 71 insertions(+), 387 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 632d78a8d8bd0..00923ce29b426 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1,5 +1,3 @@ -import * as NodeModule from "node:module"; - import { historyResponseItems } from "@t3tools/provider-core/server/handoffBudget"; import { makeProviderTextDeltaCoalescer, @@ -31,7 +29,7 @@ import { ThreadId, TurnItemId, } from "@t3tools/contracts"; -import { assert, describe, expect, it } from "@effect/vitest"; +import { assert, describe, it } from "@effect/vitest"; import * as CodexClient from "effect-codex-app-server/client"; import * as CodexReplay from "effect-codex-app-server/replay"; import * as CodexError from "effect-codex-app-server/errors"; @@ -3247,101 +3245,27 @@ describe("CodexAdapterV2 post-settle continuation", () => { it.effect.each([ { - name: "multiple saved images appear once in a text-only reply", + name: "a text reply gains each saved image once", reply: "Done.", images: [ { savedPath: "/tmp/a.png" }, - { savedPath: "/tmp/b.png" }, { savedPath: "/tmp/a.png" }, - ], - expected: "Done.\n\n![]()\n\n![]()", - }, - { - name: "a link or quoted Markdown still needs a preview", - reply: - "Saved /tmp/a.png: [open](/tmp/a.png)\n`![](/tmp/a.png)`\n\n ![](/tmp/a.png)\n\n```md\n![](/tmp/a.png)\n```\n\\![](/tmp/a.png)", - images: [{ savedPath: "/tmp/a.png" }], - expected: - "Saved /tmp/a.png: [open](/tmp/a.png)\n`![](/tmp/a.png)`\n\n ![](/tmp/a.png)\n\n```md\n![](/tmp/a.png)\n```\n\\![](/tmp/a.png)\n\n![]()", - }, - { - name: "an unclosed trailing fence leaves the generated image visible", - reply: 'Created the image.\n\n~~~js\nconsole.log("done");', - images: [{ savedPath: "/tmp/generated.png" }], - expected: - 'Created the image.\n\n~~~js\nconsole.log("done");\n~~~\n\n![]()', - expectedImage: "/tmp/generated.png", - }, - { - name: "existing images are not duplicated", - reply: "![A]() ![B](/tmp/b.png) ![C][generated]\n\n[generated]: /tmp/c.png", - images: [ - { savedPath: "/tmp/a b.png" }, - { savedPath: "/tmp/b.png" }, - { savedPath: "/tmp/c.png" }, - { savedPath: "/tmp/B.png" }, - ], - expected: - "![A]() ![B](/tmp/b.png) ![C][generated]\n\n[generated]: /tmp/c.png\n\n![]()", - }, - { - name: "Windows drive paths survive mobile Markdown parsing", - reply: "Done.", - images: [{ savedPath: "C:\\Users\\me\\.codex\\a b.png" }], - expected: "Done.\n\n![]()", - }, - { - name: "existing Windows embeds are normalized without adding duplicates", - reply: - "![A](c:\\Users\\me\\.codex\\a.png) ![B](c:/users/me/.codex/b.png) ![C][unc]\n\n[unc]: <\\\\HOST\\Share\\.codex\\c.png>\n\n![D](file://host/share/.codex/d.png)", - images: [ - { savedPath: "C:\\Users\\Me\\.codex\\a.png" }, - { savedPath: "C:\\Users\\Me\\.codex\\b.png" }, - { savedPath: "\\\\host\\share\\.codex\\c.png" }, - { savedPath: "\\\\HOST\\Share\\.codex\\d.png" }, - ], - expected: - "![A]() ![B](c:/users/me/.codex/b.png) ![C][unc]\n\n[unc]: \n\n![D](file://host/share/.codex/d.png)", - }, - { - name: "reserved path characters are encoded once in new and existing embeds", - reply: - "![raw]() ![encoded](/tmp/a%23b.png) ![percent]() ![percent encoded](/tmp/100%2525.png) ![query][generated]\n\n[generated]: /tmp/q?x.png\n\n![unc](file://host/share/a#b%25.png) ![unc encoded](file://HOST/SHARE/a%23b%2525.png)", - images: [ - { savedPath: "/tmp/a#b.png" }, - { savedPath: "/tmp/100%25.png" }, - { savedPath: "/tmp/q?x.png" }, - { savedPath: "\\\\host\\share\\a#b%25.png" }, - { savedPath: "/tmp/new#image?100%25.png" }, - { savedPath: "/tmp/a©.png" }, - { savedPath: "C:\\Users\\Me#100%25\\new.png" }, - { savedPath: "/tmp/a\\!b.png" }, - { savedPath: "/tmp/line\nbreak.png" }, - ], - expected: - "![raw]() ![encoded](/tmp/a%23b.png) ![percent]() ![percent encoded](/tmp/100%2525.png) ![query][generated]\n\n[generated]: \n\n![unc]() ![unc encoded](file://HOST/SHARE/a%23b%2525.png)\n\n![]()\n\n![]()\n\n![]()\n\n![]()\n\n![]()", - }, - { - name: "an image-only UNC result produces a filesystem media message", - reply: null, - images: [{ savedPath: "\\\\host\\share\\.codex\\a.png" }], - expected: "![]()", - }, - { - name: "failed and unsaved results never become media", - reply: "No image.", - images: [ { savedPath: "/tmp/failed.png", status: "failed" }, { savedPath: "/tmp/limited.png", failure: { type: "usageLimitExceeded", limitId: "image-generation" }, }, { savedPath: null }, - { savedPath: "" }, ], - expected: "No image.", + expected: "Done.\n\n![]()", + }, + { + name: "an image-only result becomes the reply", + reply: null, + images: [{ savedPath: "/tmp/a.png" }], + expected: "![]()", }, - ])("projects generated images: $name", ({ reply, images, expected, expectedImage }) => + ])("projects generated images: $name", ({ reply, images, expected }) => Effect.gen(function* () { const scenario = "codex-generated-images"; const result = "aW1hZ2UtYnl0ZXMtbXVzdC1uZXZlci1yZWFjaC1jbGllbnRz"; @@ -3387,26 +3311,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { yield* harness.firstTerminal; const messages = assistantMessages(harness.events); - const final = messages.at(-1)?.message; - if (expectedImage !== undefined) { - // Resolve the shared package's real parser without adding server runtime dependencies. - const requireMarkdown = NodeModule.createRequire( - import.meta.resolve("@t3tools/shared/generatedImageMarkdown"), - ); - const parser: { parse(text: string): unknown } = requireMarkdown("unified") - .unified() - .use(requireMarkdown("remark-parse").default); - expect(parser.parse(final?.text ?? "")).toMatchObject({ - children: [ - { type: "paragraph" }, - { type: "code", value: 'console.log("done");' }, - { type: "paragraph", children: [{ type: "image", url: expectedImage }] }, - ], - }); - } - assert.strictEqual(final?.text, expected); - assert.strictEqual(final?.streaming, false); + assert.strictEqual(messages.at(-1)?.message.text, expected); + assert.strictEqual(messages.at(-1)?.message.streaming, false); assert.strictEqual(new Set(messages.map(({ message }) => message.id)).size, 1); + // Image bytes stay with Codex; clients load the saved file. assert.notInclude(encodeUnknownJson(harness.events), result); }).pipe( Effect.scoped, @@ -3423,13 +3331,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { const scenario = "codex-retained-image"; const base = finalAnswerTranscript(scenario, [{ id: "answer", text: "Done." }]); const completion = yield* Deferred.make(); - const secondCompletion = yield* Deferred.make(); - const projected = yield* Deferred.make(); const drained = yield* Deferred.make(); - const expected = "Done.\n\n![]()"; - const imageEvent = (phase: "started" | "completed", second = false) => ({ + const imageEvent = (phase: "started" | "completed") => ({ type: "emit_inbound" as const, - label: `image-${phase}${second ? "-second" : ""}`, + label: `image-${phase}`, frame: { method: `item/${phase}`, params: { @@ -3437,12 +3342,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { turnId: `native-${scenario}-turn`, item: { type: "imageGeneration", - id: second ? "second-image" : "late-image", + id: "late-image", status: phase === "started" ? "inProgress" : "completed", result: "", - ...(phase === "completed" - ? { savedPath: second ? "/tmp/second.png" : "/tmp/late image.png" } - : {}), + ...(phase === "completed" ? { savedPath: "/tmp/late.png" } : {}), }, }, }, @@ -3453,26 +3356,17 @@ describe("CodexAdapterV2 post-settle continuation", () => { entries: [ ...base.entries.slice(0, -2), imageEvent("started"), - imageEvent("started", true), ...base.entries.slice(-2), imageEvent("completed"), - imageEvent("completed", true), ], }), (event) => - event.type === "message.updated" && event.message.text === expected - ? Deferred.succeed(projected, undefined) - : event.type === "background_work.updated" && !event.pending - ? Deferred.succeed(drained, undefined) - : Effect.void, + event.type === "background_work.updated" && !event.pending + ? Deferred.succeed(drained, undefined) + : Effect.void, undefined, undefined, - (entry) => - entry.label === "image-completed" - ? Deferred.await(completion) - : entry.label === "image-completed-second" - ? Deferred.await(secondCompletion) - : Effect.void, + (entry) => (entry.label === "image-completed" ? Deferred.await(completion) : Effect.void), ); yield* harness.runtime.startTurn( makeCodexTestTurnInput({ @@ -3484,33 +3378,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { }), ); yield* harness.firstTerminal; - const threadPending = harness.runtime.hasPendingBackgroundWorkForThread!( - harness.providerThread, - ); - const pendingBefore = [yield* threadPending, yield* harness.hasPendingBackgroundWork]; + const pending = Effect.all([ + harness.runtime.hasPendingBackgroundWorkForThread!(harness.providerThread), + harness.hasPendingBackgroundWork, + ]); + assert.deepEqual(yield* pending, [true, true]); assert.strictEqual(assistantMessages(harness.events).at(-1)?.message.text, "Done."); if (outcome === "complete") { yield* Deferred.succeed(completion, undefined); - yield* Deferred.await(projected); - assert.deepEqual( - [yield* threadPending, yield* harness.hasPendingBackgroundWork], - [true, true], - ); - assert.isFalse( - harness.events.some( - (event) => event.type === "background_work.updated" && !event.pending, - ), - ); - yield* Deferred.succeed(secondCompletion, undefined); - yield* Deferred.await(drained); - assert.strictEqual( - assistantMessages(harness.events).at(-1)?.message.text, - `${expected}\n\n![]()`, - ); - const messages = assistantMessages(harness.events); - assert.strictEqual(messages.at(-1)?.message.id, messages[0]?.message.id); - assert.strictEqual(messages.at(-1)?.message.streaming, false); } else { yield* harness.runtime.interruptTurn({ providerThread: harness.providerThread, @@ -3519,31 +3395,28 @@ describe("CodexAdapterV2 post-settle continuation", () => { }); } yield* Deferred.await(drained); - assert.deepEqual( - [yield* threadPending, yield* harness.hasPendingBackgroundWork], - [false, false], - ); - const started = harness.events.findIndex( - (event) => event.type === "background_work.updated" && event.pending, - ); - const terminal = harness.events.findIndex((event) => event.type === "turn.terminal"); - const finished = harness.events.findIndex( - (event) => event.type === "background_work.updated" && !event.pending, - ); + assert.deepEqual(yield* pending, [false, false]); + assert.lengthOf(harness.terminalEvents(), 1); + // Run ingestion follows these updates, so they must bracket the terminal and the image. + const indexOf = (predicate: (event: ProviderAdapter.ProviderAdapterV2Event) => boolean) => + harness.events.findIndex(predicate); + const started = indexOf((e) => e.type === "background_work.updated" && e.pending); + const terminal = indexOf((e) => e.type === "turn.terminal"); + const finished = indexOf((e) => e.type === "background_work.updated" && !e.pending); assert.isAtLeast(started, 0); assert.isAbove(terminal, started); assert.isAbove(finished, terminal); if (outcome === "complete") { - assert.isAbove( - finished, - harness.events.findLastIndex( - (event) => - event.type === "turn_item.updated" && event.turnItem.type === "assistant_message", - ), + const messages = assistantMessages(harness.events); + const projected = indexOf( + (e) => + e.type === "message.updated" && e.message.text === "Done.\n\n![]()", ); + assert.isAbove(projected, terminal); + assert.isAbove(finished, projected); + assert.strictEqual(messages.at(-1)?.message.id, messages[0]?.message.id); + assert.strictEqual(messages.at(-1)?.message.streaming, false); } - assert.lengthOf(harness.terminalEvents(), 1); - assert.deepEqual(pendingBefore, [true, true]); }).pipe( Effect.scoped, Effect.provide( diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 99f89e902c054..11f4530a8290d 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1780,158 +1780,34 @@ it.effect("ends a nested native subagent and its thread when the root run is int }), ); -it.effect( - "ingests every queued generated-image update when pending work clears during root finalization", - () => - Effect.gen(function* () { - const key = "late-generated-image-chunks"; - const ids = backgroundScenarioIds(key); - const now = yield* DateTime.now; - const pending = yield* Ref.make(true); - const ingested = yield* Ref.make>([]); - const nodeId = NodeId.make(`node:${key}:assistant`); - const messageId = MessageId.make(`message:${key}:assistant`); - const text = "Here is the image."; - const imageText = `${text}\n\n![Generated image](file:///tmp/codex-08/a%23b%2525%3F.png)`; - const nodeEvent = { - type: "node.updated", - driver, - node: { - id: nodeId, - threadId: ids.threadId, - runId: ids.runId, - parentNodeId: ids.rootNodeId, - rootNodeId: ids.rootNodeId, - kind: "assistant_message", - status: "completed", - countsForRun: false, - providerThreadId: ids.providerThreadId, - providerTurnId: ids.rootProviderTurnId, - nativeItemRef: null, - runtimeRequestId: null, - checkpointScopeId: null, - startedAt: now, - completedAt: now, - }, - } satisfies ProviderAdapter.ProviderAdapterV2Event; - const messageEvent = { - type: "message.updated", - driver, - message: { - id: messageId, - threadId: ids.threadId, - runId: ids.runId, - nodeId, - role: "assistant", - createdBy: "agent", - creationSource: "provider", - text, - attachments: [], - streaming: false, - createdAt: now, - updatedAt: now, - }, - } satisfies ProviderAdapter.ProviderAdapterV2Event; - const itemEvent = { - type: "turn_item.updated", - driver, - turnItem: { - id: ids.itemId, - threadId: ids.threadId, - runId: ids.runId, - nodeId, - providerThreadId: ids.providerThreadId, +it.effect("keeps ingesting until queued background work clears after the probe goes idle", () => + Effect.gen(function* () { + // The adapter's probe runs ahead of its event queue, so only the queued update may release. + const pending = yield* Ref.make(true); + const observed = yield* runBackgroundItemScenario( + "queued-background-work", + (ids) => { + const backgroundWork = { + type: "background_work.updated", + driver, providerTurnId: ids.rootProviderTurnId, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "completed", - title: null, - startedAt: now, - completedAt: now, - updatedAt: now, - type: "assistant_message", - messageId, - text, - streaming: false, - }, - } satisfies ProviderAdapter.ProviderAdapterV2Event; - const lateMessage = { - ...messageEvent, - message: { ...messageEvent.message, text: imageText }, - }; - const lateItem = { ...itemEvent, turnItem: { ...itemEvent.turnItem, text: imageText } }; - const bothImagesText = `${imageText}\n\n![Generated image](file:///tmp/codex-08/second.png)`; - const secondLateMessage = { - ...messageEvent, - message: { ...messageEvent.message, text: bothImagesText }, - }; - const secondLateItem = { - ...itemEvent, - turnItem: { ...itemEvent.turnItem, text: bothImagesText }, - }; - const terminal = rootTerminalEvent(ids, "completed"); - const backgroundStarted = { - type: "background_work.updated", - driver, - providerTurnId: ids.rootProviderTurnId, - pending: true, - } satisfies ProviderAdapter.ProviderAdapterV2Event; - const backgroundCompleted = { ...backgroundStarted, pending: false }; - - const observed = yield* runBackgroundItemScenario( - key, - () => - Stream.fromIterable([ - backgroundStarted, - nodeEvent, - messageEvent, - itemEvent, - terminal, - ]).pipe( - // Both completions are queued when pending clears during root finalization. - // Separate chunks expose stop checks before the first node and between completions. - Stream.concat(Stream.succeed(nodeEvent)), - Stream.concat(Stream.succeed(lateMessage)), - Stream.concat(Stream.succeed(lateItem)), - Stream.concat(Stream.succeed(nodeEvent)), - Stream.concat(Stream.succeed(secondLateMessage)), - Stream.concat(Stream.succeed(secondLateItem)), - Stream.concat(Stream.succeed(backgroundCompleted)), - ), - { - keepEventStreamOpen: true, - hasPendingBackgroundWorkForThread: (providerThread) => { - assert.equal(providerThread.id, ids.providerThreadId); - return Ref.get(pending); - }, - onFinalized: () => Ref.set(pending, false), - onIngest: (event) => Ref.update(ingested, (current) => [...current, event]), - }, - ); - - assert.deepEqual(yield* Ref.get(ingested), [ - backgroundStarted, - nodeEvent, - messageEvent, - itemEvent, - terminal, - nodeEvent, - lateMessage, - lateItem, - nodeEvent, - secondLateMessage, - secondLateItem, - backgroundCompleted, - ]); - assert.deepEqual(observed, [ - "turn_item:completed", - "root-finalized", - "turn_item:completed", - "turn_item:completed", - ]); - assert.isFalse(yield* Ref.get(pending)); - }), + pending: true, + } satisfies ProviderAdapter.ProviderAdapterV2Event; + return [ + backgroundWork, + rootTerminalEvent(ids, "completed"), + backgroundTurnItemEvent(ids, "dynamic_tool", "completed", 1), + { ...backgroundWork, pending: false }, + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Ref.get(pending), + onFinalized: () => Ref.set(pending, false), + }, + ); + assert.deepEqual(observed, ["root-finalized", "turn_item:completed"]); + }), ); it.effect("keeps ingesting a child thread's late background item completion", () => @@ -2064,64 +1940,6 @@ it.effect("keeps ingesting a failed root's background item when its work continu }), ); -it.effect("releases a settled run when a transferred child still has pending background work", () => - Effect.gen(function* () { - const consumed = yield* Deferred.make(); - const result = yield* Effect.raceFirst( - runBackgroundItemScenario( - "child-transfer-pending-work", - (ids) => { - const childProviderThreadId = ProviderThreadId.make(`${ids.providerThreadId}:child`); - const childProviderTurnId = ProviderTurnId.make(`${ids.rootProviderTurnId}:child`); - const running = subagentEvent(ids, "running"); - if (running.type !== "subagent.updated") throw new Error("Missing subagent"); - const linked = { - ...running, - subagent: { - ...running.subagent, - origin: "provider_native" as const, - childThreadId: ids.childThreadId, - providerThreadId: childProviderThreadId, - }, - }; - return [ - childThreadCreatedEvent(ids), - linked, - { - type: "provider_turn.updated", - driver, - threadId: ids.childThreadId, - providerTurn: { - id: childProviderTurnId, - providerThreadId: childProviderThreadId, - runAttemptId: null, - status: "completed", - }, - } as ProviderAdapter.ProviderAdapterV2Event, - { - type: "background_work.updated", - driver, - providerTurnId: childProviderTurnId, - pending: true, - }, - rootTerminalEvent(ids, "completed"), - { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, - ]; - }, - { - keepEventStreamOpen: true, - onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), - }, - ).pipe(Effect.asSome), - Deferred.await(consumed).pipe(Effect.as(Option.none())), - ); - assert.isTrue( - Option.isSome(result), - "the transferred child's pending work must not pin ingestion", - ); - }), -); - it.effect("does not pin ingestion on background items when the root turn is interrupted", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-interrupted", (ids) => [ @@ -3737,17 +3555,12 @@ function rootTerminalEvent( function runBackgroundItemScenario( key: string, - makeEvents: ( - ids: BackgroundScenarioIds, - ) => - | ReadonlyArray - | Stream.Stream, + makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray, options?: { readonly keepEventStreamOpen?: boolean; readonly hasPendingBackgroundWorkForThread?: ( providerThread: OrchestrationV2ProviderThread, ) => Effect.Effect; - readonly onIngest?: (event: ProviderAdapter.ProviderAdapterV2Event) => Effect.Effect; readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect< ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }> >; @@ -3797,7 +3610,6 @@ function runBackgroundItemScenario( ingestNormalized: (input) => Effect.gen(function* () { const event = input.event; - yield* options?.onIngest?.(event) ?? Effect.void; if (event.type === "turn_item.updated") { yield* Ref.update(observed, (current) => [ ...current, @@ -3831,8 +3643,7 @@ function runBackgroundItemScenario( : { hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread }), subscribeEvents: Effect.gen(function* () { yield* options?.onSubscribe ?? Effect.void; - const source = makeEvents(ids); - const events = (Stream.isStream(source) ? source : Stream.fromIterable(source)).pipe( + const events = Stream.fromIterable(makeEvents(ids)).pipe( Stream.concat( Stream.fromEffect( Effect.andThen( From 6153fb2ed0994d6f95893c1f69450944571891ef Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Fri, 30 Oct 2026 12:38:16 +0000 Subject: [PATCH 5/6] fix(shared): generated images stay visible after an unclosed HTML block A reply ending inside an HTML comment or a raw 
,