From c968cb2820a1012366308478f4757629e3ea29d2 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 04:33:05 +0000 Subject: [PATCH 1/3] feat(engine): emit durable agent.exited and node.status events Agent exit on a node and node death were only observable by polling. Emit durable events across the same three channels as invocation completion so they are push-visible: - agent.exited on deregisterAgentViaNode (control frame + handleAgentDisconnect), the reconcileInventory missing-agent sweep, and release completion. Carries agent_id/agent_name/node_id, correlated spawn invocation_id, and a reason (deregistered | missing_from_inventory | released). The spawn caller's mailbox is notified directly. - node.status.online/offline on liveness transitions: markNodeOffline (sweep, provider disconnect, deregister) and the register/heartbeat offline->online transition. New emitAgentExitedEffects / emitNodeStatusEffects helpers reuse InvocationCompletionDeps and are best-effort (never fail the state transition). Event taxonomy updated in @relaycast/types (schemas + subscription enum) and the python SDK ServerEventType mirror. Wildcard webhooks receive them automatically. Closes #273 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JQ8fa196XtLnvctQMQuUDx --- .../2026-07/traj_wulwzxphvrdc.trace.json | 461 ++++++++++++++++++ .../2026-07/traj_wulwzxphvrdc/summary.md | 21 + .../2026-07/traj_wulwzxphvrdc/trajectory.json | 48 ++ CHANGELOG.md | 2 + README.md | 19 +- openapi.yaml | 4 +- packages/engine/CHANGELOG.md | 2 + .../conformance/agentNodeExitEvents.test.ts | 174 +++++++ packages/engine/src/adapters/node/index.ts | 2 +- packages/engine/src/adapters/node/realtime.ts | 4 +- packages/engine/src/agent-disconnect.ts | 6 + .../__tests__/agentExitedEffects.test.ts | 169 +++++++ packages/engine/src/engine/action.ts | 60 ++- .../engine/src/engine/invocationCompletion.ts | 167 ++++++- packages/engine/src/engine/node.ts | 108 +++- packages/engine/src/routes/node.ts | 3 +- packages/engine/src/routes/presence.ts | 5 +- packages/sdk-python/src/relay_sdk/models.py | 3 + packages/types/CHANGELOG.md | 6 +- packages/types/src/events.ts | 34 ++ packages/types/src/subscription.ts | 3 + 21 files changed, 1249 insertions(+), 52 deletions(-) create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json create mode 100644 packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts create mode 100644 packages/engine/src/engine/__tests__/agentExitedEffects.test.ts diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json new file mode 100644 index 00000000..84586683 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json @@ -0,0 +1,461 @@ +{ + "version": "1.0.0", + "id": "b2fa69e0-81d6-4e05-9352-c33aa33b5578", + "timestamp": "2026-07-16T04:34:18.552Z", + "trajectory": "traj_wulwzxphvrdc", + "files": [ + { + "path": ".agentworkforce/trajectories/active/traj_wulwzxphvrdc/trajectory.json", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 1, + "end_line": 19, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "CHANGELOG.md", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 21, + "end_line": 28, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "README.md", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 403, + "end_line": 424, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "openapi.yaml", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 4275, + "end_line": 4283, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/CHANGELOG.md", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 10, + "end_line": 17, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 1, + "end_line": 174, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/adapters/node/index.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 164, + "end_line": 170, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/adapters/node/realtime.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 529, + "end_line": 541, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/agent-disconnect.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 2, + "end_line": 21, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 39, + "end_line": 45, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/engine/__tests__/agentExitedEffects.test.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 1, + "end_line": 169, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/engine/action.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 7, + "end_line": 13, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 843, + "end_line": 849, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 877, + "end_line": 886, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 890, + "end_line": 927, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1509, + "end_line": 1515, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1587, + "end_line": 1593, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/engine/invocationCompletion.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 1, + "end_line": 8, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 71, + "end_line": 246, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/engine/node.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 40, + "end_line": 47, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 473, + "end_line": 488, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 512, + "end_line": 526, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 535, + "end_line": 548, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 570, + "end_line": 576, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 588, + "end_line": 594, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 598, + "end_line": 608, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 615, + "end_line": 621, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1170, + "end_line": 1176, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1209, + "end_line": 1224, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1369, + "end_line": 1375, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1379, + "end_line": 1385, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1388, + "end_line": 1411, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1458, + "end_line": 1492, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1553, + "end_line": 1559, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1610, + "end_line": 1629, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1694, + "end_line": 1700, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 1756, + "end_line": 1762, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/routes/node.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 343, + "end_line": 351, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/engine/src/routes/presence.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 49, + "end_line": 58, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/sdk-python/src/relay_sdk/models.py", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 970, + "end_line": 978, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/types/CHANGELOG.md", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 7, + "end_line": 17, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/types/src/events.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 136, + "end_line": 169, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 388, + "end_line": 396, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + }, + { + "start_line": 439, + "end_line": 447, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + }, + { + "path": "packages/types/src/subscription.ts", + "conversations": [ + { + "contributor": { + "type": "ai" + }, + "ranges": [ + { + "start_line": 11, + "end_line": 19, + "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + } + ] + } + ] + } + ] +} \ No newline at end of file diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md new file mode 100644 index 00000000..fcc62ad0 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md @@ -0,0 +1,21 @@ +# Trajectory: Emit durable agent.exited and node.status events (issue #273) + +> **Status:** ✅ Completed +> **Confidence:** 85% +> **Started:** July 16, 2026 at 04:15 AM +> **Completed:** July 16, 2026 at 04:34 AM + +--- + +## Summary + +Emitted durable agent.exited (deregister / inventory-missing / release) and node.status.online/offline events through the invocation-completion fanout (workspace event log, webhook outbox, spawn caller mailbox); threaded optional completion deps through node/action transition sites + +**Approach:** Standard approach + +--- + +## Artifacts + +**Commits:** 625a610 +**Files changed:** 19 diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json new file mode 100644 index 00000000..9aedf0d2 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json @@ -0,0 +1,48 @@ +{ + "id": "traj_wulwzxphvrdc", + "version": 1, + "task": { + "title": "Emit durable agent.exited and node.status events (issue #273)" + }, + "status": "completed", + "startedAt": "2026-07-16T04:15:43.951Z", + "completedAt": "2026-07-16T04:34:18.521Z", + "agents": [], + "chapters": [], + "retrospective": { + "summary": "Emitted durable agent.exited (deregister / inventory-missing / release) and node.status.online/offline events through the invocation-completion fanout (workspace event log, webhook outbox, spawn caller mailbox); threaded optional completion deps through node/action transition sites", + "approach": "Standard approach", + "confidence": 0.85 + }, + "commits": [ + "625a610" + ], + "filesChanged": [ + ".agentworkforce/trajectories/active/traj_wulwzxphvrdc/trajectory.json", + "CHANGELOG.md", + "README.md", + "openapi.yaml", + "packages/engine/CHANGELOG.md", + "packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts", + "packages/engine/src/adapters/node/index.ts", + "packages/engine/src/adapters/node/realtime.ts", + "packages/engine/src/agent-disconnect.ts", + "packages/engine/src/engine/__tests__/agentExitedEffects.test.ts", + "packages/engine/src/engine/action.ts", + "packages/engine/src/engine/invocationCompletion.ts", + "packages/engine/src/engine/node.ts", + "packages/engine/src/routes/node.ts", + "packages/engine/src/routes/presence.ts", + "packages/sdk-python/src/relay_sdk/models.py", + "packages/types/CHANGELOG.md", + "packages/types/src/events.ts", + "packages/types/src/subscription.ts" + ], + "projectId": "git/AgentWorkforce/relaycast", + "tags": [], + "_trace": { + "startRef": "2e2b7a2b0719db1057555e151ae326154903e685", + "endRef": "625a6100becac9779c5feb824d01f568d1e1fdea", + "traceId": "b2fa69e0-81d6-4e05-9352-c33aa33b5578" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index b7e69b52..e270a2aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,8 @@ Packages without a separate changelog are covered by the cross-package notes bel ### Added - Message retention remains opt-in (history is kept forever by default); self-host deployments can now opt in to a deployment-wide message TTL via `RELAYCAST_MESSAGE_TTL_DAYS`. +- Durable `agent.exited` event when a node-hosted agent leaves (deregister, missing from an inventory sync, or release), carrying `agent_id`, `agent_name`, `node_id`, the spawn `invocation_id`, and a `reason`; the spawn's caller is notified directly. +- Durable `node.status.online` / `node.status.offline` events on node liveness transitions (offline carries a `reason` like `liveness_timeout`). Wildcard webhook subscriptions (`events: ["*"]`) receive all three new events automatically. ### Fixed diff --git a/README.md b/README.md index 38e95c17..7a422eb2 100644 --- a/README.md +++ b/README.md @@ -403,13 +403,22 @@ and outbound subscriptions: `message.created`, `message.reacted`, `message.read` `delivery.accepted`, `delivery.delivered`, `delivery.deferred`, `delivery.failed`, `agent.status.changed`, `agent.status.active`, `agent.status.idle`, `agent.status.blocked`, `agent.status.waiting`, `agent.status.offline`, +`agent.exited`, `node.status.online`, `node.status.offline`, `action.invoked`, `action.completed`, `action.failed`, and `action.denied`. -Fleet node presence is published to workspace-key observer streams as -`node.online`, `node.heartbeat`, and `node.offline`. Each carries a `node` -payload matching the `GET /nodes` roster entry (capabilities, tags, `load`, -`active_agents`/`max_agents`, `handlers_live`, `last_heartbeat_at`), so a single -event fully refreshes a node's row. +`agent.exited` is a durable record that an agent hosted by a node left — via +deregister, an inventory sync that dropped it, or a release — carrying +`agent_id`, `agent_name`, `node_id`, the correlated spawn `invocation_id`, and a +`reason` (`deregistered` | `missing_from_inventory` | `released`); the spawn's +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`). + +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, +tags, `load`, `active_agents`/`max_agents`, `handlers_live`, +`last_heartbeat_at`), so a single event fully refreshes a node's row. Nodes are first-class delivery hosts and every agent has a node route. `kind` describes transport (`ws`, `http_push`, or `poll`), `role` describes ownership diff --git a/openapi.yaml b/openapi.yaml index 9ea33ce6..89fc48bd 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -4275,7 +4275,9 @@ paths: post: summary: Create outbound event subscription description: | - Subscribe an external webhook endpoint to workspace events. Delivery payloads are JSON: + Subscribe an external webhook endpoint to workspace events. `events` is a free-form array of + dotted event names (or `["*"]` to receive every event, including new ones such as + `agent.exited` and `node.status.online` / `node.status.offline`). Delivery payloads are JSON: `{ type, workspace_id, timestamp, data }`. If `secret` is provided, Relaycast signs the exact JSON request body with HMAC-SHA256 and sends `X-Relay-Signature: sha256=` along with `X-Relay-Event` and `X-Relay-Timestamp`. Custom `headers` are included on delivery; Relay's diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index 85f155d5..d41cde5a 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -10,6 +10,8 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased - Minor] ### Added +- Emit durable `agent.exited` on every server-observable agent exit — `deregisterAgentViaNode` (deregister control frame + `handleAgentDisconnect`), the `reconcileInventory` missing-agent sweep, and release completion (`completeNodeInvocation`) — fanned out via the new `emitAgentExitedEffects` helper to the spawn caller's mailbox, the durable workspace log, and the webhook outbox. +- Emit durable `node.status.online` / `node.status.offline` on node liveness transitions via `emitNodeStatusEffects`: `markNodeOffline` (liveness sweep, provider disconnect, deregister) and the register/heartbeat offline→online transition. Both helpers reuse `InvocationCompletionDeps` and are best-effort (never fail the underlying state transition). - Message retention remains opt-in (`pruneExpired` still defaults `messageTtlDays` to `null`). Self-host can now opt in to a deployment-wide message TTL: `startServer` accepts `eventQueue` (`DurableEventQueueOptions`, including `retention`), and the `relaycast-engine` CLI exposes `RELAYCAST_MESSAGE_TTL_DAYS` (positive = prune after N days; unset or `0`/negative = keep forever). - Exported the provider-attach arbitration policy from `@relaycast/engine/node-control`: `providerAttachDecision()` plus `PROVIDER_ATTACH_LIVENESS_MS`, so an out-of-process socket owner (a hosted NodeDO) mirrors the spec §3.1 decision from one source of truth instead of hand-copying the constant and logic. diff --git a/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts b/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts new file mode 100644 index 00000000..883671d2 --- /dev/null +++ b/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts @@ -0,0 +1,174 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { and, eq } from 'drizzle-orm'; +import { + makeNodeStack, + createWorkspace, + FakeSocket, + type TestStack, +} from './harness.js'; +import { actionInvocations, nodes, workspaceEvents } from '../../db/schema.js'; +import { sweepOfflineNodes } from '../../engine/node.js'; +import { NODE_LIVENESS_TTL_MS } from '../../engine/placement.js'; + +/** + * Durable observability for agent exit + node liveness (issue #273): every + * exit / node transition lands in the durable workspace event log AND the + * webhook outbox, so it is push-visible instead of poll-only. + * + * These assert the durable `workspace_events` log (never drained). The webhook + * outbox (`pending_events`) is proven by the helper unit tests instead — the + * Node adapter's queue polls and deletes subscriber-less outbox rows right after + * `send`, so a row-count assertion here would be racy. + */ +describe('agent.exited + node.status durable events', () => { + let stack: TestStack; + beforeEach(() => { stack = makeNodeStack({ ttlMs: 60_000 }); }); + afterEach(() => stack.close()); + + async function enrollNode(workspaceKey: string, nodeId: string, name: string) { + const res = await stack.app.request('/v1/nodes', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${workspaceKey}` }, + body: JSON.stringify({ node_id: nodeId, name, role: 'broker', capabilities: ['spawn:claude'], max_agents: 4, tags: ['test'], version: 'v0' }), + }); + expect(res.status).toBe(201); + } + + function attach(workspaceId: string, nodeId: string) { + const sock = new FakeSocket(); + const handle = stack.runtime.realtime.attachNodeSocket(workspaceId, nodeId, sock); + return { sock, handle }; + } + + async function bringNodeOnline(ws: { workspaceKey: string; workspaceId: string }, nodeId: string, name: string) { + await enrollNode(ws.workspaceKey, nodeId, name); + const { sock, handle } = attach(ws.workspaceId, nodeId); + 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 registerAgentViaNode(handle: { handleMessage(raw: string): Promise }, sock: FakeSocket, name: string) { + await handle.handleMessage(JSON.stringify({ v: 1, type: 'agent.register', name, resumable: true })); + const reply = sock.ofType('reply').at(-1) as { ok: boolean; data: { agent_id: string; name?: string } }; + expect(reply?.ok).toBe(true); + return reply.data.agent_id; + } + + const db = () => stack.runtime.deps.db; + + async function workspaceEventsOfType(workspaceId: string, type: string) { + return db().select().from(workspaceEvents) + .where(and(eq(workspaceEvents.workspaceId, workspaceId), eq(workspaceEvents.type, type))); + } + + it('(a) agent.deregister control frame emits a durable agent.exited', async () => { + const ws = await createWorkspace(stack.app, 'exit-a'); + const { sock, handle } = await bringNodeOnline(ws, 'node_a', 'alpha'); + const agentId = await registerAgentViaNode(handle, sock, 'worker-a'); + + await handle.handleMessage(JSON.stringify({ v: 1, type: 'agent.deregister', agent_id: agentId })); + + const logged = await workspaceEventsOfType(ws.workspaceId, 'agent.exited'); + expect(logged).toHaveLength(1); + expect(JSON.parse(logged[0]!.payload)).toMatchObject({ + type: 'agent.exited', agent_id: agentId, agent_name: 'worker-a', node_id: 'node_a', reason: 'deregistered', + }); + }); + + it('(b) inventory.sync with a missing agent emits agent.exited', async () => { + const ws = await createWorkspace(stack.app, 'exit-b'); + const { sock, handle } = await bringNodeOnline(ws, 'node_b', 'beta'); + const keepId = await registerAgentViaNode(handle, sock, 'keep'); + const dropId = await registerAgentViaNode(handle, sock, 'drop'); + + // Inventory lists only "keep"; "drop" is missing and must be flipped + emitted. + await handle.handleMessage(JSON.stringify({ + v: 1, type: 'inventory.sync', agents: [{ agent_id: keepId, name: 'keep' }], + })); + + const logged = await workspaceEventsOfType(ws.workspaceId, 'agent.exited'); + expect(logged).toHaveLength(1); + expect(JSON.parse(logged[0]!.payload)).toMatchObject({ + agent_id: dropId, agent_name: 'drop', node_id: 'node_b', reason: 'missing_from_inventory', + }); + }); + + it('(c) release completion emits agent.exited', async () => { + const ws = await createWorkspace(stack.app, 'exit-c'); + const { sock, handle } = await bringNodeOnline(ws, 'node_c', 'gamma'); + const agentId = await registerAgentViaNode(handle, sock, 'releasee'); + + await db().insert(actionInvocations).values({ + id: 'inv_release_c', + workspaceId: ws.workspaceId, + actionName: 'release', + input: { name: 'releasee' }, + status: 'dispatched', + dispatchedNodeId: 'node_c', + dispatchedProvider: 'default', + dispatchedAt: new Date(), + }); + + await handle.handleMessage(JSON.stringify({ + v: 1, type: 'action.result', invocation_id: 'inv_release_c', output: {}, + })); + + const logged = await workspaceEventsOfType(ws.workspaceId, 'agent.exited'); + expect(logged).toHaveLength(1); + expect(JSON.parse(logged[0]!.payload)).toMatchObject({ + agent_id: agentId, agent_name: 'releasee', node_id: 'node_c', reason: 'released', + }); + }); + + it('(d) node liveness sweep emits node.status.offline', async () => { + const ws = await createWorkspace(stack.app, 'exit-d'); + await bringNodeOnline(ws, 'node_d', 'delta'); + + // Age the heartbeat past the liveness TTL so the sweep flips it offline. + await db().update(nodes) + .set({ lastHeartbeatAt: new Date(Date.now() - NODE_LIVENESS_TTL_MS - 5_000) }) + .where(and(eq(nodes.workspaceId, ws.workspaceId), eq(nodes.id, 'node_d'))); + + const swept = await sweepOfflineNodes(db(), stack.runtime.realtime, stack.runtime.deps); + expect(swept).toBeGreaterThanOrEqual(1); + + const logged = await workspaceEventsOfType(ws.workspaceId, 'node.status.offline'); + expect(logged).toHaveLength(1); + expect(JSON.parse(logged[0]!.payload)).toMatchObject({ + type: 'node.status.offline', node_id: 'node_d', node_name: 'delta', reason: 'liveness_timeout', + }); + }); + + it('(e) re-heartbeat after offline emits node.status.online exactly once', async () => { + const ws = await createWorkspace(stack.app, 'exit-e'); + // Initial register already produces the first offline->online transition. + await bringNodeOnline(ws, 'node_e', 'epsilon'); + expect(await workspaceEventsOfType(ws.workspaceId, 'node.status.online')).toHaveLength(1); + + // Force offline via the sweep. + await db().update(nodes) + .set({ lastHeartbeatAt: new Date(Date.now() - NODE_LIVENESS_TTL_MS - 5_000) }) + .where(and(eq(nodes.workspaceId, ws.workspaceId), eq(nodes.id, 'node_e'))); + await sweepOfflineNodes(db(), stack.runtime.realtime, stack.runtime.deps); + + // Reconnect: a fresh register + two heartbeats. Exactly one online transition. + const { handle } = attach(ws.workspaceId, 'node_e'); + await handle.handleMessage(JSON.stringify({ + v: 1, type: 'node.register', name: 'epsilon', node_id: 'node_e', + 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 })); + await handle.handleMessage(JSON.stringify({ v: 1, type: 'node.heartbeat', load: 0, active_agents: 0, handlers_live: true })); + + // One from initial register + one from the reconnect = two total, not more. + expect(await workspaceEventsOfType(ws.workspaceId, 'node.status.online')).toHaveLength(2); + }); +}); diff --git a/packages/engine/src/adapters/node/index.ts b/packages/engine/src/adapters/node/index.ts index d349dfaf..85f8e8b7 100644 --- a/packages/engine/src/adapters/node/index.ts +++ b/packages/engine/src/adapters/node/index.ts @@ -164,7 +164,7 @@ export function createNodeRuntime(options: NodeRuntimeOptions): NodeRuntime { const runDeliveryMaintenance = createDeliveryMaintenanceRunner(deps); const sweepTimer = setInterval(() => { - void sweepOfflineNodes(db, realtime).catch(() => {}); + void sweepOfflineNodes(db, realtime, deps).catch(() => {}); void sweepTimedOutInvocations(db, realtime).catch(() => {}); void runDeliveryMaintenance(); }, 15_000); diff --git a/packages/engine/src/adapters/node/realtime.ts b/packages/engine/src/adapters/node/realtime.ts index 20eacc05..92540dbc 100644 --- a/packages/engine/src/adapters/node/realtime.ts +++ b/packages/engine/src/adapters/node/realtime.ts @@ -529,13 +529,13 @@ export class InProcessRealtime implements RealtimeBus, ConnectionRegistry, NodeC // would keep stale capabilities with no signal. Log it; a close still must // not throw. if (providerName) { - await handleProviderDisconnect(this.db, this, workspaceId, nodeId, providerName, hasRemaining).catch((err) => { + await handleProviderDisconnect(this.db, this, workspaceId, nodeId, providerName, hasRemaining, this.nodeCompletionDeps).catch((err) => { // Pass the error object so the runtime logs its stack, not just the message. console.error('[node.teardown] provider disconnect failed', { workspace_id: workspaceId, node_id: nodeId, provider: providerName }, err); }); } else if (!hasRemaining) { // Connection dropped before it bound a provider and it was the node's last. - await markNodeOffline(this.db, this, workspaceId, nodeId).catch((err) => { + await markNodeOffline(this.db, this, workspaceId, nodeId, { deps: this.nodeCompletionDeps, reason: 'disconnected' }).catch((err) => { console.error('[node.teardown] mark node offline failed', { workspace_id: workspaceId, node_id: nodeId }, err); }); } diff --git a/packages/engine/src/agent-disconnect.ts b/packages/engine/src/agent-disconnect.ts index f04231dd..cc9cdd36 100644 --- a/packages/engine/src/agent-disconnect.ts +++ b/packages/engine/src/agent-disconnect.ts @@ -2,15 +2,20 @@ import { and, eq } from 'drizzle-orm'; import type { EngineDb } from './ports/index.js'; import { agents } from './db/schema.js'; import { deregisterAgentViaNode, directNodeIdForAgent } from './engine/node.js'; +import type { InvocationCompletionDeps } from './engine/invocationCompletion.js'; /** * Disconnect an agent currently hosted by an explicit node when the caller has * no node-control socket frame to pass through `handleNodeControlMessage`. + * + * `deps` are optional so this stays infallible for callers without the fanout + * ports; when provided, the deregister emits a durable `agent.exited`. */ export async function handleAgentDisconnect( db: EngineDb, workspaceId: string, agentId: string, + deps?: InvocationCompletionDeps, ): Promise { const [agent] = await db .select({ @@ -34,6 +39,7 @@ export async function handleAgentDisconnect( workspaceId, agent.locationNodeId, { agent_id: agentId }, + deps, ); return disconnected !== null; } diff --git a/packages/engine/src/engine/__tests__/agentExitedEffects.test.ts b/packages/engine/src/engine/__tests__/agentExitedEffects.test.ts new file mode 100644 index 00000000..c5637106 --- /dev/null +++ b/packages/engine/src/engine/__tests__/agentExitedEffects.test.ts @@ -0,0 +1,169 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + emitAgentExitedEffects, + emitNodeStatusEffects, + type InvocationCompletionDeps, +} from '../invocationCompletion.js'; +import { sendNodeDeliveriesToAgents } from '../nodeDeliver.js'; + +vi.mock('../nodeDeliver.js', () => ({ + sendNodeDeliveriesToAgents: vi.fn(async () => {}), +})); + +/** + * Build a deps stub whose `db` resolves a caller lookup to `callerId` and lets + * the durable outbox insert succeed/fail on demand. `select().from().where()` + * returns the invocation caller row; `insert()` throws when `outboxThrows`. + */ +function makeDeps(opts: { callerId?: string | null; outboxThrows?: boolean } = {}) { + const publishToWorkspaceStream = vi.fn(async () => {}); + const webhookSend = vi.fn(async () => {}); + const db = { + select: () => ({ + from: () => ({ + where: async () => (opts.callerId === undefined ? [] : [{ callerId: opts.callerId }]), + }), + }), + insert: () => { + if (opts.outboxThrows) throw new Error('skip durable outbox in test'); + return { values: async () => {} }; + }, + }; + const deps = { + db, + realtime: { publishToWorkspaceStream }, + nodeConnections: {}, + webhookQueue: { send: webhookSend }, + config: {}, + } as unknown as InvocationCompletionDeps; + return { deps, publishToWorkspaceStream, webhookSend }; +} + +describe('emitAgentExitedEffects', () => { + it('fans out to all three channels and delivers to the spawn caller exactly once', async () => { + const sendMock = vi.mocked(sendNodeDeliveriesToAgents); + sendMock.mockClear(); + const { deps, publishToWorkspaceStream, webhookSend } = makeDeps({ callerId: 'caller-1' }); + + await emitAgentExitedEffects(deps, 'ws-1', { + agentId: 'agent-9', + agentName: 'worker', + nodeId: 'node-3', + invocationId: 'inv-7', + reason: 'released', + }); + + // (1) caller mailbox + expect(sendMock).toHaveBeenCalledTimes(1); + expect(sendMock.mock.calls[0]?.[1]).toMatchObject({ + agentIds: ['caller-1'], + event: 'agent.exited', + eventKey: 'agent-9_inv-7_released', + data: { + agent_id: 'agent-9', + agent_name: 'worker', + node_id: 'node-3', + invocation_id: 'inv-7', + reason: 'released', + }, + }); + // (2) durable log + stream + expect(publishToWorkspaceStream).toHaveBeenCalledTimes(1); + const published = publishToWorkspaceStream.mock.calls[0]?.[0] as { event: Record }; + expect(published.event).toMatchObject({ type: 'agent.exited', agent_id: 'agent-9', reason: 'released' }); + // (3) webhook outbox + expect(webhookSend).toHaveBeenCalledTimes(1); + expect(webhookSend.mock.calls[0]?.[0]).toMatchObject({ type: 'agent.exited', workspaceId: 'ws-1' }); + }); + + it('skips the caller mailbox when there is no correlated invocation', async () => { + const sendMock = vi.mocked(sendNodeDeliveriesToAgents); + sendMock.mockClear(); + const { deps, publishToWorkspaceStream, webhookSend } = makeDeps(); + + await emitAgentExitedEffects(deps, 'ws-1', { + agentId: 'agent-9', + agentName: 'worker', + nodeId: 'node-3', + invocationId: null, + reason: 'deregistered', + }); + + expect(sendMock).not.toHaveBeenCalled(); + expect(publishToWorkspaceStream).toHaveBeenCalledTimes(1); + expect(webhookSend).toHaveBeenCalledTimes(1); + }); + + it('skips the caller mailbox when the invocation has no live caller', async () => { + const sendMock = vi.mocked(sendNodeDeliveriesToAgents); + sendMock.mockClear(); + const { deps, publishToWorkspaceStream } = makeDeps({ callerId: null }); + + await emitAgentExitedEffects(deps, 'ws-1', { + agentId: 'agent-9', + agentName: 'worker', + nodeId: null, + invocationId: 'inv-orphan', + reason: 'missing_from_inventory', + }); + + expect(sendMock).not.toHaveBeenCalled(); + expect(publishToWorkspaceStream).toHaveBeenCalledTimes(1); + }); + + it('still publishes and sends the webhook when the durable outbox insert fails', async () => { + const sendMock = vi.mocked(sendNodeDeliveriesToAgents); + sendMock.mockClear(); + const { deps, publishToWorkspaceStream, webhookSend } = makeDeps({ callerId: null, outboxThrows: true }); + + await emitAgentExitedEffects(deps, 'ws-1', { + agentId: 'agent-9', + agentName: 'worker', + nodeId: 'node-3', + invocationId: null, + reason: 'deregistered', + }); + + expect(publishToWorkspaceStream).toHaveBeenCalledTimes(1); + // Webhook send still attempted (without an outboxId) despite the insert failure. + expect(webhookSend).toHaveBeenCalledTimes(1); + expect(webhookSend.mock.calls[0]?.[0]).not.toHaveProperty('outboxId'); + }); +}); + +describe('emitNodeStatusEffects', () => { + it('publishes node.status.offline with a reason and enqueues the webhook', async () => { + const { deps, publishToWorkspaceStream, webhookSend } = makeDeps({ callerId: null, outboxThrows: true }); + + await emitNodeStatusEffects(deps, 'ws-1', { + status: 'offline', + nodeId: 'node-3', + nodeName: 'alpha', + reason: 'liveness_timeout', + }); + + expect(publishToWorkspaceStream).toHaveBeenCalledTimes(1); + const published = publishToWorkspaceStream.mock.calls[0]?.[0] as { event: Record }; + expect(published.event).toMatchObject({ + type: 'node.status.offline', + node_id: 'node-3', + node_name: 'alpha', + reason: 'liveness_timeout', + }); + expect(webhookSend).toHaveBeenCalledTimes(1); + }); + + it('publishes node.status.online without a reason field', async () => { + const { deps, publishToWorkspaceStream } = makeDeps({ callerId: null, outboxThrows: true }); + + await emitNodeStatusEffects(deps, 'ws-1', { + status: 'online', + nodeId: 'node-3', + nodeName: 'alpha', + }); + + const published = publishToWorkspaceStream.mock.calls[0]?.[0] as { event: Record }; + expect(published.event).toMatchObject({ type: 'node.status.online', node_id: 'node-3', node_name: 'alpha' }); + expect(published.event).not.toHaveProperty('reason'); + }); +}); diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index c6497e1b..7a6b7d95 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -7,6 +7,7 @@ import { toFleetWireJson } from './deliveryWire.js'; import type { NodeConnectionRegistry } from '../ports/realtime.js'; import { claimSpawnNode, chooseNodeForAction, isNodeLive, releaseNodeCapacity, reserveNodeCapacity } from './placement.js'; import { DEFAULT_PROVIDER_NAME, capacityProviderName, getProvider, isProviderLive } from './nodeProvider.js'; +import { emitAgentExitedEffects, fleetInvocationId, type InvocationCompletionDeps } from './invocationCompletion.js'; type Db = ReturnType; type ActionRow = typeof actions.$inferSelect; @@ -842,6 +843,7 @@ async function applyReleaseCompletionEffect( nodeId: string, invocation: Pick, data: { error?: string }, + deps?: InvocationCompletionDeps, ): Promise { if (!isReleaseInvocation(invocation.actionName) || data.error) return; @@ -875,6 +877,10 @@ async function applyReleaseCompletionEffect( .returning({ id: agentNodeBindings.id }); if (!deactivatedBinding) return; + // 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({ @@ -884,29 +890,38 @@ async function applyReleaseCompletionEffect( if (input.delete_agent === true) { await db.delete(agents).where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agent.id))); - return; + } else { + const existingMetadata = agent.metadata ?? {}; + const { spawn: _spawn, cli: _cli, ...restMetadata } = existingMetadata; + await db + .update(agents) + .set({ + status: 'offline', + // Clear the node location so the agent is no longer routable to the released + // node and a repeat release can't re-decrement the node's active count. + locationType: 'self_connected', + locationNodeId: null, + lastSeen: new Date(), + metadata: { + ...restMetadata, + release: { + reason: typeof input.reason === 'string' ? input.reason : null, + released_at: new Date().toISOString(), + }, + }, + }) + .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agent.id))); } - const existingMetadata = agent.metadata ?? {}; - const { spawn: _spawn, cli: _cli, ...restMetadata } = existingMetadata; - await db - .update(agents) - .set({ - status: 'offline', - // Clear the node location so the agent is no longer routable to the released - // node and a repeat release can't re-decrement the node's active count. - locationType: 'self_connected', - locationNodeId: null, - lastSeen: new Date(), - metadata: { - ...restMetadata, - release: { - reason: typeof input.reason === 'string' ? input.reason : null, - released_at: new Date().toISOString(), - }, - }, - }) - .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, agent.id))); + if (deps) { + await emitAgentExitedEffects(deps, workspaceId, { + agentId: exited.agentId, + agentName: exited.agentName, + nodeId, + invocationId: exited.invocationId, + reason: 'released', + }); + } } async function dispatchNodeAttempt( @@ -1494,6 +1509,7 @@ export async function completeNodeInvocation( output?: unknown; error?: string; }, + deps?: InvocationCompletionDeps, ) { const [existing] = await db .select({ @@ -1571,7 +1587,7 @@ export async function completeNodeInvocation( } if (updated) { - await applyReleaseCompletionEffect(db, workspaceId, nodeId, existing, data); + await applyReleaseCompletionEffect(db, workspaceId, nodeId, existing, data, deps); } return updated ? publicInvocation(updated) : null; diff --git a/packages/engine/src/engine/invocationCompletion.ts b/packages/engine/src/engine/invocationCompletion.ts index 9b1b704b..9366a820 100644 --- a/packages/engine/src/engine/invocationCompletion.ts +++ b/packages/engine/src/engine/invocationCompletion.ts @@ -1,5 +1,8 @@ +import { and, eq } from 'drizzle-orm'; import { transformForClient } from './wsTransform.js'; import { enqueueEvent } from './eventQueue.js'; +import { actionInvocations } from '../db/schema.js'; +import type { EngineDb } from '../ports/database.js'; import type { EngineDeps } from '../ports/index.js'; import { sendNodeDeliveriesToAgents } from './nodeDeliver.js'; import { appendAndPublishWorkspaceEvent } from './workspaceEvents.js'; @@ -68,20 +71,176 @@ export async function emitInvocationCompletionEffects( ); await Promise.allSettled(fanoutTasks); + await enqueueAndSendWebhook(deps, workspaceId, eventType, result as Record); +} + +/** + * Push one event through the durable outbox + external webhook queue. Both steps + * are best-effort — a caller's state transition must never fail because the + * outbox insert or queue send did. + */ +async function enqueueAndSendWebhook( + deps: Pick, + workspaceId: string, + eventType: string, + data: Record, +): Promise { let outboxId: string | undefined; try { - outboxId = await enqueueEvent(deps.db, workspaceId, eventType, result as Record); + outboxId = await enqueueEvent(deps.db, workspaceId, eventType, data); } catch { // Best-effort: if the durable outbox row can't be inserted, still try the queue send. } try { - await deps.webhookQueue.send(outboxId ? { type: eventType, workspaceId, data: result, outboxId } : { + await deps.webhookQueue.send(outboxId ? { type: eventType, workspaceId, data, outboxId } : { type: eventType, workspaceId, - data: result, + data, }); } catch { - // Best-effort side effect; completion must not fail because webhook delivery did. + // Best-effort side effect; the state transition must not fail because webhook delivery did. + } +} + +/** + * Extract the spawn invocation id an agent was registered under, if any. The + * `metadata.fleet.invocation_id` correlation is written by `registerAgentViaNode` + * on every via-node agent. + */ +export function fleetInvocationId(metadata: unknown): string | null { + if (metadata && typeof metadata === 'object' && !Array.isArray(metadata)) { + const fleet = (metadata as Record).fleet; + if (fleet && typeof fleet === 'object' && !Array.isArray(fleet)) { + const invocationId = (fleet as Record).invocation_id; + if (typeof invocationId === 'string') return invocationId; + } + } + return null; +} + +/** Resolve the caller mailbox of a spawn invocation; best-effort (null on any error). */ +async function lookupSpawnCaller(db: EngineDb, workspaceId: string, invocationId: string): Promise { + try { + const [row] = await db + .select({ callerId: actionInvocations.callerId }) + .from(actionInvocations) + .where(and(eq(actionInvocations.workspaceId, workspaceId), eq(actionInvocations.id, invocationId))); + return row?.callerId ?? null; + } catch { + return null; } } + +/** Distinguishes the three server-observable sources of an agent exit. */ +export type AgentExitedReason = 'deregistered' | 'missing_from_inventory' | 'released'; + +export interface AgentExitedInput { + agentId: string; + agentName: string; + /** The node the agent exited from (authoritative), or null when unknown. */ + nodeId: string | null; + /** The spawn invocation the agent was registered under, if correlated. */ + invocationId: string | null; + reason: AgentExitedReason; +} + +/** + * Fan out a durable `agent.exited` event across the same three channels as + * invocation completion: (1) the spawn caller's mailbox (when the exit + * correlates to a spawn invocation with a live caller), (2) the durable + * workspace event log + live observer stream, and (3) the webhook outbox. + * + * Best-effort by contract — every channel is isolated so the underlying agent + * state transition (deregister / inventory reconcile / release) never fails + * because an emission did. + */ +export async function emitAgentExitedEffects( + deps: InvocationCompletionDeps, + workspaceId: string, + input: AgentExitedInput, +): Promise { + const eventType = 'agent.exited'; + const eventData: Record = { + agent_id: input.agentId, + agent_name: input.agentName, + node_id: input.nodeId, + invocation_id: input.invocationId, + reason: input.reason, + }; + const event = buildEvent(eventType, workspaceId, eventData); + const payload = transformForClient(event); + + const fanoutTasks: Promise[] = []; + // Notify the spawn caller's mailbox so a spawner learns its agent exited. The + // stable eventKey dedupes redundant emissions of the same logical exit. + const callerId = input.invocationId ? await lookupSpawnCaller(deps.db, workspaceId, input.invocationId) : null; + if (callerId) { + fanoutTasks.push( + sendNodeDeliveriesToAgents({ + db: deps.db, + nodeConnections: deps.nodeConnections, + workspaceId, + environment: deps.config?.environment, + httpPushProxy: deps.config?.httpPushProxy, + }, { + agentIds: [callerId], + event: eventType, + eventKey: `${input.agentId}_${input.invocationId ?? 'none'}_${input.reason}`, + data: eventData, + messageId: `agent_exited_${input.agentId}`, + }), + ); + } + fanoutTasks.push( + appendAndPublishWorkspaceEvent( + { db: deps.db, realtime: deps.realtime }, + workspaceId, + { type: eventType, payload }, + ), + ); + await Promise.allSettled(fanoutTasks); + + await enqueueAndSendWebhook(deps, workspaceId, eventType, eventData); +} + +/** Reason a node flipped offline, when cheaply known at the transition point. */ +export type NodeOfflineReason = 'liveness_timeout' | 'disconnected' | 'deregistered'; + +export interface NodeStatusInput { + status: 'online' | 'offline'; + nodeId: string; + nodeName: string | null; + /** Only meaningful for `offline`; omitted for `online`. */ + reason?: NodeOfflineReason; +} + +/** + * Fan out a durable `node.status.online` / `node.status.offline` event. Node + * death has no single caller mailbox, so only the durable workspace log + live + * stream and the webhook outbox apply. Best-effort by contract. + */ +export async function emitNodeStatusEffects( + deps: Pick, + workspaceId: string, + input: NodeStatusInput, +): Promise { + const eventType = input.status === 'online' ? 'node.status.online' : 'node.status.offline'; + const eventData: Record = { + node_id: input.nodeId, + node_name: input.nodeName, + ...(input.status === 'offline' && input.reason ? { reason: input.reason } : {}), + }; + const event = buildEvent(eventType, workspaceId, eventData); + const payload = transformForClient(event); + + await Promise.allSettled([ + appendAndPublishWorkspaceEvent( + { db: deps.db, realtime: deps.realtime }, + workspaceId, + { type: eventType, payload }, + ), + ]); + + await enqueueAndSendWebhook(deps, workspaceId, eventType, eventData); +} diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 4f180ced..f6e66546 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -40,8 +40,8 @@ import { rescheduleInvocationsForLostNode, rescheduleNodeInvocation, } from './action.js'; -import { emitInvocationCompletionEffects } from './invocationCompletion.js'; -import type { InvocationCompletionDeps } from './invocationCompletion.js'; +import { emitInvocationCompletionEffects, emitAgentExitedEffects, emitNodeStatusEffects, fleetInvocationId } from './invocationCompletion.js'; +import type { InvocationCompletionDeps, NodeOfflineReason } from './invocationCompletion.js'; import { ackDeliveriesUpToSeq, deliverPendingToNode } from './delivery.js'; type Db = ReturnType; @@ -473,7 +473,16 @@ export async function markNodeOffline( registry: NodeConnectionRegistry, workspaceId: string, nodeId: string, + effects?: { deps?: InvocationCompletionDeps; reason?: NodeOfflineReason }, ) { + // Capture the node's name + prior status BEFORE flipping it so a durable + // node.status.offline is emitted only on a real online -> offline transition + // (never on a node that was already offline). + const [before] = await db + .select({ status: nodes.status, name: nodes.name }) + .from(nodes) + .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, nodeId))); + await db .update(nodes) .set({ @@ -503,6 +512,15 @@ export async function markNodeOffline( )); await rescheduleInvocationsForLostNode(db, registry, workspaceId, nodeId); + + if (effects?.deps && before && before.status === 'online') { + await emitNodeStatusEffects(effects.deps, workspaceId, { + status: 'offline', + nodeId, + nodeName: before.name, + reason: effects.reason, + }); + } } /** @@ -517,13 +535,14 @@ export async function handleProviderDisconnect( nodeId: string, providerName: string, hasRemainingConnections: boolean, + deps?: InvocationCompletionDeps, ) { // Serialized with node-control operations so a teardown can't race a // concurrent register's provider upsert / aggregate recompute (or, on // better-sqlite3, deadlock its isolated transaction). return serializeNodeOp(workspaceId, nodeId, async () => { if (!hasRemainingConnections) { - await markNodeOffline(db, registry, workspaceId, nodeId); + await markNodeOffline(db, registry, workspaceId, nodeId, { deps, reason: 'disconnected' }); return; } await markProviderOffline(db, workspaceId, nodeId, providerName); @@ -551,6 +570,7 @@ export async function deregisterProvider( workspaceId: string, nodeId: string, providerName: string, + deps?: InvocationCompletionDeps, ) { await removeProvider(db, workspaceId, nodeId, providerName); registry.detachProvider(workspaceId, nodeId, providerName); @@ -568,7 +588,7 @@ export async function deregisterProvider( .from(nodeProviders) .where(and(eq(nodeProviders.workspaceId, workspaceId), eq(nodeProviders.nodeId, nodeId))); if (!remaining || remaining.count === 0) { - await markNodeOffline(db, registry, workspaceId, nodeId); + await markNodeOffline(db, registry, workspaceId, nodeId, { deps, reason: 'deregistered' }); } else { // Other providers remain: recompute the node aggregate but don't reschedule // node-wide — that would disturb the surviving providers' in-flight invokes. @@ -578,7 +598,11 @@ export async function deregisterProvider( } } -export async function sweepOfflineNodes(db: Db, registry: NodeConnectionRegistry): Promise { +export async function sweepOfflineNodes( + db: Db, + registry: NodeConnectionRegistry, + deps?: InvocationCompletionDeps, +): Promise { const rows = await db.select().from(nodes).where(eq(nodes.status, 'online')); const stale = rows.filter((node) => !isNodeLive(node)); let swept = 0; @@ -591,7 +615,7 @@ export async function sweepOfflineNodes(db: Db, registry: NodeConnectionRegistry .from(nodes) .where(and(eq(nodes.workspaceId, node.workspaceId), eq(nodes.id, node.id))); if (!current || current.status !== 'online' || isNodeLive(current)) return false; - await markNodeOffline(db, registry, node.workspaceId, node.id); + await markNodeOffline(db, registry, node.workspaceId, node.id, { deps, reason: 'liveness_timeout' }); return true; }); if (offlined) swept++; @@ -1146,6 +1170,7 @@ export async function deregisterAgentViaNode( workspaceId: string, nodeId: string, message: { agent_id?: string; name?: string }, + deps?: InvocationCompletionDeps, ) { const conditions = [ eq(agents.workspaceId, workspaceId), @@ -1184,6 +1209,16 @@ export async function deregisterAgentViaNode( .where(and(eq(agents.workspaceId, workspaceId), eq(agents.id, updated.id))); await markDirectNodeOfflineForAgent(db, workspaceId, updated.id); await releaseNodeAgentSlots(db, workspaceId, activeNodeIds.filter((activeNodeId) => activeNodeId !== nodeId)); + + if (deps) { + await emitAgentExitedEffects(deps, workspaceId, { + agentId: updated.id, + agentName: updated.name, + nodeId, + invocationId: fleetInvocationId(updated.metadata), + reason: 'deregistered', + }); + } } return updated ?? null; } @@ -1334,7 +1369,7 @@ export async function reconcileInventory( invocation_id: item.invocation_id ?? null, session_ref: item.session_ref ?? null, }, - }); + }, completionDeps); if (completed) { completedInvocations++; if (completionDeps) { @@ -1344,7 +1379,7 @@ export async function reconcileInventory( } const nodeAgents = await db - .select({ id: agents.id, name: agents.name }) + .select({ id: agents.id, name: agents.name, metadata: agents.metadata }) .from(agents) .where(and( eq(agents.workspaceId, workspaceId), @@ -1353,12 +1388,24 @@ export async function reconcileInventory( eq(agents.providerName, providerName), eq(agents.status, 'active'), )); - const missing = nodeAgents.filter((agent) => !names.has(agent.name)).map((agent) => agent.id); + const missingAgents = nodeAgents.filter((agent) => !names.has(agent.name)); + const missing = missingAgents.map((agent) => agent.id); if (missing.length > 0) { await db .update(agents) .set({ status: 'offline', lastSeen: new Date() }) .where(inArray(agents.id, missing)); + if (completionDeps) { + for (const agent of missingAgents) { + await emitAgentExitedEffects(completionDeps, workspaceId, { + agentId: agent.id, + agentName: agent.name, + nodeId, + invocationId: fleetInvocationId(agent.metadata), + reason: 'missing_from_inventory', + }); + } + } } let rescheduledInvocations = 0; @@ -1411,6 +1458,35 @@ export async function getPublicNode(db: Db, workspaceId: string, name: string) { return node ? publicNode(node) : null; } +/** The node's persisted `status` column, or undefined when the node is gone. */ +async function readNodeStatus(db: Db, workspaceId: string, nodeId: string): Promise { + const [row] = await db + .select({ status: nodes.status }) + .from(nodes) + .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, nodeId))); + return row?.status; +} + +/** + * Emit a durable `node.status.online` only on a real offline -> online + * transition: the node was not `online` before the register/heartbeat and the + * resulting descriptor reports online. A steady heartbeat (already online) or a + * register that leaves the node offline (no live provider) emits nothing. + */ +async function emitNodeOnlineTransition( + deps: InvocationCompletionDeps | undefined, + workspaceId: string, + priorStatus: string | undefined, + node: { id: string; name: string; status: string } | null, +): Promise { + if (!deps || !node || priorStatus === 'online' || node.status !== 'online') return; + await emitNodeStatusEffects(deps, workspaceId, { + status: 'online', + nodeId: node.id, + nodeName: node.name, + }); +} + function sendControl(socket: NodeSocketLike | undefined, payload: Record): boolean { if (!socket) return false; try { @@ -1477,6 +1553,7 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg throw codedError(conflict.message, conflict.code, 409); } } + const priorStatus = await readNodeStatus(args.db, args.workspaceId, args.nodeId); const registered = await registerNode(args.db, args.workspaceId, args.nodeId, message, provider); const readinessSupported = supportsProviderDeliveryReadiness(args.registry); const acceptance = registered.acceptance.map((capability) => ( @@ -1533,16 +1610,20 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg { providerName: provider.name }, ).catch(() => {}); } + await emitNodeOnlineTransition(args.completionDeps, args.workspaceId, priorStatus, registered.node); return; } - case 'node.heartbeat': - await heartbeatNode(args.db, args.workspaceId, args.nodeId, frameProviderName, message); + case 'node.heartbeat': { + const priorStatus = await readNodeStatus(args.db, args.workspaceId, args.nodeId); + const beat = await heartbeatNode(args.db, args.workspaceId, args.nodeId, frameProviderName, message); // Heartbeat refreshes online/capacity state; re-drain as a backstop in // case a queued spawn could not reserve capacity at register time. await args.registry.drainNode(args.workspaceId, args.nodeId); + await emitNodeOnlineTransition(args.completionDeps, args.workspaceId, priorStatus, beat); return; + } case 'node.deregister': - await deregisterProvider(args.db, args.registry, args.workspaceId, args.nodeId, frameProviderName); + await deregisterProvider(args.db, args.registry, args.workspaceId, args.nodeId, frameProviderName, args.completionDeps); return; case 'node.spawn': { // Handler-context `ctx.spawnAgent`: capacity-direct delegation to this @@ -1613,7 +1694,7 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg return; } case 'agent.deregister': - await deregisterAgentViaNode(args.db, args.workspaceId, args.nodeId, message); + await deregisterAgentViaNode(args.db, args.workspaceId, args.nodeId, message, args.completionDeps); return; case 'inventory.sync': { const result = await reconcileInventory( @@ -1675,6 +1756,7 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg ...(Object.prototype.hasOwnProperty.call(message, 'output') ? { output: message.output } : {}), ...(message.error ? { error: message.error } : {}), }, + args.completionDeps, ); if (completed && args.completionDeps) { await emitInvocationCompletionEffects(args.completionDeps, args.workspaceId, completed); diff --git a/packages/engine/src/routes/node.ts b/packages/engine/src/routes/node.ts index cdb2eb8f..ead726c2 100644 --- a/packages/engine/src/routes/node.ts +++ b/packages/engine/src/routes/node.ts @@ -343,8 +343,9 @@ nodeRoutes.delete('/nodes/:node/providers/:name', requireWorkspaceKey, rateLimit } // Serialize with node-control operations so removal can't race a concurrent // register/heartbeat's aggregate recompute. + const engine = c.get('engine'); await serializeNodeOp(workspace.id, node.id, () => - nodeEngine.deregisterProvider(db, c.get('engine').nodeConnections, workspace.id, node.id, c.req.param('name')), + nodeEngine.deregisterProvider(db, engine.nodeConnections, workspace.id, node.id, c.req.param('name'), engine), ); return jsonNoContent(c); } catch (err: unknown) { diff --git a/packages/engine/src/routes/presence.ts b/packages/engine/src/routes/presence.ts index fef447d1..8bf36594 100644 --- a/packages/engine/src/routes/presence.ts +++ b/packages/engine/src/routes/presence.ts @@ -49,9 +49,10 @@ presenceRoutes.post('/agents/disconnect', requireAgentToken, rateLimit, async (c const db = c.get('db'); const agent = c.get('agent')!; const workspace = c.get('workspace'); - const { presence } = c.get('engine'); + const engine = c.get('engine'); + const { presence } = engine; - await handleAgentDisconnect(db, workspace.id, agent.id); + await handleAgentDisconnect(db, workspace.id, agent.id, engine); await presence.disconnect(workspace.id, agent.id, agent.name); emitServerEvent(c, workspace.id, 'relaycast_server_presence_disconnected', { diff --git a/packages/sdk-python/src/relay_sdk/models.py b/packages/sdk-python/src/relay_sdk/models.py index a0544f21..3fb50316 100644 --- a/packages/sdk-python/src/relay_sdk/models.py +++ b/packages/sdk-python/src/relay_sdk/models.py @@ -970,6 +970,9 @@ class PongEvent(BaseModel): "agent.status.blocked", "agent.status.waiting", "agent.status.offline", + "agent.exited", + "node.status.online", + "node.status.offline", "channel.created", "channel.archived", "message.read", diff --git a/packages/types/CHANGELOG.md b/packages/types/CHANGELOG.md index 389852fc..dbb327b3 100644 --- a/packages/types/CHANGELOG.md +++ b/packages/types/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Minor] + +### Added + +- `AgentExitedEventSchema`, `NodeStatusOnlineEventSchema`, and `NodeStatusOfflineEventSchema` added to `ServerEventSchema` / `WsClientEventSchema`, and `agent.exited` / `node.status.online` / `node.status.offline` added to `SubscribableEventTypeSchema` so webhooks can subscribe to them. ## [6.0.3] - 2026-07-12 diff --git a/packages/types/src/events.ts b/packages/types/src/events.ts index 2b2d3772..b532db46 100644 --- a/packages/types/src/events.ts +++ b/packages/types/src/events.ts @@ -136,6 +136,34 @@ export const AgentStatusChangedEventSchema = z.object({ }); export type AgentStatusChangedEvent = z.infer; +// Durable agent exit — an agent hosted by a node was deregistered, dropped from +// a node inventory sync, or released. Correlates back to the spawn invocation. +export const AgentExitedEventSchema = z.object({ + type: z.literal('agent.exited'), + agent_id: z.string(), + agent_name: z.string(), + node_id: z.string().nullable(), + invocation_id: z.string().nullable(), + reason: z.enum(['deregistered', 'missing_from_inventory', 'released']), +}); +export type AgentExitedEvent = z.infer; + +// Durable node liveness transitions. Emitted once per online <-> offline flip. +export const NodeStatusOnlineEventSchema = z.object({ + type: z.literal('node.status.online'), + node_id: z.string(), + node_name: z.string().nullable(), +}); +export type NodeStatusOnlineEvent = z.infer; + +export const NodeStatusOfflineEventSchema = z.object({ + type: z.literal('node.status.offline'), + node_id: z.string(), + node_name: z.string().nullable(), + reason: z.enum(['liveness_timeout', 'disconnected', 'deregistered']).optional(), +}); +export type NodeStatusOfflineEvent = z.infer; + export type AgentStatusEvent = | AgentStatusActiveEvent | AgentStatusIdleEvent @@ -360,6 +388,9 @@ export const ServerEventSchema = z.discriminatedUnion('type', [ AgentStatusWaitingEventSchema, AgentStatusOfflineEventSchema, AgentStatusChangedEventSchema, + AgentExitedEventSchema, + NodeStatusOnlineEventSchema, + NodeStatusOfflineEventSchema, ChannelCreatedEventSchema, ChannelUpdatedEventSchema, ChannelArchivedEventSchema, @@ -408,6 +439,9 @@ export const WsClientEventSchema = z.discriminatedUnion('type', [ AgentStatusWaitingEventSchema, AgentStatusOfflineEventSchema, AgentStatusChangedEventSchema, + AgentExitedEventSchema, + NodeStatusOnlineEventSchema, + NodeStatusOfflineEventSchema, ChannelCreatedEventSchema, ChannelUpdatedEventSchema, ChannelArchivedEventSchema, diff --git a/packages/types/src/subscription.ts b/packages/types/src/subscription.ts index c8697b96..24d4097d 100644 --- a/packages/types/src/subscription.ts +++ b/packages/types/src/subscription.ts @@ -11,6 +11,9 @@ export const SubscribableEventTypeSchema = z.enum([ 'agent.status.blocked', 'agent.status.waiting', 'agent.status.offline', + 'agent.exited', + 'node.status.online', + 'node.status.offline', 'channel.created', 'channel.updated', 'channel.archived', From eb264b19dafbfe39067dfaf08318deb5e48ed17f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 14:51:08 +0000 Subject: [PATCH 2/3] fix(engine): address PR #279 review findings - observerToken: map agent.exited to agents:read so agents:read-only observer tokens receive it (node.status.* already maps via node.* -> nodes:read) - deregisterProvider: emit node.status.offline when deregistering the last online provider flips the node offline via recomputeNodeAggregate (leftover offline provider row keeps remaining.count > 0) - node teardown: serialize the unbound-socket-close markNodeOffline under serializeNodeOp and re-check connectivity, so a racing reconnect can't be clobbered by a stale offline transition - tests: fix agentExitedEffects insert mock to exercise the real .values().returning() seq-stamping path and assert seq; add conformance tests for deregister idempotency and the provider-deregister offline flip - changelog: rewrite engine bullets to impact-first, user-facing language - trajectory: correct stale commit SHA 625a610 -> c968cb2 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JQ8fa196XtLnvctQMQuUDx --- .../2026-07/traj_wulwzxphvrdc.trace.json | 90 +++++++++---------- .../2026-07/traj_wulwzxphvrdc/summary.md | 2 +- .../2026-07/traj_wulwzxphvrdc/trajectory.json | 4 +- packages/engine/CHANGELOG.md | 4 +- .../conformance/agentNodeExitEvents.test.ts | 52 +++++++++++ packages/engine/src/adapters/node/realtime.ts | 11 ++- .../__tests__/agentExitedEffects.test.ts | 9 +- packages/engine/src/engine/node.ts | 30 ++++++- packages/engine/src/engine/observerToken.ts | 2 +- 9 files changed, 147 insertions(+), 57 deletions(-) diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json index 84586683..887935fb 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json +++ b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc.trace.json @@ -15,7 +15,7 @@ { "start_line": 1, "end_line": 19, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -32,7 +32,7 @@ { "start_line": 21, "end_line": 28, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -49,7 +49,7 @@ { "start_line": 403, "end_line": 424, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -66,7 +66,7 @@ { "start_line": 4275, "end_line": 4283, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -83,7 +83,7 @@ { "start_line": 10, "end_line": 17, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -100,7 +100,7 @@ { "start_line": 1, "end_line": 174, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -117,7 +117,7 @@ { "start_line": 164, "end_line": 170, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -134,7 +134,7 @@ { "start_line": 529, "end_line": 541, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -151,12 +151,12 @@ { "start_line": 2, "end_line": 21, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 39, "end_line": 45, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -173,7 +173,7 @@ { "start_line": 1, "end_line": 169, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -190,32 +190,32 @@ { "start_line": 7, "end_line": 13, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 843, "end_line": 849, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 877, "end_line": 886, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 890, "end_line": 927, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1509, "end_line": 1515, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1587, "end_line": 1593, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -232,12 +232,12 @@ { "start_line": 1, "end_line": 8, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 71, "end_line": 246, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -254,92 +254,92 @@ { "start_line": 40, "end_line": 47, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 473, "end_line": 488, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 512, "end_line": 526, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 535, "end_line": 548, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 570, "end_line": 576, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 588, "end_line": 594, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 598, "end_line": 608, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 615, "end_line": 621, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1170, "end_line": 1176, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1209, "end_line": 1224, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1369, "end_line": 1375, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1379, "end_line": 1385, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1388, "end_line": 1411, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1458, "end_line": 1492, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1553, "end_line": 1559, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1610, "end_line": 1629, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1694, "end_line": 1700, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 1756, "end_line": 1762, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -356,7 +356,7 @@ { "start_line": 343, "end_line": 351, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -373,7 +373,7 @@ { "start_line": 49, "end_line": 58, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -390,7 +390,7 @@ { "start_line": 970, "end_line": 978, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -407,7 +407,7 @@ { "start_line": 7, "end_line": 17, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -424,17 +424,17 @@ { "start_line": 136, "end_line": 169, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 388, "end_line": 396, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" }, { "start_line": 439, "end_line": 447, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } @@ -451,7 +451,7 @@ { "start_line": 11, "end_line": 19, - "revision": "625a6100becac9779c5feb824d01f568d1e1fdea" + "revision": "c968cb2820a1012366308478f4757629e3ea29d2" } ] } diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md index fcc62ad0..6f38c91d 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/summary.md @@ -17,5 +17,5 @@ Emitted durable agent.exited (deregister / inventory-missing / release) and node ## Artifacts -**Commits:** 625a610 +**Commits:** c968cb2 **Files changed:** 19 diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json index 9aedf0d2..8b8a89da 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-07/traj_wulwzxphvrdc/trajectory.json @@ -15,7 +15,7 @@ "confidence": 0.85 }, "commits": [ - "625a610" + "c968cb2" ], "filesChanged": [ ".agentworkforce/trajectories/active/traj_wulwzxphvrdc/trajectory.json", @@ -42,7 +42,7 @@ "tags": [], "_trace": { "startRef": "2e2b7a2b0719db1057555e151ae326154903e685", - "endRef": "625a6100becac9779c5feb824d01f568d1e1fdea", + "endRef": "c968cb2820a1012366308478f4757629e3ea29d2", "traceId": "b2fa69e0-81d6-4e05-9352-c33aa33b5578" } } \ No newline at end of file diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index d41cde5a..3c8b7a6c 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -10,8 +10,8 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased - Minor] ### Added -- Emit durable `agent.exited` on every server-observable agent exit — `deregisterAgentViaNode` (deregister control frame + `handleAgentDisconnect`), the `reconcileInventory` missing-agent sweep, and release completion (`completeNodeInvocation`) — fanned out via the new `emitAgentExitedEffects` helper to the spawn caller's mailbox, the durable workspace log, and the webhook outbox. -- Emit durable `node.status.online` / `node.status.offline` on node liveness transitions via `emitNodeStatusEffects`: `markNodeOffline` (liveness sweep, provider disconnect, deregister) and the register/heartbeat offline→online transition. Both helpers reuse `InvocationCompletionDeps` and are best-effort (never fail the underlying state transition). +- Added durable `agent.exited` events on every node-hosted agent exit (deregistration, missing from an inventory sync, and release), delivered to the durable workspace event log, webhook subscribers, and the spawn caller's mailbox, and carrying `agent_id`, `agent_name`, `node_id`, `invocation_id`, and a `reason`. +- Added durable `node.status.online` / `node.status.offline` events on node liveness transitions (offline carries a `reason` such as `liveness_timeout`, `disconnected`, or `deregistered`), delivered to the workspace event log and webhook subscribers. - Message retention remains opt-in (`pruneExpired` still defaults `messageTtlDays` to `null`). Self-host can now opt in to a deployment-wide message TTL: `startServer` accepts `eventQueue` (`DurableEventQueueOptions`, including `retention`), and the `relaycast-engine` CLI exposes `RELAYCAST_MESSAGE_TTL_DAYS` (positive = prune after N days; unset or `0`/negative = keep forever). - Exported the provider-attach arbitration policy from `@relaycast/engine/node-control`: `providerAttachDecision()` plus `PROVIDER_ATTACH_LIVENESS_MS`, so an out-of-process socket owner (a hosted NodeDO) mirrors the spec §3.1 decision from one source of truth instead of hand-copying the constant and logic. diff --git a/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts b/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts index 883671d2..d9d299db 100644 --- a/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts +++ b/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts @@ -146,6 +146,58 @@ describe('agent.exited + node.status durable events', () => { }); }); + it('(f) repeated agent.deregister frames emit agent.exited exactly once', async () => { + const ws = await createWorkspace(stack.app, 'exit-f'); + const { sock, handle } = await bringNodeOnline(ws, 'node_f', 'zeta'); + const agentId = await registerAgentViaNode(handle, sock, 'worker-f'); + + // First deregister transitions the agent and re-homes its location to the + // implicit direct node, so subsequent frames for node_f no longer match. + await handle.handleMessage(JSON.stringify({ v: 1, type: 'agent.deregister', agent_id: agentId })); + await handle.handleMessage(JSON.stringify({ v: 1, type: 'agent.deregister', agent_id: agentId })); + await handle.handleMessage(JSON.stringify({ v: 1, type: 'agent.deregister', name: 'worker-f' })); + + expect(await workspaceEventsOfType(ws.workspaceId, 'agent.exited')).toHaveLength(1); + }); + + it('(g) deregistering the last online provider emits node.status.offline even with a leftover offline provider row', async () => { + const ws = await createWorkspace(stack.app, 'exit-g'); + await enrollNode(ws.workspaceKey, 'node_g', 'eta'); + + // Two providers on the node: py + rb. + const py = attach(ws.workspaceId, 'node_g'); + await py.handle.handleMessage(JSON.stringify({ + v: 1, id: 'reg-py', type: 'node.register', name: 'eta', node_id: 'node_g', + provider: { name: 'py', instance_id: 'py-i1' }, + capabilities: [{ name: 'spawn:claude', kind: 'capacity' }], max_agents: 4, tags: ['test'], version: 'v1', resume_cursor: null, + })); + await py.handle.handleMessage(JSON.stringify({ v: 1, type: 'node.heartbeat', provider: { name: 'py', instance_id: 'py-i1' }, load: 0, active_agents: 0, handlers_live: true })); + const rb = attach(ws.workspaceId, 'node_g'); + await rb.handle.handleMessage(JSON.stringify({ + v: 1, id: 'reg-rb', type: 'node.register', name: 'eta', node_id: 'node_g', + provider: { name: 'rb', instance_id: 'rb-i1' }, + capabilities: [{ name: 'build', kind: 'action' }], max_agents: 4, tags: ['test'], version: 'v1', resume_cursor: null, + })); + await rb.handle.handleMessage(JSON.stringify({ v: 1, type: 'node.heartbeat', provider: { name: 'rb', instance_id: 'rb-i1' }, load: 0, active_agents: 0, handlers_live: true })); + + // rb's socket drops while py remains: rb becomes a persisted *offline* provider + // row and the node stays online, so nothing is emitted yet. + await rb.handle.handleClose(); + expect(await workspaceEventsOfType(ws.workspaceId, 'node.status.offline')).toHaveLength(0); + + // Deregister py (the last online provider). The leftover rb row keeps + // remaining.count > 0, routing through recomputeNodeAggregate — which flips the + // node offline. The durable node.status.offline must still be emitted. + await py.handle.handleMessage(JSON.stringify({ v: 1, type: 'node.deregister', provider: { name: 'py', instance_id: 'py-i1' } })); + + const logged = await workspaceEventsOfType(ws.workspaceId, 'node.status.offline'); + expect(logged.length).toBeGreaterThanOrEqual(1); + expect(logged.some((row) => { + const p = JSON.parse(row.payload) as { node_id?: string; reason?: string }; + return p.node_id === 'node_g' && p.reason === 'deregistered'; + })).toBe(true); + }); + it('(e) re-heartbeat after offline emits node.status.online exactly once', async () => { const ws = await createWorkspace(stack.app, 'exit-e'); // Initial register already produces the first offline->online transition. diff --git a/packages/engine/src/adapters/node/realtime.ts b/packages/engine/src/adapters/node/realtime.ts index 92540dbc..b7880c9d 100644 --- a/packages/engine/src/adapters/node/realtime.ts +++ b/packages/engine/src/adapters/node/realtime.ts @@ -11,6 +11,7 @@ import { and, eq, gt, isNull, or } from 'drizzle-orm'; import type { EngineDb } from '../../ports/database.js'; import { observerTokens } from '../../db/schema.js'; import { handleNodeControlMessage, handleProviderDisconnect, markNodeOffline } from '../../engine/node.js'; +import { serializeNodeOp } from '../../engine/nodeLock.js'; import { DEFAULT_PROVIDER_NAME } from '../../engine/nodeProvider.js'; import { providerAttachDecision } from '../../engine/placement.js'; import { drainNodeInvocations } from '../../engine/action.js'; @@ -535,7 +536,15 @@ export class InProcessRealtime implements RealtimeBus, ConnectionRegistry, NodeC }); } else if (!hasRemaining) { // Connection dropped before it bound a provider and it was the node's last. - await markNodeOffline(this.db, this, workspaceId, nodeId, { deps: this.nodeCompletionDeps, reason: 'disconnected' }).catch((err) => { + // Serialize with node-control ops (register/heartbeat) and re-check + // connectivity under the lock: a reconnect racing this close may have bound + // a fresh socket, in which case the node must stay online. Without this a + // stale close could append node.status.offline after the reconnect's online + // event and leave the node marked offline despite a live socket. + await serializeNodeOp(workspaceId, nodeId, async () => { + if (this.isNodeConnected(workspaceId, nodeId)) return; + await markNodeOffline(this.db, this, workspaceId, nodeId, { deps: this.nodeCompletionDeps, reason: 'disconnected' }); + }).catch((err) => { console.error('[node.teardown] mark node offline failed', { workspace_id: workspaceId, node_id: nodeId }, err); }); } diff --git a/packages/engine/src/engine/__tests__/agentExitedEffects.test.ts b/packages/engine/src/engine/__tests__/agentExitedEffects.test.ts index c5637106..5551f1c9 100644 --- a/packages/engine/src/engine/__tests__/agentExitedEffects.test.ts +++ b/packages/engine/src/engine/__tests__/agentExitedEffects.test.ts @@ -14,6 +14,10 @@ vi.mock('../nodeDeliver.js', () => ({ * Build a deps stub whose `db` resolves a caller lookup to `callerId` and lets * the durable outbox insert succeed/fail on demand. `select().from().where()` * returns the invocation caller row; `insert()` throws when `outboxThrows`. + * + * The insert stub mirrors the two real shapes: the workspace-event append chains + * `.values(...).returning(...)` (so the primary seq-stamping path is exercised + * and returns `seq: 1`), while the outbox enqueue awaits `.values(...)` directly. */ function makeDeps(opts: { callerId?: string | null; outboxThrows?: boolean } = {}) { const publishToWorkspaceStream = vi.fn(async () => {}); @@ -26,7 +30,8 @@ function makeDeps(opts: { callerId?: string | null; outboxThrows?: boolean } = { }), insert: () => { if (opts.outboxThrows) throw new Error('skip durable outbox in test'); - return { values: async () => {} }; + const chain = { returning: async () => [{ seq: 1 }] }; + return { values: () => chain }; }, }; const deps = { @@ -71,6 +76,8 @@ describe('emitAgentExitedEffects', () => { expect(publishToWorkspaceStream).toHaveBeenCalledTimes(1); const published = publishToWorkspaceStream.mock.calls[0]?.[0] as { event: Record }; expect(published.event).toMatchObject({ type: 'agent.exited', agent_id: 'agent-9', reason: 'released' }); + // Primary (non-throwing) append path assigns and stamps the durable seq. + expect(published.event).toMatchObject({ seq: 1 }); // (3) webhook outbox expect(webhookSend).toHaveBeenCalledTimes(1); expect(webhookSend.mock.calls[0]?.[0]).toMatchObject({ type: 'agent.exited', workspaceId: 'ws-1' }); diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index f6e66546..31d31eeb 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -590,11 +590,33 @@ export async function deregisterProvider( if (!remaining || remaining.count === 0) { await markNodeOffline(db, registry, workspaceId, nodeId, { deps, reason: 'deregistered' }); } else { - // Other providers remain: recompute the node aggregate but don't reschedule - // node-wide — that would disturb the surviving providers' in-flight invokes. - // Work dispatched to the removed provider is caught by the dispatch-timeout - // sweep, matching the provider-disconnect path. + // Other provider rows remain (possibly all offline): recompute the node + // aggregate but don't reschedule node-wide — that would disturb the surviving + // providers' in-flight invokes. Work dispatched to the removed provider is + // caught by the dispatch-timeout sweep, matching the provider-disconnect path. + // Capture status across the recompute so that when the deregistered provider + // was the node's last *online* one, the resulting online -> offline flip still + // emits a durable node.status.offline — recomputeNodeAggregate flips the + // column but never emits. + const [before] = await db + .select({ status: nodes.status, name: nodes.name }) + .from(nodes) + .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, nodeId))); await recomputeNodeAggregate(db, workspaceId, nodeId); + if (deps && before?.status === 'online') { + const [after] = await db + .select({ status: nodes.status, name: nodes.name }) + .from(nodes) + .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, nodeId))); + if (after && after.status !== 'online') { + await emitNodeStatusEffects(deps, workspaceId, { + status: 'offline', + nodeId, + nodeName: after.name ?? before.name, + reason: 'deregistered', + }); + } + } } } diff --git a/packages/engine/src/engine/observerToken.ts b/packages/engine/src/engine/observerToken.ts index 40b29328..e7befd2a 100644 --- a/packages/engine/src/engine/observerToken.ts +++ b/packages/engine/src/engine/observerToken.ts @@ -541,7 +541,7 @@ function eventRequiredScopes(type: string): ObserverScope[] { if (type === 'message.reacted') return ['reactions:read']; if (type === 'dm.received' || type === 'group_dm.received') return ['dms:read']; if (type === 'file.uploaded') return ['files:read']; - if (type.startsWith('agent.status.')) return ['agents:read']; + if (type.startsWith('agent.status.') || type === 'agent.exited') return ['agents:read']; if (type.startsWith('member.') || type.startsWith('channel.')) return ['channels:read']; if (type.startsWith('delivery.')) return ['deliveries:read']; if (type.startsWith('node.')) return ['nodes:read']; From a523a022987a1f7de42578260bc1ce0c313f6f32 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 15:58:24 +0000 Subject: [PATCH 3/3] fix(engine): unique exit delivery ids, emit node online before drain - emitAgentExitedEffects: drop the agentId-only messageId so node delivery falls back to the unique eventKey (agent + invocation + reason); a messageId keyed only on agentId repeated across lifecycle sessions and a downstream dedup on it could drop later exit notifications. - fleetInvocationId: replace manual typeof/Array.isArray checks with a zod safeParse schema (AGENTS.md prefers zod); signature unchanged. - handleNodeControlMessage: emit the durable node.status.online transition right after the register/heartbeat persistence and before drainNode, with isolated error handling, so a drainNode rejection no longer skips the event. - agentNodeExitEvents conformance test (a): register the agent under a spawn invocation and assert the emitted agent.exited carries the same invocation_id. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JQ8fa196XtLnvctQMQuUDx --- .../conformance/agentNodeExitEvents.test.ts | 17 +++++++++++--- .../engine/src/engine/invocationCompletion.ts | 22 +++++++++++-------- packages/engine/src/engine/node.ts | 12 ++++++++-- 3 files changed, 37 insertions(+), 14 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts b/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts index d9d299db..e889cac7 100644 --- a/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts +++ b/packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts @@ -54,8 +54,16 @@ describe('agent.exited + node.status durable events', () => { return { sock, handle }; } - async function registerAgentViaNode(handle: { handleMessage(raw: string): Promise }, sock: FakeSocket, name: string) { - await handle.handleMessage(JSON.stringify({ v: 1, type: 'agent.register', name, resumable: true })); + async function registerAgentViaNode( + handle: { handleMessage(raw: string): Promise }, + sock: FakeSocket, + name: string, + opts: { invocationId?: string } = {}, + ) { + await handle.handleMessage(JSON.stringify({ + v: 1, type: 'agent.register', name, resumable: true, + ...(opts.invocationId ? { invocation_id: opts.invocationId } : {}), + })); const reply = sock.ofType('reply').at(-1) as { ok: boolean; data: { agent_id: string; name?: string } }; expect(reply?.ok).toBe(true); return reply.data.agent_id; @@ -71,7 +79,8 @@ describe('agent.exited + node.status durable events', () => { it('(a) agent.deregister control frame emits a durable agent.exited', async () => { const ws = await createWorkspace(stack.app, 'exit-a'); const { sock, handle } = await bringNodeOnline(ws, 'node_a', 'alpha'); - const agentId = await registerAgentViaNode(handle, sock, 'worker-a'); + // Register under a spawn invocation so the exit event must carry the same id. + const agentId = await registerAgentViaNode(handle, sock, 'worker-a', { invocationId: 'inv_spawn_a' }); await handle.handleMessage(JSON.stringify({ v: 1, type: 'agent.deregister', agent_id: agentId })); @@ -79,6 +88,8 @@ describe('agent.exited + node.status durable events', () => { expect(logged).toHaveLength(1); expect(JSON.parse(logged[0]!.payload)).toMatchObject({ type: 'agent.exited', agent_id: agentId, agent_name: 'worker-a', node_id: 'node_a', reason: 'deregistered', + // Spawn correlation: the exit carries the invocation the agent registered under. + invocation_id: 'inv_spawn_a', }); }); diff --git a/packages/engine/src/engine/invocationCompletion.ts b/packages/engine/src/engine/invocationCompletion.ts index 9366a820..422e8999 100644 --- a/packages/engine/src/engine/invocationCompletion.ts +++ b/packages/engine/src/engine/invocationCompletion.ts @@ -1,4 +1,5 @@ import { and, eq } from 'drizzle-orm'; +import { z } from 'zod'; import { transformForClient } from './wsTransform.js'; import { enqueueEvent } from './eventQueue.js'; import { actionInvocations } from '../db/schema.js'; @@ -103,20 +104,20 @@ async function enqueueAndSendWebhook( } } +/** Shape of the `metadata.fleet.invocation_id` spawn correlation. Unknown keys + * are stripped by default, so extra metadata fields are tolerated. */ +const fleetInvocationSchema = z.object({ + fleet: z.object({ invocation_id: z.string() }), +}); + /** * Extract the spawn invocation id an agent was registered under, if any. The * `metadata.fleet.invocation_id` correlation is written by `registerAgentViaNode` * on every via-node agent. */ export function fleetInvocationId(metadata: unknown): string | null { - if (metadata && typeof metadata === 'object' && !Array.isArray(metadata)) { - const fleet = (metadata as Record).fleet; - if (fleet && typeof fleet === 'object' && !Array.isArray(fleet)) { - const invocationId = (fleet as Record).invocation_id; - if (typeof invocationId === 'string') return invocationId; - } - } - return null; + const parsed = fleetInvocationSchema.safeParse(metadata); + return parsed.success ? parsed.data.fleet.invocation_id : null; } /** Resolve the caller mailbox of a spawn invocation; best-effort (null on any error). */ @@ -186,9 +187,12 @@ export async function emitAgentExitedEffects( }, { agentIds: [callerId], event: eventType, + // Omit messageId so delivery falls back to this unique eventKey (agent + + // invocation + reason). A messageId keyed only on agentId would repeat + // across an agent's lifecycle sessions and a downstream dedup on it could + // drop later exit notifications. eventKey: `${input.agentId}_${input.invocationId ?? 'none'}_${input.reason}`, data: eventData, - messageId: `agent_exited_${input.agentId}`, }), ); } diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 31d31eeb..da2d8b35 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -1613,6 +1613,11 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg accepted_capabilities: acceptance, }, }); + // The node row is already persisted online, so emit the durable + // transition before draining: a drainNode rejection must not skip the + // node.status.online event. Isolated so its own failure can't either. + await emitNodeOnlineTransition(args.completionDeps, args.workspaceId, priorStatus, registered.node) + .catch((err) => console.error('[node.status] online event emission failed', err)); // Node is now marked online: flush any queued action.invoke frames so // spawns queued while it was offline can reserve capacity and dispatch. await args.registry.drainNode(args.workspaceId, args.nodeId); @@ -1632,16 +1637,19 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg { providerName: provider.name }, ).catch(() => {}); } - await emitNodeOnlineTransition(args.completionDeps, args.workspaceId, priorStatus, registered.node); return; } case 'node.heartbeat': { const priorStatus = await readNodeStatus(args.db, args.workspaceId, args.nodeId); const beat = await heartbeatNode(args.db, args.workspaceId, args.nodeId, frameProviderName, message); + // The node row is already persisted, so emit the durable online + // transition before draining: a drainNode rejection must not skip the + // node.status.online event. Isolated so its own failure can't either. + await emitNodeOnlineTransition(args.completionDeps, args.workspaceId, priorStatus, beat) + .catch((err) => console.error('[node.status] online event emission failed', err)); // Heartbeat refreshes online/capacity state; re-drain as a backstop in // case a queued spawn could not reserve capacity at register time. await args.registry.drainNode(args.workspaceId, args.nodeId); - await emitNodeOnlineTransition(args.completionDeps, args.workspaceId, priorStatus, beat); return; } case 'node.deregister':