feat(engine): emit durable agent.exited and node.status events

.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:** c968cb2
+**Files changed:** 19

.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": [
+    "c968cb2"
+  ],
+  "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": "c968cb2820a1012366308478f4757629e3ea29d2",
+    "traceId": "b2fa69e0-81d6-4e05-9352-c33aa33b5578"
+  }
+}

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.
 - `POST /v1/agents/disconnect` accepts an optional `deregister` flag; SDK `disconnect()` and `presence.markOffline()` take `{ deregister?: boolean }` to opt into full node teardown.
 
 ### Changed

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

openapi.yaml

@@ -4294,7 +4294,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=<hex>` along with
         `X-Relay-Event` and `X-Relay-Timestamp`. Custom `headers` are included on delivery; Relay's

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
+- 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.
 - `handleAgentDisconnect` accepts an optional `{ deregister?: boolean }` argument, and `POST /v1/agents/disconnect` accepts an optional `{ deregister?: boolean }` body.

packages/engine/src/__tests__/conformance/agentNodeExitEvents.test.ts

@@ -0,0 +1,237 @@
+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<void> },
+    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;
+  }
+
+  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');
+    // 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 }));
+
+    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',
+      // Spawn correlation: the exit carries the invocation the agent registered under.
+      invocation_id: 'inv_spawn_a',
+    });
+  });
+
+  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('(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.
+    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);
+  });
+});

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);