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

Co-authored-by: M3gA-Mind <megamind@mahadao.com>

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

@@ -11,6 +11,7 @@ import { ingestRuntimeErrorSignal } from '../lib/userErrors/report';
 import {
   type ChatApprovalRequestEvent,
   type ChatDoneEvent,
+  type ChatInferenceHeartbeatEvent,
   type ChatInferenceStartEvent,
   type ChatIterationStartEvent,
   type ChatPlanReviewRequestEvent,
@@ -29,6 +30,7 @@ import { store } from '../store';
 import {
   appendProcessingProse,
   appendSubagentStreamDelta,
+  bumpInferenceHeartbeatForThread,
   clearInferenceStatusForThread,
   clearParallelRequest,
   clearPendingApprovalForThread,
@@ -481,6 +483,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,

src/openhuman/channels/providers/web/progress_bridge.rs

@@ -6,6 +6,18 @@ use crate::openhuman::threads::turn_state::{TurnStateMirror, TurnStateStore};
 use super::event_bus::publish_web_channel_event;
 use super::types::ChatRequestMetadata;
 
+/// Cadence of the `inference_heartbeat` liveness beat the bridge emits while a
+/// turn is in flight (issue #4270). The frontend silence timer in
+/// `Conversations.tsx` only fires after ~120s with NO progress signal of any
+/// kind; a long prefill on a large context, or a reasoning-tier model that
+/// buffers `reasoning_content` server-side, can legitimately stream nothing for
+/// minutes — tripping a false "no response after 2 minutes" timeout that
+/// discards the live turn. A wall-clock beat every 20s rides the same socket as
+/// the real progress events, so it keeps the timer armed while work is genuinely
+/// progressing yet stops the instant the socket/core dies — preserving the
+/// genuine-disconnect error path (6 missed beats before the 120s window lapses).
+const INFERENCE_HEARTBEAT_SECS: u64 = 20;
+
 /// Current wall-clock time as Unix-epoch milliseconds, used to stamp tracing
 /// spans (issue #3886). Saturates to `0` if the clock is before the epoch.
 fn unix_epoch_ms() -> u64 {
@@ -155,7 +167,48 @@ pub(crate) fn spawn_progress_bridge(
             None
         };
 
-        while let Some(event) = rx.recv().await {
+        // #4270: emit a periodic liveness beat for the whole in-flight turn so
+        // the frontend silence timer never false-fires during a long prefill or
+        // a buffered-reasoning phase that streams no progress events. The beat
+        // is gated on `turn_active` (set once `TurnStarted` is observed) so we
+        // never emit before the turn's `inference_start` has armed the timer.
+        let mut heartbeat =
+            tokio::time::interval(std::time::Duration::from_secs(INFERENCE_HEARTBEAT_SECS));
+        // Wall-clock cadence: a slow turn must not produce a burst of catch-up
+        // beats once it finally yields control.
+        heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        // `interval`'s first tick resolves immediately — consume it so the first
+        // real beat lands one full interval after the turn begins.
+        heartbeat.tick().await;
+        let mut turn_active = false;
+
+        loop {
+            let event = tokio::select! {
+                // Drain real progress events preferentially over the timer so a
+                // busy turn never starves event handling to emit a beat.
+                biased;
+                maybe = rx.recv() => match maybe {
+                    Some(ev) => ev,
+                    None => break,
+                },
+                _ = heartbeat.tick() => {
+                    if turn_active {
+                        log::trace!(
+                            "[web_channel][bridge] inference_heartbeat thread_id={} request_id={}",
+                            thread_id,
+                            request_id,
+                        );
+                        publish_web_channel_event(WebChannelEvent {
+                            event: "inference_heartbeat".to_string(),
+                            client_id: client_id.clone(),
+                            thread_id: thread_id.clone(),
+                            request_id: request_id.clone(),
+                            ..Default::default()
+                        });
+                    }
+                    continue;
+                }
+            };
             events_seen += 1;
             turn_state.observe(&event);
             if let Some(collector) = span_collector.as_mut() {
@@ -245,6 +298,8 @@ pub(crate) fn spawn_progress_bridge(
             }
             match event {
                 AgentProgress::TurnStarted => {
+                    // Turn is live — start emitting liveness beats (issue #4270).
+                    turn_active = true;
                     ledger_upsert_agent_run(
                         &config,
                         AgentRunUpsert {
@@ -953,6 +1008,10 @@ pub(crate) fn spawn_progress_bridge(
                 }
                 AgentProgress::TurnCompleted { iterations } => {
                     parent_completed = true;
+                    // Turn is done — stop liveness beats (issue #4270). The FE
+                    // clears its silence timer on `chat_done`/`chat_error`; this
+                    // also prevents a stray beat racing the channel close.
+                    turn_active = false;
                     let completed_at = chrono::Utc::now();
                     ledger_upsert_agent_run(
                         &config,
@@ -1129,4 +1188,156 @@ mod tests {
         );
         assert_eq!(d.dirty_status, Some(true));
     }
+
+    // ── #4270 inference heartbeat ────────────────────────────────────────────
+
+    use crate::openhuman::agent::progress::AgentProgress;
+    use crate::openhuman::config::Config;
+    use std::time::Duration;
+    use tokio::sync::broadcast::error::TryRecvError;
+
+    /// Await the next web-channel event published for `thread_id`, skipping
+    /// events for other threads (the bus is a process-global broadcast) and
+    /// tolerating broadcast lag. Panics if the channel closes first.
+    async fn recv_for_thread(
+        rx: &mut tokio::sync::broadcast::Receiver<WebChannelEvent>,
+        thread_id: &str,
+    ) -> WebChannelEvent {
+        loop {
+            match rx.recv().await {
+                Ok(ev) if ev.thread_id == thread_id => return ev,
+                Ok(_) => continue,
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                Err(err) => panic!("web-channel bus closed before event: {err}"),
+            }
+        }
+    }
+
+    fn spawn_test_bridge(
+        thread_id: &str,
+        request_id: &str,
+    ) -> tokio::sync::mpsc::Sender<AgentProgress> {
+        let (tx, rx) = tokio::sync::mpsc::channel::<AgentProgress>(16);
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = TurnStateStore::new(dir.path().to_path_buf());
+        // Keep the tempdir alive for the bridge task's lifetime by leaking it —
+        // a test-only allocation; the OS reclaims it on process exit.
+        std::mem::forget(dir);
+        spawn_progress_bridge(
+            rx,
+            "client-hb-4270".to_string(),
+            thread_id.to_string(),
+            request_id.to_string(),
+            store,
+            ChatRequestMetadata::default(),
+            Config::default(),
+        );
+        tx
+    }
+
+    /// Repro-gone guard: once a turn is in flight, the bridge emits a periodic
+    /// `inference_heartbeat` even though no other progress event has streamed —
+    /// this is the signal the FE silence timer rearms on to avoid the false
+    /// "no response after 2 minutes" timeout (#4270).
+    #[tokio::test(start_paused = true)]
+    async fn emits_inference_heartbeat_while_turn_in_flight() {
+        let mut events = super::super::event_bus::subscribe_web_channel_events();
+        let thread_id = "thread-hb-emit-4270";
+        let request_id = "req-hb-emit-4270";
+        let tx = spawn_test_bridge(thread_id, request_id);
+
+        // Turn begins — arms the liveness beat.
+        tx.send(AgentProgress::TurnStarted).await.unwrap();
+
+        // inference_start first, then a heartbeat after the interval elapses
+        // (the paused clock auto-advances while the test awaits the bus).
+        let start = recv_for_thread(&mut events, thread_id).await;
+        assert_eq!(start.event, "inference_start");
+
+        let beat = recv_for_thread(&mut events, thread_id).await;
+        assert_eq!(beat.event, "inference_heartbeat");
+        assert_eq!(beat.thread_id, thread_id);
+        assert_eq!(beat.request_id, request_id);
+
+        drop(tx);
+    }
+
+    /// Lifecycle: once `TurnCompleted` lands the bridge stops beating, so a beat
+    /// can't race the channel close after the FE has already cleared its timer
+    /// on `chat_done`/`chat_error`. Exercises the `turn_active = false` arm and
+    /// the channel-closed `break`.
+    #[tokio::test(start_paused = true)]
+    async fn stops_heartbeat_after_turn_completed() {
+        let mut events = super::super::event_bus::subscribe_web_channel_events();
+        let thread_id = "thread-hb-stop-4270";
+        let tx = spawn_test_bridge(thread_id, "req-hb-stop-4270");
+
+        tx.send(AgentProgress::TurnStarted).await.unwrap();
+        // Drain through the first heartbeat to prove the turn was beating.
+        loop {
+            if recv_for_thread(&mut events, thread_id).await.event == "inference_heartbeat" {
+                break;
+            }
+        }
+
+        // Complete the turn, then drop the sender so the bridge loop breaks.
+        tx.send(AgentProgress::TurnCompleted { iterations: 1 })
+            .await
+            .unwrap();
+        drop(tx);
+
+        // Let the bridge process TurnCompleted + observe the closed channel.
+        for _ in 0..8 {
+            tokio::task::yield_now().await;
+        }
+        // Advance well past several intervals — no further beats must appear.
+        tokio::time::advance(Duration::from_secs(INFERENCE_HEARTBEAT_SECS * 4)).await;
+        for _ in 0..8 {
+            tokio::task::yield_now().await;
+        }
+
+        loop {
+            match events.try_recv() {
+                Ok(ev) => assert_ne!(
+                    (ev.thread_id.as_str(), ev.event.as_str()),
+                    (thread_id, "inference_heartbeat"),
+                    "heartbeat emitted after TurnCompleted"
+                ),
+                Err(TryRecvError::Empty | TryRecvError::Closed) => break,
+                Err(TryRecvError::Lagged(_)) => continue,
+            }
+        }
+    }
+
+    /// Gate check: before `TurnStarted` the bridge must NOT beat — otherwise a
+    /// beat could land before the FE has armed its timer. Exercises the
+    /// `turn_active == false` branch of the heartbeat tick.
+    #[tokio::test(start_paused = true)]
+    async fn no_heartbeat_before_turn_started() {
+        let mut events = super::super::event_bus::subscribe_web_channel_events();
+        let thread_id = "thread-hb-gate-4270";
+        let tx = spawn_test_bridge(thread_id, "req-hb-gate-4270");
+
+        // Advance well past several heartbeat intervals with no TurnStarted.
+        tokio::time::advance(Duration::from_secs(INFERENCE_HEARTBEAT_SECS * 4)).await;
+        // Let the bridge task run its (no-op) heartbeat ticks.
+        for _ in 0..8 {
+            tokio::task::yield_now().await;
+        }
+
+        // No event of any kind should have been published for this thread.
+        loop {
+            match events.try_recv() {
+                Ok(ev) => assert_ne!(
+                    ev.thread_id, thread_id,
+                    "unexpected pre-turn event {} for {thread_id}",
+                    ev.event
+                ),
+                Err(TryRecvError::Empty | TryRecvError::Closed) => break,
+                Err(TryRecvError::Lagged(_)) => continue,
+            }
+        }
+
+        drop(tx);
+    }
 }