Prevent chat renderer hangs on long sessions

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

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.

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()

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,14 +45,54 @@ 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"})
 	lr.publish(agent.Event{Type: agent.EventText, Delta: "y"})
 	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 {

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"
   },

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)
+  })
+})

web/src/lib/chatStreamQueue.ts

@@ -0,0 +1,48 @@
+export type StreamSegment = 'text' | 'reasoning'
+
+export type QueuedStreamPatch<T> =
+  | { 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<T>(
+  queue: QueuedStreamPatch<T>[],
+  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<T>(
+  queue: QueuedStreamPatch<T>[],
+): Map<string, QueuedStreamPatch<T>[]> {
+  const grouped = new Map<string, QueuedStreamPatch<T>[]>()
+  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
+}