From 99d373591fb68a190b92562e22e1048a9a094fd4 Mon Sep 17 00:00:00 2001 From: ibelick Date: Thu, 12 Feb 2026 13:21:42 +0100 Subject: [PATCH 1/4] fix: improve chat streaming UX and live updates --- apps/webclaw/src/routeTree.gen.ts | 21 ++ apps/webclaw/src/routes/api/send.ts | 12 +- apps/webclaw/src/routes/api/stream.ts | 109 ++++++ apps/webclaw/src/screens/chat/chat-screen.tsx | 249 ++++++++++++- apps/webclaw/src/server/gateway.ts | 333 +++++++++++++++++- lefthook.yml | 2 +- 6 files changed, 705 insertions(+), 21 deletions(-) create mode 100644 apps/webclaw/src/routes/api/stream.ts diff --git a/apps/webclaw/src/routeTree.gen.ts b/apps/webclaw/src/routeTree.gen.ts index 97661c6..9fc7637 100644 --- a/apps/webclaw/src/routeTree.gen.ts +++ b/apps/webclaw/src/routeTree.gen.ts @@ -13,6 +13,7 @@ import { Route as NewRouteImport } from './routes/new' import { Route as ConnectRouteImport } from './routes/connect' import { Route as IndexRouteImport } from './routes/index' import { Route as ChatSessionKeyRouteImport } from './routes/chat/$sessionKey' +import { Route as ApiStreamRouteImport } from './routes/api/stream' import { Route as ApiSessionsRouteImport } from './routes/api/sessions' import { Route as ApiSendRouteImport } from './routes/api/send' import { Route as ApiPingRouteImport } from './routes/api/ping' @@ -39,6 +40,11 @@ const ChatSessionKeyRoute = ChatSessionKeyRouteImport.update({ path: '/chat/$sessionKey', getParentRoute: () => rootRouteImport, } as any) +const ApiStreamRoute = ApiStreamRouteImport.update({ + id: '/api/stream', + path: '/api/stream', + getParentRoute: () => rootRouteImport, +} as any) const ApiSessionsRoute = ApiSessionsRouteImport.update({ id: '/api/sessions', path: '/api/sessions', @@ -74,6 +80,7 @@ export interface FileRoutesByFullPath { '/api/ping': typeof ApiPingRoute '/api/send': typeof ApiSendRoute '/api/sessions': typeof ApiSessionsRoute + '/api/stream': typeof ApiStreamRoute '/chat/$sessionKey': typeof ChatSessionKeyRoute } export interface FileRoutesByTo { @@ -85,6 +92,7 @@ export interface FileRoutesByTo { '/api/ping': typeof ApiPingRoute '/api/send': typeof ApiSendRoute '/api/sessions': typeof ApiSessionsRoute + '/api/stream': typeof ApiStreamRoute '/chat/$sessionKey': typeof ChatSessionKeyRoute } export interface FileRoutesById { @@ -97,6 +105,7 @@ export interface FileRoutesById { '/api/ping': typeof ApiPingRoute '/api/send': typeof ApiSendRoute '/api/sessions': typeof ApiSessionsRoute + '/api/stream': typeof ApiStreamRoute '/chat/$sessionKey': typeof ChatSessionKeyRoute } export interface FileRouteTypes { @@ -110,6 +119,7 @@ export interface FileRouteTypes { | '/api/ping' | '/api/send' | '/api/sessions' + | '/api/stream' | '/chat/$sessionKey' fileRoutesByTo: FileRoutesByTo to: @@ -121,6 +131,7 @@ export interface FileRouteTypes { | '/api/ping' | '/api/send' | '/api/sessions' + | '/api/stream' | '/chat/$sessionKey' id: | '__root__' @@ -132,6 +143,7 @@ export interface FileRouteTypes { | '/api/ping' | '/api/send' | '/api/sessions' + | '/api/stream' | '/chat/$sessionKey' fileRoutesById: FileRoutesById } @@ -144,6 +156,7 @@ export interface RootRouteChildren { ApiPingRoute: typeof ApiPingRoute ApiSendRoute: typeof ApiSendRoute ApiSessionsRoute: typeof ApiSessionsRoute + ApiStreamRoute: typeof ApiStreamRoute ChatSessionKeyRoute: typeof ChatSessionKeyRoute } @@ -177,6 +190,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ChatSessionKeyRouteImport parentRoute: typeof rootRouteImport } + '/api/stream': { + id: '/api/stream' + path: '/api/stream' + fullPath: '/api/stream' + preLoaderRoute: typeof ApiStreamRouteImport + parentRoute: typeof rootRouteImport + } '/api/sessions': { id: '/api/sessions' path: '/api/sessions' @@ -224,6 +244,7 @@ const rootRouteChildren: RootRouteChildren = { ApiPingRoute: ApiPingRoute, ApiSendRoute: ApiSendRoute, ApiSessionsRoute: ApiSessionsRoute, + ApiStreamRoute: ApiStreamRoute, ChatSessionKeyRoute: ChatSessionKeyRoute, } export const routeTree = rootRouteImport diff --git a/apps/webclaw/src/routes/api/send.ts b/apps/webclaw/src/routes/api/send.ts index b05f09a..5243ba1 100644 --- a/apps/webclaw/src/routes/api/send.ts +++ b/apps/webclaw/src/routes/api/send.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto' import { createFileRoute } from '@tanstack/react-router' import { json } from '@tanstack/react-start' -import { gatewayRpc } from '../../server/gateway' +import { gatewayRpc, gatewayRpcShared } from '../../server/gateway' type SessionsResolveResponse = { ok?: boolean @@ -70,18 +70,22 @@ export const Route = createFileRoute('/api/send')({ sessionKey = 'main' } - const res = await gatewayRpc<{ runId: string }>('chat.send', { + const res = await gatewayRpcShared<{ runId: string }>( + 'chat.send', + { sessionKey, message, thinking, attachments, - deliver: false, + deliver: true, timeoutMs: 120_000, idempotencyKey: typeof body.idempotencyKey === 'string' ? body.idempotencyKey : randomUUID(), - }) + }, + sessionKey, + ) return json({ ok: true, ...res, sessionKey }) } catch (err) { diff --git a/apps/webclaw/src/routes/api/stream.ts b/apps/webclaw/src/routes/api/stream.ts new file mode 100644 index 0000000..ffba067 --- /dev/null +++ b/apps/webclaw/src/routes/api/stream.ts @@ -0,0 +1,109 @@ +import { createFileRoute } from '@tanstack/react-router' +import { acquireGatewayClient, gatewayRpcShared } from '../../server/gateway' + +type StreamEventPayload = { + event: string + payload?: unknown + seq?: number + stateVersion?: number +} + +export const Route = createFileRoute('/api/stream')({ + server: { + handlers: { + GET: ({ request }) => { + const url = new URL(request.url) + const sessionKey = url.searchParams.get('sessionKey')?.trim() || '' + const friendlyId = url.searchParams.get('friendlyId')?.trim() || '' + const encoder = new TextEncoder() + + let releaseClient: (() => void) | null = null + let closed = false + + const stream = new ReadableStream({ + start(controller) { + function send(data: StreamEventPayload) { + if (closed) return + try { + controller.enqueue( + encoder.encode(`data: ${JSON.stringify(data)}\n\n`), + ) + } catch { + closed = true + } + } + + const heartbeat = setInterval(() => { + controller.enqueue(encoder.encode('event: ping\ndata: {}\n\n')) + }, 15000) + + const key = sessionKey || friendlyId + if (key) { + void acquireGatewayClient(key, { + onEvent(event) { + send({ + event: event.event, + payload: event.payload, + seq: event.seq, + stateVersion: event.stateVersion, + }) + }, + onError(error) { + send({ event: 'error', payload: error.message }) + }, + }) + .then((handle) => { + if (closed) { + handle.release() + return + } + releaseClient = handle.release + if (sessionKey) { + void gatewayRpcShared( + 'chat.history', + { sessionKey, limit: 1 }, + sessionKey, + ) + } + }) + .catch((error: unknown) => { + if (closed) return + const message = error instanceof Error ? error.message : String(error) + send({ event: 'error', payload: message }) + }) + } + + request.signal.addEventListener( + 'abort', + () => { + if (closed) return + closed = true + clearInterval(heartbeat) + releaseClient?.() + try { + controller.close() + } catch { + return + } + }, + { once: true }, + ) + }, + cancel() { + if (closed) return + closed = true + releaseClient?.() + }, + }) + + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }, + }) + }, + }, + }, +}) diff --git a/apps/webclaw/src/screens/chat/chat-screen.tsx b/apps/webclaw/src/screens/chat/chat-screen.tsx index 6137d7c..c6bcc06 100644 --- a/apps/webclaw/src/screens/chat/chat-screen.tsx +++ b/apps/webclaw/src/screens/chat/chat-screen.tsx @@ -11,6 +11,7 @@ import { useQuery, useQueryClient } from '@tanstack/react-query' import { deriveFriendlyIdFromKey, + getMessageTimestamp, isMissingGatewayAuth, readError, textFromMessage, @@ -22,6 +23,7 @@ import { clearHistoryMessages, fetchGatewayStatus, removeHistoryMessageByClientId, + updateHistoryMessages, updateHistoryMessageByClientId, updateSessionLastMessage, } from './chat-queries' @@ -47,7 +49,7 @@ import { useChatMobile } from './hooks/use-chat-mobile' import { useChatSessions } from './hooks/use-chat-sessions' import type { AttachmentFile } from '@/components/attachment-button' import type { ChatComposerHelpers } from './components/chat-composer' -import type { HistoryResponse } from './types' +import type { GatewayMessage, HistoryResponse } from './types' import { useExport } from '@/hooks/use-export' import { cn } from '@/lib/utils' @@ -81,8 +83,15 @@ export function ChatScreen({ const [pinToTop, setPinToTop] = useState( () => hasPendingSend() || hasPendingGeneration(), ) - const streamTimer = useRef(null) const streamIdleTimer = useRef(null) + const streamRefetchInFlight = useRef(false) + const lastStreamStateVersion = useRef(null) + const lastStreamSeq = useRef(null) + const streamSourceRef = useRef(null) + const streamReconnectTimer = useRef(null) + const streamReconnectAttempt = useRef(0) + const streamFinalRefetchTimer = useRef(null) + const lastStreamFinalRunId = useRef('') const lastAssistantSignature = useRef('') const refreshHistoryRef = useRef<() => void>(() => {}) const pendingStartRef = useRef(false) @@ -157,32 +166,242 @@ export function ChatScreen({ navigate({ to: '/new', replace: true }) }, [navigate]) const streamStop = useCallback(() => { - if (streamTimer.current) { - window.clearInterval(streamTimer.current) - streamTimer.current = null - } if (streamIdleTimer.current) { window.clearTimeout(streamIdleTimer.current) streamIdleTimer.current = null } + if (streamReconnectTimer.current) { + window.clearTimeout(streamReconnectTimer.current) + streamReconnectTimer.current = null + } + if (streamFinalRefetchTimer.current) { + window.clearTimeout(streamFinalRefetchTimer.current) + streamFinalRefetchTimer.current = null + } + if (streamSourceRef.current) { + streamSourceRef.current.close() + streamSourceRef.current = null + } + streamRefetchInFlight.current = false }, []) const streamFinish = useCallback(() => { streamStop() setPendingGeneration(false) setWaitingForResponse(false) }, [streamStop]) - const streamStart = useCallback(() => { - if (!activeFriendlyId || isNewChat) return - if (streamTimer.current) window.clearInterval(streamTimer.current) - streamTimer.current = window.setInterval(() => { - refreshHistoryRef.current() - }, 350) - }, [activeFriendlyId, isNewChat]) const stableContentStyle = useMemo(() => ({}), []) refreshHistoryRef.current = function refreshHistory() { void historyQuery.refetch() } + useEffect(function setupStream() { + if (!activeFriendlyId || isNewChat || isRedirecting) return + let cancelled = false + + function startStream() { + if (cancelled) return + if (streamSourceRef.current) { + streamSourceRef.current.close() + streamSourceRef.current = null + } + const params = new URLSearchParams() + const streamSessionKey = resolvedSessionKey || sessionKeyForHistory + if (streamSessionKey) params.set('sessionKey', streamSessionKey) + if (activeFriendlyId) params.set('friendlyId', activeFriendlyId) + const source = new EventSource(`/api/stream?${params.toString()}`) + streamSourceRef.current = source + + function handleStreamEvent(event: MessageEvent) { + try { + const parsed = JSON.parse(String(event.data || '{}')) as { + event?: string + payload?: unknown + seq?: number + stateVersion?: number + } + if ( + typeof parsed.stateVersion === 'number' && + parsed.stateVersion === lastStreamStateVersion.current + ) { + return + } + if (typeof parsed.stateVersion === 'number') { + lastStreamStateVersion.current = parsed.stateVersion + } + if (typeof parsed.seq === 'number') { + if (parsed.seq === lastStreamSeq.current) return + lastStreamSeq.current = parsed.seq + } + + if (parsed.event === 'chat.history') { + const payload = parsed.payload as { messages?: Array } | null + if (payload && Array.isArray(payload.messages)) { + console.info('[stream] apply history payload', { + count: payload.messages.length, + }) + queryClient.setQueryData( + chatQueryKeys.history(activeFriendlyId, sessionKeyForHistory), + { + sessionKey: sessionKeyForHistory, + messages: payload.messages, + }, + ) + return + } + return + } + if (!parsed.event) return + if (parsed.event === 'chat') { + const payload = parsed.payload as + | { + runId?: string + sessionKey?: string + state?: string + message?: GatewayMessage + } + | null + if (payload?.message && typeof payload.message === 'object') { + const payloadSessionKey = payload.sessionKey + if ( + payloadSessionKey && + resolvedSessionKey && + payloadSessionKey !== resolvedSessionKey && + payloadSessionKey !== sessionKeyForHistory + ) { + return + } + const streamRunId = + typeof payload.runId === 'string' ? payload.runId : '' + const nextMessage = { + ...payload.message, + __streamRunId: streamRunId || undefined, + } + function upsert(messages: Array) { + if (streamRunId) { + const index = messages.findIndex( + (message) => + (message as { __streamRunId?: string }).__streamRunId === + streamRunId, + ) + if (index >= 0) { + const next = [...messages] + next[index] = nextMessage + return next + } + } + if (nextMessage.role === 'assistant') { + const nextTime = getMessageTimestamp(nextMessage) + const index = [...messages] + .reverse() + .findIndex((message) => message.role === 'assistant') + if (index >= 0) { + const target = messages.length - 1 - index + const targetTime = getMessageTimestamp(messages[target]) + if (Math.abs(nextTime - targetTime) <= 15000) { + const next = [...messages] + next[target] = nextMessage + return next + } + } + } + return [...messages, nextMessage] + } + + updateHistoryMessages( + queryClient, + activeFriendlyId, + sessionKeyForHistory, + upsert, + ) + if (payloadSessionKey && payloadSessionKey !== sessionKeyForHistory) { + updateHistoryMessages( + queryClient, + activeFriendlyId, + payloadSessionKey, + upsert, + ) + } + if (payloadSessionKey) { + updateSessionLastMessage( + queryClient, + payloadSessionKey, + activeFriendlyId, + nextMessage, + ) + } + if (payload.state === 'final') { + const nextRunId = streamRunId || 'final' + if (lastStreamFinalRunId.current !== nextRunId) { + lastStreamFinalRunId.current = nextRunId + if (streamFinalRefetchTimer.current) { + window.clearTimeout(streamFinalRefetchTimer.current) + } + streamFinalRefetchTimer.current = window.setTimeout(() => { + streamFinalRefetchTimer.current = null + refreshHistoryRef.current() + }, 350) + } + } + } + return + } + if (!parsed.event.startsWith('chat.')) { + return + } + } catch { + // ignore parse errors + } + if (streamRefetchInFlight.current) return + streamRefetchInFlight.current = true + const refetchStart = performance.now() + Promise.resolve(refreshHistoryRef.current()).finally(() => { + streamRefetchInFlight.current = false + void refetchStart + }) + } + + function handleStreamOpen() { + streamReconnectAttempt.current = 0 + console.info('[stream] open') + refreshHistoryRef.current() + } + + function handleStreamError() { + console.info('[stream] error') + if (cancelled) return + if (streamReconnectTimer.current) return + if (streamSourceRef.current) { + streamSourceRef.current.close() + streamSourceRef.current = null + } + streamReconnectAttempt.current += 1 + const backoff = Math.min(8000, 1000 * streamReconnectAttempt.current) + streamReconnectTimer.current = window.setTimeout(() => { + streamReconnectTimer.current = null + startStream() + }, backoff) + } + + source.addEventListener('message', handleStreamEvent) + source.addEventListener('open', handleStreamOpen) + source.addEventListener('error', handleStreamError) + } + + startStream() + + return () => { + cancelled = true + streamStop() + } + }, [ + activeFriendlyId, + isNewChat, + isRedirecting, + resolvedSessionKey, + sessionKeyForHistory, + streamStop, + ]) + useEffect(() => { if (isRedirecting) { if (error) setError(null) @@ -279,7 +498,7 @@ export function ChatScreen({ } streamIdleTimer.current = window.setTimeout(() => { streamFinish() - }, 4000) + }, 12000) } }, [historyMessages, streamFinish]) @@ -406,7 +625,7 @@ export function ChatScreen({ }) .then(async (res) => { if (!res.ok) throw new Error(await readError(res)) - streamStart() + refreshHistoryRef.current() }) .catch((err) => { const messageText = err instanceof Error ? err.message : String(err) diff --git a/apps/webclaw/src/server/gateway.ts b/apps/webclaw/src/server/gateway.ts index e6d6e5e..ccbd5f6 100644 --- a/apps/webclaw/src/server/gateway.ts +++ b/apps/webclaw/src/server/gateway.ts @@ -10,7 +10,13 @@ type GatewayFrame = payload?: unknown error?: { code: string; message: string; details?: unknown } } - | { type: 'event'; event: string; payload?: unknown; seq?: number } + | { + type: 'event' + event: string + payload?: unknown + seq?: number + stateVersion?: number + } type ConnectParams = { minProtocol: number @@ -33,6 +39,44 @@ type GatewayWaiter = { handleMessage: (evt: MessageEvent) => void } +type GatewayEventFrame = { + type: 'event' + event: string + payload?: unknown + seq?: number + stateVersion?: number +} + +type GatewayEventStreamOptions = { + sessionKey?: string + friendlyId?: string + signal?: AbortSignal + onEvent: (event: GatewayEventFrame) => void + onError?: (error: Error) => void +} + +type GatewayClient = { + connect: () => Promise + sendReq: (method: string, params?: unknown) => Promise + close: () => void + setOnEvent: (handler?: (event: GatewayEventFrame) => void) => void + setOnError: (handler?: (error: Error) => void) => void + isClosed: () => boolean +} + +type GatewayClientEntry = { + key: string + refs: number + client: GatewayClient +} + +type GatewayClientHandle = { + client: GatewayClient + release: () => void +} + +const sharedGatewayClients = new Map() + function getGatewayConfig() { const url = process.env.CLAWDBOT_GATEWAY_URL?.trim() || 'ws://127.0.0.1:18789' const token = process.env.CLAWDBOT_GATEWAY_TOKEN?.trim() || '' @@ -69,6 +113,293 @@ function buildConnectParams(token: string, password: string): ConnectParams { } } +async function connectGateway(ws: WebSocket): Promise { + const { token, password } = getGatewayConfig() + await wsOpen(ws) + const connectId = randomUUID() + const connectParams = buildConnectParams(token, password) + const connectReq: GatewayFrame = { + type: 'req', + id: connectId, + method: 'connect', + params: connectParams, + } + const waiter = createGatewayWaiter() + ws.addEventListener('message', waiter.handleMessage) + ws.send(JSON.stringify(connectReq)) + await waiter.waitForRes(connectId) + ws.removeEventListener('message', waiter.handleMessage) +} + +function createGatewayClient(): GatewayClient { + const { url, token, password } = getGatewayConfig() + const ws = new WebSocket(url) + let closed = false + let connected = false + let onEvent: ((event: GatewayEventFrame) => void) | undefined + let onError: ((error: Error) => void) | undefined + const waiters = new Map< + string, + { + resolve: (v: unknown) => void + reject: (e: Error) => void + } + >() + + function rejectAll(error: Error) { + for (const [, waiter] of waiters) { + waiter.reject(error) + } + waiters.clear() + } + + function handleMessage(evt: MessageEvent) { + try { + const data = typeof evt.data === 'string' ? evt.data : '' + const parsed = JSON.parse(data) as GatewayFrame + if (parsed.type === 'event') { + if (onEvent) onEvent(parsed) + return + } + if (parsed.type !== 'res') return + const waiter = waiters.get(parsed.id) + if (!waiter) return + waiters.delete(parsed.id) + if (parsed.ok) waiter.resolve(parsed.payload) + else waiter.reject(new Error(parsed.error?.message ?? 'gateway error')) + } catch { + // ignore parse errors + } + } + + function handleError(err: Event) { + if (onError) { + onError( + new Error(`Gateway client error: ${String((err as any)?.message ?? err)}`), + ) + } + } + + function handleClose() { + if (closed) return + closed = true + rejectAll(new Error('Gateway client closed')) + } + + ws.addEventListener('message', handleMessage) + ws.addEventListener('error', handleError) + ws.addEventListener('close', handleClose) + + async function connect() { + if (connected || closed) return + await wsOpen(ws) + const connectId = randomUUID() + const connectParams = buildConnectParams(token, password) + const connectReq: GatewayFrame = { + type: 'req', + id: connectId, + method: 'connect', + params: connectParams, + } + const waitForRes = new Promise((resolve, reject) => { + waiters.set(connectId, { resolve, reject }) + }) + ws.send(JSON.stringify(connectReq)) + await waitForRes + connected = true + } + + function sendReq(method: string, params?: unknown) { + if (closed) { + return Promise.reject(new Error('Gateway client closed')) + } + const id = randomUUID() + const req: GatewayFrame = { + type: 'req', + id, + method, + params, + } + const waitForRes = new Promise((resolve, reject) => { + waiters.set(id, { resolve, reject }) + }) + ws.send(JSON.stringify(req)) + return waitForRes as Promise + } + + function close() { + if (closed) return + closed = true + ws.removeEventListener('message', handleMessage) + ws.removeEventListener('error', handleError) + ws.removeEventListener('close', handleClose) + rejectAll(new Error('Gateway client closed')) + void wsClose(ws) + } + + function setOnEvent(handler?: (event: GatewayEventFrame) => void) { + onEvent = handler + } + + function setOnError(handler?: (error: Error) => void) { + onError = handler + } + + function isClosed() { + return closed + } + + return { connect, sendReq, close, setOnEvent, setOnError, isClosed } +} + +export async function acquireGatewayClient( + key: string, + options?: { + onEvent?: (event: GatewayEventFrame) => void + onError?: (error: Error) => void + }, +): Promise { + const existing = sharedGatewayClients.get(key) + if (existing && !existing.client.isClosed()) { + existing.refs += 1 + if (options?.onEvent) existing.client.setOnEvent(options.onEvent) + if (options?.onError) existing.client.setOnError(options.onError) + return { + client: existing.client, + release: function release() { + releaseGatewayClient(key) + }, + } + } + + const client = createGatewayClient() + if (options?.onEvent) client.setOnEvent(options.onEvent) + if (options?.onError) client.setOnError(options.onError) + await client.connect() + sharedGatewayClients.set(key, { key, refs: 1, client }) + return { + client, + release: function release() { + releaseGatewayClient(key) + }, + } +} + +function releaseGatewayClient(key: string) { + const entry = sharedGatewayClients.get(key) + if (!entry) return + entry.refs -= 1 + if (entry.refs > 0) return + entry.client.close() + sharedGatewayClients.delete(key) +} + +export async function gatewayRpcShared( + method: string, + params: unknown, + key?: string, +): Promise { + if (key) { + const entry = sharedGatewayClients.get(key) + if (entry && !entry.client.isClosed()) { + await entry.client.connect() + return entry.client.sendReq(method, params) + } + } + return gatewayRpc(method, params) +} + +export function gatewayEventStream({ + sessionKey, + friendlyId, + signal, + onEvent, + onError, +}: GatewayEventStreamOptions) { + const { url } = getGatewayConfig() + const ws = new WebSocket(url) + let closed = false + + function handleMessage(evt: MessageEvent) { + try { + const data = typeof evt.data === 'string' ? evt.data : '' + const parsed = JSON.parse(data) as GatewayFrame + if (parsed.type !== 'event') return + onEvent(parsed) + } catch { + // ignore parse errors + } + } + + function handleError(err: Event) { + if (onError) { + onError( + new Error(`Gateway event stream error: ${String((err as any)?.message ?? err)}`), + ) + } + } + + function handleClose() { + if (closed) return + closed = true + } + + ws.addEventListener('message', handleMessage) + ws.addEventListener('error', handleError) + ws.addEventListener('close', handleClose) + + void connectGateway(ws) + .then(async () => { + if (!sessionKey && !friendlyId) return + const subscribeReq: GatewayFrame = { + type: 'req', + id: randomUUID(), + method: 'chat.subscribe', + params: { + sessionKey: sessionKey || undefined, + friendlyId: friendlyId || undefined, + }, + } + const waiter = createGatewayWaiter() + ws.addEventListener('message', waiter.handleMessage) + try { + ws.send(JSON.stringify(subscribeReq)) + await waiter.waitForRes(subscribeReq.id) + } catch (err) { + if (onError) { + onError(err instanceof Error ? err : new Error(String(err))) + } + close() + } finally { + ws.removeEventListener('message', waiter.handleMessage) + } + }) + .catch((err) => { + if (onError) onError(err instanceof Error ? err : new Error(String(err))) + }) + + if (signal) { + signal.addEventListener( + 'abort', + () => { + close() + }, + { once: true }, + ) + } + + function close() { + if (closed) return + closed = true + ws.removeEventListener('message', handleMessage) + ws.removeEventListener('error', handleError) + ws.removeEventListener('close', handleClose) + void wsClose(ws) + } + + return close +} + function createGatewayWaiter(): GatewayWaiter { const waiters = new Map< string, diff --git a/lefthook.yml b/lefthook.yml index d8139d3..61039fe 100644 --- a/lefthook.yml +++ b/lefthook.yml @@ -3,5 +3,5 @@ pre-commit: commands: eslint-webclaw: glob: "apps/webclaw/**/*.{js,jsx,ts,tsx}" - run: "pnpm -C apps/webclaw eslint --fix {staged_files}" + run: "pnpm -C apps/webclaw exec eslint --fix ." stage_fixed: true From 991d56f6110bd5000c66f590b9904c0735b1d32b Mon Sep 17 00:00:00 2001 From: ibelick Date: Thu, 12 Feb 2026 13:25:23 +0100 Subject: [PATCH 2/4] fix: eslint errors --- .../prompt-kit/code-block/index.tsx | 4 ++-- .../components/prompt-kit/scroll-button.tsx | 1 - .../src/components/prompt-kit/thinking.tsx | 10 +++++----- .../src/components/prompt-kit/tool.tsx | 10 +++++----- apps/webclaw/src/routes/api/send.ts | 20 +++++++++---------- apps/webclaw/src/routes/api/stream.ts | 3 ++- apps/webclaw/src/screens/chat/chat-queries.ts | 4 +--- .../src/screens/chat/chat-screen-utils.ts | 4 ++-- apps/webclaw/src/screens/chat/chat-screen.tsx | 2 +- .../screens/chat/components/message-item.tsx | 12 ++++++----- .../chat/components/settings-dialog.tsx | 2 +- .../screens/chat/hooks/use-chat-history.ts | 18 ++++++++--------- .../src/screens/chat/hooks/use-chat-mobile.ts | 2 +- apps/webclaw/src/screens/chat/pending-send.ts | 2 +- .../src/screens/chat/session-tombstones.ts | 8 ++------ 15 files changed, 48 insertions(+), 54 deletions(-) diff --git a/apps/webclaw/src/components/prompt-kit/code-block/index.tsx b/apps/webclaw/src/components/prompt-kit/code-block/index.tsx index 10dab87..ac99878 100644 --- a/apps/webclaw/src/components/prompt-kit/code-block/index.tsx +++ b/apps/webclaw/src/components/prompt-kit/code-block/index.tsx @@ -3,7 +3,6 @@ import { HugeiconsIcon } from '@hugeicons/react' import { Copy01Icon, Tick02Icon } from '@hugeicons/core-free-icons' import { createHighlighterCore } from 'shiki/core' import { createJavaScriptRegexEngine } from 'shiki/engine/javascript' -import type { HighlighterCore } from 'shiki/core' import vitesseDark from '@shikijs/themes/vitesse-dark' import vitesseLight from '@shikijs/themes/vitesse-light' import langBash from '@shikijs/langs/bash' @@ -35,10 +34,11 @@ import langTypescript from '@shikijs/langs/typescript' import langTsx from '@shikijs/langs/tsx' import langXml from '@shikijs/langs/xml' import langYaml from '@shikijs/langs/yaml' +import { formatLanguageName, normalizeLanguage, resolveLanguage } from './utils' +import type { HighlighterCore } from 'shiki/core' import { useResolvedTheme } from '@/hooks/use-chat-settings' import { cn } from '@/lib/utils' import { Button } from '@/components/ui/button' -import { formatLanguageName, normalizeLanguage, resolveLanguage } from './utils' type CodeBlockProps = { content: string diff --git a/apps/webclaw/src/components/prompt-kit/scroll-button.tsx b/apps/webclaw/src/components/prompt-kit/scroll-button.tsx index 32a7ffa..a327908 100644 --- a/apps/webclaw/src/components/prompt-kit/scroll-button.tsx +++ b/apps/webclaw/src/components/prompt-kit/scroll-button.tsx @@ -46,7 +46,6 @@ function ScrollButton({ } const observer = new MutationObserver(() => { - if (!element) return if (element.scrollTop !== lastScrollTopRef.current) { lastScrollTopRef.current = element.scrollTop } diff --git a/apps/webclaw/src/components/prompt-kit/thinking.tsx b/apps/webclaw/src/components/prompt-kit/thinking.tsx index 11ae2f1..bed2122 100644 --- a/apps/webclaw/src/components/prompt-kit/thinking.tsx +++ b/apps/webclaw/src/components/prompt-kit/thinking.tsx @@ -1,12 +1,12 @@ 'use client' -import { - Collapsible, - CollapsibleTrigger, - CollapsiblePanel, -} from '@/components/ui/collapsible' import { HugeiconsIcon } from '@hugeicons/react' import { ArrowDown01Icon } from '@hugeicons/core-free-icons' +import { + Collapsible, + CollapsiblePanel, + CollapsibleTrigger, +} from '@/components/ui/collapsible' import { Button } from '@/components/ui/button' export type ThinkingProps = { diff --git a/apps/webclaw/src/components/prompt-kit/tool.tsx b/apps/webclaw/src/components/prompt-kit/tool.tsx index f2ef34d..8004917 100644 --- a/apps/webclaw/src/components/prompt-kit/tool.tsx +++ b/apps/webclaw/src/components/prompt-kit/tool.tsx @@ -1,12 +1,12 @@ 'use client' -import { - Collapsible, - CollapsibleTrigger, - CollapsiblePanel, -} from '@/components/ui/collapsible' import { HugeiconsIcon } from '@hugeicons/react' import { ArrowDown01Icon } from '@hugeicons/core-free-icons' +import { + Collapsible, + CollapsiblePanel, + CollapsibleTrigger, +} from '@/components/ui/collapsible' import { Button } from '@/components/ui/button' export type ToolPart = { diff --git a/apps/webclaw/src/routes/api/send.ts b/apps/webclaw/src/routes/api/send.ts index 5243ba1..b0b19f3 100644 --- a/apps/webclaw/src/routes/api/send.ts +++ b/apps/webclaw/src/routes/api/send.ts @@ -73,16 +73,16 @@ export const Route = createFileRoute('/api/send')({ const res = await gatewayRpcShared<{ runId: string }>( 'chat.send', { - sessionKey, - message, - thinking, - attachments, - deliver: true, - timeoutMs: 120_000, - idempotencyKey: - typeof body.idempotencyKey === 'string' - ? body.idempotencyKey - : randomUUID(), + sessionKey, + message, + thinking, + attachments, + deliver: true, + timeoutMs: 120_000, + idempotencyKey: + typeof body.idempotencyKey === 'string' + ? body.idempotencyKey + : randomUUID(), }, sessionKey, ) diff --git a/apps/webclaw/src/routes/api/stream.ts b/apps/webclaw/src/routes/api/stream.ts index ffba067..d978cd7 100644 --- a/apps/webclaw/src/routes/api/stream.ts +++ b/apps/webclaw/src/routes/api/stream.ts @@ -68,7 +68,8 @@ export const Route = createFileRoute('/api/stream')({ }) .catch((error: unknown) => { if (closed) return - const message = error instanceof Error ? error.message : String(error) + const message = + error instanceof Error ? error.message : String(error) send({ event: 'error', payload: message }) }) } diff --git a/apps/webclaw/src/screens/chat/chat-queries.ts b/apps/webclaw/src/screens/chat/chat-queries.ts index 03844d1..2d08b5c 100644 --- a/apps/webclaw/src/screens/chat/chat-queries.ts +++ b/apps/webclaw/src/screens/chat/chat-queries.ts @@ -162,9 +162,7 @@ export function moveHistoryMessages( ) { const fromKey = chatQueryKeys.history(fromFriendlyId, fromSessionKey) const toKey = chatQueryKeys.history(toFriendlyId, toSessionKey) - const fromData = queryClient.getQueryData(fromKey) as - | HistoryResponse - | undefined + const fromData = queryClient.getQueryData(fromKey) if (!fromData) return const messages = Array.isArray(fromData.messages) ? fromData.messages : [] queryClient.setQueryData(toKey, { diff --git a/apps/webclaw/src/screens/chat/chat-screen-utils.ts b/apps/webclaw/src/screens/chat/chat-screen-utils.ts index af9715e..59c7b99 100644 --- a/apps/webclaw/src/screens/chat/chat-screen-utils.ts +++ b/apps/webclaw/src/screens/chat/chat-screen-utils.ts @@ -9,7 +9,7 @@ type OptimisticMessagePayload = { export function createOptimisticMessage( body: string, - attachments?: AttachmentFile[], + attachments?: Array, ): OptimisticMessagePayload { const clientId = crypto.randomUUID() const optimisticId = `opt-${clientId}` @@ -23,7 +23,7 @@ export function createOptimisticMessage( if (attachments && attachments.length > 0) { for (const att of attachments) { - if (att.type === 'image' && att.base64) { + if (att.base64) { content.push({ type: 'image', source: { diff --git a/apps/webclaw/src/screens/chat/chat-screen.tsx b/apps/webclaw/src/screens/chat/chat-screen.tsx index c6bcc06..ff4debc 100644 --- a/apps/webclaw/src/screens/chat/chat-screen.tsx +++ b/apps/webclaw/src/screens/chat/chat-screen.tsx @@ -23,8 +23,8 @@ import { clearHistoryMessages, fetchGatewayStatus, removeHistoryMessageByClientId, - updateHistoryMessages, updateHistoryMessageByClientId, + updateHistoryMessages, updateSessionLastMessage, } from './chat-queries' import { chatUiQueryKey, getChatUiState, setChatUiState } from './chat-ui' diff --git a/apps/webclaw/src/screens/chat/components/message-item.tsx b/apps/webclaw/src/screens/chat/components/message-item.tsx index 310c18e..e1ed241 100644 --- a/apps/webclaw/src/screens/chat/components/message-item.tsx +++ b/apps/webclaw/src/screens/chat/components/message-item.tsx @@ -146,16 +146,18 @@ type ImagePart = { * @param msg - The gateway message to extract images from * @returns Array of image parts with base64 data */ -function imagesFromMessage(msg: GatewayMessage): ImagePart[] { +function imagesFromMessage(msg: GatewayMessage): Array { const parts = Array.isArray(msg.content) ? msg.content : [] - const images: ImagePart[] = [] + const images: Array = [] for (const part of parts) { + const partType = (part as { type?: string }).type + const imagePart = part as unknown as ImagePart if ( - part.type === 'image' && + partType === 'image' && 'source' in part && - typeof (part as ImagePart).source?.data === 'string' + typeof imagePart.source.data === 'string' ) { - images.push(part as ImagePart) + images.push(imagePart) } } return images diff --git a/apps/webclaw/src/screens/chat/components/settings-dialog.tsx b/apps/webclaw/src/screens/chat/components/settings-dialog.tsx index 1bc15d8..350a850 100644 --- a/apps/webclaw/src/screens/chat/components/settings-dialog.tsx +++ b/apps/webclaw/src/screens/chat/components/settings-dialog.tsx @@ -6,6 +6,7 @@ import { Sun01Icon, } from '@hugeicons/core-free-icons' import type { PathsPayload } from '../types' +import type { ThemeMode } from '@/hooks/use-chat-settings' import { DialogClose, DialogContent, @@ -16,7 +17,6 @@ import { import { Switch } from '@/components/ui/switch' import { Tabs, TabsList, TabsTab } from '@/components/ui/tabs' import { useChatSettings } from '@/hooks/use-chat-settings' -import type { ThemeMode } from '@/hooks/use-chat-settings' import { Button } from '@/components/ui/button' type SettingsSectionProps = { diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts index f89b18c..9287886 100644 --- a/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts @@ -1,8 +1,9 @@ import { useMemo, useRef } from 'react' -import { useQuery, type QueryClient } from '@tanstack/react-query' +import { useQuery } from '@tanstack/react-query' import { chatQueryKeys, fetchHistory } from '../chat-queries' import { getMessageTimestamp, textFromMessage } from '../utils' +import type { QueryClient } from '@tanstack/react-query' import type { GatewayMessage, HistoryResponse } from '../types' type UseChatHistoryInput = { @@ -35,9 +36,7 @@ export function useChatHistory({ const historyQuery = useQuery({ queryKey: historyKey, queryFn: async function fetchHistoryForSession() { - const cached = queryClient.getQueryData(historyKey) as - | HistoryResponse - | undefined + const cached = queryClient.getQueryData(historyKey) const optimisticMessages = Array.isArray(cached?.messages) ? cached.messages.filter((message) => { if (message.status === 'sending') return true @@ -79,12 +78,11 @@ export function useChatHistory({ const messages = Array.isArray(historyQuery.data?.messages) ? historyQuery.data.messages : [] - const last = messages[messages.length - 1] - const lastId = - last && typeof (last as { id?: string }).id === 'string' - ? (last as { id?: string }).id - : '' - const signature = `${messages.length}:${last?.role ?? ''}:${lastId}:${textFromMessage(last ?? { role: 'user', content: [] }).slice(-32)}` + const last = messages.at(-1) + const lastId = typeof last?.id === 'string' ? last.id : '' + const lastRole = typeof last?.role === 'string' ? last.role : '' + const lastText = last ? textFromMessage(last) : '' + const signature = `${messages.length}:${lastRole}:${lastId}:${lastText.slice(-32)}` if (signature === stableHistorySignatureRef.current) { return stableHistoryMessagesRef.current } diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-mobile.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-mobile.ts index f12df8b..d3929e3 100644 --- a/apps/webclaw/src/screens/chat/hooks/use-chat-mobile.ts +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-mobile.ts @@ -1,7 +1,7 @@ import { useLayoutEffect, useState } from 'react' +import { setChatUiState } from '../chat-ui' import type { QueryClient } from '@tanstack/react-query' -import { setChatUiState } from '../chat-ui' export function useChatMobile(queryClient: QueryClient) { const [isMobile, setIsMobile] = useState(false) diff --git a/apps/webclaw/src/screens/chat/pending-send.ts b/apps/webclaw/src/screens/chat/pending-send.ts index 7f09973..c930fa0 100644 --- a/apps/webclaw/src/screens/chat/pending-send.ts +++ b/apps/webclaw/src/screens/chat/pending-send.ts @@ -6,7 +6,7 @@ export type PendingSendPayload = { friendlyId: string message: string optimisticMessage: GatewayMessage - attachments?: AttachmentFile[] + attachments?: Array } let pendingSend: PendingSendPayload | null = null diff --git a/apps/webclaw/src/screens/chat/session-tombstones.ts b/apps/webclaw/src/screens/chat/session-tombstones.ts index d41117e..36f3a9a 100644 --- a/apps/webclaw/src/screens/chat/session-tombstones.ts +++ b/apps/webclaw/src/screens/chat/session-tombstones.ts @@ -21,7 +21,6 @@ export function filterSessionsWithTombstones< >(sessions: Array) { if (tombstones.size === 0) return sessions const now = Date.now() - let changed = false const next = sessions.filter((session) => { const keyTombstone = tombstones.get(session.key) const friendlyTombstone = tombstones.get(session.friendlyId) @@ -37,11 +36,8 @@ export function filterSessionsWithTombstones< } return true } - if (keyTombstone || friendlyTombstone) { - changed = true - return false - } + if (keyTombstone || friendlyTombstone) return false return true }) - return changed ? next : sessions + return next } From 69753174420a433d16c6b265f526cf527eb117db Mon Sep 17 00:00:00 2001 From: ibelick Date: Thu, 12 Feb 2026 14:24:17 +0100 Subject: [PATCH 3/4] fix: stabilize chat streaming and generation state --- apps/webclaw/src/screens/chat/chat-screen.tsx | 507 +++--------------- .../chat/hooks/use-chat-error-state.ts | 74 +++ .../chat/hooks/use-chat-generation-guard.ts | 76 +++ .../chat/hooks/use-chat-idle-finish.ts | 44 ++ .../chat/hooks/use-chat-pending-send.ts | 116 ++++ .../screens/chat/hooks/use-chat-redirect.ts | 62 +++ .../src/screens/chat/hooks/use-chat-stream.ts | 226 ++++++++ 7 files changed, 680 insertions(+), 425 deletions(-) create mode 100644 apps/webclaw/src/screens/chat/hooks/use-chat-error-state.ts create mode 100644 apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts create mode 100644 apps/webclaw/src/screens/chat/hooks/use-chat-idle-finish.ts create mode 100644 apps/webclaw/src/screens/chat/hooks/use-chat-pending-send.ts create mode 100644 apps/webclaw/src/screens/chat/hooks/use-chat-redirect.ts create mode 100644 apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts diff --git a/apps/webclaw/src/screens/chat/chat-screen.tsx b/apps/webclaw/src/screens/chat/chat-screen.tsx index ff4debc..f16095d 100644 --- a/apps/webclaw/src/screens/chat/chat-screen.tsx +++ b/apps/webclaw/src/screens/chat/chat-screen.tsx @@ -1,21 +1,8 @@ -import { - useCallback, - useEffect, - useLayoutEffect, - useMemo, - useRef, - useState, -} from 'react' +import { useCallback, useMemo, useRef, useState } from 'react' import { useNavigate } from '@tanstack/react-router' import { useQuery, useQueryClient } from '@tanstack/react-query' -import { - deriveFriendlyIdFromKey, - getMessageTimestamp, - isMissingGatewayAuth, - readError, - textFromMessage, -} from './utils' +import { deriveFriendlyIdFromKey, isMissingGatewayAuth, readError } from './utils' import { createOptimisticMessage } from './chat-screen-utils' import { appendHistoryMessage, @@ -24,7 +11,6 @@ import { fetchGatewayStatus, removeHistoryMessageByClientId, updateHistoryMessageByClientId, - updateHistoryMessages, updateSessionLastMessage, } from './chat-queries' import { chatUiQueryKey, getChatUiState, setChatUiState } from './chat-ui' @@ -34,11 +20,9 @@ import { ChatMessageList } from './components/chat-message-list' import { ChatComposer } from './components/chat-composer' import { GatewayStatusMessage } from './components/gateway-status-message' import { - consumePendingSend, hasPendingGeneration, hasPendingSend, isRecentSession, - resetPendingSend, setPendingGeneration, setRecentSession, stashPendingSend, @@ -47,9 +31,13 @@ import { useChatMeasurements } from './hooks/use-chat-measurements' import { useChatHistory } from './hooks/use-chat-history' import { useChatMobile } from './hooks/use-chat-mobile' import { useChatSessions } from './hooks/use-chat-sessions' +import { useChatStream } from './hooks/use-chat-stream' +import { useChatPendingSend } from './hooks/use-chat-pending-send' +import { useChatGenerationGuard } from './hooks/use-chat-generation-guard' +import { useChatErrorState } from './hooks/use-chat-error-state' +import { useChatRedirect } from './hooks/use-chat-redirect' import type { AttachmentFile } from '@/components/attachment-button' import type { ChatComposerHelpers } from './components/chat-composer' -import type { GatewayMessage, HistoryResponse } from './types' import { useExport } from '@/hooks/use-export' import { cn } from '@/lib/utils' @@ -83,18 +71,7 @@ export function ChatScreen({ const [pinToTop, setPinToTop] = useState( () => hasPendingSend() || hasPendingGeneration(), ) - const streamIdleTimer = useRef(null) - const streamRefetchInFlight = useRef(false) - const lastStreamStateVersion = useRef(null) - const lastStreamSeq = useRef(null) - const streamSourceRef = useRef(null) - const streamReconnectTimer = useRef(null) - const streamReconnectAttempt = useRef(0) - const streamFinalRefetchTimer = useRef(null) - const lastStreamFinalRunId = useRef('') - const lastAssistantSignature = useRef('') - const refreshHistoryRef = useRef<() => void>(() => {}) - const pendingStartRef = useRef(false) + const sendRefreshTimersRef = useRef>([]) const { isMobile } = useChatMobile(queryClient) const { sessionsQuery, @@ -165,282 +142,7 @@ export function ChatScreen({ setIsRedirecting(true) navigate({ to: '/new', replace: true }) }, [navigate]) - const streamStop = useCallback(() => { - if (streamIdleTimer.current) { - window.clearTimeout(streamIdleTimer.current) - streamIdleTimer.current = null - } - if (streamReconnectTimer.current) { - window.clearTimeout(streamReconnectTimer.current) - streamReconnectTimer.current = null - } - if (streamFinalRefetchTimer.current) { - window.clearTimeout(streamFinalRefetchTimer.current) - streamFinalRefetchTimer.current = null - } - if (streamSourceRef.current) { - streamSourceRef.current.close() - streamSourceRef.current = null - } - streamRefetchInFlight.current = false - }, []) - const streamFinish = useCallback(() => { - streamStop() - setPendingGeneration(false) - setWaitingForResponse(false) - }, [streamStop]) const stableContentStyle = useMemo(() => ({}), []) - refreshHistoryRef.current = function refreshHistory() { - void historyQuery.refetch() - } - - useEffect(function setupStream() { - if (!activeFriendlyId || isNewChat || isRedirecting) return - let cancelled = false - - function startStream() { - if (cancelled) return - if (streamSourceRef.current) { - streamSourceRef.current.close() - streamSourceRef.current = null - } - const params = new URLSearchParams() - const streamSessionKey = resolvedSessionKey || sessionKeyForHistory - if (streamSessionKey) params.set('sessionKey', streamSessionKey) - if (activeFriendlyId) params.set('friendlyId', activeFriendlyId) - const source = new EventSource(`/api/stream?${params.toString()}`) - streamSourceRef.current = source - - function handleStreamEvent(event: MessageEvent) { - try { - const parsed = JSON.parse(String(event.data || '{}')) as { - event?: string - payload?: unknown - seq?: number - stateVersion?: number - } - if ( - typeof parsed.stateVersion === 'number' && - parsed.stateVersion === lastStreamStateVersion.current - ) { - return - } - if (typeof parsed.stateVersion === 'number') { - lastStreamStateVersion.current = parsed.stateVersion - } - if (typeof parsed.seq === 'number') { - if (parsed.seq === lastStreamSeq.current) return - lastStreamSeq.current = parsed.seq - } - - if (parsed.event === 'chat.history') { - const payload = parsed.payload as { messages?: Array } | null - if (payload && Array.isArray(payload.messages)) { - console.info('[stream] apply history payload', { - count: payload.messages.length, - }) - queryClient.setQueryData( - chatQueryKeys.history(activeFriendlyId, sessionKeyForHistory), - { - sessionKey: sessionKeyForHistory, - messages: payload.messages, - }, - ) - return - } - return - } - if (!parsed.event) return - if (parsed.event === 'chat') { - const payload = parsed.payload as - | { - runId?: string - sessionKey?: string - state?: string - message?: GatewayMessage - } - | null - if (payload?.message && typeof payload.message === 'object') { - const payloadSessionKey = payload.sessionKey - if ( - payloadSessionKey && - resolvedSessionKey && - payloadSessionKey !== resolvedSessionKey && - payloadSessionKey !== sessionKeyForHistory - ) { - return - } - const streamRunId = - typeof payload.runId === 'string' ? payload.runId : '' - const nextMessage = { - ...payload.message, - __streamRunId: streamRunId || undefined, - } - function upsert(messages: Array) { - if (streamRunId) { - const index = messages.findIndex( - (message) => - (message as { __streamRunId?: string }).__streamRunId === - streamRunId, - ) - if (index >= 0) { - const next = [...messages] - next[index] = nextMessage - return next - } - } - if (nextMessage.role === 'assistant') { - const nextTime = getMessageTimestamp(nextMessage) - const index = [...messages] - .reverse() - .findIndex((message) => message.role === 'assistant') - if (index >= 0) { - const target = messages.length - 1 - index - const targetTime = getMessageTimestamp(messages[target]) - if (Math.abs(nextTime - targetTime) <= 15000) { - const next = [...messages] - next[target] = nextMessage - return next - } - } - } - return [...messages, nextMessage] - } - - updateHistoryMessages( - queryClient, - activeFriendlyId, - sessionKeyForHistory, - upsert, - ) - if (payloadSessionKey && payloadSessionKey !== sessionKeyForHistory) { - updateHistoryMessages( - queryClient, - activeFriendlyId, - payloadSessionKey, - upsert, - ) - } - if (payloadSessionKey) { - updateSessionLastMessage( - queryClient, - payloadSessionKey, - activeFriendlyId, - nextMessage, - ) - } - if (payload.state === 'final') { - const nextRunId = streamRunId || 'final' - if (lastStreamFinalRunId.current !== nextRunId) { - lastStreamFinalRunId.current = nextRunId - if (streamFinalRefetchTimer.current) { - window.clearTimeout(streamFinalRefetchTimer.current) - } - streamFinalRefetchTimer.current = window.setTimeout(() => { - streamFinalRefetchTimer.current = null - refreshHistoryRef.current() - }, 350) - } - } - } - return - } - if (!parsed.event.startsWith('chat.')) { - return - } - } catch { - // ignore parse errors - } - if (streamRefetchInFlight.current) return - streamRefetchInFlight.current = true - const refetchStart = performance.now() - Promise.resolve(refreshHistoryRef.current()).finally(() => { - streamRefetchInFlight.current = false - void refetchStart - }) - } - - function handleStreamOpen() { - streamReconnectAttempt.current = 0 - console.info('[stream] open') - refreshHistoryRef.current() - } - - function handleStreamError() { - console.info('[stream] error') - if (cancelled) return - if (streamReconnectTimer.current) return - if (streamSourceRef.current) { - streamSourceRef.current.close() - streamSourceRef.current = null - } - streamReconnectAttempt.current += 1 - const backoff = Math.min(8000, 1000 * streamReconnectAttempt.current) - streamReconnectTimer.current = window.setTimeout(() => { - streamReconnectTimer.current = null - startStream() - }, backoff) - } - - source.addEventListener('message', handleStreamEvent) - source.addEventListener('open', handleStreamOpen) - source.addEventListener('error', handleStreamError) - } - - startStream() - - return () => { - cancelled = true - streamStop() - } - }, [ - activeFriendlyId, - isNewChat, - isRedirecting, - resolvedSessionKey, - sessionKeyForHistory, - streamStop, - ]) - - useEffect(() => { - if (isRedirecting) { - if (error) setError(null) - return - } - if (shouldRedirectToNew) { - if (error) setError(null) - return - } - if (sessionsQuery.isSuccess && !activeExists) { - if (error) setError(null) - return - } - const messageText = sessionsError ?? historyError ?? gatewayStatusError - if (!messageText) { - if (error?.startsWith('Failed to load')) { - setError(null) - } - return - } - if (isMissingGatewayAuth(messageText)) { - navigate({ to: '/connect', replace: true }) - } - const message = sessionsError - ? `Failed to load sessions. ${sessionsError}` - : historyError - ? `Failed to load history. ${historyError}` - : gatewayStatusError - ? `Gateway unavailable. ${gatewayStatusError}` - : null - if (message) setError(message) - }, [ - error, - gatewayStatusError, - historyError, - isRedirecting, - navigate, - sessionsError, - ]) const shouldRedirectToNew = !isNewChat && @@ -452,126 +154,8 @@ export function ChatScreen({ !historyQuery.isFetching && !historyQuery.isSuccess - useEffect(() => { - if (!isRedirecting) return - if (isNewChat) { - setIsRedirecting(false) - return - } - if (!shouldRedirectToNew && sessionsQuery.isSuccess) { - setIsRedirecting(false) - } - }, [isNewChat, isRedirecting, sessionsQuery.isSuccess, shouldRedirectToNew]) - - useEffect(() => { - if (isNewChat) return - if (!sessionsQuery.isSuccess) return - if (sessions.length === 0) return - if (!shouldRedirectToNew) return - resetPendingSend() - clearHistoryMessages(queryClient, activeFriendlyId, sessionKeyForHistory) - navigate({ to: '/new', replace: true }) - }, [ - activeFriendlyId, - historyQuery.isFetching, - historyQuery.isSuccess, - isNewChat, - navigate, - queryClient, - sessionKeyForHistory, - sessions, - sessionsQuery.isSuccess, - shouldRedirectToNew, - ]) - const hideUi = shouldRedirectToNew || isRedirecting - useEffect(() => { - if (historyMessages.length === 0) return - const latestMessage = historyMessages[historyMessages.length - 1] - if (latestMessage.role !== 'assistant') return - const signature = `${historyMessages.length}:${textFromMessage(latestMessage).slice(-64)}` - if (signature !== lastAssistantSignature.current) { - lastAssistantSignature.current = signature - if (streamIdleTimer.current) { - window.clearTimeout(streamIdleTimer.current) - } - streamIdleTimer.current = window.setTimeout(() => { - streamFinish() - }, 12000) - } - }, [historyMessages, streamFinish]) - - useEffect(() => { - if (pendingStartRef.current) { - pendingStartRef.current = false - return - } - if (hasPendingSend() || hasPendingGeneration()) { - setWaitingForResponse(true) - setPinToTop(true) - return - } - streamStop() - lastAssistantSignature.current = '' - setWaitingForResponse(false) - setPinToTop(false) - }, [activeFriendlyId, isNewChat, streamStop]) - - useLayoutEffect(() => { - if (isNewChat) return - const pending = consumePendingSend( - forcedSessionKey || resolvedSessionKey || activeSessionKey, - activeFriendlyId, - ) - if (!pending) return - pendingStartRef.current = true - const historyKey = chatQueryKeys.history( - pending.friendlyId, - pending.sessionKey, - ) - const cached = queryClient.getQueryData(historyKey) - const cachedMessages = Array.isArray(cached?.messages) - ? cached.messages - : [] - const alreadyHasOptimistic = cachedMessages.some((message) => { - if (pending.optimisticMessage.clientId) { - if (message.clientId === pending.optimisticMessage.clientId) return true - if (message.__optimisticId === pending.optimisticMessage.clientId) - return true - } - if (pending.optimisticMessage.__optimisticId) { - if (message.__optimisticId === pending.optimisticMessage.__optimisticId) - return true - } - return false - }) - if (!alreadyHasOptimistic) { - appendHistoryMessage( - queryClient, - pending.friendlyId, - pending.sessionKey, - pending.optimisticMessage, - ) - } - setWaitingForResponse(true) - setPinToTop(true) - sendMessage( - pending.sessionKey, - pending.friendlyId, - pending.message, - true, - pending.attachments, - ) - }, [ - activeFriendlyId, - activeSessionKey, - forcedSessionKey, - isNewChat, - queryClient, - resolvedSessionKey, - ]) - function sendMessage( sessionKey: string, friendlyId: string, @@ -611,6 +195,19 @@ export function ChatScreen({ content: a.base64, })) + function schedulePostSendRefreshes() { + for (const timer of sendRefreshTimersRef.current) { + window.clearTimeout(timer) + } + sendRefreshTimersRef.current = [] + const delays = [0, 2000, 6000] + sendRefreshTimersRef.current = delays.map((delay) => + window.setTimeout(() => { + refreshHistory() + }, delay), + ) + } + fetch('/api/send', { method: 'POST', headers: { 'content-type': 'application/json' }, @@ -625,7 +222,7 @@ export function ChatScreen({ }) .then(async (res) => { if (!res.ok) throw new Error(await readError(res)) - refreshHistoryRef.current() + schedulePostSendRefreshes() }) .catch((err) => { const messageText = err instanceof Error ? err.message : String(err) @@ -809,6 +406,66 @@ export function ChatScreen({ ) }, [gatewayError, handleGatewayRefetch, showGatewayNotice]) + const refreshHistory = useCallback(() => { + void historyQuery.refetch() + }, [historyQuery]) + + const { stopStream } = useChatStream({ + activeFriendlyId, + isNewChat, + isRedirecting, + resolvedSessionKey, + sessionKeyForHistory, + queryClient, + refreshHistory, + }) + + useChatErrorState({ + error, + setError, + isRedirecting, + shouldRedirectToNew, + sessionsReady: sessionsQuery.isSuccess, + activeExists, + sessionsError, + historyError, + gatewayStatusError, + }) + + useChatRedirect({ + activeFriendlyId, + isNewChat, + isRedirecting, + shouldRedirectToNew, + sessionsReady: sessionsQuery.isSuccess, + sessionsCount: sessions.length, + sessionKeyForHistory, + queryClient, + setIsRedirecting, + }) + + useChatGenerationGuard({ + waitingForResponse, + historyMessages, + streamStop: stopStream, + refreshHistory, + setWaitingForResponse, + setPinToTop, + }) + + useChatPendingSend({ + activeFriendlyId, + activeSessionKey, + forcedSessionKey, + isNewChat, + queryClient, + resolvedSessionKey, + setWaitingForResponse, + setPinToTop, + streamStop: stopStream, + sendMessage, + }) + const sidebar = ( void + isRedirecting: boolean + shouldRedirectToNew: boolean + sessionsReady: boolean + activeExists: boolean + sessionsError: string | null + historyError: string | null + gatewayStatusError: string | null +} + +export function useChatErrorState({ + error, + setError, + isRedirecting, + shouldRedirectToNew, + sessionsReady, + activeExists, + sessionsError, + historyError, + gatewayStatusError, +}: UseChatErrorStateInput) { + const navigate = useNavigate() + + useEffect(() => { + if (isRedirecting) { + if (error) setError(null) + return + } + if (shouldRedirectToNew) { + if (error) setError(null) + return + } + if (sessionsReady && !activeExists) { + if (error) setError(null) + return + } + const messageText = sessionsError ?? historyError ?? gatewayStatusError + if (!messageText) { + if (error?.startsWith('Failed to load')) { + setError(null) + } + return + } + if (isMissingGatewayAuth(messageText)) { + navigate({ to: '/connect', replace: true }) + } + const message = sessionsError + ? `Failed to load sessions. ${sessionsError}` + : historyError + ? `Failed to load history. ${historyError}` + : gatewayStatusError + ? `Gateway unavailable. ${gatewayStatusError}` + : null + if (message) setError(message) + }, [ + activeExists, + error, + gatewayStatusError, + historyError, + isRedirecting, + navigate, + sessionsError, + sessionsReady, + setError, + shouldRedirectToNew, + ]) +} diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts new file mode 100644 index 0000000..e876d58 --- /dev/null +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts @@ -0,0 +1,76 @@ +import { useEffect, useRef } from 'react' + +import { getMessageTimestamp } from '../utils' +import { setPendingGeneration } from '../pending-send' +import type { GatewayMessage } from '../types' + +type UseChatGenerationGuardInput = { + waitingForResponse: boolean + historyMessages: Array + streamStop: () => void + refreshHistory: () => void + setWaitingForResponse: (value: boolean) => void + setPinToTop: (value: boolean) => void +} + +export function useChatGenerationGuard({ + waitingForResponse, + historyMessages, + streamStop, + refreshHistory, + setWaitingForResponse, + setPinToTop, +}: UseChatGenerationGuardInput) { + const timeoutTimer = useRef(null) + const waitingRef = useRef(waitingForResponse) + + function finish() { + streamStop() + setPendingGeneration(false) + setWaitingForResponse(false) + } + + useEffect(() => { + waitingRef.current = waitingForResponse + }, [waitingForResponse]) + + useEffect(() => { + if (!waitingForResponse) { + if (timeoutTimer.current) { + window.clearTimeout(timeoutTimer.current) + timeoutTimer.current = null + } + return + } + + const lastAssistant = [...historyMessages] + .reverse() + .find((message) => message.role === 'assistant') + const lastUser = [...historyMessages] + .reverse() + .find((message) => message.role === 'user') + const assistantTime = lastAssistant ? getMessageTimestamp(lastAssistant) : 0 + const userTime = lastUser ? getMessageTimestamp(lastUser) : 0 + if (assistantTime > userTime) { + finish() + return + } + + if (!timeoutTimer.current) { + timeoutTimer.current = window.setTimeout(() => { + timeoutTimer.current = null + if (!waitingRef.current) return + refreshHistory() + finish() + }, 30000) + } + + }, [ + historyMessages, + refreshHistory, + setPinToTop, + setWaitingForResponse, + streamStop, + waitingForResponse, + ]) +} diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-idle-finish.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-idle-finish.ts new file mode 100644 index 0000000..7c6f7df --- /dev/null +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-idle-finish.ts @@ -0,0 +1,44 @@ +import { useEffect, useRef } from 'react' + +import { textFromMessage } from '../utils' +import { setPendingGeneration } from '../pending-send' +import type { GatewayMessage } from '../types' + +type UseChatIdleFinishInput = { + historyMessages: Array + streamStop: () => void + setWaitingForResponse: (value: boolean) => void +} + +export function useChatIdleFinish({ + historyMessages, + streamStop, + setWaitingForResponse, +}: UseChatIdleFinishInput) { + const lastAssistantSignature = useRef('') + const streamIdleTimer = useRef(null) + + useEffect(() => { + if (historyMessages.length === 0) return + const latestMessage = historyMessages[historyMessages.length - 1] + if (latestMessage.role !== 'assistant') return + const signature = `${historyMessages.length}:${textFromMessage(latestMessage).slice(-64)}` + if (signature !== lastAssistantSignature.current) { + lastAssistantSignature.current = signature + if (streamIdleTimer.current) { + window.clearTimeout(streamIdleTimer.current) + } + streamIdleTimer.current = window.setTimeout(() => { + streamStop() + setPendingGeneration(false) + setWaitingForResponse(false) + }, 12000) + } + return () => { + if (streamIdleTimer.current) { + window.clearTimeout(streamIdleTimer.current) + streamIdleTimer.current = null + } + } + }, [historyMessages, setWaitingForResponse, streamStop]) +} diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-pending-send.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-pending-send.ts new file mode 100644 index 0000000..97ab279 --- /dev/null +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-pending-send.ts @@ -0,0 +1,116 @@ +import { useEffect, useLayoutEffect, useRef } from 'react' +import { appendHistoryMessage, chatQueryKeys } from '../chat-queries' +import { + consumePendingSend, + hasPendingGeneration, + hasPendingSend, +} from '../pending-send' +import type { QueryClient } from '@tanstack/react-query' + +import type { AttachmentFile } from '@/components/attachment-button' +import type { HistoryResponse } from '../types' + +type UseChatPendingSendInput = { + activeFriendlyId: string + activeSessionKey: string + forcedSessionKey?: string + isNewChat: boolean + queryClient: QueryClient + resolvedSessionKey: string + setWaitingForResponse: (value: boolean) => void + setPinToTop: (value: boolean) => void + streamStop: () => void + sendMessage: ( + sessionKey: string, + friendlyId: string, + body: string, + skipOptimistic: boolean, + attachments?: Array, + ) => void +} + +export function useChatPendingSend({ + activeFriendlyId, + activeSessionKey, + forcedSessionKey, + isNewChat, + queryClient, + resolvedSessionKey, + setWaitingForResponse, + setPinToTop, + streamStop, + sendMessage, +}: UseChatPendingSendInput) { + const pendingStartRef = useRef(false) + + useEffect(() => { + if (pendingStartRef.current) { + pendingStartRef.current = false + return + } + if (hasPendingSend() || hasPendingGeneration()) { + setWaitingForResponse(true) + setPinToTop(true) + return + } + streamStop() + setWaitingForResponse(false) + }, [activeFriendlyId, isNewChat, setPinToTop, setWaitingForResponse, streamStop]) + + useLayoutEffect(() => { + if (isNewChat) return + const pending = consumePendingSend( + forcedSessionKey || resolvedSessionKey || activeSessionKey, + activeFriendlyId, + ) + if (!pending) return + pendingStartRef.current = true + const historyKey = chatQueryKeys.history( + pending.friendlyId, + pending.sessionKey, + ) + const cached = queryClient.getQueryData(historyKey) + const cachedMessages = Array.isArray(cached?.messages) + ? cached.messages + : [] + const alreadyHasOptimistic = cachedMessages.some((message) => { + if (pending.optimisticMessage.clientId) { + if (message.clientId === pending.optimisticMessage.clientId) return true + if (message.__optimisticId === pending.optimisticMessage.clientId) + return true + } + if (pending.optimisticMessage.__optimisticId) { + if (message.__optimisticId === pending.optimisticMessage.__optimisticId) + return true + } + return false + }) + if (!alreadyHasOptimistic) { + appendHistoryMessage( + queryClient, + pending.friendlyId, + pending.sessionKey, + pending.optimisticMessage, + ) + } + setWaitingForResponse(true) + setPinToTop(true) + sendMessage( + pending.sessionKey, + pending.friendlyId, + pending.message, + true, + pending.attachments, + ) + }, [ + activeFriendlyId, + activeSessionKey, + forcedSessionKey, + isNewChat, + queryClient, + resolvedSessionKey, + sendMessage, + setPinToTop, + setWaitingForResponse, + ]) +} diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-redirect.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-redirect.ts new file mode 100644 index 0000000..dd7d198 --- /dev/null +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-redirect.ts @@ -0,0 +1,62 @@ +import { useEffect } from 'react' +import { useNavigate } from '@tanstack/react-router' + +import { clearHistoryMessages } from '../chat-queries' +import { resetPendingSend } from '../pending-send' +import type { QueryClient } from '@tanstack/react-query' + +type UseChatRedirectInput = { + activeFriendlyId: string + isNewChat: boolean + isRedirecting: boolean + shouldRedirectToNew: boolean + sessionsReady: boolean + sessionsCount: number + sessionKeyForHistory: string + queryClient: QueryClient + setIsRedirecting: (value: boolean) => void +} + +export function useChatRedirect({ + activeFriendlyId, + isNewChat, + isRedirecting, + shouldRedirectToNew, + sessionsReady, + sessionsCount, + sessionKeyForHistory, + queryClient, + setIsRedirecting, +}: UseChatRedirectInput) { + const navigate = useNavigate() + + useEffect(() => { + if (!isRedirecting) return + if (isNewChat) { + setIsRedirecting(false) + return + } + if (!shouldRedirectToNew && sessionsReady) { + setIsRedirecting(false) + } + }, [isNewChat, isRedirecting, sessionsReady, setIsRedirecting, shouldRedirectToNew]) + + useEffect(() => { + if (isNewChat) return + if (!sessionsReady) return + if (sessionsCount === 0) return + if (!shouldRedirectToNew) return + resetPendingSend() + clearHistoryMessages(queryClient, activeFriendlyId, sessionKeyForHistory) + navigate({ to: '/new', replace: true }) + }, [ + activeFriendlyId, + isNewChat, + navigate, + queryClient, + sessionKeyForHistory, + sessionsCount, + sessionsReady, + shouldRedirectToNew, + ]) +} diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts new file mode 100644 index 0000000..4bc08c6 --- /dev/null +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts @@ -0,0 +1,226 @@ +import { useCallback, useEffect, useRef } from 'react' + +import { getMessageTimestamp } from '../utils' +import { + chatQueryKeys, + updateHistoryMessages, + updateSessionLastMessage, +} from '../chat-queries' +import type { QueryClient } from '@tanstack/react-query' +import type { GatewayMessage } from '../types' + +type UseChatStreamInput = { + activeFriendlyId: string + isNewChat: boolean + isRedirecting: boolean + resolvedSessionKey: string + sessionKeyForHistory: string + queryClient: QueryClient + refreshHistory: () => void +} + +export function useChatStream({ + activeFriendlyId, + isNewChat, + isRedirecting, + resolvedSessionKey, + sessionKeyForHistory, + queryClient, + refreshHistory, +}: UseChatStreamInput) { + const streamSourceRef = useRef(null) + const streamReconnectTimer = useRef(null) + const streamReconnectAttempt = useRef(0) + const refreshHistoryRef = useRef(refreshHistory) + + useEffect(() => { + refreshHistoryRef.current = refreshHistory + }, [refreshHistory]) + + const stopStream = useCallback(() => { + if (streamReconnectTimer.current) { + window.clearTimeout(streamReconnectTimer.current) + streamReconnectTimer.current = null + } + if (streamSourceRef.current) { + streamSourceRef.current.close() + streamSourceRef.current = null + } + }, []) + + useEffect(() => { + if (!activeFriendlyId || isNewChat || isRedirecting) return + let cancelled = false + + function startStream() { + if (cancelled) return + if (streamSourceRef.current) { + streamSourceRef.current.close() + streamSourceRef.current = null + } + const params = new URLSearchParams() + const streamSessionKey = resolvedSessionKey || sessionKeyForHistory + if (streamSessionKey) params.set('sessionKey', streamSessionKey) + if (activeFriendlyId) params.set('friendlyId', activeFriendlyId) + const source = new EventSource(`/api/stream?${params.toString()}`) + streamSourceRef.current = source + + function handleStreamEvent(event: MessageEvent) { + try { + const parsed = JSON.parse(String(event.data || '{}')) as { + event?: string + payload?: unknown + } + if (parsed.event === 'chat.history') { + const payload = parsed.payload as { messages?: Array } | null + if (payload && Array.isArray(payload.messages)) { + queryClient.setQueryData( + chatQueryKeys.history(activeFriendlyId, sessionKeyForHistory), + { + sessionKey: sessionKeyForHistory, + messages: payload.messages, + }, + ) + return + } + return + } + if (!parsed.event) return + if (parsed.event === 'chat') { + const payload = parsed.payload as + | { + runId?: string + sessionKey?: string + state?: string + message?: GatewayMessage + } + | null + if (payload?.message && typeof payload.message === 'object') { + const payloadSessionKey = payload.sessionKey + if ( + payloadSessionKey && + resolvedSessionKey && + payloadSessionKey !== resolvedSessionKey && + payloadSessionKey !== sessionKeyForHistory + ) { + return + } + const streamRunId = + typeof payload.runId === 'string' ? payload.runId : '' + const nextMessage = { + ...payload.message, + __streamRunId: streamRunId || undefined, + } + function upsert(messages: Array) { + if (streamRunId) { + const index = messages.findIndex( + (message) => + (message as { __streamRunId?: string }).__streamRunId === + streamRunId, + ) + if (index >= 0) { + const next = [...messages] + next[index] = nextMessage + return next + } + } + if (nextMessage.role === 'assistant') { + const nextTime = getMessageTimestamp(nextMessage) + const index = [...messages] + .reverse() + .findIndex((message) => message.role === 'assistant') + if (index >= 0) { + const target = messages.length - 1 - index + const targetTime = getMessageTimestamp(messages[target]) + if (Math.abs(nextTime - targetTime) <= 15000) { + const next = [...messages] + next[target] = nextMessage + return next + } + } + } + return [...messages, nextMessage] + } + + updateHistoryMessages( + queryClient, + activeFriendlyId, + sessionKeyForHistory, + upsert, + ) + if (payloadSessionKey && payloadSessionKey !== sessionKeyForHistory) { + updateHistoryMessages( + queryClient, + activeFriendlyId, + payloadSessionKey, + upsert, + ) + } + if (payloadSessionKey) { + updateSessionLastMessage( + queryClient, + payloadSessionKey, + activeFriendlyId, + nextMessage, + ) + } + if ( + payload.state === 'final' || + payload.state === 'error' || + payload.state === 'aborted' + ) { + refreshHistoryRef.current() + } + } + return + } + if (!parsed.event.startsWith('chat.')) { + return + } + } catch { + // ignore parse errors + } + } + + function handleStreamOpen() { + streamReconnectAttempt.current = 0 + refreshHistoryRef.current() + } + + function handleStreamError() { + if (cancelled) return + if (streamReconnectTimer.current) return + if (streamSourceRef.current) { + streamSourceRef.current.close() + streamSourceRef.current = null + } + streamReconnectAttempt.current += 1 + const backoff = Math.min(8000, 1000 * streamReconnectAttempt.current) + streamReconnectTimer.current = window.setTimeout(() => { + streamReconnectTimer.current = null + startStream() + }, backoff) + } + + source.addEventListener('message', handleStreamEvent) + source.addEventListener('open', handleStreamOpen) + source.addEventListener('error', handleStreamError) + } + + startStream() + + return () => { + cancelled = true + stopStream() + } + }, [ + activeFriendlyId, + isNewChat, + isRedirecting, + resolvedSessionKey, + sessionKeyForHistory, + stopStream, + ]) + + return { stopStream } +} From 65baa7ac3ac9e260b14643378a26285c3857bf84 Mon Sep 17 00:00:00 2001 From: ibelick Date: Thu, 12 Feb 2026 14:55:24 +0100 Subject: [PATCH 4/4] fix: make chat streaming run-aware --- apps/webclaw/src/screens/chat/chat-screen.tsx | 133 ++++++++++++++---- .../chat/hooks/use-chat-generation-guard.ts | 27 +--- .../screens/chat/hooks/use-chat-history.ts | 65 +++++++-- .../src/screens/chat/hooks/use-chat-stream.ts | 87 ++++++++++-- 4 files changed, 238 insertions(+), 74 deletions(-) diff --git a/apps/webclaw/src/screens/chat/chat-screen.tsx b/apps/webclaw/src/screens/chat/chat-screen.tsx index f16095d..68097bb 100644 --- a/apps/webclaw/src/screens/chat/chat-screen.tsx +++ b/apps/webclaw/src/screens/chat/chat-screen.tsx @@ -1,8 +1,12 @@ -import { useCallback, useMemo, useRef, useState } from 'react' +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import { useNavigate } from '@tanstack/react-router' import { useQuery, useQueryClient } from '@tanstack/react-query' -import { deriveFriendlyIdFromKey, isMissingGatewayAuth, readError } from './utils' +import { + deriveFriendlyIdFromKey, + isMissingGatewayAuth, + readError, +} from './utils' import { createOptimisticMessage } from './chat-screen-utils' import { appendHistoryMessage, @@ -71,7 +75,8 @@ export function ChatScreen({ const [pinToTop, setPinToTop] = useState( () => hasPendingSend() || hasPendingGeneration(), ) - const sendRefreshTimersRef = useRef>([]) + const pendingRunIdsRef = useRef(new Set()) + const pendingRunTimersRef = useRef(new Map()) const { isMobile } = useChatMobile(queryClient) const { sessionsQuery, @@ -84,7 +89,6 @@ export function ChatScreen({ } = useChatSessions({ activeFriendlyId, isNewChat, forcedSessionKey }) const { historyQuery, - historyMessages, displayMessages, historyError, resolvedSessionKey, @@ -154,8 +158,69 @@ export function ChatScreen({ !historyQuery.isFetching && !historyQuery.isSuccess + const refreshHistory = useCallback(() => { + void historyQuery.refetch() + }, [historyQuery]) + const hideUi = shouldRedirectToNew || isRedirecting + const finishRun = useCallback( + (runId: string) => { + if (!runId) return + const timer = pendingRunTimersRef.current.get(runId) + if (typeof timer === 'number') { + window.clearTimeout(timer) + } + pendingRunTimersRef.current.delete(runId) + pendingRunIdsRef.current.delete(runId) + if (pendingRunIdsRef.current.size === 0) { + setPendingGeneration(false) + setWaitingForResponse(false) + } + }, + [setWaitingForResponse], + ) + + const startRun = useCallback( + (runId: string) => { + if (!runId) return + pendingRunIdsRef.current.add(runId) + const existingTimer = pendingRunTimersRef.current.get(runId) + if (typeof existingTimer === 'number') { + window.clearTimeout(existingTimer) + } + const timeout = window.setTimeout(() => { + pendingRunTimersRef.current.delete(runId) + pendingRunIdsRef.current.delete(runId) + refreshHistory() + if (pendingRunIdsRef.current.size === 0) { + setPendingGeneration(false) + setWaitingForResponse(false) + } + }, 120000) + pendingRunTimersRef.current.set(runId, timeout) + setPendingGeneration(true) + setWaitingForResponse(true) + }, + [refreshHistory], + ) + + const finishAllRuns = useCallback(() => { + for (const [, timer] of pendingRunTimersRef.current) { + window.clearTimeout(timer) + } + pendingRunTimersRef.current.clear() + pendingRunIdsRef.current.clear() + setPendingGeneration(false) + setWaitingForResponse(false) + }, []) + + useEffect(() => { + return () => { + finishAllRuns() + } + }, [finishAllRuns]) + function sendMessage( sessionKey: string, friendlyId: string, @@ -195,19 +260,6 @@ export function ChatScreen({ content: a.base64, })) - function schedulePostSendRefreshes() { - for (const timer of sendRefreshTimersRef.current) { - window.clearTimeout(timer) - } - sendRefreshTimersRef.current = [] - const delays = [0, 2000, 6000] - sendRefreshTimersRef.current = delays.map((delay) => - window.setTimeout(() => { - refreshHistory() - }, delay), - ) - } - fetch('/api/send', { method: 'POST', headers: { 'content-type': 'application/json' }, @@ -222,7 +274,16 @@ export function ChatScreen({ }) .then(async (res) => { if (!res.ok) throw new Error(await readError(res)) - schedulePostSendRefreshes() + const payload = (await res.json().catch(() => ({}))) as { + runId?: string + } + if ( + typeof payload.runId === 'string' && + payload.runId.trim().length > 0 + ) { + startRun(payload.runId.trim()) + } + refreshHistory() }) .catch((err) => { const messageText = err instanceof Error ? err.message : String(err) @@ -406,10 +467,6 @@ export function ChatScreen({ ) }, [gatewayError, handleGatewayRefetch, showGatewayNotice]) - const refreshHistory = useCallback(() => { - void historyQuery.refetch() - }, [historyQuery]) - const { stopStream } = useChatStream({ activeFriendlyId, isNewChat, @@ -418,6 +475,35 @@ export function ChatScreen({ sessionKeyForHistory, queryClient, refreshHistory, + onChatEvent(payload) { + const payloadSessionKey = + typeof payload.sessionKey === 'string' ? payload.sessionKey : '' + if ( + payloadSessionKey && + resolvedSessionKey && + payloadSessionKey !== resolvedSessionKey && + payloadSessionKey !== sessionKeyForHistory + ) { + return + } + const runId = typeof payload.runId === 'string' ? payload.runId : '' + const state = typeof payload.state === 'string' ? payload.state : '' + if (runId && state === 'delta') { + startRun(runId) + } + if ( + runId && + (state === 'final' || state === 'error' || state === 'aborted') + ) { + finishRun(runId) + } + if ( + !runId && + (state === 'final' || state === 'error' || state === 'aborted') + ) { + finishAllRuns() + } + }, }) useChatErrorState({ @@ -446,11 +532,8 @@ export function ChatScreen({ useChatGenerationGuard({ waitingForResponse, - historyMessages, - streamStop: stopStream, refreshHistory, setWaitingForResponse, - setPinToTop, }) useChatPendingSend({ diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts index e876d58..8f3075c 100644 --- a/apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-generation-guard.ts @@ -1,31 +1,22 @@ import { useEffect, useRef } from 'react' -import { getMessageTimestamp } from '../utils' import { setPendingGeneration } from '../pending-send' -import type { GatewayMessage } from '../types' type UseChatGenerationGuardInput = { waitingForResponse: boolean - historyMessages: Array - streamStop: () => void refreshHistory: () => void setWaitingForResponse: (value: boolean) => void - setPinToTop: (value: boolean) => void } export function useChatGenerationGuard({ waitingForResponse, - historyMessages, - streamStop, refreshHistory, setWaitingForResponse, - setPinToTop, }: UseChatGenerationGuardInput) { const timeoutTimer = useRef(null) const waitingRef = useRef(waitingForResponse) function finish() { - streamStop() setPendingGeneration(false) setWaitingForResponse(false) } @@ -43,34 +34,18 @@ export function useChatGenerationGuard({ return } - const lastAssistant = [...historyMessages] - .reverse() - .find((message) => message.role === 'assistant') - const lastUser = [...historyMessages] - .reverse() - .find((message) => message.role === 'user') - const assistantTime = lastAssistant ? getMessageTimestamp(lastAssistant) : 0 - const userTime = lastUser ? getMessageTimestamp(lastUser) : 0 - if (assistantTime > userTime) { - finish() - return - } - if (!timeoutTimer.current) { timeoutTimer.current = window.setTimeout(() => { timeoutTimer.current = null if (!waitingRef.current) return refreshHistory() finish() - }, 30000) + }, 120000) } }, [ - historyMessages, refreshHistory, - setPinToTop, setWaitingForResponse, - streamStop, waitingForResponse, ]) } diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts index 9287886..465d0c9 100644 --- a/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-history.ts @@ -37,24 +37,35 @@ export function useChatHistory({ queryKey: historyKey, queryFn: async function fetchHistoryForSession() { const cached = queryClient.getQueryData(historyKey) - const optimisticMessages = Array.isArray(cached?.messages) - ? cached.messages.filter((message) => { - if (message.status === 'sending') return true - if (message.__optimisticId) return true - return Boolean(message.clientId) - }) + const cachedMessages = Array.isArray(cached?.messages) + ? cached.messages : [] + const optimisticMessages = cachedMessages.filter((message) => { + if (message.status === 'sending') return true + if (message.__optimisticId) return true + return Boolean(message.clientId) + }) + const streamingMessages = cachedMessages.filter((message) => { + const runId = (message as { __streamRunId?: unknown }).__streamRunId + return typeof runId === 'string' && runId.trim().length > 0 + }) const serverData = await fetchHistory({ sessionKey: sessionKeyForHistory, friendlyId: activeFriendlyId, }) - if (!optimisticMessages.length) return serverData + if (!optimisticMessages.length && !streamingMessages.length) { + return serverData + } - const merged = mergeOptimisticHistoryMessages( + const mergedWithOptimistic = mergeOptimisticHistoryMessages( serverData.messages, optimisticMessages, ) + const merged = mergeStreamingHistoryMessages( + mergedWithOptimistic, + streamingMessages, + ) return { ...serverData, @@ -114,6 +125,44 @@ export function useChatHistory({ } } +function mergeStreamingHistoryMessages( + serverMessages: Array, + streamingMessages: Array, +): Array { + if (!streamingMessages.length) return serverMessages + + const merged = [...serverMessages] + for (const streamingMessage of streamingMessages) { + const runId = (streamingMessage as { __streamRunId?: unknown }).__streamRunId + if (typeof runId !== 'string' || runId.trim().length === 0) continue + + const hasMatch = merged.some((serverMessage) => { + const serverRunId = (serverMessage as { __streamRunId?: unknown }) + .__streamRunId + if ( + typeof serverRunId === 'string' && + serverRunId.trim().length > 0 && + serverRunId === runId + ) { + return true + } + if (serverMessage.role !== streamingMessage.role) return false + const streamingText = textFromMessage(streamingMessage) + if (!streamingText) return false + if (streamingText !== textFromMessage(serverMessage)) return false + const streamingTime = getMessageTimestamp(streamingMessage) + const serverTime = getMessageTimestamp(serverMessage) + return Math.abs(streamingTime - serverTime) <= 15000 + }) + + if (!hasMatch) { + merged.push(streamingMessage) + } + } + + return merged +} + function mergeOptimisticHistoryMessages( serverMessages: Array, optimisticMessages: Array, diff --git a/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts b/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts index 4bc08c6..06606e9 100644 --- a/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts +++ b/apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts @@ -1,6 +1,6 @@ import { useCallback, useEffect, useRef } from 'react' -import { getMessageTimestamp } from '../utils' +import { getMessageTimestamp, textFromMessage } from '../utils' import { chatQueryKeys, updateHistoryMessages, @@ -17,6 +17,12 @@ type UseChatStreamInput = { sessionKeyForHistory: string queryClient: QueryClient refreshHistory: () => void + onChatEvent?: (payload: { + runId?: string + sessionKey?: string + state?: string + message?: GatewayMessage + }) => void } export function useChatStream({ @@ -27,10 +33,12 @@ export function useChatStream({ sessionKeyForHistory, queryClient, refreshHistory, + onChatEvent, }: UseChatStreamInput) { const streamSourceRef = useRef(null) const streamReconnectTimer = useRef(null) const streamReconnectAttempt = useRef(0) + const streamRunTextRef = useRef(new Map()) const refreshHistoryRef = useRef(refreshHistory) useEffect(() => { @@ -46,6 +54,7 @@ export function useChatStream({ streamSourceRef.current.close() streamSourceRef.current = null } + streamRunTextRef.current.clear() }, []) useEffect(() => { @@ -95,6 +104,9 @@ export function useChatStream({ message?: GatewayMessage } | null + if (payload) { + onChatEvent?.(payload) + } if (payload?.message && typeof payload.message === 'object') { const payloadSessionKey = payload.sessionKey if ( @@ -107,11 +119,53 @@ export function useChatStream({ } const streamRunId = typeof payload.runId === 'string' ? payload.runId : '' - const nextMessage = { + const state = typeof payload.state === 'string' ? payload.state : '' + let nextMessage: GatewayMessage = { ...payload.message, __streamRunId: streamRunId || undefined, } + + if (streamRunId && state === 'delta') { + const deltaText = textFromMessage(nextMessage) + const previousText = streamRunTextRef.current.get(streamRunId) ?? '' + const cumulativeText = `${previousText}${deltaText}` + if (cumulativeText.length > 0) { + streamRunTextRef.current.set(streamRunId, cumulativeText) + nextMessage = { + ...nextMessage, + content: [{ type: 'text', text: cumulativeText }], + } + } + } + + if (streamRunId && state === 'final') { + const finalText = textFromMessage(nextMessage) + if (!finalText) { + const bufferedText = streamRunTextRef.current.get(streamRunId) + if (bufferedText) { + nextMessage = { + ...nextMessage, + content: [{ type: 'text', text: bufferedText }], + } + } + } + streamRunTextRef.current.delete(streamRunId) + } + + if ( + streamRunId && + (state === 'error' || state === 'aborted') + ) { + streamRunTextRef.current.delete(streamRunId) + } + function upsert(messages: Array) { + const lastUserIndex = [...messages] + .reverse() + .findIndex((message) => message.role === 'user') + const resolvedLastUserIndex = + lastUserIndex >= 0 ? messages.length - 1 - lastUserIndex : -1 + if (streamRunId) { const index = messages.findIndex( (message) => @@ -119,9 +173,12 @@ export function useChatStream({ streamRunId, ) if (index >= 0) { - const next = [...messages] - next[index] = nextMessage - return next + if (index > resolvedLastUserIndex) { + const next = [...messages] + next[index] = nextMessage + return next + } + return [...messages, nextMessage] } } if (nextMessage.role === 'assistant') { @@ -131,10 +188,17 @@ export function useChatStream({ .findIndex((message) => message.role === 'assistant') if (index >= 0) { const target = messages.length - 1 - index - const targetTime = getMessageTimestamp(messages[target]) - if (Math.abs(nextTime - targetTime) <= 15000) { + if (target > resolvedLastUserIndex) { + const targetTime = getMessageTimestamp(messages[target]) + if (Math.abs(nextTime - targetTime) <= 15000) { + const next = [...messages] + next[target] = nextMessage + return next + } + } + if (resolvedLastUserIndex >= 0 && target <= resolvedLastUserIndex) { const next = [...messages] - next[target] = nextMessage + next.push(nextMessage) return next } } @@ -164,13 +228,6 @@ export function useChatStream({ nextMessage, ) } - if ( - payload.state === 'final' || - payload.state === 'error' || - payload.state === 'aborted' - ) { - refreshHistoryRef.current() - } } return }