fix(orchestrator): reap a lifecycle that took a batch slot and never placed an agent (#303)

src/cli/diagnose.test.ts

@@ -174,6 +174,64 @@ describe('factory diagnose --deployed (#295)', () => {
     expect(report.verdict).toContain('stalled')
   })
 
+  // #303: a wedged batch was the one dispatch-gating condition every surface
+  // reported as healthy. `factory diagnose` has to name it, or the operator is
+  // back to reading the state document.
+  it('names a wedged batch when everything else reads healthy', async () => {
+    const out = buffer()
+    const code = await runFleetCli(['diagnose', '--deployed', BASE, '--json'], {
+      stdout: out,
+      stderr: buffer(),
+      env: HERMETIC_ENV,
+      diagnoseFetch: stubFetch({
+        healthz: {
+          status: 200,
+          body: {
+            ok: true,
+            phase: 'running',
+            health: {
+              schemaVersion: 1,
+              ok: true,
+              status: 'degraded',
+              stale: false,
+              loopStatus: 'running',
+              degradedSubsystems: ['dispatchCapacity'],
+              readinessReconcile: {
+                state: 'healthy',
+                consecutiveFailures: 0,
+                failureThreshold: 3,
+                intervalMs: 60_000,
+                lastStartedAtMs: NOW_MS - 30_000,
+                lastCompletedAtMs: NOW_MS - 29_000,
+              },
+              eventListener: { state: 'subscribed' },
+              dispatchCapacity: {
+                state: 'stalled',
+                batchSize: 1,
+                active: 1,
+                waiting: 7,
+                waitWarnMs: 1_800_000,
+                agentlessHoldTimeoutMs: 1_800_000,
+                longestWaitMs: 46_800_000,
+                agentlessOccupants: 1,
+              },
+            },
+          },
+        },
+      }),
+    })
+
+    expect(code).not.toBe(0)
+    const report = JSON.parse(out.text()) as { dispatching: boolean; verdict: string }
+    expect(report.dispatching).toBe(false)
+    expect(report.verdict).toContain('7 issue(s) have been waiting for batch capacity')
+    expect(report.verdict).toContain('1/1 slot(s) occupied')
+    expect(report.verdict).toContain('never placed an agent and are past the 30m 0s reap deadline')
+    // `longestWaitMs` is a queue wait, so the verdict must not present it as
+    // how long the slots have been held (#303 review, cubic).
+    expect(report.verdict).not.toContain('slot(s) have been occupied for')
+  })
+
   // The deployed container serves the block inside its heartbeat projection.
   it('reads the health block where the container actually serves it', async () => {
     const out = buffer()

src/cli/diagnose.ts

@@ -263,6 +263,29 @@ function verdictFor(diagnosis: Omit<DeployedFactoryDiagnosis, 'verdict' | 'dispa
       verdict: 'not dispatching: the daemon is not listening for Relayfile events.',
     }
   }
+  // #303. A wedged batch is the one dispatch-gating condition that fails
+  // without failing: nothing throws, no counter moves, and every other line in
+  // this report reads green while no issue can be promoted out of `queued`.
+  const capacity = health.dispatchCapacity
+  if (capacity?.state === 'stalled') {
+    const agentless = capacity.agentlessOccupants ?? 0
+    return {
+      dispatching: false,
+      verdict:
+        // `longestWaitMs` is the oldest *queue* wait, not how long the slots
+        // have been held; saying otherwise sends an operator looking for an
+        // occupant that has been there that long (#303 review, cubic).
+        `not dispatching: ${capacity.waiting} issue(s) have been waiting for batch capacity for up ` +
+        `to ${formatDuration(capacity.longestWaitMs)}, with ${capacity.active}/${capacity.batchSize} ` +
+        'slot(s) occupied' +
+        (agentless > 0
+          ? `. ${agentless} occupied slot(s) never placed an agent and are past the ` +
+            `${formatDuration(capacity.agentlessHoldTimeoutMs)} reap deadline, so they cannot finish ` +
+            'on their own'
+          : '') +
+        '. Pass --token to read /evidence for the issues holding the slots.',
+    }
+  }
   if (health.status === 'unknown') {
     return {
       dispatching: false,
@@ -444,6 +467,22 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st
       lines.push(`    lastCompletedAt    : ${formatInstant(readiness.lastCompletedAtMs)}`)
       lines.push(`    lastFailureAt      : ${formatInstant(readiness.lastFailureAtMs)}`)
     }
+    const capacity = health.dispatchCapacity
+    if (capacity) {
+      lines.push('  dispatchCapacity:')
+      lines.push(`    state              : ${capacity.state}`)
+      lines.push(`    slots              : ${capacity.active}/${capacity.batchSize} occupied`)
+      lines.push(`    waiting            : ${capacity.waiting} issue(s)`)
+      lines.push(
+        `    longest queue wait : ${formatDuration(capacity.longestWaitMs)} (warn past ${formatDuration(capacity.waitWarnMs)})`,
+      )
+      if (capacity.agentlessOccupants !== undefined) {
+        lines.push(
+          `    unreaped wedges    : ${capacity.agentlessOccupants} occupied slot(s) never placed an ` +
+          `agent and are past the ${formatDuration(capacity.agentlessHoldTimeoutMs)} reap deadline`,
+        )
+      }
+    }
     lines.push(`  eventListener        : ${health.eventListener?.state ?? 'unknown'}`)
   } else if (diagnosis.unreadable) {
     lines.push('  health block         : none — this response carried no Factory health')

src/cli/fleet.ts

@@ -1293,6 +1293,14 @@ async function factoryStatusWithMountHealth(
   const fleetControlPlane = liveness.ok
     ? heartbeat?.fleetControlPlane
     : observableStatus.fleetControlPlane
+  // Same rule as readinessReconcile (#303): a live daemon owns the batch, and
+  // a fresh local Factory instance holds no lifecycles. Falling back to that
+  // instance when a live daemon predates the field would publish its empty
+  // view as "the batch is free" — the exact misreport this exists to prevent,
+  // so an older daemon reports nothing here instead (#303 review, cubic).
+  const dispatchCapacity = liveness.ok
+    ? heartbeat?.dispatchCapacity
+    : observableStatus.dispatchCapacity
   const eventListener = liveness.ok
     ? heartbeat?.eventListener ?? {
       state: 'unknown' as const,
@@ -1311,6 +1319,7 @@ async function factoryStatusWithMountHealth(
     eventListener,
     readinessReconcile,
     fleetControlPlane,
+    ...(dispatchCapacity ? { dispatchCapacity } : {}),
   }
   return {
     ...observableStatus,
@@ -1320,6 +1329,7 @@ async function factoryStatusWithMountHealth(
     eventListener,
     readinessReconcile,
     fleetControlPlane,
+    ...(dispatchCapacity ? { dispatchCapacity } : {}),
     localMountDegraded: health.degraded,
     ...(health.reason ? { localMountDegradedReason: health.reason } : {}),
     ...(health.localDir ? { localMountRoot: health.localDir } : {}),

src/config/schema.test.ts

@@ -32,6 +32,10 @@ describe('FactoryConfigSchema', () => {
     expect(parsed.dispatch).toEqual({
       errorCooldownMs: 60_000,
       maxAttempts: 2,
+      // Much shorter than the placed-agent hold: a slot-occupying lifecycle
+      // that never placed an agent has nothing that can move it (#303).
+      agentlessHoldTimeoutMs: 30 * 60_000,
+      capacityWaitWarnMs: 30 * 60_000,
       agentHoldTimeoutMs: 4 * 60 * 60_000,
     })
     expect(parsed.fleetHealth).toEqual({

src/config/schema.ts

@@ -84,6 +84,21 @@ const liveSubscriptionSchema = z.object({
 
 export const DEFAULT_AGENT_HOLD_TIMEOUT_MS = 4 * 60 * 60_000
 
+/**
+ * How long a lifecycle may occupy a batch slot without ever placing an agent.
+ *
+ * Much shorter than `agentHoldTimeoutMs` because the two bound different
+ * things: that one bounds a team that is plausibly working, this one bounds a
+ * row that definitionally is not — nothing but a placement can move it, and no
+ * placement ever happened. It still has to clear the whole promote-to-spawn
+ * window (clone, worktree prep, fleet spawn, roster adoption) with room to
+ * spare, or the reaper races a dispatch that was about to succeed (#303).
+ */
+export const DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS = 30 * 60_000
+
+/** Capacity wait past which a full batch stops reading as ordinary backpressure. */
+export const DEFAULT_CAPACITY_WAIT_WARN_MS = 30 * 60_000
+
 const dispatchSchema = z.object({
   errorCooldownMs: z.number().int().min(0).default(60_000),
   maxAttempts: z.number().int().min(1).max(5).default(2),
@@ -93,6 +108,21 @@ const dispatchSchema = z.object({
   // agent placement until terminal cleanup.
   agentHoldTimeoutMs: z.number().int().min(1).max(7 * 24 * 60 * 60_000)
     .default(DEFAULT_AGENT_HOLD_TIMEOUT_MS),
+  // A slot-occupying lifecycle with no successful placement has no other
+  // deadline: `agentHoldTimeoutMs` is anchored on a placement that never
+  // happened, so before #303 nothing could ever reap it.
+  agentlessHoldTimeoutMs: z.number().int().min(1).max(7 * 24 * 60 * 60_000)
+    .default(DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS),
+  /**
+   * Wall-clock capacity wait past which dispatch is reported degraded (#303).
+   *
+   * A full batch is normal; a batch that has been full for hours is how a
+   * total dispatch outage looked from every operator surface. Deployments that
+   * legitimately run multi-hour issues against a small `batchSize` should
+   * raise this rather than lower `batchSize`'s usefulness.
+   */
+  capacityWaitWarnMs: z.number().int().min(1_000).max(7 * 24 * 60 * 60_000)
+    .default(DEFAULT_CAPACITY_WAIT_WARN_MS),
 }).default({})
 
 const fleetHealthSchema = z.object({

src/orchestrator/batch-tracker.ts

@@ -27,6 +27,13 @@ export interface InFlightIssue {
   dispatchClaim?: FactoryDispatchClaimStatus
   /** Wall-clock anchor set when the first agent placement succeeds. */
   heldSinceAtMs?: number
+  /**
+   * Wall-clock anchor set when this record took a `batchSize` slot (#303).
+   *
+   * The only deadline a lifecycle that never placed an agent has: it holds a
+   * slot, `heldSinceAtMs` is never stamped, and nothing else will move it.
+   */
+  slotHeldSinceAtMs?: number
   /** Latest durable phase, used only for operator-facing held-agent status. */
   lifecyclePhase?: DispatchLifecyclePhase
 }
@@ -315,6 +322,7 @@ export class BatchTracker {
       invocationIds: new Set(record.invocationIds),
       result: record.result ? structuredClone(record.result) : undefined,
       heldSinceAtMs: record.heldSinceAtMs,
+      slotHeldSinceAtMs: record.slotHeldSinceAtMs,
       lifecyclePhase: record.lifecyclePhase,
     }
     this.#inFlight.set(key, restored)