Fix agent history catch-up + avoid mobile background stream deltas (#18)

* Update files * Fix agent history sync + mobile stream gating
2026-07-29 12:01:31 +00:00 · 2026-02-06 13:18:15 +07:00
parent 4633a7e8fe
commit d724988542
8 changed files with 169 additions and 117 deletions
--- 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<string, PendingPermission>;
+  /**
+   * 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<FlatList<StreamItem>>(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 (
+        <View style={[stylesheet.emptyState, stylesheet.contentWrapper]}>
+          <ActivityIndicator
+            size="small"
+            color={theme.colors.foregroundMuted}
+          />
+          <Text style={stylesheet.emptyStateText}>Catching up…</Text>
+        </View>
+      );
+    }
+
+    return (
+      <View style={[stylesheet.emptyState, stylesheet.contentWrapper]}>
+        <Text style={stylesheet.emptyStateText}>
+          Start chatting with this agent...
+        </Text>
+      </View>
+    );
+  }, [
+    agent.status,
+    isSyncingHistory,
+    pendingPermissionItems.length,
+    streamHead,
+    theme.colors.foregroundMuted,
+  ]);
+
  return (
    <ToolCallSheetProvider>
      <View style={stylesheet.container}>
@@ -607,13 +650,7 @@ export function AgentStreamView({
              style={stylesheet.list}
              onScroll={handleScroll}
              scrollEventThrottle={16}
-              ListEmptyComponent={
-                <View style={[stylesheet.emptyState, stylesheet.contentWrapper]}>
-                  <Text style={stylesheet.emptyStateText}>
-                    Start chatting with this agent...
-                  </Text>
-                </View>
-              }
+              ListEmptyComponent={listEmptyComponent}
              ListHeaderComponent={listHeaderComponent}
              extraData={flatListExtraData}
              maintainVisibleContentPosition={
--- 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(
--- 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<void> => {
@@ -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 };
 }
-
--- 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(() => {
--- 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({
            <ReanimatedAnimated.View
              style={[styles.content, animatedKeyboardStyle]}
            >
-              {isInitializing && !shouldUseOptimisticStream ? (
-                <View style={styles.loadingContainer}>
-                  <ActivityIndicator
-                    size="large"
-                    color={theme.colors.primary}
-                  />
-                  <Text style={styles.loadingText}>Loading agent...</Text>
-                </View>
-              ) : (
-                <AgentStreamView
-                  agentId={effectiveAgent.id}
-                  serverId={serverId}
-                  agent={effectiveAgent}
-                  streamItems={shouldUseOptimisticStream ? mergedStreamItems : streamItems}
-                  pendingPermissions={pendingPermissions}
-                />
-              )}
+              <AgentStreamView
+                agentId={effectiveAgent.id}
+                serverId={serverId}
+                agent={effectiveAgent}
+                streamItems={
+                  shouldUseOptimisticStream ? mergedStreamItems : streamItems
+                }
+                pendingPermissions={pendingPermissions}
+                isSyncingHistory={isInitializing && !shouldUseOptimisticStream}
+              />
            </ReanimatedAnimated.View>
          </View>

          {/* Agent Input Area */}
-          {!isInitializing && agent && resolvedAgentId && (
+          {agent && resolvedAgentId && (
            <AgentInputArea agentId={resolvedAgentId} serverId={serverId} autoFocus onAddImages={handleAddImagesCallback} />
          )}

--- 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 =
--- 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<void>;
  resolve: () => void;
  reject: (error: Error) => void;
+  timeoutId: ReturnType<typeof setTimeout> | null;
 }

 const initPromises = new Map<string, DeferredInit>();
@@ -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<typeof setTimeout>): 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);
 }
-
--- 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),