From d72498854236070de7423c86fd22496650dd0431 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 6 Feb 2026 13:18:15 +0700 Subject: [PATCH] Fix agent history catch-up + avoid mobile background stream deltas (#18) * Update files * Fix agent history sync + mobile stream gating --- .../app/src/components/agent-stream-view.tsx | 51 ++++++++++++-- packages/app/src/contexts/session-context.tsx | 47 ++----------- .../app/src/hooks/use-agent-initialization.ts | 46 ++++++------ packages/app/src/hooks/use-client-activity.ts | 5 +- .../src/screens/agent/agent-ready-screen.tsx | 70 ++++++++++++++----- packages/app/src/types/stream.ts | 24 ++----- .../app/src/utils/agent-initialization.ts | 25 ++++++- packages/server/src/server/session.ts | 18 +++++ 8 files changed, 169 insertions(+), 117 deletions(-) diff --git a/packages/app/src/components/agent-stream-view.tsx b/packages/app/src/components/agent-stream-view.tsx index 303ed00c7..5a655e6b9 100644 --- a/packages/app/src/components/agent-stream-view.tsx +++ b/packages/app/src/components/agent-stream-view.tsx @@ -84,6 +84,11 @@ export interface AgentStreamViewProps { agent: Agent; streamItems: StreamItem[]; pendingPermissions: Map; + /** + * True when we expect a history snapshot to arrive shortly (e.g. after opening an agent + * or resuming after connectivity changes) and should avoid showing the "empty" state. + */ + isSyncingHistory?: boolean; } export function AgentStreamView({ @@ -92,6 +97,7 @@ export function AgentStreamView({ agent, streamItems, pendingPermissions, + isSyncingHistory = false, }: AgentStreamViewProps) { const flatListRef = useRef>(null); const { theme } = useUnistyles(); @@ -590,6 +596,43 @@ export function AgentStreamView({ return { marginBottom: tightGap }; }, [listHeaderComponent, tightGap]); + const listEmptyComponent = useMemo(() => { + const hasPermissions = pendingPermissionItems.length > 0; + const hasHeadItems = (streamHead?.length ?? 0) > 0; + if (hasPermissions || hasHeadItems) { + return null; + } + + const shouldShowSyncing = + isSyncingHistory || agent.status === "running"; + + if (shouldShowSyncing) { + return ( + + + Catching up… + + ); + } + + return ( + + + Start chatting with this agent... + + + ); + }, [ + agent.status, + isSyncingHistory, + pendingPermissionItems.length, + streamHead, + theme.colors.foregroundMuted, + ]); + return ( @@ -607,13 +650,7 @@ export function AgentStreamView({ style={stylesheet.list} onScroll={handleScroll} scrollEventThrottle={16} - ListEmptyComponent={ - - - Start chatting with this agent... - - - } + ListEmptyComponent={listEmptyComponent} ListHeaderComponent={listHeaderComponent} extraData={flatListExtraData} maintainVisibleContentPosition={ diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 595917587..5501a0487 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -915,16 +915,6 @@ export function SessionProvider({ }); } - setInitializingAgents(serverId, (prev) => { - const currentState = prev.get(agentId); - if (currentState === false) { - return prev; - } - const next = new Map(prev); - next.set(agentId, false); - return next; - }); - // NOTE: We don't update lastActivityAt on every stream event to prevent // cascading rerenders. The agent_update handler updates agent.lastActivityAt // on status changes, which is sufficient for sorting and display purposes. @@ -956,7 +946,7 @@ export function SessionProvider({ clearAgentStreamHead(serverId, agentId); setInitializingAgents(serverId, (prev) => { - if (!prev.has(agentId)) { + if (prev.get(agentId) !== true) { return prev; } const next = new Map(prev); @@ -990,16 +980,6 @@ export function SessionProvider({ requestId: (message.payload as any).requestId, }); } - if (status === "agent_initialized" && "agentId" in message.payload) { - const agentId = (message.payload as any).agentId as string | undefined; - if (agentId) { - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, false); - return next; - }); - } - } }); const unsubPermissionRequest = client.on( @@ -1443,12 +1423,6 @@ export function SessionProvider({ return next; }); - setAgentStreamTail(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, []); - return next; - }); - clearAgentStreamHead(serverId, agentId); if (!client) { console.warn("[Session] initializeAgent skipped: daemon unavailable"); setInitializingAgents(serverId, (prev) => { @@ -1470,24 +1444,17 @@ export function SessionProvider({ }); }); }, - [serverId, client, setAgentStreamTail, setInitializingAgents, clearAgentStreamHead] + [serverId, client, setInitializingAgents] ); const refreshAgent = useCallback( - ({ agentId, requestId }: { agentId: string; requestId?: string }) => { + ({ agentId }: { agentId: string }) => { setInitializingAgents(serverId, (prev) => { const next = new Map(prev); next.set(agentId, true); return next; }); - setAgentStreamTail(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, []); - return next; - }); - clearAgentStreamHead(serverId, agentId); - refreshAgentMutation .mutateAsync({ agentId }) .catch((error) => { @@ -1499,13 +1466,7 @@ export function SessionProvider({ }); }); }, - [ - serverId, - refreshAgentMutation, - setAgentStreamTail, - setInitializingAgents, - clearAgentStreamHead, - ] + [serverId, refreshAgentMutation, setInitializingAgents] ); const sendAgentMessage = useCallback( diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index 139847209..337db27a2 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -1,19 +1,18 @@ import { useCallback } from "react"; import { useSessionStore } from "@/stores/session-store"; import { + attachInitTimeout, createInitDeferred, getInitDeferred, getInitKey, rejectInitDeferred, } from "@/utils/agent-initialization"; -const INIT_TIMEOUT_MS = 10000; +const INIT_TIMEOUT_MS = 5 * 60_000; export function useAgentInitialization(serverId: string) { const client = useSessionStore((state) => state.sessions[serverId]?.client ?? null); const setInitializingAgents = useSessionStore((state) => state.setInitializingAgents); - const setAgentStreamTail = useSessionStore((state) => state.setAgentStreamTail); - const clearAgentStreamHead = useSessionStore((state) => state.clearAgentStreamHead); const ensureAgentIsInitialized = useCallback( (agentId: string): Promise => { @@ -24,10 +23,23 @@ export function useAgentInitialization(serverId: string) { } const deferred = createInitDeferred(key); - const timeoutId = setTimeout(() => { - rejectInitDeferred(key, new Error(`Agent initialization timed out after ${INIT_TIMEOUT_MS}ms`)); + setInitializingAgents(serverId, (prev) => { + if (prev.get(agentId) !== true) { + return prev; + } + const next = new Map(prev); + next.set(agentId, false); + return next; + }); + rejectInitDeferred( + key, + new Error( + `History sync timed out after ${Math.round(INIT_TIMEOUT_MS / 1000)}s` + ) + ); }, INIT_TIMEOUT_MS); + attachInitTimeout(key, timeoutId); setInitializingAgents(serverId, (prev) => { const next = new Map(prev); @@ -35,15 +47,7 @@ export function useAgentInitialization(serverId: string) { return next; }); - setAgentStreamTail(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, []); - return next; - }); - clearAgentStreamHead(serverId, agentId); - if (!client) { - clearTimeout(timeoutId); setInitializingAgents(serverId, (prev) => { const next = new Map(prev); next.set(agentId, false); @@ -56,10 +60,10 @@ export function useAgentInitialization(serverId: string) { client .initializeAgent(agentId) .then(() => { - clearTimeout(timeoutId); + // No-op: the actual "timeline hydrated" signal is the `agent_stream_snapshot` + // message, handled in SessionContext. }) .catch((error) => { - clearTimeout(timeoutId); setInitializingAgents(serverId, (prev) => { const next = new Map(prev); next.set(agentId, false); @@ -73,7 +77,7 @@ export function useAgentInitialization(serverId: string) { return deferred.promise; }, - [clearAgentStreamHead, client, serverId, setAgentStreamTail, setInitializingAgents] + [client, serverId, setInitializingAgents] ); const refreshAgent = useCallback( @@ -87,13 +91,6 @@ export function useAgentInitialization(serverId: string) { return next; }); - setAgentStreamTail(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, []); - return next; - }); - clearAgentStreamHead(serverId, agentId); - try { await client.refreshAgent(agentId); } catch (error) { @@ -105,9 +102,8 @@ export function useAgentInitialization(serverId: string) { throw error; } }, - [clearAgentStreamHead, client, serverId, setAgentStreamTail, setInitializingAgents] + [client, serverId, setInitializingAgents] ); return { ensureAgentIsInitialized, refreshAgent }; } - diff --git a/packages/app/src/hooks/use-client-activity.ts b/packages/app/src/hooks/use-client-activity.ts index 600457c89..1a0192f6f 100644 --- a/packages/app/src/hooks/use-client-activity.ts +++ b/packages/app/src/hooks/use-client-activity.ts @@ -55,12 +55,13 @@ export function useClientActivity({ client, focusedAgentId }: ClientActivityOpti appVisibleRef.current = nextState === "active"; if (nextState === "active") { recordUserActivity(); - maybeSendImmediateHeartbeat(); } + // Send immediately on visibility changes so the server can adapt streaming behavior. + sendHeartbeat(); }); return () => subscription.remove(); - }, [maybeSendImmediateHeartbeat, recordUserActivity]); + }, [recordUserActivity, sendHeartbeat]); // Track user activity on web for accurate staleness. useEffect(() => { diff --git a/packages/app/src/screens/agent/agent-ready-screen.tsx b/packages/app/src/screens/agent/agent-ready-screen.tsx index b9508bc8c..feee689cd 100644 --- a/packages/app/src/screens/agent/agent-ready-screen.tsx +++ b/packages/app/src/screens/agent/agent-ready-screen.tsx @@ -7,6 +7,7 @@ import { ScrollView, Platform, BackHandler, + AppState, } from "react-native"; import { useRouter } from "expo-router"; import { useFocusEffect } from "@react-navigation/native"; @@ -503,6 +504,46 @@ function AgentScreenContent({ }); }, [resolvedAgentId, ensureAgentIsInitialized, isConnected]); + // When the app comes back to the foreground, re-sync history for the focused agent. + // This covers cases where the OS/backgrounding caused us to miss stream events. + const lastAppStateRef = useRef(AppState.currentState); + const lastResumeSyncAtRef = useRef(0); + useEffect(() => { + if (!resolvedAgentId) { + return; + } + + const subscription = AppState.addEventListener("change", (nextState) => { + const prev = lastAppStateRef.current; + lastAppStateRef.current = nextState; + + if (nextState !== "active" || prev === "active") { + return; + } + if (!isConnected) { + return; + } + + const now = Date.now(); + // Avoid accidental double-syncs on rapid transitions. + if (now - lastResumeSyncAtRef.current < 2000) { + return; + } + lastResumeSyncAtRef.current = now; + + ensureAgentIsInitialized(resolvedAgentId).catch((error) => { + console.warn("[AgentScreen] Agent initialization failed on resume", { + agentId: resolvedAgentId, + error, + }); + }); + }); + + return () => { + subscription.remove(); + }; + }, [ensureAgentIsInitialized, isConnected, resolvedAgentId]); + useEffect(() => { if (Platform.OS !== "web") { return; @@ -723,28 +764,21 @@ function AgentScreenContent({ - {isInitializing && !shouldUseOptimisticStream ? ( - - - Loading agent... - - ) : ( - - )} + {/* Agent Input Area */} - {!isInitializing && agent && resolvedAgentId && ( + {agent && resolvedAgentId && ( )} diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index ff8114ec2..e204fd181 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -46,25 +46,11 @@ function createUniqueTimelineId( timestamp: Date ): string { const base = createTimelineId(prefix, text, timestamp); - let suffixSeed = state.length; - let candidate = `${base}_${suffixSeed.toString(36)}`; - - // Fast path when the generated id hasn't been used yet - const hasCollision = state.some((entry) => entry.id === candidate); - if (!hasCollision) { - return candidate; - } - - // If the id already exists (e.g. prior items were pruned/replaced), - // spin until we find an unused suffix. Building the lookup Set lazily keeps - // the common case cheap while still guaranteeing uniqueness. - const usedIds = new Set(state.map((entry) => entry.id)); - while (usedIds.has(candidate)) { - suffixSeed += 1; - candidate = `${base}_${suffixSeed.toString(36)}`; - } - - return candidate; + // We only ever append new timeline items, and we incorporate the current + // length as a monotonic suffix, so uniqueness is guaranteed without an O(n) + // collision scan (important for large hydration snapshots). + const suffixSeed = state.length; + return `${base}_${suffixSeed.toString(36)}`; } export type StreamItem = diff --git a/packages/app/src/utils/agent-initialization.ts b/packages/app/src/utils/agent-initialization.ts index 32a6f9025..f9f87c53f 100644 --- a/packages/app/src/utils/agent-initialization.ts +++ b/packages/app/src/utils/agent-initialization.ts @@ -2,6 +2,7 @@ export interface DeferredInit { promise: Promise; resolve: () => void; reject: (error: Error) => void; + timeoutId: ReturnType | null; } const initPromises = new Map(); @@ -23,14 +24,30 @@ export function createInitDeferred(key: string): DeferredInit { reject = rej; }); - const deferred: DeferredInit = { promise, resolve, reject }; + const deferred: DeferredInit = { promise, resolve, reject, timeoutId: null }; initPromises.set(key, deferred); return deferred; } +export function attachInitTimeout(key: string, timeoutId: ReturnType): void { + const deferred = initPromises.get(key); + if (!deferred) { + clearTimeout(timeoutId); + return; + } + deferred.timeoutId = timeoutId; +} + export function resolveInitDeferred(key: string): void { const deferred = initPromises.get(key); - deferred?.resolve(); + if (!deferred) { + return; + } + if (deferred.timeoutId) { + clearTimeout(deferred.timeoutId); + } + initPromises.delete(key); + deferred.resolve(); } export function rejectInitDeferred(key: string, error: Error): void { @@ -38,7 +55,9 @@ export function rejectInitDeferred(key: string, error: Error): void { if (!deferred) { return; } + if (deferred.timeoutId) { + clearTimeout(deferred.timeoutId); + } initPromises.delete(key); deferred.reject(error); } - diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 8c72c2ffe..639aa7a47 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -578,6 +578,24 @@ export class Session { return; } + // Reduce bandwidth/CPU on mobile: only forward high-frequency agent stream events + // while the app is visible and the user is focused on that agent. + // + // History catch-up is handled via explicit `initialize_agent_request` which emits a + // batched `agent_stream_snapshot`. + const activity = this.clientActivity; + if (activity?.deviceType === "mobile") { + if (!activity.appVisible) { + return; + } + if (!activity.focusedAgentId) { + return; + } + if (activity.focusedAgentId !== event.agentId) { + return; + } + } + const payload = { agentId: event.agentId, event: serializeAgentStreamEvent(event.event),