fix(factory): fail closed on broker control-plane faults

README.md

@@ -150,6 +150,34 @@ names and the provider-claim state (`pending`, `verified`, or `degraded`). This
 view is read from Factory's local in-flight registry, so it remains available
 when GitHub lifecycle writeback is the degraded subsystem.
 
+It also reports `fleetControlPlane`. Factory bounds its read-only roster probe,
+pauses a live dispatch immediately when that probe fails, and opens a circuit
+after repeated failures. While the circuit is open, new spawn and resume calls
+fail fast. After `fleetHealth.resetTimeoutMs`, one half-open roster probe may
+close the circuit. Configure `fleetHealth.rosterTimeoutMs`,
+`fleetHealth.failureThreshold`, and `fleetHealth.resetTimeoutMs` when the fleet
+has intentionally higher control-plane latency. Mutating calls are never
+abandoned behind a local timeout because an accepted-but-late spawn would be
+ambiguous.
+
+A paused `run-once` or `loop` exits non-zero and logs `dispatch paused by the
+circuit`; `factory status` reads the daemon heartbeat and reports the circuit's
+`state`, `lastError`, and `retryAtMs`. If the broker is healthy but slower than
+the configured bound, raise `fleetHealth.rosterTimeoutMs` (maximum 60 seconds)
+to match measured control-plane latency. If the control path is unavailable,
+repair the isolated broker and wait until `retryAtMs`; the next successful
+half-open roster probe closes the circuit automatically. Do not treat an open
+circuit as an empty queue or successful no-work iteration.
+
+For production, set `fleetHealth.requireDedicatedBroker` to `true` and point
+`AGENT_RELAY_STATE_DIR` at a state directory that is distinct from the
+project's `.agentworkforce/relay`. Factory then refuses startup if it would
+silently reuse the interactive project broker.
+
+Factory now defaults `batchSize` to `1`. Operators may explicitly raise it to
+at most `5` after isolating Factory workers from an interactive broker and
+measuring control-plane latency under the intended recipe mix.
+
 For a live daemon, `factory status` also includes `readinessReconcile`. Its
 state advances from `retrying` to `degraded` after three consecutive periodic
 discovery failures and returns to `healthy` after a successful checkpoint.

src/cli/fleet.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it, vi } from 'vitest'
 import { existsSync } from 'node:fs'
-import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { mkdir, mkdtemp, readFile, rm, symlink, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { dirname, join } from 'node:path'
 
@@ -23,7 +23,7 @@ import type { GithubConnectionRead, GithubConnectionWrite, GithubIssueLookup, Lo
 import type { HarnessDriverClientLike } from '../fleet/internal-fleet-client'
 import { factoryGithubIssueCommentDraftName } from '../github/writeback-paths'
 import { ensureLocalMount as runLocalMountPreflight } from '../mount/local-mount-preflight'
-import { formatLogArgs, installFactoryStopSignalHandlers, parseFleetCommand, parseGithubIssueSelector, parseGlobalOptions, reportFactoryVersionDrift, resolveBrokerConnectionPath, runFleetCli } from './fleet'
+import { formatLogArgs, installFactoryStopSignalHandlers, parseFleetCommand, parseGithubIssueSelector, parseGlobalOptions, reportFactoryVersionDrift, resolveBrokerConnectionPath, resolveFactoryBrokerConnectionPath, runFleetCli } from './fleet'
 
 const issuePath = '/linear/issues/AR-77__uuid-77.json'
 
@@ -530,6 +530,73 @@ describe('fleet CLI parsing', () => {
       await rm(root, { recursive: true, force: true })
     }
   })
+
+  it('requires a separate explicit relay state directory when dedicated broker isolation is enabled', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-dedicated-broker-'))
+    try {
+      const projectStateDir = join(root, '.agentworkforce', 'relay')
+      const dedicatedStateDir = join(root, '.agentworkforce', 'factory-relay')
+      await mkdir(projectStateDir, { recursive: true })
+      await writeFile(join(projectStateDir, 'connection.json'), JSON.stringify({ port: 3890 }))
+
+      expect(() => resolveFactoryBrokerConnectionPath(root, {}, true))
+        .toThrow(/requires AGENT_RELAY_STATE_DIR/u)
+      expect(() => resolveFactoryBrokerConnectionPath(root, { AGENT_RELAY_STATE_DIR: projectStateDir }, true))
+        .toThrow(/resolves to the project broker/u)
+      expect(() => resolveFactoryBrokerConnectionPath(root, {
+        AGENT_RELAY_STATE_DIR: join(projectStateDir, '..', 'relay'),
+      }, true)).toThrow(/resolves to the project broker/u)
+      expect(resolveFactoryBrokerConnectionPath(root, { AGENT_RELAY_STATE_DIR: dedicatedStateDir }, true))
+        .toBe(join(dedicatedStateDir, 'connection.json'))
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('rejects the shared project relay path before its connection file exists', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-empty-dedicated-broker-'))
+    try {
+      const projectStateDir = join(root, '.agentworkforce', 'relay')
+      expect(() => resolveFactoryBrokerConnectionPath(root, { AGENT_RELAY_STATE_DIR: projectStateDir }, true))
+        .toThrow(/resolves to the project broker/u)
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('rejects the project relay path even when broker discovery finds an ancestor first', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-ancestor-dedicated-broker-'))
+    try {
+      const project = join(root, 'project')
+      const projectStateDir = join(project, '.agentworkforce', 'relay')
+      const ancestorConnectionPath = join(root, '.agentworkforce', 'relay', 'connection.json')
+      await mkdir(dirname(ancestorConnectionPath), { recursive: true })
+      await mkdir(project, { recursive: true })
+      await writeFile(ancestorConnectionPath, JSON.stringify({ port: 3890 }))
+
+      expect(() => resolveFactoryBrokerConnectionPath(project, {
+        AGENT_RELAY_STATE_DIR: projectStateDir,
+      }, true)).toThrow(/resolves to the project broker/u)
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('rejects a symlink that resolves to the shared project relay path', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-symlink-dedicated-broker-'))
+    try {
+      const projectStateDir = join(root, '.agentworkforce', 'relay')
+      const stateAlias = join(root, 'relay-alias')
+      await mkdir(projectStateDir, { recursive: true })
+      await symlink(projectStateDir, stateAlias, 'dir')
+
+      expect(() => resolveFactoryBrokerConnectionPath(root, {
+        AGENT_RELAY_STATE_DIR: stateAlias,
+      }, true)).toThrow(/resolves to the project broker/u)
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
 })
 
 describe('parseGithubIssueSelector normalization matrix', () => {
@@ -2709,7 +2776,7 @@ describe('fleet CLI runtime', () => {
     }
   })
 
-  it('surfaces degraded readiness reconciliation from the live daemon heartbeat', async () => {
+  it('surfaces degraded readiness and an open fleet circuit from the live daemon heartbeat', async () => {
     const root = await mkdtemp(join(tmpdir(), 'fleet-cli-reconcile-status-'))
     try {
       const heartbeatPath = join(root, 'heartbeat.json')
@@ -2730,6 +2797,16 @@ describe('fleet CLI runtime', () => {
           lastFailureAtMs: now - 1_000,
           lastError: 'discovery sweep lease expired',
         },
+        fleetControlPlane: {
+          state: 'open',
+          consecutiveFailures: 2,
+          timeoutMs: 5_000,
+          failureThreshold: 2,
+          resetTimeoutMs: 60_000,
+          lastFailureAtMs: now - 500,
+          retryAtMs: now + 59_500,
+          lastError: 'TimeoutError (FACTORY_FLEET_CONTROL_TIMEOUT)',
+        },
       }))
       const output = buffer()
       const factory = {
@@ -2742,6 +2819,13 @@ describe('fleet CLI runtime', () => {
             consecutiveFailures: 0,
             failureThreshold: 3,
           },
+          fleetControlPlane: {
+            state: 'closed' as const,
+            consecutiveFailures: 0,
+            timeoutMs: 5_000,
+            failureThreshold: 2,
+            resetTimeoutMs: 60_000,
+          },
         })),
       } as unknown as Factory
 
@@ -2761,7 +2845,58 @@ describe('fleet CLI runtime', () => {
           failureThreshold: 3,
           lastError: 'discovery sweep lease expired',
         },
+        fleetControlPlane: {
+          state: 'open',
+          consecutiveFailures: 2,
+          lastError: 'TimeoutError (FACTORY_FLEET_CONTROL_TIMEOUT)',
+        },
+      })
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('does not report a fresh local closed circuit for a live older daemon heartbeat', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'fleet-cli-legacy-circuit-status-'))
+    try {
+      const heartbeatPath = join(root, 'heartbeat.json')
+      const configPath = await writeConfig(root, { loop: { heartbeatPath, heartbeatStaleMs: 10_000 } })
+      const now = Date.now()
+      await writeFile(heartbeatPath, JSON.stringify({
+        pid: process.pid,
+        status: 'running',
+        iteration: 0,
+        maxIterations: 0,
+        updatedAt: new Date(now).toISOString(),
+        updatedAtMs: now,
+        eventListener: { state: 'subscribed' },
+      }))
+      const output = buffer()
+      const factory = {
+        status: vi.fn(() => ({
+          inFlight: [],
+          queued: [],
+          counters: {},
+          fleetControlPlane: {
+            state: 'closed' as const,
+            consecutiveFailures: 0,
+            timeoutMs: 5_000,
+            failureThreshold: 2,
+            resetTimeoutMs: 60_000,
+          },
+        })),
+      } as unknown as Factory
+
+      const code = await runFleetCli(['status', '--config', configPath], {
+        fleet: new FakeFleetClient(),
+        mount: new FakeMountClient(),
+        createFactory: () => factory,
+        stdout: output,
+        stderr: buffer(),
       })
+
+      expect(code).toBe(0)
+      expect(JSON.parse(output.text())).not.toHaveProperty('fleetControlPlane')
     } finally {
       await rm(root, { recursive: true, force: true })
     }

src/cli/fleet.ts

@@ -1,6 +1,6 @@
-import { existsSync } from 'node:fs'
+import { existsSync, realpathSync } from 'node:fs'
 import { readFile } from 'node:fs/promises'
-import { dirname, join, resolve } from 'node:path'
+import { basename, dirname, join, resolve } from 'node:path'
 import { createHash, randomUUID } from 'node:crypto'
 import readline from 'node:readline/promises'
 import { ensureCloudSession, type CloudSession } from '@agent-relay/cloud'
@@ -1160,7 +1160,9 @@ async function factoryStatusWithMountHealth(
   heartbeatStaleMs: number,
   versionInfo?: FactoryVersionInfo,
   stateStoreStatus?: { backend: string },
-): Promise<ReturnType<Factory['status']> & {
+): Promise<Omit<ReturnType<Factory['status']>, 'fleetControlPlane'> & {
+  /** Undefined when a live older daemon predates fleet control-plane reporting. */
+  fleetControlPlane?: ReturnType<Factory['status']>['fleetControlPlane']
   stateStore?: { backend: string }
   version?: string
   installedAt?: string
@@ -1198,6 +1200,9 @@ async function factoryStatusWithMountHealth(
   const readinessReconcile = liveness.ok
     ? heartbeat?.readinessReconcile
     : observableStatus.readinessReconcile
+  const fleetControlPlane = liveness.ok
+    ? heartbeat?.fleetControlPlane
+    : observableStatus.fleetControlPlane
   const eventListener = liveness.ok
     ? heartbeat?.eventListener ?? {
       state: 'unknown' as const,
@@ -1215,6 +1220,7 @@ async function factoryStatusWithMountHealth(
     heldAgents,
     eventListener,
     readinessReconcile,
+    fleetControlPlane,
   }
   return {
     ...observableStatus,
@@ -1223,6 +1229,7 @@ async function factoryStatusWithMountHealth(
     heldAgents,
     eventListener,
     readinessReconcile,
+    fleetControlPlane,
     localMountDegraded: health.degraded,
     ...(health.reason ? { localMountDegradedReason: health.reason } : {}),
     ...(health.localDir ? { localMountRoot: health.localDir } : {}),
@@ -1583,7 +1590,13 @@ async function buildFleet(
 
   const cwd = process.cwd()
   const env = deps.env ?? process.env
-  const connectionPath = resolveBrokerConnectionPath(cwd, env)
+  const connectionPath = globals.backend === 'internal'
+    ? resolveFactoryBrokerConnectionPath(
+        cwd,
+        env,
+        loaded?.config.fleetHealth.requireDedicatedBroker ?? false,
+      )
+    : resolveBrokerConnectionPath(cwd, env)
 
   // An injected createFleet owns fleet construction entirely (tests), so skip the
   // real broker bootstrap.
@@ -1678,6 +1691,60 @@ export function resolveBrokerConnectionPath(
   }
 }
 
+export function resolveFactoryBrokerConnectionPath(
+  startCwd = process.cwd(),
+  env: NodeJS.ProcessEnv = process.env,
+  requireDedicatedBroker = false,
+): string | undefined {
+  const connectionPath = resolveBrokerConnectionPath(startCwd, env)
+  if (!requireDedicatedBroker) return connectionPath
+
+  const explicitStateDir = env.AGENT_RELAY_STATE_DIR?.trim()
+  if (!explicitStateDir) {
+    throw new Error(
+      'fleetHealth.requireDedicatedBroker requires AGENT_RELAY_STATE_DIR to name Factory\'s isolated relay state directory',
+    )
+  }
+  const inheritedConnectionPath = resolveBrokerConnectionPath(startCwd, {
+    ...env,
+    AGENT_RELAY_STATE_DIR: '',
+  })
+  // Always reject the checkout-local path, even if broker discovery currently
+  // walks past it and finds an ancestor connection. An interactive relay
+  // started in this checkout can claim the local path later.
+  const sharedStateDirs = [
+    join(resolve(startCwd), '.agentworkforce', 'relay'),
+    ...(inheritedConnectionPath ? [dirname(inheritedConnectionPath)] : []),
+  ]
+  if (sharedStateDirs.some((stateDir) => sameFilesystemPath(explicitStateDir, stateDir))) {
+    throw new Error(
+      'AGENT_RELAY_STATE_DIR resolves to the project broker; choose a separate state directory for Factory',
+    )
+  }
+  return connectionPath
+}
+
+function sameFilesystemPath(left: string, right: string): boolean {
+  return canonicalPath(left) === canonicalPath(right)
+}
+
+function canonicalPath(path: string): string {
+  const absolute = resolve(path)
+  const missingSegments: string[] = []
+  let existingAncestor = absolute
+  while (!existsSync(existingAncestor)) {
+    const parent = dirname(existingAncestor)
+    if (parent === existingAncestor) return absolute
+    missingSegments.unshift(basename(existingAncestor))
+    existingAncestor = parent
+  }
+  try {
+    return resolve(realpathSync.native(existingAncestor), ...missingSegments)
+  } catch {
+    return absolute
+  }
+}
+
 async function prepareFactoryIntegrations(
   command: ParsedCommand,
   mount: MountClient,

src/config/schema.test.ts

@@ -28,12 +28,18 @@ describe('FactoryConfigSchema', () => {
     expect(parsed.repos.byProject).toEqual({})
     expect(parsed.repos.keywordRules).toEqual([])
     expect(parsed.repos.clonePaths).toEqual({})
-    expect(parsed.batchSize).toBe(5)
+    expect(parsed.batchSize).toBe(1)
     expect(parsed.dispatch).toEqual({
       errorCooldownMs: 60_000,
       maxAttempts: 2,
       agentHoldTimeoutMs: 4 * 60 * 60_000,
     })
+    expect(parsed.fleetHealth).toEqual({
+      rosterTimeoutMs: 5_000,
+      failureThreshold: 2,
+      resetTimeoutMs: 60_000,
+      requireDedicatedBroker: false,
+    })
     expect(parsed.models).toEqual({ babysitter: 'sonnet' })
     // Agent CLI per role defaults to today's behavior: codex implements, claude
     // reviews/babysits — so existing configs are unaffected unless set.