From c280a96975beddec1591ec9af2a6e9715551bd0f Mon Sep 17 00:00:00 2001 From: Barry Cape Date: Thu, 6 Aug 2026 05:32:58 -0400 Subject: [PATCH 1/8] fix(engine): restore agent presence lifecycle --- README.md | 13 ++ openapi.yaml | 36 ++++- .../conformance/agentLifecycle.test.ts | 139 ++++++++++++++++++ packages/engine/src/adapters/node/index.ts | 2 + packages/engine/src/engine/action.ts | 139 ++++++++++++++---- packages/engine/src/engine/agent.ts | 60 +++++--- packages/engine/src/index.ts | 1 + packages/engine/src/routes/agent.ts | 5 +- 8 files changed, 336 insertions(+), 59 deletions(-) create mode 100644 packages/engine/src/__tests__/conformance/agentLifecycle.test.ts diff --git a/README.md b/README.md index 7850cf30..12067df0 100644 --- a/README.md +++ b/README.md @@ -461,6 +461,17 @@ caller also receives it directly. `node.status.online` / `node.status.offline` durably record node liveness transitions (offline carries a `reason` such as `liveness_timeout` | `disconnected` | `deregistered`). +Agent roster presence is lease-based, not a write-once registration flag. +`active` means Relaycast observed authenticated agent activity within the last +five minutes (`last_seen`); an older persisted `active` (or legacy `online`) +record is reported and durably swept as `offline`. Registration or subsequent +authenticated activity renews the lease. Releasing an agent dispatches to its +live host when one exists. If the host is absent or offline, a normal release +fails explicitly with `agent_host_unavailable` instead of creating an ownerless +pending invocation. A `delete_agent` request can be completed locally in that +case: Relaycast deactivates bindings and deletes the record and its implicit +direct node. + Fleet node presence is also published to workspace-key observer streams as the ephemeral `node.online`, `node.heartbeat`, and `node.offline` events. Each carries a `node` payload matching the `GET /nodes` roster entry (capabilities, @@ -567,6 +578,8 @@ immediate delivery and receive a rejected capability result. Queue/cron-backed adapters that own node dispatch outside the Node adapter should call `drainNodeInvocations` after node reconnect/register/heartbeat and `sweepTimedOutInvocations` from cron via `@relaycast/engine/node-invocations`. +They should also call `sweepStaleAgents` from `@relaycast/engine` to persist +agent lease expiry; roster reads derive the same status even before that sweep. Actions are async fire-and-forget: invoking an action returns an ack with `invocation_id` and dispatches an `action.invoke` frame to the handler's node. Agent diff --git a/openapi.yaml b/openapi.yaml index 072f15d2..12302abe 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -146,6 +146,11 @@ components: status: type: string enum: [active, idle, blocked, waiting, offline, online] + description: >- + Presence-aware lifecycle status. `active` means the engine observed + authenticated activity within the five-minute agent liveness TTL; + a persisted active/legacy-online row older than that is reported and + durably swept as `offline`. token: type: string description: Agent token (only returned on registration) @@ -226,7 +231,10 @@ components: additionalProperties: true status: type: string - description: Invocation dispatch state, typically `pending` or `dispatched`. + description: >- + Invocation lifecycle state. `pending` or `dispatched` means a live + host owns the request; `completed` can be returned immediately when + release reaps an agent that has no live host. created_at: type: string format: date-time @@ -1695,7 +1703,11 @@ paths: get: summary: List agents - description: List all agents in the workspace. Observer tokens require `agents:read`. + description: >- + List all agents in the workspace. Active presence is derived from + server-observed `last_seen` using a five-minute TTL, so stale persisted + active rows are returned as offline and swept durably. Observer tokens + require `agents:read`. tags: - Agents security: @@ -1706,7 +1718,7 @@ paths: in: query schema: type: string - enum: [online, offline] + enum: [active, idle, blocked, waiting, offline, online, all] responses: '200': description: List of agents @@ -1875,9 +1887,13 @@ paths: post: summary: Request agent release description: | - Request the node that owns an agent to release it by dispatching the - built-in `release` action. The engine marks or deletes the agent only - after the node reports successful completion. + If the agent has a live host, request release by dispatching the built-in + `release` action and apply the lifecycle change after the host confirms. + If no host is bound or the hosting connection is not live, a normal + release fails explicitly with `503 agent_host_unavailable`; it never + creates an ownerless pending invocation. With `delete_agent`, the engine + can reap the database record directly and returns a completed invocation, + deleting the agent and any implicit direct node. tags: - Agents security: @@ -1904,7 +1920,7 @@ paths: description: If true, permanently delete the agent responses: '201': - description: Release action invoked + description: Release dispatched to a live host or completed locally content: application/json: schema: @@ -1920,6 +1936,12 @@ paths: application/json: schema: $ref: '#/components/schemas/ErrorResponse' + '503': + description: Agent has no live host to receive a normal release + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' /channels: post: diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts new file mode 100644 index 00000000..8bd5135a --- /dev/null +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -0,0 +1,139 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { and, eq } from 'drizzle-orm'; +import { actionInvocations, agentNodeBindings, agents, nodes } from '../../db/schema.js'; +import { AGENT_LIVENESS_TTL_MS } from '../../engine/agent.js'; +import { + attachDirectNodeSocket, + createWorkspace, + makeNodeStack, + registerAgent, + type TestStack, +} from './harness.js'; + +describe('agent presence and release lifecycle', () => { + let stack: TestStack; + + beforeEach(() => { stack = makeNodeStack(); }); + afterEach(() => stack.close()); + + it('derives presence from last_seen and persists stale active agents offline', async () => { + const ws = await createWorkspace(stack.app, 'agent-presence-expiry'); + const stale = await registerAgent(stack.app, ws.workspaceKey, 'stale-agent'); + await stack.runtime.deps.db + .update(agents) + .set({ + status: 'active', + lastSeen: new Date(Date.now() - AGENT_LIVENESS_TTL_MS - 1_000), + }) + .where(eq(agents.id, stale.agentId)); + + const response = await stack.app.request('/v1/agents?status=active', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + expect(response.status).toBe(200); + const body = await response.json() as { data: Array<{ name: string }> }; + expect(body.data.map((agent) => agent.name)).not.toContain('stale-agent'); + + const [persisted] = await stack.runtime.deps.db + .select({ status: agents.status }) + .from(agents) + .where(eq(agents.id, stale.agentId)); + expect(persisted.status).toBe('offline'); + }); + + it('fails release explicitly when the agent has no live host', async () => { + const ws = await createWorkspace(stack.app, 'hostless-agent-release'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'hostless-agent'); + const nodeId = `node_direct_${target.agentId}`; + + const response = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: target.name, reason: 'stale cleanup' }), + }); + expect(response.status).toBe(503); + const body = await response.json() as { error: { code: string; message: string } }; + expect(body.error).toEqual({ + code: 'agent_host_unavailable', + message: 'Agent "hostless-agent" has no live host node; cannot dispatch release', + }); + + const [agent] = await stack.runtime.deps.db + .select({ status: agents.status, locationNodeId: agents.locationNodeId }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(agent).toMatchObject({ status: 'active', locationNodeId: nodeId }); + + const [binding] = await stack.runtime.deps.db + .select({ status: agentNodeBindings.status }) + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, ws.workspaceId), + eq(agentNodeBindings.agentId, target.agentId), + eq(agentNodeBindings.nodeId, nodeId), + )); + expect(binding.status).toBe('active'); + + const [invocation] = await stack.runtime.deps.db + .select({ status: actionInvocations.status, error: actionInvocations.error }) + .from(actionInvocations) + .where(and( + eq(actionInvocations.workspaceId, ws.workspaceId), + eq(actionInvocations.actionName, 'release'), + )); + expect(invocation).toMatchObject({ + status: 'failed', + error: 'agent_host_unavailable', + }); + }); + + it('deletes a hostless agent and its implicit direct node', async () => { + const ws = await createWorkspace(stack.app, 'hostless-agent-delete'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'delete-me'); + const nodeId = `node_direct_${target.agentId}`; + // Reproduce a legacy/orphaned roster row with no dispatchable location. + await stack.runtime.deps.db + .update(agents) + .set({ locationType: 'self_connected', locationNodeId: null }) + .where(eq(agents.id, target.agentId)); + + const response = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: target.name, delete_agent: true }), + }); + expect(response.status).toBe(201); + expect((await response.json() as { data: { status: string } }).data.status).toBe('completed'); + + expect(await stack.runtime.deps.db.select().from(agents).where(eq(agents.id, target.agentId))).toHaveLength(0); + expect(await stack.runtime.deps.db.select().from(nodes).where(eq(nodes.id, nodeId))).toHaveLength(0); + }); + + it('continues dispatching release to a live host', async () => { + const ws = await createWorkspace(stack.app, 'live-agent-release'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'live-agent'); + const { sock, handle, nodeId } = await attachDirectNodeSocket(stack, ws.workspaceId, target); + + const response = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: target.name }), + }); + expect(response.status).toBe(201); + const body = await response.json() as { + data: { status: string; dispatched_node_id: string | null }; + }; + expect(body.data).toMatchObject({ status: 'dispatched', dispatched_node_id: nodeId }); + expect(sock.ofType('action.invoke').at(-1)).toMatchObject({ action: 'release' }); + await handle.handleClose(); + }); +}); diff --git a/packages/engine/src/adapters/node/index.ts b/packages/engine/src/adapters/node/index.ts index 2f2e8068..7050f93c 100644 --- a/packages/engine/src/adapters/node/index.ts +++ b/packages/engine/src/adapters/node/index.ts @@ -14,6 +14,7 @@ import { InProcessKeyValueStore } from './kv.js'; import { DurableEventQueue, InProcessEventQueue, type DurableEventQueueOptions } from './event-queue.js'; import { LocalFileStorage, createFileRouteHandler, FILE_ROUTE_PREFIX } from './files.js'; import { sweepOfflineNodes } from '../../engine/node.js'; +import { sweepStaleAgents } from '../../engine/agent.js'; import { sweepTimedOutInvocations } from '../../engine/action.js'; import { sendNodePresenceContext } from '../../engine/nodeContext.js'; import { createDeliveryMaintenanceRunner } from './delivery-maintenance.js'; @@ -164,6 +165,7 @@ export function createNodeRuntime(options: NodeRuntimeOptions): NodeRuntime { const runDeliveryMaintenance = createDeliveryMaintenanceRunner(deps); const sweepTimer = setInterval(() => { + void sweepStaleAgents(db).catch(() => {}); void sweepOfflineNodes(db, realtime, deps).catch(() => {}); void sweepTimedOutInvocations(db, realtime, { completionDeps: deps }).catch(() => {}); void runDeliveryMaintenance(); diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index f4649259..89704b38 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -662,6 +662,7 @@ async function dispatchNodeProviderInvocation(args: { async function dispatchRelease(args: { db: Db; registry?: NodeConnectionRegistry; + completionDeps?: InvocationCompletionDeps; workspaceId: string; data: { input?: Record; @@ -669,9 +670,6 @@ async function dispatchRelease(args: { caller_name?: string; }; }) { - if (!args.registry) { - throw codedError('Node dispatch is not available', 'node_dispatch_unavailable', 503); - } const input = recordInput(args.data.input); const name = typeof input.name === 'string' ? input.name : null; if (!name) { @@ -685,35 +683,106 @@ async function dispatchRelease(args: { if (!agent) { throw codedError(`Agent "${name}" not found`, 'agent_not_found', 404); } - if (agent.locationType !== 'via_node' || !agent.locationNodeId) { - throw codedError(`Agent "${name}" is not bound to a node`, 'agent_not_node_bound', 409); - } - const invocation = await createInvocation(args.db, args.workspaceId, null, { input, caller_id: args.data.caller_id, caller_name: args.data.caller_name, action_name: 'release', }); + const completeLocally = async () => { + await applyReleaseCompletionEffect( + args.db, + args.workspaceId, + agent.locationNodeId, + invocation, + {}, + args.completionDeps, + { allowMissingBinding: true, expectedAgentId: agent.id }, + ); + await args.db + .update(actionInvocations) + .set({ + status: 'completed', + output: { released: true, deleted: input.delete_agent === true, reaped_locally: true }, + completedAt: new Date(), + }) + .where(and( + eq(actionInvocations.workspaceId, args.workspaceId), + eq(actionInvocations.id, invocation.id), + inArray(actionInvocations.status, OPEN_INVOCATION_STATUSES), + )); + return { + invocation_id: invocation.id, + action_name: 'release', + handler_agent_id: null, + handler_node_id: agent.locationNodeId, + dispatched_node_id: null, + input, + status: 'completed', + created_at: invocation.createdAt.toISOString(), + }; + }; + const failClosed = async (): Promise => { + await args.db + .update(actionInvocations) + .set({ status: 'failed', error: 'agent_host_unavailable', completedAt: new Date() }) + .where(and( + eq(actionInvocations.workspaceId, args.workspaceId), + eq(actionInvocations.id, invocation.id), + inArray(actionInvocations.status, OPEN_INVOCATION_STATUSES), + )); + throw codedError( + `Agent "${name}" has no live host node; cannot dispatch release`, + 'agent_host_unavailable', + 503, + ); + }; + + const registry = args.registry; + const nodeId = agent.locationNodeId; + const hostLive = !!registry + && agent.locationType === 'via_node' + && !!nodeId + && await isHandlerConnectionLive( + args.db, + registry, + args.workspaceId, + nodeId, + agent.providerName, + ); + + if (!hostLive) { + return input.delete_agent === true ? completeLocally() : failClosed(); + } + + if (!registry || !nodeId) { + throw codedError(`Agent "${name}" has no live host node`, 'agent_host_unavailable', 503); + } // Release is a capacity operation handled by the provider hosting the agent. const dispatched = await dispatchNodeInvocation({ db: args.db, - registry: args.registry, + registry, workspaceId: args.workspaceId, invocationId: invocation.id, - nodeId: agent.locationNodeId, + nodeId, providerName: agent.providerName, action: 'release', input, }); + // The provider can disconnect between the liveness check and send. Complete + // the DB lifecycle locally instead of creating an ownerless pending request. + if (!dispatched.accepted) { + return input.delete_agent === true ? completeLocally() : failClosed(); + } + return { invocation_id: invocation.id, action_name: 'release', handler_agent_id: null, - handler_node_id: agent.locationNodeId, - dispatched_node_id: dispatched.accepted ? agent.locationNodeId : null, + handler_node_id: nodeId, + dispatched_node_id: dispatched.accepted ? nodeId : null, input: recordInput(invocation.input), status: dispatched.accepted ? (dispatched.pending ? 'pending' : 'dispatched') : 'pending', created_at: invocation.createdAt.toISOString(), @@ -917,6 +986,7 @@ export async function invokeAction( }, options: { nodeConnections?: NodeConnectionRegistry; + completionDeps?: InvocationCompletionDeps; /** Resolve plain node-scoped actions too (message triggers bind by name * without a node); the resolved row is dispatched node-addressed. */ includeNodeScoped?: boolean; @@ -937,6 +1007,7 @@ export async function invokeAction( return dispatchRelease({ db, registry: options.nodeConnections, + completionDeps: options.completionDeps, workspaceId, data, }); @@ -1111,16 +1182,17 @@ function publicInvocation(row: InvocationRow) { async function applyReleaseCompletionEffect( db: Db, workspaceId: string, - nodeId: string, + nodeId: string | null, invocation: Pick, data: { error?: string }, deps?: InvocationCompletionDeps, -): Promise { - if (!isReleaseInvocation(invocation.actionName) || data.error) return; + options: { allowMissingBinding?: boolean; expectedAgentId?: string } = {}, +): Promise { + if (!isReleaseInvocation(invocation.actionName) || data.error) return false; const input = recordInput(invocation.input); const name = typeof input.name === 'string' ? input.name : null; - if (!name) return; + if (!name) return false; const [agent] = await db .select() @@ -1128,39 +1200,47 @@ async function applyReleaseCompletionEffect( .where(and( eq(agents.workspaceId, workspaceId), eq(agents.name, name), - eq(agents.locationType, 'via_node'), - eq(agents.locationNodeId, nodeId), + ...(options.expectedAgentId ? [eq(agents.id, options.expectedAgentId)] : []), + ...(!options.allowMissingBinding && nodeId ? [ + eq(agents.locationType, 'via_node'), + eq(agents.locationNodeId, nodeId), + ] : []), )); - if (!agent) return; + if (!agent) return false; // Only proceed if an active binding actually flipped to inactive. This guards // against a second release (e.g. a retry) double-decrementing activeAgents for // an agent that was already released from this node. - const [deactivatedBinding] = await db + const deactivatedBindings = await db .update(agentNodeBindings) .set({ status: 'inactive', updatedAt: new Date() }) .where(and( eq(agentNodeBindings.workspaceId, workspaceId), - eq(agentNodeBindings.nodeId, nodeId), eq(agentNodeBindings.agentId, agent.id), eq(agentNodeBindings.status, 'active'), + ...(nodeId ? [eq(agentNodeBindings.nodeId, nodeId)] : []), )) - .returning({ id: agentNodeBindings.id }); - if (!deactivatedBinding) return; + .returning({ nodeId: agentNodeBindings.nodeId }); + if (deactivatedBindings.length === 0 && !options.allowMissingBinding) return false; // Capture exit correlation BEFORE the mutation deletes the row or strips the // spawn/cli metadata, so a durable agent.exited can still be emitted. const exited = { agentId: agent.id, agentName: agent.name, invocationId: fleetInvocationId(agent.metadata) }; - await db - .update(nodes) - .set({ - activeAgents: sql`CASE WHEN ${nodes.activeAgents} > 0 THEN ${nodes.activeAgents} - 1 ELSE 0 END`, - }) - .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, nodeId))); + const deactivatedNodeIds = Array.from(new Set(deactivatedBindings.map((binding) => binding.nodeId))); + if (deactivatedNodeIds.length > 0) { + await db + .update(nodes) + .set({ + activeAgents: sql`CASE WHEN ${nodes.activeAgents} > 0 THEN ${nodes.activeAgents} - 1 ELSE 0 END`, + }) + .where(and(eq(nodes.workspaceId, workspaceId), inArray(nodes.id, deactivatedNodeIds))); + } if (input.delete_agent === true) { await db.delete(agents).where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agent.id))); + const implicitNodeId = `node_direct_${agent.id}`; + await db.delete(nodes).where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, implicitNodeId))); } else { const existingMetadata = agent.metadata ?? {}; const { spawn: _spawn, cli: _cli, ...restMetadata } = existingMetadata; @@ -1184,7 +1264,7 @@ async function applyReleaseCompletionEffect( .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agent.id))); } - if (deps) { + if (deps && nodeId) { await emitAgentExitedEffects(deps, workspaceId, { agentId: exited.agentId, agentName: exited.agentName, @@ -1193,6 +1273,7 @@ async function applyReleaseCompletionEffect( reason: 'released', }); } + return true; } async function dispatchNodeAttempt( diff --git a/packages/engine/src/engine/agent.ts b/packages/engine/src/engine/agent.ts index a3c94c5e..ef2906c1 100644 --- a/packages/engine/src/engine/agent.ts +++ b/packages/engine/src/engine/agent.ts @@ -8,7 +8,24 @@ import { directNodeIdForAgent, ensureDirectNodeForAgent } from './node.js'; type Db = ReturnType; -const STALE_THRESHOLD_MS = 5 * 60 * 1000; // 5 minutes +/** How long an authenticated agent can be silent before it is no longer present. */ +export const AGENT_LIVENESS_TTL_MS = 5 * 60 * 1000; + +type AgentPresenceRow = Pick; + +/** + * Resolve the public presence status from server-observed activity. Persisted + * `active` is only the last reported state; it is not proof of current life. + */ +export function effectiveAgentStatus(agent: AgentPresenceRow, now = Date.now()): string { + if ( + (agent.status === 'active' || agent.status === 'online') + && now - agent.lastSeen.getTime() > AGENT_LIVENESS_TTL_MS + ) { + return 'offline'; + } + return agent.status === 'online' ? 'active' : agent.status; +} /** Detect unique constraint violations across D1, SQLite, and drizzle error shapes. */ function isUniqueConstraintError(err: unknown): boolean { @@ -107,33 +124,27 @@ export async function registerAgent( } export async function listAgents(db: Db, workspaceId: string, status?: string) { - let rows; - if (status && status !== 'all') { - rows = await db - .select() - .from(agents) - .where( - and(eq(agents.workspaceId, workspaceId), eq(agents.status, status)), - ); - } else { - rows = await db - .select() - .from(agents) - .where(eq(agents.workspaceId, workspaceId)); - } + // Keep the durable state aligned as a cleanup side effect, while still + // deriving below so correctness never depends on a cron/sweep having run. + await sweepStaleAgents(db, workspaceId); + const rows = await db + .select() + .from(agents) + .where(eq(agents.workspaceId, workspaceId)); + const requestedStatus = status === 'online' ? 'active' : status; return rows.map((a) => ({ id: a.id, name: a.name, handle: `@${a.name}`, type: a.type, - status: a.status, + status: effectiveAgentStatus(a), persona: a.persona, capabilities: a.capabilities ?? null, created_at: a.createdAt.toISOString(), last_seen: a.lastSeen.toISOString(), metadata: a.metadata, - })); + })).filter((agent) => !requestedStatus || requestedStatus === 'all' || agent.status === requestedStatus); } export async function getAgentByName(db: Db, workspaceId: string, name: string) { @@ -202,7 +213,7 @@ export async function getAgentByName(db: Db, workspaceId: string, name: string) name: agent.name, handle: agent.handle ?? `@${agent.name}`, type: agent.type, - status: agent.status, + status: effectiveAgentStatus(agent), persona: agent.persona, capabilities: agent.capabilities ?? null, created_at: agent.createdAt.toISOString(), @@ -266,7 +277,7 @@ export async function updateAgent( name: updated.name, handle: `@${updated.name}`, type: updated.type, - status: updated.status, + status: effectiveAgentStatus(updated), persona: updated.persona, capabilities: updated.capabilities ?? null, created_at: updated.createdAt.toISOString(), @@ -295,14 +306,19 @@ export async function touchLastSeen(db: Db, agentId: string): Promise { .where(eq(agents.id, agentId)); } -export async function sweepStaleAgents(db: Db): Promise { - const cutoff = new Date(Date.now() - STALE_THRESHOLD_MS); +export async function sweepStaleAgents(db: Db, workspaceId?: string): Promise { + const cutoff = new Date(Date.now() - AGENT_LIVENESS_TTL_MS); + const stale = and( + inArray(agents.status, ['active', 'online']), + lt(agents.lastSeen, cutoff), + ...(workspaceId ? [eq(agents.workspaceId, workspaceId)] : []), + ); const result = await db .update(agents) .set({ status: 'offline' }) // Sweep both 'active' and legacy 'online' agents during the transition period - .where(and(inArray(agents.status, ['active', 'online']), lt(agents.lastSeen, cutoff))) + .where(stale) .returning({ id: agents.id }); return result.length; diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 7c45c046..ac443515 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -60,6 +60,7 @@ export { deliverPendingToNode } from './engine/delivery.js'; export { handleNodeReconnect } from './node-reconnect.js'; export { handleAgentDisconnect } from './agent-disconnect.js'; export { drainNodeInvocations, sweepTimedOutInvocations } from './node-invocations.js'; +export { AGENT_LIVENESS_TTL_MS, effectiveAgentStatus, sweepStaleAgents } from './engine/agent.js'; export type { SweepTimedOutInvocationsOptions } from './node-invocations.js'; export { handleNodeControlMessage } from './engine/node.js'; export type { HandleNodeControlMessageArgs, NodeSocketLike } from './engine/node.js'; diff --git a/packages/engine/src/routes/agent.ts b/packages/engine/src/routes/agent.ts index 6af50a83..afd60708 100644 --- a/packages/engine/src/routes/agent.ts +++ b/packages/engine/src/routes/agent.ts @@ -621,7 +621,10 @@ agentRoutes.post( caller_id: callerAgent?.id, caller_name: callerAgent?.name ?? callerNode?.name ?? 'workspace', }, - { nodeConnections: c.get('engine').nodeConnections }, + { + nodeConnections: c.get('engine').nodeConnections, + completionDeps: c.get('engine'), + }, ); const eventData = { From cdd8968c5586b03771b06ac9c198b02c36d01456 Mon Sep 17 00:00:00 2001 From: Barry Cape Date: Thu, 6 Aug 2026 06:23:29 -0400 Subject: [PATCH 2/8] fix(engine): harden agent lifecycle follow-up --- .../conformance/agentLifecycle.test.ts | 161 +++++++++++++++++- packages/engine/src/engine/action.ts | 155 ++++++++++++++--- packages/engine/src/engine/agent.ts | 138 ++++++++++----- 3 files changed, 389 insertions(+), 65 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index 8bd5135a..4e7ad1e7 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -41,6 +41,111 @@ describe('agent presence and release lifecycle', () => { expect(persisted.status).toBe('offline'); }); + it('persists stale presence before returning agent detail', async () => { + const ws = await createWorkspace(stack.app, 'agent-detail-presence-expiry'); + const stale = await registerAgent(stack.app, ws.workspaceKey, 'stale-detail-agent'); + await stack.runtime.deps.db + .update(agents) + .set({ + status: 'active', + lastSeen: new Date(Date.now() - AGENT_LIVENESS_TTL_MS - 1_000), + }) + .where(eq(agents.id, stale.agentId)); + + const response = await stack.app.request(`/v1/agents/${stale.name}`, { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + expect(response.status).toBe(200); + expect((await response.json() as { data: { status: string } }).data.status).toBe('offline'); + + const [persisted] = await stack.runtime.deps.db + .select({ status: agents.status }) + .from(agents) + .where(eq(agents.id, stale.agentId)); + expect(persisted.status).toBe('offline'); + }); + + it('clamps a future last_seen before applying the liveness window', async () => { + const ws = await createWorkspace(stack.app, 'agent-future-presence'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'future-agent'); + const beforeRead = Date.now(); + await stack.runtime.deps.db + .update(agents) + .set({ + status: 'active', + lastSeen: new Date(beforeRead + 14 * 60 * 1000), + }) + .where(eq(agents.id, target.agentId)); + + const response = await stack.app.request(`/v1/agents/${target.name}`, { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + expect(response.status).toBe(200); + expect((await response.json() as { data: { status: string } }).data.status).toBe('active'); + + const afterRead = Date.now(); + const [persisted] = await stack.runtime.deps.db + .select({ lastSeen: agents.lastSeen }) + .from(agents) + .where(eq(agents.id, target.agentId)); + // SQLite timestamp mode stores whole seconds. + expect(persisted.lastSeen.getTime()).toBeGreaterThanOrEqual(beforeRead - 1_000); + expect(persisted.lastSeen.getTime()).toBeLessThanOrEqual(afterRead); + }); + + it('atomically registers human rows with an implicit direct binding', async () => { + const ws = await createWorkspace(stack.app, 'human-direct-registration'); + const response = await stack.app.request('/v1/agents', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: 'direct-human', type: 'human' }), + }); + expect(response.status).toBe(201); + const body = await response.json() as { data: { id: string } }; + const nodeId = `node_direct_${body.data.id}`; + + const [agent] = await stack.runtime.deps.db + .select({ type: agents.type, locationType: agents.locationType, locationNodeId: agents.locationNodeId }) + .from(agents) + .where(eq(agents.id, body.data.id)); + expect(agent).toEqual({ type: 'human', locationType: 'via_node', locationNodeId: nodeId }); + expect(await stack.runtime.deps.db + .select() + .from(nodes) + .where(and(eq(nodes.workspaceId, ws.workspaceId), eq(nodes.id, nodeId)))) + .toHaveLength(1); + expect(await stack.runtime.deps.db + .select() + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, ws.workspaceId), + eq(agentNodeBindings.agentId, body.data.id), + eq(agentNodeBindings.nodeId, nodeId), + eq(agentNodeBindings.status, 'active'), + ))) + .toHaveLength(1); + + const duplicate = await stack.app.request('/v1/agents', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: 'direct-human', type: 'human' }), + }); + expect(duplicate.status).toBe(409); + // The failed batch inserted its generated direct node before it hit the + // duplicate agent name; rollback must leave no orphan node behind. + expect(await stack.runtime.deps.db + .select() + .from(nodes) + .where(eq(nodes.workspaceId, ws.workspaceId))) + .toHaveLength(1); + }); + it('fails release explicitly when the agent has no live host', async () => { const ws = await createWorkspace(stack.app, 'hostless-agent-release'); const target = await registerAgent(stack.app, ws.workspaceKey, 'hostless-agent'); @@ -115,10 +220,64 @@ describe('agent presence and release lifecycle', () => { expect(await stack.runtime.deps.db.select().from(nodes).where(eq(nodes.id, nodeId))).toHaveLength(0); }); - it('continues dispatching release to a live host', async () => { + it('rolls back every local reap mutation when invocation completion fails', async () => { + const ws = await createWorkspace(stack.app, 'hostless-agent-delete-rollback'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'keep-me'); + const nodeId = `node_direct_${target.agentId}`; + await stack.runtime.deps.db + .update(agents) + .set({ locationType: 'self_connected', locationNodeId: null }) + .where(eq(agents.id, target.agentId)); + stack.runtime.handle.sqlite.exec(` + CREATE TRIGGER fail_local_release_completion + BEFORE UPDATE ON action_invocations + WHEN NEW.status = 'completed' AND NEW.action_name = 'release' + BEGIN + SELECT RAISE(ABORT, 'forced invocation completion failure'); + END + `); + + const response = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: target.name, delete_agent: true }), + }); + expect(response.status).toBe(500); + + expect(await stack.runtime.deps.db.select().from(agents).where(eq(agents.id, target.agentId))).toHaveLength(1); + expect(await stack.runtime.deps.db.select().from(nodes).where(eq(nodes.id, nodeId))).toHaveLength(1); + const [binding] = await stack.runtime.deps.db + .select({ status: agentNodeBindings.status }) + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, ws.workspaceId), + eq(agentNodeBindings.agentId, target.agentId), + eq(agentNodeBindings.nodeId, nodeId), + )); + expect(binding.status).toBe('active'); + const [invocation] = await stack.runtime.deps.db + .select({ status: actionInvocations.status }) + .from(actionInvocations) + .where(and( + eq(actionInvocations.workspaceId, ws.workspaceId), + eq(actionInvocations.actionName, 'release'), + )); + expect(invocation.status).toBe('pending'); + }); + + it('dispatches release through a live implicit direct binding', async () => { const ws = await createWorkspace(stack.app, 'live-agent-release'); const target = await registerAgent(stack.app, ws.workspaceKey, 'live-agent'); const { sock, handle, nodeId } = await attachDirectNodeSocket(stack, ws.workspaceId, target); + // Legacy directly registered rows can lack a durable location even though + // their implicit node binding and connection are both live. + await stack.runtime.deps.db + .update(agents) + .set({ locationType: 'self_connected', locationNodeId: null }) + .where(eq(agents.id, target.agentId)); const response = await stack.app.request('/v1/agents/release', { method: 'POST', diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index 89704b38..c29a0095 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -11,7 +11,7 @@ import { type InvocationCompletionDeps, } from './invocationCompletion.js'; import type { NodeConnectionRegistry } from '../ports/realtime.js'; -import { runAtomic } from '../ports/database.js'; +import { runAtomic, runAtomicWrites, type AtomicWrite } from '../ports/database.js'; import { claimSpawnNode, chooseNodeForAction, isNodeLive, releaseNodeCapacity, reserveNodeCapacity } from './placement.js'; import { DEFAULT_PROVIDER_NAME, capacityProviderName, getProvider, isProviderLive } from './nodeProvider.js'; @@ -690,27 +690,127 @@ async function dispatchRelease(args: { action_name: 'release', }); const completeLocally = async () => { - await applyReleaseCompletionEffect( - args.db, - args.workspaceId, - agent.locationNodeId, - invocation, - {}, - args.completionDeps, - { allowMissingBinding: true, expectedAgentId: agent.id }, - ); - await args.db - .update(actionInvocations) - .set({ - status: 'completed', - output: { released: true, deleted: input.delete_agent === true, reaped_locally: true }, - completedAt: new Date(), - }) + const activeBindings = await args.db + .select({ nodeId: agentNodeBindings.nodeId }) + .from(agentNodeBindings) .where(and( - eq(actionInvocations.workspaceId, args.workspaceId), - eq(actionInvocations.id, invocation.id), - inArray(actionInvocations.status, OPEN_INVOCATION_STATUSES), + eq(agentNodeBindings.workspaceId, args.workspaceId), + eq(agentNodeBindings.agentId, agent.id), + eq(agentNodeBindings.status, 'active'), )); + const activeNodeIds = Array.from(new Set(activeBindings.map((binding) => binding.nodeId))); + const completedAt = new Date(); + const invocationIsOpen = sql`EXISTS ( + SELECT 1 FROM ${actionInvocations} + WHERE ${actionInvocations.workspaceId} = ${args.workspaceId} + AND ${actionInvocations.id} = ${invocation.id} + AND ${actionInvocations.status} IN ('pending', 'dispatched', 'invoked') + )`; + + const results = await runAtomicWrites(args.db, (writeDb) => { + const writes: AtomicWrite[] = []; + + if (activeNodeIds.length > 0) { + // Decrement only nodes whose binding is still active when this atomic + // unit begins, so a retry cannot consume another agent's capacity. + writes.push(writeDb + .update(nodes) + .set({ + activeAgents: sql`CASE WHEN ${nodes.activeAgents} > 0 THEN ${nodes.activeAgents} - 1 ELSE 0 END`, + }) + .where(and( + eq(nodes.workspaceId, args.workspaceId), + inArray(nodes.id, activeNodeIds), + invocationIsOpen, + sql`EXISTS ( + SELECT 1 FROM ${agentNodeBindings} + WHERE ${agentNodeBindings.workspaceId} = ${args.workspaceId} + AND ${agentNodeBindings.agentId} = ${agent.id} + AND ${agentNodeBindings.nodeId} = ${nodes.id} + AND ${agentNodeBindings.status} = 'active' + )`, + ))); + } + + writes.push(writeDb + .update(agentNodeBindings) + .set({ status: 'inactive', updatedAt: completedAt }) + .where(and( + eq(agentNodeBindings.workspaceId, args.workspaceId), + eq(agentNodeBindings.agentId, agent.id), + eq(agentNodeBindings.status, 'active'), + invocationIsOpen, + ))); + + if (input.delete_agent === true) { + writes.push(writeDb + .delete(agents) + .where(and( + eq(agents.workspaceId, args.workspaceId), + eq(agents.id, agent.id), + invocationIsOpen, + ))); + writes.push(writeDb + .delete(nodes) + .where(and( + eq(nodes.workspaceId, args.workspaceId), + eq(nodes.id, `node_direct_${agent.id}`), + invocationIsOpen, + ))); + } else { + const existingMetadata = agent.metadata ?? {}; + const { spawn: _spawn, cli: _cli, ...restMetadata } = existingMetadata; + writes.push(writeDb + .update(agents) + .set({ + status: 'offline', + locationType: 'self_connected', + locationNodeId: null, + lastSeen: completedAt, + metadata: { + ...restMetadata, + release: { + reason: typeof input.reason === 'string' ? input.reason : null, + released_at: completedAt.toISOString(), + }, + }, + }) + .where(and( + eq(agents.workspaceId, args.workspaceId), + eq(agents.id, agent.id), + invocationIsOpen, + ))); + } + + writes.push(writeDb + .update(actionInvocations) + .set({ + status: 'completed', + output: { released: true, deleted: input.delete_agent === true, reaped_locally: true }, + completedAt, + }) + .where(and( + eq(actionInvocations.workspaceId, args.workspaceId), + eq(actionInvocations.id, invocation.id), + inArray(actionInvocations.status, OPEN_INVOCATION_STATUSES), + )) + .returning({ id: actionInvocations.id })); + + return writes; + }); + const completed = results.at(-1) as Array<{ id: string }>; + + // External completion effects belong after the durable atomic unit: an + // aborted local reap must never publish agent.exited. + if (completed.length > 0 && args.completionDeps && agent.locationNodeId) { + await emitAgentExitedEffects(args.completionDeps, args.workspaceId, { + agentId: agent.id, + agentName: agent.name, + nodeId: agent.locationNodeId, + invocationId: fleetInvocationId(agent.metadata), + reason: 'released', + }); + } return { invocation_id: invocation.id, action_name: 'release', @@ -739,9 +839,20 @@ async function dispatchRelease(args: { }; const registry = args.registry; - const nodeId = agent.locationNodeId; + const activeBindings = await args.db + .select({ nodeId: agentNodeBindings.nodeId }) + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, args.workspaceId), + eq(agentNodeBindings.agentId, agent.id), + eq(agentNodeBindings.status, 'active'), + )); + const implicitDirectNodeId = `node_direct_${agent.id}`; + const nodeId = activeBindings.find((binding) => binding.nodeId === agent.locationNodeId)?.nodeId + ?? activeBindings.find((binding) => binding.nodeId === implicitDirectNodeId)?.nodeId + ?? activeBindings[0]?.nodeId + ?? (agent.locationType === 'via_node' ? agent.locationNodeId : null); const hostLive = !!registry - && agent.locationType === 'via_node' && !!nodeId && await isHandlerConnectionLive( args.db, diff --git a/packages/engine/src/engine/agent.ts b/packages/engine/src/engine/agent.ts index ef2906c1..b3b3ac69 100644 --- a/packages/engine/src/engine/agent.ts +++ b/packages/engine/src/engine/agent.ts @@ -1,10 +1,11 @@ -import { eq, and, lt, inArray } from 'drizzle-orm'; +import { eq, and, gt, lt, inArray } from 'drizzle-orm'; import type { getDb } from '../db/index.js'; -import { agents, channels, channelMembers, actions, deliveries, nodes } from '../db/schema.js'; +import { agents, agentNodeBindings, channels, channelMembers, actions, deliveries, nodes } from '../db/schema.js'; import { randomHex, sha256Hex } from '../lib/crypto.js'; import { generateId } from './snowflake.js'; import { codedError } from '../lib/httpError.js'; -import { directNodeIdForAgent, ensureDirectNodeForAgent } from './node.js'; +import { directNodeIdForAgent } from './node.js'; +import { runAtomicWrites, type AtomicWrite } from '../ports/database.js'; type Db = ReturnType; @@ -18,9 +19,13 @@ type AgentPresenceRow = Pick; * `active` is only the last reported state; it is not proof of current life. */ export function effectiveAgentStatus(agent: AgentPresenceRow, now = Date.now()): string { + // A client-supplied/faulty clock must not create a negative age. The + // workspace sweep durably clamps future timestamps to its server clock so + // they subsequently expire normally after one TTL window. + const observedAt = Math.min(agent.lastSeen.getTime(), now); if ( (agent.status === 'active' || agent.status === 'online') - && now - agent.lastSeen.getTime() > AGENT_LIVENESS_TTL_MS + && now - observedAt > AGENT_LIVENESS_TTL_MS ) { return 'offline'; } @@ -57,29 +62,82 @@ export async function registerAgent( const agentId = generateId(); const token = `at_live_${randomHex(16)}`; const tokenHash = await sha256Hex(token); + const directNodeId = directNodeIdForAgent(agentId); + const directNodeTokenHash = await sha256Hex(`implicit_direct:${workspaceId}:${agentId}:${randomHex(16)}`); + const now = new Date(); - // Use INSERT directly and let the unique index (workspace_id, name) enforce - // uniqueness. Avoids TOCTOU race between SELECT check and INSERT that causes - // false "already exists" errors on D1 read replicas after delete+re-register. + const [generalChannel] = await db + .select() + .from(channels) + .where(and(eq(channels.workspaceId, workspaceId), eq(channels.name, 'general'))); + + // Register the identity, implicit direct node, active binding, and default + // membership as one atomic unit. A failed direct-node write must not leave a + // live but undispatchable roster row behind. let agent; try { - [agent] = await db - .insert(agents) - .values({ - id: agentId, + const results = await runAtomicWrites(db, (writeDb) => { + const writes: AtomicWrite[] = [writeDb.insert(nodes).values({ + id: directNodeId, + workspaceId, + name: `direct-${agentId}`, + tokenHash: directNodeTokenHash, + kind: 'ws', + role: 'direct', + deliveryAdapter: 'ws.node.v1', + deliveryConfig: { implicit: true, agent_id: agentId, agent_name: data.name }, + capabilities: [], + maxAgents: 1, + activeAgents: 1, + tags: ['implicit', 'direct'], + version: 'implicit', + status: 'offline', + handlersLive: false, + load: 0, + lastHeartbeatAt: null, + createdAt: now, + }), writeDb + .insert(agents) + .values({ + id: agentId, + workspaceId, + name: data.name, + handle: `@${data.name}`, + type: data.type || 'agent', + // Set status explicitly: the column DEFAULT is the deprecated 'online' + // on databases migrated before 0013, and SQLite can't ALTER a default. + status: 'active', + tokenHash, + persona: data.persona ?? null, + metadata: data.metadata ?? {}, + capabilities: data.capabilities ?? null, + locationType: 'via_node', + locationNodeId: directNodeId, + }) + .returning()]; + + if (generalChannel) { + writes.push(writeDb.insert(channelMembers).values({ + channelId: generalChannel.id, + agentId, + role: 'member', + })); + } + + writes.push(writeDb.insert(agentNodeBindings).values({ + id: `anb_${generateId()}`, workspaceId, - name: data.name, - handle: `@${data.name}`, - type: data.type || 'agent', - // Set status explicitly: the column DEFAULT is the deprecated 'online' - // on databases migrated before 0013, and SQLite can't ALTER a default. + agentId, + nodeId: directNodeId, status: 'active', - tokenHash, - persona: data.persona ?? null, - metadata: data.metadata ?? {}, - capabilities: data.capabilities ?? null, - }) - .returning(); + sessionRef: null, + priority: 0, + createdAt: now, + updatedAt: now, + })); + return writes; + }); + [agent] = results[1] as (typeof agents.$inferSelect)[]; } catch (insertErr: unknown) { // Unique constraint violation on (workspace_id, name) → agent already exists // D1 uses .code = 'SQLITE_CONSTRAINT_UNIQUE', drizzle may wrap in its own error, @@ -90,24 +148,6 @@ export async function registerAgent( throw insertErr; } - // Auto-join #general - const [generalChannel] = await db - .select() - .from(channels) - .where( - and(eq(channels.workspaceId, workspaceId), eq(channels.name, 'general')), - ); - - if (generalChannel) { - await db.insert(channelMembers).values({ - channelId: generalChannel.id, - agentId, - role: 'member', - }); - } - - await ensureDirectNodeForAgent(db, workspaceId, agent); - return { id: agentId, // Return the workspace id so a client that joined by workspace key (and @@ -148,6 +188,9 @@ export async function listAgents(db: Db, workspaceId: string, status?: string) { } export async function getAgentByName(db: Db, workspaceId: string, name: string) { + // Match roster reads: detail consumers should observe both derived and + // durable presence consistently within this workspace. + await sweepStaleAgents(db, workspaceId); const [agent] = await db .select() .from(agents) @@ -307,7 +350,18 @@ export async function touchLastSeen(db: Db, agentId: string): Promise { } export async function sweepStaleAgents(db: Db, workspaceId?: string): Promise { - const cutoff = new Date(Date.now() - AGENT_LIVENESS_TTL_MS); + const now = new Date(); + const cutoff = new Date(now.getTime() - AGENT_LIVENESS_TTL_MS); + const future = and( + inArray(agents.status, ['active', 'online']), + gt(agents.lastSeen, now), + ...(workspaceId ? [eq(agents.workspaceId, workspaceId)] : []), + ); + const normalized = await db + .update(agents) + .set({ lastSeen: now }) + .where(future) + .returning({ id: agents.id }); const stale = and( inArray(agents.status, ['active', 'online']), lt(agents.lastSeen, cutoff), @@ -321,5 +375,5 @@ export async function sweepStaleAgents(db: Db, workspaceId?: string): Promise Date: Thu, 6 Aug 2026 21:43:01 +0200 Subject: [PATCH 3/8] fix(engine): harden local release reaping --- .../conformance/agentLifecycle.test.ts | 54 +++++++++ packages/engine/src/engine/action.ts | 110 ++++++------------ 2 files changed, 90 insertions(+), 74 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index 4e7ad1e7..fa65c7a7 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -220,6 +220,60 @@ describe('agent presence and release lifecycle', () => { expect(await stack.runtime.deps.db.select().from(nodes).where(eq(nodes.id, nodeId))).toHaveLength(0); }); + it('releases capacity from the binding that local reaping deactivates', async () => { + const ws = await createWorkspace(stack.app, 'hostless-agent-binding-capacity'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'reap-bound-agent'); + const enrolled = await stack.app.request('/v1/nodes', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ + node_id: 'node_reap_target', + name: 'reap-target', + role: 'broker', + max_agents: 1, + }), + }); + expect(enrolled.status).toBe(201); + const bound = await stack.app.request('/v1/nodes/reap-target/agents', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ agent_name: target.name }), + }); + expect(bound.status).toBe(201); + + const response = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: target.name, delete_agent: true }), + }); + expect(response.status).toBe(201); + + const [node] = await stack.runtime.deps.db + .select({ activeAgents: nodes.activeAgents }) + .from(nodes) + .where(and(eq(nodes.workspaceId, ws.workspaceId), eq(nodes.id, 'node_reap_target'))); + expect(node.activeAgents).toBe(0); + expect(await stack.runtime.deps.db + .select() + .from(agentNodeBindings) + .where(and( + eq(agentNodeBindings.workspaceId, ws.workspaceId), + eq(agentNodeBindings.agentId, target.agentId), + eq(agentNodeBindings.nodeId, 'node_reap_target'), + eq(agentNodeBindings.status, 'active'), + ))) + .toHaveLength(0); + }); + it('rolls back every local reap mutation when invocation completion fails', async () => { const ws = await createWorkspace(stack.app, 'hostless-agent-delete-rollback'); const target = await registerAgent(stack.app, ws.workspaceKey, 'keep-me'); diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index c29a0095..ccbd6bae 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -690,15 +690,6 @@ async function dispatchRelease(args: { action_name: 'release', }); const completeLocally = async () => { - const activeBindings = await args.db - .select({ nodeId: agentNodeBindings.nodeId }) - .from(agentNodeBindings) - .where(and( - eq(agentNodeBindings.workspaceId, args.workspaceId), - eq(agentNodeBindings.agentId, agent.id), - eq(agentNodeBindings.status, 'active'), - )); - const activeNodeIds = Array.from(new Set(activeBindings.map((binding) => binding.nodeId))); const completedAt = new Date(); const invocationIsOpen = sql`EXISTS ( SELECT 1 FROM ${actionInvocations} @@ -710,27 +701,25 @@ async function dispatchRelease(args: { const results = await runAtomicWrites(args.db, (writeDb) => { const writes: AtomicWrite[] = []; - if (activeNodeIds.length > 0) { - // Decrement only nodes whose binding is still active when this atomic - // unit begins, so a retry cannot consume another agent's capacity. - writes.push(writeDb - .update(nodes) - .set({ - activeAgents: sql`CASE WHEN ${nodes.activeAgents} > 0 THEN ${nodes.activeAgents} - 1 ELSE 0 END`, - }) - .where(and( - eq(nodes.workspaceId, args.workspaceId), - inArray(nodes.id, activeNodeIds), - invocationIsOpen, - sql`EXISTS ( - SELECT 1 FROM ${agentNodeBindings} - WHERE ${agentNodeBindings.workspaceId} = ${args.workspaceId} - AND ${agentNodeBindings.agentId} = ${agent.id} - AND ${agentNodeBindings.nodeId} = ${nodes.id} - AND ${agentNodeBindings.status} = 'active' - )`, - ))); - } + // Resolve the active binding in this atomic unit instead of from a + // pre-transaction snapshot. A concurrent rebind therefore decrements + // the node that is actually deactivated below. + writes.push(writeDb + .update(nodes) + .set({ + activeAgents: sql`CASE WHEN ${nodes.activeAgents} > 0 THEN ${nodes.activeAgents} - 1 ELSE 0 END`, + }) + .where(and( + eq(nodes.workspaceId, args.workspaceId), + invocationIsOpen, + sql`EXISTS ( + SELECT 1 FROM ${agentNodeBindings} + WHERE ${agentNodeBindings.workspaceId} = ${args.workspaceId} + AND ${agentNodeBindings.agentId} = ${agent.id} + AND ${agentNodeBindings.nodeId} = ${nodes.id} + AND ${agentNodeBindings.status} = 'active' + )`, + ))); writes.push(writeDb .update(agentNodeBindings) @@ -742,51 +731,28 @@ async function dispatchRelease(args: { invocationIsOpen, ))); - if (input.delete_agent === true) { - writes.push(writeDb - .delete(agents) - .where(and( - eq(agents.workspaceId, args.workspaceId), - eq(agents.id, agent.id), - invocationIsOpen, - ))); - writes.push(writeDb - .delete(nodes) - .where(and( - eq(nodes.workspaceId, args.workspaceId), - eq(nodes.id, `node_direct_${agent.id}`), - invocationIsOpen, - ))); - } else { - const existingMetadata = agent.metadata ?? {}; - const { spawn: _spawn, cli: _cli, ...restMetadata } = existingMetadata; - writes.push(writeDb - .update(agents) - .set({ - status: 'offline', - locationType: 'self_connected', - locationNodeId: null, - lastSeen: completedAt, - metadata: { - ...restMetadata, - release: { - reason: typeof input.reason === 'string' ? input.reason : null, - released_at: completedAt.toISOString(), - }, - }, - }) - .where(and( - eq(agents.workspaceId, args.workspaceId), - eq(agents.id, agent.id), - invocationIsOpen, - ))); - } + // This helper is only used for delete_agent releases. Non-delete + // releases fail closed when no live host can receive the invocation. + writes.push(writeDb + .delete(agents) + .where(and( + eq(agents.workspaceId, args.workspaceId), + eq(agents.id, agent.id), + invocationIsOpen, + ))); + writes.push(writeDb + .delete(nodes) + .where(and( + eq(nodes.workspaceId, args.workspaceId), + eq(nodes.id, `node_direct_${agent.id}`), + invocationIsOpen, + ))); writes.push(writeDb .update(actionInvocations) .set({ status: 'completed', - output: { released: true, deleted: input.delete_agent === true, reaped_locally: true }, + output: { released: true, deleted: true, reaped_locally: true }, completedAt, }) .where(and( @@ -866,10 +832,6 @@ async function dispatchRelease(args: { return input.delete_agent === true ? completeLocally() : failClosed(); } - if (!registry || !nodeId) { - throw codedError(`Agent "${name}" has no live host node`, 'agent_host_unavailable', 503); - } - // Release is a capacity operation handled by the provider hosting the agent. const dispatched = await dispatchNodeInvocation({ db: args.db, From 78011801cfd9d9642199edd9474e9653b4b88244 Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 6 Aug 2026 21:54:43 +0200 Subject: [PATCH 4/8] fix(engine): emit exits for legacy local reaps --- .../conformance/agentLifecycle.test.ts | 17 +++++++++++++++-- packages/engine/src/engine/action.ts | 7 ++++--- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index fa65c7a7..848b47e0 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -1,6 +1,6 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { and, eq } from 'drizzle-orm'; -import { actionInvocations, agentNodeBindings, agents, nodes } from '../../db/schema.js'; +import { actionInvocations, agentNodeBindings, agents, nodes, workspaceEvents } from '../../db/schema.js'; import { AGENT_LIVENESS_TTL_MS } from '../../engine/agent.js'; import { attachDirectNodeSocket, @@ -214,10 +214,23 @@ describe('agent presence and release lifecycle', () => { body: JSON.stringify({ name: target.name, delete_agent: true }), }); expect(response.status).toBe(201); - expect((await response.json() as { data: { status: string } }).data.status).toBe('completed'); + expect((await response.json() as { data: { status: string; handler_node_id: string | null } }).data) + .toMatchObject({ status: 'completed', handler_node_id: nodeId }); expect(await stack.runtime.deps.db.select().from(agents).where(eq(agents.id, target.agentId))).toHaveLength(0); expect(await stack.runtime.deps.db.select().from(nodes).where(eq(nodes.id, nodeId))).toHaveLength(0); + const [exited] = await stack.runtime.deps.db + .select({ payload: workspaceEvents.payload }) + .from(workspaceEvents) + .where(and( + eq(workspaceEvents.workspaceId, ws.workspaceId), + eq(workspaceEvents.type, 'agent.exited'), + )); + expect(JSON.parse(exited.payload)).toMatchObject({ + agent_id: target.agentId, + node_id: nodeId, + reason: 'released', + }); }); it('releases capacity from the binding that local reaping deactivates', async () => { diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index ccbd6bae..c0325775 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -768,11 +768,12 @@ async function dispatchRelease(args: { // External completion effects belong after the durable atomic unit: an // aborted local reap must never publish agent.exited. - if (completed.length > 0 && args.completionDeps && agent.locationNodeId) { + const exitNodeId = nodeId ?? agent.locationNodeId; + if (completed.length > 0 && args.completionDeps && exitNodeId) { await emitAgentExitedEffects(args.completionDeps, args.workspaceId, { agentId: agent.id, agentName: agent.name, - nodeId: agent.locationNodeId, + nodeId: exitNodeId, invocationId: fleetInvocationId(agent.metadata), reason: 'released', }); @@ -781,7 +782,7 @@ async function dispatchRelease(args: { invocation_id: invocation.id, action_name: 'release', handler_agent_id: null, - handler_node_id: agent.locationNodeId, + handler_node_id: exitNodeId, dispatched_node_id: null, input, status: 'completed', From 4af9e7ec15a62a85d53d5c792015384b5549ee15 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 7 Aug 2026 14:57:58 +0200 Subject: [PATCH 5/8] fix(engine): release a name by tombstone, not by DELETE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The local reap inlined a bare `DELETE` on `agents` inside the same atomic unit as the binding update and the invocation completion. Four foreign keys reference `agents.id` without `onDelete` — channels.created_by (schema.ts:455), messages.agent_id (:503), files.uploaded_by (:666), webhooks.created_by (:759) — so SQLite refuses the delete for any agent that has ever created a channel, sent a message, uploaded a file, or created a webhook. Because the statement sits inside `runAtomicWrites`, that refusal aborted the whole unit, so the invocation never completed either: a transaction abort rather than a legible error, on exactly the agents the reap exists to clean up. Every existing `delete_agent` fixture registered a fresh agent and released it immediately, so the suite could not observe this. The added fixture posts one message first and reproduced it as `SQLITE_CONSTRAINT_FOREIGNKEY: FOREIGN KEY constraint failed` (HTTP 500) before this change. Cascade is not an alternative — it would delete the agent's message history, which is the thing worth keeping — and `messages.agent_id` is NOT NULL, so `set null` cannot apply. That leaves the tombstone rename proposed in #309: the unique key is `(workspace_id, name)`, so freeing the name only requires the name to stop colliding, not the row to disappear. The released row keeps its id, so every FK target stays valid and every message keeps its sender. It is renamed to `#released-`, marked `released`, and stamped with `metadata.release`. Two deliberate choices beyond #309's sketch: - the tombstone is keyed on the agent id rather than a timestamp. It runs inside an atomic batch, where a unique-constraint violation would abort the whole unit — reintroducing the failure being fixed. The id is already unique per workspace, so the name cannot collide and a repeat release is idempotent. The release time is preserved in `metadata.release.releasedAt`. - `token_hash` is rotated to an unheld value. The row survives the release, and `token_hash` is NOT NULL UNIQUE so it cannot be cleared; without the rotation a released agent's old token would keep authenticating. `listAgents` now excludes released rows so `agent list` does not fill with tombstones. Refs #309 --- .../conformance/agentLifecycle.test.ts | 72 ++++++++++++++++++- packages/engine/src/engine/action.ts | 44 +++++++++++- packages/engine/src/engine/agent.ts | 25 ++++++- 3 files changed, 136 insertions(+), 5 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index 848b47e0..355201b8 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -217,7 +217,21 @@ describe('agent presence and release lifecycle', () => { expect((await response.json() as { data: { status: string; handler_node_id: string | null } }).data) .toMatchObject({ status: 'completed', handler_node_id: nodeId }); - expect(await stack.runtime.deps.db.select().from(agents).where(eq(agents.id, target.agentId))).toHaveLength(0); + // The name is freed; the row is retained as a tombstone so the agent's + // history keeps its author (relaycast#309). + expect(await stack.runtime.deps.db + .select() + .from(agents) + .where(and(eq(agents.workspaceId, ws.workspaceId), eq(agents.name, target.name)))) + .toHaveLength(0); + const [tombstone] = await stack.runtime.deps.db + .select({ name: agents.name, status: agents.status, tokenHash: agents.tokenHash }) + .from(agents) + .where(eq(agents.id, target.agentId)); + expect(tombstone).toMatchObject({ + name: `${target.name}#released-${target.agentId}`, + status: 'released', + }); expect(await stack.runtime.deps.db.select().from(nodes).where(eq(nodes.id, nodeId))).toHaveLength(0); const [exited] = await stack.runtime.deps.db .select({ payload: workspaceEvents.payload }) @@ -233,6 +247,62 @@ describe('agent presence and release lifecycle', () => { }); }); + it('reaps a hostless agent that has already spoken', async () => { + const ws = await createWorkspace(stack.app, 'hostless-agent-delete-with-history'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'talkative-agent'); + const nodeId = `node_direct_${target.agentId}`; + + // Every agent worth reaping has history. Four FKs reference agents.id + // without onDelete (channels.created_by, messages.agent_id, files.uploaded_by, + // webhooks.created_by), so a bare DELETE on the row is refused for any agent + // that has ever spoken — and inside runAtomicWrites that refusal aborts the + // binding update and the invocation completion along with it. + const posted = await stack.app.request('/v1/channels/general/messages', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${target.token}`, + }, + body: JSON.stringify({ text: 'i have said something' }), + }); + expect(posted.status).toBe(201); + + await stack.runtime.deps.db + .update(agents) + .set({ locationType: 'self_connected', locationNodeId: null }) + .where(eq(agents.id, target.agentId)); + + const response = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${ws.workspaceKey}`, + }, + body: JSON.stringify({ name: target.name, delete_agent: true }), + }); + expect(response.status).toBe(201); + expect((await response.json() as { data: { status: string } }).data) + .toMatchObject({ status: 'completed' }); + + // The name is released and the implicit direct node is gone... + expect(await stack.runtime.deps.db + .select() + .from(agents) + .where(and(eq(agents.workspaceId, ws.workspaceId), eq(agents.name, target.name)))) + .toHaveLength(0); + expect(await stack.runtime.deps.db.select().from(nodes).where(eq(nodes.id, nodeId))).toHaveLength(0); + + // ...and the invocation actually completed rather than being aborted. + const [invocation] = await stack.runtime.deps.db + .select({ status: actionInvocations.status }) + .from(actionInvocations) + .where(and( + eq(actionInvocations.workspaceId, ws.workspaceId), + eq(actionInvocations.actionName, 'release'), + )); + expect(invocation.status).toBe('completed'); + }); + it('releases capacity from the binding that local reaping deactivates', async () => { const ws = await createWorkspace(stack.app, 'hostless-agent-binding-capacity'); const target = await registerAgent(stack.app, ws.workspaceKey, 'reap-bound-agent'); diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index c0325775..f1a0b84b 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -2,6 +2,8 @@ import { and, asc, eq, inArray, isNotNull, isNull, lte, or, sql } from 'drizzle- import type { getDb } from '../db/index.js'; import { actions, actionInvocations, agents, agentNodeBindings, nodes } from '../db/schema.js'; import { generateId } from './snowflake.js'; +import { RELEASED_AGENT_STATUS, releasedAgentName } from './agent.js'; +import { randomHex, sha256Hex } from '../lib/crypto.js'; import { codedError } from '../lib/httpError.js'; import { toFleetWireJson } from './deliveryWire.js'; import { @@ -691,6 +693,13 @@ async function dispatchRelease(args: { }); const completeLocally = async () => { const completedAt = new Date(); + // Keyed on the agent id, not on the clock: the id is already unique per + // workspace, so the tombstone can never collide with an existing row (a + // second release of the same row is idempotent). A timestamped name would + // reintroduce a unique-constraint abort into the very path this is fixing. + // The release time is preserved in `metadata.release.releasedAt`. + const releasedName = releasedAgentName(agent.name, agent.id); + const releasedTokenHash = await sha256Hex(`released:${agent.id}:${randomHex(16)}`); const invocationIsOpen = sql`EXISTS ( SELECT 1 FROM ${actionInvocations} WHERE ${actionInvocations.workspaceId} = ${args.workspaceId} @@ -733,8 +742,32 @@ async function dispatchRelease(args: { // This helper is only used for delete_agent releases. Non-delete // releases fail closed when no live host can receive the invocation. + // + // Tombstone-rename rather than DELETE (relaycast#309). Four FKs reference + // `agents.id` without `onDelete` — channels.created_by (schema.ts:455), + // messages.agent_id (:503), files.uploaded_by (:666), + // webhooks.created_by (:759) — so a bare DELETE is refused for any agent + // that has ever spoken, and inside this atomic unit that refusal aborts + // the binding update and the invocation completion along with it. Cascade + // is not an option either: it would destroy the agent's message history, + // and `messages.agent_id` is NOT NULL so `set null` cannot apply. + // + // Renaming frees the unique `(workspace_id, name)` immediately while + // every FK target stays valid and every message keeps its sender. writes.push(writeDb - .delete(agents) + .update(agents) + .set({ + name: releasedName, + handle: `@${releasedName}`, + status: RELEASED_AGENT_STATUS, + // The row survives, so its credential must not. `token_hash` is + // NOT NULL UNIQUE and cannot be cleared, so rotate it to a value + // nobody holds; the released agent's old token stops authenticating. + tokenHash: releasedTokenHash, + metadata: sql`json_patch(COALESCE(${agents.metadata}, '{}'), ${JSON.stringify({ + release: { reason: 'released', releasedAt: completedAt.toISOString(), previousName: agent.name }, + })})`, + }) .where(and( eq(agents.workspaceId, args.workspaceId), eq(agents.id, agent.id), @@ -752,7 +785,14 @@ async function dispatchRelease(args: { .update(actionInvocations) .set({ status: 'completed', - output: { released: true, deleted: true, reaped_locally: true }, + output: { + released: true, + // The roster row is retained as a tombstone so the agent's history + // keeps its author; the name is what the caller gets back. + deleted: false, + reaped_locally: true, + released_name: releasedName, + }, completedAt, }) .where(and( diff --git a/packages/engine/src/engine/agent.ts b/packages/engine/src/engine/agent.ts index b3b3ac69..f08aadf2 100644 --- a/packages/engine/src/engine/agent.ts +++ b/packages/engine/src/engine/agent.ts @@ -1,4 +1,4 @@ -import { eq, and, gt, lt, inArray } from 'drizzle-orm'; +import { eq, and, gt, lt, ne, inArray } from 'drizzle-orm'; import type { getDb } from '../db/index.js'; import { agents, agentNodeBindings, channels, channelMembers, actions, deliveries, nodes } from '../db/schema.js'; import { randomHex, sha256Hex } from '../lib/crypto.js'; @@ -14,6 +14,25 @@ export const AGENT_LIVENESS_TTL_MS = 5 * 60 * 1000; type AgentPresenceRow = Pick; +/** + * Status of a row whose name has been released back to the workspace + * (relaycast#309). The row is retained so the four RESTRICT foreign keys to + * `agents.id` stay valid and the agent's history keeps its author; only the + * name is freed. A released row is not a roster member and never serves as a + * delivery target. + */ +export const RELEASED_AGENT_STATUS = 'released'; + +/** + * Tombstone name for a released agent. Keyed on the agent id rather than a + * timestamp so it is unique by construction and idempotent on repeat release — + * the reap runs inside an atomic batch where a unique-constraint violation + * would abort the whole unit rather than fail just this statement. + */ +export function releasedAgentName(name: string, agentId: string): string { + return `${name}#released-${agentId}`; +} + /** * Resolve the public presence status from server-observed activity. Persisted * `active` is only the last reported state; it is not proof of current life. @@ -170,7 +189,9 @@ export async function listAgents(db: Db, workspaceId: string, status?: string) { const rows = await db .select() .from(agents) - .where(eq(agents.workspaceId, workspaceId)); + // Released rows are tombstones retained only to keep history attributable; + // they are not roster members, so `agent list` must not fill with them. + .where(and(eq(agents.workspaceId, workspaceId), ne(agents.status, RELEASED_AGENT_STATUS))); const requestedStatus = status === 'online' ? 'active' : status; return rows.map((a) => ({ From 9f681246cd830ced773d5c54588ae7f6bcee02f4 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 7 Aug 2026 14:58:13 +0200 Subject: [PATCH 6/8] fix(engine): gate name reclaim on observed silence, not on status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restoring the presence sweep loosened an identity boundary as a side effect, and the trigger was a read. `registerAgentViaNode` reclaims a name via `onConflictDoUpdate` and overwrites `token_hash`, so a permitted reclaim is a full credential handover: the incumbent's token stops working and the claiming node is handed a live `at_live_` token for the same row. That decision was guarded by `setWhere: or(ne(agents.status, 'active'), )`. `status` is maintained by `sweepStaleAgents`, which this branch calls synchronously from `listAgents` and `getAgentByName`. So a plain `agent list` rewrote the column, and every record it flipped to 'offline' satisfied the first disjunct and moved from "reclaimable only by its own node" to "reclaimable by any node, on name alone". A read widened who may claim an identity. Presence and identity are different questions and must not share a field. The first disjunct now gates on observed silence — `last_seen` older than `AGENT_RECLAIM_GRACE_MS` — which reads cannot write. The owning-node disjunct is unchanged, so a node restart still re-registers freely. An agent is absent from the roster after 5 minutes of silence and its name is reclaimable by a stranger after 24 hours. Between those points it reads as away and its identity is still its own. The grace value is measured, not guessed (relaycast-cloud, 2026-08-07): of the 1,578 records this governs, silence was <5m: 5, 5m-24h: 9, 1d-7d: 1, >7d: 1,568. At 24h the eligible set is 1,569 against 1,578, so it costs essentially nothing steady-state while protecting the ~14 identities a human would still call live. The reasoning is recorded on the constant. Scope, measured on the same data: this workspace has 8 genuinely loosened records, not the 305 first estimated — 300 of its 308 stale-active rows sit on their implicit direct node and were already reclaimable by any node with no deploy at all. Fleet-wide the split is 1,578 loosened against 14,074 already open. The larger standing hole is the `location_node_id = 'node_direct_' || id` disjunct, which this change does not touch; that is #311's subject. Also fixes the reverse defect: a row stored 'active' but silent for weeks was previously NOT reclaimable, so a name stranded by a dead node stayed stranded. The grace window now expires into recovery. --- .../conformance/agentNameReclaim.test.ts | 146 ++++++++++++++++++ packages/engine/src/engine/agent.ts | 24 +++ packages/engine/src/engine/node.ts | 26 +++- 3 files changed, 194 insertions(+), 2 deletions(-) create mode 100644 packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts diff --git a/packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts b/packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts new file mode 100644 index 00000000..d764cbfb --- /dev/null +++ b/packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts @@ -0,0 +1,146 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { and, eq } from 'drizzle-orm'; +import { createWorkspace, FakeSocket, makeNodeStack, type TestStack } from './harness.js'; +import { agents } from '../../db/schema.js'; +import { AGENT_LIVENESS_TTL_MS, AGENT_RECLAIM_GRACE_MS } from '../../engine/agent.js'; + +/** + * Who may take an agent's name and be issued a token for it. + * + * `registerAgentViaNode` overwrites `token_hash` on conflict, so a permitted + * reclaim is a full credential handover: the incumbent's token stops working + * and the claiming node is handed a live one for the same row. The guard on + * that decision therefore reads observed silence (`last_seen`) rather than the + * `status` column, which `sweepStaleAgents` rewrites on every roster read. + */ +describe('agent name reclaim across nodes', () => { + let stack: TestStack; + beforeEach(() => { stack = makeNodeStack(); }); + afterEach(() => stack.close()); + + const db = () => stack.runtime.deps.db; + + async function bringNodeOnline( + ws: { workspaceKey: string; workspaceId: string }, + nodeId: string, + name: string, + ) { + const enrolled = await stack.app.request('/v1/nodes', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${ws.workspaceKey}` }, + body: JSON.stringify({ + node_id: nodeId, name, role: 'broker', + capabilities: ['spawn:claude'], max_agents: 4, tags: ['test'], version: 'v0', + }), + }); + expect(enrolled.status).toBe(201); + const sock = new FakeSocket(); + const handle = stack.runtime.realtime.attachNodeSocket(ws.workspaceId, nodeId, sock); + await handle.handleMessage(JSON.stringify({ + v: 1, type: 'node.register', name, node_id: nodeId, + capabilities: [{ name: 'spawn:claude', kind: 'capacity' }], + max_agents: 4, tags: ['test'], version: 'v1', resume_cursor: null, + })); + await handle.handleMessage(JSON.stringify({ + v: 1, type: 'node.heartbeat', load: 0, active_agents: 0, handlers_live: true, + })); + return { sock, handle }; + } + + async function registerViaNode( + node: { sock: FakeSocket; handle: { handleMessage(raw: string): Promise } }, + name: string, + ) { + await node.handle.handleMessage(JSON.stringify({ + v: 1, type: 'agent.register', name, resumable: true, + })); + } + + async function agentRow(workspaceId: string, name: string) { + const [row] = await db() + .select({ + id: agents.id, + tokenHash: agents.tokenHash, + locationNodeId: agents.locationNodeId, + status: agents.status, + }) + .from(agents) + .where(and(eq(agents.workspaceId, workspaceId), eq(agents.name, name))); + return row; + } + + /** Backdate observed activity without touching anything else. */ + async function silentFor(agentId: string, ms: number) { + await db().update(agents).set({ lastSeen: new Date(Date.now() - ms) }).where(eq(agents.id, agentId)); + } + + it('a roster read does not make a recently-active agent reclaimable by another node', async () => { + const ws = await createWorkspace(stack.app, 'reclaim-roster-read'); + const alpha = await bringNodeOnline(ws, 'node_alpha', 'alpha'); + const beta = await bringNodeOnline(ws, 'node_beta', 'beta'); + + await registerViaNode(alpha, 'contested'); + const before = await agentRow(ws.workspaceId, 'contested'); + expect(before.locationNodeId).toBe('node_alpha'); + + // Silent long enough to be absent from the roster, far short of the + // reclaim grace. This is the ordinary state of a working agent between + // bursts of activity. + await silentFor(before.id, AGENT_LIVENESS_TTL_MS * 2); + + // A plain roster read. This sweeps, so it rewrites `status` — which is + // exactly the write that used to widen the reclaim guard. + const roster = await stack.app.request('/v1/agents', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + expect(roster.status).toBe(200); + const swept = await agentRow(ws.workspaceId, 'contested'); + // Confirm the read really did flip the column, so this test cannot pass + // by the sweep silently not running. + expect(swept.status).toBe('offline'); + + // A foreign node now claims the name. + await registerViaNode(beta, 'contested'); + + const after = await agentRow(ws.workspaceId, 'contested'); + expect(after.locationNodeId).toBe('node_alpha'); + expect(after.tokenHash).toBe(before.tokenHash); + expect(after.id).toBe(before.id); + }); + + it('an agent silent beyond the reclaim grace can be reclaimed by another node', async () => { + const ws = await createWorkspace(stack.app, 'reclaim-after-grace'); + const alpha = await bringNodeOnline(ws, 'node_alpha', 'alpha'); + const beta = await bringNodeOnline(ws, 'node_beta', 'beta'); + + await registerViaNode(alpha, 'abandoned'); + const before = await agentRow(ws.workspaceId, 'abandoned'); + await silentFor(before.id, AGENT_RECLAIM_GRACE_MS + 60_000); + + await registerViaNode(beta, 'abandoned'); + + // The grace window must expire into something, or a name stranded by a + // dead node would be unrecoverable. + const after = await agentRow(ws.workspaceId, 'abandoned'); + expect(after.locationNodeId).toBe('node_beta'); + expect(after.tokenHash).not.toBe(before.tokenHash); + }); + + it('the owning node can re-register its own agent inside the grace window', async () => { + const ws = await createWorkspace(stack.app, 'reclaim-own-node'); + const alpha = await bringNodeOnline(ws, 'node_alpha', 'alpha'); + + await registerViaNode(alpha, 'restarted'); + const before = await agentRow(ws.workspaceId, 'restarted'); + await silentFor(before.id, AGENT_LIVENESS_TTL_MS * 2); + + // A node restart must never be blocked by the grace window — that would + // make the guard a fail-closed gate with no recovery path. + await registerViaNode(alpha, 'restarted'); + + const after = await agentRow(ws.workspaceId, 'restarted'); + expect(after.id).toBe(before.id); + expect(after.locationNodeId).toBe('node_alpha'); + expect(after.status).toBe('active'); + }); +}); diff --git a/packages/engine/src/engine/agent.ts b/packages/engine/src/engine/agent.ts index f08aadf2..d6dfb62d 100644 --- a/packages/engine/src/engine/agent.ts +++ b/packages/engine/src/engine/agent.ts @@ -12,6 +12,30 @@ type Db = ReturnType; /** How long an authenticated agent can be silent before it is no longer present. */ export const AGENT_LIVENESS_TTL_MS = 5 * 60 * 1000; +/** + * How long an agent must be silent before a DIFFERENT node may reclaim its + * name and overwrite its `token_hash` (`registerAgentViaNode`). + * + * Deliberately NOT `AGENT_LIVENESS_TTL_MS`, and deliberately not the `status` + * column. Presence and identity are different questions: "should the roster + * show this agent as here" is cheap to get wrong and self-corrects on the next + * heartbeat, whereas "may another node take this name and be issued a working + * token for it" is a credential handover that cannot be undone. They must not + * share a threshold, and they must not share a field that a roster read can + * write — gating identity on `status` meant a stale-agent sweep, triggered by + * an `agent list`, silently made records claimable. + * + * 24h chosen against measured production data (relaycast-cloud, 2026-08-07): + * of the 1,578 records whose reclaim eligibility this governs, silence was + * <5m: 5, 5m-24h: 9, 1d-7d: 1, >7d: 1,568. So 99.4% have been silent over a + * week and a 24h grace costs essentially nothing steady-state (1,569 eligible + * vs 1,578) while protecting the ~14 identities a human would still call live. + * + * Lower this only with fresh measurements — shortening it converts live agents + * into reclaimable ones. + */ +export const AGENT_RECLAIM_GRACE_MS = 24 * 60 * 60 * 1000; + type AgentPresenceRow = Pick; /** diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 3ce93381..eb4ed75a 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -1,4 +1,4 @@ -import { and, eq, inArray, ne, or, sql } from 'drizzle-orm'; +import { and, eq, inArray, lt, ne, or, sql } from 'drizzle-orm'; import type { FleetAgentRegisterMessage, FleetBrokerToRelaycastMessage, @@ -22,6 +22,7 @@ import { runAtomic } from '../ports/database.js'; import type { EngineDb } from '../ports/database.js'; import { isProviderAgentDeliveryReady, type NodeConnectionRegistry } from '../ports/realtime.js'; import { generateId } from './snowflake.js'; +import { AGENT_RECLAIM_GRACE_MS } from './agent.js'; import { isNodeLive, nodeHasCapability } from './placement.js'; import { DEFAULT_PROVIDER_NAME, @@ -1182,8 +1183,29 @@ export async function registerAgentViaNode( resumable: message.resumable ?? false, sessionRef: message.session_ref ?? null, }, + // Who may take this name and be issued a token for it. + // + // The first disjunct gates on OBSERVED SILENCE (`last_seen`), not on + // the `status` column. Those are not the same question and must not + // share a field. `status` is maintained by `sweepStaleAgents`, which + // runs on every roster read — so while identity was gated on it, an + // `agent list` flipped records to 'offline' and thereby moved them + // from "reclaimable only by their own node" to "reclaimable by any + // node, on name alone, with a `token_hash` overwrite". A read must + // never widen who may claim an identity; reads do not move + // `last_seen`, so gating here makes that structurally impossible. + // + // The grace window is far longer than the presence TTL: an agent goes + // 'offline' on the roster after 5 minutes of silence, but its name is + // not reclaimable by a stranger until AGENT_RECLAIM_GRACE_MS. Between + // those two points the agent reads as away and its identity is still + // its own. + // + // The second disjunct is unchanged: the agent's own node may always + // re-register it, so a node restart or reconnect is never blocked by + // the grace window. setWhere: or( - ne(agents.status, 'active'), + lt(agents.lastSeen, new Date(Date.now() - AGENT_RECLAIM_GRACE_MS)), and( eq(agents.locationType, 'via_node'), or( From bd0f9412b12e80c514d14569ef52caab2d700c7d Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 7 Aug 2026 15:16:56 +0200 Subject: [PATCH 7/8] test(engine): record why the owning-node reclaim test must keep passing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit That test asserts no new behaviour — it passes before and after the guard change. It pins a decision: gating reclaim on node identity alone was rejected because an agent whose node dies and respawns elsewhere could never reclaim its own name, and a stranded name has no recovery path short of relaycast#309. A test that guards a decision rather than a behaviour reads like a redundant one, and is the first to be deleted by someone simplifying the disjunct it protects. Say so in the test. --- .../conformance/agentNameReclaim.test.ts | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts b/packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts index d764cbfb..294615ee 100644 --- a/packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts +++ b/packages/engine/src/__tests__/conformance/agentNameReclaim.test.ts @@ -134,8 +134,20 @@ describe('agent name reclaim across nodes', () => { const before = await agentRow(ws.workspaceId, 'restarted'); await silentFor(before.id, AGENT_LIVENESS_TTL_MS * 2); - // A node restart must never be blocked by the grace window — that would - // make the guard a fail-closed gate with no recovery path. + // This test passes both before and after the guard change, deliberately. + // It does not assert new behaviour — it pins a DECISION. + // + // The alternative considered for the reclaim guard was to drop the status + // disjunct entirely and gate on node identity alone. That was rejected + // because it takes hostages: an agent whose broker node dies and respawns + // elsewhere could never reclaim its own name, and there is no recovery + // path for a stranded name short of relaycast#309's tombstone. A guard + // that protects an identity by making it unrecoverable has traded one + // outage for another. + // + // So this asserts the honest caller's path stays open. Anyone later + // "simplifying" the owning-node disjunct away turns this red, and finds + // this comment, rather than discovering the consequence in production. await registerViaNode(alpha, 'restarted'); const after = await agentRow(ws.workspaceId, 'restarted'); From 9df4b949d4f9230fd4b9d8eb1ac661451a239102 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 7 Aug 2026 15:27:01 +0200 Subject: [PATCH 8/8] fix(engine): reserve the tombstone namespace and close three review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four findings from cubic on the previous two commits. All verified against the code before being actioned; one falsifies a claim I made in a comment. 1. The id-keyed tombstone was NOT collision-free, as claimed. The argument was "the agent id is unique per workspace, so the name cannot collide". That holds only if nothing else can occupy the namespace, and agent names are validated as `z.string().min(1)` — arbitrary strings. A caller could pre-register `#released-`, and the victim's release would then hit `UNIQUE(workspace_id, name)` inside the atomic batch and abort the whole unit: exactly the failure the tombstone exists to avoid, reachable on demand. Fixed at the root rather than by adding entropy: `#released-` is now a reserved marker rejected on both registration paths (`registerAgent` and `registerAgentViaNode`), which is what makes the id-keyed name actually collision-free. Production has zero existing names containing the marker, so nothing is grandfathered out. 2. `/v1/agents/presence` still listed released tombstones. `listAgents` was filtered but `getPresence` runs its own query, so releasing a name made it reappear as a permanently offline agent instead of disappearing. Filtering one roster surface and not the other is worse than filtering neither. 3. The local reap discarded the caller's release reason, hardcoding `reason: 'released'`, while the dispatched path records the supplied one. Now uses the same `release: { reason, released_at, previous_name }` shape, so an audit does not have to know which path released the agent. 4. The reclaim guard's comment overclaimed. It said reads do not move `last_seen`; the sweep does write it, clamping a FUTURE timestamp back to the server clock. The security conclusion is unchanged — the clamp writes `now`, and the gate needs `now - AGENT_RECLAIM_GRACE_MS`, so a read still cannot make a row claimable — but "reads never touch this column" was false, and a security-sensitive comment that overstates its guarantee is how the next person justifies a change it does not actually cover. Each fix has a test that fails without it. --- .../conformance/agentLifecycle.test.ts | 77 +++++++++++++++++++ packages/engine/src/engine/action.ts | 8 +- packages/engine/src/engine/agent.ts | 42 ++++++++-- packages/engine/src/engine/node.ts | 16 +++- packages/engine/src/engine/presence.ts | 7 +- 5 files changed, 139 insertions(+), 11 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts index 355201b8..1464858d 100644 --- a/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts +++ b/packages/engine/src/__tests__/conformance/agentLifecycle.test.ts @@ -303,6 +303,83 @@ describe('agent presence and release lifecycle', () => { expect(invocation.status).toBe('completed'); }); + it('refuses to register into the reserved released-agent namespace', async () => { + const ws = await createWorkspace(stack.app, 'reserved-tombstone-namespace'); + // The tombstone name is only collision-free while nothing else can occupy + // that namespace. Agent names are otherwise arbitrary strings, so without + // this guard a caller could pre-register `#released-` + // and make the victim's release abort the whole atomic unit — the exact + // failure the tombstone exists to avoid. + const squatted = await stack.app.request('/v1/agents', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${ws.workspaceKey}` }, + body: JSON.stringify({ name: 'victim#released-12345' }), + }); + expect(squatted.status).toBe(400); + expect((await squatted.json() as { error: { code: string } }).error.code).toBe('invalid_agent_name'); + }); + + it('keeps released tombstones out of the roster and the presence view', async () => { + const ws = await createWorkspace(stack.app, 'tombstone-not-a-roster-member'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'ghost-agent'); + await stack.runtime.deps.db + .update(agents) + .set({ locationType: 'self_connected', locationNodeId: null }) + .where(eq(agents.id, target.agentId)); + + const released = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${ws.workspaceKey}` }, + body: JSON.stringify({ name: target.name, delete_agent: true }), + }); + expect(released.status).toBe(201); + + // A tombstone is retained only so history stays attributable. Every + // consumer that answers "who is in this workspace" must exclude it — + // otherwise releasing a name makes it look like a second, permanently + // offline agent rather than making it disappear. + const roster = await stack.app.request('/v1/agents', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + const rosterNames = (await roster.json() as { data: Array<{ name: string }> }).data.map((a) => a.name); + expect(rosterNames).not.toContain(target.name); + expect(rosterNames.some((n) => n.includes('#released-'))).toBe(false); + + const presence = await stack.app.request('/v1/agents/presence', { + headers: { authorization: `Bearer ${ws.workspaceKey}` }, + }); + expect(presence.status).toBe(200); + const presenceNames = (await presence.json() as { data: Array<{ agent_name: string }> }) + .data.map((p) => p.agent_name); + expect(presenceNames).not.toContain(target.name); + expect(presenceNames.some((n) => n.includes('#released-'))).toBe(false); + }); + + it('records the caller-supplied release reason on the tombstone', async () => { + const ws = await createWorkspace(stack.app, 'tombstone-release-reason'); + const target = await registerAgent(stack.app, ws.workspaceKey, 'audited-agent'); + await stack.runtime.deps.db + .update(agents) + .set({ locationType: 'self_connected', locationNodeId: null }) + .where(eq(agents.id, target.agentId)); + + const response = await stack.app.request('/v1/agents/release', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${ws.workspaceKey}` }, + body: JSON.stringify({ name: target.name, delete_agent: true, reason: 'node decommissioned' }), + }); + expect(response.status).toBe(201); + + const [tombstone] = await stack.runtime.deps.db + .select({ metadata: agents.metadata }) + .from(agents) + .where(eq(agents.id, target.agentId)); + // Same `release` shape the dispatched path writes, so an audit does not + // have to know which path released the agent. + expect((tombstone.metadata as { release?: Record }).release) + .toMatchObject({ reason: 'node decommissioned', previous_name: target.name }); + }); + it('releases capacity from the binding that local reaping deactivates', async () => { const ws = await createWorkspace(stack.app, 'hostless-agent-binding-capacity'); const target = await registerAgent(stack.app, ws.workspaceKey, 'reap-bound-agent'); diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index f1a0b84b..dad8982e 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -764,8 +764,14 @@ async function dispatchRelease(args: { // NOT NULL UNIQUE and cannot be cleared, so rotate it to a value // nobody holds; the released agent's old token stops authenticating. tokenHash: releasedTokenHash, + // Same `release` shape the dispatched path writes, so an audit does + // not have to know which path released the agent. metadata: sql`json_patch(COALESCE(${agents.metadata}, '{}'), ${JSON.stringify({ - release: { reason: 'released', releasedAt: completedAt.toISOString(), previousName: agent.name }, + release: { + reason: typeof input.reason === 'string' ? input.reason : null, + released_at: completedAt.toISOString(), + previous_name: agent.name, + }, })})`, }) .where(and( diff --git a/packages/engine/src/engine/agent.ts b/packages/engine/src/engine/agent.ts index d6dfb62d..77de087b 100644 --- a/packages/engine/src/engine/agent.ts +++ b/packages/engine/src/engine/agent.ts @@ -48,13 +48,44 @@ type AgentPresenceRow = Pick; export const RELEASED_AGENT_STATUS = 'released'; /** - * Tombstone name for a released agent. Keyed on the agent id rather than a - * timestamp so it is unique by construction and idempotent on repeat release — - * the reap runs inside an atomic batch where a unique-constraint violation - * would abort the whole unit rather than fail just this statement. + * Marker separating a released agent's original name from its tombstone + * suffix. Reserved: registration rejects it, which is what makes + * {@link releasedAgentName} collision-free (see {@link assertRegistrableAgentName}). + */ +export const RELEASED_NAME_MARKER = '#released-'; + +/** + * Tombstone name for a released agent. + * + * Keyed on the agent id, not a timestamp: the release runs inside an atomic + * batch where a `UNIQUE(workspace_id, name)` violation aborts the WHOLE unit + * rather than failing just this statement — which is the exact defect the + * tombstone exists to avoid. A timestamped name can collide; an id-keyed one + * cannot, because the id is already unique per workspace. + * + * That argument only holds while no ordinary agent can occupy this namespace. + * Agent names are otherwise arbitrary strings, so without the registration + * guard a caller could pre-register `#released-` and make the + * victim's release abort. Hence the reserved marker. */ export function releasedAgentName(name: string, agentId: string): string { - return `${name}#released-${agentId}`; + return `${name}${RELEASED_NAME_MARKER}${agentId}`; +} + +/** + * Reject names that would land in the reserved tombstone namespace. + * + * Called on every registration path. Without it the release path is not + * atomic-safe: see {@link releasedAgentName}. + */ +export function assertRegistrableAgentName(name: string): void { + if (name.includes(RELEASED_NAME_MARKER)) { + throw codedError( + `Agent name "${name}" is reserved: "${RELEASED_NAME_MARKER}" marks a released agent`, + 'invalid_agent_name', + 400, + ); + } } /** @@ -102,6 +133,7 @@ export async function registerAgent( capabilities?: Record; }, ) { + assertRegistrableAgentName(data.name); const agentId = generateId(); const token = `at_live_${randomHex(16)}`; const tokenHash = await sha256Hex(token); diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index eb4ed75a..eae4b6fd 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -22,7 +22,7 @@ import { runAtomic } from '../ports/database.js'; import type { EngineDb } from '../ports/database.js'; import { isProviderAgentDeliveryReady, type NodeConnectionRegistry } from '../ports/realtime.js'; import { generateId } from './snowflake.js'; -import { AGENT_RECLAIM_GRACE_MS } from './agent.js'; +import { AGENT_RECLAIM_GRACE_MS, assertRegistrableAgentName } from './agent.js'; import { isNodeLive, nodeHasCapability } from './placement.js'; import { DEFAULT_PROVIDER_NAME, @@ -1124,6 +1124,7 @@ export async function registerAgentViaNode( message: FleetAgentRegisterMessage, options: { deliveryCursorSupported?: boolean } = {}, ): Promise { + assertRegistrableAgentName(message.name); return runAtomic(db, async (tx) => { const [node] = await tx .select() @@ -1192,8 +1193,17 @@ export async function registerAgentViaNode( // `agent list` flipped records to 'offline' and thereby moved them // from "reclaimable only by their own node" to "reclaimable by any // node, on name alone, with a `token_hash` overwrite". A read must - // never widen who may claim an identity; reads do not move - // `last_seen`, so gating here makes that structurally impossible. + // never widen who may claim an identity. + // + // Precisely: the sweep's status update cannot affect this gate at all + // any more. The sweep does also clamp a FUTURE `last_seen` back to the + // server clock, which is a write — but it can only move the reclaim + // moment later-or-equal relative to that bogus timestamp, never make a + // row claimable now, since the clamped value is `now` and this + // predicate needs `now - AGENT_RECLAIM_GRACE_MS`. That clamp exists so + // a client with a skewed clock cannot make its name permanently + // unreclaimable; it is deliberate, and it is the only path by which a + // read touches this column. // // The grace window is far longer than the presence TTL: an agent goes // 'offline' on the roster after 5 minutes of silence, but its name is diff --git a/packages/engine/src/engine/presence.ts b/packages/engine/src/engine/presence.ts index 1be5b5b4..b6bc3e2b 100644 --- a/packages/engine/src/engine/presence.ts +++ b/packages/engine/src/engine/presence.ts @@ -1,7 +1,8 @@ -import { eq } from 'drizzle-orm'; +import { and, eq, ne } from 'drizzle-orm'; import { agents } from '../db/schema.js'; import type { getDb } from '../db/index.js'; import type { PresenceTracker } from '../ports/presence.js'; +import { RELEASED_AGENT_STATUS } from './agent.js'; /** * Get presence status for all agents in a workspace. @@ -16,7 +17,9 @@ export async function getPresence( db .select({ id: agents.id, name: agents.name }) .from(agents) - .where(eq(agents.workspaceId, workspaceId)), + // Released rows are tombstones kept only so history stays attributable. + // They are not roster members and must not appear as presence entries. + .where(and(eq(agents.workspaceId, workspaceId), ne(agents.status, RELEASED_AGENT_STATUS))), presence.getOnline(workspaceId), ]);