diff --git a/src/cli/diagnose.test.ts b/src/cli/diagnose.test.ts index ac3f017c..92263866 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,40 @@ 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()}`) + }) + + 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`. @@ -316,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'], { @@ -765,4 +842,3 @@ describe('factory diagnose --deployed (#295)', () => { } }) }) - diff --git a/src/cli/diagnose.ts b/src/cli/diagnose.ts index 2a1ed36a..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) } : {}), @@ -523,6 +528,20 @@ export function renderDeployedDiagnosis(diagnosis: DeployedFactoryDiagnosis): st ) } } + const fleetConnect = health.fleetConnect + 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)}`) + 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') @@ -547,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/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/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..9de22526 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' @@ -37,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 @@ -124,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 @@ -1910,3 +1913,184 @@ 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 only dialed until the socket produces a confirming event', async () => { + const messaging = new FakeMessaging() + let now = 1_000 + const fleet = createClient(messaging, { now: () => ++now }) + fleet.onAgentExit(() => {}) + await flush() + + expect(fleet.fleetConnectStatus()).toMatchObject({ + state: 'dialed', + attempts: 1, + lastDialedAtMs: 1_002, + }) + expect(fleet.fleetConnectStatus().firstEventAtMs).toBeUndefined() + expect(messaging.connected).toBe(1) + + 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') + }) + + /** + * 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() + healthy.emit('any', { type: 'connected' }) + + 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://') + }) +}) + +/** + * 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 a49b7677..1e61302d 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 @@ -725,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`, ) } @@ -876,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 }, ) @@ -889,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 }, ) @@ -896,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 }, ) @@ -916,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`, ) } @@ -954,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`, ) } @@ -981,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 }, ) @@ -988,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', ) } @@ -1004,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`, ) } @@ -1197,21 +1217,56 @@ 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)}`) }) } + 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.#eventUnsubscribers.push(messaging.events.on('any', (event) => this.#handleEvent(event))) + // 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, + ...(this.#fleetConnect === statusBeforeDial ? { state: 'dialed' as const } : {}), + lastDialedAtMs: this.#now(), + } } #handleEvent(event: RelayMessagingEvent): void { + this.#observeFleetConnectionEvent(event) switch (event.type) { case 'dmReceived': case 'groupDmReceived': @@ -1234,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 @@ -1558,13 +1650,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 } 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/orchestrator/factory.ts b/src/orchestrator/factory.ts index 191d4fcd..d19e838a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -4995,6 +4995,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(), @@ -8203,6 +8206,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 26c4221e..fad9bdc4 100644 --- a/src/orchestrator/public-health.test.ts +++ b/src/orchestrator/public-health.test.ts @@ -744,6 +744,81 @@ 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({ + 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. */ + 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() + }) + + 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) + }) +}) + describe('sweep counters on the public surface (#355)', () => { const swept = ( overrides: Partial> = {}, diff --git a/src/orchestrator/public-health.ts b/src/orchestrator/public-health.ts index 7942db3d..f79996f1 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 { FACTORY_SWEEP_SKIP_REASON_CODES, @@ -14,6 +15,7 @@ import type { FactoryPublicDispatchCapacityHealth, FactoryPublicDispatchSlotOccupant, FactoryPublicEventListenerHealth, + FactoryPublicFleetConnectHealth, FactoryPublicFleetControlPlaneHealth, FactoryPublicHealth, FactoryPublicReadinessReconcileHealth, @@ -74,6 +76,14 @@ const EVENT_LISTENER_STATES: readonly FactoryEventListenerStatus['state'][] = [ 'polling', ] +const FLEET_CONNECT_STATES: readonly FleetConnectStatus['state'][] = [ + 'never-attempted', + 'connecting', + 'dialed', + 'connected', + 'failed', +] + const FLEET_CONTROL_PLANE_STATES: readonly FleetControlPlaneStatus['state'][] = [ 'closed', 'open', @@ -526,6 +536,29 @@ 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('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. + } +} + function fleetControlPlaneHealth( status: FleetControlPlaneStatus, ): FactoryPublicFleetControlPlaneHealth { @@ -573,6 +606,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 @@ -653,6 +687,7 @@ export function publicHealthFromHeartbeat( ...(readinessReconcile ? { readinessReconcile } : {}), ...(eventListener ? { eventListener } : {}), ...(fleetControlPlane ? { fleetControlPlane } : {}), + ...(fleetConnect ? { fleetConnect } : {}), ...(dispatchCapacity ? { dispatchCapacity } : {}), } } @@ -678,6 +713,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 @@ -748,6 +784,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 a032024b..47df3e84 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -132,6 +132,40 @@ export type AgentUsage = { } export type FleetTrackedAgent = { invocationId?: string; node?: string } +/** + * 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' + +/** + * 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 + /** 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. */ + lastError?: string +} + export interface FleetClient { /** Backend-wide placement locality, used when recovering a spawn ack crash gap. */ readonly placementLocality?: 'local' | 'remote' @@ -157,6 +191,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/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 f9a1d94f..a4b36615 100644 --- a/src/types.ts +++ b/src/types.ts @@ -14,6 +14,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 @@ -185,6 +186,17 @@ export interface FactoryLoopHeartbeat { dispatchCapacity?: FactoryDispatchCapacityStatus /** Daemon-owned dispatch admission state; status readers must prefer this over a fresh local Factory instance. */ fleetControlPlane?: FleetControlPlaneStatus + /** + * 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 + * agent and never connected read as healthy on every surface. + */ + fleetConnect?: FleetConnectStatus /** * Redacted projection of this record, safe to serve unauthenticated (#295). * @@ -417,6 +429,22 @@ 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 + lastDialedAtMs?: number + firstEventAtMs?: number + lastConnectedAtMs?: number + lastFailureAtMs?: number +} + export interface FactoryPublicFleetControlPlaneHealth { state: FleetControlPlaneStatus['state'] | 'unknown' consecutiveFailures: number @@ -457,6 +485,8 @@ export interface FactoryPublicHealth { readinessReconcile?: FactoryPublicReadinessReconcileHealth eventListener?: FactoryPublicEventListenerHealth fleetControlPlane?: FactoryPublicFleetControlPlaneHealth + /** Fleet event socket. NOT dispatch-gating: see DISPATCH_GATING_SUBSYSTEMS. */ + fleetConnect?: FactoryPublicFleetConnectHealth dispatchCapacity?: FactoryPublicDispatchCapacityHealth } @@ -634,6 +664,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. */