From 763302017f7538438b043fec4ba5148a99ef3716 Mon Sep 17 00:00:00 2001 From: Jihad Irfansyah Date: Sat, 1 Aug 2026 09:22:29 +0700 Subject: [PATCH] Prevent chat renderer hangs on long sessions --- Makefile | 6 +- internal/server/handlers_chat.go | 21 ++++--- internal/server/livechat.go | 28 ++++++++- internal/server/livechat_test.go | 55 +++++++++++++--- web/package.json | 1 + web/src/lib/chatStreamQueue.test.mjs | 47 ++++++++++++++ web/src/lib/chatStreamQueue.ts | 48 ++++++++++++++ web/src/pages/ChatPage.tsx | 93 +++++++++++++++++++--------- 8 files changed, 250 insertions(+), 49 deletions(-) create mode 100644 web/src/lib/chatStreamQueue.test.mjs create mode 100644 web/src/lib/chatStreamQueue.ts diff --git a/Makefile b/Makefile index 6f734ed5..1d2ea9fb 100644 --- a/Makefile +++ b/Makefile @@ -94,6 +94,10 @@ vet: ## Run go vet typecheck: ## Typecheck the frontend @cd web && $(BUN) x tsc -b --noEmit +.PHONY: web-test +web-test: ## Run frontend regression tests + @cd web && $(BUN) test + .PHONY: smoke smoke: build ## Build, serve, and load every dashboard route in a real browser @ANTARES_PORT=8799 ANTARES_HOME=$$(mktemp -d) ./bin/antares serve >/tmp/antares-smoke.log 2>&1 & \ @@ -103,7 +107,7 @@ smoke: build ## Build, serve, and load every dashboard route in a real browser SMOKE_BASE=http://127.0.0.1:8799 $(BUN) scripts/smoke.mjs .PHONY: check -check: vet test typecheck ## Run every check (add `make smoke` for the browser pass) +check: vet test typecheck web-test ## Run every check (add `make smoke` for the browser pass) .PHONY: fmt fmt: ## Format Go sources diff --git a/internal/server/handlers_chat.go b/internal/server/handlers_chat.go index aaff517b..9b8086ab 100644 --- a/internal/server/handlers_chat.go +++ b/internal/server/handlers_chat.go @@ -159,12 +159,12 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { }() agentReq := agent.Request{ - SessionID: req.SessionID, - Message: req.Message, - Images: decodeImages(req.Images), - Role: req.Role, - Platform: "web", - UserID: req.UserID, + SessionID: req.SessionID, + Message: req.Message, + Images: decodeImages(req.Images), + Role: req.Role, + Platform: "web", + UserID: req.UserID, Model: req.Model, Toolset: req.Toolset, ReasoningEffort: req.ReasoningEffort, @@ -219,7 +219,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { }() // Follow the run until it finishes or this client disconnects. - _ = lr.follow(ctx, 0, func(e agent.Event) error { return sse.send(e) }) + _ = lr.follow(ctx, 0, func(e agent.Event, _ int) error { return sse.send(e) }) } // handleChatAttach reconnects a client to a turn already in flight for a session @@ -260,7 +260,12 @@ func (s *Server) handleChatAttach(w http.ResponseWriter, r *http.Request) { } }() - _ = lr.follow(ctx, cursor, func(e agent.Event) error { return sse.send(e) }) + _ = lr.follow(ctx, cursor, func(e agent.Event, nextCursor int) error { + return sse.send(struct { + agent.Event + Cursor int `json:"cursor"` + }{Event: e, Cursor: nextCursor}) + }) } // decodeImages accepts data URLs and bare base64 payloads. diff --git a/internal/server/livechat.go b/internal/server/livechat.go index 7280df6e..7ff65961 100644 --- a/internal/server/livechat.go +++ b/internal/server/livechat.go @@ -2,6 +2,7 @@ package server import ( "context" + "strings" "sync" "github.com/enowdev/antares/internal/agent" @@ -22,7 +23,7 @@ const maxLiveEvents = 4000 type liveRun struct { mu sync.Mutex events []agent.Event - base int // absolute index of events[0] (count of events already trimmed) + base int // absolute index of events[0] (count of events already trimmed) done bool updated chan struct{} // closed on every change; replaced under the lock } @@ -61,7 +62,7 @@ func (lr *liveRun) finish() { // follow replays events from cursor, then blocks for new ones until the run // finishes or ctx is cancelled (the client disconnected). send stops the follow // early by returning an error. -func (lr *liveRun) follow(ctx context.Context, cursor int, send func(agent.Event) error) error { +func (lr *liveRun) follow(ctx context.Context, cursor int, send func(agent.Event, int) error) error { i := cursor // absolute event index for { lr.mu.Lock() @@ -73,8 +74,29 @@ func (lr *liveRun) follow(ctx context.Context, cursor int, send func(agent.Event for i-lr.base < len(lr.events) { e := lr.events[i-lr.base] i++ + // A reconnect may have thousands of token-sized deltas waiting. Collapse + // adjacent text/reasoning events while the backlog is already available; + // the returned cursor still advances past every original event. + if e.Type == agent.EventText || e.Type == agent.EventReasoning { + var merged strings.Builder + for i-lr.base < len(lr.events) { + next := lr.events[i-lr.base] + if next.Type != e.Type { + break + } + if merged.Len() == 0 { + merged.Grow(len(e.Delta) + len(next.Delta)) + merged.WriteString(e.Delta) + } + merged.WriteString(next.Delta) + i++ + } + if merged.Len() > 0 { + e.Delta = merged.String() + } + } lr.mu.Unlock() - if err := send(e); err != nil { + if err := send(e, i); err != nil { return err } lr.mu.Lock() diff --git a/internal/server/livechat_test.go b/internal/server/livechat_test.go index 8dd842a8..8c41a6ca 100644 --- a/internal/server/livechat_test.go +++ b/internal/server/livechat_test.go @@ -16,19 +16,16 @@ func TestLiveRun_ReplayThenFollow(t *testing.T) { // A follower joining late must first replay the backlog, then see new events. got := make(chan string, 8) go func() { - _ = lr.follow(context.Background(), 0, func(e agent.Event) error { + _ = lr.follow(context.Background(), 0, func(e agent.Event, _ int) error { got <- e.Delta return nil }) close(got) }() - // Backlog. - if v := <-got; v != "a" { - t.Fatalf("want a, got %q", v) - } - if v := <-got; v != "b" { - t.Fatalf("want b, got %q", v) + // Adjacent backlog deltas are coalesced into one frame. + if v := <-got; v != "ab" { + t.Fatalf("want ab, got %q", v) } // Live. lr.publish(agent.Event{Type: agent.EventText, Delta: "c"}) @@ -48,6 +45,46 @@ func TestLiveRun_ReplayThenFollow(t *testing.T) { } } +func TestLiveRun_CoalescesBacklogAndReportsAbsoluteCursor(t *testing.T) { + lr := newLiveRun() + for i := 0; i < 4000; i++ { + lr.publish(agent.Event{Type: agent.EventReasoning, Delta: "x"}) + } + lr.publish(agent.Event{Type: agent.EventUsage, InputTokens: 10}) + lr.finish() + + var frames []agent.Event + var cursors []int + if err := lr.follow(context.Background(), 0, func(e agent.Event, cursor int) error { + frames = append(frames, e) + cursors = append(cursors, cursor) + return nil + }); err != nil { + t.Fatal(err) + } + if len(frames) != 2 { + t.Fatalf("4001 backlog events produced %d frames, want 2", len(frames)) + } + if got := len(frames[0].Delta); got != 3999 { + t.Fatalf("coalesced retained reasoning length = %d, want 3999", got) + } + if cursors[0] != 4000 || cursors[1] != 4001 { + t.Fatalf("absolute cursors = %v, want [4000 4001]", cursors) + } + + // Reattaching at the reported cursor must not replay the reasoning backlog. + var replayed []agent.Event + if err := lr.follow(context.Background(), cursors[0], func(e agent.Event, _ int) error { + replayed = append(replayed, e) + return nil + }); err != nil { + t.Fatal(err) + } + if len(replayed) != 1 || replayed[0].Type != agent.EventUsage { + t.Fatalf("cursor replay = %#v, want only usage event", replayed) + } +} + func TestLiveRun_FollowFromCursor(t *testing.T) { lr := newLiveRun() lr.publish(agent.Event{Type: agent.EventText, Delta: "x"}) @@ -55,7 +92,7 @@ func TestLiveRun_FollowFromCursor(t *testing.T) { lr.finish() var seen []string - _ = lr.follow(context.Background(), 1, func(e agent.Event) error { + _ = lr.follow(context.Background(), 1, func(e agent.Event, _ int) error { seen = append(seen, e.Delta) return nil }) @@ -69,7 +106,7 @@ func TestLiveRun_FollowStopsOnContextCancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { - done <- lr.follow(ctx, 0, func(agent.Event) error { return nil }) + done <- lr.follow(ctx, 0, func(agent.Event, int) error { return nil }) }() cancel() select { diff --git a/web/package.json b/web/package.json index 4784f2a4..9775464f 100644 --- a/web/package.json +++ b/web/package.json @@ -6,6 +6,7 @@ "scripts": { "dev": "vite", "build": "tsc -b && vite build", + "test": "bun test", "preview": "vite preview", "typecheck": "tsc -b --noEmit" }, diff --git a/web/src/lib/chatStreamQueue.test.mjs b/web/src/lib/chatStreamQueue.test.mjs new file mode 100644 index 00000000..d907908a --- /dev/null +++ b/web/src/lib/chatStreamQueue.test.mjs @@ -0,0 +1,47 @@ +import { describe, expect, test } from 'bun:test' +import { + groupStreamPatches, + queueStreamDelta, + shouldRefreshAfterAttach, +} from './chatStreamQueue.ts' + +describe('stream patch batching', () => { + test('coalesces thousands of adjacent replay deltas', () => { + const queue = [] + for (let i = 0; i < 4000; i++) queueStreamDelta(queue, 'assistant', 'reasoning', 'x') + + expect(queue).toHaveLength(1) + expect(queue[0]).toEqual({ + id: 'assistant', + kind: 'delta', + segment: 'reasoning', + delta: 'x'.repeat(4000), + }) + }) + + test('preserves boundaries between messages, segments, and apply patches', () => { + const queue = [] + queueStreamDelta(queue, 'a', 'text', 'one') + queueStreamDelta(queue, 'a', 'reasoning', 'two') + queueStreamDelta(queue, 'b', 'text', 'three') + queue.push({ id: 'a', kind: 'apply', fn: (message) => ({ text: message.text + '!' }) }) + queueStreamDelta(queue, 'a', 'text', 'four') + + expect(queue).toHaveLength(5) + const grouped = groupStreamPatches(queue) + expect(grouped.get('a')).toHaveLength(4) + expect(grouped.get('b')).toHaveLength(1) + }) +}) + +describe('attach hydration policy', () => { + test('refreshes only the first idle attachment', () => { + expect(shouldRefreshAfterAttach(false, false)).toBe(true) + expect(shouldRefreshAfterAttach(false, true)).toBe(false) + }) + + test('always refreshes after a live stream emitted events', () => { + expect(shouldRefreshAfterAttach(true, false)).toBe(true) + expect(shouldRefreshAfterAttach(true, true)).toBe(true) + }) +}) diff --git a/web/src/lib/chatStreamQueue.ts b/web/src/lib/chatStreamQueue.ts new file mode 100644 index 00000000..23305c8f --- /dev/null +++ b/web/src/lib/chatStreamQueue.ts @@ -0,0 +1,48 @@ +export type StreamSegment = 'text' | 'reasoning' + +export type QueuedStreamPatch = + | { id: string; kind: 'apply'; fn: (message: T) => T } + | { id: string; kind: 'delta'; segment: StreamSegment; delta: string } + +/** + * Add a text/reasoning delta while collapsing an adjacent run of compatible + * deltas. A reconnect can replay thousands of token-sized events synchronously; + * keeping those as individual closures makes the browser concatenate an + * ever-growing string thousands of times before it can paint. + */ +export function queueStreamDelta( + queue: QueuedStreamPatch[], + id: string, + segment: StreamSegment, + delta: string, +): void { + if (!delta) return + const last = queue[queue.length - 1] + if (last?.kind === 'delta' && last.id === id && last.segment === segment) { + last.delta += delta + return + } + queue.push({ id, kind: 'delta', segment, delta }) +} + +/** Group patches once before walking the transcript. */ +export function groupStreamPatches( + queue: QueuedStreamPatch[], +): Map[]> { + const grouped = new Map[]>() + for (const patch of queue) { + const patches = grouped.get(patch.id) + if (patches) patches.push(patch) + else grouped.set(patch.id, [patch]) + } + return grouped +} + +/** + * Hydrate after a completed live stream, and once after the first idle attach to + * close the initial-load race. Repeated idle polls must not rebuild the whole + * transcript every time. + */ +export function shouldRefreshAfterAttach(hadEvents: boolean, idleRefreshDone: boolean): boolean { + return hadEvents || !idleRefreshDone +} diff --git a/web/src/pages/ChatPage.tsx b/web/src/pages/ChatPage.tsx index fe915bba..ab5cd58b 100644 --- a/web/src/pages/ChatPage.tsx +++ b/web/src/pages/ChatPage.tsx @@ -18,6 +18,12 @@ import { X, } from '@phosphor-icons/react' import { ApiError, get, post, streamGet, streamPost, type StreamEvent } from '@/lib/api' +import { + groupStreamPatches, + queueStreamDelta, + shouldRefreshAfterAttach, + type QueuedStreamPatch, +} from '@/lib/chatStreamQueue' import { copyText } from '@/lib/clipboard' import { useI18n, useTimeAgo, type MessageKey } from '@/lib/i18n' import { cn } from '@/lib/utils' @@ -377,30 +383,47 @@ export default function ChatPage() { const virtuosoRef = useRef(null) const textareaRef = useRef(null) - // A streaming turn emits hundreds of tiny events (text deltas, tool - // progress). Applying each one with its own setMessages triggers a full - // O(n) array map per event and saturates the main thread on long tasks — - // the dashboard visibly freezes (issue #5). Instead we queue the per-event - // patches and drain them once per animation frame, so a burst of N events - // costs a single render that folds all N patches into the assistant message. - const patchQueue = useRef<{ id: string; fn: (m: ChatMessage) => ChatMessage }[]>([]) + // A streaming turn emits hundreds of tiny events. Queue them for one render, + // grouping by message first so flushing is O(messages + patches), not + // O(messages * patches). Consecutive text/reasoning deltas are coalesced too: + // replaying a long live-run must not create thousands of closures and perform + // thousands of ever-growing string concatenations on the browser main thread. + const patchQueue = useRef[]>([]) const flushHandle = useRef(null) const flushPatches = useCallback(() => { flushHandle.current = null const queued = patchQueue.current if (queued.length === 0) return patchQueue.current = [] + const byMessage = groupStreamPatches(queued) setMessages((prev) => - prev.map((m) => { - let next = m - for (const p of queued) if (p.id === m.id) next = p.fn(next) + prev.map((message) => { + const patches = byMessage.get(message.id) + if (!patches) return message + let next = message + for (const patch of patches) { + next = + patch.kind === 'delta' + ? appendSeg(next, patch.segment, patch.delta) + : patch.fn(next) + } return next }), ) }, []) const enqueuePatch = useCallback( (id: string, fn: (m: ChatMessage) => ChatMessage) => { - patchQueue.current.push({ id, fn }) + patchQueue.current.push({ id, kind: 'apply', fn }) + if (flushHandle.current == null) { + flushHandle.current = requestAnimationFrame(flushPatches) + } + }, + [flushPatches], + ) + const enqueueDelta = useCallback( + (id: string, segment: 'text' | 'reasoning', delta: string) => { + if (!delta) return + queueStreamDelta(patchQueue.current, id, segment, delta) if (flushHandle.current == null) { flushHandle.current = requestAnimationFrame(flushPatches) } @@ -486,10 +509,10 @@ export default function ChatPage() { ) break case 'text': - patchAssistant((m) => appendSeg(m, 'text', String(event.delta ?? ''))) + enqueueDelta(assistantId, 'text', String(event.delta ?? '')) break case 'reasoning': - patchAssistant((m) => appendSeg(m, 'reasoning', String(event.delta ?? ''))) + enqueueDelta(assistantId, 'reasoning', String(event.delta ?? '')) break case 'tool_call': setLive((s) => ({ turn: s.turn + 1, tool: String(event.name ?? '') })) @@ -570,7 +593,7 @@ export default function ChatPage() { break } }, - [t, enqueuePatch], + [t, enqueuePatch, enqueueDelta], ) // Reconnect to a turn still running for this session (after navigating away @@ -595,6 +618,11 @@ export default function ChatPage() { let alive = true let close: (() => void) | undefined let assistantId: string | null = null + let cursor = 0 + // The initial hydrate can race a turn finishing just before attach. Refresh + // once after the first idle `done` to close that race, then stop downloading + // and rebuilding the entire transcript on every 1.5-second idle poll. + let idleRefreshDone = false const connect = () => { if (!alive) return @@ -605,7 +633,6 @@ export default function ChatPage() { window.setTimeout(connect, 1500) return } - assistantId = null const ensure = () => { if (assistantId) return assistantId assistantId = `live_${Date.now()}_a` @@ -618,24 +645,34 @@ export default function ChatPage() { return assistantId } close = streamGet( - `/chat/attach?session_id=${encodeURIComponent(sid)}&cursor=0`, + `/chat/attach?session_id=${encodeURIComponent(sid)}&cursor=${cursor}`, (event) => { + const eventCursor = Number(event.cursor ?? cursor) + if (Number.isFinite(eventCursor) && eventCursor >= cursor) cursor = eventCursor if (event.type === 'done') { drainPatches() setStreaming(false) close?.() // stop EventSource from auto-reconnecting - // Swap whatever we have for the canonical persisted turn. - get(`/sessions/${sid}`) - .then((d) => { - setMessages(hydrate(d)) - setTitle(d.session.title || t('chat.conversation')) - }) - .catch(() => {}) - // Reconnect shortly so a server-initiated wake turn is not missed. - // The attach returns 'done' at once when nothing is live, so this - // is a light poll of "is a new turn streaming?" — one open SSE, not - // a request loop. - if (alive) window.setTimeout(connect, 1500) + const shouldRefresh = shouldRefreshAfterAttach(assistantId !== null, idleRefreshDone) + assistantId = null + idleRefreshDone = true + // A later server-initiated turn is a new liveRun with its own + // zero-based cursor. Reset only after the current run is done. + cursor = 0 + const refresh = shouldRefresh + ? get(`/sessions/${sid}`) + .then((d) => { + if (!alive) return + setMessages(hydrate(d)) + setTitle(d.session.title || t('chat.conversation')) + }) + .catch(() => {}) + : Promise.resolve() + // Do not overlap canonical hydration with the next attachment: a + // stale response could otherwise overwrite fresh live deltas. + void refresh.finally(() => { + if (alive) window.setTimeout(connect, 1500) + }) return } applyEvent(ensure(), event, (_id, evtTitle) => {