diff --git a/cli/src/agent/backends/acp/AcpMessageHandler.test.ts b/cli/src/agent/backends/acp/AcpMessageHandler.test.ts index 55e14b04..0922ce0c 100644 --- a/cli/src/agent/backends/acp/AcpMessageHandler.test.ts +++ b/cli/src/agent/backends/acp/AcpMessageHandler.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { join } from 'node:path'; import { fileURLToPath } from 'node:url'; import type { AgentMessage } from '@/agent/types'; @@ -16,6 +16,14 @@ function getToolResult(messages: AgentMessage[], id: string): Extract { + beforeEach(() => { + vi.spyOn(Date, 'now').mockReturnValue(0); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + it('does not synthesize {status} output when tool completes without payload', () => { const messages: AgentMessage[] = []; const handler = new AcpMessageHandler((message) => messages.push(message)); @@ -759,6 +767,98 @@ describe('AcpMessageHandler', () => { ]); }); + it('streams throttled reasoning snapshots with a stable id before final flush', () => { + let now = 0; + vi.spyOn(Date, 'now').mockImplementation(() => now); + + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'first ' } + }); + expect(messages).toEqual([]); + + now = 300; + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'second' } + }); + + expect(messages).toHaveLength(1); + expect(messages[0]).toMatchObject({ + type: 'reasoning', + text: 'first second', + live: true + }); + const streamId = (messages[0] as Extract).id; + expect(streamId).toEqual(expect.any(String)); + + handler.flushReasoning(); + + expect(messages).toHaveLength(2); + expect(messages[1]).toEqual({ + type: 'reasoning', + text: 'first second', + id: streamId + }); + }); + + it('does not split reasoning on ignored agent message chunks', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'first ' } + }); + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: '' } + }); + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { + type: 'text', + text: 'user-only bookkeeping', + annotations: { audience: ['user'] } + } + }); + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'second' } + }); + handler.drainBuffers(); + + expect(messages).toEqual([ + { type: 'reasoning', text: 'first second' } + ]); + }); + + it('does not split reasoning on unknown ACP bookkeeping updates', () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message)); + + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'first ' } + }); + handler.handleUpdate({ + sessionUpdate: 'session_status', + status: 'running' + }); + handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'second' } + }); + handler.drainBuffers(); + + expect(messages).toEqual([ + { type: 'reasoning', text: 'first second' } + ]); + }); + it('emits buffered reasoning before a tool_call boundary', () => { const messages: AgentMessage[] = []; const handler = new AcpMessageHandler((message) => messages.push(message)); @@ -789,10 +889,10 @@ describe('AcpMessageHandler', () => { expect(messages[1]).toMatchObject({ type: 'tool_call', id: 'tc-1' }); }); - // Locks the flush-before-every-non-thought-boundary contract introduced - // in this fix: a future refactor that forgets to call flushReasoning() in - // one branch of handleUpdate would otherwise silently regress reasoning - // ordering for that update type. + // Locks the flush-before-visible-boundary contract: a future refactor + // that forgets to call flushReasoning() in one visible branch of + // handleUpdate would otherwise silently regress reasoning ordering for + // that update type. it.each([ [ 'agentMessageChunk', diff --git a/cli/src/agent/backends/acp/AcpMessageHandler.ts b/cli/src/agent/backends/acp/AcpMessageHandler.ts index 3f956eb2..8998d2e7 100644 --- a/cli/src/agent/backends/acp/AcpMessageHandler.ts +++ b/cli/src/agent/backends/acp/AcpMessageHandler.ts @@ -1,4 +1,5 @@ import type { AgentMessage, PlanItem } from '@/agent/types'; +import { randomUUID } from 'node:crypto'; import { asString, isObject } from '@hapi/protocol'; import { deriveToolNameWithSource, isPlaceholderToolName } from '@/agent/utils'; import { parseRateLimitText } from '@/agent/rateLimitParser'; @@ -14,6 +15,8 @@ function normalizeStatus(status: unknown): 'pending' | 'in_progress' | 'complete type DerivedToolName = ReturnType; +const REASONING_SNAPSHOT_INTERVAL_MS = 250; + /** * Extracts _meta.kind from the first diff block in a content array. * Returns null when content is not an array, is empty, or the first block @@ -278,6 +281,10 @@ export class AcpMessageHandler { // otherwise incur — a 10k-token reasoning trace allocates 10k full-buffer // copies if we use `+=`. private bufferedReasoning: string[] = []; + private reasoningStreamId: string | null = null; + private lastReasoningSnapshotAt: number | null = null; + private lastReasoningSnapshotText = ''; + private reasoningSnapshotEmitted = false; constructor(private readonly onMessage: (message: AgentMessage) => void) {} @@ -299,14 +306,14 @@ export class AcpMessageHandler { /** * Emits buffered thought chunks as a single reasoning message and clears * the buffer. ACP agents (notably OpenCode/Zen) stream thoughts at the - * granularity of one chunk per token; emitting each chunk inline would - * make the web reducer render one row per token. Coalescing here keeps - * the reasoning block intact while preserving its position relative to - * adjacent text segments and tool events. + * granularity of one chunk per token; raw per-token messages would make + * the web reducer render one row per token. We stream throttled full-text + * snapshots with a stable id while the buffer is open, then emit one final + * message with the same id at the boundary. * - * Called automatically before every non-thought update inside - * `handleUpdate`, and externally at turn boundaries by `drainBuffers` - * from AcpSdkBackend. + * Called automatically before visible boundaries inside `handleUpdate` + * (assistant text, tool lifecycle, plan), and externally at turn + * boundaries by `drainBuffers` from AcpSdkBackend. * * Whitespace-only buffers are dropped: a turn that happens to emit a * single whitespace token would otherwise render an empty Reasoning row @@ -317,11 +324,12 @@ export class AcpMessageHandler { return; } const text = this.bufferedReasoning.join(''); - this.bufferedReasoning = []; + const id = this.reasoningSnapshotEmitted ? this.reasoningStreamId ?? undefined : undefined; + this.resetReasoningState(); if (text.trim().length === 0) { return; } - this.onMessage({ type: 'reasoning', text }); + this.onMessage(id ? { type: 'reasoning', text, id } : { type: 'reasoning', text }); } /** @@ -375,6 +383,56 @@ export class AcpMessageHandler { this.bufferedText += text; } + private appendReasoningChunk(text: string): void { + if (!text) { + return; + } + this.bufferedReasoning.push(text); + if (!this.reasoningStreamId) { + this.reasoningStreamId = randomUUID(); + } + this.emitReasoningSnapshotIfDue(); + } + + private emitReasoningSnapshotIfDue(): void { + if (!this.reasoningStreamId) { + return; + } + + const now = Date.now(); + if (this.lastReasoningSnapshotAt === null) { + this.lastReasoningSnapshotAt = now; + return; + } + if (now - this.lastReasoningSnapshotAt < REASONING_SNAPSHOT_INTERVAL_MS) { + return; + } + + const text = this.bufferedReasoning.join(''); + if (text.trim().length === 0 || text === this.lastReasoningSnapshotText) { + this.lastReasoningSnapshotAt = now; + return; + } + + this.lastReasoningSnapshotAt = now; + this.lastReasoningSnapshotText = text; + this.reasoningSnapshotEmitted = true; + this.onMessage({ + type: 'reasoning', + text, + id: this.reasoningStreamId, + live: true + }); + } + + private resetReasoningState(): void { + this.bufferedReasoning = []; + this.reasoningStreamId = null; + this.lastReasoningSnapshotAt = null; + this.lastReasoningSnapshotText = ''; + this.reasoningSnapshotEmitted = false; + } + handleUpdate(update: unknown): void { if (!isObject(update)) return; const updateType = asString(update.sessionUpdate); @@ -394,17 +452,11 @@ export class AcpMessageHandler { // should not cause the reasoning to be silently dropped. const content = update.content; if (isObject(content) && content.type === 'text' && typeof content.text === 'string' && content.text.length > 0) { - this.bufferedReasoning.push(content.text); + this.appendReasoningChunk(content.text); } return; } - // Any non-thought update is a reasoning-segment boundary: emit the - // accumulated thought now so it arrives before the next event in - // the same arrival order that streamed in. Tool calls / plans - // additionally flush the text buffer below. - this.flushReasoning(); - if (updateType === ACP_SESSION_UPDATE_TYPES.agentMessageChunk) { const content = update.content; const text = extractTextContent(content); @@ -423,6 +475,7 @@ export class AcpMessageHandler { if (rateLimit.suppress) { return; } + this.flushReasoning(); this.flushText(); this.onMessage(rateLimit.message); return; @@ -435,12 +488,20 @@ export class AcpMessageHandler { } return; } + // Visible assistant text is a reasoning-segment boundary: + // emit accumulated thoughts first so the rendered turn keeps + // Reasoning above the answer. Empty / filtered message chunks + // are not boundaries; OpenCode can interleave bookkeeping + // updates while streaming thoughts, and flushing on those + // would split reasoning back into one row per token. + this.flushReasoning(); this.appendTextChunk(text); } return; } if (updateType === ACP_SESSION_UPDATE_TYPES.toolCall) { + this.flushReasoning(); // A new tool invocation closes the preceding text segment. // Flushing here preserves the arrival order between text and // tool lifecycle events without disturbing cumulative dedup @@ -451,16 +512,20 @@ export class AcpMessageHandler { } if (updateType === ACP_SESSION_UPDATE_TYPES.toolCallUpdate) { - // Do not flush here: a toolCallUpdate is a lifecycle event on - // an already-open tool call, not a boundary between text + this.flushReasoning(); + // Do not flush text here: a toolCallUpdate is a lifecycle event + // on an already-open tool call, not a boundary between text // segments. If the agent streams a new text segment while the - // tool is running, flushing here would leak that segment - // across the tool_result boundary. + // tool is running, flushing text here would leak that segment + // across the tool_result boundary. Reasoning is separate and is + // flushed above so tool results still appear after the thought + // that led to them. this.handleToolCallUpdate(update); return; } if (updateType === ACP_SESSION_UPDATE_TYPES.plan) { + this.flushReasoning(); this.flushText(); const items = normalizePlanEntries(update.entries); if (items.length > 0) { diff --git a/cli/src/agent/messageConverter.test.ts b/cli/src/agent/messageConverter.test.ts index 97dae110..628c7495 100644 --- a/cli/src/agent/messageConverter.test.ts +++ b/cli/src/agent/messageConverter.test.ts @@ -35,4 +35,18 @@ describe('convertAgentMessage', () => { is_error: true }); }); + + it('preserves stable reasoning id when provided', () => { + const converted = convertAgentMessage({ + type: 'reasoning', + text: 'thinking', + id: 'reasoning-stream-1' + }); + + expect(converted).toEqual({ + type: 'reasoning', + message: 'thinking', + id: 'reasoning-stream-1' + }); + }); }); diff --git a/cli/src/agent/messageConverter.ts b/cli/src/agent/messageConverter.ts index c51d982f..c3313a7e 100644 --- a/cli/src/agent/messageConverter.ts +++ b/cli/src/agent/messageConverter.ts @@ -28,7 +28,7 @@ export function convertAgentMessage(message: AgentMessage): CodexMessage | null // AgentMessage uses `text` (consistent with the `text` variant); // the wire-level CodexMessage uses `message` to match the // existing reasoning format emitted by the Codex path. - return { type: 'reasoning', message: message.text, id: randomUUID() }; + return { type: 'reasoning', message: message.text, id: message.id ?? randomUUID() }; case 'tool_call': return { type: 'tool-call', diff --git a/cli/src/agent/types.ts b/cli/src/agent/types.ts index 9011937a..5b0e5dbe 100644 --- a/cli/src/agent/types.ts +++ b/cli/src/agent/types.ts @@ -30,7 +30,7 @@ export type PlanItem = { export type AgentMessage = | { type: 'text'; text: string } - | { type: 'reasoning'; text: string } + | { type: 'reasoning'; text: string; id?: string; live?: boolean } | { type: 'tool_call'; id: string; name: string; input: unknown; status: 'pending' | 'in_progress' | 'completed' | 'failed' } | { type: 'tool_result'; id: string; output: unknown; status: 'completed' | 'failed' } | { type: 'plan'; items: PlanItem[] } diff --git a/cli/src/gemini/geminiRemoteLauncher.ts b/cli/src/gemini/geminiRemoteLauncher.ts index 26691714..c130182d 100644 --- a/cli/src/gemini/geminiRemoteLauncher.ts +++ b/cli/src/gemini/geminiRemoteLauncher.ts @@ -221,6 +221,9 @@ class GeminiRemoteLauncher extends RemoteLauncherBase { this.messageBuffer.addMessage(message.text, 'assistant'); break; case 'reasoning': + if (message.live) { + break; + } this.messageBuffer.addMessage(`[Thinking] ${message.text.substring(0, 100)}...`, 'system'); break; case 'tool_call': diff --git a/cli/src/opencode/opencodeRemoteLauncher.ts b/cli/src/opencode/opencodeRemoteLauncher.ts index ab1d22ee..38ec88b8 100644 --- a/cli/src/opencode/opencodeRemoteLauncher.ts +++ b/cli/src/opencode/opencodeRemoteLauncher.ts @@ -243,6 +243,9 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { this.messageBuffer.addMessage(message.text, 'assistant'); break; case 'reasoning': + if (message.live) { + break; + } this.messageBuffer.addMessage(`[Thinking] ${message.text.substring(0, 100)}...`, 'system'); break; case 'tool_call': diff --git a/web/src/chat/normalize.test.ts b/web/src/chat/normalize.test.ts index 982cd58a..fd18148d 100644 --- a/web/src/chat/normalize.test.ts +++ b/web/src/chat/normalize.test.ts @@ -155,6 +155,29 @@ describe('normalizeDecryptedMessage', () => { }) }) + it('keeps Codex/OpenCode reasoning stream ids for snapshot merging', () => { + const normalized = normalizeDecryptedMessage(makeMessage({ + role: 'agent', + content: { + type: 'codex', + data: { + type: 'reasoning', + id: 'reasoning-stream-1', + message: 'thinking' + } + } + })) + + expect(normalized).toMatchObject({ + role: 'agent', + content: [{ + type: 'reasoning', + text: 'thinking', + streamId: 'reasoning-stream-1' + }] + }) + }) + it('treats non-sidechain string user output as sidechain', () => { const message = makeMessage({ role: 'agent', diff --git a/web/src/chat/normalizeAgent.ts b/web/src/chat/normalizeAgent.ts index e5512346..64fbc174 100644 --- a/web/src/chat/normalizeAgent.ts +++ b/web/src/chat/normalizeAgent.ts @@ -596,13 +596,14 @@ export function normalizeAgentRecord( } if (data.type === 'reasoning' && typeof data.message === 'string') { + const streamId = asString(data.id) ?? messageId return { id: messageId, localId, createdAt, role: 'agent', isSidechain: false, - content: [{ type: 'reasoning', text: data.message, uuid: messageId, parentUUID: null }], + content: [{ type: 'reasoning', text: data.message, uuid: messageId, streamId, parentUUID: null }], meta } } diff --git a/web/src/chat/reducerTimeline.test.ts b/web/src/chat/reducerTimeline.test.ts index 291ba0eb..d88b7bac 100644 --- a/web/src/chat/reducerTimeline.test.ts +++ b/web/src/chat/reducerTimeline.test.ts @@ -255,6 +255,46 @@ describe('reduceTimeline', () => { expect(agentTextBlock.model).toBeUndefined() }) + it('collapses reasoning snapshots with the same stream id', () => { + const first: TracedMessage = { + id: 'reasoning-row-1', + localId: null, + createdAt: 1_700_000_000_000, + role: 'agent', + content: [{ + type: 'reasoning', + text: 'first ', + uuid: 'reasoning-row-1', + streamId: 'reasoning-stream-1', + parentUUID: null + }], + isSidechain: false + } as TracedMessage + const second: TracedMessage = { + id: 'reasoning-row-2', + localId: null, + createdAt: 1_700_000_000_100, + role: 'agent', + content: [{ + type: 'reasoning', + text: 'first second', + uuid: 'reasoning-row-2', + streamId: 'reasoning-stream-1', + parentUUID: null + }], + isSidechain: false + } as TracedMessage + + const { blocks } = reduceTimeline([first, second], makeContext()) + const reasoningBlocks = blocks.filter((block) => block.kind === 'agent-reasoning') + + expect(reasoningBlocks).toHaveLength(1) + expect(reasoningBlocks[0]).toMatchObject({ + id: 'reasoning-row-1:0', + text: 'first second' + }) + }) + it('falls back to the last duration-bearing block when targetMessageId resolves to a non-duration block', () => { // Regression: the matcher used to take the first id-prefix match and // then silently drop the duration when that block was not duration- diff --git a/web/src/chat/reducerTimeline.ts b/web/src/chat/reducerTimeline.ts index 8efae80f..4a872f15 100644 --- a/web/src/chat/reducerTimeline.ts +++ b/web/src/chat/reducerTimeline.ts @@ -191,7 +191,7 @@ function normalizeTraceMessage( ...base, id: traceId, role: 'agent', - content: [{ type: 'reasoning', text: data.message, uuid: traceId, parentUUID: null }] + content: [{ type: 'reasoning', text: data.message, uuid: traceId, streamId: traceId, parentUUID: null }] } as TracedMessage] } @@ -270,6 +270,7 @@ export function reduceTimeline( const agentRunCardByAgentId = new Map() const agentRunTraceMessagesByCardId = new Map() const pendingAgentRunCardByFingerprint = new Map() + const reasoningBlocksByStreamId = new Map() let hasReadyEvent = false const ensureAgentRunBlock = ( @@ -746,7 +747,20 @@ export function reduceTimeline( } if (c.type === 'reasoning') { - blocks.push({ + const streamId = asString(c.streamId) + if (streamId) { + const existing = reasoningBlocksByStreamId.get(streamId) + if (existing) { + existing.text = c.text + existing.usage = msg.usage + existing.model = msg.model + existing.meta = msg.meta + existing.invokedAt = msg.invokedAt + continue + } + } + + const block: AgentReasoningBlock = { kind: 'agent-reasoning', id: `${msg.id}:${idx}`, localId: msg.localId, @@ -756,7 +770,11 @@ export function reduceTimeline( model: msg.model, text: c.text, meta: msg.meta - }) + } + blocks.push(block) + if (streamId) { + reasoningBlocksByStreamId.set(streamId, block) + } continue } diff --git a/web/src/chat/types.ts b/web/src/chat/types.ts index c66ed3e5..4641e913 100644 --- a/web/src/chat/types.ts +++ b/web/src/chat/types.ts @@ -93,6 +93,7 @@ export type NormalizedAgentContent = type: 'reasoning' text: string uuid: string + streamId?: string parentUUID: string | null } | ToolUse diff --git a/web/src/hooks/queries/useSession.test.ts b/web/src/hooks/queries/useSession.test.ts new file mode 100644 index 00000000..b5adcfdc --- /dev/null +++ b/web/src/hooks/queries/useSession.test.ts @@ -0,0 +1,13 @@ +import { describe, expect, it } from 'vitest' +import { isSessionNotFoundError } from './useSession' + +describe('isSessionNotFoundError', () => { + it('matches hub 404 session responses', () => { + expect(isSessionNotFoundError(new Error('HTTP 404 Not Found: {"error":"Session not found"}'))).toBe(true) + }) + + it('does not match unrelated errors', () => { + expect(isSessionNotFoundError(new Error('HTTP 500 Internal Server Error'))).toBe(false) + expect(isSessionNotFoundError(null)).toBe(false) + }) +}) diff --git a/web/src/hooks/queries/useSession.ts b/web/src/hooks/queries/useSession.ts index 655e5e77..d9d6e5be 100644 --- a/web/src/hooks/queries/useSession.ts +++ b/web/src/hooks/queries/useSession.ts @@ -3,10 +3,16 @@ import type { ApiClient } from '@/api/client' import type { Session } from '@/types/api' import { queryKeys } from '@/lib/query-keys' +export function isSessionNotFoundError(error: unknown): boolean { + return error instanceof Error + && (error.message.includes('HTTP 404') || error.message.includes('Session not found')) +} + export function useSession(api: ApiClient | null, sessionId: string | null): { session: Session | null isLoading: boolean error: string | null + notFound: boolean refetch: () => Promise } { const resolvedSessionId = sessionId ?? 'unknown' @@ -19,12 +25,19 @@ export function useSession(api: ApiClient | null, sessionId: string | null): { return await api.getSession(sessionId) }, enabled: Boolean(api && sessionId), + retry: (failureCount, error) => { + if (isSessionNotFoundError(error)) { + return false + } + return failureCount < 2 + }, }) return { session: query.data?.session ?? null, isLoading: query.isLoading, error: query.error instanceof Error ? query.error.message : query.error ? 'Failed to load session' : null, + notFound: isSessionNotFoundError(query.error) && !query.isFetching, refetch: query.refetch, } } diff --git a/web/src/router.tsx b/web/src/router.tsx index 6b2ac171..3f56e3f0 100644 --- a/web/src/router.tsx +++ b/web/src/router.tsx @@ -1,4 +1,4 @@ -import { useCallback, useMemo } from 'react' +import { useCallback, useEffect, useMemo } from 'react' import { useQueryClient } from '@tanstack/react-query' import { Navigate, @@ -255,6 +255,7 @@ function SessionPage() { const { sessionId } = useParams({ from: '/sessions/$sessionId' }) const { session, + error: sessionError, refetch: refetchSession, } = useSession(api, sessionId) const { @@ -360,6 +361,30 @@ function SessionPage() { }, [refetchMessages, refetchSession]) if (!session) { + if (sessionError) { + return ( +
+
Session unavailable
+
{sessionError}
+
+ + +
+
+ ) + } return (
@@ -394,11 +419,29 @@ function SessionPage() { } function SessionDetailRoute() { + const { api } = useAppContext() const pathname = useLocation({ select: location => location.pathname }) const { sessionId } = useParams({ from: '/sessions/$sessionId' }) + const navigate = useNavigate() + const { notFound: sessionNotFound } = useSession(api, sessionId) const basePath = `/sessions/${sessionId}` const isChat = pathname === basePath || pathname === `${basePath}/` + useEffect(() => { + if (!sessionNotFound) { + return + } + navigate({ to: '/sessions', replace: true }) + }, [navigate, sessionNotFound]) + + if (sessionNotFound) { + return ( +
+ +
+ ) + } + return isChat ? : }