Mobile Realtime Sync Research

apps/ade-cli/src/services/sync/syncHostService.ts

@@ -169,6 +169,8 @@ const DEFAULT_TERMINAL_HISTORY_PAGE_BYTES = 262_144;
 const MIN_TERMINAL_HISTORY_PAGE_BYTES = 4_096;
 const MAX_TERMINAL_HISTORY_PAGE_BYTES = 524_288;
 const PEER_BACKPRESSURE_BYTES = 4 * 1024 * 1024;
+export const SYNC_HOST_CHAT_ACTIVE_BACKGROUND_BACKPRESSURE_BYTES = 512 * 1024;
+export const SYNC_HOST_CHAT_ACTIVE_CHANGESET_BATCH_BYTES = 64 * 1024;
 const MOBILE_COMMAND_RESULT_CACHE_TTL_MS = 30 * 60 * 1000;
 const MOBILE_COMMAND_RESULT_CACHE_MAX_ENTRIES = 512;
 const CHANGESET_ACK_TIMEOUT_MS = 10_000;
@@ -543,6 +545,26 @@ export function syncHeartbeatMissLimitForPeerMetadata(metadata: Pick<SyncPeerMet
     : DEFAULT_SYNC_HEARTBEAT_MISS_LIMIT;
 }
 
+export function shouldDeferSyncHostBackgroundChangesForChat(args: {
+  subscribedChatSessionCount: number;
+  bufferedAmount: number;
+}): boolean {
+  return args.subscribedChatSessionCount > 0
+    && args.bufferedAmount >= SYNC_HOST_CHAT_ACTIVE_BACKGROUND_BACKPRESSURE_BYTES;
+}
+
+export function syncHostChangesetBatchOptionsForChat(args: {
+  subscribedChatSessionCount: number;
+  maxRows: number;
+  maxBytes: number;
+}): { maxRows?: number; maxBytes?: number } | undefined {
+  if (args.subscribedChatSessionCount <= 0) return undefined;
+  return {
+    maxRows: Math.min(args.maxRows, 64),
+    maxBytes: Math.min(args.maxBytes, SYNC_HOST_CHAT_ACTIVE_CHANGESET_BATCH_BYTES),
+  };
+}
+
 /**
  * Flushed transcript size in bytes, or null when the session has no
  * transcript (file missing / untracked). The transcript WriteStream buffers,
@@ -1497,12 +1519,18 @@ export function createSyncHostService(args: SyncHostServiceArgs) {
   });
 
   const pollTimer = setInterval(() => {
-    void pumpChanges().catch((error) => {
-      args.logger.warn("sync_host.poll_failed", { error: error instanceof Error ? error.message : String(error) });
-    });
-    void pumpChatEvents().catch((error) => {
-      args.logger.warn("sync_host.chat_poll_failed", { error: error instanceof Error ? error.message : String(error) });
-    });
+    void (async () => {
+      try {
+        await pumpChatEvents();
+      } catch (error) {
+        args.logger.warn("sync_host.chat_poll_failed", { error: error instanceof Error ? error.message : String(error) });
+      }
+      try {
+        await pumpChanges();
+      } catch (error) {
+        args.logger.warn("sync_host.poll_failed", { error: error instanceof Error ? error.message : String(error) });
+      }
+    })();
   }, pollIntervalMs);
   const heartbeatTimer = setInterval(() => {
     pruneExpiredPairFailures();
@@ -2290,6 +2318,17 @@ export function createSyncHostService(args: SyncHostServiceArgs) {
     return peer.ws.bufferedAmount >= PEER_BACKPRESSURE_BYTES;
   }
 
+  function shouldPrioritizeChatForPeer(peer: PeerState): boolean {
+    return peer.subscribedChatSessionIds.size > 0;
+  }
+
+  function shouldDeferBackgroundChangesForChat(peer: PeerState): boolean {
+    return shouldDeferSyncHostBackgroundChangesForChat({
+      subscribedChatSessionCount: peer.subscribedChatSessionIds.size,
+      bufferedAmount: peer.ws.bufferedAmount,
+    });
+  }
+
   function sendAndWait<TPayload>(
     ws: WebSocket,
     type: SyncEnvelope["type"],
@@ -2355,15 +2394,16 @@ export function createSyncHostService(args: SyncHostServiceArgs) {
     fromDbVersion: number,
     toDbVersion: number,
     changes: CrsqlChangeRow[],
+    options: { maxRows?: number; maxBytes?: number } = {},
   ): PendingChangesetBatch | null {
     const payload = buildChangesetBatchPayload({
       deviceId: peer.metadata?.deviceId ?? peer.pairedDeviceId ?? "peer",
       reason,
       fromDbVersion,
       toDbVersion,
       changes,
-      maxRows: maxChangesetBatchRows,
-      maxBytes: maxChangesetBatchBytes,
+      maxRows: options.maxRows ?? maxChangesetBatchRows,
+      maxBytes: options.maxBytes ?? maxChangesetBatchBytes,
     });
     if (!payload) return null;
     const batch: PendingChangesetBatch = {
@@ -2834,6 +2874,14 @@ export function createSyncHostService(args: SyncHostServiceArgs) {
     for (const peer of peers) {
       if (!peer.authenticated || !peer.metadata || peer.ws.readyState !== WebSocket.OPEN) continue;
       if (isPeerBackpressured(peer)) continue;
+      if (shouldDeferBackgroundChangesForChat(peer)) {
+        args.logger.debug("sync_host.changeset_deferred_chat_backpressure", {
+          peerDeviceId: peer.metadata?.deviceId ?? null,
+          bufferedAmount: peer.ws.bufferedAmount,
+          thresholdBytes: SYNC_HOST_CHAT_ACTIVE_BACKGROUND_BACKPRESSURE_BYTES,
+        });
+        continue;
+      }
       if (peer.pendingChangesetBatch) {
         if (nowMs - peer.pendingChangesetBatch.sentAtMs >= CHANGESET_ACK_TIMEOUT_MS) {
           const pending = peer.pendingChangesetBatch;
@@ -2901,7 +2949,18 @@ export function createSyncHostService(args: SyncHostServiceArgs) {
         });
         continue;
       }
-      const pending = sendNextChangesetBatch(peer, "broadcast", peer.lastKnownServerDbVersion, exportedThroughDbVersion, changes);
+      const pending = sendNextChangesetBatch(
+        peer,
+        "broadcast",
+        peer.lastKnownServerDbVersion,
+        exportedThroughDbVersion,
+        changes,
+        syncHostChangesetBatchOptionsForChat({
+          subscribedChatSessionCount: peer.subscribedChatSessionIds.size,
+          maxRows: maxChangesetBatchRows,
+          maxBytes: maxChangesetBatchBytes,
+        }),
+      );
       if (pending) {
         if (peerSupportsChangesetAck(peer)) {
           peer.pendingChangesetBatch = pending;

apps/ade-cli/src/services/sync/syncRemoteCommandService.ts

@@ -435,6 +435,9 @@ function parseListLanesArgs(value: Record<string, unknown>): ListLanesArgs {
   return {
     includeArchived: asOptionalBoolean(value.includeArchived),
     includeStatus: asOptionalBoolean(value.includeStatus),
+    includeConflictStatus: asOptionalBoolean(value.includeConflictStatus),
+    includeRebaseSuggestions: asOptionalBoolean(value.includeRebaseSuggestions),
+    includeAutoRebaseStatus: asOptionalBoolean(value.includeAutoRebaseStatus),
   };
 }
 
@@ -1866,13 +1869,20 @@ function summarizeLaneRuntime(
 async function buildLaneListSnapshots(
   args: SyncRemoteCommandServiceArgs,
   lanes: Awaited<ReturnType<ReturnType<typeof createLaneService>["list"]>>,
+  options: ListLanesArgs = {},
 ): Promise<LaneListSnapshot[]> {
   const [sessions, rebaseSuggestions, autoRebaseStatuses, stateSnapshots, batchAssessment] = await Promise.all([
     Promise.resolve(args.sessionService.list({ limit: 500 })),
-    Promise.resolve(args.rebaseSuggestionService?.listSuggestions() ?? []),
-    Promise.resolve(args.autoRebaseService?.listStatuses() ?? []),
+    options.includeRebaseSuggestions === false
+      ? Promise.resolve([])
+      : Promise.resolve(args.rebaseSuggestionService?.listSuggestions({ lanes }) ?? []),
+    options.includeAutoRebaseStatus === false
+      ? Promise.resolve([])
+      : Promise.resolve(args.autoRebaseService?.listStatuses({ lanes }) ?? []),
     Promise.resolve(args.laneService.listStateSnapshots()),
-    args.conflictService?.getBatchAssessment({ lanes }).catch(() => null) ?? Promise.resolve(null),
+    options.includeConflictStatus === false
+      ? Promise.resolve(null)
+      : args.conflictService?.getBatchAssessment({ lanes }).catch(() => null) ?? Promise.resolve(null),
   ]);
 
   const rebaseByLaneId = new Map(rebaseSuggestions.map((entry) => [entry.laneId, entry] as const));
@@ -1892,7 +1902,7 @@ async function buildLaneListSnapshots(
 }
 
 async function buildLaneDetailPayload(args: SyncRemoteCommandServiceArgs, laneId: string): Promise<LaneDetailPayload> {
-  const lane = (await args.laneService.list({ includeArchived: true, includeStatus: true })).find((entry) => entry.id === laneId) ?? null;
+  const lane = await args.laneService.getSummary(laneId, { includeStatus: true });
   if (!lane) throw new Error(`Lane not found: ${laneId}`);
 
   const [
@@ -1916,8 +1926,8 @@ async function buildLaneDetailPayload(args: SyncRemoteCommandServiceArgs, laneId
     args.laneService.getChildren(laneId),
     Promise.resolve(args.sessionService.list({ laneId, limit: 200 })),
     args.agentChatService?.listSessions(laneId, { includeAutomation: true }) ?? Promise.resolve([]),
-    Promise.resolve(args.rebaseSuggestionService?.listSuggestions() ?? []),
-    Promise.resolve(args.autoRebaseService?.listStatuses() ?? []),
+    Promise.resolve(args.rebaseSuggestionService?.listSuggestions({ lanes: [lane] }) ?? []),
+    Promise.resolve(args.autoRebaseService?.listStatuses({ lanes: [lane] }) ?? []),
     Promise.resolve(args.laneService.getStateSnapshot(laneId)),
     args.gitService?.listRecentCommits({ laneId, limit: 20 }) ?? Promise.resolve([]),
     args.diffService?.getChanges(laneId).catch(() => null) ?? Promise.resolve(null),
@@ -1965,10 +1975,11 @@ type RemoteCommandRegistrationDeps = {
 function registerLaneRemoteCommands({ args, register }: RemoteCommandRegistrationDeps): void {
   register("lanes.list", { viewerAllowed: true }, async (payload) => args.laneService.list(parseListLanesArgs(payload)));
   register("lanes.refreshSnapshots", { viewerAllowed: true }, async (payload) => {
-    const refreshed = await args.laneService.refreshSnapshots(parseListLanesArgs(payload));
+    const listArgs = parseListLanesArgs(payload);
+    const refreshed = await args.laneService.refreshSnapshots(listArgs);
     return {
       ...refreshed,
-      snapshots: await buildLaneListSnapshots(args, refreshed.lanes),
+      snapshots: await buildLaneListSnapshots(args, refreshed.lanes, listArgs),
     };
   });
   register("lanes.getDetail", { viewerAllowed: true }, async (payload) =>

apps/desktop/src/main/services/lanes/laneService.ts

@@ -2911,6 +2911,66 @@ export function createLaneService({
       return await listLanes(args);
     },
 
+    async getSummary(laneId: string, options: { includeStatus?: boolean } = {}): Promise<LaneSummary | null> {
+      const row = getLaneRow(laneId);
+      if (!row) return null;
+
+      const rowsById = getRowsById(true);
+      const parent = row.parent_lane_id ? rowsById.get(row.parent_lane_id) ?? null : null;
+      let status: LaneStatus = cloneLaneStatus(DEFAULT_LANE_STATUS);
+      let parentStatus: LaneStatus | null = parent ? cloneLaneStatus(DEFAULT_LANE_STATUS) : null;
+      let worktreeAvailable: boolean | undefined;
+
+      if (options.includeStatus !== false) {
+        try {
+          worktreeAvailable = await isExpectedGitWorktreeRoot(row.worktree_path);
+          if (worktreeAvailable) {
+            status = await computeLaneStatus(
+              row.worktree_path,
+              rowTracksParent(row, parent) ? parent?.branch_ref ?? row.base_ref : row.base_ref,
+              row.branch_ref,
+            );
+          }
+        } catch {
+          logger.warn("laneService.getSummary.status_failed", { laneId });
+          worktreeAvailable = false;
+          status = cloneLaneStatus(DEFAULT_LANE_STATUS);
+        }
+
+        if (parent) {
+          try {
+            const grandParent = parent.parent_lane_id ? rowsById.get(parent.parent_lane_id) ?? null : null;
+            parentStatus = await computeLaneStatus(
+              parent.worktree_path,
+              rowTracksParent(parent, grandParent) ? grandParent?.branch_ref ?? parent.base_ref : parent.base_ref,
+              parent.branch_ref,
+            );
+          } catch {
+            logger.warn("laneService.getSummary.parent_status_failed", { laneId, parentLaneId: parent.id });
+            parentStatus = cloneLaneStatus(DEFAULT_LANE_STATUS);
+          }
+        }
+
+        upsertLaneStateSnapshot({
+          laneId: row.id,
+          status,
+          updatedAt: new Date().toISOString(),
+        });
+      }
+
+      return toLaneSummary({
+        row,
+        status,
+        parentStatus,
+        childCount: getChildrenRows(row.id, false).length,
+        stackDepth: computeStackDepth({ laneId: row.id, rowsById, memo: new Map() }),
+        worktreeAvailable,
+        activeBranchProfile: ensureBranchProfileForRow(row),
+        linearIssue: getLaneLinearIssue(row.id),
+        linearIssueLinks: getLaneLinearIssueLinks(row.id),
+      });
+    },
+
     async listUnregisteredWorktrees(): Promise<UnregisteredLaneCandidate[]> {
       return listUnregisteredWorktreeCandidates();
     },
@@ -2955,7 +3015,7 @@ export function createLaneService({
       invalidateLaneListCache();
       const summaries = await listLanes({
         includeArchived: args.includeArchived ?? true,
-        includeStatus: true,
+        includeStatus: args.includeStatus ?? true,
       });
       return {
         refreshedCount: summaries.length,

apps/desktop/src/main/services/sync/syncHostService.test.ts

@@ -10,7 +10,11 @@ import { isCrsqliteAvailable } from "../state/crsqliteExtension";
 import {
   createSyncHostService,
   parseNativeLanDiscoveryProcessList,
+  shouldDeferSyncHostBackgroundChangesForChat,
+  syncHostChangesetBatchOptionsForChat,
   syncHeartbeatMissLimitForPeerMetadata,
+  SYNC_HOST_CHAT_ACTIVE_BACKGROUND_BACKPRESSURE_BYTES,
+  SYNC_HOST_CHAT_ACTIVE_CHANGESET_BATCH_BYTES,
 } from "./syncHostService";
 import type { SyncPinStore } from "./syncPinStore";
 import { encodeSyncEnvelope, parseSyncEnvelope } from "./syncProtocol";
@@ -313,6 +317,43 @@ it("allows a wider heartbeat grace window for mobile peers", () => {
   );
 });
 
+it("prioritizes active chat by deferring and slicing background sync batches", () => {
+  expect(shouldDeferSyncHostBackgroundChangesForChat({
+    subscribedChatSessionCount: 0,
+    bufferedAmount: SYNC_HOST_CHAT_ACTIVE_BACKGROUND_BACKPRESSURE_BYTES,
+  })).toBe(false);
+  expect(shouldDeferSyncHostBackgroundChangesForChat({
+    subscribedChatSessionCount: 1,
+    bufferedAmount: SYNC_HOST_CHAT_ACTIVE_BACKGROUND_BACKPRESSURE_BYTES - 1,
+  })).toBe(false);
+  expect(shouldDeferSyncHostBackgroundChangesForChat({
+    subscribedChatSessionCount: 1,
+    bufferedAmount: SYNC_HOST_CHAT_ACTIVE_BACKGROUND_BACKPRESSURE_BYTES,
+  })).toBe(true);
+
+  expect(syncHostChangesetBatchOptionsForChat({
+    subscribedChatSessionCount: 0,
+    maxRows: 250,
+    maxBytes: 256 * 1024,
+  })).toBeUndefined();
+  expect(syncHostChangesetBatchOptionsForChat({
+    subscribedChatSessionCount: 1,
+    maxRows: 250,
+    maxBytes: 256 * 1024,
+  })).toEqual({
+    maxRows: 64,
+    maxBytes: SYNC_HOST_CHAT_ACTIVE_CHANGESET_BATCH_BYTES,
+  });
+  expect(syncHostChangesetBatchOptionsForChat({
+    subscribedChatSessionCount: 1,
+    maxRows: 12,
+    maxBytes: 16 * 1024,
+  })).toEqual({
+    maxRows: 12,
+    maxBytes: 16 * 1024,
+  });
+});
+
 it("parses ADE dns-sd discovery processes for orphan recovery", () => {
   const stdout = [
     " 111 1 dns-sd -R ADE Sync lappy 8788 _ade-sync._tcp local 8788 version=1",

apps/desktop/src/main/services/sync/syncRemoteCommandService.test.ts

@@ -176,6 +176,7 @@ function createLogger() {
 function createMockLaneService() {
   return {
     list: vi.fn().mockResolvedValue([]),
+    getSummary: vi.fn().mockResolvedValue(null),
     refreshSnapshots: vi.fn().mockResolvedValue({ lanes: [] }),
     create: vi.fn().mockResolvedValue({ id: "lane-1" }),
     createChild: vi.fn().mockResolvedValue({ id: "child-1" }),
@@ -785,9 +786,48 @@ describe("createSyncRemoteCommandService", () => {
       expect(laneService.list).toHaveBeenCalledWith({
         includeArchived: true,
         includeStatus: undefined,
+        includeConflictStatus: undefined,
+        includeRebaseSuggestions: undefined,
+        includeAutoRebaseStatus: undefined,
       });
     });
 
+    it("lanes.refreshSnapshots forwards light-refresh flags and skips expensive decorations", async () => {
+      const lane = { id: "lane-1", name: "Lane one" };
+      laneService.refreshSnapshots.mockResolvedValue({ lanes: [lane] });
+
+      const result = await service.execute(makePayload("lanes.refreshSnapshots", {
+        includeArchived: true,
+        includeStatus: false,
+        includeConflictStatus: false,
+        includeRebaseSuggestions: false,
+        includeAutoRebaseStatus: false,
+      })) as { snapshots: Array<{ lane: unknown }> };
+
+      expect(laneService.refreshSnapshots).toHaveBeenCalledWith({
+        includeArchived: true,
+        includeStatus: false,
+        includeConflictStatus: false,
+        includeRebaseSuggestions: false,
+        includeAutoRebaseStatus: false,
+      });
+      expect(conflictService.getBatchAssessment).not.toHaveBeenCalled();
+      expect(rebaseSuggestionService.listSuggestions).not.toHaveBeenCalled();
+      expect(result.snapshots.map((entry) => entry.lane)).toEqual([lane]);
+    });
+
+    it("lanes.getDetail loads only the requested lane summary", async () => {
+      const lane = { id: "lane-1", name: "Lane one" };
+      laneService.getSummary.mockResolvedValue(lane);
+
+      const result = await service.execute(makePayload("lanes.getDetail", { laneId: "lane-1" })) as { lane: unknown };
+
+      expect(laneService.getSummary).toHaveBeenCalledWith("lane-1", { includeStatus: true });
+      expect(laneService.list).not.toHaveBeenCalled();
+      expect(rebaseSuggestionService.listSuggestions).toHaveBeenCalledWith({ lanes: [lane] });
+      expect(result.lane).toBe(lane);
+    });
+
     it("lanes.create parses name and routes to laneService.create", async () => {
       await service.execute(makePayload("lanes.create", {
         name: "my-lane",
@@ -2527,6 +2567,9 @@ describe("createSyncRemoteCommandService", () => {
       expect(laneService.list).toHaveBeenCalledWith({
         includeArchived: undefined,
         includeStatus: undefined,
+        includeConflictStatus: undefined,
+        includeRebaseSuggestions: undefined,
+        includeAutoRebaseStatus: undefined,
       });
     });