fix(orchestrator): bound each relayfile call so a hung read cannot wedge reconcile (#351)

docs/deployed-diagnostics.md

@@ -92,13 +92,26 @@ logic of its own by design: the boundary lives in one place, in this repo, with
   A cold container legitimately spends minutes in its first pass (#36 measured 61 minutes while the
   Relayfile mirror hydrated), so check `lastCompletedAtMs`: absent means "first pass since boot,
   still hydrating"; present and hours old means "was fine, then wedged".
-- **How long a stall can last** — a sweep is bounded at `liveSubscription.reconcileTimeoutMs`,
-  90 minutes by default (#296). On expiry the *wait* fails, so `consecutiveFailures` starts rising
-  and the loop schedules the next pass; the sweep itself is not cancelled, because it holds a durable
-  discovery lease, so `inFlightSinceMs` keeps ageing until it really finishes. A `stalled` state that
-  never turns into a rising `consecutiveFailures` therefore means the process is not running the loop
-  at all, which is a restart, not a wait. The deadline sits above #36's 61-minute measurement on
+- **How long a stall can last** — two deadlines, at different scales.
+
+  `liveSubscription.relayfileOperationTimeoutMs` bounds ONE relayfile call, five minutes by default
+  (#351). This is the one that catches a wedge. Expiry cancels the request, fails the pass with
+  `lastErrorClass: "RelayfileOperationTimeoutError"` and a `lastError` naming the call
+  (`relayfile listTree did not respond within 300000ms (GitHub issue ingestion)`), and unwinds the
+  sweep — which releases the discovery lease, so the next cycle starts clean.
+
+  `liveSubscription.reconcileTimeoutMs` bounds the whole sweep, 90 minutes by default (#296). It is
+  the outer backstop only. On expiry the *wait* fails, so `consecutiveFailures` starts rising and the
+  loop schedules the next pass; the sweep itself is not cancelled, because it holds a durable
+  discovery lease, so `inFlightSinceMs` keeps ageing until it really finishes — and the next pass
+  coalesces onto that same running `runOnce()`. The deadline sits above #36's 61-minute measurement on
   purpose: setting it below realistic cold-mirror hydration would turn a slow boot into a crash loop.
+  Per-call bounds can be far tighter precisely because that cold-mirror cost is spread across
+  thousands of calls rather than concentrated in one.
+
+  A `stalled` state that never turns into a rising `consecutiveFailures` means either the process is
+  not running the loop at all, or it predates #351 — on a current build a hung call fails within
+  `relayfileOperationTimeoutMs`.
 
 ### Why `ok` stays `true` while `status` goes amber
 

src/cli/fleet.ts

@@ -2203,6 +2203,9 @@ async function buildMount(
   mount = await (deps.cloudMountFromConfig ?? RelayfileCloudMountClient.fromConfig)({
     workspaceId: loaded.config.workspaceId,
     localMountRoot: loaded.config.localMountRoot,
+    // The transport half of #351: without a signal the SDK issues a bare
+    // fetch() with no deadline, which is what wedged the reconcile loop.
+    operationTimeoutMs: loaded.config.liveSubscription.relayfileOperationTimeoutMs,
     logger: observability.logger,
     onLocalMountHealth: observability.onLocalMountHealth,
     isAllowedDraft: (path, content, opts) => isAllowedFactoryDraft(path, content, opts, mount, loaded.config),

src/config/schema.ts

@@ -7,6 +7,7 @@ import {
   DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
   DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
 } from '../fleet/control-plane-circuit'
+import { DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS } from '../mount/relayfile-operation-timeout'
 
 import { KubernetesEnvironmentConfigSchema } from '../environments/connection-registry.js'
 
@@ -70,6 +71,19 @@ const liveSubscriptionSchema = z.object({
   /** Bounds one sweep so a hung dependency call cannot stop the loop forever. */
   reconcileTimeoutMs: z.number().int().min(50).max(6 * 60 * 60_000)
     .default(DEFAULT_READINESS_RECONCILE_TIMEOUT_MS),
+  /**
+   * Bounds ONE relayfile call (#351).
+   *
+   * Distinct from `reconcileTimeoutMs`, and deliberately far tighter. The
+   * cold-mirror cost that forces the sweep deadline up to 90 minutes is spread
+   * across thousands of calls; no single call has ever needed minutes. So a
+   * per-call bound catches a wedge in five minutes instead of ninety without
+   * re-creating the crash-loop risk described above — and unlike the sweep
+   * deadline it actually unwinds the pass, releasing the discovery lease so the
+   * next cycle starts clean rather than coalescing onto the hung one.
+   */
+  relayfileOperationTimeoutMs: z.number().int().min(50).max(60 * 60_000)
+    .default(DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS),
 }).superRefine((value, ctx) => {
   // A deadline below the interval kills every pass that takes longer than one
   // tick, which is most of them on a cold mirror.

src/mount/relayfile-cloud-mount-client.test.ts

@@ -62,7 +62,7 @@ class FakeRelayFileClient implements RelayFileClientLike {
     path: string
     baseRevision: string
   }> = []
-  readonly listTreeCalls: Array<{ workspaceId: string; options?: { path?: string; depth?: number; cursor?: string } }> = []
+  readonly listTreeCalls: Array<{ workspaceId: string; options?: { path?: string; depth?: number; cursor?: string; signal?: AbortSignal } }> = []
   readonly getEventsCalls: Array<{ workspaceId: string; opts?: { cursor?: string; limit?: number; provider?: string; last?: number } }> = []
   readonly listLastNChangesCalls: Array<{ limit: number; context?: { workspaceId: string } }> = []
   readonly getOpCalls: Array<{ workspaceId: string; opId: string }> = []
@@ -1456,8 +1456,11 @@ describe('RelayfileCloudMountClient', () => {
     expect(fake.readFileCalls[0]).toEqual({ workspaceId: 'rw_test', path: '/linear/issues/AR-1.json' })
     expect(fake.listTreeCalls[0]).toEqual({
       workspaceId: 'rw_test',
-      options: { path: '/linear/issues', cursor: undefined },
+      // #351: reads carry the per-operation deadline's signal, so a relayfile
+      // call that stops answering is cancelled instead of waited on forever.
+      options: { path: '/linear/issues', cursor: undefined, signal: expect.any(AbortSignal) },
     })
+    expect(fake.listTreeCalls[0]?.options?.signal?.aborted).toBe(false)
     expect(fake.getEventsCalls[0]).toEqual({ workspaceId: 'rw_test', opts: { cursor: 'evt-0', limit: 10 } })
   })
 
@@ -1486,6 +1489,104 @@ describe('RelayfileCloudMountClient', () => {
     expect(fake.listTreeCalls.map((call) => call.options?.cursor)).toEqual([undefined, '2', '4'])
   })
 
+  // #351: the SDK attaches an AbortSignal to its fetch only when the caller
+  // supplies one, and nothing here did — so every relayfile read was a bare
+  // fetch() with no deadline. One of them stopped answering on 2026-08-23 and
+  // the readiness reconcile cycle that issued it never finished.
+  describe('bounded relayfile reads', () => {
+    class HangingListTreeClient extends FakeRelayFileClient {
+      seenSignal?: AbortSignal
+
+      override async listTree(
+        workspaceId: string,
+        options?: { path?: string; depth?: number; cursor?: string; signal?: AbortSignal },
+      ): Promise<never> {
+        this.listTreeCalls.push({ workspaceId, options })
+        this.seenSignal = options?.signal
+        return await new Promise<never>((_resolve, reject) => {
+          options?.signal?.addEventListener('abort', () => {
+            reject((options.signal as AbortSignal & { reason?: unknown }).reason)
+          })
+        })
+      }
+    }
+
+    it('cancels a read that stops answering and names the operation', async () => {
+      const client = new HangingListTreeClient()
+      const mount = new RelayfileCloudMountClient({
+        workspaceId: 'rw_test',
+        client,
+        operationTimeoutMs: 25,
+      })
+
+      await expect(mount.listTree('/github/repos')).rejects.toMatchObject({
+        name: 'RelayfileOperationTimeoutError',
+        operation: 'listTree',
+      })
+      // Cancelled, not merely abandoned. An abandoned wait leaves the socket
+      // and the SDK's retry loop live, and hands the next caller the same
+      // wedged in-flight read.
+      expect(client.seenSignal?.aborted).toBe(true)
+    })
+
+    it('honours the ensureSubRoot timeout it used to accept and discard', async () => {
+      const client = new HangingListTreeClient()
+      const mount = new RelayfileCloudMountClient({
+        workspaceId: 'rw_test',
+        client,
+        // Client-wide budget far past the caller's, so only the per-call
+        // argument can end this. It was silently dropped before #351, which is
+        // why `#ensureGithubIngestionReady` passing 90_000 bounded nothing.
+        operationTimeoutMs: 60_000,
+      })
+
+      await expect(mount.ensureSubRoot('/github/issues', { timeoutMs: 25 })).rejects.toMatchObject({
+        name: 'RelayfileOperationTimeoutError',
+        operation: 'ensureSubRoot',
+      })
+    })
+
+    it('caps an explicit ensureSubRoot timeout at the tighter client-wide budget', async () => {
+      const client = new HangingListTreeClient()
+      const mount = new RelayfileCloudMountClient({
+        workspaceId: 'rw_test',
+        client,
+        operationTimeoutMs: 25,
+      })
+
+      // The caller's argument must cap, not replace: a client-wide budget
+      // tighter than the argument has to still cancel at the transport, or the
+      // orchestrator's backstop fires first and abandons the wait instead
+      // (codex on #354).
+      await expect(mount.ensureSubRoot('/github/issues', { timeoutMs: 60_000 })).rejects.toMatchObject({
+        name: 'RelayfileOperationTimeoutError',
+        operation: 'ensureSubRoot',
+        timeoutMs: 25,
+      })
+      expect(client.seenSignal?.aborted).toBe(true)
+    })
+
+    it('leaves the call unbounded when no budget is configured', async () => {
+      const client = new HangingListTreeClient()
+      const mount = new RelayfileCloudMountClient({
+        workspaceId: 'rw_test',
+        client,
+        operationTimeoutMs: 0,
+      })
+
+      const pending = mount.listTree('/github/repos')
+      const settled = await Promise.race([
+        pending.then(() => 'settled' as const, () => 'settled' as const),
+        new Promise<'pending'>((resolve) => { setTimeout(() => resolve('pending'), 50) }),
+      ])
+      expect(settled).toBe('pending')
+      expect(client.seenSignal).toBeUndefined()
+      // Nothing will ever settle this; drop the reference explicitly so the
+      // rejection-free pending promise is not mistaken for a leak.
+      void pending.catch(() => undefined)
+    })
+  })
+
   it('uses recent change-log events for provider-filtered getEvents tail reads', async () => {
     const fake = new FakeRelayFileClient()
     fake.events = [

src/mount/relayfile-cloud-mount-client.ts

@@ -22,6 +22,15 @@ import {
   type WriteQueuedResponse,
 } from '@relayfile/sdk'
 import { RelayfileSetup } from '@relayfile/sdk/cli'
+import {
+  DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS,
+  RelayfileOperationTimeoutError,
+  isRelayfileCallAbort,
+  relayfileCallBudgetMs,
+  relayfileCallDeadline,
+  tighterRelayfileBudgetMs,
+  withRelayfileCallDeadline,
+} from './relayfile-operation-timeout'
 import { existsSync } from 'node:fs'
 import { isAbsolute, join, resolve } from 'node:path'
 
@@ -260,10 +269,26 @@ export interface RelayfileCloudMountClientConfig {
   skipRegisteredMirrorLookup?: boolean
   isAllowedDraft?: (path: string, content: unknown, opts?: { guarded?: boolean }) => boolean | Promise<boolean>
   isAllowedDelete?: (path: string, currentContent: unknown) => boolean | Promise<boolean>
+  /**
+   * Deadline for one relayfile read, in milliseconds (#351).
+   *
+   * The SDK attaches an `AbortSignal` to its `fetch` only when the caller
+   * supplies one, so without this every read is a bare `fetch()` that can wait
+   * forever. Zero or a non-finite value restores that unbounded behaviour.
+   */
+  operationTimeoutMs?: number
 }
 
 export type RelayFileClientLike = {
-    readFile(workspaceId: string, path: string): Promise<FileReadResponse>
+    // `correlationId`/`signal` are the SDK's own trailing arguments. Declared
+    // here so a read can carry a deadline (#351); optional, so a narrower fake
+    // still satisfies the type.
+    readFile(
+      workspaceId: string,
+      path: string,
+      correlationId?: string,
+      signal?: AbortSignal,
+    ): Promise<FileReadResponse>
     writeFile(input: WriteFileInput): Promise<WriteQueuedResponse>
     deleteFile(input: DeleteFileInput): Promise<WriteQueuedResponse>
     listTree(workspaceId: string, options?: ListTreeOptions): Promise<TreeResponse>
@@ -335,6 +360,7 @@ export class RelayfileCloudMountClient implements MountClient {
   #disposed = false
   #isAllowedDraft?: (path: string, content: unknown, opts?: { guarded?: boolean }) => boolean | Promise<boolean>
   readonly #isAllowedDelete?: (path: string, currentContent: unknown) => boolean | Promise<boolean>
+  readonly #operationTimeoutMs: number
   readonly #lastOpByPath = new Map<string, string>()
   readonly #confirmedExternalIdByPath = new Map<string, string>()
   readonly #confirmedFailureReasonByPath = new Map<string, string>()
@@ -345,6 +371,7 @@ export class RelayfileCloudMountClient implements MountClient {
     }
 
     this.workspaceId = config.workspaceId ?? DEFAULT_WORKSPACE_ID
+    this.#operationTimeoutMs = config.operationTimeoutMs ?? DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS
     this.#client = config.client
     this.#tokenProvider = config.tokenProvider ?? (() => this.#client.getToken?.())
     this.#baseUrl = config.baseUrl ?? this.#client.getBaseUrl?.()
@@ -646,7 +673,8 @@ export class RelayfileCloudMountClient implements MountClient {
   }
 
   async readFile(path: string): Promise<{ content: unknown; revision?: string }> {
-    const response = await this.#client.readFile(this.workspaceId, path)
+    const response = await this.#bounded('readFile', this.#operationTimeoutMs, (signal) =>
+      this.#client.readFile(this.workspaceId, path, undefined, signal))
     return {
       content: parseRemoteContent(response),
       revision: response.revision,
@@ -735,18 +763,88 @@ export class RelayfileCloudMountClient implements MountClient {
   }
 
   async listTree(prefix: string): Promise<string[]> {
-    const paths: string[] = []
-    let cursor: string | undefined
-    for (;;) {
-      const response = await this.#client.listTree(this.workspaceId, {
-        path: prefix,
-        cursor,
+    // ONE deadline for the whole walk, not one per page. The cursor loop is
+    // unbounded, so a per-page budget would let a server that keeps handing
+    // back a `nextCursor` run for as long as it likes — the loop would stay
+    // unbounded while every individual call looked bounded (#351).
+    const budgetMs = relayfileCallBudgetMs(this.#operationTimeoutMs)
+    const deadline = relayfileCallDeadline('listTree', budgetMs)
+    const deadlineAtMs = budgetMs === undefined ? undefined : Date.now() + budgetMs
+    try {
+      return await this.#named('listTree', budgetMs, async () => {
+        const paths: string[] = []
+        let cursor: string | undefined
+        for (;;) {
+          const response = await withRelayfileCallDeadline(
+            'listTree',
+            undefined,
+            this.#remainingMs('listTree', deadlineAtMs),
+            () => this.#client.listTree(this.workspaceId, {
+              path: prefix,
+              cursor,
+              ...(deadline.signal ? { signal: deadline.signal } : {}),
+            }),
+          )
+          paths.push(...response.entries.map((entry) => entry.path))
+          if (!response.nextCursor) break
+          cursor = response.nextCursor
+        }
+        return paths.sort()
       })
-      paths.push(...response.entries.map((entry) => entry.path))
-      if (!response.nextCursor) break
-      cursor = response.nextCursor
+    } finally {
+      deadline.dispose()
+    }
+  }
+
+  /** What is left of a multi-call operation's budget, refusing at zero. */
+  #remainingMs(operation: string, deadlineAtMs: number | undefined): number | undefined {
+    if (deadlineAtMs === undefined) return undefined
+    const remainingMs = deadlineAtMs - Date.now()
+    // Refuse rather than fall through: `withRelayfileCallDeadline` treats a
+    // non-positive budget as "no budget" and would run the call unbounded.
+    if (remainingMs <= 0) {
+      throw new RelayfileOperationTimeoutError(operation, this.#operationTimeoutMs)
+    }
+    return remainingMs
+  }
+
+  /**
+   * One relayfile read under a deadline: cancelled at the transport where the
+   * SDK honours the signal, and abandoned by the race where it does not.
+   */
+  async #bounded<T>(
+    operation: string,
+    budgetMs: number | undefined,
+    start: (signal?: AbortSignal) => Promise<T>,
+  ): Promise<T> {
+    const usableMs = relayfileCallBudgetMs(budgetMs)
+    const deadline = relayfileCallDeadline(operation, usableMs)
+    try {
+      return await this.#named(operation, usableMs, () =>
+        withRelayfileCallDeadline(operation, undefined, usableMs, () => start(deadline.signal)))
+    } finally {
+      deadline.dispose()
+    }
+  }
+
+  /**
+   * Give the deadline a name an operator can act on.
+   *
+   * A signal expiry surfaces as a bare `TimeoutError` reading "the operation
+   * was aborted" — true and useless. `readinessReconcile.lastError` has to say
+   * which relayfile call it was waiting on, which is the whole point of #351.
+   * These paths pass no other signal, so an abort here can only be ours.
+   */
+  async #named<T>(operation: string, budgetMs: number | undefined, run: () => Promise<T>): Promise<T> {
+    try {
+      return await run()
+    } catch (error) {
+      if (budgetMs !== undefined && isRelayfileCallAbort(error)) {
+        if (error instanceof RelayfileOperationTimeoutError) throw error
+        throw new RelayfileOperationTimeoutError(operation, budgetMs)
+      }
+      throw error
     }
-    return paths.sort()
   }
 
   subscribe(globs: string[], onChange: (event: ChangeEvent) => void, opts?: SubscribeOptions): Subscription {
@@ -970,9 +1068,21 @@ export class RelayfileCloudMountClient implements MountClient {
     return this.#confirmedExternalIdByPath.get(path)
   }
 
-  async ensureSubRoot(prefix: string, _opts?: { timeoutMs?: number }): Promise<'ready' | 'absent'> {
+  // `timeoutMs` used to be accepted and discarded, so `#ensureGithubIngestionReady`
+  // passed 90_000 and got no bound at all — the call site believed it was
+  // bounded and was not (#351). It is honoured now, and it *caps* rather than
+  // replaces the client-wide budget: a config that tightened
+  // `relayfileOperationTimeoutMs` below the caller's argument would otherwise
+  // let the transport run past the orchestrator's backstop, abandoning the wait
+  // instead of cancelling the call (codex on #354).
+  async ensureSubRoot(prefix: string, opts?: { timeoutMs?: number }): Promise<'ready' | 'absent'> {
     try {
-      await this.#client.listTree(this.workspaceId, { path: prefix, depth: 1 })
+      await this.#bounded('ensureSubRoot', tighterRelayfileBudgetMs(opts?.timeoutMs, this.#operationTimeoutMs), (signal) =>
+        this.#client.listTree(this.workspaceId, {
+          path: prefix,
+          depth: 1,
+          ...(signal ? { signal } : {}),
+        }))
       return 'ready'
     } catch (error) {
       if (isHttpStatus(error, 404)) return 'absent'