fix(mobile): stream and event durability

apps/mobile/src/components/agents/mobile-session-page-adapter.test.ts

@@ -93,6 +93,7 @@ describe('fetchMobileSessionSnapshotPage', () => {
     mocks.getSessionMessagesPageQuery.mockResolvedValueOnce({
       kiloSessionId: 'ses_123',
       history: page,
+      watermarkEventId: 42,
     });
 
     const { fetchMobileSessionSnapshotPage } = await importAdapter();
@@ -104,6 +105,7 @@ describe('fetchMobileSessionSnapshotPage', () => {
       messages: page.messages,
       nextCursor: 'opaque-cursor',
       omittedItemCount: 2,
+      watermarkEventId: 42,
     });
     // No `cursor` was supplied, so the tRPC layer must see only `session_id`
     // and let its input schema default the bounded page size to 50.
@@ -239,4 +241,30 @@ describe('fetchMobileSessionSnapshotPage', () => {
     expect(call).toBeDefined();
     expect(call.limit).toBeUndefined();
   });
+
+  it('carries watermarkEventId through when the web router includes it', async () => {
+    mocks.getSessionMessagesPageQuery.mockResolvedValueOnce({
+      kiloSessionId: 'ses_123',
+      history: historyPage({ nextCursor: null }),
+      watermarkEventId: 77,
+    });
+
+    const { fetchMobileSessionSnapshotPage } = await importAdapter();
+    const result = await fetchMobileSessionSnapshotPage(kiloSessionId('ses_123'), {});
+
+    expect(result).toMatchObject({ kind: 'success', watermarkEventId: 77 });
+  });
+
+  it('omits watermarkEventId from the page when the web router does not include it', async () => {
+    mocks.getSessionMessagesPageQuery.mockResolvedValueOnce({
+      kiloSessionId: 'ses_123',
+      history: historyPage({ nextCursor: null }),
+    });
+
+    const { fetchMobileSessionSnapshotPage } = await importAdapter();
+    const result = await fetchMobileSessionSnapshotPage(kiloSessionId('ses_123'), {});
+
+    expect(result).toMatchObject({ kind: 'success' });
+    expect((result as Record<string, unknown>).watermarkEventId).toBeUndefined();
+  });
 });

apps/mobile/src/components/agents/mobile-session-page-adapter.ts

@@ -20,7 +20,8 @@ function isHistoryPage(history: KiloSdkMessageHistory): history is KiloSdkMessag
  *
  * Maps the shared `KiloSdkMessageHistory` union to the SDK's
  * `SessionSnapshotPageOutcome`:
- * - Page variant → success outcome with messages, cursor, and omitted count
+ * - Page variant → success outcome with messages, cursor, omitted count,
+ *   and the Cloud Agent event-log watermark
  * - Failure variants → passed through verbatim (retryable, too_large, invalid_data)
  * - Null history → empty success page (existing session with zero messages). A genuine
  *   NOT_FOUND is returned as a thrown/rejected TRPCError, never as `history === null`.
@@ -45,6 +46,7 @@ export async function fetchMobileSessionSnapshotPage(
       messages: [],
       nextCursor: null,
       omittedItemCount: 0,
+      watermarkEventId: result.watermarkEventId ?? undefined,
     };
   }
 
@@ -55,6 +57,7 @@ export async function fetchMobileSessionSnapshotPage(
       messages: history.messages as SessionSnapshotPage['messages'],
       nextCursor: history.nextCursor,
       omittedItemCount: history.omittedItemCount,
+      watermarkEventId: result.watermarkEventId ?? undefined,
     };
   }
 

apps/mobile/src/components/kilo-chat/conversation-list-screen.tsx

@@ -38,6 +38,7 @@ import {
 } from './hooks/use-conversations';
 import { useInstancePresence } from './hooks/use-instance-presence';
 import { useNowTicker } from './hooks/use-now-ticker';
+import { useAppActiveAndFocused } from './hooks/use-app-active-and-focused';
 
 type Props = {
   sandboxId: string;
@@ -105,6 +106,7 @@ export function ConversationListScreen({ sandboxId, sandboxLabel }: Props) {
   const { fontScale } = useWindowDimensions();
   const client = useKiloChatClient();
   const eventClient = useEventServiceClient();
+  const activeAndFocused = useAppActiveAndFocused();
   const listQuery = useConversations(client, sandboxId);
   const createConversation = useCreateConversation(client);
   const leaveConversation = useLeaveConversation(client);
@@ -137,7 +139,7 @@ export function ConversationListScreen({ sandboxId, sandboxLabel }: Props) {
   );
 
   useInstancePresence(sandboxId);
-  useBotStatus(client, eventClient, sandboxId);
+  useBotStatus(client, eventClient, sandboxId, activeAndFocused);
 
   function handleRowPress(conversationId: string) {
     void Haptics.selectionAsync();

apps/mobile/src/components/kilo-chat/conversation-screen.tsx

@@ -29,6 +29,7 @@ import { useMessageCacheUpdater, useMessages } from './hooks/use-messages';
 import { useNowTicker } from './hooks/use-now-ticker';
 import { useCurrentUserId } from './hooks/use-current-user-id';
 import { useKiloChatTokenError } from './kilo-chat-provider';
+import { useAppActiveAndFocused } from './hooks/use-app-active-and-focused';
 import {
   instanceOrgId,
   useAllKiloClawInstances,
@@ -56,6 +57,7 @@ export function ConversationScreen({
 }: Props) {
   const client = useKiloChatClient();
   const eventClient = useEventServiceClient();
+  const activeAndFocused = useAppActiveAndFocused();
   const router = useRouter();
   const currentUserId = useCurrentUserId();
   const tokenError = useKiloChatTokenError();
@@ -67,7 +69,7 @@ export function ConversationScreen({
   const { data: instances } = useAllKiloClawInstances();
   const currentInstance = instances?.find(instance => instance.sandboxId === sandboxId);
   const instanceStatus = instanceStatusQuery.data?.status ?? currentInstance?.status ?? null;
-  const botStatus = useBotStatus(client, eventClient, sandboxId);
+  const botStatus = useBotStatus(client, eventClient, sandboxId, activeAndFocused);
   const botPresence = botStatus ? { online: botStatus.online, lastAt: botStatus.at } : undefined;
   const hasAttachmentsCapability = botStatus?.capabilities?.includes('attachments') ?? false;
   const now = useNowTicker(10_000);

apps/mobile/src/components/kilo-chat/hooks/use-conversation-event-subscription.ts

@@ -29,7 +29,7 @@ export function useConversationEventSubscription(
     if (!conversationId) {
       return undefined;
     }
-    return eventService.onReconnect(() => {
+    return eventService.onResync(() => {
       void queryClient.invalidateQueries({ queryKey: messagesKey(conversationId) });
     });
   }, [eventService, queryClient, conversationId]);

apps/mobile/src/components/kilo-chat/kilo-chat-provider.mounted.test.tsx

@@ -0,0 +1,166 @@
+/* eslint-disable typescript-eslint/no-deprecated -- react-test-renderer is the DOM-free renderer used to mount React trees under vitest (same pattern as src/test/render-with-providers.tsx) */
+/* eslint-disable eslint/max-classes-per-file -- hoisted mock classes for two different packages */
+import { createElement, type ReactElement, type ReactNode } from 'react';
+import TestRenderer, { act } from 'react-test-renderer';
+import { describe, expect, it, vi } from 'vitest';
+
+import { KiloChatProvider } from './kilo-chat-provider';
+
+// ── Hoisted mocks ──────────────────────────────────────────────────────────
+
+const mocks = vi.hoisted(() => {
+  const acquire = vi.fn<() => Promise<void>>().mockResolvedValue(undefined);
+  const release = vi.fn();
+
+  // Must be a real class (not vi.fn()) because it is instantiated with `new`.
+  class MockEventServiceClient {
+    acquire = acquire;
+    release = release;
+  }
+
+  // eslint-disable-next-line typescript-eslint/no-extraneous-class -- hoisted mock stub for constructor
+  class MockKiloChatClient {}
+
+  const subscribeToKiloChatTokenResponses = vi.fn(() => vi.fn());
+  const clearKiloChatTokenCache = vi.fn();
+  const useKiloChatTokenGetter = vi.fn(() => vi.fn().mockResolvedValue('mock-token'));
+  const useKiloChatTokenResponseGetter = vi.fn(() =>
+    vi.fn().mockResolvedValue({ userId: 'test-user', token: 'mock-token' })
+  );
+
+  const useAppActiveAndFocused = vi.fn<() => boolean>();
+
+  return {
+    acquire,
+    release,
+    EventServiceClient: MockEventServiceClient,
+    KiloChatClient: MockKiloChatClient,
+    subscribeToKiloChatTokenResponses,
+    clearKiloChatTokenCache,
+    useKiloChatTokenGetter,
+    useKiloChatTokenResponseGetter,
+    useAppActiveAndFocused,
+  };
+});
+
+// ── Module mocks ───────────────────────────────────────────────────────────
+
+vi.mock('@kilocode/event-service', () => ({
+  EventServiceClient: mocks.EventServiceClient,
+}));
+
+vi.mock('@kilocode/kilo-chat', () => ({
+  KiloChatClient: mocks.KiloChatClient,
+}));
+
+vi.mock('@kilocode/kilo-chat-hooks', () => ({
+  KiloChatHooksProvider: ({ children }: { children: ReactNode; value: unknown }) =>
+    children as ReactElement,
+}));
+
+vi.mock('@/lib/config', () => ({
+  EVENT_SERVICE_URL: 'https://events.test',
+  KILO_CHAT_URL: 'https://chat.test',
+}));
+
+vi.mock('./hooks/use-app-active-and-focused', () => ({
+  useAppActiveAndFocused: mocks.useAppActiveAndFocused,
+}));
+
+vi.mock('./hooks/use-kilo-chat-token', () => ({
+  subscribeToKiloChatTokenResponses: mocks.subscribeToKiloChatTokenResponses,
+  clearKiloChatTokenCache: mocks.clearKiloChatTokenCache,
+  useKiloChatTokenGetter: mocks.useKiloChatTokenGetter,
+  useKiloChatTokenResponseGetter: mocks.useKiloChatTokenResponseGetter,
+}));
+
+// ── Flat render helper ─────────────────────────────────────────────────────
+
+async function mountProvider(): Promise<TestRenderer.ReactTestRenderer> {
+  const ref: { current: TestRenderer.ReactTestRenderer | undefined } = { current: undefined };
+  await act(async () => {
+    ref.current = TestRenderer.create(createElement(KiloChatProvider, null));
+    await Promise.resolve();
+  });
+  const created = ref.current;
+  if (!created) {
+    throw new Error('mountProvider: renderer was not created');
+  }
+  return created;
+}
+
+async function unmountProvider(renderer: TestRenderer.ReactTestRenderer): Promise<void> {
+  await act(async () => {
+    renderer.unmount();
+    await Promise.resolve();
+  });
+}
+
+// ── Tests ──────────────────────────────────────────────────────────────────
+
+describe('KiloChatProvider lifecycle', () => {
+  it('mounts active, calls acquire, unmounts and releases the hold', async () => {
+    mocks.useAppActiveAndFocused.mockReturnValue(true);
+    vi.clearAllMocks();
+
+    const renderer = await mountProvider();
+
+    expect(mocks.acquire).toHaveBeenCalledTimes(1);
+    expect(mocks.release).not.toHaveBeenCalled();
+
+    await unmountProvider(renderer);
+
+    // Active mount → cleanup must release the hold.
+    expect(mocks.release).toHaveBeenCalledTimes(1);
+  });
+
+  it('mounts inactive, calls neither acquire nor release', async () => {
+    mocks.useAppActiveAndFocused.mockReturnValue(false);
+    vi.clearAllMocks();
+
+    const renderer = await mountProvider();
+
+    // Inactive: acquires nothing, releases nothing — no unpaired release.
+    expect(mocks.acquire).not.toHaveBeenCalled();
+    expect(mocks.release).not.toHaveBeenCalled();
+
+    await unmountProvider(renderer);
+
+    // Cleanup: holding was false, so no release on unmount either.
+    expect(mocks.release).not.toHaveBeenCalled();
+  });
+
+  it('handles active → inactive → active transition without unmounting', async () => {
+    mocks.useAppActiveAndFocused.mockReturnValue(true);
+    vi.clearAllMocks();
+
+    const renderer = await mountProvider();
+    expect(mocks.acquire).toHaveBeenCalledTimes(1);
+    expect(mocks.release).not.toHaveBeenCalled();
+
+    // Transition to inactive — cleanup from previous active effect releases the hold.
+    mocks.useAppActiveAndFocused.mockReturnValue(false);
+    vi.clearAllMocks();
+    await act(async () => {
+      renderer.update(createElement(KiloChatProvider, null));
+      await Promise.resolve();
+    });
+    expect(mocks.release).toHaveBeenCalledTimes(1);
+    expect(mocks.acquire).not.toHaveBeenCalled();
+
+    // Transition back to active — reacquires.
+    mocks.useAppActiveAndFocused.mockReturnValue(true);
+    vi.clearAllMocks();
+    await act(async () => {
+      renderer.update(createElement(KiloChatProvider, null));
+      await Promise.resolve();
+    });
+    expect(mocks.acquire).toHaveBeenCalledTimes(1);
+    expect(mocks.release).not.toHaveBeenCalled();
+
+    // Unmount releases the reacquired hold.
+    vi.clearAllMocks();
+    await unmountProvider(renderer);
+    expect(mocks.release).toHaveBeenCalledTimes(1);
+  });
+});

apps/mobile/src/components/kilo-chat/kilo-chat-provider.tsx

@@ -6,6 +6,7 @@ import { KiloChatHooksProvider } from '@kilocode/kilo-chat-hooks';
 
 import { EVENT_SERVICE_URL, KILO_CHAT_URL } from '@/lib/config';
 
+import { useAppActiveAndFocused } from './hooks/use-app-active-and-focused';
 import {
   clearKiloChatTokenCache,
   subscribeToKiloChatTokenResponses,
@@ -41,6 +42,7 @@ export function KiloChatProvider({ children }: KiloChatProviderProps) {
   const [currentUserId, setCurrentUserId] = useState<string | null>(null);
   const [tokenError, setTokenError] = useState(false);
   const [retryCount, setRetryCount] = useState(0);
+  const activeAndFocused = useAppActiveAndFocused();
 
   const [value] = useState(() => {
     const eventService = new EventServiceClient({
@@ -64,11 +66,17 @@ export function KiloChatProvider({ children }: KiloChatProviderProps) {
   });
 
   useEffect(() => {
-    void value.eventService.connect();
+    let holding = false;
+    if (activeAndFocused) {
+      void value.eventService.acquire();
+      holding = true;
+    }
     return () => {
-      value.eventService.disconnect();
+      if (holding) {
+        value.eventService.release();
+      }
     };
-  }, [value]);
+  }, [value, activeAndFocused]);
 
   useEffect(() => {
     let cancelled = false;