From 424bd9f6cf753b1d9b2da313d99dfb9ea90651fc Mon Sep 17 00:00:00 2001 From: Derrick Koko Date: Wed, 9 Sep 2026 15:03:47 +0100 Subject: [PATCH 1/2] fix(timeline): surface reasoning streams as thinking rows Turns from models that narrate through their reasoning channel (Gemini via Antigravity/OpenCode, Claude thinking blocks, Codex reasoning) rendered as bare tool lists: ingestion dropped every non-assistant content delta, so only the final answer ever showed. Buffer reasoning_text/reasoning_summary_text per turn and flush them as thinking activities interleaved with the tool calls they explain (before tool rows, approval pauses, bubble completion, and turn end). Split at 8k chars per activity; whitespace-only buffers are skipped. Web and mobile render thinking as standalone expandable rows outside collapsed tool groups. --- apps/mobile/src/lib/threadActivity.test.ts | 43 ++++ apps/mobile/src/lib/threadActivity.ts | 14 +- .../Layers/ProviderRuntimeIngestion.test.ts | 207 +++++++++++++++++- .../Layers/ProviderRuntimeIngestion.ts | 165 ++++++++++++++ .../chat/MessagesTimeline.logic.test.ts | 50 +++++ .../components/chat/MessagesTimeline.logic.ts | 7 + apps/web/src/session-logic.test.ts | 24 ++ apps/web/src/session-logic.ts | 12 + packages/contracts/src/orchestration.ts | 14 ++ 9 files changed, 532 insertions(+), 4 deletions(-) diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index 38ce75f24b08..f4a59c616a37 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -373,6 +373,49 @@ describe("buildThreadFeed", () => { expect(nextRows.some((row) => row.type === "activity-group")).toBe(true); }); + it("keeps thinking activities in their own feed group between tool calls", () => { + const turnId = TurnId.make("turn-1"); + const thread = makeThread({ + id: ThreadId.make("feed-thinking"), + projectId: ProjectId.make("project-1"), + title: "Feed thinking", + messages: [], + activities: [ + makeActivity({ + id: EventId.make("tool-1"), + kind: "tool.completed", + summary: "Ran command", + tone: "tool", + createdAt: "2026-04-01T00:00:01.000Z", + turnId, + payload: { itemType: "command_execution", status: "completed" }, + }), + makeActivity({ + id: EventId.make("thinking-1"), + kind: "thinking", + summary: "Thinking", + tone: "info", + createdAt: "2026-04-01T00:00:02.000Z", + turnId, + payload: { detail: "Checking the checklist first." }, + }), + makeActivity({ + id: EventId.make("tool-2"), + kind: "tool.completed", + summary: "Ran command", + tone: "tool", + createdAt: "2026-04-01T00:00:03.000Z", + turnId, + payload: { itemType: "command_execution", status: "completed" }, + }), + ], + }); + + const feed = buildThreadFeed(thread); + expect(feed.map((row) => row.id)).toEqual(["tool-1", "thinking-1", "tool-2"]); + expect(feed[1]).toMatchObject({ type: "activity-group" }); + }); + it("regroups cached activities for message changes and pagination", () => { const messages = [2, 4].map((second) => ({ id: MessageId.make(`message-${second}`), diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 0ab9986b2f40..059832792f11 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -4,7 +4,11 @@ import { requestKindFromRequestType, type PendingApproval, } from "@t3tools/client-runtime/pending-requests"; -import { UserInputAttachmentAnswerPayload, isToolLifecycleItemType } from "@t3tools/contracts"; +import { + THINKING_ACTIVITY_KIND, + UserInputAttachmentAnswerPayload, + isToolLifecycleItemType, +} from "@t3tools/contracts"; import type { OrchestrationLatestTurn, OrchestrationThread, @@ -1555,12 +1559,16 @@ function groupAdjacentActivities(entries: ReadonlyArray): Th } const isCompaction = entry.activity.workEntry.sourceActivityKind === "context-compaction"; - if (isCompaction || firstActivityEntry?.turnId !== entry.turnId) { + // Thinking rows stay visible outside collapsed tool groups (mirrors web): + // they break the run on both sides so a reasoning-narrated turn keeps its + // narrative without expanding dozens of tool calls. + const isThinking = entry.activity.workEntry.sourceActivityKind === THINKING_ACTIVITY_KIND; + if (isCompaction || isThinking || firstActivityEntry?.turnId !== entry.turnId) { flushGroup(); } firstActivityEntry ??= entry; openGroupActivities.push(entry.activity); - if (isCompaction) { + if (isCompaction || isThinking) { flushGroup(); } } diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 1094ab48b7ac..6ca57fd25949 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1322,7 +1322,9 @@ describe("ProviderRuntimeIngestion", () => { const harness = await createHarness(); const initial = await harness.readModel(); - for (const streamKind of ["reasoning_text", "command_output", "file_change_output"] as const) { + // Reasoning streams are persisted as thinking activities (tested below); + // tool-output streams stay ignored because tool rows already show them. + for (const streamKind of ["command_output", "file_change_output", "unknown"] as const) { harness.emit({ type: "content.delta", eventId: asEventId(`evt-ignored-${streamKind}`), @@ -1341,6 +1343,209 @@ describe("ProviderRuntimeIngestion", () => { expect(await harness.readModel()).toEqual(initial); }); + it("persists reasoning deltas as thinking activities ordered before tool rows", async () => { + const harness = await createHarness(); + const thoughtAt = "2026-01-01T00:00:00.000Z"; + const toolAt = "2026-01-01T00:00:01.000Z"; + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-reason-1"), + provider: ProviderDriverKind.make("antigravity"), + createdAt: thoughtAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-thinking"), + payload: { + streamKind: "reasoning_text", + delta: "Checking the ", + }, + }); + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-reason-2"), + provider: ProviderDriverKind.make("antigravity"), + createdAt: thoughtAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-thinking"), + payload: { + streamKind: "reasoning_text", + delta: "checklist first.", + }, + }); + harness.emit({ + type: "item.completed", + eventId: asEventId("evt-thinking-tool"), + provider: ProviderDriverKind.make("antigravity"), + createdAt: toolAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-thinking"), + itemId: asItemId("tool-1"), + payload: { + itemType: "command_execution", + status: "completed", + title: "Ran command", + detail: "git status", + }, + }); + + await harness.drain(); + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ); + const thinking = thread.activities.find( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ); + expect(thinking?.tone).toBe("info"); + expect(thinking?.summary).toBe("Thinking"); + expect(thinking?.turnId).toBe("turn-thinking"); + expect(thinking?.createdAt).toBe(thoughtAt); + expect((thinking?.payload as Record | undefined)?.detail).toBe( + "Checking the checklist first.", + ); + const kinds = thread.activities.map( + (activity: ProviderRuntimeTestActivity) => activity.kind, + ); + expect(kinds.indexOf("thinking")).toBeLessThan(kinds.indexOf("tool.completed")); + }); + + it("flushes buffered thinking when the turn completes", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-summary-reason"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-9"), + payload: { + streamKind: "reasoning_summary_text", + delta: "ci, websearch", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-9-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: now, + turnId: asTurnId("turn-9"), + payload: { state: "completed" }, + }); + + await harness.drain(); + const thread = await waitForThread( + harness.readModel, + (entry) => + entry.session?.status === "ready" && + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ); + const thinking = thread.activities.filter( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ); + expect(thinking).toHaveLength(1); + expect((thinking[0]?.payload as Record | undefined)?.detail).toBe( + "ci, websearch", + ); + expect(thinking[0]?.turnId).toBe("turn-9"); + }); + + it("splits long thinking buffers across activities without losing text", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + const longThought = "x".repeat(9_000); + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-long-reason"), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-long"), + payload: { + streamKind: "reasoning_text", + delta: longThought, + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-long-completed"), + provider: ProviderDriverKind.make("opencode"), + threadId: asThreadId("thread-1"), + createdAt: now, + turnId: asTurnId("turn-long"), + payload: { state: "completed" }, + }); + + await harness.drain(); + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ); + const thinking = thread.activities.filter( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ); + expect(thinking.length).toBeGreaterThan(1); + const thinkingText = (activity: ProviderRuntimeTestActivity): string => { + const detail = (activity.payload as Record | undefined)?.detail; + return typeof detail === "string" ? detail : ""; + }; + expect(thinking.map(thinkingText).join("")).toBe(longThought); + for (const activity of thinking) { + expect(thinkingText(activity).length).toBeLessThanOrEqual(8_000); + } + }); + + it("skips whitespace-only thinking buffers", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-blank-reason"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-blank"), + payload: { + streamKind: "reasoning_text", + delta: " \n ", + }, + }); + harness.emit({ + type: "item.completed", + eventId: asEventId("evt-blank-tool"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-blank"), + itemId: asItemId("tool-blank"), + payload: { + itemType: "dynamic_tool_call", + status: "completed", + title: "Tool call", + }, + }); + + await harness.drain(); + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "tool.completed", + ), + ); + expect( + thread.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ).toBe(false); + }); + it("maps canonical content delta/item completed into finalized assistant messages", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 8d34fee4f981..6a007ea8b537 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -9,6 +9,8 @@ import { classifyTaskAgentKind, EventId, isToolLifecycleItemType, + THINKING_ACTIVITY_KIND, + THINKING_ACTIVITY_SUMMARY, ThreadId, type ThreadTokenUsageSnapshot, TurnId, @@ -55,6 +57,8 @@ import { canReplaceThreadTitle } from "../threadTitles.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`; const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`; +const thinkingBufferKey = (threadId: ThreadId, turnId: TurnId | undefined) => + turnId ? providerTurnKey(threadId, turnId) : `${threadId}:no-turn`; // Fallback when the in-memory description cache no longer has the task name // (server restart, session-exit sweep, TTL/capacity eviction): earlier @@ -97,6 +101,12 @@ interface AssistantSegmentState { activeMessageId: MessageId | null; } +interface BufferedThinking { + readonly text: string; + /** When the first buffered delta arrived: the thinking activity's timestamp. */ + readonly firstCreatedAt: string; +} + const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000; const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120); const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000; @@ -106,6 +116,12 @@ const BUFFERED_PROPOSED_PLAN_BY_ID_TTL = Duration.minutes(120); const TASK_DESCRIPTION_BY_TASK_CACHE_CAPACITY = 10_000; const TASK_DESCRIPTION_BY_TASK_TTL = Duration.minutes(120); const MAX_BUFFERED_ASSISTANT_CHARS = 24_000; +// One thinking activity holds at most this much text; longer buffers split +// across several activities so a thinking-heavy turn (e.g. Gemini) cannot +// stuff an unbounded blob into a single activity payload. +const MAX_THINKING_CHARS_PER_ACTIVITY = 8_000; +const BUFFERED_THINKING_TEXT_BY_TURN_CACHE_CAPACITY = 10_000; +const BUFFERED_THINKING_TEXT_BY_TURN_TTL = Duration.minutes(120); const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0"; type TurnStartRequestedDomainEvent = Extract< @@ -942,6 +958,17 @@ const make = Effect.gen(function* () { lookup: () => Effect.succeed({ text: "", createdAt: "" }), }); + // Reasoning streams (`reasoning_text` / `reasoning_summary_text`) buffer + // here until a flush point (tool activity, approval pause, assistant + // completion, turn end) persists them as `thinking` activities. Without + // this, providers that narrate through their reasoning channel read as + // bare tool lists with only a final answer. + const bufferedThinkingTextByTurnKey = yield* Cache.make({ + capacity: BUFFERED_THINKING_TEXT_BY_TURN_CACHE_CAPACITY, + timeToLive: BUFFERED_THINKING_TEXT_BY_TURN_TTL, + lookup: () => Effect.succeed({ text: "", firstCreatedAt: "" }), + }); + // Task names arrive on task.started/task.progress but not on task.completed, // so remember them per task to title the completion activity. const taskDescriptionByTaskKey = yield* Cache.make({ @@ -1134,6 +1161,83 @@ const make = Effect.gen(function* () { const clearBufferedAssistantText = (messageId: MessageId) => Cache.invalidate(bufferedAssistantTextByMessageId, messageId); + const appendBufferedThinkingText = ( + threadId: ThreadId, + turnId: TurnId | undefined, + delta: string, + createdAt: string, + ) => + Cache.getOption(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId)).pipe( + Effect.flatMap((existing) => { + const previous = Option.getOrElse(existing, () => ({ text: "", firstCreatedAt: "" })); + return Cache.set(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId), { + text: `${previous.text}${delta}`, + firstCreatedAt: + previous.firstCreatedAt.length > 0 ? previous.firstCreatedAt : createdAt, + }); + }), + ); + + const takeBufferedThinkingText = ( + threadId: ThreadId, + turnId: TurnId | undefined, + ): Effect.Effect => + Cache.getOption(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId)).pipe( + Effect.flatMap((existing) => + Cache.invalidate(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId)).pipe( + Effect.as( + Option.getOrElse(existing, (): BufferedThinking => ({ text: "", firstCreatedAt: "" })), + ), + ), + ), + ); + + const splitThinkingText = (text: string): Array => { + const chunks: Array = []; + for (let index = 0; index < text.length; index += MAX_THINKING_CHARS_PER_ACTIVITY) { + chunks.push(text.slice(index, index + MAX_THINKING_CHARS_PER_ACTIVITY)); + } + return chunks; + }; + + // Persists buffered reasoning as `thinking` activities, oldest chunk first, + // so the rows interleave with the tool calls the thinking explains. Consumes + // the buffer; a later flush for the same turn is a no-op until new deltas + // arrive. Long buffers split across several activities instead of dropping. + // Activities carry the first delta's timestamp so they sort before the tool + // rows that flushed them. + const flushBufferedThinkingForTurn = (input: { + event: ProviderRuntimeEvent; + threadId: ThreadId; + turnId?: TurnId; + createdAt: string; + commandTag: string; + }) => + Effect.gen(function* () { + const buffered = yield* takeBufferedThinkingText(input.threadId, input.turnId); + if (!hasRenderableAssistantText(buffered.text)) { + return; + } + const chunks = splitThinkingText(buffered.text); + for (const [chunkIndex, chunk] of chunks.entries()) { + yield* orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: yield* providerCommandId(input.event, `${input.commandTag}:${chunkIndex}`), + threadId: input.threadId, + activity: { + id: EventId.make(`${input.event.eventId}:thinking:${chunkIndex}`), + createdAt: buffered.firstCreatedAt, + tone: "info", + kind: THINKING_ACTIVITY_KIND, + summary: THINKING_ACTIVITY_SUMMARY, + payload: { detail: chunk }, + turnId: input.turnId ?? null, + }, + createdAt: input.createdAt, + }); + } + }); + const appendBufferedProposedPlan = (planId: string, delta: string, createdAt: string) => Cache.getOption(bufferedProposedPlanById, planId).pipe( Effect.flatMap((existingEntry) => { @@ -1477,6 +1581,24 @@ const make = Effect.gen(function* () { const processRuntimeEvent = (event: ProviderRuntimeEvent) => Effect.gen(function* () { if (event.type === "content.delta" && event.payload.streamKind !== "assistant_text") { + // Reasoning streams buffer into `thinking` activities (flushed at the + // points below). Every other non-assistant stream stays ignored: it + // carries tool output the timeline already shows through tool rows. + if ( + (event.payload.streamKind === "reasoning_text" || + event.payload.streamKind === "reasoning_summary_text") && + event.payload.delta.length > 0 + ) { + const thread = yield* resolveThreadRuntimeContext(event.threadId); + if (thread) { + yield* appendBufferedThinkingText( + thread.id, + toTurnId(event.turnId), + event.payload.delta, + event.createdAt, + ); + } + } return; } @@ -1702,6 +1824,14 @@ const make = Effect.gen(function* () { ? toTurnId(event.turnId) : undefined; if (pauseForUserTurnId) { + // Thinking precedes the approval pause it led to. + yield* flushBufferedThinkingForTurn({ + event, + threadId: thread.id, + turnId: pauseForUserTurnId, + createdAt: now, + commandTag: "thinking-flush-on-request-opened", + }); const hasProjectedMessage = yield* projectionThreadMessages.hasAssistantMessageForTurn({ threadId: thread.id, turnId: pauseForUserTurnId, @@ -1766,6 +1896,15 @@ const make = Effect.gen(function* () { : undefined; if (assistantCompletion) { + // Thinking that shares the completed segment renders above the bubble. + const completionTurnId = toTurnId(event.turnId); + yield* flushBufferedThinkingForTurn({ + event, + threadId: thread.id, + ...(completionTurnId ? { turnId: completionTurnId } : {}), + createdAt: now, + commandTag: "thinking-flush-on-assistant-complete", + }); const turnId = toTurnId(event.turnId); const activeAssistantMessageId = turnId ? yield* getActiveAssistantMessageIdForTurn(thread.id, turnId) @@ -1835,6 +1974,14 @@ const make = Effect.gen(function* () { if (isTerminalTurn) { const turnId = toTurnId(event.turnId); + // Trailing thinking renders above the turn's final bubble. + yield* flushBufferedThinkingForTurn({ + event, + threadId: thread.id, + ...(turnId ? { turnId } : {}), + createdAt: now, + commandTag: "thinking-flush-on-turn-end", + }); if (turnId) { const userInputActivities = yield* projectionThreadActivityRepository.listUserInputLifecycleByThreadId({ @@ -1911,6 +2058,14 @@ const make = Effect.gen(function* () { } if (event.type === "session.exited") { + const exitedActiveTurnId = thread.session?.activeTurnId ?? undefined; + yield* flushBufferedThinkingForTurn({ + event, + threadId: thread.id, + ...(exitedActiveTurnId ? { turnId: exitedActiveTurnId } : {}), + createdAt: now, + commandTag: "thinking-flush-on-session-exit", + }); yield* clearTurnStateForSession(thread.id); } @@ -2126,6 +2281,16 @@ const make = Effect.gen(function* () { } } + // Thinking buffered so far renders above this event's own rows + // (tool calls, task updates, ...), preserving the order the model + // produced them in. + yield* flushBufferedThinkingForTurn({ + event, + threadId: thread.id, + ...(eventTurnId ? { turnId: eventTurnId } : {}), + createdAt: now, + commandTag: "thinking-flush-before-activities", + }); const activities = runtimeEventToActivities(activityEvent, taskTitle); yield* Effect.forEach(activities, (activity) => providerCommandId(event, "thread-activity-append").pipe( diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index e1cf7f0dc6e7..10ff146dfcd2 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -1120,6 +1120,56 @@ describe("deriveMessagesTimelineRows", () => { ]); }); + it("keeps thinking rows visible outside collapsed tool groups", () => { + const toolEntry = (id: string, createdAt: string) => ({ + id, + kind: "work" as const, + createdAt, + entry: { + id, + createdAt, + label: "Ran command", + tone: "tool" as const, + itemType: "command_execution" as const, + toolLifecycleStatus: "completed" as const, + }, + }); + const rows = deriveMessagesTimelineRows({ + timelineEntries: [ + toolEntry("tool-before", "2026-01-01T00:00:01Z"), + { + id: "thinking-1", + kind: "work" as const, + createdAt: "2026-01-01T00:00:02Z", + entry: { + id: "thinking-1", + createdAt: "2026-01-01T00:00:02Z", + label: "Thinking", + detail: "Checking the checklist first.", + tone: "info" as const, + sourceActivityKind: "thinking", + }, + }, + toolEntry("tool-after", "2026-01-01T00:00:03Z"), + ], + isWorking: false, + activeTurnStartedAt: null, + turnDiffSummaries: [], + supportsConversationRollback: false, + }); + + expect(rows.map((row) => row.id)).toEqual([ + "tool-before", + "work-toggle:thinking-1", + "tool-after", + ]); + expect(rows[1]).toMatchObject({ + kind: "work-toggle", + summary: "Thinking", + hiddenCount: 1, + }); + }); + it("only enables assistant copy for the terminal assistant message in a turn", () => { const rows = deriveMessagesTimelineRows({ timelineEntries: [ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index a87531673506..3205196c6131 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -20,6 +20,7 @@ import { formatDuration, inferCheckpointTurnCountByTurnId, isStreamingMessageTextUpdate, + isThinkingWorkEntry, workEntryDisplayIndicatesToolFailure, workEntryIndicatesToolSuccess, workEntryIndicatesToolNeutralStatus, @@ -1054,15 +1055,21 @@ export function deriveMessagesTimelineRows(input: { continue; } const groupedEntries = [timelineEntry.entry]; + // Thinking rows stay visible outside collapsed tool groups: they break + // the run on both sides so a reasoning-narrated turn keeps its narrative + // without expanding dozens of tool calls. + const anchorIsThinking = isThinkingWorkEntry(timelineEntry.entry); let cursor = index + 1; while (cursor < input.timelineEntries.length) { const nextEntry = input.timelineEntries[cursor]; if ( !nextEntry || + anchorIsThinking || nextEntry.kind !== "work" || nextEntry.entry.agentSpawn !== undefined || nextEntry.entry.sourceActivityKind === "context-compaction" || nextEntry.entry.tone === "error" || + isThinkingWorkEntry(nextEntry.entry) || activeWorkEntryIds.has(nextEntry.id) || collapsedEntryIds.has(nextEntry.id) || foldsByAnchorEntryId.has(nextEntry.id) diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index 401934bce5ba..fb106c3e2901 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -497,6 +497,30 @@ describe("deriveWorkLogEntries", () => { expect(entries.map((entry) => entry.id)).toEqual(["tool-complete"]); }); + it("derives thinking entries with their full text as detail", () => { + const entries = deriveWorkLogEntries([ + makeActivity({ + id: "thinking-1", + kind: "thinking", + summary: "Thinking", + tone: "info", + turnId: "turn-1", + payload: { detail: "Checking the checklist first." }, + sequence: 1, + }), + ]); + + expect(entries).toMatchObject([ + { + id: "thinking-1", + label: "Thinking", + detail: "Checking the checklist first.", + tone: "info", + sourceActivityKind: "thinking", + }, + ]); + }); + it("omits routine setup updates before work starts and after later turn activity", () => { const setupActivities = [ makeActivity({ diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index f84c978a04c8..ce6057b14f5a 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -25,6 +25,7 @@ import { type OrchestrationLatestTurn, type OrchestrationThreadActivity, type OrchestrationProposedPlanId, + THINKING_ACTIVITY_KIND, type ToolLifecycleItemType, type ThreadId, type TurnId, @@ -96,6 +97,17 @@ export interface WorkLogEntry { const workLogCollapseKey = Symbol(); +/** + * Persisted model thinking (`thinking` activities from reasoning streams). + * Thinking rows stay visible outside collapsed tool groups so a turn that + * narrates through its reasoning channel still reads as a narrative. + */ +export function isThinkingWorkEntry( + entry: Pick, +): boolean { + return entry.sourceActivityKind === THINKING_ACTIVITY_KIND; +} + interface DerivedWorkLogEntry extends WorkLogEntry { sourceActivityKind: OrchestrationThreadActivity["kind"]; [workLogCollapseKey]?: string; diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 4d2f80a1101a..92d81b764e7d 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -562,6 +562,20 @@ export const OrchestrationThreadActivityTone = Schema.Literals([ ]); export type OrchestrationThreadActivityTone = typeof OrchestrationThreadActivityTone.Type; +/** + * Activity kind for persisted model thinking/reasoning streams + * (`reasoning_text` / `reasoning_summary_text` content deltas). Providers that + * narrate through their reasoning channel (notably Gemini) would otherwise + * render as bare tool lists: ingestion buffers those streams and flushes them + * as `thinking` activities so every client can render them as collapsible + * rows interleaved with the tool calls they explain. Kind stays a plain + * string (like every other activity kind) so older readers keep decoding. + */ +export const THINKING_ACTIVITY_KIND = "thinking"; + +/** Row label for thinking activities. Clients fall back to the summary. */ +export const THINKING_ACTIVITY_SUMMARY = "Thinking"; + export const OrchestrationThreadActivity = Schema.Struct({ id: EventId, tone: OrchestrationThreadActivityTone, From b34067a1c25a5d83436896c9c6cac8b652a15e0a Mon Sep 17 00:00:00 2001 From: Derrick Koko Date: Wed, 9 Sep 2026 15:52:52 +0100 Subject: [PATCH 2/2] fix(timeline): harden thinking buffering against unbounded growth and loss Reviewers flagged that buffered reasoning could grow a single cache value without bound before flushing, was consumed before a flush finished, split astral characters across activities, stranded superseded turns on session exit, and hid thinking rows inside the live work group on web while a turn runs. - Cap the per-turn buffer at three activity chunks; crossing it spills the oldest full chunks as thinking activities while appending (persisting while buffering) instead of accumulating text forever. - Split and flush on code-point boundaries so surrogate pairs stay whole. - Consume the buffer only after every chunk dispatch commits; a failed flush keeps the text for the next flush point (deterministic chunk ids make re-appends idempotent in the projection). - Flush every buffered turn for the thread on session.exited, not just the active turn, so steering-superseded reasoning reaches the timeline. - Scope reasoning deltas without a turn id to the active turn. - Exclude thinking entries from the web active-work collector so they render standalone during a running turn. Tests: spill-without-loss, astral chunking, unscoped-scoping, exit flush-all (including a no-turn buffer and a duplicate exit) on the server; a running- turn regression case on the web timeline. Built with Muse Spark (deepseek-v4-flash) via the OpenCode harness. --- .../Layers/ProviderRuntimeIngestion.test.ts | 299 ++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 267 ++++++++++++---- .../chat/MessagesTimeline.logic.test.ts | 67 ++++ .../components/chat/MessagesTimeline.logic.ts | 4 + 4 files changed, 583 insertions(+), 54 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 6ca57fd25949..29c22ca71f58 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1546,6 +1546,305 @@ describe("ProviderRuntimeIngestion", () => { ).toBe(false); }); + it("spills oversized reasoning buffers into activities while buffering without losing text", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + const longThought = "y".repeat(25_000); + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-spill-reason"), + provider: ProviderDriverKind.make("opencode"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-spill"), + payload: { + streamKind: "reasoning_text", + delta: longThought, + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-spill-completed"), + provider: ProviderDriverKind.make("opencode"), + threadId: asThreadId("thread-1"), + createdAt: now, + turnId: asTurnId("turn-spill"), + payload: { state: "completed" }, + }); + + await harness.drain(); + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ); + const thinking = thread.activities.filter( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ); + expect(thinking.length).toBeGreaterThan(1); + const thinkingText = (activity: ProviderRuntimeTestActivity): string => { + const detail = (activity.payload as Record | undefined)?.detail; + return typeof detail === "string" ? detail : ""; + }; + expect(thinking.map(thinkingText).join("")).toBe(longThought); + for (const activity of thinking) { + expect(Array.from(thinkingText(activity)).length).toBeLessThanOrEqual(8_000); + } + }); + + it("keeps astral characters whole when splitting thinking buffers", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + // 7,999 BMP chars put the astral char exactly on the 8,000th UTF-16 unit + // boundary of the first activity; a UTF-16 split would halve its pair. + const astralBoundaryThought = `${"a".repeat(7_999)}𝄞${"b".repeat(2_000)}`; + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-astral-reason"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-astral"), + payload: { + streamKind: "reasoning_text", + delta: astralBoundaryThought, + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-astral-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: now, + turnId: asTurnId("turn-astral"), + payload: { state: "completed" }, + }); + + await harness.drain(); + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ); + const thinking = thread.activities.filter( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ); + expect(thinking.length).toBeGreaterThan(1); + const details = thinking.map((activity) => { + const detail = (activity.payload as Record | undefined)?.detail; + return typeof detail === "string" ? detail : ""; + }); + expect(details.join("")).toBe(astralBoundaryThought); + for (const detail of details) { + // No lone surrogates: every chunk decodes to whole code points. + expect(Array.from(detail).join("")).toBe(detail); + expect(Array.from(detail).length).toBeLessThanOrEqual(8_000); + } + }); + + it("scopes reasoning deltas without a turn id to the active turn", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + const turnId = asTurnId("turn-scoped"); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-scoped-turn-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId, + }); + await waitForThread( + harness.readModel, + (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === turnId, + ); + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-unscoped-reason"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + streamKind: "reasoning_text", + delta: "Reasoning without a turn id.", + }, + }); + harness.emit({ + type: "item.completed", + eventId: asEventId("evt-scoped-tool"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId, + itemId: asItemId("tool-scoped"), + payload: { + itemType: "dynamic_tool_call", + status: "completed", + title: "Tool call", + }, + }); + + await harness.drain(); + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ); + const thinking = thread.activities.find( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ); + expect(thinking?.turnId).toBe(turnId); + expect((thinking?.payload as Record | undefined)?.detail).toBe( + "Reasoning without a turn id.", + ); + }); + + it("flushes buffered thinking for every stranded turn when the session exits", async () => { + const harness = await createHarness(); + const threadId = asThreadId("thread-1"); + const oldTurnId = asTurnId("turn-steered-over"); + const newTurnId = asTurnId("turn-from-steer"); + const createdAt = "2026-01-01T00:00:00.000Z"; + + // Reasoning buffered before any turn exists lands on the thread's + // "no-turn" key; steering strands a second buffer under the superseded + // turn; session exit must flush both along with the active turn's. + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-no-turn-reason"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId, + payload: { + streamKind: "reasoning_text", + delta: "preamble", + }, + }); + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-steered-over-started"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId, + turnId: oldTurnId, + }); + await waitForThread( + harness.readModel, + (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === oldTurnId, + ); + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-steered-over-reason"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId, + turnId: oldTurnId, + payload: { + streamKind: "reasoning_text", + delta: "superseded thinking", + }, + }); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-steer-away"), + threadId, + message: { + messageId: asMessageId("msg-steer-away"), + role: "user", + text: "do it differently", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + }), + ); + harness.setProviderSession({ + provider: ProviderDriverKind.make("codex"), + status: "running", + runtimeMode: "approval-required", + threadId, + createdAt, + updatedAt: createdAt, + activeTurnId: newTurnId, + }); + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-steer-turn-started"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId, + turnId: newTurnId, + }); + await waitForThread( + harness.readModel, + (thread) => thread.session?.status === "running" && thread.session?.activeTurnId === newTurnId, + ); + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-steer-reason"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId, + turnId: newTurnId, + payload: { + streamKind: "reasoning_text", + delta: "active turn thinking", + }, + }); + + harness.emit({ + type: "session.exited", + eventId: asEventId("evt-session-exited-stranded"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId, + }); + await harness.drain(); + + const thread = await waitForThread(harness.readModel, (entry) => { + const thinkingCount = entry.activities.filter( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ).length; + return thinkingCount >= 3; + }); + const thinking = thread.activities.filter( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ); + expect(thinking).toHaveLength(3); + const byDetail = (detail: string) => + thinking.find( + (activity) => (activity.payload as Record | undefined)?.detail === detail, + ); + expect(byDetail("preamble")?.turnId).toBeNull(); + expect(byDetail("superseded thinking")?.turnId).toBe(oldTurnId); + expect(byDetail("active turn thinking")?.turnId).toBe(newTurnId); + + // A duplicate exit finds no buffers left and must not duplicate rows. + harness.emit({ + type: "session.exited", + eventId: asEventId("evt-duplicate-session-exited-stranded"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:02.000Z", + threadId, + }); + await harness.drain(); + const afterSecondExit = (await harness.readModel()).threads.find( + (entry) => entry.id === threadId, + ); + expect( + afterSecondExit?.activities.filter( + (activity: ProviderRuntimeTestActivity) => activity.kind === "thinking", + ), + ).toHaveLength(3); + }); + it("maps canonical content delta/item completed into finalized assistant messages", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 6a007ea8b537..7cafcb1391ef 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -57,8 +57,14 @@ import { canReplaceThreadTitle } from "../threadTitles.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`; const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`; +// Turn-keyed thinking buffers reuse the provider turn key; reasoning that +// arrived while no turn was active lands on the thread's "no-turn" key. Both +// share the thread-id prefix so a session-exit sweep can find every buffer +// the thread still holds (the "no-turn" suffix is reserved: provider turn ids +// never equal it). +const NO_TURN_BUFFER_SUFFIX = "no-turn"; const thinkingBufferKey = (threadId: ThreadId, turnId: TurnId | undefined) => - turnId ? providerTurnKey(threadId, turnId) : `${threadId}:no-turn`; + turnId ? providerTurnKey(threadId, turnId) : `${threadId}:${NO_TURN_BUFFER_SUFFIX}`; // Fallback when the in-memory description cache no longer has the task name // (server restart, session-exit sweep, TTL/capacity eviction): earlier @@ -116,10 +122,16 @@ const BUFFERED_PROPOSED_PLAN_BY_ID_TTL = Duration.minutes(120); const TASK_DESCRIPTION_BY_TASK_CACHE_CAPACITY = 10_000; const TASK_DESCRIPTION_BY_TASK_TTL = Duration.minutes(120); const MAX_BUFFERED_ASSISTANT_CHARS = 24_000; -// One thinking activity holds at most this much text; longer buffers split -// across several activities so a thinking-heavy turn (e.g. Gemini) cannot -// stuff an unbounded blob into a single activity payload. -const MAX_THINKING_CHARS_PER_ACTIVITY = 8_000; +// One thinking activity holds at most this many code points; longer buffers +// split across several activities so a thinking-heavy turn (e.g. Gemini) +// cannot stuff an unbounded blob into a single activity payload. Splitting on +// code points (not UTF-16 units) keeps astral characters whole. +const MAX_THINKING_CODE_POINTS_PER_ACTIVITY = 8_000; +// Reasoning buffers at most this many full activity chunks between flush +// points. A turn can reason for a long time without flushing (no tool rows, +// no pauses); crossing the cap spills the oldest completed chunks as thinking +// activities while appending so a single cache value stays bounded. +const MAX_BUFFERED_THINKING_CHUNKS = 3; const BUFFERED_THINKING_TEXT_BY_TURN_CACHE_CAPACITY = 10_000; const BUFFERED_THINKING_TEXT_BY_TURN_TTL = Duration.minutes(120); const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0"; @@ -960,9 +972,11 @@ const make = Effect.gen(function* () { // Reasoning streams (`reasoning_text` / `reasoning_summary_text`) buffer // here until a flush point (tool activity, approval pause, assistant - // completion, turn end) persists them as `thinking` activities. Without - // this, providers that narrate through their reasoning channel read as - // bare tool lists with only a final answer. + // completion, turn end, session exit) persists them as `thinking` + // activities. Without this, providers that narrate through their reasoning + // channel read as bare tool lists with only a final answer. Entries hold at + // most MAX_BUFFERED_THINKING_CHUNKS per-activity budgets of text; overflow + // spills to activities while appending (see appendBufferedThinkingText). const bufferedThinkingTextByTurnKey = yield* Cache.make({ capacity: BUFFERED_THINKING_TEXT_BY_TURN_CACHE_CAPACITY, timeToLive: BUFFERED_THINKING_TEXT_BY_TURN_TTL, @@ -1161,49 +1175,130 @@ const make = Effect.gen(function* () { const clearBufferedAssistantText = (messageId: MessageId) => Cache.invalidate(bufferedAssistantTextByMessageId, messageId); + // Reasoning deltas accumulate in one cache entry per turn until a flush + // point (tool activity, approval pause, assistant completion, turn end, + // session exit) persists them as `thinking` activities. The value is capped + // so a turn that reasons without any flush point cannot grow one entry + // without bound: crossing the cap spills the oldest full chunks as + // activities right away (persisting while buffering) and keeps only the + // newest tail buffered. The returned spill chunks carry the buffer's first + // delta timestamp so every chunk of one reasoning stretch sorts together. const appendBufferedThinkingText = ( threadId: ThreadId, turnId: TurnId | undefined, delta: string, createdAt: string, - ) => + ): Effect.Effect< + | { readonly firstCreatedAt: string; readonly chunks: ReadonlyArray } + | null + > => Cache.getOption(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId)).pipe( Effect.flatMap((existing) => { const previous = Option.getOrElse(existing, () => ({ text: "", firstCreatedAt: "" })); - return Cache.set(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId), { - text: `${previous.text}${delta}`, - firstCreatedAt: - previous.firstCreatedAt.length > 0 ? previous.firstCreatedAt : createdAt, - }); - }), - ); + const combinedText = `${previous.text}${delta}`; + const firstCreatedAt = + previous.firstCreatedAt.length > 0 ? previous.firstCreatedAt : createdAt; + // Fast path: text within one per-activity budget in UTF-16 units also + // fits it in code points, so no code-point scan is needed. + if (combinedText.length <= MAX_THINKING_CODE_POINTS_PER_ACTIVITY) { + return Cache.set( + bufferedThinkingTextByTurnKey, + thinkingBufferKey(threadId, turnId), + { text: combinedText, firstCreatedAt }, + ).pipe(Effect.as(null)); + } - const takeBufferedThinkingText = ( - threadId: ThreadId, - turnId: TurnId | undefined, - ): Effect.Effect => - Cache.getOption(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId)).pipe( - Effect.flatMap((existing) => - Cache.invalidate(bufferedThinkingTextByTurnKey, thinkingBufferKey(threadId, turnId)).pipe( - Effect.as( - Option.getOrElse(existing, (): BufferedThinking => ({ text: "", firstCreatedAt: "" })), - ), - ), - ), + // Split on code points so a surrogate pair never straddles the spill + // boundary or an activity chunk. The oldest whole chunks spill; the + // tail keeps at most MAX_BUFFERED_THINKING_CHUNKS of them buffered. + const codePoints = Array.from(combinedText); + const combinedChunkCount = Math.ceil( + codePoints.length / MAX_THINKING_CODE_POINTS_PER_ACTIVITY, + ); + if (combinedChunkCount <= MAX_BUFFERED_THINKING_CHUNKS) { + return Cache.set( + bufferedThinkingTextByTurnKey, + thinkingBufferKey(threadId, turnId), + { text: combinedText, firstCreatedAt }, + ).pipe(Effect.as(null)); + } + const spillChunkCount = combinedChunkCount - MAX_BUFFERED_THINKING_CHUNKS; + const spillPointCount = spillChunkCount * MAX_THINKING_CODE_POINTS_PER_ACTIVITY; + const spillText = codePoints.slice(0, spillPointCount).join(""); + const tailText = codePoints.slice(spillPointCount).join(""); + return Cache.set( + bufferedThinkingTextByTurnKey, + thinkingBufferKey(threadId, turnId), + { text: tailText, firstCreatedAt }, + ).pipe( + Effect.as({ + firstCreatedAt, + chunks: splitThinkingText(spillText), + }), + ); + }), ); + // Splits reasoning text into per-activity chunks on code-point boundaries. + // A whole chunk keeps its length in code points within the per-activity + // budget; astral characters (surrogate pairs in UTF-16) stay intact. const splitThinkingText = (text: string): Array => { + const codePoints = Array.from(text); const chunks: Array = []; - for (let index = 0; index < text.length; index += MAX_THINKING_CHARS_PER_ACTIVITY) { - chunks.push(text.slice(index, index + MAX_THINKING_CHARS_PER_ACTIVITY)); + for ( + let index = 0; + index < codePoints.length; + index += MAX_THINKING_CODE_POINTS_PER_ACTIVITY + ) { + chunks.push( + codePoints.slice(index, index + MAX_THINKING_CODE_POINTS_PER_ACTIVITY).join(""), + ); } return chunks; }; + const dispatchThinkingChunk = (input: { + event: ProviderRuntimeEvent; + threadId: ThreadId; + turnId?: TurnId; + /** Activity timestamp: the chunked reasoning's first delta. */ + activityCreatedAt: string; + /** Command timestamp: the event that triggered this flush or spill. */ + createdAt: string; + commandTag: string; + chunkIdTag: string; + chunkIndex: number; + chunk: string; + }) => + Effect.gen(function* () { + yield* orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: yield* providerCommandId( + input.event, + `${input.commandTag}:${input.chunkIndex}`, + ), + threadId: input.threadId, + activity: { + id: EventId.make( + `${input.event.eventId}:thinking:${input.chunkIdTag}:${input.turnId ?? NO_TURN_BUFFER_SUFFIX}:${input.chunkIndex}`, + ), + createdAt: input.activityCreatedAt, + tone: "info", + kind: THINKING_ACTIVITY_KIND, + summary: THINKING_ACTIVITY_SUMMARY, + payload: { detail: input.chunk }, + turnId: input.turnId ?? null, + }, + createdAt: input.createdAt, + }); + }); + // Persists buffered reasoning as `thinking` activities, oldest chunk first, - // so the rows interleave with the tool calls the thinking explains. Consumes - // the buffer; a later flush for the same turn is a no-op until new deltas - // arrive. Long buffers split across several activities instead of dropping. + // so the rows interleave with the tool calls the thinking explains. The + // buffer is only consumed once every chunk committed: a flush that fails + // halfway keeps the whole text so the next flush point retries it (chunk + // activity ids are deterministic, so re-appending a chunk that already + // committed replaces it in the projection instead of duplicating it). // Activities carry the first delta's timestamp so they sort before the tool // rows that flushed them. const flushBufferedThinkingForTurn = (input: { @@ -1214,28 +1309,68 @@ const make = Effect.gen(function* () { commandTag: string; }) => Effect.gen(function* () { - const buffered = yield* takeBufferedThinkingText(input.threadId, input.turnId); + const key = thinkingBufferKey(input.threadId, input.turnId); + const existing = yield* Cache.getOption(bufferedThinkingTextByTurnKey, key); + const buffered = Option.getOrElse( + existing, + (): BufferedThinking => ({ text: "", firstCreatedAt: "" }), + ); if (!hasRenderableAssistantText(buffered.text)) { + // Whitespace-only (or absent) buffer: drop the entry so a later flush + // for the same turn stays a no-op until new deltas arrive. + yield* Cache.invalidate(bufferedThinkingTextByTurnKey, key); return; } const chunks = splitThinkingText(buffered.text); - for (const [chunkIndex, chunk] of chunks.entries()) { - yield* orchestrationEngine.dispatch({ - type: "thread.activity.append", - commandId: yield* providerCommandId(input.event, `${input.commandTag}:${chunkIndex}`), + yield* Effect.forEach(chunks, (chunk, chunkIndex) => + dispatchThinkingChunk({ + event: input.event, threadId: input.threadId, - activity: { - id: EventId.make(`${input.event.eventId}:thinking:${chunkIndex}`), - createdAt: buffered.firstCreatedAt, - tone: "info", - kind: THINKING_ACTIVITY_KIND, - summary: THINKING_ACTIVITY_SUMMARY, - payload: { detail: chunk }, - turnId: input.turnId ?? null, - }, + ...(input.turnId ? { turnId: input.turnId } : {}), + activityCreatedAt: buffered.firstCreatedAt, createdAt: input.createdAt, - }); - } + commandTag: input.commandTag, + chunkIdTag: "flush", + chunkIndex, + chunk, + }), + ); + yield* Cache.invalidate(bufferedThinkingTextByTurnKey, key); + }); + + // Session exit can strand reasoning from turns the provider superseded + // without completing them (steering, abrupt stops). Flush every buffered + // stream the thread still holds — not just the active turn's — so none of + // them is dropped from the timeline. + const flushBufferedThinkingForThread = (input: { + event: ProviderRuntimeEvent; + threadId: ThreadId; + createdAt: string; + commandTag: string; + }) => + Effect.gen(function* () { + const threadPrefix = `${input.threadId}:`; + const keys = Array.from(yield* Cache.keys(bufferedThinkingTextByTurnKey)); + yield* Effect.forEach( + keys, + (key) => + Effect.gen(function* () { + if (!key.startsWith(threadPrefix)) { + return; + } + const turnKeySegment = key.slice(threadPrefix.length); + yield* flushBufferedThinkingForTurn({ + event: input.event, + threadId: input.threadId, + ...(turnKeySegment === NO_TURN_BUFFER_SUFFIX + ? {} + : { turnId: TurnId.make(turnKeySegment) }), + createdAt: input.createdAt, + commandTag: input.commandTag, + }); + }), + { concurrency: 1 }, + ).pipe(Effect.asVoid); }); const appendBufferedProposedPlan = (planId: string, delta: string, createdAt: string) => @@ -1591,12 +1726,38 @@ const make = Effect.gen(function* () { ) { const thread = yield* resolveThreadRuntimeContext(event.threadId); if (thread) { - yield* appendBufferedThinkingText( + // Providers do not always tag reasoning deltas with the running + // turn; scope them to the active turn so they flush alongside its + // tool rows instead of lingering unscoped until a no-turn flush + // or TTL expiry. + const reasoningTurnId = + toTurnId(event.turnId) ?? thread.session?.activeTurnId ?? undefined; + const buffered = yield* appendBufferedThinkingText( thread.id, - toTurnId(event.turnId), + reasoningTurnId, event.payload.delta, event.createdAt, ); + if (buffered !== null) { + // The buffer cap overflowed: persist the oldest full chunks as + // activities right away (the retained tail stays buffered). If + // this dispatch fails the spilled text is dropped — the same + // trade-off as the assistant buffer spill — while the retained + // tail still flushes at the next flush point. + yield* Effect.forEach(buffered.chunks, (chunk, chunkIndex) => + dispatchThinkingChunk({ + event, + threadId: thread.id, + ...(reasoningTurnId ? { turnId: reasoningTurnId } : {}), + activityCreatedAt: buffered.firstCreatedAt, + createdAt: event.createdAt, + commandTag: "thinking-spill", + chunkIdTag: "spill", + chunkIndex, + chunk, + }), + ); + } } } return; @@ -2058,11 +2219,9 @@ const make = Effect.gen(function* () { } if (event.type === "session.exited") { - const exitedActiveTurnId = thread.session?.activeTurnId ?? undefined; - yield* flushBufferedThinkingForTurn({ + yield* flushBufferedThinkingForThread({ event, threadId: thread.id, - ...(exitedActiveTurnId ? { turnId: exitedActiveTurnId } : {}), createdAt: now, commandTag: "thinking-flush-on-session-exit", }); diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 10ff146dfcd2..b98ed10ffcdd 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -1170,6 +1170,73 @@ describe("deriveMessagesTimelineRows", () => { }); }); + it("keeps thinking visible outside the live work group while the turn runs", () => { + const turnId = "turn-1" as never; + const rows = deriveMessagesTimelineRows({ + timelineEntries: [ + { + id: "running-command-entry", + kind: "work" as const, + createdAt: "2026-01-01T00:00:01Z", + entry: { + id: "running-command", + createdAt: "2026-01-01T00:00:01Z", + turnId, + label: "Running tests", + command: "vp test run", + requestKind: "command", + tone: "tool" as const, + toolLifecycleStatus: "inProgress" as const, + }, + }, + { + id: "thinking-1", + kind: "work" as const, + createdAt: "2026-01-01T00:00:02Z", + entry: { + id: "thinking-1", + createdAt: "2026-01-01T00:00:02Z", + turnId, + label: "Thinking", + detail: "Checking the checklist first.", + tone: "info" as const, + sourceActivityKind: "thinking", + }, + }, + ], + latestTurn: { + turnId, + state: "running", + startedAt: "2026-01-01T00:00:00Z", + completedAt: null, + }, + isWorking: true, + activeTurnStartedAt: "2026-01-01T00:00:00Z", + turnDiffSummaries: [], + supportsConversationRollback: false, + }); + + const liveRow = rows.find((row) => row.kind === "work-live"); + expect(liveRow).toBeDefined(); + expect(liveRow).toMatchObject({ + entry: { id: "running-command" }, + active: true, + }); + // The thinking entry must not be swept into the collapsed live group. + expect( + rows.some( + (row) => + row.kind === "work-live" && + row.groupedEntries.some((entry) => entry.sourceActivityKind === "thinking"), + ), + ).toBe(false); + expect(rows.map((row) => row.id)).toEqual([ + "working-indicator-row", + "work-live:running-command-entry", + "work-toggle:thinking-1", + ]); + }); + it("only enables assistant copy for the terminal assistant message in a turn", () => { const rows = deriveMessagesTimelineRows({ timelineEntries: [ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 3205196c6131..fb0bb303f72c 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -921,6 +921,10 @@ export function deriveMessagesTimelineRows(input: { entry.kind !== "work" || entry.entry.agentSpawn !== undefined || entry.entry.sourceActivityKind === "context-compaction" || + // Thinking rows stay standalone outside the live work group (see the + // grouping boundary below); collecting them here would hide them inside + // the collapsed `work-live` group while the turn runs. + isThinkingWorkEntry(entry.entry) || entry.entry.tone === "error" ) { break;