fix(chat): heartbeat keeps long reasoning turns alive past 2-min timeout (#4270)

Cargo.toml

@@ -279,6 +279,11 @@ sentry = { version = "0.47.0", default-features = false, features = ["test"] }
 wiremock = "0.6"
 # Used in json_rpc_e2e to backdate mtime on stale lock files.
 filetime = "0.2"
+# `test-util` enables tokio's paused virtual clock (`start_paused`,
+# `time::advance`) so the #4270 inference-heartbeat tests assert the periodic
+# beat without real-time waits. Test-only — the runtime feature set in
+# `[dependencies]` (`full`) intentionally excludes it.
+tokio = { version = "1", features = ["test-util"] }
 
 [features]
 default = ["tokenjuice-treesitter"]

app/src/pages/Conversations.tsx

@@ -354,6 +354,12 @@ const Conversations = ({
   const streamingAssistantByThread = useAppSelector(
     state => state.chatRuntime.streamingAssistantByThread
   );
+  // #4270: per-thread liveness counter bumped on each `inference_heartbeat`.
+  // Watched by the silence-timer rearm effect so a long prefill / buffered
+  // reasoning phase that streams no other progress still keeps the timer armed.
+  const inferenceHeartbeatByThread = useAppSelector(
+    state => state.chatRuntime.inferenceHeartbeatByThread
+  );
   const parallelStreamsByThread = useAppSelector(
     state => state.chatRuntime.parallelStreamsByThread
   );
@@ -798,6 +804,10 @@ const Conversations = ({
         streamingAssistantByThread[threadId],
         toolTimelineByThread[threadId],
         taskBoardByThread[threadId],
+        // #4270: liveness beat. Kept LAST so the done-transition probe on
+        // `current[0]` (status) is unaffected; a beat alone still flips the
+        // `changed` check and rearms the timer through a silent reasoning phase.
+        inferenceHeartbeatByThread[threadId],
       ] as const;
       const previous = turnSignatureByThreadRef.current.get(threadId);
       const status = current[0];
@@ -820,6 +830,7 @@ const Conversations = ({
     streamingAssistantByThread,
     toolTimelineByThread,
     taskBoardByThread,
+    inferenceHeartbeatByThread,
   ]);
 
   useEffect(() => {

app/src/pages/__tests__/Conversations.render.test.tsx

@@ -21,6 +21,7 @@ import agentProfileReducer from '../../store/agentProfileSlice';
 import chatRuntimeReducer, {
   appendProcessingProse,
   beginInferenceTurn,
+  bumpInferenceHeartbeatForThread,
   clearFollowupsForThread,
   enqueueFollowup,
   setInferenceStatusForThread,
@@ -1296,6 +1297,84 @@ describe('Conversations — smoke render (#1123 welcome-lock removal)', () => {
     }
   });
 
+  it('rearms the silence timer on inference heartbeat beats during a silent reasoning phase (#4270)', async () => {
+    // Repro for #4270: a long prefill on a large context, or a reasoning-tier
+    // model that buffers `reasoning_content` server-side, streams NO status /
+    // text / tool / board signal for minutes. The core now emits a periodic
+    // `inference_heartbeat`; the rearm effect must treat it as liveness so the
+    // 120s silence timer never false-fires while the turn is genuinely working.
+    vi.useFakeTimers({ shouldAdvanceTime: true });
+    try {
+      const { textarea, store, thread } = await renderSelectedConversation();
+
+      await act(async () => {
+        fireEvent.change(textarea, {
+          target: { value: 'summarize a big codebase in reasoning mode' },
+        });
+      });
+      await act(async () => {
+        fireEvent.click(screen.getByRole('button', { name: 'Send message' }));
+      });
+      await waitFor(() => {
+        expect(chatSend).toHaveBeenCalledTimes(1);
+      });
+
+      // 200s elapse in 20s steps — only a heartbeat each step, nothing else.
+      // Without the #4270 fix the 120s timer would fire around the 6th step.
+      for (let i = 0; i < 10; i++) {
+        await act(async () => {
+          await vi.advanceTimersByTimeAsync(20_000);
+        });
+        await act(async () => {
+          store!.dispatch(bumpInferenceHeartbeatForThread({ threadId: thread.id }));
+        });
+      }
+
+      // The beats kept rearming the timer → the turn is still marked active
+      // (a fired safety timeout would have dispatched `clearThreadInferenceActive`).
+      expect(store!.getState().thread.activeThreadIds[thread.id]).toBe(true);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('still fails fast when heartbeats stop — genuine disconnect surfaces (#4270 regression safety)', async () => {
+    // Regression safety: the heartbeat is the liveness signal, so a real
+    // connectivity drop (core/socket dead → no more beats) MUST still trip the
+    // 120s silence timer rather than hanging forever.
+    vi.useFakeTimers({ shouldAdvanceTime: true });
+    try {
+      const { textarea, store, thread } = await renderSelectedConversation();
+
+      await act(async () => {
+        fireEvent.change(textarea, { target: { value: 'task whose connection dies' } });
+      });
+      await act(async () => {
+        fireEvent.click(screen.getByRole('button', { name: 'Send message' }));
+      });
+      await waitFor(() => {
+        expect(chatSend).toHaveBeenCalledTimes(1);
+      });
+
+      // A couple of early beats, then silence (the socket died).
+      await act(async () => {
+        await vi.advanceTimersByTimeAsync(20_000);
+      });
+      await act(async () => {
+        store!.dispatch(bumpInferenceHeartbeatForThread({ threadId: thread.id }));
+      });
+
+      // No more beats for a full 120s window → the silence timer fires and
+      // drops the thread from the active set.
+      await act(async () => {
+        await vi.advanceTimersByTimeAsync(120_000);
+      });
+      expect(store!.getState().thread.activeThreadIds[thread.id]).toBeFalsy();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it('does NOT rearm the silence timer on an unrelated thread’s updates', async () => {
     // Regression for the per-thread dependency scoping: the rearm effect must
     // react only to the SENDING thread's slices. A different thread churning

app/src/providers/ChatRuntimeProvider.tsx

@@ -7,6 +7,7 @@ import { ingestRuntimeErrorSignal } from '../lib/userErrors/report';
 import {
   type ChatApprovalRequestEvent,
   type ChatDoneEvent,
+  type ChatInferenceHeartbeatEvent,
   type ChatInferenceStartEvent,
   type ChatIterationStartEvent,
   type ChatPlanReviewRequestEvent,
@@ -25,6 +26,7 @@ import { store } from '../store';
 import {
   appendProcessingProse,
   appendSubagentStreamDelta,
+  bumpInferenceHeartbeatForThread,
   clearInferenceStatusForThread,
   clearParallelRequest,
   clearPendingApprovalForThread,
@@ -473,6 +475,20 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
           })
         );
       },
+      onInferenceHeartbeat: (event: ChatInferenceHeartbeatEvent) => {
+        // #4270: liveness beat — bump the per-thread counter so the
+        // Conversations silence timer rearms even when the turn is in a long
+        // prefill / buffered-reasoning phase that emits no other progress.
+        rtLog('inference_heartbeat', { thread: event.thread_id, request: event.request_id });
+        // A parallel (forked) turn streams into its own lane and must NOT keep
+        // the thread's primary silence timer alive — otherwise a sibling branch
+        // would mask a stalled primary turn. Mirror the text/thinking-delta
+        // routing: ignore heartbeats owned by a parallel request.
+        if (store.getState().chatRuntime.parallelRequestThreads[event.request_id] !== undefined) {
+          return;
+        }
+        dispatch(bumpInferenceHeartbeatForThread({ threadId: event.thread_id }));
+      },
       onIterationStart: (event: ChatIterationStartEvent) => {
         const prev = inferenceStatusRef.current[event.thread_id];
         rtLog('iteration_start', {

app/src/providers/__tests__/ChatRuntimeProvider.test.tsx

@@ -364,6 +364,31 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria
       expect(after.streamingAssistantByThread['t-par']?.content).toBe('P');
     });
 
+    it('bumps the heartbeat counter only for the primary turn, never a parallel branch (#4282)', () => {
+      const listeners = renderProvider();
+
+      // Primary turn's heartbeat advances the thread's liveness counter.
+      act(() => {
+        listeners.onInferenceHeartbeat?.({ thread_id: 't-par', request_id: 'primary' });
+      });
+      expect(store.getState().chatRuntime.inferenceHeartbeatByThread['t-par']).toBe(1);
+
+      // A registered parallel branch's heartbeat must NOT rearm the primary
+      // silence timer — otherwise a sibling would mask a stalled primary turn.
+      act(() => {
+        store.dispatch(registerParallelRequest({ threadId: 't-par', requestId: 'branch' }));
+        listeners.onInferenceHeartbeat?.({ thread_id: 't-par', request_id: 'branch' });
+        listeners.onInferenceHeartbeat?.({ thread_id: 't-par', request_id: 'branch' });
+      });
+      expect(store.getState().chatRuntime.inferenceHeartbeatByThread['t-par']).toBe(1);
+
+      // The primary turn keeps beating independently.
+      act(() => {
+        listeners.onInferenceHeartbeat?.({ thread_id: 't-par', request_id: 'primary' });
+      });
+      expect(store.getState().chatRuntime.inferenceHeartbeatByThread['t-par']).toBe(2);
+    });
+
     it('drops duplicate chat_done events with the same thread/request', async () => {
       const listeners = renderProvider();
 

app/src/services/__tests__/chatService.test.ts

@@ -68,6 +68,22 @@ describe('chatService.subscribeChatEvents', () => {
     expect(subscribedEvents).not.toContain('chat:error');
   });
 
+  it('forwards inference_heartbeat through onInferenceHeartbeat (#4270)', () => {
+    const socket = createMockSocket();
+    vi.mocked(socketService.getSocket).mockReturnValue(socket as never);
+    const onInferenceHeartbeat = vi.fn();
+
+    subscribeChatEvents({ onInferenceHeartbeat });
+
+    const subscribedEvents = socket.on.mock.calls.map(call => call[0]);
+    expect(subscribedEvents).toEqual(['inference_heartbeat']);
+
+    const beat = { thread_id: 't1', request_id: 'r1' };
+    socket.emit('inference_heartbeat', beat);
+    expect(onInferenceHeartbeat).toHaveBeenCalledTimes(1);
+    expect(onInferenceHeartbeat).toHaveBeenCalledWith(beat);
+  });
+
   it('does not process alias events when only canonical subscriptions are active', () => {
     const socket = createMockSocket();
     vi.mocked(socketService.getSocket).mockReturnValue(socket as never);

app/src/services/chatService.ts

@@ -275,6 +275,18 @@ export interface ChatInferenceStartEvent {
   request_id: string;
 }
 
+/**
+ * Periodic liveness beat emitted by the core progress bridge for the whole
+ * duration of an in-flight turn (issue #4270). Carries no payload beyond the
+ * ids — its sole purpose is to rearm the frontend silence timer so a long
+ * prefill or a buffered-reasoning phase that streams no other progress event
+ * does not trip a false "no response after 2 minutes" timeout.
+ */
+export interface ChatInferenceHeartbeatEvent {
+  thread_id: string;
+  request_id: string;
+}
+
 /** Emitted at the start of each LLM iteration in the tool loop. */
 export interface ChatIterationStartEvent {
   thread_id: string;
@@ -493,6 +505,7 @@ export interface ChatTaskBoardUpdatedEvent {
 
 export interface ChatEventListeners {
   onInferenceStart?: (event: ChatInferenceStartEvent) => void;
+  onInferenceHeartbeat?: (event: ChatInferenceHeartbeatEvent) => void;
   onIterationStart?: (event: ChatIterationStartEvent) => void;
   onToolCall?: (event: ChatToolCallEvent) => void;
   onToolResult?: (event: ChatToolResultEvent) => void;
@@ -529,6 +542,7 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
   // processing the same logical event twice.
   const EVENTS = {
     inferenceStart: 'inference_start',
+    inferenceHeartbeat: 'inference_heartbeat',
     iterationStart: 'iteration_start',
     toolCall: 'tool_call',
     toolResult: 'tool_result',
@@ -566,6 +580,21 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
     handlers.push([EVENTS.inferenceStart, cb]);
   }
 
+  if (listeners.onInferenceHeartbeat) {
+    const cb = (payload: unknown) => {
+      const e = payload as ChatInferenceHeartbeatEvent;
+      chatLog(
+        '%s thread_id=%s request_id=%s',
+        EVENTS.inferenceHeartbeat,
+        e.thread_id,
+        e.request_id
+      );
+      listeners.onInferenceHeartbeat?.(e);
+    };
+    socket.on(EVENTS.inferenceHeartbeat, cb);
+    handlers.push([EVENTS.inferenceHeartbeat, cb]);
+  }
+
   if (listeners.onIterationStart) {
     const cb = (payload: unknown) => {
       const e = payload as ChatIterationStartEvent;

app/src/store/__tests__/chatRuntimeSlice.test.ts

@@ -3,6 +3,7 @@ import { describe, expect, it } from 'vitest';
 import type { PersistedTurnState } from '../../types/turnState';
 import reducer, {
   beginInferenceTurn,
+  bumpInferenceHeartbeatForThread,
   clearAllChatRuntime,
   clearArtifactsForThread,
   clearInferenceStatusForThread,
@@ -49,6 +50,32 @@ describe('chatRuntimeSlice', () => {
     expect(cleared.inferenceStatusByThread['thread-1']).toBeUndefined();
   });
 
+  it('bumps the per-thread inference heartbeat counter and clears it on runtime reset (#4270)', () => {
+    const once = reducer(undefined, bumpInferenceHeartbeatForThread({ threadId: 'thread-1' }));
+    expect(once.inferenceHeartbeatByThread['thread-1']).toBe(1);
+
+    // Each beat advances the counter — the silence-timer rearm watches the
+    // *change*, so a monotonically-different value per beat is what matters.
+    const twice = reducer(once, bumpInferenceHeartbeatForThread({ threadId: 'thread-1' }));
+    expect(twice.inferenceHeartbeatByThread['thread-1']).toBe(2);
+
+    // A different thread tracks its own counter independently.
+    const other = reducer(twice, bumpInferenceHeartbeatForThread({ threadId: 'thread-2' }));
+    expect(other.inferenceHeartbeatByThread['thread-2']).toBe(1);
+    expect(other.inferenceHeartbeatByThread['thread-1']).toBe(2);
+
+    // Turn teardown drops the liveness counter so a stale value can't rearm a
+    // fresh turn's timer.
+    const cleared = reducer(other, clearRuntimeForThread({ threadId: 'thread-1' }));
+    expect(cleared.inferenceHeartbeatByThread['thread-1']).toBeUndefined();
+    expect(cleared.inferenceHeartbeatByThread['thread-2']).toBe(1);
+
+    // Slice-wide reset wipes every thread's heartbeat too — no stale counter
+    // survives a full chat-runtime clear (CodeRabbit #4282).
+    const wiped = reducer(other, clearAllChatRuntime());
+    expect(wiped.inferenceHeartbeatByThread).toEqual({});
+  });
+
   it('stores and clears streaming assistant content by thread', () => {
     const withStreaming = reducer(
       undefined,

app/src/store/chatRuntimeSlice.ts

@@ -411,6 +411,15 @@ export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect' | 'parall
 interface ChatRuntimeState {
   inferenceStatusByThread: Record<string, InferenceStatus>;
   streamingAssistantByThread: Record<string, StreamingAssistantState>;
+  /**
+   * Monotonically-bumped liveness counter per thread, advanced on every
+   * `inference_heartbeat` socket event the core emits while a turn is in flight
+   * (issue #4270). The Conversations silence timer watches this alongside the
+   * status/stream/tool/board slices, so a long prefill or buffered-reasoning
+   * phase that emits no other progress still rearms the timer and avoids a
+   * false "no response after 2 minutes" timeout. Cleared on turn end.
+   */
+  inferenceHeartbeatByThread: Record<string, number>;
   /**
    * Threads with an optimistic user send in flight, set the instant the user
    * sends (before `addMessageLocal` resolves and before any streaming state
@@ -503,6 +512,7 @@ export interface QueuedFollowup {
 const initialState: ChatRuntimeState = {
   inferenceStatusByThread: {},
   streamingAssistantByThread: {},
+  inferenceHeartbeatByThread: {},
   pendingSendThreadIds: {},
   parallelStreamsByThread: {},
   parallelRequestThreads: {},
@@ -756,6 +766,15 @@ const chatRuntimeSlice = createSlice({
     clearInferenceStatusForThread: (state, action: PayloadAction<{ threadId: string }>) => {
       delete state.inferenceStatusByThread[action.payload.threadId];
     },
+    /**
+     * Bump a thread's liveness counter on each `inference_heartbeat` (issue
+     * #4270). The value is opaque — only the *change* matters to the silence
+     * timer's signature comparison. Wraps via modulo to stay a small integer.
+     */
+    bumpInferenceHeartbeatForThread: (state, action: PayloadAction<{ threadId: string }>) => {
+      const prev = state.inferenceHeartbeatByThread[action.payload.threadId] ?? 0;
+      state.inferenceHeartbeatByThread[action.payload.threadId] = (prev + 1) % 1_000_000;
+    },
     setStreamingAssistantForThread: (
       state,
       action: PayloadAction<{ threadId: string; streaming: StreamingAssistantState }>
@@ -1181,6 +1200,7 @@ const chatRuntimeSlice = createSlice({
     clearRuntimeForThread: (state, action: PayloadAction<{ threadId: string }>) => {
       delete state.inferenceStatusByThread[action.payload.threadId];
       delete state.streamingAssistantByThread[action.payload.threadId];
+      delete state.inferenceHeartbeatByThread[action.payload.threadId];
       // Drop any parallel (forked) streams for this thread and their
       // request→thread mappings — a hard per-thread reset covers every branch.
       const parallelStreams = state.parallelStreamsByThread[action.payload.threadId];
@@ -1208,6 +1228,7 @@ const chatRuntimeSlice = createSlice({
     clearAllChatRuntime: state => {
       state.inferenceStatusByThread = {};
       state.streamingAssistantByThread = {};
+      state.inferenceHeartbeatByThread = {};
       state.parallelStreamsByThread = {};
       state.parallelRequestThreads = {};
       state.toolTimelineByThread = {};
@@ -1398,6 +1419,7 @@ const chatRuntimeSlice = createSlice({
 export const {
   setInferenceStatusForThread,
   clearInferenceStatusForThread,
+  bumpInferenceHeartbeatForThread,
   setStreamingAssistantForThread,
   clearStreamingAssistantForThread,
   markThreadSendPending,