fix(remote): isolate shared control request timeouts (#9016)

Co-authored-by: Orca <help@stably.ai>

src/shared/remote-runtime-shared-control-connection.test.ts

@@ -454,32 +454,37 @@ describe('RemoteRuntimeSharedControlConnection', () => {
     connection.close()
   })
 
-  it('tears down the socket when a request times out and reconnects active subscriptions', async () => {
-    const server = await createServer({ silentMethods: ['worktree.hang'] })
-    const connection = new RemoteRuntimeSharedControlConnection(server.pairing)
-    const onResponse = vi.fn()
-    const onClose = vi.fn()
-
-    await connection.subscribe('runtime.clientEvents.subscribe', null, 1000, {
-      onResponse,
-      onError: vi.fn(),
-      onClose
+  it('keeps unrelated pending requests alive when one request times out', async () => {
+    const server = await createServer({
+      silentMethods: ['worktree.hang'],
+      delayedMethods: ['worktree.ps']
     })
-    await vi.waitFor(() => expect(onResponse).toHaveBeenCalled())
+    const connection = new RemoteRuntimeSharedControlConnection(server.pairing)
 
-    // Why: mirrors RemoteRuntimeRequestConnection — a request the server never
-    // answered marks the socket as suspect and must not leave it 'ready'.
-    await expect(connection.request('worktree.hang', undefined, 50)).rejects.toThrow('Timed out')
-
-    await vi.waitFor(() => expect(server.connectionCount()).toBe(2), { timeout: 5000 })
-    await vi.waitFor(
-      () =>
-        expect(
-          server.requests.filter((request) => request.method === 'runtime.clientEvents.subscribe')
-        ).toHaveLength(2),
-      { timeout: 5000 }
+    const timedOut = connection.request('worktree.hang', undefined, 250)
+    void timedOut.catch(() => undefined)
+    await vi.waitFor(() =>
+      expect(server.requests.map(({ method }) => method)).toContain('worktree.hang')
     )
-    expect(onClose).not.toHaveBeenCalled()
+    const survivor = connection.request('worktree.ps', undefined, 1000).then(
+      (response) => ({ ok: true as const, response }),
+      (error: unknown) => ({ ok: false as const, error })
+    )
+    await vi.waitFor(() =>
+      expect(server.requests.map(({ method }) => method)).toContain('worktree.ps')
+    )
+
+    await expect(timedOut).rejects.toThrow('Timed out')
+    // Why: a single slow method is not evidence that a shared socket is dead;
+    // liveness monitoring owns connection-wide failure detection.
+    expect(connection.getDiagnostics()).toMatchObject({ state: 'ready', pendingRequestCount: 1 })
+
+    server.flushDelayedResponses()
+    await expect(survivor).resolves.toMatchObject({
+      ok: true,
+      response: { ok: true, result: { method: 'worktree.ps' } }
+    })
+    expect(server.connectionCount()).toBe(1)
 
     connection.close()
   })
@@ -515,6 +520,7 @@ async function createServer(
     // Why: half-open simulation — the socket stays open but never answers
     // protocol pings, like a wedged tunnel that swallows frames silently.
     disableAutoPong?: boolean
+    delayedMethods?: string[]
     silentMethods?: string[]
   } = {}
 ): Promise<TestServer> {
@@ -613,6 +619,7 @@ function handleRequest(
     sendUnknownResponseBeforeResponse?: boolean
     closeAfterStreamingResponse?: () => boolean
     closeBeforeResponse?: boolean
+    delayedMethods?: string[]
     silentMethods?: string[]
   },
   delayedResponses: (() => void)[]
@@ -665,6 +672,10 @@ function handleRequest(
     delayedResponses.push(sendResponse)
     return
   }
+  if (options.delayedMethods?.includes(request.method)) {
+    delayedResponses.push(sendResponse)
+    return
+  }
   if (options.responseDelayMs !== undefined) {
     setTimeout(() => {
       sendResponse()

src/shared/remote-runtime-shared-control-connection.ts

@@ -74,10 +74,7 @@ export class RemoteRuntimeSharedControlConnection {
       timeoutMs,
       ensureReady: () => this.ensureReadyWithTimeout(timeoutMs),
       send: (requestId, requestMethod, requestParams) =>
-        this.sendRequest(requestId, requestMethod, requestParams),
-      // Why: a timed-out request marks the socket as suspect (#7718) — tear
-      // it down so reconnect+replay runs instead of keeping a zombie socket.
-      onTimeout: (error) => this.handleSocketClosed(error)
+        this.sendRequest(requestId, requestMethod, requestParams)
     })
   }
 

src/shared/remote-runtime-shared-control-keepalive-refresh.test.ts

@@ -22,10 +22,8 @@ describe('shared control keepalive timeout refresh semantics', () => {
   function startRequest(options: { refreshTimeoutOnKeepalive?: boolean } = {}): {
     pendingRequests: Map<string, SharedControlPendingRequest<unknown>>
     promise: Promise<unknown>
-    onTimeout: ReturnType<typeof vi.fn>
   } {
     const pendingRequests = new Map<string, SharedControlPendingRequest<unknown>>()
-    const onTimeout = vi.fn()
     const promise = requestSharedControl({
       pendingRequests,
       method: 'git.status',
@@ -35,16 +33,15 @@ describe('shared control keepalive timeout refresh semantics', () => {
       // server never answers, modelling a genuinely stuck server-side call.
       ensureReady: () => Promise.resolve(),
       send: () => undefined,
-      onTimeout,
       refreshTimeoutOnKeepalive: options.refreshTimeoutOnKeepalive
     })
     // Swallow the eventual rejection so unhandled-rejection noise doesn't leak.
     promise.catch(() => undefined)
-    return { pendingRequests, promise, onTimeout }
+    return { pendingRequests, promise }
   }
 
   it('times out a stuck short RPC even while keepalive frames keep arriving', async () => {
-    const { pendingRequests, promise, onTimeout } = startRequest()
+    const { pendingRequests, promise } = startRequest()
 
     // Periodic keepalives arrive faster than the 1000ms deadline — as they
     // would while a long-poll subscription streams over the same socket.
@@ -55,13 +52,11 @@ describe('shared control keepalive timeout refresh semantics', () => {
     await vi.advanceTimersByTimeAsync(1)
 
     await expect(promise).rejects.toThrow()
-    // The stuck-request path tears the connection down so reconnect+replay runs.
-    expect(onTimeout).toHaveBeenCalledTimes(1)
     expect(pendingRequests.size).toBe(0)
   })
 
   it('keeps refreshing a long-poll request that opted into keepalive refresh', async () => {
-    const { pendingRequests, promise, onTimeout } = startRequest({
+    const { pendingRequests, promise } = startRequest({
       refreshTimeoutOnKeepalive: true
     })
 
@@ -72,7 +67,6 @@ describe('shared control keepalive timeout refresh semantics', () => {
       refreshSharedControlPendingRequestTimeouts(pendingRequests)
     }
 
-    expect(onTimeout).not.toHaveBeenCalled()
     expect(pendingRequests.size).toBe(1)
 
     // It still resolves normally once the server finally answers.
@@ -87,12 +81,11 @@ describe('shared control keepalive timeout refresh semantics', () => {
   })
 
   it('fires the deadline for a short RPC when no keepalives arrive', async () => {
-    const { pendingRequests, promise, onTimeout } = startRequest()
+    const { pendingRequests, promise } = startRequest()
 
     await vi.advanceTimersByTimeAsync(1001)
 
     await expect(promise).rejects.toThrow()
-    expect(onTimeout).toHaveBeenCalledTimes(1)
     expect(pendingRequests.size).toBe(0)
   })
 })

src/shared/remote-runtime-shared-control-requests.ts

@@ -1,9 +1,5 @@
 import { randomUUID } from 'node:crypto'
-import type { RemoteRuntimeClientError } from './remote-runtime-client-error'
-import {
-  remoteRuntimeTimeoutError,
-  remoteRuntimeUnavailableError
-} from './remote-runtime-request-frames'
+import { remoteRuntimeTimeoutError } from './remote-runtime-request-frames'
 import type { RuntimeRpcResponse } from './runtime-rpc-envelope'
 import { toRemoteRuntimeClientError } from './remote-runtime-shared-control-protocol'
 import { rejectSharedControlPendingRequest } from './remote-runtime-shared-control-state'
@@ -16,7 +12,6 @@ export function requestSharedControl<TResult>(args: {
   timeoutMs: number
   ensureReady: () => Promise<void>
   send: (requestId: string, method: string, params: unknown) => void
-  onTimeout?: (error: RemoteRuntimeClientError) => void
   // Why: default off — ordinary short RPCs keep an absolute deadline. Only
   // long-polls routed through this path opt in so keepalives extend them.
   refreshTimeoutOnKeepalive?: boolean
@@ -29,16 +24,9 @@ export function requestSharedControl<TResult>(args: {
         return
       }
       args.pendingRequests.delete(requestId)
+      // Why: one stalled method does not prove the shared socket is dead;
+      // socket liveness owns connection-wide teardown so other RPCs survive.
       pending.reject(remoteRuntimeTimeoutError())
-      // Why: a request the server never answered means the socket is suspect
-      // (half-open tunnels swallow frames silently); mirror
-      // RemoteRuntimeRequestConnection and hand the connection a teardown
-      // error so reconnect+replay runs instead of keeping a zombie socket.
-      args.onTimeout?.(
-        remoteRuntimeUnavailableError(
-          'Remote Orca runtime did not answer in time; resetting the control connection.'
-        )
-      )
     }, args.timeoutMs)
     args.pendingRequests.set(requestId, {
       method: args.method,