From b7356fc024839cfe4e11e9285a48ed82053a088a Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Fri, 18 Sep 2026 18:48:00 -0700 Subject: [PATCH 1/3] fix(nodes): adopt node provider and commit agent move atomically on bind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A broker-spawned agent is HTTP-registered first (provider "default", implicit direct-* location), then the broker's create-only agent.register fails and it falls back to POST /v1/nodes/:name/agents. upsertAgentNodeBinding moved locationType/locationNodeId but providerName stayed "default", so deliveries routed to the right node yet were pushed to a provider connection that does not exist — the spawned agent was never woken (AgentWorkforce/relay#1794). bindAgentToNode now adopts a provider the node actually serves (served -> sole -> "default" -> keep), stamps origin, and refuses to steal an agent active on another live node. The binding row, location move, old-binding retirement and old-node slot refund commit as one unit via runAtomicWrites (interactive transaction, D1 batch, or loud refusal), so no failure prefix can leave a node charged for a retired binding or holding an unreserved one. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- CHANGELOG.md | 6 +- .../engine/src/__tests__/atomicity.test.ts | 87 +-- .../conformance/agentLifecycle.test.ts | 578 +++++++++++++++++- .../src/__tests__/conformance/harness.ts | 113 +++- packages/engine/src/engine/node.ts | 244 ++++++-- 5 files changed, 896 insertions(+), 132 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1aed40bf..60713864 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,11 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). Packages without a separate changelog are covered by the cross-package notes below. -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- Binding an agent to a node now adopts that node's provider and guards against stealing an agent that is active on another live node. A broker-spawned agent that was HTTP-registered first (provider `default`) and then bound through the node-agents fallback kept a provider its node did not serve, so its deliveries were routed to the node but never pushed — the spawned agent was never woken. The bind's binding row, location move and node capacity counters now commit as one atomic unit, so a failure part-way through can no longer leave a node charged for a binding it retired, or holding an active binding it never reserved capacity for. ## [8.11.2] - 2026-09-19 diff --git a/packages/engine/src/__tests__/atomicity.test.ts b/packages/engine/src/__tests__/atomicity.test.ts index de63b930..a0c0f759 100644 --- a/packages/engine/src/__tests__/atomicity.test.ts +++ b/packages/engine/src/__tests__/atomicity.test.ts @@ -2,9 +2,13 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { readFileSync } from 'node:fs'; import { and, eq } from 'drizzle-orm'; import { + attachFakeBatch as attachFakeBatchOn, + injectInsertFailure, + injectUpdateFailure, makeNodeStack, createWorkspace, registerAgent, + stripTransactionCapability, type TestStack, } from './conformance/harness.js'; import { @@ -32,7 +36,7 @@ import { applyStatusEventEffect, recordSessionEventWithIdempotency, } from '../engine/sessionEvent.js'; -import type { AtomicWrite, EngineDb, TransactionCapability } from '../ports/database.js'; +import type { EngineDb, TransactionCapability } from '../ports/database.js'; /** * Atomicity of multi-statement write paths. @@ -78,87 +82,10 @@ describe('atomic write paths', () => { return { ws, alice, bob, channelId: channel.id, db }; } - /** - * Wrap a built statement so it fails when *executed* (awaited), not when - * built. Write paths build their statement list up front, so a build-time - * throw would abort before any write executes and never exercise rollback; - * an execution-time failure lands mid-transaction / mid-batch / mid-sequence - * — the crash the atomicity machinery exists for. Builder chaining and - * `toSQL()` still delegate to the real statement. - */ - function failOnExecute(target: T, message: string): T { - return new Proxy(target, { - get(obj, prop) { - if (prop === 'then') { - return (onFulfilled?: (v: unknown) => unknown, onRejected?: (e: unknown) => unknown) => - Promise.reject(new Error(message)).then(onFulfilled, onRejected); - } - const value = Reflect.get(obj, prop) as unknown; - if (typeof value === 'function') { - return (...args: unknown[]) => { - const result = (value as (...a: unknown[]) => unknown).apply(obj, args); - return result && typeof result === 'object' ? failOnExecute(result as object, message) : result; - }; - } - return value; - }, - }); - } - - /** Make statements inserting into `table` fail at execution; returns a restore function. */ - function injectInsertFailure(db: EngineDb, table: unknown, message: string): () => void { - const handle = db as unknown as { insert: (t: unknown) => object }; - const real = handle.insert.bind(db); - handle.insert = (t: unknown) => { - const builder = real(t); - return t === table ? failOnExecute(builder, message) : builder; - }; - return () => { handle.insert = real; }; - } + const stripCapability = stripTransactionCapability; - /** Make statements updating `table` fail at execution; returns a restore function. */ - function injectUpdateFailure(db: EngineDb, table: unknown, message: string): () => void { - const handle = db as unknown as { update: (t: unknown) => object }; - const real = handle.update.bind(db); - handle.update = (t: unknown) => { - const builder = real(t); - return t === table ? failOnExecute(builder, message) : builder; - }; - return () => { handle.update = real; }; - } - - function stripCapability(db: EngineDb): void { - delete (db as Partial).withTransaction; - } - - /** - * Turn the Node handle into a D1-shaped one: no `withTransaction`, but a - * `batch()` that executes every statement inside one underlying SQLite - * transaction (all-or-nothing, like D1) and records each batch's SQL. - */ function attachFakeBatch(db: EngineDb, beforeExecute?: () => Promise): string[][] { - stripCapability(db); - const sqlite = stack.runtime.handle.sqlite; - const batches: string[][] = []; - (db as unknown as Record).batch = async ( - statements: ReadonlyArray, - ): Promise => { - batches.push(statements.map((s) => s.toSQL().sql)); - await beforeExecute?.(); - sqlite.exec('BEGIN IMMEDIATE'); - try { - const results: unknown[] = []; - for (const statement of statements) { - results.push(await statement); - } - sqlite.exec('COMMIT'); - return results; - } catch (err) { - if (sqlite.inTransaction) sqlite.exec('ROLLBACK'); - throw err; - } - }; - return batches; + return attachFakeBatchOn(stack, db, beforeExecute); } function expectStatementOn(batch: string[], verb: 'insert' | 'update', table: string): void { diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index b8826a2c..c2562615 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -1,16 +1,25 @@ import { invokeWithConcurrentReplay } from './invocationReplay.js'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { and, eq } from 'drizzle-orm'; -import { actionInvocations, agentNodeBindings, agents, channelMembers, nodes, pendingEvents, workspaceEvents } from '../../db/schema.js'; +import { actionInvocations, agentNodeBindings, agents, channelMembers, deliveries, nodeProviders, nodes, pendingEvents, workspaceEvents } from '../../db/schema.js'; import { AGENT_LIVENESS_TTL_MS, sweepStaleAgents } from '../../engine/agent.js'; +import { bindAgentToNode } from '../../engine/node.js'; +import { NODE_LIVENESS_TTL_MS } from '../../engine/placement.js'; import { attachDirectNodeSocket, + attachFakeBatch, createWorkspace, + deliverFramesOfType, + FakeSocket, + injectInsertFailure, + injectUpdateFailure, makeNodeStack, registerAgent, + stripTransactionCapability, type TestStack, } from './harness.js'; import { sha256Hex } from '../../lib/crypto.js'; +import type { EngineDb, TransactionCapability } from '../../ports/database.js'; describe('agent presence and release lifecycle', () => { let stack: TestStack; @@ -1745,3 +1754,570 @@ describe('agent presence and release lifecycle', () => { await handle.handleClose(); }); }); + +/** + * A node bind is a location move, not just a roster row. The broker's spawn + * path HTTP-registers an agent first (leaving it on its implicit `direct-*` + * pseudo-node), then falls back to this endpoint when its create-only + * `agent.register` loses to the row that already exists. Delivery routing + * joins bindings only where `agents.location_node_id` matches the bound node, + * so a bind that leaves the agent row untouched strands the spawned agent on a + * dead pseudo-node and it is never woken. + */ +describe('node agent binding adopts the agent location', () => { + let stack: TestStack; + + beforeEach(() => { stack = makeNodeStack(); }); + afterEach(() => stack.close()); + + async function enrollNode(workspaceKey: string, nodeId: string, name: string) { + const res = await stack.app.request('/v1/nodes', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${workspaceKey}` }, + body: JSON.stringify({ node_id: nodeId, name, role: 'broker', max_agents: 4, tags: ['test'], version: 'v0' }), + }); + expect(res.status).toBe(201); + return (await res.json() as { data: { token: string } }).data.token; + } + + function bindAgent( + workspaceKey: string, + nodeName: string, + agentName: string, + body: Record = {}, + ) { + return stack.app.request(`/v1/nodes/${nodeName}/agents`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${workspaceKey}` }, + body: JSON.stringify({ agent_name: agentName, ...body }), + }); + } + + /** Every agent column the bind path writes. */ + async function readAgent(agentId: string) { + const [row] = await stack.runtime.deps.db + .select({ + locationType: agents.locationType, + locationNodeId: agents.locationNodeId, + status: agents.status, + providerName: agents.providerName, + originNodeId: agents.originNodeId, + sessionRef: agents.sessionRef, + lastSeen: agents.lastSeen, + }) + .from(agents) + .where(eq(agents.id, agentId)); + return row; + } + + async function activeBindingNodeIds(workspaceId: string, agentId: string): Promise { + const rows = await stack.runtime.deps.db + .select({ nodeId: agentNodeBindings.nodeId }) + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, workspaceId), + eq(agentNodeBindings.agentId, agentId), + eq(agentNodeBindings.status, 'active'), + )); + return rows.map((row) => row.nodeId).sort(); + } + + /** The capacity counters a bind reserves on its target and releases elsewhere. */ + async function nodeSlots(nodeId: string) { + const [row] = await stack.runtime.deps.db + .select({ activeAgents: nodes.activeAgents, reservedAgents: nodes.reservedAgents }) + .from(nodes) + .where(eq(nodes.id, nodeId)); + return row; + } + + /** Age a node's heartbeat past the liveness TTL, leaving its bindings alone. */ + async function expireNode(workspaceId: string, nodeId: string) { + await stack.runtime.deps.db + .update(nodes) + .set({ lastHeartbeatAt: new Date(Date.now() - NODE_LIVENESS_TTL_MS - 60_000) }) + .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, nodeId))); + } + + /** HTTP registration always mints `default`; other providers come from a node. */ + async function setAgentProvider(agentId: string, providerName: string) { + await stack.runtime.deps.db + .update(agents) + .set({ providerName }) + .where(eq(agents.id, agentId)); + } + + /** Attach a node-control socket and register `providerName` on the node. */ + async function attachProvider(workspaceId: string, nodeId: string, nodeName: string, providerName: string) { + const sock = new FakeSocket(); + const handle = stack.runtime.realtime.attachNodeSocket(workspaceId, nodeId, sock); + await handle.handleMessage(JSON.stringify({ + v: 1, + id: `reg-${nodeId}-${providerName}`, + type: 'node.register', + node_id: nodeId, + name: nodeName, + provider: { name: providerName, instance_id: `${providerName}-i1` }, + capabilities: [{ name: 'spawn:claude', kind: 'capacity' }], + max_agents: 4, + tags: ['test'], + version: 'v1', + resume_cursor: null, + })); + expect(sock.ofType('error')).toEqual([]); + return { sock, handle }; + } + + it('moves an HTTP-registered agent off its implicit direct node onto the bound node', async () => { + const ws = await createWorkspace(stack.app, 'bind-adopts-location'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'fallback-bound-agent'); + await enrollNode(ws.workspaceKey, 'node_broker', 'broker-host'); + + // The HTTP registration parked the agent on its implicit pseudo-node. + const [before] = await stack.runtime.deps.db + .select({ locationType: agents.locationType, locationNodeId: agents.locationNodeId }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(before).toEqual({ locationType: 'via_node', locationNodeId: `node_direct_${target.agentId}` }); + + expect((await bindAgent(ws.workspaceKey, 'broker-host', target.name)).status).toBe(201); + + const [located] = await stack.runtime.deps.db + .select({ + locationType: agents.locationType, + locationNodeId: agents.locationNodeId, + status: agents.status, + originNodeId: agents.originNodeId, + }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(located).toEqual({ + locationType: 'via_node', + locationNodeId: 'node_broker', + status: 'active', + // Location moves; origin does not. HTTP registration stamped the + // pseudo-node as this agent's origin and that is where identity + // recovery authority stays. + originNodeId: `node_direct_${target.agentId}`, + }); + expect(await stack.runtime.deps.db + .select({ id: agentNodeBindings.id }) + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, ws.workspaceId), + eq(agentNodeBindings.agentId, target.agentId), + eq(agentNodeBindings.nodeId, 'node_broker'), + eq(agentNodeBindings.status, 'active'), + ))) + .toHaveLength(1); + }); + + it('routes a channel delivery for a fallback-bound agent through its bound node', async () => { + const ws = await createWorkspace(stack.app, 'bind-adopts-routing'); + const speaker = await registerAgent(stack.app, ws.workspaceKey, 'speaker'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'routed-bound-agent'); + await enrollNode(ws.workspaceKey, 'node_broker', 'broker-host'); + const { sock } = await attachProvider(ws.workspaceId, 'node_broker', 'broker-host', 'broker'); + + expect((await bindAgent(ws.workspaceKey, 'broker-host', target.name)).status).toBe(201); + + const posted = await stack.app.request('/v1/channels/general/messages', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${speaker.token}` }, + body: JSON.stringify({ text: 'wake up' }), + }); + expect(posted.status).toBe(201); + const message = await posted.json() as { data: { id: string } }; + + const [route] = await stack.runtime.deps.db + .select({ routeNodeId: deliveries.routeNodeId }) + .from(deliveries) + .where(and( + eq(deliveries.workspaceId, ws.workspaceId), + eq(deliveries.messageId, message.data.id), + eq(deliveries.agentId, target.agentId), + )); + expect(route).toEqual({ routeNodeId: 'node_broker' }); + + // Fanout publishes its completion through waitUntil. + await stack.settle(); + expect(deliverFramesOfType(sock, 'message.created')).toEqual([ + expect.objectContaining({ + type: 'deliver', + msg_id: message.data.id, + agent: target.name, + }), + ]); + }); + + it('refuses to steal an agent that is active on another live node', async () => { + const ws = await createWorkspace(stack.app, 'bind-location-conflict'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'contested-agent'); + await enrollNode(ws.workspaceKey, 'node_owner', 'owner-host'); + await enrollNode(ws.workspaceKey, 'node_thief', 'thief-host'); + await attachProvider(ws.workspaceId, 'node_owner', 'owner-host', 'broker'); + + expect((await bindAgent(ws.workspaceKey, 'owner-host', target.name)).status).toBe(201); + + const stolen = await bindAgent(ws.workspaceKey, 'thief-host', target.name); + expect(stolen.status).toBe(409); + expect((await stolen.json() as { error: { code: string } }).error) + .toMatchObject({ code: 'agent_location_conflict' }); + + const [held] = await stack.runtime.deps.db + .select({ locationNodeId: agents.locationNodeId }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(held).toEqual({ locationNodeId: 'node_owner' }); + expect(await stack.runtime.deps.db + .select({ id: agentNodeBindings.id }) + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, ws.workspaceId), + eq(agentNodeBindings.agentId, target.agentId), + eq(agentNodeBindings.nodeId, 'node_thief'), + eq(agentNodeBindings.status, 'active'), + ))) + .toHaveLength(0); + }); + + it('adopts the sole provider of the node it is bound to', async () => { + const ws = await createWorkspace(stack.app, 'bind-adopts-provider'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'adopted-agent'); + await enrollNode(ws.workspaceKey, 'node_broker', 'broker-host'); + await attachProvider(ws.workspaceId, 'node_broker', 'broker-host', 'broker'); + + const [registered] = await stack.runtime.deps.db + .select({ providerName: agents.providerName }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(registered).toEqual({ providerName: 'default' }); + expect(await stack.runtime.deps.db + .select({ name: nodeProviders.name }) + .from(nodeProviders) + .where(and(eq(nodeProviders.workspaceId, ws.workspaceId), eq(nodeProviders.nodeId, 'node_broker')))) + .toEqual([{ name: 'broker' }]); + + expect((await bindAgent(ws.workspaceKey, 'broker-host', target.name)).status).toBe(201); + + const [adopted] = await stack.runtime.deps.db + .select({ providerName: agents.providerName, locationNodeId: agents.locationNodeId }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(adopted).toEqual({ providerName: 'broker', locationNodeId: 'node_broker' }); + }); + + it('stamps session, liveness and origin with the move, and keeps the first origin on a later one', async () => { + const ws = await createWorkspace(stack.app, 'bind-stamps-identity'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'stamped-agent'); + await enrollNode(ws.workspaceKey, 'node_first', 'first-host'); + await enrollNode(ws.workspaceKey, 'node_second', 'second-host'); + const before = await readAgent(target.agentId); + + expect((await bindAgent(ws.workspaceKey, 'first-host', target.name, { session_ref: 'sess-1' })).status).toBe(201); + + const stamped = await readAgent(target.agentId); + expect(stamped).toMatchObject({ + locationNodeId: 'node_first', + status: 'active', + sessionRef: 'sess-1', + originNodeId: `node_direct_${target.agentId}`, + }); + expect(stamped.lastSeen.getTime()).toBeGreaterThanOrEqual(before.lastSeen.getTime()); + expect(await stack.runtime.deps.db + .select({ sessionRef: agentNodeBindings.sessionRef }) + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.agentId, target.agentId), + eq(agentNodeBindings.nodeId, 'node_first'), + ))) + .toEqual([{ sessionRef: 'sess-1' }]); + + // A second move omits the session ref: the binding row for the new node + // carries none, while the agent keeps the session it is still running. + expect((await bindAgent(ws.workspaceKey, 'second-host', target.name)).status).toBe(201); + + expect(await readAgent(target.agentId)).toMatchObject({ + locationNodeId: 'node_second', + sessionRef: 'sess-1', + // Origin records where the agent came from, so a move never rewrites it. + originNodeId: `node_direct_${target.agentId}`, + }); + expect(await activeBindingNodeIds(ws.workspaceId, target.agentId)).toEqual(['node_second']); + }); + + it('moves an agent off its implicit direct node even while that pseudo-node is live', async () => { + const ws = await createWorkspace(stack.app, 'bind-live-direct-node'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'connected-agent'); + await enrollNode(ws.workspaceKey, 'node_broker', 'broker-host'); + const direct = await attachDirectNodeSocket(stack, ws.workspaceId, target); + + // The pseudo-node is as live as a node gets — online with a fresh + // heartbeat. It still never owns the agent against a real node, or the + // spawn fallback could never bind an agent that is already connected. + const [directNode] = await stack.runtime.deps.db + .select({ status: nodes.status, lastHeartbeatAt: nodes.lastHeartbeatAt }) + .from(nodes) + .where(eq(nodes.id, direct.nodeId)); + expect(directNode.status).toBe('online'); + expect(Date.now() - directNode.lastHeartbeatAt!.getTime()).toBeLessThan(NODE_LIVENESS_TTL_MS); + + expect((await bindAgent(ws.workspaceKey, 'broker-host', target.name)).status).toBe(201); + + expect(await readAgent(target.agentId)).toMatchObject({ locationNodeId: 'node_broker' }); + expect(await activeBindingNodeIds(ws.workspaceId, target.agentId)).toEqual(['node_broker']); + await direct.handle.handleClose(); + }); + + it('takes over an agent whose owning node has gone stale, releasing its slot', async () => { + const ws = await createWorkspace(stack.app, 'bind-stale-owner'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'stranded-agent'); + await enrollNode(ws.workspaceKey, 'node_owner', 'owner-host'); + await enrollNode(ws.workspaceKey, 'node_rescue', 'rescue-host'); + await attachProvider(ws.workspaceId, 'node_owner', 'owner-host', 'broker'); + + expect((await bindAgent(ws.workspaceKey, 'owner-host', target.name)).status).toBe(201); + expect(await nodeSlots('node_owner')).toEqual({ activeAgents: 1, reservedAgents: 0 }); + + // The host stopped heartbeating: its claim on the agent expires with it. + await expireNode(ws.workspaceId, 'node_owner'); + expect((await bindAgent(ws.workspaceKey, 'rescue-host', target.name)).status).toBe(201); + + expect(await readAgent(target.agentId)).toMatchObject({ locationNodeId: 'node_rescue' }); + expect(await activeBindingNodeIds(ws.workspaceId, target.agentId)).toEqual(['node_rescue']); + expect(await nodeSlots('node_owner')).toEqual({ activeAgents: 0, reservedAgents: 0 }); + expect(await nodeSlots('node_rescue')).toEqual({ activeAgents: 1, reservedAgents: 0 }); + }); + + it('takes over an agent whose location node was pruned out from under it', async () => { + const ws = await createWorkspace(stack.app, 'bind-pruned-owner'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'orphaned-agent'); + await enrollNode(ws.workspaceKey, 'node_rescue', 'rescue-host'); + // Deleting a node nulls the location it owned (`on delete set null`), so a + // pruned host leaves an active agent nowhere. Nothing is left to hold the + // identity: the bind is a recovery, not a steal. + await stack.runtime.deps.db + .update(agents) + .set({ locationNodeId: null, status: 'active' }) + .where(eq(agents.id, target.agentId)); + + expect((await bindAgent(ws.workspaceKey, 'rescue-host', target.name)).status).toBe(201); + expect(await readAgent(target.agentId)).toMatchObject({ + locationType: 'via_node', + locationNodeId: 'node_rescue', + }); + // A legacy row with no origin gets one stamped by the node adopting it. + await stack.runtime.deps.db + .update(agents) + .set({ originNodeId: null }) + .where(eq(agents.id, target.agentId)); + await enrollNode(ws.workspaceKey, 'node_later', 'later-host'); + expect((await bindAgent(ws.workspaceKey, 'later-host', target.name)).status).toBe(201); + expect(await readAgent(target.agentId)).toMatchObject({ originNodeId: 'node_later' }); + }); + + /** + * Provider adoption keeps the agent addressable: deliveries are pushed to + * the `(node, provider)` socket named by `agents.provider_name`, so the bind + * may only rewrite it when the node's own providers make the answer clear. + */ + const providerAdoption = [ + { + what: 'adopts the only provider a node serves', + providers: ['broker'], + current: 'default', + expected: 'broker', + }, + { + what: 'falls back to the synthetic default when the node serves several and none is the agent\'s', + providers: ['broker', 'default'], + current: 'codex', + expected: 'default', + }, + { + what: 'keeps a provider the node already serves instead of the default', + providers: ['broker', 'default'], + current: 'broker', + expected: 'broker', + }, + { + what: 'keeps the agent provider when a multi-provider node offers no default', + providers: ['broker', 'codex'], + current: 'default', + expected: 'default', + }, + { + what: 'keeps the agent provider when the node has registered no providers', + providers: [], + current: 'codex', + expected: 'codex', + }, + ]; + + for (const [index, adoption] of providerAdoption.entries()) { + it(adoption.what, async () => { + const ws = await createWorkspace(stack.app, `bind-provider-${index}`); + const target = await registerAgent(stack.app, ws.workspaceKey, 'provider-agent'); + await enrollNode(ws.workspaceKey, 'node_broker', 'broker-host'); + for (const providerName of adoption.providers) { + await attachProvider(ws.workspaceId, 'node_broker', 'broker-host', providerName); + } + await setAgentProvider(target.agentId, adoption.current); + + expect(await stack.runtime.deps.db + .select({ name: nodeProviders.name }) + .from(nodeProviders) + .where(and( + eq(nodeProviders.workspaceId, ws.workspaceId), + eq(nodeProviders.nodeId, 'node_broker'), + )) + .then((rows) => rows.map((row) => row.name).sort())) + .toEqual([...adoption.providers].sort()); + + expect((await bindAgent(ws.workspaceKey, 'broker-host', target.name)).status).toBe(201); + + expect(await readAgent(target.agentId)).toMatchObject({ + providerName: adoption.expected, + locationNodeId: 'node_broker', + }); + }); + } + + /** + * The bind writes a binding row, the agent's location/provider/origin, the + * slot it reserves on the node it moves onto and the slot it gives back on + * the node it moves off. A failure anywhere in that sequence must leave none + * of it behind, on every adapter shape — including D1, which has no + * interactive transaction and keeps whatever already ran. + * + * Half a move is worse than no move, because the retry cannot see that it is + * half done and reads the leftovers as work already finished: + * + * - a binding that committed while its reservation was compensated away + * looks bound-and-charged, so the retry reserves nothing and the node runs + * one agent over `max_agents` forever; + * - a binding retired without its slot being given back looks released, so + * the retry finds nothing active on the old node and never refunds it. + * + * So the move commits as one unit, and every failure point below is checked + * for exactly that: nothing changed, and the retry still completes the move + * with both slot counters right. + */ + describe('a failed bind leaves nothing behind', () => { + const INJECTED = 'injected bind failure'; + + /** + * Park the agent on `owner-host`, then let that host go stale so the bind + * to `rescue-host` is a genuine move: it reserves a slot on the rescue + * node, retires the owner's binding and refunds the owner's slot. + */ + async function stagedMove(label: string) { + const ws = await createWorkspace(stack.app, `bind-failure-${label}`); + const target = await registerAgent(stack.app, ws.workspaceKey, 'moved-agent'); + await enrollNode(ws.workspaceKey, 'node_owner', 'owner-host'); + await enrollNode(ws.workspaceKey, 'node_rescue', 'rescue-host'); + await attachProvider(ws.workspaceId, 'node_owner', 'owner-host', 'broker'); + await attachProvider(ws.workspaceId, 'node_rescue', 'rescue-host', 'rescue'); + + expect((await bindAgent(ws.workspaceKey, 'owner-host', target.name)).status).toBe(201); + expect(await nodeSlots('node_owner')).toEqual({ activeAgents: 1, reservedAgents: 0 }); + await expireNode(ws.workspaceId, 'node_owner'); + + return { ws, target, db: stack.runtime.deps.db as unknown as EngineDb }; + } + + /** Everything the move touches, in one comparable value. */ + async function snapshot(workspaceId: string, agentId: string) { + return { + agent: await readAgent(agentId), + bindings: await activeBindingNodeIds(workspaceId, agentId), + owner: await nodeSlots('node_owner'), + rescue: await nodeSlots('node_rescue'), + }; + } + + const shapes = [ + { + what: 'transactional', + // The Node adapter's handle rolls the whole move back. + apply: () => {}, + }, + { + what: 'D1 batch', + // D1 has no interactive transaction; its atomicity is `batch()`. + apply: (db: EngineDb) => { attachFakeBatch(stack, db); }, + }, + ]; + + const failurePoints = [ + { + at: 'the binding insert', + inject: (db: EngineDb) => injectInsertFailure(db, agentNodeBindings, INJECTED), + }, + { + at: 'the agent location move', + inject: (db: EngineDb) => injectUpdateFailure(db, agents, INJECTED), + }, + { + at: 'the old binding retirement', + inject: (db: EngineDb) => injectUpdateFailure(db, agentNodeBindings, INJECTED), + }, + { + at: 'the old slot refund', + // Both slot writes update `nodes`: the rescue reservation is built + // first, the owner's refund second. One shot only, so the reservation's + // own compensating release still runs. + inject: (db: EngineDb) => injectUpdateFailure(db, nodes, INJECTED, { skip: 1, times: 1 }), + }, + ]; + + for (const shape of shapes) { + for (const point of failurePoints) { + it(`restores every row and both slot counters when ${point.at} fails (${shape.what})`, async () => { + const { ws, target, db } = await stagedMove(`${shape.what}-${point.at}`.replace(/\s+/g, '-')); + shape.apply(db); + const before = await snapshot(ws.workspaceId, target.agentId); + + const restore = point.inject(db); + await expect(bindAgentToNode(db, ws.workspaceId, 'rescue-host', target.name)) + .rejects.toThrow(INJECTED); + restore(); + + expect(await snapshot(ws.workspaceId, target.agentId)).toEqual(before); + + // The retry reads untouched state, so it reserves and refunds exactly + // once and the move lands whole. + await bindAgentToNode(db, ws.workspaceId, 'rescue-host', target.name); + expect(await snapshot(ws.workspaceId, target.agentId)).toMatchObject({ + agent: expect.objectContaining({ locationNodeId: 'node_rescue', providerName: 'rescue' }), + bindings: ['node_rescue'], + owner: { activeAgents: 0, reservedAgents: 0 }, + rescue: { activeAgents: 1, reservedAgents: 0 }, + }); + }); + } + } + + it('refuses the move on a handle that can neither roll back nor batch', async () => { + const { ws, target, db } = await stagedMove('bare-handle'); + const capability = (db as EngineDb & TransactionCapability).withTransaction; + stripTransactionCapability(db); + const before = await snapshot(ws.workspaceId, target.agentId); + + // Nothing can undo a half-applied move here, so the bind never starts + // one: it is refused before the first statement and the reservation it + // took is handed straight back. + await expect(bindAgentToNode(db, ws.workspaceId, 'rescue-host', target.name)) + .rejects.toThrow(/Atomic write capability required/); + expect(await snapshot(ws.workspaceId, target.agentId)).toEqual(before); + + (db as EngineDb & TransactionCapability).withTransaction = capability; + await bindAgentToNode(db, ws.workspaceId, 'rescue-host', target.name); + expect(await snapshot(ws.workspaceId, target.agentId)).toMatchObject({ + bindings: ['node_rescue'], + owner: { activeAgents: 0, reservedAgents: 0 }, + rescue: { activeAgents: 1, reservedAgents: 0 }, + }); + }); + }); +}); diff --git a/packages/engine/src/__tests__/conformance/harness.ts b/packages/engine/src/__tests__/conformance/harness.ts index 6191cb65..101940d4 100644 --- a/packages/engine/src/__tests__/conformance/harness.ts +++ b/packages/engine/src/__tests__/conformance/harness.ts @@ -6,7 +6,8 @@ import type { Hono } from 'hono'; import { createEngine } from '../../engine.js'; import { createNodeRuntime, type NodeRuntime, type EngineSocket } from '../../adapters/node/index.js'; import type { AppEnv } from '../../env.js'; -import type { EngineConfig, EntitlementsProvider } from '../../ports/index.js'; +import type { EngineConfig, EngineDb, EntitlementsProvider } from '../../ports/index.js'; +import type { AtomicWrite, TransactionCapability } from '../../ports/database.js'; export interface TestStack { app: Hono; @@ -202,3 +203,113 @@ export function deliverFramesOfType(sock: FakeSocket, type: string): Record[] { return sock.ofType('context.update').filter((frame) => frame.event === event); } + +/** + * Wrap a built statement so it fails when *executed* (awaited), not when built. + * Write paths build their statement list up front, so a build-time throw would + * abort before any write executes and never exercise rollback or compensation; + * an execution-time failure lands mid-transaction / mid-batch / mid-sequence — + * the crash the atomicity machinery exists for. Builder chaining and `toSQL()` + * still delegate to the real statement. + */ +export function failOnExecute(target: T, message: string): T { + return new Proxy(target, { + get(obj, prop) { + if (prop === 'then') { + return (onFulfilled?: (v: unknown) => unknown, onRejected?: (e: unknown) => unknown) => + Promise.reject(new Error(message)).then(onFulfilled, onRejected); + } + const value = Reflect.get(obj, prop) as unknown; + if (typeof value === 'function') { + return (...args: unknown[]) => { + const result = (value as (...a: unknown[]) => unknown).apply(obj, args); + return result && typeof result === 'object' ? failOnExecute(result as object, message) : result; + }; + } + return value; + }, + }); +} + +/** + * Aim a failure at one statement when a write path touches the same table more + * than once. The counter advances per *built* statement and every path builds + * in execution order, so `{ skip: 1 }` spares the first and fails the second. + * `times` bounds how many fail: `{ times: 1 }` is a one-shot outage, which is + * what a compensating write issued from the same path has to survive. + */ +export interface InjectFailureOptions { skip?: number; times?: number } + +function injectBuilderFailure( + db: EngineDb, + verb: 'insert' | 'update', + table: unknown, + message: string, + options: InjectFailureOptions, +): () => void { + const handle = db as unknown as Record<'insert' | 'update', (t: unknown) => object>; + const real = handle[verb].bind(db); + const skip = options.skip ?? 0; + const times = options.times ?? Number.POSITIVE_INFINITY; + let built = 0; + handle[verb] = (t: unknown) => { + const builder = real(t); + if (t !== table) return builder; + const index = built++; + return index >= skip && index < skip + times ? failOnExecute(builder, message) : builder; + }; + return () => { handle[verb] = real; }; +} + +/** Make statements inserting into `table` fail at execution; returns a restore function. */ +export function injectInsertFailure( + db: EngineDb, table: unknown, message: string, options: InjectFailureOptions = {}, +): () => void { + return injectBuilderFailure(db, 'insert', table, message, options); +} + +/** Make statements updating `table` fail at execution; returns a restore function. */ +export function injectUpdateFailure( + db: EngineDb, table: unknown, message: string, options: InjectFailureOptions = {}, +): () => void { + return injectBuilderFailure(db, 'update', table, message, options); +} + +/** Drop the interactive-transaction capability, as the D1 handle never has it. */ +export function stripTransactionCapability(db: EngineDb): void { + delete (db as Partial).withTransaction; +} + +/** + * Turn the Node handle into a D1-shaped one: no `withTransaction`, but a + * `batch()` that executes every statement inside one underlying SQLite + * transaction (all-or-nothing, like D1) and records each batch's SQL. + */ +export function attachFakeBatch( + stack: TestStack, + db: EngineDb, + beforeExecute?: () => Promise, +): string[][] { + stripTransactionCapability(db); + const sqlite = stack.runtime.handle.sqlite; + const batches: string[][] = []; + (db as unknown as Record).batch = async ( + statements: ReadonlyArray, + ): Promise => { + batches.push(statements.map((s) => s.toSQL().sql)); + await beforeExecute?.(); + sqlite.exec('BEGIN IMMEDIATE'); + try { + const results: unknown[] = []; + for (const statement of statements) { + results.push(await statement); + } + sqlite.exec('COMMIT'); + return results; + } catch (err) { + if (sqlite.inTransaction) sqlite.exec('ROLLBACK'); + throw err; + } + }; + return batches; +} diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 37f0de9b..541f7f74 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -23,8 +23,8 @@ import { actionInvocations, agents, agentNodeBindings, channelMembers, channels, import { randomHex, sha256Hex } from '../lib/crypto.js'; import { codedError } from '../lib/httpError.js'; import { acceptTaskInvocation, completeTaskInvocation } from './taskInvocation.js'; -import { runAtomic } from '../ports/database.js'; -import type { EngineDb } from '../ports/database.js'; +import { runAtomic, runAtomicWrites } from '../ports/database.js'; +import type { AtomicWrite, EngineDb } from '../ports/database.js'; import { isProviderAgentDeliveryReady, type NodeConnectionRegistry } from '../ports/realtime.js'; import { generateId } from './snowflake.js'; import { assertRegistrableAgentName } from './agent.js'; @@ -101,6 +101,31 @@ function isImplicitDirectLocation(agent: Pick return agent.locationNodeId === directNodeIdForAgent(agent.id); } +/** + * An agent adopted onto a node keeps a provider identity that node can serve: + * deliveries are pushed to the `(node, provider)` socket named by + * `agents.provider_name`, so a provider the node does not serve is a silent + * dead end. Precedence, most specific first: + * + * 1. The node already serves the agent's provider — keep it. It routes today, + * and it is the closest thing to evidence of which provider owns the agent; + * rewriting it would hand the agent to a sibling provider on the same host. + * 2. The node serves exactly one provider — adopt it; there is no other answer. + * 3. The node serves the synthetic `default` — adopt it as the generic + * fallback, since the agent's own provider is not served here. + * 4. Otherwise (several named providers, none of them the agent's) nothing + * identifies the owner, so the existing provider stands rather than guessing. + * + * A node with no registered providers at all falls through to (4): no socket + * exists to adopt, and rewriting to a name nobody serves would not help. + */ +function adoptNodeProviderName(current: string, nodeProviderNames: string[]): string { + if (nodeProviderNames.includes(current)) return current; + if (nodeProviderNames.length === 1) return nodeProviderNames[0]; + if (nodeProviderNames.includes(DEFAULT_PROVIDER_NAME)) return DEFAULT_PROVIDER_NAME; + return current; +} + function normalizeCapabilities(capabilities: CapabilityLike[]): FleetCapability[] { return capabilities.map((capability) => ( typeof capability === 'string' ? { name: capability } : capability @@ -930,57 +955,101 @@ async function autoJoinGeneral(db: Db, workspaceId: string, agentId: string) { } } -async function upsertAgentNodeBinding( +interface AgentNodeBindingOpts { + sessionRef?: string | null; + priority?: number; + deactivateExisting?: boolean; + /** Provider identity to adopt with the move; omitted leaves it untouched. */ + providerName?: string; + /** Origin node to stamp when the agent has none; omitted leaves it untouched. */ + originNodeId?: string; +} + +/** + * Statements that bind `agent` to `nodeId` and move its routable location onto + * that node. Built, not executed: a caller that owns the whole move hands them + * to {@link runAtomicWrites} so the binding row, the location move and the + * retirement of the bindings left behind commit as one unit — on D1 too, where + * there is no interactive transaction to roll a half-applied move back. + * + * Statement order is the last-resort story for a caller that still runs them + * one at a time inside its own transaction: every prefix has to leave the agent + * routable, so the new binding is inserted first (inert on its own — delivery + * joins bindings only where `agents.location_node_id` matches), then the agent + * moves onto it, then the bindings it left behind are retired (inert once the + * agent has moved off them). Deactivating first, or moving the agent before its + * binding exists, would strand the agent with no active binding at its + * location. Callers that mutate `agents` themselves must write through + * `providerName`/`originNodeId` here rather than updating the row up front, + * for the same reason. + */ +function agentNodeBindingWrites( db: Db, workspaceId: string, agent: Pick, nodeId: string, - opts: { sessionRef?: string | null; priority?: number; deactivateExisting?: boolean } = {}, -) { + opts: AgentNodeBindingOpts = {}, +): AtomicWrite[] { + const now = new Date(); + const writes: AtomicWrite[] = [ + db + .insert(agentNodeBindings) + .values({ + id: `anb_${generateId()}`, + workspaceId, + agentId: agent.id, + nodeId, + status: 'active', + sessionRef: opts.sessionRef ?? null, + priority: opts.priority ?? 0, + updatedAt: now, + }) + .onConflictDoUpdate({ + target: [agentNodeBindings.agentId, agentNodeBindings.nodeId], + set: { + status: 'active', + sessionRef: opts.sessionRef ?? null, + priority: opts.priority ?? 0, + updatedAt: now, + }, + }), + db + .update(agents) + .set({ + locationType: 'via_node', + locationNodeId: nodeId, + sessionRef: opts.sessionRef ?? undefined, + providerName: opts.providerName ?? undefined, + originNodeId: opts.originNodeId ?? undefined, + status: 'active', + lastSeen: now, + }) + .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agent.id))), + ]; + if (opts.deactivateExisting ?? true) { - await db + writes.push(db .update(agentNodeBindings) - .set({ status: 'inactive', updatedAt: new Date() }) + .set({ status: 'inactive', updatedAt: now }) .where(and( eq(agentNodeBindings.workspaceId, workspaceId), eq(agentNodeBindings.agentId, agent.id), eq(agentNodeBindings.status, 'active'), ne(agentNodeBindings.nodeId, nodeId), - )); + ))); } + return writes; +} - await db - .insert(agentNodeBindings) - .values({ - id: `anb_${generateId()}`, - workspaceId, - agentId: agent.id, - nodeId, - status: 'active', - sessionRef: opts.sessionRef ?? null, - priority: opts.priority ?? 0, - updatedAt: new Date(), - }) - .onConflictDoUpdate({ - target: [agentNodeBindings.agentId, agentNodeBindings.nodeId], - set: { - status: 'active', - sessionRef: opts.sessionRef ?? null, - priority: opts.priority ?? 0, - updatedAt: new Date(), - }, - }); - - await db - .update(agents) - .set({ - locationType: 'via_node', - locationNodeId: nodeId, - sessionRef: opts.sessionRef ?? undefined, - status: 'active', - lastSeen: new Date(), - }) - .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agent.id))); +/** Run {@link agentNodeBindingWrites} one statement at a time. */ +async function upsertAgentNodeBinding( + db: Db, + workspaceId: string, + agent: Pick, + nodeId: string, + opts: AgentNodeBindingOpts = {}, +) { + for (const write of agentNodeBindingWrites(db, workspaceId, agent, nodeId, opts)) await write; } async function activeBindingNodeIdsForAgent(db: Db, workspaceId: string, agentId: string): Promise { @@ -1113,10 +1182,18 @@ async function rejectMigrationCanceledSpawnRegistration( } } -async function releaseNodeAgentSlots(db: Db, workspaceId: string, nodeIds: string[]): Promise { +/** + * The statement (at most one) that gives back one agent slot on each of + * `nodeIds`. Built, not executed, so a caller can commit the decrement in the + * same atomic unit as the binding retirement that justifies it: a decrement + * that lands without the retirement lets the node overshoot `maxAgents`, and a + * retirement that lands without the decrement charges capacity to a binding + * nothing will ever release. + */ +function releaseNodeAgentSlotWrites(db: Db, workspaceId: string, nodeIds: string[]): AtomicWrite[] { const uniqueNodeIds = [...new Set(nodeIds)]; - if (uniqueNodeIds.length === 0) return; - await db + if (uniqueNodeIds.length === 0) return []; + return [db .update(nodes) .set({ activeAgents: sql`CASE WHEN ${nodes.activeAgents} > 0 THEN ${nodes.activeAgents} - 1 ELSE 0 END`, @@ -1124,7 +1201,11 @@ async function releaseNodeAgentSlots(db: Db, workspaceId: string, nodeIds: strin .where(and( eq(nodes.workspaceId, workspaceId), inArray(nodes.id, uniqueNodeIds), - )); + ))]; +} + +async function releaseNodeAgentSlots(db: Db, workspaceId: string, nodeIds: string[]): Promise { + for (const write of releaseNodeAgentSlotWrites(db, workspaceId, nodeIds)) await write; } export async function ensureDirectNodeForAgent( @@ -1332,6 +1413,15 @@ export async function bindAgentToNode( agentName: string, opts: { session_ref?: string | null; priority?: number } = {}, ) { + // `runAtomic` below already owns a transaction on an adapter that has one, so + // the move's statements must not re-enter `withTransaction` (the shared + // connection serializes transactions and would deadlock on itself) — they run + // in order inside it instead. Without one, the move has to be a D1 batch, and + // a handle offering neither is refused before a single statement lands rather + // than committing half a move it cannot undo. + const hasInteractiveTransaction = typeof (db as EngineDb & { + withTransaction?: unknown; + }).withTransaction === 'function'; return runAtomic(db, async (tx) => { const node = await getNodeByName(tx, workspaceId, nodeName); if (!node) throw codedError(`Node "${nodeName}" not found`, 'node_not_found', 404); @@ -1342,6 +1432,40 @@ export async function bindAgentToNode( .where(and(eq(agents.workspaceId, workspaceId), eq(agents.name, agentName))); if (!agent) throw codedError(`Agent "${agentName}" not found`, 'agent_not_found', 404); + // A bind moves the agent's routable location onto this node. Delivery + // routing joins bindings only where `agents.location_node_id` matches the + // bound node, so a binding row alone leaves an HTTP-registered agent + // stranded on its implicit `direct-*` pseudo-node and never woken — the + // exact shape of the broker's create-only register falling back to this + // endpoint. `unbindAgentFromNode` already assumes this move happened. + if ( + agent.status === 'active' + && agent.locationNodeId + && agent.locationNodeId !== node.id + && !isImplicitDirectLocation(agent) + ) { + const [locatedNode] = await tx + .select() + .from(nodes) + .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, agent.locationNodeId))); + // A dead or missing prior location is not a conflict — it is exactly what + // a bind is for. Only a live one still owns the identity. + if (locatedNode && isNodeLive(locatedNode)) { + throw codedError( + `Agent "${agentName}" is already active on another live location`, + 'agent_location_conflict', + 409, + ); + } + } + + const nodeProviderNames = (await tx + .select({ name: nodeProviders.name }) + .from(nodeProviders) + .where(and(eq(nodeProviders.workspaceId, workspaceId), eq(nodeProviders.nodeId, node.id)))) + .map((row) => row.name); + const adoptedProviderName = adoptNodeProviderName(agent.providerName, nodeProviderNames); + const activeNodeIds = await activeBindingNodeIdsForAgent(tx, workspaceId, agent.id); const targetWasActive = activeNodeIds.includes(node.id); let reservedTargetSlot = false; @@ -1350,12 +1474,34 @@ export async function bindAgentToNode( await reserveNodeAgentSlot(tx, workspaceId, node); reservedTargetSlot = true; } - await upsertAgentNodeBinding(tx, workspaceId, agent, node.id, { - sessionRef: opts.session_ref ?? null, - priority: opts.priority ?? 0, - }); - await releaseNodeAgentSlots(tx, workspaceId, activeNodeIds.filter((nodeId) => nodeId !== node.id)); + // The binding row, the location move, the provider adoption, the + // retirement of the bindings left behind and the slots those bindings + // held are one mutation. Committing any subset is a corruption that no + // retry repairs: a binding without its reservation lets the node exceed + // `maxAgents`, and a retired binding whose slot was never given back + // charges the old node forever, because the retry that would fix either + // reads the half-applied state as already done. + const buildMoveWrites = (writeDb: Db): AtomicWrite[] => [ + ...agentNodeBindingWrites(writeDb, workspaceId, agent, node.id, { + sessionRef: opts.session_ref ?? null, + priority: opts.priority ?? 0, + providerName: adoptedProviderName, + originNodeId: agent.originNodeId ?? node.id, + }), + ...releaseNodeAgentSlotWrites( + writeDb, + workspaceId, + activeNodeIds.filter((nodeId) => nodeId !== node.id), + ), + ]; + if (hasInteractiveTransaction) { + for (const write of buildMoveWrites(tx)) await write; + } else { + await runAtomicWrites(tx, buildMoveWrites, { requireAtomic: true }); + } } catch (err) { + // Every statement of the move either landed or did not, so a throw means + // none of it did and the slot reserved for it belongs to nobody. if (reservedTargetSlot) { await releaseNodeAgentSlots(tx, workspaceId, [node.id]); } From de75a31e855cfda755523eac80f69ac632136629 Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Fri, 18 Sep 2026 19:31:05 -0700 Subject: [PATCH 2/3] fix(nodes): mark bound agents delivery-ready and adopt only live providers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps in the bind path left adopted agents unable to receive pushes: - Cursor-aware (agent_scoped) providers gate deliveries per identity until it is announced. The bind moved the agent onto the provider but never marked it ready, so deliveries queued forever. The bind now marks the adopted identity delivery-ready after the move commits and drains its pending queue, and returns delivery_ack_seq so the caller learns the authoritative cursor — the same contract agent.register/recover carry. - Provider adoption read every persisted provider row including offline ones kept for the capability manifest. An agent whose own provider had gone stale could be kept on (or moved to) a provider with no live socket. Adoption now considers only live providers. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- CHANGELOG.md | 2 +- openapi.yaml | 6 ++ .../conformance/agentLifecycle.test.ts | 82 ++++++++++++++++++- packages/engine/src/engine/node.ts | 68 +++++++++++++-- packages/engine/src/routes/node.ts | 1 + 5 files changed, 150 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 60713864..19de88d9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,7 @@ Packages without a separate changelog are covered by the cross-package notes bel ### Fixed -- Binding an agent to a node now adopts that node's provider and guards against stealing an agent that is active on another live node. A broker-spawned agent that was HTTP-registered first (provider `default`) and then bound through the node-agents fallback kept a provider its node did not serve, so its deliveries were routed to the node but never pushed — the spawned agent was never woken. The bind's binding row, location move and node capacity counters now commit as one atomic unit, so a failure part-way through can no longer leave a node charged for a binding it retired, or holding an active binding it never reserved capacity for. +- Binding an agent to a node now adopts one of that node's live providers, marks the agent delivery-ready on cursor-aware providers, and guards against stealing an agent that is active on another live node. A broker-spawned agent that was HTTP-registered first (provider `default`) and then bound through the node-agents fallback kept a provider its node did not serve, so its deliveries were routed to the node but never pushed — the spawned agent was never woken. The bind's binding row, location move and node capacity counters now commit as one atomic unit, so a failure part-way through can no longer leave a node charged for a binding it retired, or holding an active binding it never reserved capacity for. The bind response also returns the agent's `delivery_ack_seq` so cursor-aware providers learn the adopted identity's authoritative delivery cursor. ## [8.11.2] - 2026-09-19 diff --git a/openapi.yaml b/openapi.yaml index 44e9e15b..710694d2 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1307,6 +1307,12 @@ components: nullable: true priority: type: integer + delivery_ack_seq: + type: integer + description: > + The agent's authoritative delivery cursor. Returned only by the bind + endpoint so cursor-aware providers learn the identity's cursor state + when the binding marks it delivery-ready. created_at: type: string format: date-time diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index c2562615..250a79cc 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -1848,7 +1848,13 @@ describe('node agent binding adopts the agent location', () => { } /** Attach a node-control socket and register `providerName` on the node. */ - async function attachProvider(workspaceId: string, nodeId: string, nodeName: string, providerName: string) { + async function attachProvider( + workspaceId: string, + nodeId: string, + nodeName: string, + providerName: string, + capabilities: Array<{ name: string; kind?: string }> = [{ name: 'spawn:claude', kind: 'capacity' }], + ) { const sock = new FakeSocket(); const handle = stack.runtime.realtime.attachNodeSocket(workspaceId, nodeId, sock); await handle.handleMessage(JSON.stringify({ @@ -1858,7 +1864,7 @@ describe('node agent binding adopts the agent location', () => { node_id: nodeId, name: nodeName, provider: { name: providerName, instance_id: `${providerName}-i1` }, - capabilities: [{ name: 'spawn:claude', kind: 'capacity' }], + capabilities, max_agents: 4, tags: ['test'], version: 'v1', @@ -2007,6 +2013,78 @@ describe('node agent binding adopts the agent location', () => { expect(adopted).toEqual({ providerName: 'broker', locationNodeId: 'node_broker' }); }); + it('adopts the live provider when the agent\u2019s own provider row is offline', async () => { + const ws = await createWorkspace(stack.app, 'bind-live-provider-only'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'stale-provider-agent'); + await enrollNode(ws.workspaceKey, 'node_broker', 'broker-host'); + + // The node's `default` provider disconnected: its row persists as history + // while `broker` keeps the node serving. Adoption must see only the live + // side — keeping `default` would point the agent at a socket that is gone. + const stale = await attachProvider(ws.workspaceId, 'node_broker', 'broker-host', 'default'); + await stale.handle.handleClose(); + const [offline] = await stack.runtime.deps.db + .select({ status: nodeProviders.status, handlersLive: nodeProviders.handlersLive }) + .from(nodeProviders) + .where(and( + eq(nodeProviders.workspaceId, ws.workspaceId), + eq(nodeProviders.nodeId, 'node_broker'), + eq(nodeProviders.name, 'default'), + )); + expect(offline).toEqual({ status: 'offline', handlersLive: false }); + await attachProvider(ws.workspaceId, 'node_broker', 'broker-host', 'broker'); + + expect((await bindAgent(ws.workspaceKey, 'broker-host', target.name)).status).toBe(201); + + const [adopted] = await stack.runtime.deps.db + .select({ providerName: agents.providerName }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(adopted).toEqual({ providerName: 'broker' }); + }); + + it('marks a bound agent delivery-ready on a cursor-aware provider and drains its queue', async () => { + const ws = await createWorkspace(stack.app, 'bind-cursor-readiness'); + const speaker = await registerAgent(stack.app, ws.workspaceKey, 'speaker'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'cursor-bound-agent'); + await enrollNode(ws.workspaceKey, 'node_broker', 'broker-host'); + const { sock } = await attachProvider(ws.workspaceId, 'node_broker', 'broker-host', 'broker', [ + { name: 'spawn:claude', kind: 'capacity' }, + { name: 'relay:delivery-cursor-v1', kind: 'capacity' }, + ]); + + // A message lands while the agent still sits on its implicit pseudo-node: + // the durable delivery is queued with no socket to push through. + const posted = await stack.app.request('/v1/channels/general/messages', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${speaker.token}` }, + body: JSON.stringify({ text: 'queued while unroutable' }), + }); + expect(posted.status).toBe(201); + await stack.settle(); + expect(deliverFramesOfType(sock, 'message.created')).toEqual([]); + expect( + stack.runtime.realtime.isProviderAgentDeliveryReady(ws.workspaceId, 'node_broker', 'broker', target.agentId), + ).toBe(false); + + const bound = await bindAgent(ws.workspaceKey, 'broker-host', target.name); + expect(bound.status).toBe(201); + // The bind response hands the caller the agent's authoritative cursor, the + // same contract the agent.register/agent.recover replies carry. + expect((await bound.json() as { data: { delivery_ack_seq: number } }).data.delivery_ack_seq) + .toBe(0); + + // Agent-scoped readiness now names the adopted identity, and the bind's own + // drain replays the delivery that queued before the move. + expect( + stack.runtime.realtime.isProviderAgentDeliveryReady(ws.workspaceId, 'node_broker', 'broker', target.agentId), + ).toBe(true); + await stack.settle(); + expect(deliverFramesOfType(sock, 'message.created')).toEqual([ + expect.objectContaining({ type: 'deliver', agent: target.name }), + ]); + }); + it('stamps session, liveness and origin with the move, and keeps the first origin on a later one', async () => { const ws = await createWorkspace(stack.app, 'bind-stamps-identity'); const target = await registerAgent(stack.app, ws.workspaceKey, 'stamped-agent'); diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 541f7f74..823e2b11 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -116,8 +116,10 @@ function isImplicitDirectLocation(agent: Pick * 4. Otherwise (several named providers, none of them the agent's) nothing * identifies the owner, so the existing provider stands rather than guessing. * - * A node with no registered providers at all falls through to (4): no socket - * exists to adopt, and rewriting to a name nobody serves would not help. + * A node with no live providers at all falls through to (4): no socket exists + * to adopt, and rewriting to a name nobody serves would not help. Callers pass + * only live provider names — a persisted-but-offline row is a historical + * manifest entry, not a connection deliveries can reach. */ function adoptNodeProviderName(current: string, nodeProviderNames: string[]): string { if (nodeProviderNames.includes(current)) return current; @@ -1387,6 +1389,7 @@ function serializeBinding(row: { status: string; sessionRef: string | null; priority: number; + deliveryAckSeq?: number | null; createdAt: Date; updatedAt: Date | null; }) { @@ -1401,6 +1404,12 @@ function serializeBinding(row: { status: row.status, session_ref: row.sessionRef, priority: row.priority, + // Only the bind response selects this: it hands the caller the agent's + // authoritative delivery cursor, the same contract `agent.register` and + // `agent.recover` replies carry for cursor-aware providers. + ...(row.deliveryAckSeq !== undefined && row.deliveryAckSeq !== null + ? { delivery_ack_seq: row.deliveryAckSeq } + : {}), created_at: row.createdAt.toISOString(), updated_at: row.updatedAt?.toISOString() ?? null, }; @@ -1412,6 +1421,7 @@ export async function bindAgentToNode( nodeName: string, agentName: string, opts: { session_ref?: string | null; priority?: number } = {}, + deps: { nodeConnections?: NodeConnectionRegistry } = {}, ) { // `runAtomic` below already owns a transaction on an adapter that has one, so // the move's statements must not re-enter `withTransaction` (the shared @@ -1422,7 +1432,11 @@ export async function bindAgentToNode( const hasInteractiveTransaction = typeof (db as EngineDb & { withTransaction?: unknown; }).withTransaction === 'function'; - return runAtomic(db, async (tx) => { + // The post-commit registry work needs the resolved ids; they only become + // authoritative once `runAtomic` has returned, so they are read off the + // pending value afterwards rather than trusted mid-transaction. + let pendingMove: { nodeId: string; agentId: string; providerName: string } | undefined; + const binding = await runAtomic(db, async (tx) => { const node = await getNodeByName(tx, workspaceId, nodeName); if (!node) throw codedError(`Node "${nodeName}" not found`, 'node_not_found', 404); @@ -1459,12 +1473,23 @@ export async function bindAgentToNode( } } - const nodeProviderNames = (await tx - .select({ name: nodeProviders.name }) + // Provider rows persist after a disconnect so the node keeps its capability + // manifest; they are history, not sockets. Adopt only among providers that + // can actually receive a push right now — keeping or choosing an offline + // row while a live sibling serves the node would point the agent at a + // provider connection that does not exist. + const liveProviderNames = (await tx + .select({ + name: nodeProviders.name, + status: nodeProviders.status, + handlersLive: nodeProviders.handlersLive, + lastHeartbeatAt: nodeProviders.lastHeartbeatAt, + }) .from(nodeProviders) .where(and(eq(nodeProviders.workspaceId, workspaceId), eq(nodeProviders.nodeId, node.id)))) + .filter((provider) => isProviderLive(provider)) .map((row) => row.name); - const adoptedProviderName = adoptNodeProviderName(agent.providerName, nodeProviderNames); + const adoptedProviderName = adoptNodeProviderName(agent.providerName, liveProviderNames); const activeNodeIds = await activeBindingNodeIdsForAgent(tx, workspaceId, agent.id); const targetWasActive = activeNodeIds.includes(node.id); @@ -1520,6 +1545,7 @@ export async function bindAgentToNode( status: agentNodeBindings.status, sessionRef: agentNodeBindings.sessionRef, priority: agentNodeBindings.priority, + deliveryAckSeq: agents.deliveryAckSeq, createdAt: agentNodeBindings.createdAt, updatedAt: agentNodeBindings.updatedAt, }) @@ -1532,8 +1558,38 @@ export async function bindAgentToNode( eq(agentNodeBindings.nodeId, node.id), )); + pendingMove = { nodeId: node.id, agentId: agent.id, providerName: adoptedProviderName }; return serializeBinding(binding); }); + // Cursor-aware providers gate every delivery on a per-identity ready mark; + // `agent.register`/`agent.recover` grant it after replying with the agent's + // authoritative cursor, and the response above carries `delivery_ack_seq` for + // the same reason here. A bound agent needs that transition too or it points + // at a socket that will never push to it. This runs strictly after the move + // commits — marking a rolled-back agent ready would deliver to a location + // that does not own it — and only mutates an agent_scoped ready-set; + // immediate-mode providers ignore it. + const registry = deps.nodeConnections; + if (pendingMove && registry) { + registry.markProviderAgentsDeliveryReady?.( + workspaceId, + pendingMove.nodeId, + pendingMove.providerName, + undefined, + [pendingMove.agentId], + ); + try { + await deliverPendingToNode(db, registry, workspaceId, pendingMove.nodeId, { + providerName: pendingMove.providerName, + agentIds: [pendingMove.agentId], + }); + } catch (err) { + // The move is durable; a replay failure leaves the pending queue intact + // for the next drain rather than reporting a bind failure that committed. + console.error('[node.bind] pending delivery replay failed', err); + } + } + return binding; } export async function unbindAgentFromNode(db: Db, workspaceId: string, nodeName: string, agentName: string) { diff --git a/packages/engine/src/routes/node.ts b/packages/engine/src/routes/node.ts index f6174a54..b1bd5c79 100644 --- a/packages/engine/src/routes/node.ts +++ b/packages/engine/src/routes/node.ts @@ -265,6 +265,7 @@ nodeRoutes.post('/nodes/:name/agents', requireWorkspaceKey, rateLimit, async (c) session_ref: parsed.data.session_ref, priority: parsed.data.priority, }, + { nodeConnections: c.get('engine').nodeConnections }, ); return jsonCreated(c, result); } catch (err: unknown) { From c9c7f706fe36c911d66f1480b3a8be4fef8acff8 Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Fri, 18 Sep 2026 19:46:15 -0700 Subject: [PATCH 3/3] fix(nodes): order bind readiness+replay after the cursor-bearing response MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cursor Bugbot: marking the adopted agent ready and draining its queue inside bindAgentToNode let deliver frames reach the provider socket before the HTTP response carrying delivery_ack_seq reached the caller. agent.register and agent.recover order their cursor-bearing reply ahead of the ready mark and replay; the bind path now does the same — bindAgentToNode returns the binding plus the resolved move, the route sends the response first, then runs the ready mark and scoped replay in the request background. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../conformance/agentLifecycle.test.ts | 5 +- packages/engine/src/engine/node.ts | 74 +++++++++---------- packages/engine/src/routes/node.ts | 21 +++++- 3 files changed, 57 insertions(+), 43 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index 250a79cc..774e34fb 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -2075,11 +2075,12 @@ describe('node agent binding adopts the agent location', () => { .toBe(0); // Agent-scoped readiness now names the adopted identity, and the bind's own - // drain replays the delivery that queued before the move. + // drain replays the delivery that queued before the move. Both run in the + // request background, strictly after the cursor-bearing response. + await stack.settle(); expect( stack.runtime.realtime.isProviderAgentDeliveryReady(ws.workspaceId, 'node_broker', 'broker', target.agentId), ).toBe(true); - await stack.settle(); expect(deliverFramesOfType(sock, 'message.created')).toEqual([ expect.objectContaining({ type: 'deliver', agent: target.name }), ]); diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 823e2b11..003c90ad 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -1421,8 +1421,10 @@ export async function bindAgentToNode( nodeName: string, agentName: string, opts: { session_ref?: string | null; priority?: number } = {}, - deps: { nodeConnections?: NodeConnectionRegistry } = {}, -) { +): Promise<{ + binding: ReturnType; + move: { nodeId: string; agentId: string; providerName: string }; +}> { // `runAtomic` below already owns a transaction on an adapter that has one, so // the move's statements must not re-enter `withTransaction` (the shared // connection serializes transactions and would deadlock on itself) — they run @@ -1432,11 +1434,7 @@ export async function bindAgentToNode( const hasInteractiveTransaction = typeof (db as EngineDb & { withTransaction?: unknown; }).withTransaction === 'function'; - // The post-commit registry work needs the resolved ids; they only become - // authoritative once `runAtomic` has returned, so they are read off the - // pending value afterwards rather than trusted mid-transaction. - let pendingMove: { nodeId: string; agentId: string; providerName: string } | undefined; - const binding = await runAtomic(db, async (tx) => { + return runAtomic(db, async (tx) => { const node = await getNodeByName(tx, workspaceId, nodeName); if (!node) throw codedError(`Node "${nodeName}" not found`, 'node_not_found', 404); @@ -1558,38 +1556,38 @@ export async function bindAgentToNode( eq(agentNodeBindings.nodeId, node.id), )); - pendingMove = { nodeId: node.id, agentId: agent.id, providerName: adoptedProviderName }; - return serializeBinding(binding); + return { + binding: serializeBinding(binding), + move: { nodeId: node.id, agentId: agent.id, providerName: adoptedProviderName }, + }; + }); +} + +/** + * The second half of a bind, run by the caller only after the cursor-bearing + * response is on the wire — `agent.register`/`agent.recover` order their + * cursor reply ahead of the ready mark and replay for the same reason. + * Cursor-aware providers gate every delivery on a per-identity ready mark, + * so a bound agent needs this transition or it points at a socket that will + * never push to it; immediate-mode providers treat the mark as a no-op. + */ +export async function completeBoundAgentDelivery( + db: Db, + registry: NodeConnectionRegistry, + workspaceId: string, + move: { nodeId: string; agentId: string; providerName: string }, +): Promise { + registry.markProviderAgentsDeliveryReady?.( + workspaceId, + move.nodeId, + move.providerName, + undefined, + [move.agentId], + ); + await deliverPendingToNode(db, registry, workspaceId, move.nodeId, { + providerName: move.providerName, + agentIds: [move.agentId], }); - // Cursor-aware providers gate every delivery on a per-identity ready mark; - // `agent.register`/`agent.recover` grant it after replying with the agent's - // authoritative cursor, and the response above carries `delivery_ack_seq` for - // the same reason here. A bound agent needs that transition too or it points - // at a socket that will never push to it. This runs strictly after the move - // commits — marking a rolled-back agent ready would deliver to a location - // that does not own it — and only mutates an agent_scoped ready-set; - // immediate-mode providers ignore it. - const registry = deps.nodeConnections; - if (pendingMove && registry) { - registry.markProviderAgentsDeliveryReady?.( - workspaceId, - pendingMove.nodeId, - pendingMove.providerName, - undefined, - [pendingMove.agentId], - ); - try { - await deliverPendingToNode(db, registry, workspaceId, pendingMove.nodeId, { - providerName: pendingMove.providerName, - agentIds: [pendingMove.agentId], - }); - } catch (err) { - // The move is durable; a replay failure leaves the pending queue intact - // for the next drain rather than reporting a bind failure that committed. - console.error('[node.bind] pending delivery replay failed', err); - } - } - return binding; } export async function unbindAgentFromNode(db: Db, workspaceId: string, nodeName: string, agentName: string) { diff --git a/packages/engine/src/routes/node.ts b/packages/engine/src/routes/node.ts index b1bd5c79..8332ab3c 100644 --- a/packages/engine/src/routes/node.ts +++ b/packages/engine/src/routes/node.ts @@ -256,7 +256,7 @@ nodeRoutes.post('/nodes/:name/agents', requireWorkspaceKey, rateLimit, async (c) if (!parsed.ok) { return parsed.response; } - const result = await nodeEngine.bindAgentToNode( + const { binding, move } = await nodeEngine.bindAgentToNode( c.get('db'), c.get('workspace').id, c.req.param('name'), @@ -265,9 +265,24 @@ nodeRoutes.post('/nodes/:name/agents', requireWorkspaceKey, rateLimit, async (c) session_ref: parsed.data.session_ref, priority: parsed.data.priority, }, - { nodeConnections: c.get('engine').nodeConnections }, ); - return jsonCreated(c, result); + const response = jsonCreated(c, binding); + // The response — which carries delivery_ack_seq — must be on the wire + // before any deliver frame reaches the provider socket: agent.register and + // agent.recover order their cursor-bearing reply ahead of the ready mark + // and replay, and a bound agent's drain honors the same ordering. + runInBackground( + c, + Promise.resolve().then(() => + nodeEngine.completeBoundAgentDelivery( + c.get('db'), + c.get('engine').nodeConnections, + c.get('workspace').id, + move, + )), + 'bind delivery readiness+replay', + ); + return response; } catch (err: unknown) { return errorResponse(c, err); }