diff --git a/Cargo.toml b/Cargo.toml index a224b8b77..1231cdbf0 100644 --- a/Cargo.toml +++ b/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"] diff --git a/app/src/pages/Conversations.tsx b/app/src/pages/Conversations.tsx index 88beb9135..e5307e1b0 100644 --- a/app/src/pages/Conversations.tsx +++ b/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(() => { diff --git a/app/src/pages/__tests__/Conversations.render.test.tsx b/app/src/pages/__tests__/Conversations.render.test.tsx index b39045abf..b89253973 100644 --- a/app/src/pages/__tests__/Conversations.render.test.tsx +++ b/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 diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index 2950eca9e..aadab9151 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/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', { diff --git a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx index ee62fd02c..7fbb37663 100644 --- a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx +++ b/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(); diff --git a/app/src/services/__tests__/chatService.test.ts b/app/src/services/__tests__/chatService.test.ts index 40d5342fb..93c30cc77 100644 --- a/app/src/services/__tests__/chatService.test.ts +++ b/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); diff --git a/app/src/services/chatService.ts b/app/src/services/chatService.ts index 60c0b0aef..79d420751 100644 --- a/app/src/services/chatService.ts +++ b/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; diff --git a/app/src/store/__tests__/chatRuntimeSlice.test.ts b/app/src/store/__tests__/chatRuntimeSlice.test.ts index 7188d2e61..c3bec69ed 100644 --- a/app/src/store/__tests__/chatRuntimeSlice.test.ts +++ b/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, diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index e4197ce7a..849cdda70 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -411,6 +411,15 @@ export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect' | 'parall interface ChatRuntimeState { inferenceStatusByThread: Record; streamingAssistantByThread: Record; + /** + * 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; /** * 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, diff --git a/src/openhuman/channels/providers/web/progress_bridge.rs b/src/openhuman/channels/providers/web/progress_bridge.rs index 429f72374..d25cf8419 100644 --- a/src/openhuman/channels/providers/web/progress_bridge.rs +++ b/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, + 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 { + let (tx, rx) = tokio::sync::mpsc::channel::(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); + } }