From 2afd34475e4f842ac640b3aaf9a40fe69630f2d3 Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 10:28:29 +0200
Subject: [PATCH 1/8] fix: pause dispatch when broker control plane is
 unhealthy

---
 README.md                               |  19 +++
 src/cli/fleet.test.ts                   |  21 ++-
 src/cli/fleet.ts                        |  34 +++-
 src/config/schema.test.ts               |   8 +-
 src/config/schema.ts                    |  14 +-
 src/fleet/control-plane-circuit.test.ts |  71 ++++++++
 src/fleet/control-plane-circuit.ts      | 215 ++++++++++++++++++++++++
 src/index.ts                            |   1 +
 src/orchestrator/factory.test.ts        |  30 ++++
 src/orchestrator/factory.ts             |  28 ++-
 src/types.ts                            |   3 +
 11 files changed, 439 insertions(+), 5 deletions(-)
 create mode 100644 src/fleet/control-plane-circuit.test.ts
 create mode 100644 src/fleet/control-plane-circuit.ts

diff --git a/README.md b/README.md
index 6b5ab33..d18795c 100644
--- a/README.md
+++ b/README.md
@@ -149,6 +149,25 @@ 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.
+
+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.
diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index a66049a..0442db6 100644
--- a/src/cli/fleet.test.ts
+++ b/src/cli/fleet.test.ts
@@ -21,7 +21,7 @@ import { FakeFleetClient, FakeMountClient } from '../testing'
 import type { GithubConnectionRead, GithubConnectionWrite, GithubIssueLookup, LocalMountOptions, SpawnInput, SpawnResult } from '../ports'
 import type { HarnessDriverClientLike } from '../fleet/internal-fleet-client'
 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'
 
@@ -514,6 +514,25 @@ 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: dedicatedStateDir }, true))
+        .toBe(join(dedicatedStateDir, 'connection.json'))
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
 })
 
 describe('parseGithubIssueSelector normalization matrix', () => {
diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts
index 6dec180..f5ae51e 100644
--- a/src/cli/fleet.ts
+++ b/src/cli/fleet.ts
@@ -1520,7 +1520,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.
@@ -1615,6 +1621,32 @@ 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: '',
+  })
+  if (connectionPath === inheritedConnectionPath) {
+    throw new Error(
+      'AGENT_RELAY_STATE_DIR resolves to the project broker; choose a separate state directory for Factory',
+    )
+  }
+  return connectionPath
+}
+
 async function prepareFactoryIntegrations(
   command: ParsedCommand,
   mount: MountClient,
diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts
index 0a9023d..d0e7be6 100644
--- a/src/config/schema.test.ts
+++ b/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.
diff --git a/src/config/schema.ts b/src/config/schema.ts
index 2ca7247..f7c5f1e 100644
--- a/src/config/schema.ts
+++ b/src/config/schema.ts
@@ -57,6 +57,17 @@ const dispatchSchema = z.object({
     .default(DEFAULT_AGENT_HOLD_TIMEOUT_MS),
 }).default({})
 
+const fleetHealthSchema = z.object({
+  // Roster is read-only, so Factory can safely bound it locally. Mutating
+  // spawn/resume calls are never abandoned behind a local timeout.
+  rosterTimeoutMs: z.number().int().min(100).max(60_000).default(5_000),
+  failureThreshold: z.number().int().min(1).max(10).default(2),
+  resetTimeoutMs: z.number().int().min(1_000).max(15 * 60_000).default(60_000),
+  // Production launchers can require an explicit, non-project broker state
+  // directory so Factory never silently shares an interactive broker.
+  requireDedicatedBroker: z.boolean().default(false),
+}).default({})
+
 const loopSchema = z.object({
   maxIterations: z.number().int().min(1).max(5).default(3),
   maxConsecutiveFailures: z.number().int().min(1).max(5).default(3),
@@ -301,10 +312,11 @@ const WorkspaceConfigObjectSchema = z.object({
   subscription: subscriptionSchema,
   liveSubscription: liveSubscriptionSchema,
   dispatch: dispatchSchema,
+  fleetHealth: fleetHealthSchema,
   loop: loopSchema,
   triage: triageSchema,
   repos: workspaceReposSchema,
-  batchSize: z.number().int().min(1).max(5).default(5),
+  batchSize: z.number().int().min(1).max(5).default(1),
   models: modelsSchema,
   agentCapabilities: agentCapabilitiesSchema,
   slack: slackSchema,
diff --git a/src/fleet/control-plane-circuit.test.ts b/src/fleet/control-plane-circuit.test.ts
new file mode 100644
index 0000000..6c8c7e8
--- /dev/null
+++ b/src/fleet/control-plane-circuit.test.ts
@@ -0,0 +1,71 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import type { RosterEntry } from '../ports/fleet'
+import { FakeFleetClient } from '../testing/fakes'
+import {
+  FleetControlPlaneCircuit,
+  FleetControlPlaneCircuitOpenError,
+  guardFleetControlPlane,
+  isFleetControlPlaneFailure,
+} from './control-plane-circuit'
+
+const roster: RosterEntry = { agents: [], nodes: [] }
+
+describe('FleetControlPlaneCircuit', () => {
+  it('opens after bounded roster failures and rejects further probes without calling the broker', async () => {
+    let now = 1_000
+    const circuit = new FleetControlPlaneCircuit({
+      timeoutMs: 5,
+      failureThreshold: 2,
+      resetTimeoutMs: 60_000,
+      now: () => now,
+    })
+    const never = vi.fn(() => new Promise<RosterEntry>(() => undefined))
+
+    await expect(circuit.probe(never)).rejects.toMatchObject({ name: 'TimeoutError' })
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 })
+    await expect(circuit.probe(never)).rejects.toMatchObject({ name: 'TimeoutError' })
+    expect(circuit.status()).toMatchObject({ state: 'open', consecutiveFailures: 2, retryAtMs: 61_000 })
+
+    await expect(circuit.probe(never)).rejects.toBeInstanceOf(FleetControlPlaneCircuitOpenError)
+    expect(never).toHaveBeenCalledTimes(2)
+    now += 60_000
+    await expect(circuit.probe(async () => roster)).resolves.toEqual(roster)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
+  })
+
+  it('coalesces concurrent roster probes', async () => {
+    let resolveProbe: ((value: RosterEntry) => void) | undefined
+    const call = vi.fn(() => new Promise<RosterEntry>((resolve) => { resolveProbe = resolve }))
+    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 2, resetTimeoutMs: 1_000 })
+    const first = circuit.probe(call)
+    const second = circuit.probe(call)
+    await Promise.resolve()
+    resolveProbe?.(roster)
+    await expect(Promise.all([first, second])).resolves.toEqual([roster, roster])
+    expect(call).toHaveBeenCalledTimes(1)
+  })
+
+  it('blocks new spawns while open without force-timing an accepted mutation', async () => {
+    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 5, failureThreshold: 1, resetTimeoutMs: 60_000 })
+    const fleet = new FakeFleetClient()
+    const guarded = guardFleetControlPlane(fleet, circuit)
+    circuit.recordFailure(new Error('broker unavailable'))
+
+    await expect(guarded.spawn({
+      name: 'blocked-worker',
+      capability: 'spawn:codex',
+    })).rejects.toBeInstanceOf(FleetControlPlaneCircuitOpenError)
+    expect(fleet.spawns).toEqual([])
+  })
+
+  it('recognizes timeout and transport failures without classifying domain errors', () => {
+    expect(isFleetControlPlaneFailure(Object.assign(new Error('connect failed'), { code: 'ECONNREFUSED' }))).toBe(true)
+    expect(isFleetControlPlaneFailure(Object.assign(new Error('aborted'), { name: 'AbortError' }))).toBe(true)
+    expect(isFleetControlPlaneFailure(new Error('agent already exists'))).toBe(false)
+
+    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 1, resetTimeoutMs: 1_000 })
+    circuit.recordFailure(Object.assign(new Error('https://broker.invalid?token=must-not-leak'), { code: 'ECONNREFUSED' }))
+    expect(circuit.status().lastError).toBe('Error (ECONNREFUSED)')
+  })
+})
diff --git a/src/fleet/control-plane-circuit.ts b/src/fleet/control-plane-circuit.ts
new file mode 100644
index 0000000..123dd4d
--- /dev/null
+++ b/src/fleet/control-plane-circuit.ts
@@ -0,0 +1,215 @@
+import type { FleetClient, RosterEntry, SpawnInput, SpawnResult } from '../ports/fleet'
+
+export type FleetControlPlaneState = 'closed' | 'open' | 'half-open'
+
+export interface FleetControlPlaneStatus {
+  state: FleetControlPlaneState
+  consecutiveFailures: number
+  timeoutMs: number
+  failureThreshold: number
+  resetTimeoutMs: number
+  lastFailureAtMs?: number
+  retryAtMs?: number
+  lastError?: string
+}
+
+export interface FleetControlPlaneCircuitOptions {
+  timeoutMs: number
+  failureThreshold: number
+  resetTimeoutMs: number
+  now?: () => number
+}
+
+export class FleetControlPlaneTimeoutError extends Error {
+  readonly code = 'FACTORY_FLEET_CONTROL_TIMEOUT'
+
+  constructor(readonly timeoutMs: number) {
+    super(`fleet control-plane roster probe timed out after ${timeoutMs}ms`)
+    this.name = 'TimeoutError'
+  }
+}
+
+export class FleetControlPlaneCircuitOpenError extends Error {
+  readonly code = 'FACTORY_FLEET_CONTROL_CIRCUIT_OPEN'
+
+  constructor(readonly retryAtMs: number, readonly state: 'open' | 'half-open' = 'open') {
+    super(state === 'open'
+      ? `fleet control-plane circuit is open until ${new Date(retryAtMs).toISOString()}`
+      : 'fleet control-plane circuit requires a successful roster probe before dispatch')
+    this.name = 'FleetControlPlaneCircuitOpenError'
+  }
+}
+
+/**
+ * Bounds the read-only roster probe and prevents new worker mutations after
+ * repeated control-plane failures. Mutating operations are deliberately not
+ * raced against a local timer: abandoning a spawn after its side effect has
+ * reached the broker would create an ambiguous orphan.
+ */
+export class FleetControlPlaneCircuit {
+  readonly #timeoutMs: number
+  readonly #failureThreshold: number
+  readonly #resetTimeoutMs: number
+  readonly #now: () => number
+  #consecutiveFailures = 0
+  #lastFailureAtMs?: number
+  #retryAtMs?: number
+  #lastError?: string
+  #probeInFlight?: Promise<RosterEntry>
+
+  constructor(options: FleetControlPlaneCircuitOptions) {
+    this.#timeoutMs = options.timeoutMs
+    this.#failureThreshold = options.failureThreshold
+    this.#resetTimeoutMs = options.resetTimeoutMs
+    this.#now = options.now ?? Date.now
+  }
+
+  status(): FleetControlPlaneStatus {
+    const state: FleetControlPlaneState = this.#consecutiveFailures < this.#failureThreshold
+      ? 'closed'
+      : this.#retryAtMs !== undefined && this.#now() < this.#retryAtMs
+        ? 'open'
+        : 'half-open'
+    return {
+      state,
+      consecutiveFailures: this.#consecutiveFailures,
+      timeoutMs: this.#timeoutMs,
+      failureThreshold: this.#failureThreshold,
+      resetTimeoutMs: this.#resetTimeoutMs,
+      ...(this.#lastFailureAtMs === undefined ? {} : { lastFailureAtMs: this.#lastFailureAtMs }),
+      ...(this.#retryAtMs === undefined ? {} : { retryAtMs: this.#retryAtMs }),
+      ...(this.#lastError === undefined ? {} : { lastError: this.#lastError }),
+    }
+  }
+
+  async probe(roster: () => Promise<RosterEntry>): Promise<RosterEntry> {
+    if (this.#probeInFlight) return this.#probeInFlight
+    const status = this.status()
+    if (status.state === 'open') {
+      throw new FleetControlPlaneCircuitOpenError(status.retryAtMs!)
+    }
+
+    const probe = withTimeout(roster, this.#timeoutMs)
+      .then((result) => {
+        this.#recordSuccess()
+        return result
+      })
+      .catch((error: unknown) => {
+        this.recordFailure(error)
+        throw error
+      })
+      .finally(() => {
+        if (this.#probeInFlight === probe) this.#probeInFlight = undefined
+      })
+    this.#probeInFlight = probe
+    return probe
+  }
+
+  assertMutationAllowed(): void {
+    const status = this.status()
+    if (status.state === 'closed') return
+    throw new FleetControlPlaneCircuitOpenError(status.retryAtMs ?? this.#now(), status.state)
+  }
+
+  recordFailure(error: unknown): void {
+    const now = this.#now()
+    const wasOpen = this.#retryAtMs !== undefined && now < this.#retryAtMs
+    this.#lastFailureAtMs = now
+    this.#lastError = describeControlPlaneError(error)
+    if (wasOpen) return
+    this.#consecutiveFailures += 1
+    if (this.#consecutiveFailures >= this.#failureThreshold) {
+      this.#retryAtMs = now + this.#resetTimeoutMs
+    }
+  }
+
+  #recordSuccess(): void {
+    this.#consecutiveFailures = 0
+    this.#retryAtMs = undefined
+    this.#lastError = undefined
+  }
+}
+
+export function guardFleetControlPlane(
+  fleet: FleetClient,
+  circuit: FleetControlPlaneCircuit,
+): FleetClient {
+  const guardedMutation = async <T>(operation: () => Promise<T>): Promise<T> => {
+    circuit.assertMutationAllowed()
+    try {
+      return await operation()
+    } catch (error) {
+      if (isFleetControlPlaneFailure(error)) circuit.recordFailure(error)
+      throw error
+    }
+  }
+
+  return new Proxy(fleet, {
+    get(target, property) {
+      if (property === 'roster') {
+        return (): Promise<RosterEntry> => circuit.probe(() => target.roster())
+      }
+      if (property === 'spawn') {
+        return (input: SpawnInput): Promise<SpawnResult> => guardedMutation(() => target.spawn(input))
+      }
+      if (property === 'resume') {
+        return (input: Parameters<FleetClient['resume']>[0]): Promise<SpawnResult> =>
+          guardedMutation(() => target.resume(input))
+      }
+      const value = Reflect.get(target, property, target)
+      return typeof value === 'function' ? value.bind(target) : value
+    },
+  }) as FleetClient
+}
+
+export function isFleetControlPlaneFailure(error: unknown): boolean {
+  if (!(error instanceof Error)) return false
+  const candidate = error as Error & { code?: unknown }
+  if (error.name === 'TimeoutError' || error.name === 'AbortError') return true
+  if (typeof candidate.code === 'string' && [
+    'ECONNREFUSED',
+    'ECONNRESET',
+    'EPIPE',
+    'ETIMEDOUT',
+    'UND_ERR_CONNECT_TIMEOUT',
+  ].includes(candidate.code)) return true
+  return /(?:operation\s+timed\s+out|operation\s+was\s+aborted|no\s+running\s+broker|broker\s+unavailable|socket\s+hang\s+up)/iu
+    .test(error.message)
+}
+
+function describeControlPlaneError(error: unknown): string {
+  if (!(error instanceof Error)) return 'unknown control-plane failure'
+  const code = (error as Error & { code?: unknown }).code
+  const safeCode = typeof code === 'string' && /^[A-Z0-9_]{1,80}$/u.test(code) ? ` (${code})` : ''
+  // This value is exposed through `factory status`; do not persist arbitrary
+  // transport messages because they may contain URLs or credential material.
+  return `${error.name || 'Error'}${safeCode}`
+}
+
+function withTimeout<T>(operation: () => Promise<T>, timeoutMs: number): Promise<T> {
+  return new Promise<T>((resolve, reject) => {
+    let settled = false
+    const timer = setTimeout(() => {
+      if (settled) return
+      settled = true
+      reject(new FleetControlPlaneTimeoutError(timeoutMs))
+    }, timeoutMs)
+
+    Promise.resolve()
+      .then(operation)
+      .then(
+        (value) => {
+          if (settled) return
+          settled = true
+          clearTimeout(timer)
+          resolve(value)
+        },
+        (error: unknown) => {
+          if (settled) return
+          settled = true
+          clearTimeout(timer)
+          reject(error)
+        },
+      )
+  })
+}
diff --git a/src/index.ts b/src/index.ts
index 438cc95..19f31e9 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -458,6 +458,7 @@ export type {
   TriageDecision,
   TriageEngine,
 } from './types'
+export type { FleetControlPlaneState, FleetControlPlaneStatus } from './fleet/control-plane-circuit'
 export {
   LOAD_EVIDENCE_CONTRACT,
   LoadMeasurementsSchema,
diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts
index c9dd3f4..57d6672 100644
--- a/src/orchestrator/factory.test.ts
+++ b/src/orchestrator/factory.test.ts
@@ -47,6 +47,36 @@ import {
 import { InternalFleetClient, type HarnessDriverClientLike } from '../fleet/internal-fleet-client'
 import type { ConversationMessage, ConversationSessionState, DiscoverySweepClaim, DiscoverySweepRenewal, DispatchLifecycle } from '../ports/state'
 
+describe('fleet control-plane admission', () => {
+  it('fails closed before discovery or spawn when the roster probe stalls', async () => {
+    class StalledRosterFleet extends FakeFleetClient {
+      override async roster(): Promise<never> {
+        return new Promise(() => undefined)
+      }
+    }
+
+    const mount = new FakeMountClient({ [issuePath(991)]: issueFile(991) })
+    const fleet = new StalledRosterFleet()
+    const factory = createFactory(config({
+      fleetHealth: {
+        rosterTimeoutMs: 100,
+        failureThreshold: 1,
+        resetTimeoutMs: 1_000,
+      },
+    }), { mount, fleet, triage: new StaticTriage(), logger: {} })
+
+    await expect(factory.runOnce()).rejects.toThrow(
+      'Factory dispatch paused because the fleet control plane is unavailable',
+    )
+    expect(fleet.spawns).toEqual([])
+    expect(mount.reads).toEqual([])
+    expect(factory.status().fleetControlPlane).toMatchObject({
+      state: 'open',
+      consecutiveFailures: 1,
+    })
+  })
+})
+
 const ready = 'b9bec744-b60c-4745-8022-d90d6ab59ae3'
 const implementing = '39b9881d-1196-4c95-8b80-a20f0c7263f7'
 const humanReview = '24462e2d-9946-4dd1-a798-931cdd678498'
diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts
index cd825fc..2ebb2d9 100644
--- a/src/orchestrator/factory.ts
+++ b/src/orchestrator/factory.ts
@@ -121,6 +121,7 @@ import {
 } from '../observability/events'
 import { boundedRunCostTotal, CostLedger, type RunCostTotal, type UnpricedModelCostRecord } from '../cost/ledger'
 import { createTicketDispatchDelivery, type TicketDispatchDelivery } from '../delivery/ticket-dispatch'
+import { FleetControlPlaneCircuit, guardFleetControlPlane } from '../fleet/control-plane-circuit'
 
 type FactoryEvent = 'issue-queued' | 'dispatched' | 'issue-done' | 'writeback-verified' | 'error'
 type Listener = (payload: FactoryEventPayload) => void
@@ -476,6 +477,7 @@ export class FactoryLoop implements Factory {
   readonly #mount: MountClient
   readonly #states: FactoryStateResolution
   readonly #fleet: FleetClient
+  readonly #fleetControlPlane: FleetControlPlaneCircuit
   readonly #ticketDispatchDelivery: TicketDispatchDelivery
   readonly #triage: TriageEngine
   readonly #linear: LinearWriteback
@@ -705,7 +707,6 @@ export class FactoryLoop implements Factory {
     // synced records (state.name but no state.id) without the states catalog.
     this.#states = ports.stateResolution ?? stateResolutionFromIds(config.stateIds, config.linear.states)
     installFactoryDraftPredicate(this.#mount, config)
-    this.#fleet = ports.fleet
     this.#ticketDispatchDelivery = ports.ticketDispatchDelivery ?? createTicketDispatchDelivery({
       mountRoot: config.localMountRoot,
     })
@@ -736,6 +737,13 @@ export class FactoryLoop implements Factory {
     this.#probePrResolver = ports.probePrResolver ?? ((issue) => this.#resolveIssuePr(issue))
     this.#logger = normalizeLogger(ports.logger ?? console)
     this.#clock = ports.clock ?? realClock
+    this.#fleetControlPlane = new FleetControlPlaneCircuit({
+      timeoutMs: config.fleetHealth.rosterTimeoutMs,
+      failureThreshold: config.fleetHealth.failureThreshold,
+      resetTimeoutMs: config.fleetHealth.resetTimeoutMs,
+      now: () => this.#clock.now(),
+    })
+    this.#fleet = guardFleetControlPlane(ports.fleet, this.#fleetControlPlane)
     this.#processIdentityReader = ports.processIdentityReader ?? readProcessIdentity
     this.#processFinder = ports.processFinder ?? ((agentName, opts) => findAgentProcessByName(agentName, {
       readProcessIdentity: this.#processIdentityReader,
@@ -2131,6 +2139,23 @@ export class FactoryLoop implements Factory {
 
   async #runOnceWithDiscoveryFence(opts: { dryRun?: boolean }): Promise<IterationReport> {
     const sweepStartedAtMs = this.#clock.now()
+    if (!(opts.dryRun ?? this.#config.dryRun)) {
+      try {
+        await this.#fleet.roster()
+        this.#increment('fleetControlPlaneProbeSuccesses')
+      } catch (error) {
+        const health = this.#fleetControlPlane.status()
+        this.#increment('fleetControlPlaneProbeFailures')
+        if (health.state === 'open') this.#increment('fleetControlPlaneCircuitOpen')
+        this.#logger.error?.('[factory] fleet control plane unavailable; dispatch paused', {
+          state: health.state,
+          consecutiveFailures: health.consecutiveFailures,
+          retryAtMs: health.retryAtMs,
+          error: describeError(error).errorMessage,
+        })
+        throw contextualError('Factory dispatch paused because the fleet control plane is unavailable', error)
+      }
+    }
     let claim = await this.#state.claimDiscoverySweep(
       this.#workspaceId,
       this.#discoverySweepOwner,
@@ -3943,6 +3968,7 @@ export class FactoryLoop implements Factory {
         capacityBlocked: parked.capacityBlocked,
       })) ?? [],
       counters: { ...this.#counters },
+      fleetControlPlane: this.#fleetControlPlane.status(),
       slackDegraded: this.#slackDegraded,
       slackDegradedReason: this.#slackDegradedReason,
       eventListener: this.#eventListenerStatus(),
diff --git a/src/types.ts b/src/types.ts
index 2e27261..ea1eb72 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -12,6 +12,7 @@ import type { DispatchRelayflowOptions, RelayflowPolicyRegistry } from './dispat
 import type { VerificationGate } from './environments/verification-pipeline'
 import type { CostLedger } from './cost/ledger'
 import type { TicketDispatchDelivery } from './delivery/ticket-dispatch'
+import type { FleetControlPlaneStatus } from './fleet/control-plane-circuit'
 
 export interface FactoryPorts {
   mount: MountClient
@@ -278,6 +279,8 @@ export interface FactoryStatus {
     capacityBlocked: boolean
   }>
   counters: Record<string, number>
+  /** Broker/fleet mutation gate. An open circuit blocks new workers until a successful half-open roster probe. */
+  fleetControlPlane: FleetControlPlaneStatus
   slackDegraded?: boolean
   slackDegradedReason?: string
   /** Primary Relayfile subscription/poll registration, not event activity. */

From b029914638c0d5c15c18abba02fb0b640cfb7b48 Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 10:36:52 +0200
Subject: [PATCH 2/8] fix: canonicalize dedicated broker state paths

---
 src/cli/fleet.test.ts | 14 ++++++++++++++
 src/cli/fleet.ts      | 30 +++++++++++++++++++++++++++---
 2 files changed, 41 insertions(+), 3 deletions(-)

diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index 0442db6..f85aad3 100644
--- a/src/cli/fleet.test.ts
+++ b/src/cli/fleet.test.ts
@@ -527,12 +527,26 @@ describe('fleet CLI parsing', () => {
         .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 })
+    }
+  })
 })
 
 describe('parseGithubIssueSelector normalization matrix', () => {
diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts
index f5ae51e..f90e969 100644
--- a/src/cli/fleet.ts
+++ b/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'
@@ -1639,7 +1639,10 @@ export function resolveFactoryBrokerConnectionPath(
     ...env,
     AGENT_RELAY_STATE_DIR: '',
   })
-  if (connectionPath === inheritedConnectionPath) {
+  const projectStateDir = inheritedConnectionPath
+    ? dirname(inheritedConnectionPath)
+    : join(resolve(startCwd), '.agentworkforce', 'relay')
+  if (sameFilesystemPath(explicitStateDir, projectStateDir)) {
     throw new Error(
       'AGENT_RELAY_STATE_DIR resolves to the project broker; choose a separate state directory for Factory',
     )
@@ -1647,6 +1650,27 @@ export function resolveFactoryBrokerConnectionPath(
   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,

From 1bc3c06e52e0d2358b6309941e0327f5684872e9 Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 10:49:03 +0200
Subject: [PATCH 3/8] test: harden broker circuit admission evidence

---
 README.md                               |   9 ++
 src/cli/fleet.test.ts                   |  24 +++-
 src/cli/fleet.ts                        |  15 ++-
 src/config/schema.ts                    |  11 +-
 src/fleet/control-plane-circuit.test.ts | 150 +++++++++++++++++++++---
 src/fleet/control-plane-circuit.ts      |  17 ++-
 src/orchestrator/factory.test.ts        | 132 ++++++++++++++++++++-
 src/orchestrator/factory.ts             |  20 +++-
 src/types.ts                            |   2 +
 9 files changed, 353 insertions(+), 27 deletions(-)

diff --git a/README.md b/README.md
index d18795c..977b0be 100644
--- a/README.md
+++ b/README.md
@@ -159,6 +159,15 @@ 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
diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index f85aad3..c31d8d7 100644
--- a/src/cli/fleet.test.ts
+++ b/src/cli/fleet.test.ts
@@ -2692,7 +2692,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')
@@ -2713,6 +2713,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 = {
@@ -2725,6 +2735,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
 
@@ -2744,6 +2761,11 @@ 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 })
diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts
index f90e969..4176677 100644
--- a/src/cli/fleet.ts
+++ b/src/cli/fleet.ts
@@ -1143,6 +1143,9 @@ async function factoryStatusWithMountHealth(
   const readinessReconcile = liveness.ok
     ? heartbeat?.readinessReconcile
     : observableStatus.readinessReconcile
+  const fleetControlPlane = liveness.ok
+    ? heartbeat?.fleetControlPlane ?? observableStatus.fleetControlPlane
+    : observableStatus.fleetControlPlane
   const eventListener = liveness.ok
     ? heartbeat?.eventListener ?? {
       state: 'unknown' as const,
@@ -1153,13 +1156,23 @@ async function factoryStatusWithMountHealth(
       reason: liveness.reason,
     }
   const health = mount.getLocalMountHealth?.()
-  if (!health) return { ...observableStatus, ...versionInfo, heldAgents, eventListener, readinessReconcile }
+  if (!health) {
+    return {
+      ...observableStatus,
+      ...versionInfo,
+      heldAgents,
+      eventListener,
+      readinessReconcile,
+      fleetControlPlane,
+    }
+  }
   return {
     ...observableStatus,
     ...versionInfo,
     heldAgents,
     eventListener,
     readinessReconcile,
+    fleetControlPlane,
     localMountDegraded: health.degraded,
     ...(health.reason ? { localMountDegradedReason: health.reason } : {}),
     ...(health.localDir ? { localMountRoot: health.localDir } : {}),
diff --git a/src/config/schema.ts b/src/config/schema.ts
index f7c5f1e..23009df 100644
--- a/src/config/schema.ts
+++ b/src/config/schema.ts
@@ -2,6 +2,11 @@ import { homedir } from 'node:os'
 import { isAbsolute, join } from 'node:path'
 
 import { z } from 'zod'
+import {
+  DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+  DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
+  DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+} from '../fleet/control-plane-circuit'
 
 import { KubernetesEnvironmentConfigSchema } from '../environments/connection-registry.js'
 
@@ -60,9 +65,9 @@ const dispatchSchema = z.object({
 const fleetHealthSchema = z.object({
   // Roster is read-only, so Factory can safely bound it locally. Mutating
   // spawn/resume calls are never abandoned behind a local timeout.
-  rosterTimeoutMs: z.number().int().min(100).max(60_000).default(5_000),
-  failureThreshold: z.number().int().min(1).max(10).default(2),
-  resetTimeoutMs: z.number().int().min(1_000).max(15 * 60_000).default(60_000),
+  rosterTimeoutMs: z.number().int().min(100).max(60_000).default(DEFAULT_FLEET_ROSTER_TIMEOUT_MS),
+  failureThreshold: z.number().int().min(1).max(10).default(DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD),
+  resetTimeoutMs: z.number().int().min(1_000).max(15 * 60_000).default(DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS),
   // Production launchers can require an explicit, non-project broker state
   // directory so Factory never silently shares an interactive broker.
   requireDedicatedBroker: z.boolean().default(false),
diff --git a/src/fleet/control-plane-circuit.test.ts b/src/fleet/control-plane-circuit.test.ts
index 6c8c7e8..5501f51 100644
--- a/src/fleet/control-plane-circuit.test.ts
+++ b/src/fleet/control-plane-circuit.test.ts
@@ -1,8 +1,11 @@
-import { describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 
 import type { RosterEntry } from '../ports/fleet'
 import { FakeFleetClient } from '../testing/fakes'
 import {
+  DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+  DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
+  DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
   FleetControlPlaneCircuit,
   FleetControlPlaneCircuitOpenError,
   guardFleetControlPlane,
@@ -12,51 +15,170 @@ import {
 const roster: RosterEntry = { agents: [], nodes: [] }
 
 describe('FleetControlPlaneCircuit', () => {
-  it('opens after bounded roster failures and rejects further probes without calling the broker', async () => {
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('MUST FIRE: two 5s roster timeouts open for 60s and only a successful half-open probe closes', async () => {
+    vi.useFakeTimers()
     let now = 1_000
     const circuit = new FleetControlPlaneCircuit({
-      timeoutMs: 5,
-      failureThreshold: 2,
-      resetTimeoutMs: 60_000,
+      timeoutMs: DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+      failureThreshold: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      resetTimeoutMs: DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
       now: () => now,
     })
     const never = vi.fn(() => new Promise<RosterEntry>(() => undefined))
 
-    await expect(circuit.probe(never)).rejects.toMatchObject({ name: 'TimeoutError' })
+    const first = circuit.probe(never)
+    const firstFailure = expect(first).rejects.toMatchObject({
+      name: 'TimeoutError',
+      timeoutMs: DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+    })
+    await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS - 1)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
+    await vi.advanceTimersByTimeAsync(1)
+    await firstFailure
     expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 })
-    await expect(circuit.probe(never)).rejects.toMatchObject({ name: 'TimeoutError' })
+
+    const second = circuit.probe(never)
+    const secondFailure = expect(second).rejects.toMatchObject({ name: 'TimeoutError' })
+    await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS)
+    await secondFailure
     expect(circuit.status()).toMatchObject({ state: 'open', consecutiveFailures: 2, retryAtMs: 61_000 })
 
     await expect(circuit.probe(never)).rejects.toBeInstanceOf(FleetControlPlaneCircuitOpenError)
     expect(never).toHaveBeenCalledTimes(2)
-    now += 60_000
+    now += DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS - 1
+    expect(circuit.status().state).toBe('open')
+    await expect(circuit.probe(never)).rejects.toBeInstanceOf(FleetControlPlaneCircuitOpenError)
+    expect(never).toHaveBeenCalledTimes(2)
+
+    now += 1
+    expect(circuit.status().state).toBe('half-open')
     await expect(circuit.probe(async () => roster)).resolves.toEqual(roster)
     expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
   })
 
-  it('coalesces concurrent roster probes', async () => {
+  it('MUST NOT FIRE: a slow 4.999s probe and one isolated 5s failure leave the circuit closed', async () => {
+    vi.useFakeTimers()
     let resolveProbe: ((value: RosterEntry) => void) | undefined
-    const call = vi.fn(() => new Promise<RosterEntry>((resolve) => { resolveProbe = resolve }))
+    const circuit = new FleetControlPlaneCircuit({
+      timeoutMs: DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+      failureThreshold: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      resetTimeoutMs: DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
+    })
+    const slow = vi.fn(() => new Promise<RosterEntry>((resolve) => { resolveProbe = resolve }))
+
+    const slowProbe = circuit.probe(slow)
+    await Promise.resolve()
+    await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS - 1)
+    resolveProbe?.(roster)
+    await expect(slowProbe).resolves.toEqual(roster)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
+
+    const isolated = circuit.probe(() => new Promise<RosterEntry>(() => undefined))
+    const isolatedFailure = expect(isolated).rejects.toMatchObject({ name: 'TimeoutError' })
+    await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS)
+    await isolatedFailure
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 })
+
+    await expect(circuit.probe(async () => roster)).resolves.toEqual(roster)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
+  })
+
+  it('coalesces a rejected probe, rejects every waiter, and does not cache the failure', async () => {
+    const sharedError = Object.assign(new Error('socket hang up'), { code: 'ECONNRESET' })
+    let rejectProbe: ((error: Error) => void) | undefined
+    const call = vi.fn()
+      .mockImplementationOnce(() => new Promise<RosterEntry>((_resolve, reject) => { rejectProbe = reject }))
+      .mockResolvedValueOnce(roster)
     const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 2, resetTimeoutMs: 1_000 })
     const first = circuit.probe(call)
     const second = circuit.probe(call)
     await Promise.resolve()
-    resolveProbe?.(roster)
-    await expect(Promise.all([first, second])).resolves.toEqual([roster, roster])
+    rejectProbe?.(sharedError)
+
+    const results = await Promise.allSettled([first, second])
+    expect(results).toEqual([
+      { status: 'rejected', reason: sharedError },
+      { status: 'rejected', reason: sharedError },
+    ])
     expect(call).toHaveBeenCalledTimes(1)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 })
+
+    await expect(circuit.probe(call)).resolves.toEqual(roster)
+    expect(call).toHaveBeenCalledTimes(2)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
   })
 
-  it('blocks new spawns while open without force-timing an accepted mutation', async () => {
-    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 5, failureThreshold: 1, resetTimeoutMs: 60_000 })
+  it('does not start a mutation until the shared admission probe has completed', async () => {
+    let resolveProbe: ((value: RosterEntry) => void) | undefined
     const fleet = new FakeFleetClient()
+    const rosterProbe = vi.spyOn(fleet, 'roster')
+      .mockImplementation(() => new Promise<RosterEntry>((resolve) => { resolveProbe = resolve }))
+    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 2, resetTimeoutMs: 1_000 })
+    const guarded = guardFleetControlPlane(fleet, circuit)
+
+    const spawning = guarded.spawn({ name: 'pending-worker', capability: 'spawn:codex' })
+    await Promise.resolve()
+    expect(rosterProbe).toHaveBeenCalledTimes(1)
+    expect(fleet.spawns).toEqual([])
+
+    resolveProbe?.(roster)
+    await expect(spawning).resolves.toMatchObject({ name: 'pending-worker' })
+    expect(fleet.spawns).toHaveLength(1)
+  })
+
+  it('MUST FIRE at mutation admission: two wedged rosters open the circuit and the next spawn fails fast', async () => {
+    vi.useFakeTimers()
+    const fleet = new FakeFleetClient()
+    const rosterProbe = vi.spyOn(fleet, 'roster')
+      .mockImplementation(() => new Promise<RosterEntry>(() => undefined))
+    const circuit = new FleetControlPlaneCircuit({
+      timeoutMs: DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+      failureThreshold: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      resetTimeoutMs: DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
+    })
+    const guarded = guardFleetControlPlane(fleet, circuit)
+
+    const first = guarded.spawn({ name: 'worker-1', capability: 'spawn:codex' })
+    const firstFailure = expect(first).rejects.toMatchObject({ name: 'TimeoutError' })
+    await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS)
+    await firstFailure
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 })
+
+    const second = guarded.spawn({ name: 'worker-2', capability: 'spawn:codex' })
+    const secondFailure = expect(second).rejects.toMatchObject({ name: 'TimeoutError' })
+    await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS)
+    await secondFailure
+    expect(circuit.status()).toMatchObject({ state: 'open', consecutiveFailures: 2 })
+
+    await expect(guarded.spawn({ name: 'worker-3', capability: 'spawn:codex' }))
+      .rejects.toBeInstanceOf(FleetControlPlaneCircuitOpenError)
+    expect(rosterProbe).toHaveBeenCalledTimes(2)
+    expect(fleet.spawns).toEqual([])
+  })
+
+  it('blocks spawn and resume while open without calling roster or either mutation', async () => {
+    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 5, failureThreshold: 2, resetTimeoutMs: 60_000 })
+    const fleet = new FakeFleetClient()
+    const rosterProbe = vi.spyOn(fleet, 'roster')
     const guarded = guardFleetControlPlane(fleet, circuit)
     circuit.recordFailure(new Error('broker unavailable'))
+    circuit.recordFailure(new Error('broker unavailable'))
 
     await expect(guarded.spawn({
       name: 'blocked-worker',
       capability: 'spawn:codex',
     })).rejects.toBeInstanceOf(FleetControlPlaneCircuitOpenError)
+    await expect(guarded.resume({
+      name: 'blocked-worker',
+      sessionRef: 'session-blocked',
+    })).rejects.toBeInstanceOf(FleetControlPlaneCircuitOpenError)
+    expect(rosterProbe).not.toHaveBeenCalled()
     expect(fleet.spawns).toEqual([])
+    expect(fleet.resumes).toEqual([])
   })
 
   it('recognizes timeout and transport failures without classifying domain errors', () => {
diff --git a/src/fleet/control-plane-circuit.ts b/src/fleet/control-plane-circuit.ts
index 123dd4d..d7aaa41 100644
--- a/src/fleet/control-plane-circuit.ts
+++ b/src/fleet/control-plane-circuit.ts
@@ -1,5 +1,9 @@
 import type { FleetClient, RosterEntry, SpawnInput, SpawnResult } from '../ports/fleet'
 
+export const DEFAULT_FLEET_ROSTER_TIMEOUT_MS = 5_000
+export const DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD = 2
+export const DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS = 60_000
+
 export type FleetControlPlaneState = 'closed' | 'open' | 'half-open'
 
 export interface FleetControlPlaneStatus {
@@ -135,13 +139,14 @@ export function guardFleetControlPlane(
   circuit: FleetControlPlaneCircuit,
 ): FleetClient {
   const guardedMutation = async <T>(operation: () => Promise<T>): Promise<T> => {
+    // A fresh Factory instance starts with a closed circuit, so checking state
+    // alone would let direct `factory dispatch` calls bypass admission. Probe
+    // the same roster path before every spawn/resume. Concurrent mutations
+    // coalesce onto one in-flight probe, and an open circuit rejects here
+    // without calling either roster or the mutation.
+    await circuit.probe(() => fleet.roster())
     circuit.assertMutationAllowed()
-    try {
-      return await operation()
-    } catch (error) {
-      if (isFleetControlPlaneFailure(error)) circuit.recordFailure(error)
-      throw error
-    }
+    return await operation()
   }
 
   return new Proxy(fleet, {
diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts
index 57d6672..d2377de 100644
--- a/src/orchestrator/factory.test.ts
+++ b/src/orchestrator/factory.test.ts
@@ -46,9 +46,139 @@ import {
 } from '../subscriptions'
 import { InternalFleetClient, type HarnessDriverClientLike } from '../fleet/internal-fleet-client'
 import type { ConversationMessage, ConversationSessionState, DiscoverySweepClaim, DiscoverySweepRenewal, DispatchLifecycle } from '../ports/state'
+import {
+  DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+  DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+} from '../fleet/control-plane-circuit'
 
 describe('fleet control-plane admission', () => {
-  it('fails closed before discovery or spawn when the roster probe stalls', async () => {
+  class StalledRosterFleet extends FakeFleetClient {
+    rosterCalls = 0
+
+    override async roster(): Promise<never> {
+      this.rosterCalls += 1
+      return new Promise(() => undefined)
+    }
+  }
+
+  it('fails closed at the default boundary before discovery and opens after two stalls', async () => {
+    vi.useFakeTimers()
+    try {
+      const mount = new FakeMountClient({ [issuePath(991)]: issueFile(991) })
+      const fleet = new StalledRosterFleet()
+      const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage(), logger: {} })
+
+      const first = factory.runOnce()
+      const firstFailure = expect(first).rejects.toThrow(
+        'Factory dispatch paused because the fleet control plane is unavailable',
+      )
+      await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS)
+      await firstFailure
+      expect(factory.status().fleetControlPlane).toMatchObject({
+        state: 'closed',
+        consecutiveFailures: 1,
+        failureThreshold: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      })
+
+      const second = factory.runOnce()
+      const secondFailure = expect(second).rejects.toThrow(
+        'Factory dispatch paused because the fleet control plane is unavailable',
+      )
+      await vi.advanceTimersByTimeAsync(DEFAULT_FLEET_ROSTER_TIMEOUT_MS)
+      await secondFailure
+
+      expect(fleet.rosterCalls).toBe(2)
+      expect(fleet.spawns).toEqual([])
+      expect(mount.reads).toEqual([])
+      expect(factory.status().fleetControlPlane).toMatchObject({
+        state: 'open',
+        consecutiveFailures: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      })
+
+      await expect(factory.runOnce()).rejects.toThrow(
+        'Factory dispatch paused because the fleet control plane is unavailable',
+      )
+      expect(fleet.rosterCalls).toBe(2)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('gates a direct dispatch at mutation admission instead of bypassing the probe', async () => {
+    vi.useRealTimers()
+    const path = issuePath(992)
+    const file = issueFile(992)
+    const mount = new FakeMountClient({ [path]: file })
+    const fleet = new StalledRosterFleet()
+    const factory = createFactory(config({
+      fleetHealth: {
+        rosterTimeoutMs: 100,
+        failureThreshold: 1,
+        resetTimeoutMs: 60_000,
+        requireDedicatedBroker: false,
+      },
+    }), { mount, fleet, triage: new StaticTriage(), logger: {} })
+    const decision = await factory.triageIssue(parseLinearIssue(path, file))
+
+    await expect(factory.dispatch(decision)).rejects.toMatchObject({
+      cause: {
+        name: 'FleetControlPlaneCircuitOpenError',
+        code: 'FACTORY_FLEET_CONTROL_CIRCUIT_OPEN',
+      },
+    })
+
+    expect(fleet.rosterCalls).toBe(1)
+    expect(fleet.spawns).toEqual([])
+    expect(factory.status().fleetControlPlane.state).toBe('open')
+  })
+
+  it('rejects a bounded loop when the circuit opens and persists the paused state in its heartbeat', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'factory-fleet-circuit-loop-'))
+    vi.useRealTimers()
+    try {
+      const heartbeatPath = join(root, 'heartbeat.json')
+      const registryPath = join(root, 'registry.json')
+      const fleet = new StalledRosterFleet()
+      const factory = createFactory(config({
+        fleetHealth: {
+          rosterTimeoutMs: 100,
+          failureThreshold: 2,
+          resetTimeoutMs: 60_000,
+          requireDedicatedBroker: false,
+        },
+        loop: {
+          maxIterations: 3,
+          maxConsecutiveFailures: 3,
+          heartbeatPath,
+          registryPath,
+        },
+      }), {
+        mount: new FakeMountClient(),
+        fleet,
+        triage: new StaticTriage(),
+        logger: {},
+      })
+
+      await expect(factory.runLoop()).rejects.toMatchObject({
+        code: 'FACTORY_FLEET_CONTROL_CIRCUIT_OPEN',
+      })
+
+      expect(fleet.rosterCalls).toBe(2)
+      expect(await readFactoryLoopHeartbeat(heartbeatPath)).toMatchObject({
+        status: 'stopping',
+        fleetControlPlane: {
+          state: 'open',
+          consecutiveFailures: 2,
+          retryAtMs: expect.any(Number),
+        },
+      })
+    } finally {
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+
+  it('supports a deliberately tighter configured bound', async () => {
+    vi.useRealTimers()
     class StalledRosterFleet extends FakeFleetClient {
       override async roster(): Promise<never> {
         return new Promise(() => undefined)
diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts
index 2ebb2d9..31c391d 100644
--- a/src/orchestrator/factory.ts
+++ b/src/orchestrator/factory.ts
@@ -121,7 +121,11 @@ import {
 } from '../observability/events'
 import { boundedRunCostTotal, CostLedger, type RunCostTotal, type UnpricedModelCostRecord } from '../cost/ledger'
 import { createTicketDispatchDelivery, type TicketDispatchDelivery } from '../delivery/ticket-dispatch'
-import { FleetControlPlaneCircuit, guardFleetControlPlane } from '../fleet/control-plane-circuit'
+import {
+  FleetControlPlaneCircuit,
+  FleetControlPlaneCircuitOpenError,
+  guardFleetControlPlane,
+} from '../fleet/control-plane-circuit'
 
 type FactoryEvent = 'issue-queued' | 'dispatched' | 'issue-done' | 'writeback-verified' | 'error'
 type Listener = (payload: FactoryEventPayload) => void
@@ -3492,6 +3496,19 @@ export class FactoryLoop implements Factory {
           reports.push(failedIterationReport(error, opts.dryRun ?? this.#config.dryRun))
           await this.#reapDispatchFailureHandoffsNow(heartbeatPath, registryPath)
           await this.#writeLoopHeartbeat(heartbeatPath, registryPath, 'running', iteration + 1, maxIterations)
+          const fleetControlPlane = this.#fleetControlPlane.status()
+          if (fleetControlPlane.state !== 'closed') {
+            this.#increment('loopCircuitBreaks')
+            this.#logger.error?.('[factory] stopping loop because dispatch is paused by the fleet control-plane circuit', {
+              state: fleetControlPlane.state,
+              consecutiveFailures: fleetControlPlane.consecutiveFailures,
+              retryAtMs: fleetControlPlane.retryAtMs,
+            })
+            throw new FleetControlPlaneCircuitOpenError(
+              fleetControlPlane.retryAtMs ?? this.#clock.now(),
+              fleetControlPlane.state,
+            )
+          }
           if (consecutiveFailures >= maxConsecutiveFailures) {
             this.#increment('loopCircuitBreaks')
             this.#logger.error?.('[factory] stopping loop after consecutive iteration failures', {
@@ -6704,6 +6721,7 @@ export class FactoryLoop implements Factory {
       registryPath,
       eventListener: this.#eventListenerStatus(),
       readinessReconcile: this.#readinessReconcileStatus(),
+      fleetControlPlane: this.#fleetControlPlane.status(),
     }
     await mkdir(dirname(path), { recursive: true })
     await writeFile(path, `${JSON.stringify(heartbeat, null, 2)}\n`, 'utf8')
diff --git a/src/types.ts b/src/types.ts
index ea1eb72..a8ea111 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -129,6 +129,8 @@ export interface FactoryLoopHeartbeat {
   registryPath?: string
   eventListener?: FactoryEventListenerStatus
   readinessReconcile?: FactoryReadinessReconcileStatus
+  /** Daemon-owned dispatch admission state; status readers must prefer this over a fresh local Factory instance. */
+  fleetControlPlane?: FleetControlPlaneStatus
 }
 
 export interface FactoryReadinessReconcileStatus {

From 6ddb5a9e23569d683fa93161137cb0d605a7504f Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 10:52:03 +0200
Subject: [PATCH 4/8] docs: clarify broker admission paths

---
 src/fleet/control-plane-circuit.ts | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/fleet/control-plane-circuit.ts b/src/fleet/control-plane-circuit.ts
index d7aaa41..1277d05 100644
--- a/src/fleet/control-plane-circuit.ts
+++ b/src/fleet/control-plane-circuit.ts
@@ -140,10 +140,10 @@ export function guardFleetControlPlane(
 ): FleetClient {
   const guardedMutation = async <T>(operation: () => Promise<T>): Promise<T> => {
     // A fresh Factory instance starts with a closed circuit, so checking state
-    // alone would let direct `factory dispatch` calls bypass admission. Probe
-    // the same roster path before every spawn/resume. Concurrent mutations
-    // coalesce onto one in-flight probe, and an open circuit rejects here
-    // without calling either roster or the mutation.
+    // alone would let resume/cold-start paths that did not run discovery bypass
+    // admission. Probe the same roster path before every spawn/resume.
+    // Concurrent mutations coalesce onto one in-flight probe, and an open
+    // circuit rejects here without calling either roster or the mutation.
     await circuit.probe(() => fleet.roster())
     circuit.assertMutationAllowed()
     return await operation()

From 168e760f183f3053452f94900fa78c9d38a258df Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 11:05:09 +0200
Subject: [PATCH 5/8] fix: harden broker admission recovery

---
 src/cli/fleet.test.ts            | 36 ++++++++++++++++++++++++++-
 src/cli/fleet.ts                 | 12 ++++++---
 src/orchestrator/factory.test.ts | 42 +++++++++++++++++++++++++-------
 src/orchestrator/factory.ts      | 40 ++++++++++++++++++------------
 4 files changed, 101 insertions(+), 29 deletions(-)

diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index c31d8d7..38feb17 100644
--- a/src/cli/fleet.test.ts
+++ b/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'
 
@@ -547,6 +547,40 @@ describe('fleet CLI parsing', () => {
       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', () => {
diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts
index 4176677..1415038 100644
--- a/src/cli/fleet.ts
+++ b/src/cli/fleet.ts
@@ -1652,10 +1652,14 @@ export function resolveFactoryBrokerConnectionPath(
     ...env,
     AGENT_RELAY_STATE_DIR: '',
   })
-  const projectStateDir = inheritedConnectionPath
-    ? dirname(inheritedConnectionPath)
-    : join(resolve(startCwd), '.agentworkforce', 'relay')
-  if (sameFilesystemPath(explicitStateDir, projectStateDir)) {
+  // 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',
     )
diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts
index d2377de..58a05f6 100644
--- a/src/orchestrator/factory.test.ts
+++ b/src/orchestrator/factory.test.ts
@@ -104,12 +104,13 @@ describe('fleet control-plane admission', () => {
     }
   })
 
-  it('gates a direct dispatch at mutation admission instead of bypassing the probe', async () => {
+  it('gates a direct dispatch before it consumes an attempt or reaches mutation admission', async () => {
     vi.useRealTimers()
     const path = issuePath(992)
     const file = issueFile(992)
     const mount = new FakeMountClient({ [path]: file })
     const fleet = new StalledRosterFleet()
+    const stateStore = new InMemoryStateStore({ batchSize: 2 })
     const factory = createFactory(config({
       fleetHealth: {
         rosterTimeoutMs: 100,
@@ -117,21 +118,43 @@ describe('fleet control-plane admission', () => {
         resetTimeoutMs: 60_000,
         requireDedicatedBroker: false,
       },
-    }), { mount, fleet, triage: new StaticTriage(), logger: {} })
+    }), { mount, fleet, stateStore, triage: new StaticTriage(), logger: {} })
     const decision = await factory.triageIssue(parseLinearIssue(path, file))
 
-    await expect(factory.dispatch(decision)).rejects.toMatchObject({
-      cause: {
-        name: 'FleetControlPlaneCircuitOpenError',
-        code: 'FACTORY_FLEET_CONTROL_CIRCUIT_OPEN',
-      },
-    })
+    await expect(factory.dispatch(decision)).rejects.toThrow(
+      'Factory dispatch paused because the fleet control plane is unavailable',
+    )
 
     expect(fleet.rosterCalls).toBe(1)
     expect(fleet.spawns).toEqual([])
+    await expect(stateStore.getDispatchAttempts('factory-test', issueKey(decision.issue))).resolves.toBeUndefined()
     expect(factory.status().fleetControlPlane.state).toBe('open')
   })
 
+  it('logs only the sanitized circuit error when roster admission fails', async () => {
+    class RejectingRosterFleet extends FakeFleetClient {
+      override async roster(): Promise<never> {
+        throw Object.assign(new Error('https://broker.invalid?token=must-not-leak'), { code: 'ECONNREFUSED' })
+      }
+    }
+    const error = vi.fn()
+    const factory = createFactory(config({ fleetHealth: { failureThreshold: 1 } }), {
+      mount: new FakeMountClient(),
+      fleet: new RejectingRosterFleet(),
+      triage: new StaticTriage(),
+      logger: { error },
+    })
+
+    await expect(factory.runOnce()).rejects.toThrow(
+      'Factory dispatch paused because the fleet control plane is unavailable',
+    )
+    expect(error).toHaveBeenCalledWith(
+      '[factory] fleet control plane unavailable; dispatch paused',
+      expect.objectContaining({ error: 'Error (ECONNREFUSED)' }),
+    )
+    expect(JSON.stringify(error.mock.calls)).not.toContain('must-not-leak')
+  })
+
   it('rejects a bounded loop when the circuit opens and persists the paused state in its heartbeat', async () => {
     const root = await mkdtemp(join(tmpdir(), 'factory-fleet-circuit-loop-'))
     vi.useRealTimers()
@@ -213,10 +236,11 @@ const humanReview = '24462e2d-9946-4dd1-a798-931cdd678498'
 const done = '83ea5383-bfe9-425a-86ef-517b8190f09a'
 const planning = '3de351f2-90e6-4731-aa6b-4a55b77f481e'
 
-type FactoryConfigOverrides = Omit<Partial<FactoryConfig>, 'dispatch' | 'loop' | 'safety'> & {
+type FactoryConfigOverrides = Omit<Partial<FactoryConfig>, 'dispatch' | 'loop' | 'safety' | 'fleetHealth'> & {
   dispatch?: Partial<FactoryConfig['dispatch']>
   loop?: Partial<FactoryConfig['loop']>
   safety?: Partial<FactoryConfig['safety']>
+  fleetHealth?: Partial<FactoryConfig['fleetHealth']>
 }
 
 const config = (overrides: FactoryConfigOverrides = {}): FactoryConfig => FactoryConfigSchema.parse({
diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts
index 31c391d..08f2481 100644
--- a/src/orchestrator/factory.ts
+++ b/src/orchestrator/factory.ts
@@ -2141,24 +2141,28 @@ export class FactoryLoop implements Factory {
     }
   }
 
+  async #assertFleetControlPlaneAvailable(): Promise<void> {
+    try {
+      await this.#fleet.roster()
+      this.#increment('fleetControlPlaneProbeSuccesses')
+    } catch (error) {
+      const health = this.#fleetControlPlane.status()
+      this.#increment('fleetControlPlaneProbeFailures')
+      if (health.state === 'open') this.#increment('fleetControlPlaneCircuitOpen')
+      this.#logger.error?.('[factory] fleet control plane unavailable; dispatch paused', {
+        state: health.state,
+        consecutiveFailures: health.consecutiveFailures,
+        retryAtMs: health.retryAtMs,
+        error: health.lastError ?? 'unknown control-plane failure',
+      })
+      throw contextualError('Factory dispatch paused because the fleet control plane is unavailable', error)
+    }
+  }
+
   async #runOnceWithDiscoveryFence(opts: { dryRun?: boolean }): Promise<IterationReport> {
     const sweepStartedAtMs = this.#clock.now()
     if (!(opts.dryRun ?? this.#config.dryRun)) {
-      try {
-        await this.#fleet.roster()
-        this.#increment('fleetControlPlaneProbeSuccesses')
-      } catch (error) {
-        const health = this.#fleetControlPlane.status()
-        this.#increment('fleetControlPlaneProbeFailures')
-        if (health.state === 'open') this.#increment('fleetControlPlaneCircuitOpen')
-        this.#logger.error?.('[factory] fleet control plane unavailable; dispatch paused', {
-          state: health.state,
-          consecutiveFailures: health.consecutiveFailures,
-          retryAtMs: health.retryAtMs,
-          error: describeError(error).errorMessage,
-        })
-        throw contextualError('Factory dispatch paused because the fleet control plane is unavailable', error)
-      }
+      await this.#assertFleetControlPlaneAvailable()
     }
     let claim = await this.#state.claimDiscoverySweep(
       this.#workspaceId,
@@ -3669,6 +3673,12 @@ export class FactoryLoop implements Factory {
       }
     }
     this.#clearDependencyPark(batch, dispatchDecision.issue)
+    // Event-driven and direct dispatches do not necessarily pass through issue
+    // discovery. Admit them before creating previews, claiming a lifecycle, or
+    // consuming a dispatch attempt. The mutation proxy probes again at the
+    // actual spawn/resume boundary so a later control-plane fault still fails
+    // closed.
+    if (!dryRun) await this.#assertFleetControlPlaneAvailable()
     const durableDispatch = !dryRun && this.#usesDurableDispatchLifecycle()
     // Local dispatches need the same deterministic branch identity as remote
     // ones. Without it, every worker starts in the configured shared checkout

From 48bd2dfdb2e5628dbd2a2bb1fb8e280a63650aeb Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 11:14:36 +0200
Subject: [PATCH 6/8] fix: report legacy daemon circuit as unknown

---
 src/cli/fleet.test.ts | 46 +++++++++++++++++++++++++++++++++++++++++++
 src/cli/fleet.ts      |  6 ++++--
 2 files changed, 50 insertions(+), 2 deletions(-)

diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts
index 38feb17..5c2dfe9 100644
--- a/src/cli/fleet.test.ts
+++ b/src/cli/fleet.test.ts
@@ -2806,6 +2806,52 @@ describe('fleet CLI runtime', () => {
     }
   })
 
+  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 })
+    }
+  })
+
   it('lists registry-backed in-flight issues, agents, and degraded claims in factory status', async () => {
     const root = await mkdtemp(join(tmpdir(), 'fleet-cli-registry-status-'))
     try {
diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts
index 1415038..8c35a28 100644
--- a/src/cli/fleet.ts
+++ b/src/cli/fleet.ts
@@ -1106,7 +1106,9 @@ async function factoryStatusWithMountHealth(
   registryPath: string,
   heartbeatStaleMs: number,
   versionInfo?: FactoryVersionInfo,
-): 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']
   version?: string
   installedAt?: string
   latestVersion?: string
@@ -1144,7 +1146,7 @@ async function factoryStatusWithMountHealth(
     ? heartbeat?.readinessReconcile
     : observableStatus.readinessReconcile
   const fleetControlPlane = liveness.ok
-    ? heartbeat?.fleetControlPlane ?? observableStatus.fleetControlPlane
+    ? heartbeat?.fleetControlPlane
     : observableStatus.fleetControlPlane
   const eventListener = liveness.ok
     ? heartbeat?.eventListener ?? {

From c320b3aa821006e1ea64f667a0f3a027649ebe06 Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 15:22:37 +0200
Subject: [PATCH 7/8] refactor: enforce roster-only circuit health

---
 src/fleet/control-plane-circuit.test.ts | 31 +++++++++++++++-------
 src/fleet/control-plane-circuit.ts      | 34 ++++++++++++-------------
 2 files changed, 38 insertions(+), 27 deletions(-)

diff --git a/src/fleet/control-plane-circuit.test.ts b/src/fleet/control-plane-circuit.test.ts
index 5501f51..7590861 100644
--- a/src/fleet/control-plane-circuit.test.ts
+++ b/src/fleet/control-plane-circuit.test.ts
@@ -9,7 +9,6 @@ import {
   FleetControlPlaneCircuit,
   FleetControlPlaneCircuitOpenError,
   guardFleetControlPlane,
-  isFleetControlPlaneFailure,
 } from './control-plane-circuit'
 
 const roster: RosterEntry = { agents: [], nodes: [] }
@@ -165,8 +164,8 @@ describe('FleetControlPlaneCircuit', () => {
     const fleet = new FakeFleetClient()
     const rosterProbe = vi.spyOn(fleet, 'roster')
     const guarded = guardFleetControlPlane(fleet, circuit)
-    circuit.recordFailure(new Error('broker unavailable'))
-    circuit.recordFailure(new Error('broker unavailable'))
+    await expect(circuit.probe(async () => { throw new Error('broker unavailable') })).rejects.toThrow()
+    await expect(circuit.probe(async () => { throw new Error('broker unavailable') })).rejects.toThrow()
 
     await expect(guarded.spawn({
       name: 'blocked-worker',
@@ -181,13 +180,27 @@ describe('FleetControlPlaneCircuit', () => {
     expect(fleet.resumes).toEqual([])
   })
 
-  it('recognizes timeout and transport failures without classifying domain errors', () => {
-    expect(isFleetControlPlaneFailure(Object.assign(new Error('connect failed'), { code: 'ECONNREFUSED' }))).toBe(true)
-    expect(isFleetControlPlaneFailure(Object.assign(new Error('aborted'), { name: 'AbortError' }))).toBe(true)
-    expect(isFleetControlPlaneFailure(new Error('agent already exists'))).toBe(false)
-
+  it('sanitizes probe failures before exposing them through circuit status', async () => {
     const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 1, resetTimeoutMs: 1_000 })
-    circuit.recordFailure(Object.assign(new Error('https://broker.invalid?token=must-not-leak'), { code: 'ECONNREFUSED' }))
+    await expect(circuit.probe(async () => {
+      throw Object.assign(new Error('https://broker.invalid?token=must-not-leak'), { code: 'ECONNREFUSED' })
+    })).rejects.toThrow()
     expect(circuit.status().lastError).toBe('Error (ECONNREFUSED)')
   })
+
+  it('does not treat ambiguous mutation transport failures as roster probe failures', async () => {
+    const fleet = new FakeFleetClient()
+    const mutationError = Object.assign(
+      new Error('Timed out waiting for spawn invocation inv-1 to complete'),
+      { code: 'ETIMEDOUT' },
+    )
+    vi.spyOn(fleet, 'spawn').mockRejectedValue(mutationError)
+    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 2, resetTimeoutMs: 1_000 })
+    const guarded = guardFleetControlPlane(fleet, circuit)
+
+    await expect(guarded.spawn({ name: 'worker-1', capability: 'spawn:codex' })).rejects.toBe(mutationError)
+    await expect(guarded.spawn({ name: 'worker-2', capability: 'spawn:codex' })).rejects.toBe(mutationError)
+
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
+  })
 })
diff --git a/src/fleet/control-plane-circuit.ts b/src/fleet/control-plane-circuit.ts
index 1277d05..f22a25c 100644
--- a/src/fleet/control-plane-circuit.ts
+++ b/src/fleet/control-plane-circuit.ts
@@ -68,6 +68,7 @@ export class FleetControlPlaneCircuit {
     this.#now = options.now ?? Date.now
   }
 
+  /** Returns the current admission state without performing broker I/O. */
   status(): FleetControlPlaneStatus {
     const state: FleetControlPlaneState = this.#consecutiveFailures < this.#failureThreshold
       ? 'closed'
@@ -86,12 +87,13 @@ export class FleetControlPlaneCircuit {
     }
   }
 
+  /** Runs or joins one bounded roster request, recording only its outcome. */
   async probe(roster: () => Promise<RosterEntry>): Promise<RosterEntry> {
-    if (this.#probeInFlight) return this.#probeInFlight
     const status = this.status()
     if (status.state === 'open') {
       throw new FleetControlPlaneCircuitOpenError(status.retryAtMs!)
     }
+    if (this.#probeInFlight) return this.#probeInFlight
 
     const probe = withTimeout(roster, this.#timeoutMs)
       .then((result) => {
@@ -99,7 +101,7 @@ export class FleetControlPlaneCircuit {
         return result
       })
       .catch((error: unknown) => {
-        this.recordFailure(error)
+        this.#recordFailure(error)
         throw error
       })
       .finally(() => {
@@ -109,13 +111,14 @@ export class FleetControlPlaneCircuit {
     return probe
   }
 
+  /** Rejects mutations until an open or half-open circuit has recovered. */
   assertMutationAllowed(): void {
     const status = this.status()
     if (status.state === 'closed') return
     throw new FleetControlPlaneCircuitOpenError(status.retryAtMs ?? this.#now(), status.state)
   }
 
-  recordFailure(error: unknown): void {
+  #recordFailure(error: unknown): void {
     const now = this.#now()
     const wasOpen = this.#retryAtMs !== undefined && now < this.#retryAtMs
     this.#lastFailureAtMs = now
@@ -134,6 +137,12 @@ export class FleetControlPlaneCircuit {
   }
 }
 
+/**
+ * Gates roster, spawn, and resume through the read-only roster control path.
+ * Mutation rejections deliberately do not affect circuit state: a transport
+ * error can arrive after the remote side effect committed, so only a fresh,
+ * bounded roster probe is allowed to decide subsequent admission.
+ */
 export function guardFleetControlPlane(
   fleet: FleetClient,
   circuit: FleetControlPlaneCircuit,
@@ -146,6 +155,8 @@ export function guardFleetControlPlane(
     // circuit rejects here without calling either roster or the mutation.
     await circuit.probe(() => fleet.roster())
     circuit.assertMutationAllowed()
+    // Do not catch or classify operation failures here. Their remote outcome
+    // can be ambiguous; the next mutation must run a new roster admission probe.
     return await operation()
   }
 
@@ -167,21 +178,7 @@ export function guardFleetControlPlane(
   }) as FleetClient
 }
 
-export function isFleetControlPlaneFailure(error: unknown): boolean {
-  if (!(error instanceof Error)) return false
-  const candidate = error as Error & { code?: unknown }
-  if (error.name === 'TimeoutError' || error.name === 'AbortError') return true
-  if (typeof candidate.code === 'string' && [
-    'ECONNREFUSED',
-    'ECONNRESET',
-    'EPIPE',
-    'ETIMEDOUT',
-    'UND_ERR_CONNECT_TIMEOUT',
-  ].includes(candidate.code)) return true
-  return /(?:operation\s+timed\s+out|operation\s+was\s+aborted|no\s+running\s+broker|broker\s+unavailable|socket\s+hang\s+up)/iu
-    .test(error.message)
-}
-
+/** Redacts arbitrary transport text before circuit state becomes observable. */
 function describeControlPlaneError(error: unknown): string {
   if (!(error instanceof Error)) return 'unknown control-plane failure'
   const code = (error as Error & { code?: unknown }).code
@@ -191,6 +188,7 @@ function describeControlPlaneError(error: unknown): string {
   return `${error.name || 'Error'}${safeCode}`
 }
 
+/** Applies the local roster deadline without imposing a timeout on mutations. */
 function withTimeout<T>(operation: () => Promise<T>, timeoutMs: number): Promise<T> {
   return new Promise<T>((resolve, reject) => {
     let settled = false

From 0672fca30f2c1fd6ab3fd847bd05bc7e7f98ff5c Mon Sep 17 00:00:00 2001
From: Khaliq <khaliqgant@gmail.com>
Date: Mon, 17 Aug 2026 15:29:21 +0200
Subject: [PATCH 8/8] fix: fence stale broker circuit probes

---
 src/fleet/control-plane-circuit.test.ts | 101 +++++++++++++++++++++---
 src/fleet/control-plane-circuit.ts      |  55 ++++++++++---
 2 files changed, 134 insertions(+), 22 deletions(-)

diff --git a/src/fleet/control-plane-circuit.test.ts b/src/fleet/control-plane-circuit.test.ts
index 7590861..0eb3104 100644
--- a/src/fleet/control-plane-circuit.test.ts
+++ b/src/fleet/control-plane-circuit.test.ts
@@ -9,6 +9,7 @@ import {
   FleetControlPlaneCircuit,
   FleetControlPlaneCircuitOpenError,
   guardFleetControlPlane,
+  isFleetControlPlaneFailure,
 } from './control-plane-circuit'
 
 const roster: RosterEntry = { agents: [], nodes: [] }
@@ -111,6 +112,75 @@ describe('FleetControlPlaneCircuit', () => {
     expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
   })
 
+  it('MUST FIRE: concurrent mutation faults cannot let a pre-open roster probe bypass or close the circuit', async () => {
+    let now = 1_000
+    let rejectSpawn: ((error: Error) => void) | undefined
+    let rejectResume: ((error: Error) => void) | undefined
+    let resolveStaleProbe: ((value: RosterEntry) => void) | undefined
+    const fleet = new FakeFleetClient()
+    const rosterProbe = vi.spyOn(fleet, 'roster')
+    const spawn = vi.spyOn(fleet, 'spawn')
+      .mockImplementation(() => new Promise((_resolve, reject) => { rejectSpawn = reject }))
+    const resume = vi.spyOn(fleet, 'resume')
+      .mockImplementation(() => new Promise((_resolve, reject) => { rejectResume = reject }))
+    const circuit = new FleetControlPlaneCircuit({
+      timeoutMs: DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+      failureThreshold: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      resetTimeoutMs: DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
+      now: () => now,
+    })
+    const guarded = guardFleetControlPlane(fleet, circuit)
+
+    const spawning = guarded.spawn({ name: 'worker-1', capability: 'spawn:codex' })
+    await vi.waitFor(() => { expect(spawn).toHaveBeenCalledTimes(1) })
+    const resuming = guarded.resume({ name: 'worker-2', sessionRef: 'session-2' })
+    await vi.waitFor(() => { expect(resume).toHaveBeenCalledTimes(1) })
+
+    rosterProbe.mockImplementationOnce(() => new Promise<RosterEntry>((resolve) => { resolveStaleProbe = resolve }))
+    const staleProbe = guarded.roster()
+    await vi.waitFor(() => { expect(resolveStaleProbe).toBeTypeOf('function') })
+
+    rejectSpawn?.(new Error('Timed out waiting for spawn invocation inv-spawn to complete (last status: pending)'))
+    rejectResume?.(new Error('Timed out waiting for resume invocation inv-resume to complete (last status: pending)'))
+    await Promise.allSettled([spawning, resuming])
+    const openedBeforeStaleProbeSettled = circuit.status()
+
+    let laterSettled = false
+    let laterError: unknown
+    const laterProbe = guarded.roster().then(
+      (result) => {
+        laterSettled = true
+        return result
+      },
+      (error: unknown) => {
+        laterSettled = true
+        laterError = error
+        throw error
+      },
+    )
+    void laterProbe.catch(() => undefined)
+    await Promise.resolve()
+    await Promise.resolve()
+    const laterFailedFast = laterSettled && laterError instanceof FleetControlPlaneCircuitOpenError
+
+    now += DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS
+    resolveStaleProbe?.(roster)
+    const outcomes = await Promise.allSettled([staleProbe, laterProbe])
+
+    expect(openedBeforeStaleProbeSettled).toMatchObject({
+      state: 'open',
+      consecutiveFailures: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      retryAtMs: 61_000,
+    })
+    expect(laterFailedFast).toBe(true)
+    expect(outcomes.map((outcome) => outcome.status)).toEqual(['rejected', 'rejected'])
+    expect(circuit.status()).toMatchObject({ state: 'half-open', consecutiveFailures: 2 })
+
+    rosterProbe.mockResolvedValueOnce(roster)
+    await expect(guarded.roster()).resolves.toEqual(roster)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
+  })
+
   it('does not start a mutation until the shared admission probe has completed', async () => {
     let resolveProbe: ((value: RosterEntry) => void) | undefined
     const fleet = new FakeFleetClient()
@@ -180,7 +250,14 @@ describe('FleetControlPlaneCircuit', () => {
     expect(fleet.resumes).toEqual([])
   })
 
-  it('sanitizes probe failures before exposing them through circuit status', async () => {
+  it('recognizes and sanitizes transport failures without classifying domain errors', async () => {
+    expect(isFleetControlPlaneFailure(Object.assign(new Error('connect failed'), { code: 'ECONNREFUSED' }))).toBe(true)
+    expect(isFleetControlPlaneFailure(Object.assign(new Error('aborted'), { name: 'AbortError' }))).toBe(true)
+    expect(isFleetControlPlaneFailure(
+      new Error('Timed out waiting for spawn invocation inv-1 to complete (last status: pending)'),
+    )).toBe(true)
+    expect(isFleetControlPlaneFailure(new Error('agent already exists'))).toBe(false)
+
     const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 1, resetTimeoutMs: 1_000 })
     await expect(circuit.probe(async () => {
       throw Object.assign(new Error('https://broker.invalid?token=must-not-leak'), { code: 'ECONNREFUSED' })
@@ -188,19 +265,23 @@ describe('FleetControlPlaneCircuit', () => {
     expect(circuit.status().lastError).toBe('Error (ECONNREFUSED)')
   })
 
-  it('does not treat ambiguous mutation transport failures as roster probe failures', async () => {
+  it('MUST NOT FIRE: one isolated mutation fault stays closed and domain errors are not counted', async () => {
     const fleet = new FakeFleetClient()
-    const mutationError = Object.assign(
-      new Error('Timed out waiting for spawn invocation inv-1 to complete'),
-      { code: 'ETIMEDOUT' },
-    )
-    vi.spyOn(fleet, 'spawn').mockRejectedValue(mutationError)
-    const circuit = new FleetControlPlaneCircuit({ timeoutMs: 100, failureThreshold: 2, resetTimeoutMs: 1_000 })
+    const transportError = new Error('Timed out waiting for spawn invocation inv-1 to complete (last status: pending)')
+    const domainError = new Error('agent already exists')
+    vi.spyOn(fleet, 'spawn').mockRejectedValueOnce(transportError)
+    vi.spyOn(fleet, 'resume').mockRejectedValueOnce(domainError)
+    const circuit = new FleetControlPlaneCircuit({
+      timeoutMs: DEFAULT_FLEET_ROSTER_TIMEOUT_MS,
+      failureThreshold: DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD,
+      resetTimeoutMs: DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS,
+    })
     const guarded = guardFleetControlPlane(fleet, circuit)
 
-    await expect(guarded.spawn({ name: 'worker-1', capability: 'spawn:codex' })).rejects.toBe(mutationError)
-    await expect(guarded.spawn({ name: 'worker-2', capability: 'spawn:codex' })).rejects.toBe(mutationError)
+    await expect(guarded.spawn({ name: 'worker-1', capability: 'spawn:codex' })).rejects.toBe(transportError)
+    expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 1 })
 
+    await expect(guarded.resume({ name: 'worker-1', sessionRef: 'session-1' })).rejects.toBe(domainError)
     expect(circuit.status()).toMatchObject({ state: 'closed', consecutiveFailures: 0 })
   })
 })
diff --git a/src/fleet/control-plane-circuit.ts b/src/fleet/control-plane-circuit.ts
index f22a25c..6a8d5b5 100644
--- a/src/fleet/control-plane-circuit.ts
+++ b/src/fleet/control-plane-circuit.ts
@@ -59,6 +59,9 @@ export class FleetControlPlaneCircuit {
   #lastFailureAtMs?: number
   #retryAtMs?: number
   #lastError?: string
+  // A roster request that predates an open transition is not a valid
+  // half-open recovery probe, even if it resolves after the cooldown.
+  #openGeneration = 0
   #probeInFlight?: Promise<RosterEntry>
 
   constructor(options: FleetControlPlaneCircuitOptions) {
@@ -95,15 +98,25 @@ export class FleetControlPlaneCircuit {
     }
     if (this.#probeInFlight) return this.#probeInFlight
 
+    const openGeneration = this.#openGeneration
     const probe = withTimeout(roster, this.#timeoutMs)
+      .catch((error: unknown) => {
+        this.recordFailure(error)
+        throw error
+      })
       .then((result) => {
+        const settledStatus = this.status()
+        // Mutation failures can open the circuit while this read is pending.
+        // Never let that stale result satisfy waiters or reset circuit state.
+        if (settledStatus.state === 'open' || openGeneration !== this.#openGeneration) {
+          throw new FleetControlPlaneCircuitOpenError(
+            settledStatus.retryAtMs ?? this.#now(),
+            settledStatus.state === 'open' ? 'open' : 'half-open',
+          )
+        }
         this.#recordSuccess()
         return result
       })
-      .catch((error: unknown) => {
-        this.#recordFailure(error)
-        throw error
-      })
       .finally(() => {
         if (this.#probeInFlight === probe) this.#probeInFlight = undefined
       })
@@ -118,7 +131,7 @@ export class FleetControlPlaneCircuit {
     throw new FleetControlPlaneCircuitOpenError(status.retryAtMs ?? this.#now(), status.state)
   }
 
-  #recordFailure(error: unknown): void {
+  recordFailure(error: unknown): void {
     const now = this.#now()
     const wasOpen = this.#retryAtMs !== undefined && now < this.#retryAtMs
     this.#lastFailureAtMs = now
@@ -127,6 +140,7 @@ export class FleetControlPlaneCircuit {
     this.#consecutiveFailures += 1
     if (this.#consecutiveFailures >= this.#failureThreshold) {
       this.#retryAtMs = now + this.#resetTimeoutMs
+      this.#openGeneration += 1
     }
   }
 
@@ -138,10 +152,9 @@ export class FleetControlPlaneCircuit {
 }
 
 /**
- * Gates roster, spawn, and resume through the read-only roster control path.
- * Mutation rejections deliberately do not affect circuit state: a transport
- * error can arrive after the remote side effect committed, so only a fresh,
- * bounded roster probe is allowed to decide subsequent admission.
+ * Gates spawn and resume with a bounded read-only roster admission. Transport
+ * failures from admitted mutations also count toward the circuit without
+ * abandoning or timing the mutation; domain rejections remain uncounted.
  */
 export function guardFleetControlPlane(
   fleet: FleetClient,
@@ -155,9 +168,12 @@ export function guardFleetControlPlane(
     // circuit rejects here without calling either roster or the mutation.
     await circuit.probe(() => fleet.roster())
     circuit.assertMutationAllowed()
-    // Do not catch or classify operation failures here. Their remote outcome
-    // can be ambiguous; the next mutation must run a new roster admission probe.
-    return await operation()
+    try {
+      return await operation()
+    } catch (error) {
+      if (isFleetControlPlaneFailure(error)) circuit.recordFailure(error)
+      throw error
+    }
   }
 
   return new Proxy(fleet, {
@@ -178,6 +194,21 @@ export function guardFleetControlPlane(
   }) as FleetClient
 }
 
+export function isFleetControlPlaneFailure(error: unknown): boolean {
+  if (!(error instanceof Error)) return false
+  const candidate = error as Error & { code?: unknown }
+  if (error.name === 'TimeoutError' || error.name === 'AbortError') return true
+  if (typeof candidate.code === 'string' && [
+    'ECONNREFUSED',
+    'ECONNRESET',
+    'EPIPE',
+    'ETIMEDOUT',
+    'UND_ERR_CONNECT_TIMEOUT',
+  ].includes(candidate.code)) return true
+  return /(?:operation\s+timed\s+out|timed\s+out\s+waiting\s+for\b.*\binvocation\b.*\bto\s+complete|operation\s+was\s+aborted|no\s+running\s+broker|broker\s+unavailable|socket\s+hang\s+up)/iu
+    .test(error.message)
+}
+
 /** Redacts arbitrary transport text before circuit state becomes observable. */
 function describeControlPlaneError(error: unknown): string {
   if (!(error instanceof Error)) return 'unknown control-plane failure'