diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index 583cbe7098e15..58cdc44bc7fe2 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -888,6 +888,72 @@ describe("buildThreadFeed", () => { ]); }); + it.each(["commentary", undefined] as const)( + "keeps confirmed final answers visible across continuations ending with %s", + (trailingPhase) => { + const reply = (id: string, phase?: "final_answer" | "commentary") => ({ + ...assistantMessage(), + id: TurnItemId.make(id), + messageId: MessageId.make(id), + ...(phase === undefined ? {} : { phase }), + }); + const items: OrchestrationV2TurnItem[] = [ + userMessage(), + command(), + reply("first-answer", "final_answer"), + { + ...base("wake", "2026-06-20T00:00:04.000Z", 4), + type: "notification", + source: { kind: "system" }, + outcome: "completed", + summary: "Delegated task finished", + }, + reply("progress", "commentary"), + { + ...userMessage(), + id: TurnItemId.make("steer"), + messageId: MessageId.make("steer"), + inputIntent: "steer", + }, + reply("second-answer", "final_answer"), + reply("trailing", trailingPhase), + ]; + const present = (turnItems: typeof items, expanded = false) => + deriveThreadFeedPresentation( + buildThreadFeed(turnItems.map((item, position) => projected(item, position))), + null, + expanded ? new Set([runId]) : new Set(), + ); + const assistantIds = (entries: ThreadFeedEntry[]) => + entries.flatMap((entry) => + entry.type === "message" && entry.message.role === "assistant" ? [entry.id] : [], + ); + expect(assistantIds(present(items.slice(0, 3)))).toEqual(["first-answer"]); + expect(assistantIds(present(items))).toEqual([ + "first-answer", + "second-answer", + ...(trailingPhase === undefined ? ["trailing"] : []), + ]); + expect( + assistantIds( + present( + items.map((item) => + item.id === "first-answer" + ? { ...item, status: "interrupted", completedAt: null } + : item, + ), + ), + ), + ).not.toContain("first-answer"); + expect(assistantIds(present(items, true))).toEqual([ + "first-answer", + "progress", + "second-answer", + "trailing", + ]); + }, + ); + // regression: #21393 show plans and fold settled progress notes it("folds progress even when a response has no tool calls", () => { const feed = buildThreadFeed([ diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 91681731aab16..c3bb4f0518290 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -1152,7 +1152,13 @@ function deriveThreadFeedRunFolds( group.entries .filter( (entry) => - entry.id !== terminalAssistantId && + !( + entry.type === "message" && + (messageIsConfirmedFinalAnswer(entry.message) || + (entry.id === terminalAssistantId && + (entry.message.projectedItem?.item.type !== "assistant_message" || + entry.message.projectedItem.item.phase === undefined))) + ) && entry.type !== "html-render" && entry.type !== "mcp-app" && !(entry.type === "activity-group" && isContextCompactionActivityGroup(entry)) && @@ -1244,6 +1250,17 @@ function deriveThreadFeedRunFolds( return foldsByAnchorId; } +function messageIsConfirmedFinalAnswer(message: ThreadFeedMessage): boolean { + const item = message.projectedItem?.item; + return ( + item?.type === "assistant_message" && + item.phase === "final_answer" && + item.status === "completed" && + !item.streaming && + !message.streaming + ); +} + const supersededReasoningGroups = new WeakMap(); const trailingReasoningGroups = new WeakMap(); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 342f6ced28bae..9aba91324d970 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -6470,15 +6470,25 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - const usageStreamFrame = (event: unknown, parentToolUseId: string | null = null) => + const usageStreamFrame = ( + event: unknown, + parentToolUseId: string | null = null, + apiMessageId?: string, + ) => claudeSdkFrame({ type: "stream_event", event, parent_tool_use_id: parentToolUseId, session_id: WAKE_NATIVE_SESSION, uuid: "usage-stream-frame", + ...(apiMessageId === undefined ? {} : { api_message_id: apiMessageId }), }); - const usageSnapshotFrame = (id: string, uuid: string, usage: unknown) => + const usageSnapshotFrame = ( + id: string, + uuid: string, + usage: unknown, + stopReason: "end_turn" | "tool_use" | "max_tokens" | "stop_sequence" | null = null, + ) => claudeSdkFrame({ type: "assistant", message: { @@ -6487,7 +6497,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { type: "message", role: "assistant", content: [{ type: "text", text: "Working." }], - stop_reason: null, + stop_reason: stopReason, stop_sequence: null, usage, }, @@ -6508,6 +6518,55 @@ describe("ClaudeAdapterV2 background wake turns", () => { : [], ); + it.effect.each([ + { reason: "end_turn", phase: "final_answer" }, + { reason: "tool_use", phase: "commentary" }, + { reason: "max_tokens", phase: undefined }, + { reason: "stop_sequence", phase: undefined }, + { reason: null, phase: undefined }, + { reason: "end_turn", phase: "final_answer", resultOnly: true }, + ] as const)( + "preserves only confirmed assistant intent from snapshot stop reason $reason", + (input) => + Effect.gen(function* () { + const { reason, phase } = input; + const resultOnly = "resultOnly" in input; + const harness = yield* makeWakeHarness; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-answer-phase"), + text: "Continue.", + attachments: [], + }), + ); + if (!resultOnly) { + yield* Queue.offer( + harness.sdkMessages, + usageSnapshotFrame("msg_answer_phase", "answer-phase", zeroUsage, reason), + ); + } + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ uuid: "result-answer-phase", result: "Done." }), + ); + yield* Queue.take(harness.terminalReceipts); + const items = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem] + : [], + ); + assert.equal(items.at(-1)?.phase, phase); + assert.equal(items.at(-1)?.text, resultOnly ? "Done." : "Working."); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("reports context from final stream usage when the request starts at zero", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; @@ -6559,6 +6618,14 @@ describe("ClaudeAdapterV2 background wake turns", () => { const terminal = yield* Queue.take(harness.terminalReceipts); assert.equal(terminal.status, "completed"); assert.deepEqual(reportedContextTokens(harness.events), [38_123]); + const assistantItems = new Map( + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [[event.turnItem.nativeItemRef?.nativeId, event.turnItem.phase] as const] + : [], + ), + ); + assert.deepEqual([...assistantItems.values()], ["commentary", "commentary", "final_answer"]); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), @@ -6566,6 +6633,80 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect( + "matches delayed stop metadata to its native message across blocks and continuations", + () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-delayed-answer-phase"), + text: "Continue.", + attachments: [], + }), + ); + const frames = [ + usageStreamFrame({ type: "message_start", message: { id: "msg_answer" } }), + usageSnapshotFrame("msg_answer", "answer-block-1", zeroUsage), + usageSnapshotFrame("msg_answer", "answer-block-2", zeroUsage), + usageStreamFrame({ type: "message_start", message: { id: "msg_progress" } }), + usageSnapshotFrame("msg_progress", "progress-block", zeroUsage), + usageStreamFrame( + { type: "message_delta", delta: { stop_reason: "end_turn" } }, + null, + "msg_answer", + ), + usageStreamFrame( + { type: "message_delta", delta: { stop_reason: "end_turn" } }, + "child-tool", + "msg_progress", + ), + usageStreamFrame( + { type: "message_delta", delta: { stop_reason: "tool_use" } }, + null, + "msg_progress", + ), + usageSnapshotFrame("msg_answer", "answer-block-3", zeroUsage), + makeResultFrame({ uuid: "result-delayed-answer-phase", result: "Done." }), + ]; + for (const frame of frames) yield* Queue.offer(harness.sdkMessages, frame); + yield* Queue.take(harness.terminalReceipts); + const phases = new Map( + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [[event.turnItem.nativeItemRef?.nativeId, event.turnItem.phase] as const] + : [], + ), + ); + assert.deepEqual( + [...phases], + [ + ["answer-block-1", "final_answer"], + ["answer-block-2", "final_answer"], + ["progress-block", "commentary"], + ["answer-block-3", "final_answer"], + ], + ); + assert.equal( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "assistant_message" && + event.turnItem.nativeItemRef?.nativeId === "progress-block" && + event.turnItem.phase === "final_answer", + ), + false, + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("merges cumulative stream usage into the message_start counts", () => Effect.gen(function* () { const harness = yield* makeWakeHarnessWithOptions({ diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index d8a88b7dd2b96..1e5523287fcd2 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1858,6 +1858,14 @@ function assistantTextFromSdkMessage( }; } +function claudeAssistantPhase(stopReason: unknown) { + return stopReason === "end_turn" + ? ("final_answer" as const) + : stopReason === "tool_use" + ? ("commentary" as const) + : undefined; +} + function resultTextFromSdkMessage( message: SDKMessage, ): { readonly nativeItemId: string; readonly text: string } | null { @@ -3850,10 +3858,11 @@ function buildAssistantArtifacts(input: { readonly ordinal: number; readonly startedAt: DateTime.Utc; readonly completedAt: DateTime.Utc; + readonly phase?: "final_answer" | "commentary"; }): { readonly node: OrchestrationV2ExecutionNode; readonly message: OrchestrationV2ConversationMessage; - readonly turnItem: OrchestrationV2TurnItem; + readonly turnItem: Extract; } { const nodeId = input.idAllocator.derive.nodeFromProviderItem({ driver: CLAUDE_PROVIDER, @@ -3924,6 +3933,7 @@ function buildAssistantArtifacts(input: { messageId, text: input.text, streaming: false, + ...(input.phase === undefined ? {} : { phase: input.phase }), }, }; } @@ -3983,7 +3993,15 @@ interface ActiveClaudeTurnContext { readonly assistant: { fallbackText: string; fallbackNativeItemId: string; + fallbackPhase: "final_answer" | "commentary" | undefined; emittedNativeItemIds: Set; + messages: Map< + string, + { + phase?: "final_answer" | "commentary"; + items: Map>; + } + >; // The CLI repeats a notice that ends a prompt as the turn's result text. noticeTexts: Set; }; @@ -7434,6 +7452,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ordinal, startedAt: input.context.startedAt, completedAt: input.completedAt, + ...(input.status === "completed" && + input.context.assistant.fallbackPhase !== undefined + ? { phase: input.context.assistant.fallbackPhase } + : {}), }); yield* Effect.all( [ @@ -7642,15 +7664,47 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); }); + // SDK blocks arrive before their message's final stop reason. + const updateAssistantMessagePhase = Effect.fnUntraced(function* ( + context: ActiveClaudeTurnContext, + nativeMessageId: string, + stopReason: unknown, + ) { + const phase = claudeAssistantPhase(stopReason); + if (phase === undefined) return; + let message = context.assistant.messages.get(nativeMessageId); + if (message === undefined) { + message = { items: new Map() }; + context.assistant.messages.set(nativeMessageId, message); + } + if (message.phase === phase) return; + message.phase = phase; + for (const item of message.items.values()) { + const turnItem = { ...item, phase }; + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CLAUDE_PROVIDER, + turnItem, + }); + } + message.items.clear(); + }); + const emitAssistantTextArtifacts = Effect.fnUntraced(function* (input: { readonly context: ActiveClaudeTurnContext; readonly nativeItemId: string; + readonly nativeMessageId: string; readonly text: string; }) { if (input.context.assistant.emittedNativeItemIds.has(input.nativeItemId)) { return; } input.context.assistant.emittedNativeItemIds.add(input.nativeItemId); + let message = input.context.assistant.messages.get(input.nativeMessageId); + if (message === undefined) { + message = { items: new Map() }; + input.context.assistant.messages.set(input.nativeMessageId, message); + } const now = yield* DateTime.now; const ordinal = yield* resolveItemOrdinal(input.context, input.nativeItemId); const artifacts = buildAssistantArtifacts({ @@ -7662,7 +7716,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ordinal, startedAt: now, completedAt: now, + ...(message.phase === undefined ? {} : { phase: message.phase }), }); + if (message.phase === undefined) { + message.items.set(input.nativeItemId, artifacts.turnItem); + } yield* Effect.all( [ emitProviderEvent({ @@ -8355,6 +8413,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( usage === undefined ? null : { messageId: event.message.id, usage }; if (usage !== undefined) yield* emitContextUsage(context, usage, true); } else if (event.type === "message_delta") { + const nativeMessageId = field(message, "api_message_id"); + const messageId = + typeof nativeMessageId === "string" ? nativeMessageId : reasoning.messageId; + if (messageId !== null) { + yield* updateAssistantMessagePhase(context, messageId, event.delta.stop_reason); + } const usage: typeof event.usage | undefined = event.usage; const stream = context.streamUsage; if (stream?.usage && usage !== undefined) { @@ -8401,6 +8465,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } return; } + if (message.type === "assistant" && !message.parent_tool_use_id) { + yield* updateAssistantMessagePhase( + context, + message.message.id, + message.message.stop_reason, + ); + } if ( message.type === "assistant" && !message.parent_tool_use_id && @@ -8422,6 +8493,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* emitAssistantTextArtifacts({ context, nativeItemId: update.itemId, + nativeMessageId: update.messageId, text: update.text, }); } @@ -9256,10 +9328,15 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) { return; } - if (assistantText !== null && assistantText.text.length > 0) { + if ( + message.type === "assistant" && + assistantText !== null && + assistantText.text.length > 0 + ) { yield* emitAssistantTextArtifacts({ context, nativeItemId: assistantText.nativeItemId, + nativeMessageId: message.message.id, text: assistantText.text, }); return; @@ -9323,6 +9400,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) { context.assistant.fallbackText = resultText.text; context.assistant.fallbackNativeItemId = resultText.nativeItemId; + context.assistant.fallbackPhase = + message.type === "result" ? claudeAssistantPhase(message.stop_reason) : undefined; } if (message.type === "result") { @@ -10452,7 +10531,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( assistant: { fallbackText: "", fallbackNativeItemId: `assistant:${turnInput.runId}`, + fallbackPhase: undefined, emittedNativeItemIds: new Set(), + messages: new Map(), noticeTexts: new Set(), }, reasoning: { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 90ee44f3d5b6f..a136901b4ab61 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -3421,6 +3421,14 @@ describe("CodexAdapterV2 post-settle continuation", () => { assistantMessages(harness.events).map((event) => event.message.text), ["CODEX_RECOVERY_OK", "CODEX_RECOVERY_OK"], ); + assert.deepEqual( + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem.phase] + : [], + ), + ["commentary", "final_answer"], + ); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), @@ -3429,6 +3437,178 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each([false, true])( + "retains explicit answer phases without inferring missing phases (streamed: %s)", + (streamed) => + Effect.scoped( + Effect.gen(function* () { + const transcript = finalAnswerTranscript("codex-answer-phases", [ + { id: "commentary", text: "Checking.", phase: "commentary", streamed }, + { id: "answer-first", text: "First answer.", phase: "final_answer", streamed }, + { id: "unknown-null", text: "Unknown null.", phase: null, streamed }, + { id: "answer-second", text: "Second answer.", phase: "final_answer", streamed }, + { id: "unknown-omitted", text: "Unknown omitted.", omitPhase: true, streamed }, + ]); + const harness = yield* makeCodexReplayHarness(transcript); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-codex-answer-phases"), + text: "Reply with the requested recovery marker.", + }), + ); + yield* harness.awaitTerminal(1); + assert.deepEqual( + harness.events.flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.type === "assistant_message" && + event.turnItem.status === "completed" + ? [[event.turnItem.text, event.turnItem.phase]] + : [], + ), + [ + ["Checking.", "commentary"], + ["First answer.", "final_answer"], + ["Unknown null.", undefined], + ["Second answer.", "final_answer"], + ["Unknown omitted.", undefined], + ], + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect.each(["completed", "interrupted"] as const)( + "does not confirm unfinished final-answer text when the turn is %s", + (status) => + Effect.gen(function* () { + const scenario = `codex-unfinished-answer-${status}`; + const base = finalAnswerTranscript(scenario, [ + { id: "answer", text: "Unfinished answer", phase: "final_answer", streamed: true }, + ]); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario, + entries: [ + ...base.entries.slice(0, -2), + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: `native-${scenario}-thread`, + turn: makeCodexReplayTurn({ id: `native-${scenario}-turn`, status }), + }, + }, + }, + ], + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(`attempt-${scenario}`), + text: "Reply with the requested recovery marker.", + }), + ); + yield* harness.awaitTerminal(1); + const items = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem] + : [], + ); + assert.equal(items.at(-1)?.text, "Unfinished answer"); + assert.equal(items.at(-1)?.phase, "commentary"); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + Effect.scoped, + ), + ); + + it.effect("confirms a retained answer completed after the turn cleanup flush", () => + Effect.gen(function* () { + const scenario = "codex-late-answer-completion"; + const base = finalAnswerTranscript(scenario, [ + { id: "answer", text: "Done.", phase: "final_answer", streamed: true }, + ]); + const completion = yield* Deferred.make(); + const drained = yield* Deferred.make(); + const notification = (type: "imageGeneration" | "contextCompaction", id: string) => ({ + type: "emit_inbound" as const, + frame: { + method: type === "imageGeneration" ? "item/started" : "item/completed", + params: { + threadId: `native-${scenario}-thread`, + turnId: `native-${scenario}-turn`, + item: + type === "imageGeneration" + ? { type, id, status: "inProgress", result: "" } + : { type, id }, + }, + }, + }); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario, + entries: [ + ...base.entries.slice(0, -2), + notification("imageGeneration", "background-image"), + ...base.entries.slice(-1), + ...base.entries.slice(-2, -1), + notification("contextCompaction", "drained"), + ], + }), + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "compaction" && + event.turnItem.nativeItemRef?.nativeId === "drained" + ? Deferred.succeed(drained, undefined) + : Effect.void, + undefined, + undefined, + (entry) => + entry.label === "item/completed/answer" ? Deferred.await(completion) : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(`attempt-${scenario}`), + text: "Reply with the requested recovery marker.", + }), + ); + yield* harness.firstTerminal; + yield* Deferred.succeed(completion, undefined); + yield* Deferred.await(drained); + const items = harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem] + : [], + ); + assert.deepEqual( + items.map((item) => item.phase), + ["commentary", "final_answer"], + ); + assert.equal(items[0]?.id, items[1]?.id); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + Effect.scoped, + ), + ); + it.effect.each([ { name: "a text reply gains each saved image once", @@ -3500,6 +3680,14 @@ describe("CodexAdapterV2 post-settle continuation", () => { 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); + assert.deepEqual( + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem.phase] + : [], + ), + reply === null ? [undefined] : ["final_answer", "final_answer"], + ); // Image bytes stay with Codex; clients load the saved file. assert.notInclude(encodeUnknownJson(harness.events), result); }).pipe( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 821cf051c3235..7ca8d8bd409f5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1144,6 +1144,10 @@ interface ActiveCodexTurnContext { readonly subagent: CodexSubagentThreadContext | null; readonly startedAt: DateTime.Utc; readonly runningCompactionItemIds: Set; + readonly assistantMessagePhases: Map< + string, + "commentary" | "pending_final_answer" | "final_answer" + >; // Modern approval requests omit paths; keep the kinds without retaining patch bodies. readonly fileChangeKindsByItem: Map; readonly itemPositions: Map< @@ -2034,6 +2038,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( subagent: null, startedAt: input.startedAt, itemPositions: new Map(), + assistantMessagePhases: new Map(), fileChangeKindsByItem: new Map(), runningCompactionItemIds: new Set(), }; @@ -2640,6 +2645,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( subagent, startedAt: turn.startedAt, itemPositions: new Map(), + assistantMessagePhases: new Map(), fileChangeKindsByItem: new Map(), runningCompactionItemIds: new Set(), }; @@ -3302,6 +3308,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( createdAt: startedAt, updatedAt, }; + const phase = context.assistantMessagePhases.get(item.id); const turnItem: OrchestrationV2TurnItem = { id: turnItemId, threadId: context.projectionThreadId, @@ -3318,6 +3325,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( completedAt, updatedAt, type: "assistant_message", + phase: phase === "pending_final_answer" ? "commentary" : phase, messageId, text: item.text, streaming: !completed, @@ -3421,13 +3429,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( emit: /* Provider events include streaming deltas; avoid a span for each update. */ Effect.fnUntraced(function* (update) { - const context = yield* awaitActiveTurn(update.turnId); + const context = update.completed + ? (yield* resolveItemEventContext(update.turnId))?.context + : yield* awaitActiveTurn(update.turnId); if (context === undefined) { return; } - const finalAnswerItem = (yield* Ref.get(finalAnswerItemIdsByTurn)) - .get(update.turnId) - ?.has(update.itemId); + const finalAnswerItem = + (yield* Ref.get(finalAnswerItemIdsByTurn)).get(update.turnId)?.has(update.itemId) && + (!update.completed || + context.assistantMessagePhases.get(update.itemId) !== "pending_final_answer"); if (finalAnswerItem) { const finalAnswerItemIds = (yield* Ref.get(finalAnswerItemIdsByTurn)).get( update.turnId, @@ -4784,6 +4795,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } if (payload.item.type === "agentMessage") { + if (payload.item.phase != null) { + context.assistantMessagePhases.set( + payload.item.id, + payload.item.phase === "final_answer" ? "pending_final_answer" : "commentary", + ); + } if (payload.item.phase == null || payload.item.phase === "final_answer") { yield* Ref.update(finalAnswerItemIdsByTurn, (current) => { const updated = new Map(current); @@ -5148,6 +5165,18 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } const finalAnswer = payload.item.phase == null || payload.item.phase === "final_answer"; + // Turn cleanup also flushes text, so only item/completed confirms an answer. + if ( + payload.item.phase != null || + context.assistantMessagePhases.get(payload.item.id) === "pending_final_answer" + ) { + context.assistantMessagePhases.set( + payload.item.id, + payload.item.phase == null || payload.item.phase === "final_answer" + ? "final_answer" + : "commentary", + ); + } if (finalAnswer) { yield* Ref.update(finalAnswerItemIdsByTurn, (current) => { const updated = new Map(current); @@ -6495,6 +6524,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( subagent: null, startedAt: now, itemPositions: new Map(), + assistantMessagePhases: new Map(), fileChangeKindsByItem: new Map(), runningCompactionItemIds: new Set(), }; diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 4c7eba620b5d1..73399212aec21 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -1004,6 +1004,86 @@ describe("deriveMessagesTimelineRows", () => { ).toBeDefined(); }); + it.each(["commentary", undefined] as const)( + "keeps confirmed final answers visible across continuations ending with %s", + (trailingPhase) => { + const fixture = makeStreamingTimelineFixture(); + const history = fixture.visibleTurnItems.slice(0, 3); + const answer = history[2]!.item; + const prompt = history[0]!.item; + if (answer.type !== "assistant_message" || prompt.type !== "user_message") { + throw new Error("Expected the completed history response"); + } + const reply = (id: string, second: number, phase?: "final_answer" | "commentary") => ({ + ...answer, + id: TurnItemId.make(id), + messageId: MessageId.make(id), + ...(phase === undefined ? {} : { phase }), + updatedAt: DateTime.makeUnsafe(fixture.time(second)), + }); + const firstAnswer = { ...answer, phase: "final_answer" as const }; + const items: OrchestrationV2TurnItem[] = [ + prompt, + history[1]!.item, + firstAnswer, + { + ...firstAnswer, + id: TurnItemId.make("wake"), + type: "notification", + source: { kind: "system" }, + outcome: "completed", + summary: "Delegated task finished", + }, + reply("progress", 5, "commentary"), + { + ...prompt, + id: TurnItemId.make("steer"), + messageId: MessageId.make("steer"), + inputIntent: "steer", + }, + reply("second-answer", 7, "final_answer"), + reply("trailing", 8, trailingPhase), + ]; + const rows = (turnItems: typeof items, expanded = false) => + deriveMessagesTimelineRows({ + timelineEntries: deriveTimelineEntriesFromVisibleTurnItems({ + visibleTurnItems: turnItems.map((item, position) => ({ + position, + visibility: "local", + sourceThreadId: fixture.threadId, + sourceItemId: item.id, + item, + })), + optimisticMessages: [], + }), + expandedRunIds: expanded ? new Set([fixture.historyRunId]) : new Set(), + isWorking: false, + turnDiffSummaries: [], + supportsConversationRollback: false, + }); + const assistantIds = (presented: MessagesTimelineRow[]) => + presented.flatMap((row) => + row.kind === "message" && row.message.role === "assistant" ? [row.message.id] : [], + ); + expect(assistantIds(rows(items.slice(0, 3)))).toEqual([answer.messageId]); + expect(assistantIds(rows(items))).toEqual([ + answer.messageId, + "second-answer", + ...(trailingPhase === undefined ? ["trailing"] : []), + ]); + expect( + assistantIds( + rows( + items.map((item) => + item.id === answer.id ? { ...item, status: "interrupted", completedAt: null } : item, + ), + ), + ), + ).not.toContain(answer.messageId); + expect(assistantIds(rows(items.slice(0, 5), true))).toEqual([answer.messageId, "progress"]); + }, + ); + it.each([1, 2, 3])("folds %i completed activities after the terminal response", (count) => { const runId = RunId.make("turn-1"); const timelineEntries = [ @@ -2877,34 +2957,51 @@ describe("v2 run and attempt history", () => { .filter((row) => row.kind === "turn-fold") .flatMap((row) => row.deferredWorkItemIds ?? []), ).not.toContain("old-work-anchor"); + const deferredTailEntries: TimelineEntry[] = [ + timelineEntries[0]!, + timelineEntries[2]!, + timelineEntries[1]!, + ...timelineEntries.slice(3), + ].map((entry) => ({ + ...entry, + attempt: + entry.attempt.id === supersededAttemptId + ? { ...entry.attempt, completedAt: DateTime.makeUnsafe("2026-01-01T00:00:08Z") } + : entry.attempt, + ...(entry.id === "superseded-assistant-entry" + ? { + projectedItem: { + item: { + type: "assistant_message", + ordinal: 2, + status: "completed", + completedAt: DateTime.makeUnsafe("2026-01-01T00:00:03Z"), + }, + } as never, + } + : {}), + })); const deferredTailRows = deriveMessagesTimelineRows({ ...common, - timelineEntries: [ - timelineEntries[0]!, - timelineEntries[2]!, - timelineEntries[1]!, - ...timelineEntries.slice(3), - ].map((entry) => ({ - ...entry, - attempt: - entry.attempt.id === supersededAttemptId - ? { ...entry.attempt, completedAt: DateTime.makeUnsafe("2026-01-01T00:00:08Z") } - : entry.attempt, - ...(entry.id === "superseded-assistant-entry" + timelineEntries: deferredTailEntries, + }); + expect(deferredTailRows.some((row) => row.kind === "attempt-fold")).toBe(true); + const confirmedRows = deriveMessagesTimelineRows({ + ...common, + timelineEntries: deferredTailEntries.map((entry) => + entry.kind === "message" && entry.projectedItem?.item.type === "assistant_message" ? { + ...entry, projectedItem: { - item: { - type: "assistant_message", - ordinal: 2, - status: "completed", - completedAt: DateTime.makeUnsafe("2026-01-01T00:00:03Z"), - }, - } as never, + ...entry.projectedItem, + item: { ...entry.projectedItem.item, phase: "final_answer" }, + }, } - : {}), - })), + : entry, + ), }); - expect(deferredTailRows.some((row) => row.kind === "attempt-fold")).toBe(true); + expect(confirmedRows.some((row) => row.id === "superseded-assistant-entry")).toBe(true); + expect(confirmedRows.some((row) => row.kind === "attempt-fold")).toBe(true); }); describe("a steer that supersedes an attempt", () => { @@ -2930,6 +3027,7 @@ describe("v2 run and attempt history", () => { id: string, attempt: typeof supersededAttempt | typeof nextAttempt, completedSecond: number, + phase?: "final_answer" | "commentary", ): TimelineEntry => ({ id, kind: "message", @@ -2948,6 +3046,8 @@ describe("v2 run and attempt history", () => { item: { type: "assistant_message", status: "completed", + ...(phase === undefined ? {} : { phase }), + streaming: false, completedAt: DateTime.makeUnsafe(at(completedSecond)), }, } as never, @@ -2955,6 +3055,7 @@ describe("v2 run and attempt history", () => { const rowsFor = ( replyCompletedSecond: number, runStatus: "running" | "completed" = "completed", + phase?: "final_answer" | "commentary", ) => deriveMessagesTimelineRows({ timelineEntries: [ @@ -2981,7 +3082,7 @@ describe("v2 run and attempt history", () => { attempt: supersededAttempt, entry: { id: "old-work", createdAt: at(1), runId, label: "Read", tone: "tool" }, }, - assistantEntry("old-reply-entry", supersededAttempt, replyCompletedSecond), + assistantEntry("old-reply-entry", supersededAttempt, replyCompletedSecond, phase), { id: "steer-user-entry", kind: "message", @@ -3012,11 +3113,15 @@ describe("v2 run and attempt history", () => { }); it.each(["running", "completed"] as const)( - "keeps a reply that finished before the steer visible while the run is %s", + "preserves final replies and folds commentary before a steer while the run is %s", (runStatus) => { - const rows = rowsFor(4, runStatus); - expect(rows.some((row) => row.kind === "attempt-fold")).toBe(false); - expect(rows.map((row) => row.id)).toContain("old-reply-entry"); + for (const phase of [undefined, "final_answer", "commentary"] as const) { + const rows = rowsFor(4, runStatus, phase); + expect(rows.some((row) => row.kind === "attempt-fold")).toBe( + phase === "commentary" && runStatus === "running", + ); + expect(rows.some((row) => row.id === "old-reply-entry")).toBe(phase !== "commentary"); + } }, ); diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 2fe811db4d18f..062b90700ede5 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -682,7 +682,11 @@ export function resolveAssistantMessageCopyState({ } function deriveTerminalAssistantMessageIds(timelineEntries: ReadonlyArray) { - const lastAssistantMessageIdByResponseKey = new Map(); + const lastAssistantMessageByResponseKey = new Map< + string, + Extract + >(); + const lastFinalAnswerIdByResponseKey = new Map(); let nullTurnResponseIndex = 0; for (const timelineEntry of timelineEntries) { @@ -701,10 +705,33 @@ function deriveTerminalAssistantMessageIds(timelineEntries: ReadonlyArray { + const item = entry.projectedItem?.item; + const phase = item?.type === "assistant_message" ? item.phase : undefined; + const id = phase === undefined ? entry.message.id : lastFinalAnswerIdByResponseKey.get(key); + return id === undefined ? [] : [id]; + }), + ); +} + +function timelineEntryIsConfirmedFinalAnswer(entry: TimelineEntry): boolean { + if (entry.kind !== "message" || entry.message.role !== "assistant" || entry.message.streaming) { + return false; + } + const item = entry.projectedItem?.item; + return ( + item?.type === "assistant_message" && + item.phase === "final_answer" && + item.status === "completed" && + !item.streaming + ); } interface TurnFold { @@ -763,6 +790,7 @@ function completeSupersededReplyEntryIds( entry.kind === "message" && !entry.message.streaming && item?.type === "assistant_message" && + item.phase !== "commentary" && item.status === "completed" && item.completedAt !== null && supersededAt !== null && @@ -795,6 +823,7 @@ function deriveSupersededAttemptFolds( (entry.kind === "event" && entry.projectedItem.item.type === "handoff") || entry.kind === "html-render" || entry.kind === "mcp-app" || + timelineEntryIsConfirmedFinalAnswer(entry) || timelineEntryIsPersistentResourceCard(entry) || liveSubagentEntryIds.has(entry.id) || (entry.kind === "work" && entry.entry.itemType === "system_notice") @@ -1121,14 +1150,15 @@ function deriveTurnFolds(input: { ? group.entries.findIndex((entry) => entry.id === group.terminalEntry?.id) : group.entries.length; for (const [index, entry] of group.entries.entries()) { - if (entry.id === group.terminalEntry?.id) { + if (entry.id === group.terminalEntry?.id || timelineEntryIsConfirmedFinalAnswer(entry)) { continue; } const isCompaction = timelineEntryIsCompaction(entry); const isFoldableTrailingActivity = - entry.kind === "work" && - entry.entry.toolLifecycleStatus !== "inProgress" && - !workEntryDisplayIndicatesToolFailure(entry.entry); + entry.kind === "message" || + (entry.kind === "work" && + entry.entry.toolLifecycleStatus !== "inProgress" && + !workEntryDisplayIndicatesToolFailure(entry.entry)); if (!isCompaction && index > terminalEntryIndex && !isFoldableTrailingActivity) { continue; } diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index 3782307171e87..975681897ef0e 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -35,6 +35,44 @@ const decodeOrchestrationV2ProviderCapabilities = Schema.decodeUnknownSync( ); describe("orchestration V2 contracts", () => { + it.each(["final_answer", "commentary", undefined] as const)( + "retains assistant phase %s through storage and wire codecs", + (phase) => { + const stored = { + id: "answer", + threadId: "thread-1", + runId: "run-1", + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: "2026-04-20T00:00:00.000Z", + completedAt: "2026-04-20T00:00:00.000Z", + updatedAt: "2026-04-20T00:00:00.000Z", + type: "assistant_message", + messageId: "answer", + text: "Answer", + streaming: false, + ...(phase === undefined ? {} : { phase }), + }; + for (const codec of [ + OrchestrationV2TurnItemJson, + Schema.toCodecJson(OrchestrationV2TurnItem), + ]) { + const decode = Schema.decodeUnknownSync(codec); + const encode = Schema.encodeSync(codec); + const item = decode(stored); + const encoded = JSON.parse(JSON.stringify(encode(item))); + expect(encoded).toEqual(stored); + expect(decode(encoded)).toEqual(item); + } + }, + ); + it("decodes thread event types from a newer server as skippable items", () => { const decodeWireItems = Schema.decodeUnknownSync( Schema.toCodecJson(Schema.Array(OrchestrationV2RpcSchemas.subscribeThread.output)), diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 1c80128f8bbc5..965559b71cdc1 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1606,6 +1606,8 @@ export const OrchestrationV2TurnItem = Schema.Union([ ...OrchestrationV2TurnItemBaseFields, type: Schema.Literal("assistant_message"), messageId: MessageId, + /** Absent for older history and providers that cannot identify answer intent. */ + phase: Schema.optional(Schema.Literals(["final_answer", "commentary"])), text: Schema.String, textOmitted: DedupedBodyMarker, attachments: Schema.optional(Schema.Array(ChatAttachment)), @@ -2436,6 +2438,7 @@ export const OrchestrationV2TurnItemJson = Schema.Union([ ...OrchestrationV2TurnItemJsonBaseFields, type: Schema.Literal("assistant_message"), messageId: MessageId, + phase: Schema.optional(Schema.Literals(["final_answer", "commentary"])), text: Schema.String, attachments: Schema.optional(Schema.Array(ChatAttachment)), streaming: Schema.Boolean,