diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index be4f40b0..4de8e8d8 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -1378,14 +1378,23 @@ describe('RelayFleetClient', () => { }, ) - // A stale `offline` record plus an unreadable presence must never authorise a + // A stale `offline` record plus an UNREADABLE presence must never authorise a // seizure: the cost of being wrong is stranding a LIVE factory's credential, // and the token taken is the one it would have needed to recover. + // + // "Unreadable" means exactly that — the request failed, or the body was not a + // list. A presence list that WAS read and simply omits this agent is the + // opposite case and is covered by the must-fire tests below; conflating the + // two is the defect this pair exists to pin. it.each([ ['presence request throws', () => { throw new Error('presence unavailable') }], ['presence returns a non-list', () => ({ not: 'a list' })], - ['presence omits this agent', () => [{ agentName: 'someone-else', status: 'offline' }]], ['presence entry carries no status', () => [{ agentName: 'factory' }]], + // Omission only means absence if a row FOR this agent would have been + // recognised. A row we cannot name breaks that: if the SDK renamed the + // naming field, a LIVE agent's row stops matching and reads as absent. + ['a presence row carries no agent name', () => [{}]], + ['presence rows use an unrecognised naming field', () => [{ agent: 'factory', status: 'online' }]], ])('fails closed and does not take over when %s', async (_label, impl) => { const messaging = new FakeMessaging() const { agents, presenceImpl } = deprecatedAliasAgents() @@ -1411,11 +1420,99 @@ describe('RelayFleetClient', () => { const error = await fleet.roster().then(() => undefined, (err: unknown) => err) expect((error as Error).name).toBe('FactoryAgentRegistrationError') - expect((error as Error).message).toMatch(/cannot be confirmed offline|could not read presence/) + expect((error as Error).message).toMatch(/cannot be confirmed offline/) + // The refusal must not claim absence from presence — that is the other, + // now-permitted case, and an operator has to be able to tell them apart. + expect((error as Error).message).not.toMatch(/does not list this agent/) // The seizure must never have been attempted. expect(takeovers).toBe(0) }) + // The must-fire half of the pair. Absence from a presence list we actually + // READ is the strongest evidence the engine can give that the agent is + // offline; refusing to conclude it is what left an orphaned row permanently + // unreclaimable and gated dispatch (factory-cloud#55). The empty-list arm is + // the shape a single-agent cloud factory sees, where the dead row is the only + // agent there is. + it.each([ + ['presence lists only other agents', () => [{ agentName: 'someone-else', status: 'online' }]], + ['presence is readable and empty', () => []], + ])('reclaims the identity when %s', async (_label, impl) => { + const messaging = new FakeMessaging() + const { agents, presenceImpl } = deprecatedAliasAgents() + await (agents.register as (i: { name: string }) => Promise)({ name: 'factory' }) + presenceImpl.value = impl as () => unknown + const bootstrap: RelayClientLike = { messaging: { agents } as unknown as RelayMessaging } + const calls: Array<{ url: string; body: Record }> = [] + const fleet = new RelayFleetClient({ + workspaceKey: 'rk_live_test', + nodeId: 'node_test_1', + env: {}, + sleep: immediateSleep, + pollIntervalMs: 0, + fetch: (async (url: string, init: { body: string }) => { + calls.push({ url, body: JSON.parse(init.body) as Record }) + return new Response( + JSON.stringify({ ok: true, data: { agent_id: 'agent-1', name: 'factory', token: 'at_live_seized' } }), + { status: 200, headers: { 'content-type': 'application/json' } }, + ) + }) as unknown as typeof globalThis.fetch, + createRelay: (options) => (options.agentToken ? { messaging: messaging.asMessaging() } : bootstrap), + }) + + await expect(fleet.roster()).resolves.toBeDefined() + + // Resolving is not enough on its own: the takeover must actually have been + // issued, for this record, rather than the conflict being routed elsewhere. + expect(calls).toHaveLength(1) + expect(calls[0]?.url).toBe('https://cast.agentrelay.com/v1/agents/factory/takeover') + expect(calls[0]?.body.expected_agent_id).toBe('agent-1') + }) + + // Control arm. Both presence shapes run through ONE harness that differs in + // nothing but the presence stub, so no setup difference can absorb a swap of + // the two branches. The `not.toBe` assertion additionally kills the inert + // fixture — the failure mode where both arms take the same path for a reason + // unrelated to presence, which is exactly what would let a swapped + // implementation pass a pair of separately-written tests. + it('separates readable-and-absent from unreadable presence, and would notice a swap', async () => { + const run = async (impl: () => unknown): Promise<{ seized: boolean; message: string }> => { + const messaging = new FakeMessaging() + const { agents, presenceImpl } = deprecatedAliasAgents() + await (agents.register as (i: { name: string }) => Promise)({ name: 'factory' }) + presenceImpl.value = impl + const bootstrap: RelayClientLike = { messaging: { agents } as unknown as RelayMessaging } + let takeovers = 0 + const fleet = new RelayFleetClient({ + workspaceKey: 'rk_live_test', + env: {}, + sleep: immediateSleep, + pollIntervalMs: 0, + fetch: (async () => { + takeovers += 1 + return new Response( + JSON.stringify({ ok: true, data: { agent_id: 'agent-1', name: 'factory', token: 'at_live_seized' } }), + { status: 200, headers: { 'content-type': 'application/json' } }, + ) + }) as unknown as typeof globalThis.fetch, + createRelay: (options) => (options.agentToken ? { messaging: messaging.asMessaging() } : bootstrap), + }) + const error = await fleet.roster().then(() => undefined, (err: unknown) => err) + return { seized: takeovers > 0, message: error instanceof Error ? error.message : '' } + } + + const absent = await run(() => [{ agentName: 'someone-else', status: 'online' }]) + const unreadable = await run(() => { throw new Error('presence unavailable') }) + + expect(absent.seized).toBe(true) + expect(unreadable.seized).toBe(false) + // Fails if the harness cannot tell the two apart at all. + expect(absent.seized).not.toBe(unreadable.seized) + // And the refusal has to say WHICH of the two it was. + expect(unreadable.message).toMatch(/presence is unreadable/) + expect(absent.message).toBe('') + }) + it('re-reads the agent id and retries once when the identity moved mid-takeover', async () => { const messaging = new FakeMessaging() const { agents } = deprecatedAliasAgents() diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index e06b7609..a49b7677 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -921,52 +921,93 @@ export class RelayFleetClient implements FleetClient { } // Presence is the canonical liveness signal, and the agent record can be - // stale. Everything below fails CLOSED: only a presence row we actually - // read, for this agent, carrying a status we recognise as not-live, may - // authorise a seizure. A request that failed, a body we cannot parse, or a - // missing row is absence of evidence — not evidence of death. + // stale. Two outcomes are routed apart here, and collapsing them swaps one + // wrong conclusion for another: // - // This asymmetry is deliberate. Refusing to boot costs us a recoverable - // outage of our own; seizing a live agent's credential strands another - // running process and takes the very token it would need to recover. - let presence: unknown - try { - presence = await agents.presence() - } catch (error) { + // presence UNREADABLE (threw, timed out, or came back as something other + // than a list) — evidence of nothing. Fails CLOSED, because seizing a + // live agent's credential strands another running process and takes the + // very token it would need to recover. + // + // presence READABLE and omitting this agent — the strongest evidence the + // engine can give that the agent is offline. Proceeds. Treating this as + // inconclusive is what made an orphaned row permanently unreclaimable + // (factory-cloud#55): every registration attempt burned the bounded + // budget, which only a success resets, and dispatch stayed gated. + // + // A row that IS listed still has to clear the status check below: presence + // listing us is the case where the engine may still be holding the + // identity, so an unrecognised or missing status stays fail-closed. + const presence = await this.#readPresenceRows(agents) + const entry = presence + .map((row) => asRecord(row)) + .find((row) => readString(row, 'agentName', 'agent_name', 'name') === this.#agentName) + if (!entry) { + // Confirmed absence. Note this is also the correct reading of an empty + // list: a single-agent cloud factory whose only row is the dead one sees + // exactly that, and refusing it would re-latch the outage in its most + // common shape. + this.#log(`Presence is readable and does not list ${this.#agentName}; treating the existing record as offline`) + return + } + const seenStatus = readString(entry, 'status') + if (seenStatus === undefined) { throw new FactoryAgentRegistrationError( this.#agentName, - `could not read presence to confirm the agent is not live: ${errorMessage(error)}`, - { cause: error }, + 'presence lists this agent with no status, so it cannot be confirmed offline', ) } - if (!Array.isArray(presence)) { + if (LIVE_AGENT_STATUSES.has(seenStatus)) { throw new FactoryAgentRegistrationError( this.#agentName, - 'presence did not return a list, so the agent cannot be confirmed offline', + `presence reports this agent as "${seenStatus}"; another factory may still hold this identity`, ) } - const entry = presence - .map((row) => asRecord(row)) - .find((row) => readString(row, 'agentName', 'agent_name', 'name') === this.#agentName) - if (!entry) { + } + + /** + * Read the presence list, or fail closed. + * + * Every failure here means presence could not be read at all — a transport + * error, a timeout, or a body that is not a list. None of them say anything + * about whether the agent is alive, so none of them may authorise a seizure. + * The messages all name presence as UNREADABLE so an operator can tell this + * apart from a presence list that was read and simply did not list us. + */ + async #readPresenceRows(agents: RelayMessaging['agents']): Promise { + let presence: unknown + try { + presence = await agents.presence() + } catch (error) { throw new FactoryAgentRegistrationError( this.#agentName, - 'presence does not list this agent, so it cannot be confirmed offline', + `presence is unreadable, so the agent cannot be confirmed offline: ${errorMessage(error)}`, + { cause: error }, ) } - const seenStatus = readString(entry, 'status') - if (seenStatus === undefined) { + if (!Array.isArray(presence)) { throw new FactoryAgentRegistrationError( this.#agentName, - 'presence reported no status for this agent, so it cannot be confirmed offline', + 'presence is unreadable (it did not return a list), so the agent cannot be confirmed offline', ) } - if (LIVE_AGENT_STATUSES.has(seenStatus)) { + // A list whose rows we cannot name is not readable either, however + // well-formed the array around them is. Omission only means absence if a + // row FOR this agent would have been recognised — so if the SDK renames the + // naming field, or a row arrives without one, every row silently stops + // matching and a live agent reads as absent. That is the one way the + // permitted branch below could strand a running factory, so it is checked + // here, on the read, rather than inferred from the lookup missing. + const unnamed = presence.findIndex( + (row) => readString(asRecord(row), 'agentName', 'agent_name', 'name') === undefined, + ) + if (unnamed !== -1) { throw new FactoryAgentRegistrationError( this.#agentName, - `presence reports this agent as "${seenStatus}"; another factory may still hold this identity`, + `presence is unreadable (row ${unnamed} carries no agent name), so the agent cannot be confirmed offline`, ) } + return presence } async #takeOverAgent(workspaceKey: string, expectedAgentId: string): Promise {