diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index c20994423f3e9..9dfe62bc4d36b 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -77,6 +77,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 +1087,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 +1098,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 +3243,210 @@ describe("CodexAdapterV2 post-settle continuation", () => { }); }; + it.effect.each([ + { + name: "a text reply gains each saved image once", + reply: "Done.", + images: [ + { savedPath: "/tmp/a.png" }, + { savedPath: "/tmp/a.png" }, + { savedPath: "/tmp/failed.png", status: "failed" }, + { + savedPath: "/tmp/limited.png", + failure: { type: "usageLimitExceeded", limitId: "image-generation" }, + }, + { savedPath: null }, + ], + 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 }) => + 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); + 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, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect.each(["complete", "stop"] as const)( + "retains generating images after settlement until %s", + (outcome) => + Effect.gen(function* () { + const scenario = "codex-retained-image"; + const base = finalAnswerTranscript(scenario, [{ id: "answer", text: "Done." }]); + const completions = { + first: yield* Deferred.make(), + second: yield* Deferred.make(), + }; + const projected = yield* Deferred.make(); + const drained = yield* Deferred.make(); + const firstText = "Done.\n\n![]()"; + const imageEvent = (image: "first" | "second", phase: "started" | "completed") => ({ + type: "emit_inbound" as const, + label: `${image}-${phase}`, + frame: { + method: `item/${phase}`, + params: { + threadId: `native-${scenario}-thread`, + turnId: `native-${scenario}-turn`, + item: { + type: "imageGeneration", + id: image, + status: phase === "started" ? "inProgress" : "completed", + result: "", + ...(phase === "completed" ? { savedPath: `/tmp/${image}.png` } : {}), + }, + }, + }, + }); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario, + entries: [ + ...base.entries.slice(0, -2), + imageEvent("first", "started"), + imageEvent("second", "started"), + ...base.entries.slice(-2), + imageEvent("first", "completed"), + imageEvent("second", "completed"), + ], + }), + (event) => + event.type === "message.updated" && event.message.text === firstText + ? Deferred.succeed(projected, undefined) + : event.type === "background_work.updated" && !event.pending + ? Deferred.succeed(drained, undefined) + : Effect.void, + undefined, + undefined, + (entry) => + entry.label === "first-completed" + ? Deferred.await(completions.first) + : entry.label === "second-completed" + ? Deferred.await(completions.second) + : 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 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."); + const cleared = () => + harness.events.some((e) => e.type === "background_work.updated" && !e.pending); + + if (outcome === "complete") { + yield* Deferred.succeed(completions.first, undefined); + yield* Deferred.await(projected); + // The second image is still generating, so the first must not release the turn. + assert.deepEqual(yield* pending, [true, true]); + assert.isFalse(cleared()); + yield* Deferred.succeed(completions.second, undefined); + } else { + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: harness.terminalEvents()[0]!.providerTurnId, + requestRuntimeRestart: true, + }); + } + yield* Deferred.await(drained); + assert.deepEqual(yield* pending, [false, false]); + assert.lengthOf(harness.terminalEvents(), 1); + // Run ingestion follows these updates, so they must bracket the terminal and the images. + 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") { + const messages = assistantMessages(harness.events); + const projectedBoth = indexOf( + (e) => + e.type === "message.updated" && + e.message.text === `${firstText}\n\n![]()`, + ); + assert.isAbove(projectedBoth, terminal); + assert.isAbove(finished, projectedBoth); + assert.strictEqual(messages.at(-1)?.message.id, messages[0]?.message.id); + assert.strictEqual(messages.at(-1)?.message.streaming, false); + } + }).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..59c26bd40b956 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,27 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); + 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.fn("CodexAdapterV2.clearImageWork")(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 +2175,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 +2221,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 +3433,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 +4759,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 +5045,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 +5918,39 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }, ); + 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(); + }, + ); + 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..5ccfe080252ca 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1780,6 +1780,74 @@ it.effect("ends a nested native subagent and its thread when the root run is int }), ); +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. + // A child transferred to another run takes its pending work with it. + const pending = yield* Ref.make(true); + const consumed = yield* Deferred.make(); + const result = yield* Effect.raceFirst( + runBackgroundItemScenario( + "queued-background-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, + }, + }; + const backgroundWork = (providerTurnId: ProviderTurnId, pending: boolean) => + ({ + type: "background_work.updated", + driver, + providerTurnId, + pending, + }) satisfies ProviderAdapter.ProviderAdapterV2Event; + return [ + childThreadCreatedEvent(ids), + linked, + { + type: "provider_turn.updated", + driver, + threadId: ids.childThreadId, + providerTurn: { + id: childProviderTurnId, + providerThreadId: childProviderThreadId, + runAttemptId: null, + status: "completed", + }, + } as ProviderAdapter.ProviderAdapterV2Event, + backgroundWork(childProviderTurnId, true), + backgroundWork(ids.rootProviderTurnId, true), + rootTerminalEvent(ids, "completed"), + { ...linked, subagent: { ...linked.subagent, runId: RunId.make("run:replacement") } }, + backgroundTurnItemEvent(ids, "dynamic_tool", "completed", 1), + backgroundWork(ids.rootProviderTurnId, false), + ]; + }, + { + keepEventStreamOpen: true, + hasPendingBackgroundWorkForThread: () => Ref.get(pending), + onFinalized: () => Ref.set(pending, false), + onEventsConsumed: Deferred.succeed(consumed, undefined).pipe(Effect.asVoid), + }, + ).pipe(Effect.asSome), + Deferred.await(consumed).pipe(Effect.as(Option.none())), + ); + assert.deepEqual( + result, + Option.some(["subagent:running", "root-finalized", "turn_item:completed"]), + ); + }), +); + it.effect("keeps ingesting a child thread's late background item completion", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario("bg-child-item", (ids) => [ @@ -3528,6 +3596,9 @@ function runBackgroundItemScenario( makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray, options?: { readonly keepEventStreamOpen?: boolean; + readonly hasPendingBackgroundWorkForThread?: ( + providerThread: OrchestrationV2ProviderThread, + ) => Effect.Effect; readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect< ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }> >; @@ -3605,6 +3676,9 @@ 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( diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 3e3f37a15a936..e9713cfd2a654 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; @@ -1350,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))), @@ -1454,6 +1465,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.test.ts b/packages/shared/src/generatedImageMarkdown.test.ts new file mode 100644 index 0000000000000..4934ee27d24aa --- /dev/null +++ b/packages/shared/src/generatedImageMarkdown.test.ts @@ -0,0 +1,78 @@ +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: "an unclosed trailing HTML comment is closed first", + text: "\n\n![]()", + }, + { + name: "an unclosed trailing raw HTML block is closed first", + text: "
\nlog",
+    saved: ["/tmp/a.png"],
+    expected: "
\nlog\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
new file mode 100644
index 0000000000000..cf69dab1c3e56
--- /dev/null
+++ b/packages/shared/src/generatedImageMarkdown.ts
@@ -0,0 +1,69 @@
+import remarkParse from "remark-parse";
+import { unified } from "unified";
+
+import { markdownDestinationHostPath } from "./markdownLinks.ts";
+import { chatMarkdownImageSources } from "./markdownPipeline.ts";
+import { isUncPath, isWindowsDrivePath, normalizeProjectPathForComparison } from "./path.ts";
+
+const parser = unified().use(remarkParse);
+
+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}>)`;
+}
+
+// An unclosed fence or raw HTML block swallows the rest of the reply. Only a root-level block
+// can swallow appended embeds; nested ones end with their container.
+function unclosedBlockEnd(text: string): string | undefined {
+  const last = parser.parse(text).children.at(-1);
+  const source = text.slice(last?.position?.start.offset, last?.position?.end.offset);
+  if (last?.type === "html") {
+    const tag = /^ {0,3}<(script|pre|style|textarea)(?=[\s>]|$)/i.exec(source)?.[1];
+    if (tag !== undefined) {
+      return /<\/(?:script|pre|style|textarea)>/i.test(source) ? undefined : ``;
+    }
+    return /^ {0,3}") ? "-->" : undefined;
+  }
+  if (last?.type !== "code") return undefined;
+  const fence = /^ {0,3}(`{3,}|~{3,})/.exec(source)?.[1];
+  if (fence === undefined) return undefined;
+  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 ? undefined : fence;
+}
+
+/** Appends an embed for each saved image the reply does not already show. */
+export function appendGeneratedImageEmbeds(text: string, savedPaths: Iterable): string {
+  const shown = new Set(
+    chatMarkdownImageSources(text).flatMap((source) => {
+      const path = markdownDestinationHostPath(source);
+      return path === null ? [] : [normalizeProjectPathForComparison(path)];
+    }),
+  );
+  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;
+  const end = unclosedBlockEnd(text);
+  let result = end === undefined ? text : `${text}${/[\r\n]$/.test(text) ? "" : "\n"}${end}`;
+  for (const path of missing.values()) {
+    result += `${result.length > 0 ? "\n\n" : ""}${imageEmbed(path)}`;
+  }
+  return result;
+}