perf(mobile): faster thread sync, connects, and reconnects on bad wifi

apps/mobile/src/connection/runtime.ts

@@ -36,7 +36,11 @@ type ConnectionLayerSource =
 
 const layerProvidedClientConnection = layerSnapshotLoader.pipe(
   Layer.provideMerge(
-    Connection.layerWithOptions({ usageLimitSources: true, usageLimitsCommand: true }),
+    Connection.layerWithOptions({
+      usageLimitSources: true,
+      usageLimitsCommand: true,
+      providerCatalogsOnDemand: true,
+    }),
   ),
   Layer.provideMerge(
     Layer.mergeAll(

apps/mobile/src/state/server.ts

@@ -13,6 +13,7 @@ export const serverEnvironment = createServerEnvironmentAtoms(connectionAtomRunt
   initialConfigValueAtom: environmentSession.initialConfigValueAtom,
   usageLimitSources: true,
   usageLimitsCommand: true,
+  providerCatalogsOnDemand: true,
 });
 export const describeOutdatedServer = createOutdatedServerDescriptionCommand(connectionAtomRuntime);
 export const updateOutdatedServer = createOutdatedServerUpdateCommand(connectionAtomRuntime);

apps/server/src/terminal/OutputProtocol.test.ts

@@ -1,5 +1,6 @@
 import { assert, describe, it } from "@effect/vitest";
 import {
+  ORCHESTRATION_V2_WS_METHODS,
   type TerminalAttachStreamEvent,
   TerminalAttachStreamEvent as TerminalAttachStreamEventSchema,
   WS_METHODS,
@@ -13,14 +14,17 @@ import * as Stream from "effect/Stream";
 import * as TestClock from "effect/testing/TestClock";
 import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/rpc";
 
-import { terminalOutputStream, withTerminalOutputWindow } from "./OutputProtocol.ts";
+import { terminalOutputStream, withOutputWindow } from "./OutputProtocol.ts";
 
 describe("terminal output window", () => {
   it.effect.each([
     { tag: WS_METHODS.terminalAttach, size: 1, limit: 8 },
     { tag: WS_METHODS.terminalObserve, size: 1, limit: 8 },
     { tag: WS_METHODS.terminalAttach, size: 64 * 1024, limit: 1 },
     { tag: WS_METHODS.terminalObserve, size: 64 * 1024, limit: 1 },
+    { tag: WS_METHODS.subscribeServerConfig, size: 1, limit: 8 },
+    { tag: ORCHESTRATION_V2_WS_METHODS.subscribeShell, size: 1, limit: 8 },
+    { tag: ORCHESTRATION_V2_WS_METHODS.subscribeThread, size: 64 * 1024, limit: 8 },
     { tag: WS_METHODS.subscribeTerminalMetadata, size: 1, limit: 1 },
   ])("limits $tag with $size-byte values to $limit pending chunks", ({ tag, size, limit }) =>
     Effect.gen(function* () {
@@ -48,7 +52,7 @@ describe("terminal output window", () => {
       );
       yield* RpcServer.make(group).pipe(
         Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))),
-        Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
+        Effect.provideService(RpcServer.Protocol, withOutputWindow(protocol)),
         Effect.forkScoped,
       );
       const write = yield* Deferred.await(receive);
@@ -191,7 +195,7 @@ describe("terminal output stream", () => {
               ),
             ),
           ),
-          Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
+          Effect.provideService(RpcServer.Protocol, withOutputWindow(protocol)),
           Effect.forkScoped,
         );
         const write = yield* Deferred.await(receive);

apps/server/src/terminal/OutputProtocol.ts

@@ -1,4 +1,9 @@
-import { type TerminalAttachStreamEvent, type TerminalEvent, WS_METHODS } from "@t3tools/contracts";
+import {
+  ORCHESTRATION_V2_WS_METHODS,
+  type TerminalAttachStreamEvent,
+  type TerminalEvent,
+  WS_METHODS,
+} from "@t3tools/contracts";
 import type * as Arr from "effect/Array";
 import * as Cause from "effect/Cause";
 import * as Effect from "effect/Effect";
@@ -11,31 +16,62 @@ const MAX_PENDING_BYTES = 64 * 1024;
 const isFull = (sizes: number[]) =>
   sizes.length >= MAX_PENDING_CHUNKS ||
   sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES;
+// Terminal output is small chunks, so its window also caps bytes. The other streams
+// window by count only: sizing a multi-MB thread snapshot would serialize it twice.
+const BYTE_WINDOWED_STREAMS: ReadonlySet<string> = new Set([
+  WS_METHODS.terminalAttach,
+  WS_METHODS.terminalObserve,
+]);
+// One ACK round trip per chunk stalls sync on high-latency links.
+const COUNT_WINDOWED_STREAMS: ReadonlySet<string> = new Set([
+  WS_METHODS.subscribeServerConfig,
+  ORCHESTRATION_V2_WS_METHODS.subscribeShell,
+  ORCHESTRATION_V2_WS_METHODS.subscribeThread,
+]);
 
-export function withTerminalOutputWindow(
+interface Window {
+  readonly bytes: boolean;
+  readonly sizes: number[];
+}
+
+export function withOutputWindow(
   protocol: RpcServer.Protocol["Service"],
 ): RpcServer.Protocol["Service"] {
-  const windows = new Map<string, number[]>();
+  // RpcServer drops a disconnected client before its streams exit, so their Exit never
+  // reaches send; departed clients are pruned when a window opens instead.
+  const clients = new Map<number, Map<string, Window>>();
   let receive: Parameters<RpcServer.Protocol["Service"]["run"]>[0];
+  const windowFor = (clientId: number, requestId: string) => clients.get(clientId)?.get(requestId);
+  const openWindow = Effect.fnUntraced(function* (
+    clientId: number,
+    requestId: string,
+    bytes: boolean,
+  ) {
+    const connected = yield* protocol.clientIds;
+    for (const id of clients.keys()) if (!connected.has(id)) clients.delete(id);
+    let windows = clients.get(clientId);
+    if (!windows) clients.set(clientId, (windows = new Map()));
+    if (!windows.has(requestId)) windows.set(requestId, { bytes, sizes: [] });
+  });
   return {
     ...protocol,
     run: (write) => {
       receive = write;
       return protocol.run((clientId, message) =>
         Effect.suspend(() => {
-          if (
-            message._tag === "Request" &&
-            (message.tag === WS_METHODS.terminalAttach ||
-              message.tag === WS_METHODS.terminalObserve)
-          ) {
-            const key = `${clientId}:${message.id}`;
-            if (!windows.has(key)) windows.set(key, []);
+          if (message._tag === "Request") {
+            const bytes = BYTE_WINDOWED_STREAMS.has(message.tag);
+            if (bytes || COUNT_WINDOWED_STREAMS.has(message.tag)) {
+              return openWindow(clientId, String(message.id), bytes).pipe(
+                Effect.andThen(write(clientId, message)),
+              );
+            }
           } else if (message._tag === "Ack") {
-            const window = windows.get(`${clientId}:${message.requestId}`);
+            const window = windowFor(clientId, String(message.requestId));
             if (window) {
-              const wasFull = isFull(window);
-              window.shift();
-              if (!wasFull || isFull(window)) return Effect.void;
+              const wasFull = isFull(window.sizes);
+              window.sizes.shift();
+              if (!wasFull || isFull(window.sizes)) return Effect.void;
             }
           }
           return write(clientId, message);
@@ -46,14 +82,15 @@ export function withTerminalOutputWindow(
       Effect.suspend(() => {
         const send = protocol.send(clientId, response, transferables);
         if (response._tag === "Exit") {
-          windows.delete(`${clientId}:${response.requestId}`);
+          clients.get(clientId)?.delete(String(response.requestId));
         } else if (response._tag === "Chunk") {
-          const window = windows.get(`${clientId}:${response.requestId}`);
+          const window = windowFor(clientId, String(response.requestId));
           if (window) {
-            // @effect-diagnostics-next-line globalJsonInEffect:off
-            const size = Buffer.byteLength(JSON.stringify(response));
-            window.push(size);
-            if (!isFull(window)) {
+            window.sizes.push(
+              // @effect-diagnostics-next-line globalJsonInEffect:off
+              window.bytes ? Buffer.byteLength(JSON.stringify(response)) : 0,
+            );
+            if (!isFull(window.sizes)) {
               return send.pipe(
                 Effect.andThen(() =>
                   receive(clientId, {

apps/server/src/ws.test.ts

@@ -116,6 +116,7 @@ import * as WorkspacePaths from "./workspace/WorkspacePaths.ts";
 import * as ServerSecretStore from "./auth/ServerSecretStore.ts";
 import * as ServerConfigModule from "./config.ts";
 import * as Keybindings from "./keybindings.ts";
+import { USAGE_LIMITS_COMMAND } from "@t3tools/shared/usageLimits";
 import * as ServerSettings from "./serverSettings.ts";
 import * as ManagedPolicy from "./policy/ManagedPolicy.ts";
 
@@ -1024,13 +1025,21 @@ it.effect("on-demand clients get provider config without workspace snapshots", (
       auth: { status: "authenticated" },
       checkedAt: "2026-10-25T00:00:00.000Z",
       models: [],
-      slashCommands: [{ name: "machine-command" }],
-      skills: [],
+      slashCommands: [{ name: "machine-command", description: "A machine command" }],
+      skills: [
+        {
+          name: "machine-skill",
+          description: "Machine skill",
+          path: "/skills/machine-skill",
+          enabled: true,
+        },
+      ],
       workspaceSnapshots: [
         snapshotFor("/a", "2026-10-25T00:00:00.000Z"),
         snapshotFor("/b", "2026-10-25T00:00:00.000Z"),
       ],
     };
+    let scansCwd = true;
     const providerChanges = yield* Queue.unbounded<ReadonlyArray<ServerProvider>>();
     const pulled = yield* Queue.unbounded<void>();
     const responses = yield* Queue.unbounded<unknown>();
@@ -1047,6 +1056,20 @@ it.effect("on-demand clients get provider config without workspace snapshots", (
         ),
         refreshWorkspaceSnapshot: () => Effect.succeed([provider]),
       }),
+      Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({
+        listInstances: Effect.sync(() => [
+          (scansCwd
+            ? { instanceId, snapshotForCwd: () => Effect.die("unused") }
+            : { instanceId }) as never,
+        ]),
+        getInstance: () =>
+          Effect.sync(
+            () =>
+              (scansCwd
+                ? { instanceId, snapshotForCwd: () => Effect.die("unused") }
+                : { instanceId }) as never,
+          ),
+      }),
       Layer.mock(UsageLimitSources.UsageLimitSources)({
         current: Effect.succeed([]),
         streamChanges: Stream.make([]),
@@ -1116,12 +1139,17 @@ it.effect("on-demand clients get provider config without workspace snapshots", (
       _tag: "Request",
       id: configRequest,
       tag: WS_METHODS.subscribeServerConfig,
-      payload: { workspaceSnapshotsOnDemand: true },
+      payload: { workspaceSnapshotsOnDemand: true, providerCatalogsOnDemand: true },
       headers: Headers.empty,
     });
     const snapshot = (yield* Queue.take(responses)) as ServerConfigStreamEvent;
     if (snapshot.type !== "snapshot") return yield* Effect.die("Missing initial snapshot");
     assert.isUndefined(snapshot.config.providers[0]?.workspaceSnapshots);
+    // The cwd snapshot the client fetches carries the catalog, so only command names are sent.
+    assert.deepEqual(snapshot.config.providers[0]?.slashCommands, [{ name: "machine-command" }]);
+    assert.deepEqual(snapshot.config.providers[0]?.skills, [
+      { name: "machine-skill", path: "/skills/machine-skill", enabled: true },
+    ]);
     yield* server.write(0, { _tag: "Ack", requestId: configRequest });
 
     // A rescan that only touches a snapshot sends nothing; the next real change does.
@@ -1130,13 +1158,34 @@ it.effect("on-demand clients get provider config without workspace snapshots", (
     ]);
     yield* Queue.take(pulled);
     yield* TestClock.adjust("1 second");
-    yield* Queue.offer(providerChanges, [{ ...provider, status: "warning" }]);
+    // Late /usage-limits discovery still reaches the client, so it refetches its cwd snapshot.
+    yield* Queue.offer(providerChanges, [
+      {
+        ...provider,
+        status: "warning",
+        slashCommands: [...provider.slashCommands, USAGE_LIMITS_COMMAND],
+      },
+    ]);
     yield* Queue.take(pulled);
     yield* TestClock.adjust("1 second");
     const update = (yield* Queue.take(responses)) as ServerConfigStreamEvent;
     if (update.type !== "providerStatuses") return yield* Effect.die("Expected provider update");
     assert.equal(update.payload.providers[0]?.status, "warning");
     assert.isUndefined(update.payload.providers[0]?.workspaceSnapshots);
+    assert.deepEqual(update.payload.providers[0]?.slashCommands, [
+      { name: "machine-command" },
+      { name: USAGE_LIMITS_COMMAND.name },
+    ]);
+    yield* server.write(0, { _tag: "Ack", requestId: configRequest });
+    // A replacement that cannot scan a cwd sends its full machine catalog.
+    scansCwd = false;
+    yield* Queue.offer(providerChanges, [{ ...provider, status: "warning" }]);
+    yield* Queue.take(pulled);
+    yield* TestClock.adjust("1 second");
+    const replaced = (yield* Queue.take(responses)) as ServerConfigStreamEvent;
+    if (replaced.type !== "providerStatuses") return yield* Effect.die("Expected provider update");
+    assert.deepEqual(replaced.payload.providers[0]?.slashCommands, provider.slashCommands);
+    assert.deepEqual(replaced.payload.providers[0]?.skills, provider.skills);
 
     yield* server.write(0, {
       _tag: "Request",
@@ -1152,6 +1201,32 @@ it.effect("on-demand clients get provider config without workspace snapshots", (
       refreshed.providers.map((entry) => [entry.workspaceSnapshots, entry.slashCommands]),
       [[[snapshotFor("/b", "2026-10-25T00:00:00.000Z")], []]],
     );
+    // A failed scan still gives a client without the machine catalog its commands.
+    scansCwd = true;
+    yield* server.write(0, {
+      _tag: "Request",
+      id: RpcMessage.RequestId("refresh-failed-scan"),
+      tag: WS_METHODS.serverRefreshProviders,
+      payload: {
+        instanceId,
+        cwd: "/c",
+        workspaceSnapshotsOnDemand: true,
+        providerCatalogsOnDemand: true,
+      },
+      headers: Headers.empty,
+    });
+    const fallback = (yield* Queue.take(responses)) as {
+      readonly providers: ReadonlyArray<ServerProvider>;
+    };
+    assert.deepEqual(fallback.providers[0]?.workspaceSnapshots, [
+      {
+        cwd: "/c",
+        checkedAt: provider.checkedAt,
+        slashCommands: provider.slashCommands,
+        slashCommandsPending: true,
+        skills: provider.skills,
+      },
+    ]);
     yield* server.write(0, { _tag: "Interrupt", requestId: configRequest, interruptors: [] });
   }).pipe(
     Effect.provide(