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..29c22ca71f58 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,508 @@ 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("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 8d34fee4f981..7cafcb1391ef 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,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_BUFFER_SUFFIX}`; // Fallback when the in-memory description cache no longer has the task name // (server restart, session-exit sweep, TTL/capacity eviction): earlier @@ -97,6 +107,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 +122,18 @@ 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 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"; type TurnStartRequestedDomainEvent = Extract< @@ -942,6 +970,19 @@ 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, 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, + 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 +1175,204 @@ 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: "" })); + 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)); + } + + // 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 < 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. 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: { + event: ProviderRuntimeEvent; + threadId: ThreadId; + turnId?: TurnId; + createdAt: string; + commandTag: string; + }) => + Effect.gen(function* () { + 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); + yield* Effect.forEach(chunks, (chunk, chunkIndex) => + dispatchThinkingChunk({ + event: input.event, + threadId: input.threadId, + ...(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) => Cache.getOption(bufferedProposedPlanById, planId).pipe( Effect.flatMap((existingEntry) => { @@ -1477,6 +1716,50 @@ 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) { + // 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, + 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; } @@ -1702,6 +1985,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 +2057,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 +2135,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 +2219,12 @@ const make = Effect.gen(function* () { } if (event.type === "session.exited") { + yield* flushBufferedThinkingForThread({ + event, + threadId: thread.id, + createdAt: now, + commandTag: "thinking-flush-on-session-exit", + }); yield* clearTurnStateForSession(thread.id); } @@ -2126,6 +2440,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..b98ed10ffcdd 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -1120,6 +1120,123 @@ 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("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 a87531673506..fb0bb303f72c 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, @@ -920,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; @@ -1054,15 +1059,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,