From a25a0c1447858f7148d3f268e29263e2e9447f33 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 20:04:24 +0200 Subject: [PATCH 1/5] feat(fleet): give the fleet connect a status instead of a log nobody reads The dial that makes Factory's own agent `online` had NO status on any surface. `#ensureEventSubscription` starts `#subscribeEvents()` with `void ... .catch()` and reported a rejection by calling `#log` alone, so a client that registered an agent and then failed to connect was indistinguishable from a healthy one everywhere. Readers substituted `eventListener` -- which is the orchestrator's ISSUE subscription, a different subsystem that reports `subscribed` whenever `#subscription` is truthy. That conflation is why days of instruments read healthy while the fleet socket was never up, and why the only trace of the failure was a log line no health surface reads. WHAT THIS ADDS `FleetConnectStatus` on the fleet port: `never-attempted | connecting | connected | failed`, with attempts, timestamps, and a reduced `lastError`. The rejection in the void-catch now lands in that field instead of only in `#log`. `connected` is stamped AT THE DIAL, not when `#subscribeEvents` resolves: that function also returns early when the client is disposed, and reporting THAT as connected would recreate the exact false-healthy signal this removes. The method is OPTIONAL on the port, so a backend with no socket (the internal fleet) omits it and an absent value stays absent rather than being invented as healthy. The orchestrator writes it into the loop heartbeat next to `fleetControlPlane`, and `guardFleetControlPlane`'s Proxy passes it through untouched. REDACTION reuses `describeControlPlaneError`, now exported so there is ONE reducer rather than two that can drift: any cause becomes `Name (CODE)` with the code matched against /^[A-Z0-9_]{1,80}$/, so no transport message, URL or credential can reach the value. DELIBERATELY NOT DISPATCH-GATING. `fleetConnect` is published on /healthz but is not added to DISPATCH_GATING_SUBSYSTEMS: a failed socket does not itself stop dispatch (`roster()` runs over HTTP), and listing it would flip `ok` on a live deployment and hand container replacement a new reason to cycle. Publishing the fact is the goal; changing what `ok` means is a separate decision belonging to whoever owns dispatch behaviour. `lastError` stays behind the authenticated /evidence, exactly as it does for the circuit. VERIFICATION - build typecheck (tsconfig.build.json) exit 0, zero errors - 986 tests green across the affected surface: src/fleet + public-health + cli/fleet (363), health-projection-guard/atomic-json-file/reaper/diagnose (49), orchestrator/factory (574) - mutation control: reverting the catch to log-only fails exactly the three tests that assert recording (exit 1), while never-attempted and connected keep passing because they do not depend on it - a control test proves the same harness yields `connected` when the gate does not throw, so the failure arm is not passing against a client that reports failed unconditionally Refs AgentWorkforce/factory-cloud#55. Co-Authored-By: Claude Opus 5 --- src/fleet/control-plane-circuit.ts | 10 ++- src/fleet/relay-fleet-client.test.ts | 86 ++++++++++++++++++++++++++ src/fleet/relay-fleet-client.ts | 43 ++++++++++++- src/orchestrator/factory.ts | 6 ++ src/orchestrator/public-health.test.ts | 53 ++++++++++++++++ src/orchestrator/public-health.ts | 32 ++++++++++ src/ports/fleet.ts | 31 ++++++++++ src/types.ts | 26 ++++++++ 8 files changed, 285 insertions(+), 2 deletions(-) diff --git a/src/fleet/control-plane-circuit.ts b/src/fleet/control-plane-circuit.ts index eb8a633c..2ecbe1f5 100644 --- a/src/fleet/control-plane-circuit.ts +++ b/src/fleet/control-plane-circuit.ts @@ -224,7 +224,15 @@ export function isFleetControlPlaneFailure(error: unknown): boolean { } /** Redacts arbitrary transport text before circuit state becomes observable. */ -function describeControlPlaneError(error: unknown): string { +/** + * Reduce any cause to `Name (CODE)`. + * + * Exported so the fleet CONNECT status uses the identical redaction as the + * control-plane circuit: both values are published through `factory status` and + * the authenticated `/evidence`, and neither may carry a transport message, + * URL or credential. + */ +export 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})` : '' diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index 4de8e8d8..d958d3c3 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -1910,3 +1910,89 @@ describe('RelayFleetClient placement deadlines (#306)', () => { expect(reads).toBe(0) }) }) + +/** + * The fleet socket had no status anywhere. `#ensureEventSubscription` starts the + * subscription with `void ... .catch()` and reported a rejection by calling + * `#log` only, so a client that registered an agent and then failed to connect + * was indistinguishable from a healthy one on every surface. These pin the + * outcome into a field a health surface can publish. + */ +describe('RelayFleetClient fleet connect status', () => { + it('starts as never-attempted, so an unstarted socket is not reported as healthy', () => { + const messaging = new FakeMessaging() + const fleet = createClient(messaging) + const status = fleet.fleetConnectStatus() + expect(status.state).toBe('never-attempted') + expect(status.attempts).toBe(0) + expect(status.lastError).toBeUndefined() + }) + + it('reports connected only after the dial actually happens', async () => { + const messaging = new FakeMessaging() + const fleet = createClient(messaging) + fleet.onAgentExit(() => {}) + await flush() + const status = fleet.fleetConnectStatus() + expect(status.state).toBe('connected') + expect(status.attempts).toBe(1) + expect(messaging.connected).toBe(1) + expect(status.lastConnectedAtMs).toBeTypeOf('number') + }) + + /** + * THE REGRESSION THIS EXISTS FOR. Before, this produced a `#log` line and + * nothing else: every status surface still read healthy. + */ + it('records a failed subscription instead of only logging it', async () => { + const messaging = new FakeMessaging() + messaging.commands.available = () => false + const fleet = createClient(messaging) + fleet.onAgentExit(() => {}) + await flush() + const status = fleet.fleetConnectStatus() + expect(status.state).toBe('failed') + expect(status.attempts).toBe(1) + expect(status.lastFailureAtMs).toBeTypeOf('number') + expect(status.lastError).toBeDefined() + expect(messaging.connected).toBe(0) + }) + + /** + * CONTROL. The failure arm above only means something if the SAME harness can + * produce the other outcome -- otherwise it would pass against a client that + * reports 'failed' unconditionally. + */ + it('the same harness yields connected when the gate does not throw', async () => { + const failing = new FakeMessaging() + failing.commands.available = () => false + const failingFleet = createClient(failing) + failingFleet.onAgentExit(() => {}) + await flush() + + const healthy = new FakeMessaging() + const healthyFleet = createClient(healthy) + healthyFleet.onAgentExit(() => {}) + await flush() + + expect(failingFleet.fleetConnectStatus().state).toBe('failed') + expect(healthyFleet.fleetConnectStatus().state).toBe('connected') + }) + + it('reduces the cause to a name and code, never a transport message', async () => { + const messaging = new FakeMessaging() + messaging.commands.available = () => { + const error = new Error('connect failed to wss://relay.example/socket?token=at_live_abcdef0123456789') + error.name = 'FleetSocketError' + ;(error as Error & { code?: string }).code = 'ECONNREFUSED' + throw error + } + const fleet = createClient(messaging) + fleet.onAgentExit(() => {}) + await flush() + const lastError = fleet.fleetConnectStatus().lastError ?? '' + expect(lastError).toBe('FleetSocketError (ECONNREFUSED)') + expect(lastError).not.toContain('at_live_') + expect(lastError).not.toContain('wss://') + }) +}) diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index a49b7677..8034f7de 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -1,8 +1,9 @@ import { AgentRelay } from '@agent-relay/sdk' +import { describeControlPlaneError } from './control-plane-circuit' import { resolveRelayAgentToken, resolveRelayWorkspaceKey } from './relay-workspace-key' -import type { AgentLifecycleSignal, AgentMessage, AgentUsage, Capability, FleetClient, NodeCapability, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult } from '../ports/fleet' +import type { AgentLifecycleSignal, AgentMessage, AgentUsage, Capability, FleetClient, FleetConnectStatus, NodeCapability, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult } from '../ports/fleet' import type { RelayActionInvocation, RelayActionInvocationAck, @@ -217,6 +218,15 @@ export class RelayFleetClient implements FleetClient { #authenticatedAgentName: string #eventsStarted = false #disposed = false + /** + * The fleet socket's own status. + * + * Kept here rather than inferred by callers because the dial is + * fire-and-forget: `#ensureEventSubscription` cannot return its outcome, so + * without this the ONLY trace of a failed connect was a `#log` line that no + * health surface reads. + */ + #fleetConnect: FleetConnectStatus = { state: 'never-attempted', attempts: 0 } #watchTimer: ReturnType | undefined #reconciling: Promise | undefined #pendingReleaseRetry: Promise | undefined @@ -1197,17 +1207,48 @@ export class RelayFleetClient implements FleetClient { #ensureEventSubscription(): void { if (this.#eventsStarted) return this.#eventsStarted = true + this.#fleetConnect = { + ...this.#fleetConnect, + state: 'connecting', + attempts: this.#fleetConnect.attempts + 1, + lastAttemptAtMs: this.#now(), + } void this.#subscribeEvents().catch((error) => { this.#eventsStarted = false + // The rejection now lands in a field a health surface can publish. It kept + // going only to `#log` before, which is why a fleet client that never + // connected was indistinguishable from a healthy one everywhere. + this.#fleetConnect = { + ...this.#fleetConnect, + state: 'failed', + lastFailureAtMs: this.#now(), + lastError: describeControlPlaneError(error), + } this.#log(`relay fleet event subscription failed: ${errorMessage(error)}`) }) } + /** + * `connected` is stamped at the dial itself, not when `#subscribeEvents` + * resolves: that function also returns early when the client is disposed, and + * reporting THAT as connected would recreate the exact false-healthy signal + * this status exists to remove. + */ + fleetConnectStatus(): FleetConnectStatus { + return { ...this.#fleetConnect } + } + async #subscribeEvents(): Promise { const messaging = await this.#ensureMessaging() await this.#ensureLifecycleAction(messaging) if (this.#disposed) return messaging.events.connect() + this.#fleetConnect = { + ...this.#fleetConnect, + state: 'connected', + lastConnectedAtMs: this.#now(), + lastError: undefined, + } this.#eventUnsubscribers.push(messaging.events.on('any', (event) => this.#handleEvent(event))) } diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index efa118fe..cdd0c469 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -4940,6 +4940,9 @@ export class FactoryLoop implements Factory { })) ?? [], counters: { ...this.#counters }, fleetControlPlane: this.#fleetControlPlane.status(), + // Optional on the port: a backend with no socket omits it, and an absent + // value stays absent rather than being invented as healthy. + ...(this.#fleet.fleetConnectStatus ? { fleetConnect: this.#fleet.fleetConnectStatus() } : {}), slackDegraded: this.#slackDegraded, slackDegradedReason: this.#slackDegradedReason, eventListener: this.#eventListenerStatus(), @@ -8102,6 +8105,9 @@ export class FactoryLoop implements Factory { readinessReconcile: this.#readinessReconcileStatus(), dispatchCapacity: this.#dispatchCapacityStatus(), fleetControlPlane: this.#fleetControlPlane.status(), + // Optional on the port: a backend with no socket omits it, and an absent + // value stays absent rather than being invented as healthy. + ...(this.#fleet.fleetConnectStatus ? { fleetConnect: this.#fleet.fleetConnectStatus() } : {}), } // The deployed container serves `/healthz` straight out of this file and // has no redaction logic of its own, so publish the already-safe view here diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index 40d9f43f..0bb3b648 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -743,3 +743,56 @@ describe('publicHealthFromHeartbeat (#295)', () => { }) }) + +/** + * The fleet event socket is the dial that makes Factory's own agent `online`. + * It had no status on any surface, and readers substituted `eventListener` -- + * which is the orchestrator's ISSUE subscription, a different subsystem. That + * conflation is why a Factory that registered an agent and never connected read + * as healthy everywhere. + */ +describe('fleet connect health', () => { + const withConnect = ( + overrides: Partial> = {}, + ): FactoryLoopHeartbeat => + heartbeat({ + fleetConnect: { + state: 'failed', + attempts: 1, + lastAttemptAtMs: BOOT_MS - 5_000, + lastFailureAtMs: BOOT_MS - 4_000, + lastError: 'FactoryAgentRegistrationError', + ...overrides, + }, + }) + + it('publishes the socket state unauthenticated', () => { + const health = publicHealthFromHeartbeat(withConnect(), { nowMs: BOOT_MS }) + expect(health.fleetConnect?.state).toBe('failed') + expect(health.fleetConnect?.attempts).toBe(1) + }) + + /** `lastError` stays behind /evidence, exactly as it does for the circuit. */ + it('never leaks the cause to the unauthenticated surface', () => { + const health = publicHealthFromHeartbeat(withConnect(), { nowMs: BOOT_MS }) + expect(JSON.stringify(health.fleetConnect)).not.toContain('FactoryAgentRegistrationError') + expect(Object.hasOwn(health.fleetConnect ?? {}, 'lastError')).toBe(false) + }) + + /** + * Deliberately NOT dispatch-gating. Listing it would flip `ok` on a live + * deployment and hand container replacement a new reason to cycle -- a + * behaviour change well beyond publishing the fact. + */ + it('does not change what ok means', () => { + const health = publicHealthFromHeartbeat(withConnect(), { nowMs: BOOT_MS }) + expect(health.degradedSubsystems).not.toContain('fleetConnect') + expect(health.ok).toBe(true) + }) + + /** CONTROL: absent stays absent rather than being invented as healthy. */ + it('omits the block entirely when the backend has no socket', () => { + const health = publicHealthFromHeartbeat(heartbeat(), { nowMs: BOOT_MS }) + expect(health.fleetConnect).toBeUndefined() + }) +}) diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index ca13b01d..0c44effc 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -1,6 +1,7 @@ import { createHash, randomBytes } from 'node:crypto' import { telemetryErrorClassName } from '../observability/error-class.js' import type { FleetControlPlaneStatus } from '../fleet/control-plane-circuit' +import type { FleetConnectStatus } from '../ports/fleet' import { DEFAULT_AGENTLESS_HOLD_TIMEOUT_MS, DEFAULT_CAPACITY_WAIT_WARN_MS } from '../config/schema' import type { FactoryDispatchCapacityStatus, @@ -9,6 +10,7 @@ import type { FactoryPublicDispatchCapacityHealth, FactoryPublicDispatchSlotOccupant, FactoryPublicEventListenerHealth, + FactoryPublicFleetConnectHealth, FactoryPublicFleetControlPlaneHealth, FactoryPublicHealth, FactoryPublicReadinessReconcileHealth, @@ -69,6 +71,13 @@ const EVENT_LISTENER_STATES: readonly FactoryEventListenerStatus['state'][] = [ 'polling', ] +const FLEET_CONNECT_STATES: readonly FleetConnectStatus['state'][] = [ + 'never-attempted', + 'connecting', + 'connected', + 'failed', +] + const FLEET_CONTROL_PLANE_STATES: readonly FleetControlPlaneStatus['state'][] = [ 'closed', 'open', @@ -438,6 +447,27 @@ function dispatchCapacityHealth( } } +/** + * Project the fleet socket for the UNAUTHENTICATED surface. + * + * Deliberately NOT added to DISPATCH_GATING_SUBSYSTEMS. A failed socket does not + * itself stop dispatch -- `roster()` runs over HTTP -- and listing it there would + * flip `ok` on a live deployment and hand the container-replacement logic a new + * reason to cycle. Publishing the fact is the goal; changing what `ok` means is a + * separate decision belonging to whoever owns dispatch behaviour. + */ +function fleetConnectHealth(status: FleetConnectStatus): FactoryPublicFleetConnectHealth { + const attempts = counter(status.attempts) + return { + state: enumValue(status.state, FLEET_CONNECT_STATES), + ...(attempts !== undefined ? { attempts } : {}), + ...optionalTimestamp('lastAttemptAtMs', status.lastAttemptAtMs), + ...optionalTimestamp('lastConnectedAtMs', status.lastConnectedAtMs), + ...optionalTimestamp('lastFailureAtMs', status.lastFailureAtMs), + // `lastError` stays behind /evidence, exactly as it does for the circuit. + } +} + function fleetControlPlaneHealth( status: FleetControlPlaneStatus, ): FactoryPublicFleetControlPlaneHealth { @@ -485,6 +515,7 @@ export function publicHealthFromHeartbeat( const readinessReconcile = heartbeat.readinessReconcile ? readinessReconcileHealth(heartbeat.readinessReconcile, nowMs) : undefined + const fleetConnect = heartbeat.fleetConnect ? fleetConnectHealth(heartbeat.fleetConnect) : undefined const fleetControlPlane = heartbeat.fleetControlPlane ? fleetControlPlaneHealth(heartbeat.fleetControlPlane) : undefined @@ -565,6 +596,7 @@ export function publicHealthFromHeartbeat( ...(readinessReconcile ? { readinessReconcile } : {}), ...(eventListener ? { eventListener } : {}), ...(fleetControlPlane ? { fleetControlPlane } : {}), + ...(fleetConnect ? { fleetConnect } : {}), ...(dispatchCapacity ? { dispatchCapacity } : {}), } } diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index a032024b..3bfc0acb 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -132,6 +132,31 @@ export type AgentUsage = { } export type FleetTrackedAgent = { invocationId?: string; node?: string } +/** Lifecycle of the fleet event socket: the dial that makes this agent `online`. */ +export type FleetConnectState = 'never-attempted' | 'connecting' | 'connected' | 'failed' + +/** + * Why the fleet event subscription is (or is not) live. + * + * This exists because the dial had NO status anywhere. `#ensureEventSubscription` + * starts `#subscribeEvents()` with `void ... .catch()` and reports a rejection by + * calling `#log` only, so a fleet client that registered an agent and then failed + * to connect looked identical to a healthy one on every surface. Callers read + * `eventListener` and saw `subscribed` -- but that is the orchestrator's ISSUE + * subscription, a different subsystem entirely, which is how a broken fleet + * socket stayed invisible while every instrument reported healthy. + */ +export interface FleetConnectStatus { + state: FleetConnectState + /** How many times a subscription has been started, including the current one. */ + attempts: number + lastAttemptAtMs?: number + lastConnectedAtMs?: number + lastFailureAtMs?: number + /** Reduced to `Name (CODE)`; never a raw transport message. */ + lastError?: string +} + export interface FleetClient { /** Backend-wide placement locality, used when recovering a spawn ack crash gap. */ readonly placementLocality?: 'local' | 'remote' @@ -157,6 +182,12 @@ export interface FleetClient { }): Promise release(name: string, reason?: string): Promise roster(): Promise + /** + * Whether this backend's event socket is connected, and why not when it is not. + * Optional: backends with no socket (the internal fleet) simply omit it, and an + * absent value is reported as such rather than being invented as healthy. + */ + fleetConnectStatus?(): FleetConnectStatus resolveAgentPid?(name: string): Promise protectedPids?(): Promise sendMessage(input: SendInput): Promise diff --git a/src/types.ts b/src/types.ts index cdb05eb1..06dbc14c 100644 --- a/src/types.ts +++ b/src/types.ts @@ -13,6 +13,7 @@ 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' +import type { FleetConnectStatus } from './ports/fleet' export interface FactoryPorts { mount: MountClient @@ -184,6 +185,15 @@ export interface FactoryLoopHeartbeat { dispatchCapacity?: FactoryDispatchCapacityStatus /** Daemon-owned dispatch admission state; status readers must prefer this over a fresh local Factory instance. */ fleetControlPlane?: FleetControlPlaneStatus + /** + * Whether the fleet EVENT SOCKET is connected -- the dial that makes this + * Factory's agent `online`. Absent when the backend has no socket. + * + * Distinct from `eventListener`, which is the orchestrator's ISSUE + * subscription. Conflating the two is how a fleet client that registered an + * agent and never connected read as healthy on every surface. + */ + fleetConnect?: FleetConnectStatus /** * Redacted projection of this record, safe to serve unauthenticated (#295). * @@ -370,6 +380,20 @@ export interface FactoryPublicEventListenerHealth { * probe failure names sockets and paths — so only the state, the counters and * the retry instant cross. */ +/** + * Unauthenticated view of the fleet socket. State and counters only. + * + * `lastError` is deliberately absent for the same reason it is absent from the + * control-plane block: it stays behind the authenticated `/evidence`. + */ +export interface FactoryPublicFleetConnectHealth { + state: FleetConnectStatus['state'] | 'unknown' + attempts?: number + lastAttemptAtMs?: number + lastConnectedAtMs?: number + lastFailureAtMs?: number +} + export interface FactoryPublicFleetControlPlaneHealth { state: FleetControlPlaneStatus['state'] | 'unknown' consecutiveFailures: number @@ -410,6 +434,8 @@ export interface FactoryPublicHealth { readinessReconcile?: FactoryPublicReadinessReconcileHealth eventListener?: FactoryPublicEventListenerHealth fleetControlPlane?: FactoryPublicFleetControlPlaneHealth + /** Fleet event socket. NOT dispatch-gating: see DISPATCH_GATING_SUBSYSTEMS. */ + fleetConnect?: FactoryPublicFleetConnectHealth dispatchCapacity?: FactoryPublicDispatchCapacityHealth } From 9017d67fb2794ebd9c6d6a8964773676b63bf4d3 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 20:30:19 +0200 Subject: [PATCH 2/5] feat(fleet): name WHICH registration failure happened, through the same redaction Production reported `fleetControlPlane.lastError: "FactoryAgentRegistrationError"` and that string was the entire answer. TEN distinct throw sites reduce to it, and the sentence naming which one fired is discarded before it reaches any surface -- so an operator could not tell whether the name was taken and unreadable, the record was not offline, presence was unreadable, presence did not list the agent, or the takeover itself failed. Those are five different bugs with five different fixes. The reducer was already willing to answer this. `describeControlPlaneError` renders `${name}${code}` and appends the code when it matches /^[A-Z0-9_]{1,80}$/ -- `FactoryAgentRegistrationError` simply never carried one. So this adds a `code` to the class and a distinct value at each of the ten throw sites, and the answer appears on every existing surface with NO new exposure: a constrained uppercase token carries no transport text, no URL and no credential, and the reducer already validates its shape. The detail sentence still does not survive, deliberately -- two of the ten embed `errorMessage(error)` from a transport failure, so the message is exactly what must stay withheld. The code replaces it as the discriminator rather than widening what is published. Tests pin the rendering through the SAME reducer every published surface uses, a control asserts distinct sites reduce to DISTINCT strings (a suffix that does not discriminate would reproduce the bug one level up), and a third asserts a credential-bearing detail still cannot escape. 366 tests green across src/fleet + public-health + cli/fleet. Build typecheck exit 0. FLAKE NOTE, measured rather than assumed: `orchestrator/factory.test.ts` > "releases a late placement when another owner reclaimed the lifecycle (#303)" fails intermittently. Sampled 8 runs on each arm: clean origin/main failed 3/8, this branch failed 1/8. It is PRE-EXISTING and unrelated -- the assertion is about lifecycle ownership, not registration -- but it is the kind of red that gets blamed on whichever PR is in front of it. Refs AgentWorkforce/factory-cloud#55. Co-Authored-By: Claude Opus 5 --- src/fleet/relay-fleet-client.test.ts | 49 ++++++++++++++++++++++++++ src/fleet/relay-fleet-client.ts | 51 ++++++++++++++++++++++++++-- 2 files changed, 97 insertions(+), 3 deletions(-) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index d958d3c3..2cfbe88e 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest' +import { describeControlPlaneError } from './control-plane-circuit' import { FactoryAgentRegistrationError, MAX_REGISTRATION_ATTEMPTS, ReadOnlyFleetIdentityError, RelayFleetClient, type RelayClientFactoryOptions, type RelayClientLike } from './relay-fleet-client' import { runFleetCli } from '../cli/fleet' @@ -1996,3 +1997,51 @@ describe('RelayFleetClient fleet connect status', () => { expect(lastError).not.toContain('wss://') }) }) + +/** + * Production reported `fleetControlPlane.lastError: "FactoryAgentRegistrationError"` + * and that string was the whole answer: ten different throw sites reduce to it, + * and the sentence naming which one fired is discarded by redaction. These pin + * the code through the SAME reducer every published surface uses. + */ +describe('registration failures survive redaction with their cause', () => { + it('renders the discriminating code after the error name', () => { + const error = new FactoryAgentRegistrationError( + 'factory-cloud-7d6e3ca1', + 'PRESENCE_STATUS_MISSING', + 'presence reported no status for this agent, so it cannot be confirmed offline', + ) + expect(describeControlPlaneError(error)).toBe('FactoryAgentRegistrationError (PRESENCE_STATUS_MISSING)') + }) + + /** + * CONTROL. Distinct sites must reduce to DISTINCT strings, or the code adds a + * suffix without adding an answer — which is the bug being fixed, one level up. + */ + it('distinguishes the throw sites from one another', () => { + const rendered = ( + [ + 'STATUS_NOT_OFFLINE', + 'PRESENCE_STATUS_MISSING', + 'PRESENCE_UNREADABLE', + 'TAKEOVER_FAILED', + 'MAX_ATTEMPTS', + ] as const + ).map((code) => describeControlPlaneError(new FactoryAgentRegistrationError('a', code, 'detail'))) + expect(new Set(rendered).size).toBe(rendered.length) + }) + + /** The detail sentence may embed a transport message, so it must NOT survive. */ + it('still withholds the message the code replaces', () => { + const rendered = describeControlPlaneError( + new FactoryAgentRegistrationError( + 'a', + 'TAKEOVER_FAILED', + 'could not reclaim the existing identity: POST https://relay.example/v1/agents?token=at_live_abcdef0123456789 failed', + ), + ) + expect(rendered).toBe('FactoryAgentRegistrationError (TAKEOVER_FAILED)') + expect(rendered).not.toContain('at_live_') + expect(rendered).not.toContain('https://') + }) +}) diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index 8034f7de..97f78b98 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -735,6 +735,7 @@ export class RelayFleetClient implements FleetClient { if (this.#registrationAttempts > MAX_REGISTRATION_ATTEMPTS) { throw new FactoryAgentRegistrationError( this.#agentName, + 'MAX_ATTEMPTS', `gave up after ${MAX_REGISTRATION_ATTEMPTS} registration attempts`, ) } @@ -886,6 +887,7 @@ export class RelayFleetClient implements FleetClient { if (!agentId) { throw new FactoryAgentRegistrationError( this.#agentName, + 'RECORD_UNREADABLE', `name is taken but the record could not be read back: ${errorMessage(lastError)}`, { cause: lastError }, ) @@ -899,6 +901,7 @@ export class RelayFleetClient implements FleetClient { if (error instanceof FactoryAgentRegistrationError) throw error throw new FactoryAgentRegistrationError( this.#agentName, + 'TAKEOVER_FAILED', `could not reclaim the existing identity: ${errorMessage(error)}`, { cause: error }, ) @@ -906,6 +909,7 @@ export class RelayFleetClient implements FleetClient { } throw new FactoryAgentRegistrationError( this.#agentName, + 'TAKEOVER_EXHAUSTED', `could not reclaim the existing identity: ${errorMessage(lastError)}`, { cause: lastError }, ) @@ -926,6 +930,7 @@ export class RelayFleetClient implements FleetClient { if (status !== 'offline') { throw new FactoryAgentRegistrationError( this.#agentName, + 'STATUS_NOT_OFFLINE', `refusing to take over agent in status "${status ?? 'unknown'}"; another factory may still hold this identity`, ) } @@ -964,12 +969,14 @@ export class RelayFleetClient implements FleetClient { if (seenStatus === undefined) { throw new FactoryAgentRegistrationError( this.#agentName, - 'presence lists this agent with no status, so it cannot be confirmed offline', + 'PRESENCE_STATUS_MISSING', + 'presence reported no status for this agent, so it cannot be confirmed offline', ) } if (LIVE_AGENT_STATUSES.has(seenStatus)) { throw new FactoryAgentRegistrationError( this.#agentName, + 'PRESENCE_REPORTS_LIVE', `presence reports this agent as "${seenStatus}"; another factory may still hold this identity`, ) } @@ -991,6 +998,7 @@ export class RelayFleetClient implements FleetClient { } catch (error) { throw new FactoryAgentRegistrationError( this.#agentName, + 'PRESENCE_UNREADABLE', `presence is unreadable, so the agent cannot be confirmed offline: ${errorMessage(error)}`, { cause: error }, ) @@ -998,6 +1006,7 @@ export class RelayFleetClient implements FleetClient { if (!Array.isArray(presence)) { throw new FactoryAgentRegistrationError( this.#agentName, + 'PRESENCE_NOT_A_LIST', 'presence is unreadable (it did not return a list), so the agent cannot be confirmed offline', ) } @@ -1014,6 +1023,7 @@ export class RelayFleetClient implements FleetClient { if (unnamed !== -1) { throw new FactoryAgentRegistrationError( this.#agentName, + 'PRESENCE_ROW_UNNAMED', `presence is unreadable (row ${unnamed} carries no agent name), so the agent cannot be confirmed offline`, ) } @@ -1599,13 +1609,48 @@ function previewReference(value: unknown, placementNode?: string): PreviewRefere * Registration could not converge on a usable agent identity. Named so a * six-day silent 409 loop surfaces as one actionable failure instead. */ +/** + * Which of the ten registration failures happened. + * + * These exist to survive redaction. `describeControlPlaneError` reduces a cause + * to `${name}${code}` and appends the code ONLY when it matches + * /^[A-Z0-9_]{1,80}$/ -- so a bare `FactoryAgentRegistrationError` is what every + * published surface showed, with the sentence naming the actual throw site + * discarded. Production reported exactly that string and it could not be told + * whether the name was taken, presence was unreadable, the record was not + * offline, or the takeover itself failed. + * + * A constrained token carries none of the message's risk: no transport text, no + * URL, no credential. The reducer already validates the shape, so this widens + * the answer without widening the exposure. + */ +export type FactoryAgentRegistrationErrorCode = + | 'MAX_ATTEMPTS' + | 'RECORD_UNREADABLE' + | 'STATUS_NOT_OFFLINE' + | 'PRESENCE_UNREADABLE' + | 'PRESENCE_NOT_A_LIST' + | 'PRESENCE_ROW_UNNAMED' + | 'PRESENCE_STATUS_MISSING' + | 'PRESENCE_REPORTS_LIVE' + | 'TAKEOVER_FAILED' + | 'TAKEOVER_EXHAUSTED' + export class FactoryAgentRegistrationError extends Error { readonly agentName: string - - constructor(agentName: string, detail: string, options?: { cause?: unknown }) { + /** Uppercase token so `describeControlPlaneError` renders it after the name. */ + readonly code: FactoryAgentRegistrationErrorCode + + constructor( + agentName: string, + code: FactoryAgentRegistrationErrorCode, + detail: string, + options?: { cause?: unknown }, + ) { super(`Factory agent "${agentName}" could not register with the relay workspace: ${detail}`) this.name = 'FactoryAgentRegistrationError' this.agentName = agentName + this.code = code if (options && 'cause' in options) { ;(this as Error & { cause?: unknown }).cause = options.cause } From 317f7c74daa53079b36e4890ab23f7e46d7ce43b Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 24 Aug 2026 00:51:36 +0200 Subject: [PATCH 3/5] fix(fleet): observe socket lifecycle in health --- src/cli/diagnose.test.ts | 13 +++++- src/cli/diagnose.ts | 11 +++++ src/cli/fleet.test.ts | 25 +++++++++- src/cli/fleet.ts | 9 +++- src/fleet/relay-fleet-client.test.ts | 61 ++++++++++++++++++++++--- src/fleet/relay-fleet-client.ts | 63 +++++++++++++++++++++----- src/orchestrator/public-health.test.ts | 24 +++++++++- src/orchestrator/public-health.ts | 18 ++++++++ src/ports/fleet.ts | 6 ++- src/types.ts | 4 ++ 10 files changed, 212 insertions(+), 22 deletions(-) diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts index ac3f017c..3a05e6bb 100644 --- a/src/cli/diagnose.test.ts +++ b/src/cli/diagnose.test.ts @@ -61,6 +61,14 @@ const healthy = { lastStartedAtMs: NOW_MS - 30_000, lastCompletedAtMs: NOW_MS - 29_000, }, + fleetConnect: { + state: 'connected', + attempts: 1, + lastAttemptAtMs: NOW_MS - 32_000, + lastDialedAtMs: NOW_MS - 31_000, + firstEventAtMs: NOW_MS - 30_500, + lastConnectedAtMs: NOW_MS - 30_000, + }, eventListener: { state: 'subscribed' }, }, } @@ -82,6 +90,10 @@ describe('factory diagnose --deployed (#295)', () => { expect(seen).toEqual([`${BASE}/healthz`]) expect(out.text()).toContain('dispatching') expect(out.text()).toContain('readinessReconcile') + expect(out.text()).toContain('fleetConnect') + expect(out.text()).toContain('connected') + expect(out.text()).toContain(`lastAttemptAt : ${new Date(NOW_MS - 32_000).toISOString()}`) + expect(out.text()).toContain(`lastConnectedAt : ${new Date(NOW_MS - 30_000).toISOString()}`) }) // The 2026-08-19/20 outage: eight consecutive failures behind `ok: true`. @@ -765,4 +777,3 @@ describe('factory diagnose --deployed (#295)', () => { } }) }) - diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts index a9e9d4d6..6d68e187 100644 --- a/src/cli/diagnose.ts +++ b/src/cli/diagnose.ts @@ -483,6 +483,17 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st ) } } + const fleetConnect = health.fleetConnect + if (fleetConnect) { + lines.push(' fleetConnect:') + lines.push(` state : ${fleetConnect.state}`) + lines.push(` attempts : ${fleetConnect.attempts ?? '—'}`) + lines.push(` lastAttemptAt : ${formatInstant(fleetConnect.lastAttemptAtMs)}`) + lines.push(` lastDialedAt : ${formatInstant(fleetConnect.lastDialedAtMs)}`) + lines.push(` firstEventAt : ${formatInstant(fleetConnect.firstEventAtMs)}`) + lines.push(` lastConnectedAt : ${formatInstant(fleetConnect.lastConnectedAtMs)}`) + lines.push(` lastFailureAt : ${formatInstant(fleetConnect.lastFailureAtMs)}`) + } lines.push(` eventListener : ${health.eventListener?.state ?? 'unknown'}`) } else if (diagnosis.unreadable) { lines.push(' health block : none — this response carried no Factory health') diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 2b7fd1ab..fd1865c3 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -3238,6 +3238,13 @@ describe('fleet CLI runtime', () => { retryAtMs: now + 59_500, lastError: 'TimeoutError (FACTORY_FLEET_CONTROL_TIMEOUT)', }, + fleetConnect: { + state: 'failed', + attempts: 2, + lastAttemptAtMs: now - 2_000, + lastFailureAtMs: now - 400, + lastError: 'RelayEventStreamDisconnected', + }, })) const output = buffer() const factory = { @@ -3257,6 +3264,11 @@ describe('fleet CLI runtime', () => { failureThreshold: 2, resetTimeoutMs: 60_000, }, + fleetConnect: { + state: 'connected' as const, + attempts: 1, + firstEventAtMs: now - 5_000, + }, })), } as unknown as Factory @@ -3281,6 +3293,11 @@ describe('fleet CLI runtime', () => { consecutiveFailures: 2, lastError: 'TimeoutError (FACTORY_FLEET_CONTROL_TIMEOUT)', }, + fleetConnect: { + state: 'failed', + attempts: 2, + lastError: 'RelayEventStreamDisconnected', + }, }) } finally { await rm(root, { recursive: true, force: true }) @@ -3315,6 +3332,10 @@ describe('fleet CLI runtime', () => { failureThreshold: 2, resetTimeoutMs: 60_000, }, + fleetConnect: { + state: 'connected' as const, + attempts: 1, + }, })), } as unknown as Factory @@ -3327,7 +3348,9 @@ describe('fleet CLI runtime', () => { }) expect(code).toBe(0) - expect(JSON.parse(output.text())).not.toHaveProperty('fleetControlPlane') + const status = JSON.parse(output.text()) + expect(status).not.toHaveProperty('fleetControlPlane') + expect(status).not.toHaveProperty('fleetConnect') } finally { await rm(root, { recursive: true, force: true }) } diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index d242b0f0..9b4290dc 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -1264,9 +1264,11 @@ async function factoryStatusWithMountHealth( heartbeatStaleMs: number, versionInfo?: FactoryVersionInfo, stateStoreStatus?: { backend: string }, -): Promise, 'fleetControlPlane'> & { +): Promise, 'fleetControlPlane' | 'fleetConnect'> & { /** Undefined when a live older daemon predates fleet control-plane reporting. */ fleetControlPlane?: ReturnType['fleetControlPlane'] + /** Undefined when a live older daemon predates fleet socket reporting. */ + fleetConnect?: ReturnType['fleetConnect'] stateStore?: { backend: string } version?: string installedAt?: string @@ -1307,6 +1309,9 @@ async function factoryStatusWithMountHealth( const fleetControlPlane = liveness.ok ? heartbeat?.fleetControlPlane : observableStatus.fleetControlPlane + const fleetConnect = liveness.ok + ? heartbeat?.fleetConnect + : observableStatus.fleetConnect // 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 @@ -1333,6 +1338,7 @@ async function factoryStatusWithMountHealth( eventListener, readinessReconcile, fleetControlPlane, + fleetConnect, ...(dispatchCapacity ? { dispatchCapacity } : {}), } return { @@ -1343,6 +1349,7 @@ async function factoryStatusWithMountHealth( eventListener, readinessReconcile, fleetControlPlane, + fleetConnect, ...(dispatchCapacity ? { dispatchCapacity } : {}), localMountDegraded: health.degraded, ...(health.reason ? { localMountDegradedReason: health.reason } : {}), diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index 2cfbe88e..9de22526 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -38,6 +38,7 @@ class FakeMessaging { agentPresenceRows: Array<{ agentId: string; agentName: string; status: 'online' | 'offline' }> | undefined nodeRows: Array & { name: string }> = [] directError: Error | undefined + connectEvent: unknown | undefined connected = 0 disconnected = 0 nextInvocationId = 0 @@ -125,6 +126,7 @@ class FakeMessaging { readonly events = { connect: () => { this.connected += 1 + if (this.connectEvent !== undefined) this.emit('any', this.connectEvent) }, disconnect: async () => { this.disconnected += 1 @@ -1929,16 +1931,62 @@ describe('RelayFleetClient fleet connect status', () => { expect(status.lastError).toBeUndefined() }) - it('reports connected only after the dial actually happens', async () => { + it('reports only dialed until the socket produces a confirming event', async () => { const messaging = new FakeMessaging() - const fleet = createClient(messaging) + let now = 1_000 + const fleet = createClient(messaging, { now: () => ++now }) fleet.onAgentExit(() => {}) await flush() - const status = fleet.fleetConnectStatus() - expect(status.state).toBe('connected') - expect(status.attempts).toBe(1) + + expect(fleet.fleetConnectStatus()).toMatchObject({ + state: 'dialed', + attempts: 1, + lastDialedAtMs: 1_002, + }) + expect(fleet.fleetConnectStatus().firstEventAtMs).toBeUndefined() expect(messaging.connected).toBe(1) - expect(status.lastConnectedAtMs).toBeTypeOf('number') + + messaging.emit('any', { type: 'connected' }) + expect(fleet.fleetConnectStatus()).toMatchObject({ + state: 'connected', + firstEventAtMs: 1_003, + lastConnectedAtMs: 1_003, + }) + }) + + it('records an established socket dropping and its later recovery', async () => { + const messaging = new FakeMessaging() + let now = 2_000 + const fleet = createClient(messaging, { now: () => ++now }) + fleet.onAgentExit(() => {}) + await flush() + + messaging.emit('any', { type: 'connected' }) + const firstEventAtMs = fleet.fleetConnectStatus().firstEventAtMs + messaging.emit('any', { type: 'disconnected' }) + expect(fleet.fleetConnectStatus()).toMatchObject({ + state: 'failed', + lastError: 'RelayEventStreamDisconnected', + }) + + messaging.emit('any', { type: 'reconnecting', attempt: 1 }) + expect(fleet.fleetConnectStatus().state).toBe('connecting') + messaging.emit('any', { type: 'connected' }) + expect(fleet.fleetConnectStatus().state).toBe('connected') + expect(fleet.fleetConnectStatus().firstEventAtMs).toBe(firstEventAtMs) + expect(fleet.fleetConnectStatus().lastError).toBeUndefined() + }) + + it('does not overwrite a synchronous connection event with dialed', async () => { + const messaging = new FakeMessaging() + messaging.connectEvent = { type: 'connected' } + const fleet = createClient(messaging) + fleet.onAgentExit(() => {}) + await flush() + + expect(fleet.fleetConnectStatus().state).toBe('connected') + expect(fleet.fleetConnectStatus().lastDialedAtMs).toBeTypeOf('number') + expect(fleet.fleetConnectStatus().firstEventAtMs).toBeTypeOf('number') }) /** @@ -1975,6 +2023,7 @@ describe('RelayFleetClient fleet connect status', () => { const healthyFleet = createClient(healthy) healthyFleet.onAgentExit(() => {}) await flush() + healthy.emit('any', { type: 'connected' }) expect(failingFleet.fleetConnectStatus().state).toBe('failed') expect(healthyFleet.fleetConnectStatus().state).toBe('connected') diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index 97f78b98..1e61302d 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -1238,12 +1238,6 @@ export class RelayFleetClient implements FleetClient { }) } - /** - * `connected` is stamped at the dial itself, not when `#subscribeEvents` - * resolves: that function also returns early when the client is disposed, and - * reporting THAT as connected would recreate the exact false-healthy signal - * this status exists to remove. - */ fleetConnectStatus(): FleetConnectStatus { return { ...this.#fleetConnect } } @@ -1252,17 +1246,27 @@ export class RelayFleetClient implements FleetClient { const messaging = await this.#ensureMessaging() await this.#ensureLifecycleAction(messaging) if (this.#disposed) return - messaging.events.connect() + // Listen before dialing so a synchronous lifecycle event cannot race past + // the observer. `connect()` is void: returning only proves the SDK accepted + // the dial, not that the WebSocket handshake completed. + const unsubscribe = messaging.events.on('any', (event) => this.#handleEvent(event)) + const statusBeforeDial = this.#fleetConnect + try { + messaging.events.connect() + } catch (error) { + unsubscribe() + throw error + } + this.#eventUnsubscribers.push(unsubscribe) this.#fleetConnect = { ...this.#fleetConnect, - state: 'connected', - lastConnectedAtMs: this.#now(), - lastError: undefined, + ...(this.#fleetConnect === statusBeforeDial ? { state: 'dialed' as const } : {}), + lastDialedAtMs: this.#now(), } - this.#eventUnsubscribers.push(messaging.events.on('any', (event) => this.#handleEvent(event))) } #handleEvent(event: RelayMessagingEvent): void { + this.#observeFleetConnectionEvent(event) switch (event.type) { case 'dmReceived': case 'groupDmReceived': @@ -1285,6 +1289,43 @@ export class RelayFleetClient implements FleetClient { } } + /** Translate the SDK's real stream lifecycle into the published status. */ + #observeFleetConnectionEvent(event: RelayMessagingEvent): void { + const now = this.#now() + switch (event.type) { + case 'disconnected': + case 'permanentlyDisconnected': + this.#fleetConnect = { + ...this.#fleetConnect, + state: 'failed', + lastFailureAtMs: now, + lastError: 'RelayEventStreamDisconnected', + } + return + case 'error': + this.#fleetConnect = { + ...this.#fleetConnect, + state: 'failed', + lastFailureAtMs: now, + lastError: 'RelayEventStreamError', + } + return + case 'reconnecting': + this.#fleetConnect = { ...this.#fleetConnect, state: 'connecting' } + return + default: + // `connected` and every data event are confirmations that the event + // stream opened. A void `connect()` call alone never reaches this arm. + this.#fleetConnect = { + ...this.#fleetConnect, + state: 'connected', + firstEventAtMs: this.#fleetConnect.firstEventAtMs ?? now, + ...(this.#fleetConnect.state !== 'connected' ? { lastConnectedAtMs: now } : {}), + lastError: undefined, + } + } + } + #emitAgentMessage(message: RelayMessage, fallbackTarget: string): void { const from = message.from?.name if (!from || from === this.#authenticatedAgentName || from === this.#agentName) return diff --git a/src/orchestrator/public-health.test.ts b/src/orchestrator/public-health.test.ts index 0bb3b648..cc2d12c5 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -767,9 +767,14 @@ describe('fleet connect health', () => { }) it('publishes the socket state unauthenticated', () => { - const health = publicHealthFromHeartbeat(withConnect(), { nowMs: BOOT_MS }) + const health = publicHealthFromHeartbeat(withConnect({ + lastDialedAtMs: BOOT_MS - 4_500, + firstEventAtMs: BOOT_MS - 4_250, + }), { nowMs: BOOT_MS }) expect(health.fleetConnect?.state).toBe('failed') expect(health.fleetConnect?.attempts).toBe(1) + expect(health.fleetConnect?.lastDialedAtMs).toBe(BOOT_MS - 4_500) + expect(health.fleetConnect?.firstEventAtMs).toBe(BOOT_MS - 4_250) }) /** `lastError` stays behind /evidence, exactly as it does for the circuit. */ @@ -795,4 +800,21 @@ describe('fleet connect health', () => { const health = publicHealthFromHeartbeat(heartbeat(), { nowMs: BOOT_MS }) expect(health.fleetConnect).toBeUndefined() }) + + it('retains a failed socket record through normalization without retaining lastError', () => { + const published = publicHealthFromHeartbeat(withConnect({ + lastDialedAtMs: BOOT_MS - 4_500, + firstEventAtMs: BOOT_MS - 4_250, + }), { nowMs: BOOT_MS }) + const normalized = normalizePublicHealth({ + ...published, + fleetConnect: { + ...published.fleetConnect, + lastError: 'connect failed to wss://relay.example?token=secret', + }, + }) + + expect(normalized?.fleetConnect).toEqual(published.fleetConnect) + expect(Object.hasOwn(normalized?.fleetConnect ?? {}, 'lastError')).toBe(false) + }) }) diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index 0c44effc..49f732b9 100644 --- a/src/orchestrator/public-health.ts +++ b/src/orchestrator/public-health.ts @@ -74,6 +74,7 @@ const EVENT_LISTENER_STATES: readonly FactoryEventListenerStatus['state'][] = [ const FLEET_CONNECT_STATES: readonly FleetConnectStatus['state'][] = [ 'never-attempted', 'connecting', + 'dialed', 'connected', 'failed', ] @@ -462,6 +463,8 @@ function fleetConnectHealth(status: FleetConnectStatus): FactoryPublicFleetConne state: enumValue(status.state, FLEET_CONNECT_STATES), ...(attempts !== undefined ? { attempts } : {}), ...optionalTimestamp('lastAttemptAtMs', status.lastAttemptAtMs), + ...optionalTimestamp('lastDialedAtMs', status.lastDialedAtMs), + ...optionalTimestamp('firstEventAtMs', status.firstEventAtMs), ...optionalTimestamp('lastConnectedAtMs', status.lastConnectedAtMs), ...optionalTimestamp('lastFailureAtMs', status.lastFailureAtMs), // `lastError` stays behind /evidence, exactly as it does for the circuit. @@ -622,6 +625,7 @@ export function normalizePublicHealth(value: unknown): FactoryPublicHealth | und const readiness = plainRecord(record.readinessReconcile) const listener = plainRecord(record.eventListener) const fleet = plainRecord(record.fleetControlPlane) + const fleetConnect = plainRecord(record.fleetConnect) const capacity = plainRecord(record.dispatchCapacity) // Re-derive the wedge from the occupants the record carries rather than // trusting its own `agentlessOccupants`: a producer that published the @@ -691,6 +695,20 @@ export function normalizePublicHealth(value: unknown): FactoryPublicHealth | und }, } : {}), + ...(fleetConnect + ? { + fleetConnect: { + state: enumValue(fleetConnect.state, FLEET_CONNECT_STATES), + ...optionalCount('attempts', fleetConnect.attempts), + ...optionalTimestamp('lastAttemptAtMs', fleetConnect.lastAttemptAtMs), + ...optionalTimestamp('lastDialedAtMs', fleetConnect.lastDialedAtMs), + ...optionalTimestamp('firstEventAtMs', fleetConnect.firstEventAtMs), + ...optionalTimestamp('lastConnectedAtMs', fleetConnect.lastConnectedAtMs), + ...optionalTimestamp('lastFailureAtMs', fleetConnect.lastFailureAtMs), + // Never retain `lastError` from a remote unauthenticated record. + }, + } + : {}), ...(capacity ? { dispatchCapacity: { diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index 3bfc0acb..29a0a387 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -133,7 +133,7 @@ export type AgentUsage = { export type FleetTrackedAgent = { invocationId?: string; node?: string } /** Lifecycle of the fleet event socket: the dial that makes this agent `online`. */ -export type FleetConnectState = 'never-attempted' | 'connecting' | 'connected' | 'failed' +export type FleetConnectState = 'never-attempted' | 'connecting' | 'dialed' | 'connected' | 'failed' /** * Why the fleet event subscription is (or is not) live. @@ -151,6 +151,10 @@ export interface FleetConnectStatus { /** How many times a subscription has been started, including the current one. */ attempts: number lastAttemptAtMs?: number + /** The SDK accepted `connect()`; this does not prove that the socket opened. */ + lastDialedAtMs?: number + /** First event that proved this client received data from the stream. */ + firstEventAtMs?: number lastConnectedAtMs?: number lastFailureAtMs?: number /** Reduced to `Name (CODE)`; never a raw transport message. */ diff --git a/src/types.ts b/src/types.ts index 06dbc14c..39a3d343 100644 --- a/src/types.ts +++ b/src/types.ts @@ -390,6 +390,8 @@ export interface FactoryPublicFleetConnectHealth { state: FleetConnectStatus['state'] | 'unknown' attempts?: number lastAttemptAtMs?: number + lastDialedAtMs?: number + firstEventAtMs?: number lastConnectedAtMs?: number lastFailureAtMs?: number } @@ -609,6 +611,8 @@ export interface FactoryStatus { counters: Record /** Broker/fleet mutation gate. An open circuit blocks new workers until a successful half-open roster probe. */ fleetControlPlane: FleetControlPlaneStatus + /** Fleet event socket status. Absent when the backend has no socket. */ + fleetConnect?: FleetConnectStatus slackDegraded?: boolean slackDegradedReason?: string /** Primary Relayfile subscription/poll registration, not event activity. */ From 7363eb2403e1d3d11631a718d9db6033ea930b02 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 24 Aug 2026 04:48:14 +0200 Subject: [PATCH 4/5] docs(fleet): define dialed in silent workspaces --- src/ports/fleet.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index 29a0a387..47df3e84 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -132,7 +132,12 @@ export type AgentUsage = { } export type FleetTrackedAgent = { invocationId?: string; node?: string } -/** Lifecycle of the fleet event socket: the dial that makes this agent `online`. */ +/** + * Lifecycle of the fleet event socket: the dial that makes this agent `online`. + * `dialed` means the SDK accepted `connect()` but Factory has not observed a + * stream event yet. It is unconfirmed, not necessarily failed: a live socket in + * a silent workspace can remain `dialed` until its first event arrives. + */ export type FleetConnectState = 'never-attempted' | 'connecting' | 'dialed' | 'connected' | 'failed' /** From d0e48ea4a3b84ff1acb79712a3c70c59c515e6b9 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 24 Aug 2026 05:21:13 +0200 Subject: [PATCH 5/5] fix(fleet): expose connection diagnostics --- src/cli/diagnose.test.ts | 65 ++++++++++++++++++++++++++++++++++++++++ src/cli/diagnose.ts | 11 +++++++ src/index.ts | 2 ++ src/ports/index.ts | 2 ++ src/types.ts | 6 ++-- 5 files changed, 84 insertions(+), 2 deletions(-) diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts index 3a05e6bb..92263866 100644 --- a/src/cli/diagnose.test.ts +++ b/src/cli/diagnose.test.ts @@ -96,6 +96,36 @@ describe('factory diagnose --deployed (#295)', () => { expect(out.text()).toContain(`lastConnectedAt : ${new Date(NOW_MS - 30_000).toISOString()}`) }) + it('renders dialed as unconfirmed until a stream event is observed', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ...healthy, + health: { + ...healthy.health, + fleetConnect: { + ...healthy.health.fleetConnect, + state: 'dialed', + firstEventAtMs: undefined, + lastConnectedAtMs: undefined, + }, + }, + }, + }, + }), + }) + + expect(code).toBe(0) + expect(out.text()).toContain('unconfirmed — the SDK accepted connect()') + expect(out.text()).toContain('healthy silent workspace may remain dialed') + }) + // The 2026-08-19/20 outage: eight consecutive failures behind `ok: true`. it('names the failing subsystem, its failure count and its error class', async () => { const out = buffer() @@ -328,6 +358,41 @@ describe('factory diagnose --deployed (#295)', () => { expect(report.evidence?.lastError).toContain('dispatch lifecycle is already terminal') }) + it('reads and renders fleet socket errors from authenticated evidence', async () => { + const out = buffer() + const code = await runFleetCli(['diagnose', '--deployed', BASE, '--token', 'op-token'], { + stdout: out, + stderr: buffer(), + env: HERMETIC_ENV, + diagnoseFetch: stubFetch({ + healthz: { + status: 200, + body: { + ...healthy, + health: { + ...healthy.health, + fleetConnect: { + state: 'failed', + attempts: 2, + lastFailureAtMs: NOW_MS - 1_000, + }, + }, + }, + }, + evidence: { + status: 200, + body: { + phase: 'running', + fleetConnect: { lastError: 'FactoryAgentRegistrationError (AGENT_EXISTS)' }, + }, + }, + }), + }) + + expect(code).toBe(0) + expect(out.text()).toContain('fleetConnect error : FactoryAgentRegistrationError (AGENT_EXISTS)') + }) + it('still diagnoses when the evidence token is rejected', async () => { const out = buffer() const code = await runFleetCli(['diagnose', '--deployed', BASE, '--token', 'stale-token', '--json'], { diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts index e1a25387..5bfdfffc 100644 --- a/src/cli/diagnose.ts +++ b/src/cli/diagnose.ts @@ -24,6 +24,7 @@ export interface DeployedEvidenceSummary { reason?: string phase?: string lastError?: string + fleetConnectLastError?: string consecutiveFailures?: number } @@ -445,11 +446,15 @@ async function readEvidence( } const body = asRecord(evidence.body) const readiness = asRecord(body.readinessReconcile) + const fleetConnect = asRecord(body.fleetConnect) return { fetched: true, httpStatus: evidence.status, ...(asText(body.phase) ? { phase: asText(body.phase) } : {}), ...(asText(readiness.lastError) ? { lastError: asText(readiness.lastError) } : {}), + ...(asText(fleetConnect.lastError) + ? { fleetConnectLastError: asText(fleetConnect.lastError) } + : {}), ...(asCount(readiness.consecutiveFailures) !== undefined ? { consecutiveFailures: asCount(readiness.consecutiveFailures) } : {}), @@ -527,6 +532,9 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st if (fleetConnect) { lines.push(' fleetConnect:') lines.push(` state : ${fleetConnect.state}`) + if (fleetConnect.state === 'dialed') { + lines.push(' confirmation : unconfirmed — the SDK accepted connect(), but no stream event has been observed; a healthy silent workspace may remain dialed') + } lines.push(` attempts : ${fleetConnect.attempts ?? '—'}`) lines.push(` lastAttemptAt : ${formatInstant(fleetConnect.lastAttemptAtMs)}`) lines.push(` lastDialedAt : ${formatInstant(fleetConnect.lastDialedAtMs)}`) @@ -558,6 +566,9 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st ` evidence : ${evidence.fetched ? `read (HTTP ${evidence.httpStatus ?? 200})` : `not read — ${evidence.reason ?? 'unavailable'}`}`, ) if (evidence.lastError) lines.push(` lastError : ${evidence.lastError}`) + if (evidence.fleetConnectLastError) { + lines.push(` fleetConnect error : ${evidence.fleetConnectLastError}`) + } } lines.push('') diff --git a/src/index.ts b/src/index.ts index 68816348..3e5b7861 100644 --- a/src/index.ts +++ b/src/index.ts @@ -345,6 +345,8 @@ export type { Subscription, AgentSpec, FleetClient, + FleetConnectState, + FleetConnectStatus, RestartPolicy, RosterEntry, SendInput, diff --git a/src/ports/index.ts b/src/ports/index.ts index d4836146..2ecd4727 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -33,6 +33,8 @@ export type { PreviewSweepInput, PreviewSweepResult, FleetClient, + FleetConnectState, + FleetConnectStatus, RestartPolicy, RosterEntry, SendInput, diff --git a/src/types.ts b/src/types.ts index 76513ee6..a4b36615 100644 --- a/src/types.ts +++ b/src/types.ts @@ -187,8 +187,10 @@ export interface FactoryLoopHeartbeat { /** Daemon-owned dispatch admission state; status readers must prefer this over a fresh local Factory instance. */ fleetControlPlane?: FleetControlPlaneStatus /** - * Whether the fleet EVENT SOCKET is connected -- the dial that makes this - * Factory's agent `online`. Absent when the backend has no socket. + * State of the fleet EVENT SOCKET dial that makes this Factory agent + * `online`. Absent when the backend has no socket. `dialed` is unconfirmed: + * the SDK accepted `connect()`, but no stream event has proved the socket + * opened, so a healthy silent workspace may remain in that state. * * Distinct from `eventListener`, which is the orchestrator's ISSUE * subscription. Conflating the two is how a fleet client that registered an