perf(chat): batch live thinking rendering and bound timer updates (#5931)

* perf(chat): batch live thinking DOM updates

* test(chat): cover live thinking scheduler lifecycle

* fix(chat): guard background stop-state, restore live thinking text, drop source-text tests

- _closeOpenThinkingMarkup no longer overwrites currentAccumulated for
  backgrounded streams. It now mirrors the guard the delta path already uses
  (`if (!_isBg) currentAccumulated = accumulated`). Without it a backgrounded
  stream's text is written into the foreground session's stop-state, which
  abortCurrentRequest and detachCurrentStream then put in the wrong bubble.

- Split _extractLiveThinkingText into _liveThinkingText (strip every think tag)
  and _closedThinkingText (via extractThinkingBlocks). Slicing from the first
  <think> to the first </think> pinned the live box to "The" for the rest of the
  stream on the `<think>The</think>` + untagged-thinking pattern that the
  hasUnclosedThink detection deliberately keeps streaming through.

- The background transition now flushes with rich:true, so a stream that
  backgrounds mid-thinking isn't left as pre-wrap plain text permanently.

- Move the throttle to static/js/liveThinkingThrottle.js and import it. The
  .mjs suite imports the module instead of slicing it out of chat.js with
  vm.runInNewContext and marker comments.

- Replace the source-text assertions in tests/test_live_thinking_scheduler_js.py
  with behavioral coverage, per tests/TESTING_STANDARD.md. The .mjs suite grows
  from 3 to 6 cases.

- Collapse the duplicated tool_start/agent_step finalizers into one
  _endLiveThinkingSection().

* fix(chat): hoist thinking teardown out of the try block so catch can reach it

In an ES module a function declared inside `try { }` is scoped to that block,
and `catch` is a sibling scope rather than a nested one. _closeOpenThinkingMarkup
was declared inside the try and called from catch, so the call threw
ReferenceError and killed the rest of the error path: the stream never
finalized and the thinking block was never torn down.

Declare _closeOpenThinkingMarkup and a new _endThinkingOnTerminalPath next to
the existing _flushLiveThinking / _cancelLiveThinkingWork outer lets and assign
them inside the try, which is the pattern those two already use for exactly
this reason.

Verified against a live stream in a browser: before, clicking stop mid-thinking
logged "_closeOpenThinkingMarkup is not defined" and left no finalized thinking
section; after, the block collapses to "View thinking process" correctly.

* perf(chat): extract live thinking at commit cadence

* fix(chat): bound live thinking work

* test(chat): update stream invariant assertions

---------

Co-authored-by: Léo <leograndcontact@gmail.com>

tests/live_thinking_scheduler.test.mjs

@@ -0,0 +1,277 @@
+// Tests for the live-thinking throttle that bounds DOM work during long
+// reasoning streams (see static/js/liveThinkingThrottle.js).
+//
+// The throttle's contract is what the terminal paths in chat.js lean on:
+// a burst of deltas becomes ONE commit carrying the latest text; flush()
+// lands trailing text synchronously and cannot double-commit; cancel()
+// guarantees nothing lands after a stream is finished or backgrounded.
+//
+// Timers are injected, so this runs with no DOM and no real clock.
+import assert from 'node:assert/strict';
+import test from 'node:test';
+
+import {
+  createIncrementalDisplayProjector,
+  createLiveThinkingThrottle,
+  createThinkingAnalysisGate,
+  stripLiveThinkingTags,
+} from '../static/js/liveThinkingThrottle.js';
+
+function fakeTimers() {
+  let nextId = 1;
+  const callbacks = new Map();
+  const delays = [];
+  return {
+    schedule(callback, delay) {
+      const id = nextId++;
+      callbacks.set(id, callback);
+      delays.push(delay);
+      return id;
+    },
+    cancel(id) {
+      callbacks.delete(id);
+    },
+    run(id) {
+      const callback = callbacks.get(id);
+      assert.ok(callback, `missing timer ${id}`);
+      callbacks.delete(id);
+      callback();
+    },
+    pendingIds() {
+      return [...callbacks.keys()];
+    },
+    delays,
+  };
+}
+
+test('coalesces a burst and commits only the latest text after 100 ms', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), timers);
+
+  throttle.update('a');
+  throttle.update('ab');
+  throttle.update('abc');
+
+  assert.deepEqual(commits, []);
+  assert.deepEqual(timers.delays, [100], 'a burst must schedule exactly one commit');
+  const [timer] = timers.pendingIds();
+  timers.run(timer);
+  assert.deepEqual(commits, ['abc']);
+});
+
+test('commit count stays flat as the stream grows', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), timers);
+
+  // 500 deltas arriving inside one window is the regression this guards:
+  // the old code committed once per delta, so work grew with stream length.
+  let text = '';
+  for (let i = 0; i < 500; i++) {
+    text += 'token ';
+    throttle.update(text);
+  }
+  assert.deepEqual(commits, []);
+  assert.equal(timers.pendingIds().length, 1);
+  timers.run(timers.pendingIds()[0]);
+  assert.equal(commits.length, 1);
+  assert.equal(commits[0], text);
+});
+
+test('prepares a 200K cumulative stream only at scheduled commit cadence', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  let prepareCalls = 0;
+  let scannedCharacters = 0;
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), {
+    ...timers,
+    prepare(value) {
+      prepareCalls += 1;
+      scannedCharacters += value.length;
+      return stripLiveThinkingTags(value);
+    },
+  });
+
+  const delta = 'reasoning '.repeat(10); // 100 characters
+  let cumulative = '';
+  for (let i = 0; i < 2000; i++) {
+    cumulative += delta;
+    throttle.update(cumulative);
+  }
+
+  assert.equal(cumulative.length, 200_000);
+  assert.equal(prepareCalls, 0, 'cumulative extraction must not run per delta');
+  assert.equal(timers.pendingIds().length, 1);
+  timers.run(timers.pendingIds()[0]);
+  assert.equal(prepareCalls, 1);
+  assert.equal(scannedCharacters, 200_000);
+  assert.deepEqual(commits, [cumulative]);
+});
+
+test('ordinary answers and reasoning deltas do not request cumulative analysis', () => {
+  const startsReasoning = (text) => /^\s*thinking(?:\s+process)?\s*:/i.test(text);
+  const ordinaryGate = createThinkingAnalysisGate({ startsWithReasoningPrefix: startsReasoning });
+  let ordinary = '';
+  let ordinaryAnalyses = 0;
+  for (let i = 0; i < 2000; i++) {
+    ordinary += i === 0 ? 'Here is the answer. ' : 'answer '.repeat(10);
+    if (ordinaryGate.shouldAnalyze(ordinary)) ordinaryAnalyses += 1;
+  }
+  assert.equal(ordinaryAnalyses, 0);
+
+  const thinkingGate = createThinkingAnalysisGate({ startsWithReasoningPrefix: startsReasoning });
+  let thinking = 'Thin';
+  assert.equal(thinkingGate.shouldAnalyze(thinking), false);
+  thinking += 'king: inspect the problem';
+  assert.equal(thinkingGate.shouldAnalyze(thinking), true);
+  for (let i = 0; i < 2000; i++) {
+    thinking += ' reasoning'.repeat(10);
+    assert.equal(thinkingGate.shouldAnalyze(thinking, { isThinking: true, nonTagThinking: true }), false);
+  }
+  thinking += '\n\nHere is the answer';
+  assert.equal(thinkingGate.shouldAnalyze(thinking, { isThinking: true, nonTagThinking: true }), true);
+
+  const whitespaceGate = createThinkingAnalysisGate({ startsWithReasoningPrefix: startsReasoning });
+  let whitespaceThinking = ' '.repeat(250);
+  assert.equal(whitespaceGate.shouldAnalyze(whitespaceThinking), false);
+  whitespaceThinking += 'Thinking: bounded probe';
+  assert.equal(whitespaceGate.shouldAnalyze(whitespaceThinking), true);
+});
+
+test('split namespaced closes and false-close deadlines request analysis', () => {
+  let clock = 100;
+  const gate = createThinkingAnalysisGate({ now: () => clock });
+  let text = '<mm:think>x</mm:';
+  assert.equal(gate.shouldAnalyze(text, { isThinking: true }), true, 'fresh opening tag is analyzed');
+  text += 'think>answer';
+  assert.equal(gate.shouldAnalyze(text, { isThinking: true }), true, 'split namespaced close is analyzed');
+
+  text += ' still waiting';
+  assert.equal(gate.shouldAnalyze(text, { isThinking: true, recheckAt: 500 }), false);
+  clock = 500;
+  text += ' next delta';
+  assert.equal(gate.shouldAnalyze(text, { isThinking: true, recheckAt: 500 }), true);
+
+  const attributedGate = createThinkingAnalysisGate();
+  let attributed = `<think data-provider="${'x'.repeat(400)}"`;
+  assert.equal(attributedGate.shouldAnalyze(attributed), false);
+  attributed += '>reasoning';
+  assert.equal(attributedGate.shouldAnalyze(attributed), true, 'bounded carry preserves split tag attributes');
+});
+
+test('display projection is append-only and filters a structured tail once', () => {
+  let filterCalls = 0;
+  let filteredCharacters = 0;
+  const projector = createIncrementalDisplayProjector((text) => {
+    filterCalls += 1;
+    filteredCharacters += text.length;
+    return text.replace(/\[TOOL_CALL\][\s\S]*$/i, '');
+  });
+
+  let text = '';
+  for (let i = 0; i < 2000; i++) {
+    const delta = i === 0 ? 'Here is the answer. ' : 'ordinary text ';
+    text += delta;
+    assert.equal(projector.append(delta, text), text);
+  }
+  assert.equal(filterCalls, 0, 'ordinary deltas never run the cumulative filter');
+
+  text += '[TOOL_';
+  projector.append('[TOOL_', text);
+  text += 'CALL]{"name":"read"}';
+  const beforeToolPayload = projector.append('CALL]{"name":"read"}', text);
+  for (let i = 0; i < 2000; i++) {
+    const delta = 'payload ';
+    text += delta;
+    assert.equal(projector.append(delta, text), beforeToolPayload);
+  }
+  assert.equal(filterCalls, 1, 'structured payload filtering happens only at its boundary');
+  assert.ok(filteredCharacters < text.length, 'filter work is bounded by the first structured boundary');
+});
+
+test('literal escaped tags survive and malformed live tags retain trailing text', () => {
+  assert.equal(
+    stripLiveThinkingTags('&lt;think&gt;literal&lt;/think&gt;'),
+    '&lt;think&gt;literal&lt;/think&gt;',
+  );
+  assert.equal(
+    stripLiveThinkingTags('<think>first</think> middle <thinking mode="deep">trailing'),
+    'first middle trailing',
+  );
+  assert.equal(stripLiveThinkingTags('answer with 2 < 3 and 5 > 4'), 'answer with 2 < 3 and 5 > 4');
+});
+
+test('terminal flush prepares and commits the complete trailing cumulative text', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), {
+    ...timers,
+    prepare: stripLiveThinkingTags,
+  });
+
+  throttle.update('<think>reasoning without a closing tag');
+  assert.equal(throttle.flush(), true);
+  assert.deepEqual(commits, ['reasoning without a closing tag']);
+  assert.deepEqual(timers.pendingIds(), []);
+});
+
+test('independent throttles cannot commit cancelled text into another session', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const first = createLiveThinkingThrottle((value) => commits.push(['first', value]), timers);
+  const second = createLiveThinkingThrottle((value) => commits.push(['second', value]), timers);
+
+  first.update('stale first-session text');
+  second.update('current second-session text');
+  first.cancel();
+  assert.equal(second.flush(), true);
+
+  assert.deepEqual(timers.pendingIds(), []);
+  assert.deepEqual(commits, [['second', 'current second-session text']]);
+});
+
+test('flush synchronously preserves trailing text and cancels the pending callback', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), timers);
+
+  throttle.update('trailing text');
+  assert.equal(throttle.flush(), true);
+  assert.deepEqual(commits, ['trailing text']);
+  assert.deepEqual(timers.pendingIds(), []);
+  assert.equal(throttle.flush(), false, 'clean flush must not duplicate the commit');
+});
+
+test('cancel discards pending work without a late DOM commit', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), timers);
+
+  throttle.update('stale session text');
+  throttle.cancel();
+  assert.deepEqual(timers.pendingIds(), []);
+  assert.deepEqual(commits, []);
+});
+
+test('a cancelled throttle accepts new work again', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), timers);
+
+  throttle.update('discarded');
+  throttle.cancel();
+  throttle.update('fresh');
+  assert.equal(throttle.flush(), true);
+  assert.deepEqual(commits, ['fresh']);
+});
+
+test('coerces nullish updates instead of committing undefined', () => {
+  const timers = fakeTimers();
+  const commits = [];
+  const throttle = createLiveThinkingThrottle((value) => commits.push(value), timers);
+
+  throttle.update(null);
+  throttle.flush();
+  assert.deepEqual(commits, ['']);
+});

tests/test_chat_stream_scope.py

@@ -1,3 +1,4 @@
+import re
 from pathlib import Path
 
 
@@ -13,7 +14,7 @@ def test_stream_render_helpers_are_visible_to_catch_block():
     assert "let _cancelThinkingTimer = () => {};" in outer_scope
     assert "let _removeThinkingSpinner = () => {};" in outer_scope
 
-    assert "_renderStream = () => {" in try_body
+    assert re.search(r"(?m)^\s*_renderStream\s*=", try_body)
     assert "_cancelThinkingTimer = () => {" in try_body
     assert "_removeThinkingSpinner = () => {" in try_body
     assert "function _renderStream()" not in try_body

tests/test_live_thinking_chat_integration.py

@@ -0,0 +1,145 @@
+"""Source-level wiring guards for live-thinking stream lifecycle.
+
+The pure scheduler suite covers timing behavior. These assertions pin the
+browser-only integration seams that are impractical to import without the full
+application DOM.
+"""
+
+from pathlib import Path
+
+
+_CHAT = (Path(__file__).resolve().parent.parent / "static" / "js" / "chat.js").read_text(
+    encoding="utf-8"
+)
+
+
+def _between(start: str, end: str) -> str:
+    return _CHAT.split(start, 1)[1].split(end, 1)[0]
+
+
+def test_in_thinking_delta_short_circuits_before_cumulative_normalization():
+    delta_handler = _between(
+        "let _delta = json.delta;",
+        "} else if (json.type === 'research_progress')",
+    )
+    delta_path = _between(
+        "// Detect thinking-in-progress:",
+        "} else if (json.type === 'research_progress')",
+    )
+    guard = "if (!_thinkingAnalysisGate.shouldAnalyze(roundText, {"
+    normalize = "markdownModule.normalizeThinkingMarkup(roundText)"
+    assert guard in delta_path
+    assert delta_path.index(guard) < delta_path.index(normalize)
+    assert "_queueLiveThinking(roundText);" in delta_path
+    assert "createThinkingAnalysisGate" in _CHAT
+    projector_append = "_roundDisplayProjector.append(_delta, roundText);"
+    assert projector_append in delta_handler
+    assert delta_handler.index(projector_append) < delta_handler.index(guard)
+    assert "_renderStream({ knownNormal: true, displayText: _roundDisplayProjector.current() });" in delta_path
+    assert "_replyDisplayProjector.append(_delta, roundReplyText)" in delta_path
+
+
+def test_short_close_grace_expires_without_another_delta():
+    assert "function _scheduleThinkingGrace()" in _CHAT
+    grace = _between(
+        "function _scheduleThinkingGrace()",
+        "function _replyAfterClosedThinking",
+    )
+    assert "setTimeout(() =>" in grace
+    assert "_finishLiveThinkingTransition();" in grace
+    cancel = _between("_cancelLiveThinkingWork = () =>", "function _finalizeLiveThinking")
+    assert "_cancelThinkingGrace();" in cancel
+    delta_path = _between(
+        "// Detect thinking-in-progress:",
+        "} else if (json.type === 'research_progress')",
+    )
+    false_close = _between(
+        "// Detect false close:",
+        "if (hasUnclosedThink && !isThinking)",
+    )
+    assert "Do NOT require a prior unclosed delta" in false_close
+    assert "_afterClose &&" in false_close
+    assert "&& isThinking" not in false_close.split("let _falseCloseDeadline", 1)[1].split("if (isThinking)", 1)[0]
+    assert "_thinkingRecheckAt = _falseCloseDeadline || 0;" in delta_path
+
+
+def test_terminal_paths_use_one_authoritative_rich_round_render():
+    tool_path = _between(
+        "} else if (json.type === 'tool_start') {",
+        "} else if (json.type === 'tool_output') {",
+    )
+    assert "_endLiveThinkingSection({ rich: false });" in tool_path
+    assert tool_path.count("_finalizeRoundRender();") == 1
+    assert "_renderStream();" not in tool_path
+
+    agent_path = _between(
+        "} else if (json.type === 'agent_step') {",
+        "} else if (json.type === 'budget_exceeded') {",
+    )
+    assert "_endLiveThinkingSection({ rich: false });" in agent_path
+    assert agent_path.count("_finalizeRoundRender();") == 1
+    assert "if (!roundFinalized)" not in agent_path
+
+    catch_path = _between(
+        "// foreground session's text.\n      const _isBgCatch",
+        "} finally {",
+    )
+    assert "if (_isBgCatch)" in catch_path
+    assert "_cancelLiveThinkingWork();" in catch_path
+    assert "_catchTerminalView = _finalizeInterruptedView();" in catch_path
+    assert "_finalizeRoundRender();" not in catch_path
+    assert "_endThinkingOnTerminalPath({ rich: false });" in catch_path
+    assert "const _catchViewHolder = _catchTerminalView?.holder || holder;" in catch_path
+
+    round_finalizer = _between(
+        "_finalizeRoundRender = () => {",
+        "_finalizeInterruptedView = () => {",
+    )
+    assert "if (roundFinalized) return roundFinalization;" in round_finalizer
+    assert round_finalizer.index("processWithThinking") < round_finalizer.rindex("roundFinalized = true;")
+    assert "lastContentRoundHolder = terminalHolder;" in round_finalizer
+
+    interrupted_finalizer = _between(
+        "_finalizeInterruptedView = () => {",
+        "function _replyAfterClosedThinking",
+    )
+    assert "finalization?.hasContent" in interrupted_finalizer
+    assert "lastContentRoundHolder || finalization?.holder" in interrupted_finalizer
+
+    stop_path = _between(
+        "// Render whatever was accumulated so far",
+        "// Reset button state",
+    )
+    assert "const _stoppedViewHolder = _terminalView?.holder || currentHolder;" in stop_path
+    assert "_stoppedViewHolder.querySelector('.body').appendChild(stoppedIndicator);" in stop_path
+
+    done_path = _between(
+        "if (data === '[DONE]') {",
+        "try {\n              const json = JSON.parse(data);",
+    )
+    assert "_finalizeLiveThinking(_closedThinkingText(roundText), false);" in done_path
+    assert "_renderStream();" not in done_path
+
+    post_loop = _between(
+        "if (!_streamSawDone) {",
+        "// --- Final render (skip if stream was ever backgrounded or currently in background) ---",
+    )
+    assert "_cancelLiveThinkingWork();" in post_loop
+    assert "_renderStream();" not in post_loop
+
+    recovery_path = _between(
+        "function _tryAutoRecover(holder, accumulated, sessionId)",
+        "function _removeStallBanner()",
+    )
+    assert "processWithThinking" not in recovery_path
+
+
+def test_detach_synchronously_cancels_delayed_view_work():
+    registration = _between("_activeStreams.set(streamSessionId", "_syncForegroundStreamGlobals();")
+    assert "cancelViewWork: () => _cancelLiveThinkingWork()" in registration
+
+    detach = _between("export function detachCurrentStream", "// _notifyStreamComplete")
+    cancel = "if (active.cancelViewWork) active.cancelViewWork();"
+    background = "_backgroundStreams.set(sessionId"
+    assert cancel in detach
+    assert detach.index(cancel) < detach.index(background)

tests/test_live_thinking_scheduler_js.py

@@ -0,0 +1,29 @@
+"""Runs the live-thinking throttle's behavioral suite under pytest.
+
+Behavior lives in tests/live_thinking_scheduler.test.mjs (node:test, no DOM).
+This wrapper only exists so the JS suite runs in the normal pytest job.
+"""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+_REPO = Path(__file__).resolve().parent.parent
+_HAS_NODE = shutil.which("node") is not None
+
+
+@pytest.mark.skipif(not _HAS_NODE, reason="node binary not on PATH")
+def test_live_thinking_scheduler_behavior():
+    result = subprocess.run(
+        ["node", "--test", "tests/live_thinking_scheduler.test.mjs"],
+        cwd=_REPO,
+        capture_output=True,
+        timeout=30,
+        text=True,
+    )
+    if result.returncode != 0:
+        raise AssertionError(
+            f"node --test failed:\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
+        )