diff --git a/extension/control-socket.mjs b/extension/control-socket.mjs index c64b122..adbc774 100644 --- a/extension/control-socket.mjs +++ b/extension/control-socket.mjs @@ -104,10 +104,15 @@ const LIFECYCLE_EVENTS = [ "goal_updated", "goal_update", "model_change", + "model_changed", "thinking_level_change", "reasoning_effort_change", "retry_start", "retry_end", + "auto_retry_start", + "auto_retry_end", + "retry_fallback_applied", + "retry_fallback_succeeded", "session_pause", "session_resume", "session_stop", diff --git a/src/app-server.mjs b/src/app-server.mjs index 0622034..0bc06cd 100644 --- a/src/app-server.mjs +++ b/src/app-server.mjs @@ -219,6 +219,38 @@ function pendingBlockTail(streamed, complete) { return ""; } +function codexErrorInfoFor(message, code = null) { + const value = `${message || ""} ${code || ""}`.toLowerCase(); + if (/context.{0,20}(window|length|token)|maximum context/.test(value)) return "contextWindowExceeded"; + if (/quota|usage.{0,12}limit|rate.{0,8}limit|\\b429\\b/.test(value)) return "usageLimitExceeded"; + if (/unauthor|forbidden|credential|api.?key|invalid_grant|\\b40[13]\\b/.test(value)) return "unauthorized"; + if (/overload|capacity|\\b503\\b/.test(value)) return "serverOverloaded"; + if (/bad request|invalid request|\\b400\\b/.test(value)) return "badRequest"; + if (/connect|network|timeout|timed out|econn/.test(value)) { + return { httpConnectionFailed: { httpStatusCode: null } }; + } + return "other"; +} + +function protocolTurnError(error) { + const message = String(error?.message || error || "OMP turn failed"); + const code = error?.code ?? null; + return { + message, + codexErrorInfo: error?.codexErrorInfo || codexErrorInfoFor(message, code), + additionalDetails: error?.additionalDetails ?? (code == null ? null : `code: ${code}`), + }; +} + +function lifecycleModelSelector(model) { + if (typeof model === "string") return model; + if (!model || typeof model !== "object") return ""; + if (model.selector) return String(model.selector); + const id = model.id || model.model; + const provider = model.provider || model.modelProvider; + return provider && id ? `${provider}/${id}` : String(id || ""); +} + function normalizeUserInputQuestions(request = {}) { const rawQuestions = Array.isArray(request.questions) && request.questions.length ? request.questions.slice(0, 3) @@ -609,6 +641,7 @@ export class CodexAppServer { this.#settleExternalAsk(threadId, frame.event?.requestId); } if (this.activeTurns.has(threadId)) return; + await this.#publishExternalRecoveryEvent(threadId, frame.event || {}); const status = frame.status ? this.#codexStatusFromControl(frame.status) : (/^(?:agent|turn)_start$/.test(eventType) ? { type: "active", activeFlags: [] } : null); @@ -616,6 +649,65 @@ export class CodexAppServer { this.#scheduleExternalControlSync(threadId, eventType); } + async #publishExternalRecoveryEvent(threadId, event) { + const type = event?.type || ""; + if (![ + "auto_retry_start", "retry_start", "retryStart", + "retry_fallback_applied", "retryFallbackApplied", + "retry_fallback_succeeded", "retryFallbackSucceeded", + "model_change", "modelChange", + "auto_retry_end", "retry_end", "retryEnd", + ].includes(type)) return; + const thread = await this.store.getThread(threadId); + const turnId = thread?.turns?.at(-1)?.id; + if (!turnId) return; + if (["auto_retry_start", "retry_start", "retryStart"].includes(type)) { + const attempt = Number(event.attempt || 0); + const maxAttempts = Number(event.maxAttempts || 0); + const delayMs = Number(event.delayMs || 0); + const details = [ + attempt > 0 ? `attempt ${attempt}${maxAttempts > 0 ? ` of ${maxAttempts}` : ""}` : null, + delayMs > 0 ? `retrying in ${delayMs}ms` : "retrying", + ].filter(Boolean).join(", "); + await this.#broadcast(threadId, "error", { + threadId, + turnId, + error: protocolTurnError({ + message: event.errorMessage || event.message || "OMP model request failed", + additionalDetails: details || null, + code: event.errorId ?? null, + }), + willRetry: true, + }); + } else if (type === "retry_fallback_applied" || type === "retryFallbackApplied") { + const from = lifecycleModelSelector(event.from) || "the requested model"; + const to = lifecycleModelSelector(event.to) || "a fallback model"; + await this.#applyProjectedModelChange(threadId, { model: to }); + await this.#broadcast(threadId, "warning", { + threadId, + message: `${from} failed, continuing with ${to}`, + }); + } else if (type === "retry_fallback_succeeded" || type === "retryFallbackSucceeded") { + const model = lifecycleModelSelector(event.model); + if (model) await this.#applyProjectedModelChange(threadId, { model }); + } else if (["model_change", "modelChange"].includes(type)) { + const model = lifecycleModelSelector(event.model); + if (model) await this.#applyProjectedModelChange(threadId, { model }); + } else if ((type === "auto_retry_end" || type === "retry_end" || type === "retryEnd") && event.success === false) { + await this.#broadcast(threadId, "error", { + threadId, + turnId, + error: protocolTurnError({ + message: event.finalError || event.errorMessage || "OMP exhausted its model retries", + additionalDetails: Number(event.attempt || 0) > 0 ? `failed after ${event.attempt} attempts` : null, + code: "omp_retry_exhausted", + }), + willRetry: false, + }); + } + } + + #hasExternalSubscribers(threadId) { return [...this.peers].some((peer) => peer.subscriptions?.has(threadId)); @@ -2213,6 +2305,8 @@ export class CodexAppServer { threadCwd: thread.cwd, goalAtStart: thread.goal || null, backendGoalUpdated: false, + currentModel: resolvedTurnModel, + reportedTurnError: null, itemStartTimes: new Map(), ephemeral: Boolean(thread.ephemeral), startedAt: Date.now(), @@ -2376,13 +2470,19 @@ export class CodexAppServer { try { const result = await session.runTurn(backendPrompt, context); status = result?.status || (controller.signal.aborted ? "interrupted" : "completed"); + if (active.reportedTurnError && status !== "interrupted") { + status = "failed"; + turnError = active.reportedTurnError; + } } catch (error) { if (controller.signal.aborted || error?.name === "AbortError" || error?.code === "ABORT_ERR") { status = "interrupted"; } else { status = "failed"; - turnError = { message: error.message || String(error), code: error.code || null }; - await this.#broadcast(threadId, "error", { threadId, turnId, error: turnError, willRetry: false }); + turnError = active.reportedTurnError || protocolTurnError(error); + if (!active.reportedTurnError) { + await this.#broadcast(threadId, "error", { threadId, turnId, error: turnError, willRetry: false }); + } } } finally { await this.#completeOpenItems(active, status === "completed" ? "completed" : status); @@ -2748,15 +2848,92 @@ export class CodexAppServer { await this.#broadcast(threadId, "thread/goal/updated", { threadId, turnId, goal }); break; } + case "modelChanged": { + await this.#applyBackendModelChange(active, event); + break; + } + case "turnError": { + const error = protocolTurnError(event); + const willRetry = Boolean(event.willRetry); + await this.#broadcast(threadId, "error", { threadId, turnId, error, willRetry }); + if (!willRetry && event.terminal !== false) active.reportedTurnError = error; + break; + } case "notice": { - const method = event.level === "error" ? "error" : event.level === "warning" ? "warning" : "notice"; - await this.#broadcast(threadId, method, { threadId, turnId, message: String(event.message || "") }); + const message = String(event.message || ""); + if (event.level === "error") { + await this.#broadcast(threadId, "error", { + threadId, + turnId, + error: protocolTurnError(event), + willRetry: Boolean(event.willRetry), + }); + } else if (event.level === "warning" && message) { + await this.#broadcast(threadId, "warning", { threadId, message }); + } else if (message) { + debug("OMP notice", threadId, message); + } break; } default: debug("ignored backend event", event); } } + async #applyBackendModelChange(active, event) { + await this.#applyProjectedModelChange(active.threadId, event, active); + } + + async #applyProjectedModelChange(threadId, event, active = null) { + const selector = String(event.model || ""); + if (!selector || (active && selector === active.currentModel)) return; + const thread = await this.store.getThread(threadId); + const normalized = normalizeThreadModel(selector, selector, thread?.modelProvider); + if (!thread || !normalized) return; + const modelSelector = normalized.selector || selector; + if (!active && modelSelector === (thread.modelSelector || thread.model)) return; + if (active) { + active.currentModel = modelSelector; + active.params.model = modelSelector; + } + await this.store.updateThread(threadId, (stored) => { + stored.model = normalized.id; + stored.modelProvider = normalized.provider; + stored.modelSelector = modelSelector; + }); + const effort = thread.reasoningEffort || active?.params?.reasoningEffort || "medium"; + await this.#broadcast(threadId, "thread/settings/updated", { + threadId, + threadSettings: { + cwd: thread.cwd, + approvalPolicy: thread.approvalPolicy || "on-request", + approvalsReviewer: "user", + sandboxPolicy: { + type: "workspaceWrite", + writableRoots: [], + networkAccess: false, + excludeTmpdirEnvVar: false, + excludeSlashTmp: false, + }, + activePermissionProfile: null, + model: normalized.id, + modelProvider: normalized.provider, + serviceTier: "default", + effort, + summary: "auto", + collaborationMode: { + mode: "default", + settings: { + model: normalized.id, + reasoning_effort: effort, + developer_instructions: null, + }, + }, + multiAgentMode: "explicitRequestOnly", + personality: null, + }, + }); + } + async #closeOpenTextAndReasoning(active) { const threadId = active.threadId; const turnId = active.turnId; diff --git a/src/backend/omp.mjs b/src/backend/omp.mjs index c0b6b73..fdc47a0 100644 --- a/src/backend/omp.mjs +++ b/src/backend/omp.mjs @@ -458,18 +458,24 @@ function assistantContentText(content, kind = "text") { .join(kind === "thinking" ? "\n" : ""); } +function ompModelSelector(model) { + if (typeof model === "string") return model; + if (!model || typeof model !== "object") return ""; + if (model.selector) return String(model.selector); + const id = model.id || model.model; + const provider = model.provider || model.modelProvider; + return provider && id ? `${provider}/${id}` : String(id || ""); +} + export async function mapOmpSessionEvent(event, context, toolOutputsSoFar = null) { const type = event?.type || event?.eventType; if (type === "message_update" || type === "messageUpdate") { const delta = event.assistantMessageEvent || event.messageEvent || event.update || event.event || {}; const deltaType = delta.type || delta.kind; if (["text_delta", "textDelta", "output_text_delta"].includes(deltaType)) { - const snapshot = delta.partial - ? assistantBlockText({ partial: delta.partial, contentIndex: delta.contentIndex }, "text") - : ""; - await context.emit(snapshot - ? { type: "assistantText", text: snapshot } - : { type: "textDelta", delta: delta.delta ?? delta.text ?? "" }); + if (!delta.partial) { + await context.emit({ type: "textDelta", delta: delta.delta ?? delta.text ?? "" }); + } } else if (["thinking_delta", "thinkingDelta", "reasoning_delta", "reasoningDelta"].includes(deltaType)) { const snapshot = delta.partial ? assistantBlockText({ partial: delta.partial, contentIndex: delta.contentIndex }, "thinking") @@ -544,10 +550,54 @@ export async function mapOmpSessionEvent(event, context, toolOutputsSoFar = null } return; } - if (["model_change", "modelChange"].includes(type)) { - await context.emit({ type: "notice", level: "info", message: `OMP model changed to ${event.model?.id || event.model || "unknown"}` }); - } else if (["retry_start", "retryStart"].includes(type)) { - await context.emit({ type: "notice", level: "warning", message: event.message || "OMP is retrying the model request" }); + if (["model_change", "modelChange", "model_changed", "modelChanged"].includes(type)) { + const model = ompModelSelector(event.model || event.to); + if (model) { + await context.emit({ + type: "modelChanged", + model, + fallback: Boolean(event.resolvedModelIsFallback), + }); + } + } else if (["auto_retry_start", "retry_start", "retryStart"].includes(type)) { + const attempt = Number(event.attempt || 0); + const maxAttempts = Number(event.maxAttempts || 0); + const delayMs = Number(event.delayMs || 0); + const details = [ + attempt > 0 ? `attempt ${attempt}${maxAttempts > 0 ? ` of ${maxAttempts}` : ""}` : null, + delayMs > 0 ? `retrying in ${delayMs}ms` : "retrying", + ].filter(Boolean).join(", "); + await context.emit({ + type: "turnError", + message: event.errorMessage || event.message || "OMP model request failed", + additionalDetails: details || null, + code: event.errorId ?? null, + willRetry: true, + terminal: false, + }); + } else if (type === "retry_fallback_applied" || type === "retryFallbackApplied") { + const from = ompModelSelector(event.from) || "the requested model"; + const to = ompModelSelector(event.to) || "a fallback model"; + await context.emit({ type: "modelChanged", model: to, fallback: true, fallbackFrom: from, role: event.role || null }); + await context.emit({ + type: "notice", + level: "warning", + message: `${from} failed, continuing with ${to}`, + }); + } else if (type === "retry_fallback_succeeded" || type === "retryFallbackSucceeded") { + const model = ompModelSelector(event.model); + if (model) await context.emit({ type: "modelChanged", model, fallback: true, role: event.role || null }); + } else if (type === "auto_retry_end" || type === "retry_end" || type === "retryEnd") { + if (event.success === false) { + await context.emit({ + type: "turnError", + message: event.finalError || event.errorMessage || "OMP exhausted its model retries", + additionalDetails: Number(event.attempt || 0) > 0 ? `failed after ${event.attempt} attempts` : null, + code: "omp_retry_exhausted", + willRetry: false, + terminal: true, + }); + } } else if (["auto_compaction_start", "compaction_start"].includes(type)) { await context.emit({ type: "compactionStart", id: event.id }); } else if (["auto_compaction_end", "compaction_end"].includes(type)) { @@ -1334,6 +1384,7 @@ class OmpRpcSession { this.preexistingSessionPaths = null; this.pendingResponses = new Map(); this.requestSequence = 0; + this.stderrTail = ""; } async #discoverSessionFile() { @@ -1408,6 +1459,7 @@ class OmpRpcSession { ); } this.processStartedAtMs = Date.now(); + this.stderrTail = ""; log(`spawning OMP RPC: ${executable} ${args.join(" ")} [cwd: ${this.options.cwd || process.cwd()}]`); const child = spawn(executable, args, { cwd: this.options.cwd || process.cwd(), @@ -1415,7 +1467,11 @@ class OmpRpcSession { stdio: ["pipe", "pipe", "pipe"], }); this.process = child; - child.stderr.on("data", (chunk) => log(`[omp rpc stderr] ${chunk.toString().trimEnd()}`)); + child.stderr.on("data", (chunk) => { + const text = chunk.toString(); + this.stderrTail = `${this.stderrTail}${text}`.slice(-8_192); + log(`[omp rpc stderr] ${text.trimEnd()}`); + }); const lines = createInterface({ input: child.stdout, crlfDelay: Infinity }); lines.on("line", (line) => { const trimmed = line.trim(); @@ -1434,7 +1490,9 @@ class OmpRpcSession { }); child.on("exit", (code, signal) => { this.process = null; - const error = new Error(`OMP RPC process exited (${code ?? signal})`); + const detail = this.stderrTail.trim(); + const error = new Error(detail || `OMP RPC process exited (${code ?? signal})`); + error.code = code == null ? String(signal || "OMP_RPC_EXIT") : `OMP_RPC_EXIT_${code}`; this.events.emit("fatal", error); this.#rejectPendingResponses(error); }); @@ -1532,7 +1590,7 @@ class OmpRpcSession { contextUsage: event.contextUsage || frame.data?.contextUsage, }); } - if (type === "agent_end" || type === "session_stop" || (frame.type === "response" && frame.command === "prompt" && frame.data?.agentInvoked === false)) { + if ((type === "agent_end" && !event.willContinue) || type === "session_stop" || (frame.type === "response" && frame.command === "prompt" && frame.data?.agentInvoked === false)) { this.events.emit("turnEnd", event); } else if (["ui_request", "rpc_ui_request", "tool_approval_requested"].includes(type)) { await this.#handleUiRequest(frame, event, context); diff --git a/test/app-server.test.mjs b/test/app-server.test.mjs index 6fb6e57..5d590d2 100644 --- a/test/app-server.test.mjs +++ b/test/app-server.test.mjs @@ -1199,6 +1199,41 @@ test("cumulative answer snapshots append only their unseen tails", () => { }); }); +test("terminal backend errors use the native error UI and fail the turn", () => { + const backendFactory = { + async listModels() { return [{ id: "mock/omp", model: "mock/omp" }]; }, + async openSession() { + return { + async runTurn(_prompt, context) { + await context.emit({ + type: "turnError", + message: "provider rejected the API key with 401 unauthorized", + additionalDetails: "credential 2 of 2", + willRetry: false, + terminal: true, + }); + throw new Error("OMP RPC process exited after reporting the provider error"); + }, + async interrupt() {}, + async dispose() {}, + }; + }, + }; + return withHarness({ backendFactory }, async (h) => { + await h.initialize(); + const { thread } = await h.request("thread/start", { cwd: h.temp }); + const { turn } = await h.request("turn/start", { threadId: thread.id, input: [{ type: "text", text: "ask" }] }); + const error = await h.waitFor((message) => message.method === "error" && message.params.turnId === turn.id); + assert.equal(error.params.willRetry, false); + assert.equal(error.params.error.codexErrorInfo, "unauthorized"); + assert.equal(error.params.error.additionalDetails, "credential 2 of 2"); + const completed = await h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === turn.id); + assert.equal(completed.params.turn.status, "failed"); + assert.deepEqual(completed.params.turn.error, error.params.error); + assert.equal(h.messages.filter((message) => message.method === "error" && message.params?.turnId === turn.id).length, 1); + }); +}); + test("thinking from several messages in one turn is neither doubled nor glued", () => { const backendFactory = { async listModels() { return [{ id: "mock/omp", model: "mock/omp" }]; }, diff --git a/test/control.test.mjs b/test/control.test.mjs index e588176..54fa34e 100644 --- a/test/control.test.mjs +++ b/test/control.test.mjs @@ -624,6 +624,39 @@ test("app-server consumes live control events without waiting for the JSONL poll const started = await startedPromise; assert.ok(Date.now() - startedAt < 750, "live event should beat the one-second fallback poll"); + const retryPromise = harness.waitFor((message) => + message.method === "error" + && message.params?.turnId === started.params.turn.id + , 750); + mock.emit("auto_retry_start", { + type: "auto_retry_start", + attempt: 1, + maxAttempts: 3, + delayMs: 250, + errorMessage: "provider quota exhausted", + }); + const retry = await retryPromise; + assert.equal(retry.params.willRetry, true); + assert.equal(retry.params.error.codexErrorInfo, "usageLimitExceeded"); + + const warningPromise = harness.waitFor((message) => + message.method === "warning" + && /continuing with anthropic\/claude-opus-5/.test(message.params?.message || "") + , 750); + const settingsPromise = harness.waitFor((message) => + message.method === "thread/settings/updated" + && message.params?.threadId === threadId + , 750); + mock.emit("retry_fallback_applied", { + type: "retry_fallback_applied", + from: "openai-codex/gpt-5.6-sol", + to: "anthropic/claude-opus-5", + role: "default", + }); + const [, settings] = await Promise.all([warningPromise, settingsPromise]); + assert.equal(settings.params.threadSettings.model, "claude-opus-5"); + assert.equal(settings.params.threadSettings.modelProvider, "anthropic"); + const completedPromise = harness.waitFor((message) => message.method === "turn/completed" && message.params?.threadId === threadId diff --git a/test/omp-adapter.test.mjs b/test/omp-adapter.test.mjs index c0c011b..08fc648 100644 --- a/test/omp-adapter.test.mjs +++ b/test/omp-adapter.test.mjs @@ -587,7 +587,26 @@ test("RPC adapter translates the documented JSONL event stream", async () => { return; } if (message.type === "prompt") { + console.log(JSON.stringify({ + type: "auto_retry_start", + attempt: 1, + maxAttempts: 3, + delayMs: 250, + errorMessage: "provider quota exhausted", + })); + console.log(JSON.stringify({ + type: "retry_fallback_applied", + from: "openai-codex/gpt-5.6-sol", + to: "anthropic/claude-opus-5", + role: "default", + })); + console.log(JSON.stringify({ type: "agent_end", willContinue: true })); console.log(JSON.stringify({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "rpc:" + message.message } })); + console.log(JSON.stringify({ + type: "retry_fallback_succeeded", + model: "anthropic/claude-opus-5", + role: "default", + })); console.log(JSON.stringify({ type: "agent_end", isTerminal: true })); } }); @@ -600,10 +619,79 @@ test("RPC adapter translates the documented JSONL event stream", async () => { const { turn } = await h.request("turn/start", { threadId: thread.id, input: [{ type: "text", text: "hello" }] }); const completed = await h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === turn.id); assert.match(completed.params.turn.items.find((item) => item.type === "agentMessage").text, /rpc:hello/); + const retryError = h.messages.find((message) => message.method === "error" && message.params?.turnId === turn.id); + assert.equal(retryError.params.willRetry, true); + assert.equal(retryError.params.error.codexErrorInfo, "usageLimitExceeded"); + assert.match(retryError.params.error.additionalDetails, /attempt 1 of 3/); + const fallbackWarning = h.messages.find((message) => message.method === "warning"); + assert.match(fallbackWarning.params.message, /continuing with anthropic\/claude-opus-5/); + const settings = h.messages.findLast((message) => message.method === "thread/settings/updated"); + assert.equal(settings.params.threadSettings.model, "claude-opus-5"); + assert.equal(settings.params.threadSettings.modelProvider, "anthropic"); } finally { await h.server.close(); } } finally { await fs.rm(temp, { recursive: true, force: true }); } }); +test("exhausted OMP retries map to a terminal turn error", async () => { + const emitted = []; + await mapOmpSessionEvent({ + type: "auto_retry_end", + success: false, + attempt: 3, + finalError: "all provider credentials are exhausted", + }, { emit: async (event) => emitted.push(event) }); + + assert.deepEqual(emitted, [{ + type: "turnError", + message: "all provider credentials are exhausted", + additionalDetails: "failed after 3 attempts", + code: "omp_retry_exhausted", + willRetry: false, + terminal: true, + }]); +}); + +test("RPC process failures surface their stderr in the native turn error", async () => { + const temp = await fs.mkdtemp(path.join(os.tmpdir(), "omp-rpc-stderr-")); + try { + const fake = path.join(temp, "failing-omp.cjs"); + await fs.writeFile(fake, `#!/usr/bin/env node + const readline = require("node:readline"); + const rl = readline.createInterface({ input: process.stdin }); + rl.on("line", (line) => { + const message = JSON.parse(line); + if (message.type === "get_state") { + console.log(JSON.stringify({ id: message.id, type: "response", command: "get_state", success: true, data: { + sessionId: "rpc-failure", sessionFile: process.cwd() + "/rpc-failure.jsonl" + } })); + return; + } + if (message.type === "prompt") { + console.error("provider returned 503: capacity exhausted"); + process.exit(7); + } + }); + `); + await fs.chmod(fake, 0o755); + const factory = new OmpRpcBackendFactory({ ompExecutable: fake, cwd: temp, sessionDiscoveryTimeoutMs: 0 }); + const h = await makeServerHarness(factory, temp); + try { + const { thread } = await h.request("thread/start", { cwd: temp }); + const { turn } = await h.request("turn/start", { threadId: thread.id, input: [{ type: "text", text: "hello" }] }); + const error = await h.waitFor((message) => message.method === "error" && message.params?.turnId === turn.id); + assert.match(error.params.error.message, /provider returned 503: capacity exhausted/); + assert.equal(error.params.error.codexErrorInfo, "serverOverloaded"); + assert.match(error.params.error.additionalDetails, /OMP_RPC_EXIT_7/); + const completed = await h.waitFor((message) => message.method === "turn/completed" && message.params.turn.id === turn.id); + assert.equal(completed.params.turn.status, "failed"); + } finally { + await h.server.close(); + } + } finally { + await fs.rm(temp, { recursive: true, force: true }); + } +}); + test("OMP event mapper preserves rich tool kinds, plans, and goals", async () => { const emitted = []; const contextCwd = path.resolve("/tmp/omp-context"); @@ -886,7 +974,7 @@ test("a final answer with no text deltas still produces the answer", async () => assert.equal(emitted.length, before); }); -test("rpc-ui cumulative partials map to complete answer snapshots", async () => { +test("rpc-ui cumulative answer partials wait for the complete block", async () => { const emitted = []; const context = { emit: async (event) => emitted.push(event) }; const partial = (text) => ({ role: "assistant", content: [{ type: "text", text }] }); @@ -909,9 +997,18 @@ test("rpc-ui cumulative partials map to complete answer snapshots", async () => partial: partial("nothing, just a typo."), }, }, context); + assert.deepEqual(emitted, []); + await mapOmpSessionEvent({ + type: "message_update", + assistantMessageEvent: { + type: "text_end", + contentIndex: 0, + content: "nothing, just a typo.", + partial: partial("nothing, just a typo."), + }, + }, context); assert.deepEqual(emitted, [ - { type: "assistantText", text: "nothing," }, { type: "assistantText", text: "nothing, just a typo." }, ]); });