SDKs: node provider clients (TS, Python, Swift) (#243)

* SDKs: node provider clients (TS, Python, Swift) + node.spawn/node.message frames

Add a node provider client to each relaycast SDK: connect to /v1/node/ws with a
node token, declare provider identity, register capabilities with per-capability
acceptance, run action.invoke handlers, heartbeat, reconnect with a fresh
instance id, and deregister. Handler context exposes sendMessage and spawnAgent.

Wire the ctx helpers to two node-WS request frames: node.spawn (capacity-direct,
routing dispatchCapacitySpawn) and node.message (post + observer fanout).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Test node.message attribution is scoped to the node workspace

Assert a node token cannot post as an agent from another workspace: the
`from` lookup is workspace-scoped, so a cross-workspace name is rejected
with agent_not_found and no message leaks into the other workspace.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Scope node.spawn to the connection's own node

Drop target_node_id from the node.spawn frame: ctx.spawnAgent addresses the
provider's own node capacity executor, and the engine always dispatches to the
connection's node. A node credential cannot direct a spawn at another node —
cross-node placement is workspace-level with agent/workspace credentials.

Conformance asserts the spawn lands on the connection's own node and never
reaches a second node's capacity.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: reconnect on handshake drop; drop unsupported thread_id

- TS and Python clients now treat a transport drop during the register
  handshake as a reconnect (only a protocol rejection hard-fails), matching
  Swift. Fixes a Python hang and a TS non-reconnect when the socket closes
  before the node.register reply.
- Remove thread_id from node.message: the field was accepted by the wire
  schema and all three SDKs but silently dropped by the engine (postMessage
  posts top-level only). node.message is a top-level-post surface; thread
  replies are not part of it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: validate acceptance, sync handlers, spawn input type

- All three clients hard-fail registration when the reply omits a well-formed
  accepted_capabilities list (a missing/corrupt list means acceptance can't be
  confirmed); a malformed acceptance entry counts as rejected (fail-safe). Swift
  previously dropped malformed entries and could register on a corrupt reply.
- Python handlers may be sync or async (awaited only when awaitable), matching
  the TypeScript client.
- Swift spawnAgent takes [String: JSONValue] so callers can't build a node.spawn
  frame the engine's object-typed input schema rejects.
- Strengthen the reconnect test to complete re-registration after the drop.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* ctx.sendMessage posts through the canonical message route, not a parallel frame

Remove the node.message WS frame and its engine handler. Node-attributed posts
now go through the canonical POST /v1/channels/:name/messages route, which
accepts a node token (new `sender` auth requirement) with a required `from` —
an agent resolved strictly within the node token's workspace. Delivery routing,
observer events, webhooks, triggers, and idempotency come free because it is the
one posting path (spec §5.3: only the hosting connection differs).

ctx.sendMessage in all three SDKs becomes an HTTP call to that route with the
node token. node.spawn stays a WS frame (dispatchCapacitySpawn duplicates no
existing surface).

Conformance covers node-token posting: delivery to node-hosted recipients,
required `from`, unknown-agent and cross-workspace rejection, and `from`
rejected on an agent token.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: settle registration on reconnect exhaustion; null-safe reply

- All three clients reject the registration waiter when reconnects are exhausted
  before registering, so serve()/whenRegistered() no longer hang forever.
- TS onRegistered tolerates a null/undefined reply data (treats it as an invalid
  reply hard-fail) instead of throwing a TypeError that misroutes to reconnect.
- Guard the Python reconnect tests against a race reading the re-register frame.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: registration-state guard, HTTP base/encoding hardening

- Python: reconnect exhaustion no longer overwrites a completed registration, so
  wait_registered() keeps returning success after a later drop.
- ctx.sendMessage normalizes a ws(s):// base to http(s):// (all three SDKs) and
  encodes the channel as a single path segment (Swift no longer leaves `/`
  unescaped). Swift treats a non-HTTP or non-envelope 2xx response as a failure
  instead of returning null.
- Poll for the node-token delivery fanout in conformance instead of a fixed
  sleep, matching the file's other delivery assertions.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: verify full acceptance, drain on stop, stop rejects register

- Registration hard-fails unless every requested capability is acknowledged in
  the reply (an empty/partial accepted_capabilities can no longer pass) — all
  three SDKs.
- stop() drains in-flight invoke handlers (bounded) before closing the socket so
  their action.result frames aren't dropped, and rejects a still-pending
  registration waiter so whenRegistered() can't hang when stopped pre-register.
- Swift clamps the reconnect max delay against the sanitized base delay (no
  negative backoff) and its HTTP tests clear the mock URL handler via defer.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: drain gate, snapshot validation, drain self-safety

- stop() stops accepting new action.invoke frames once draining, so a late
  invoke can't start a handler whose result would be dropped after close (all
  three SDKs); the engine reschedules the undispatched invocation.
- Registration acceptance is validated against the capability snapshot sent in
  the register frame, not the live map, so a capability added mid-handshake
  can't trigger a false missing-acknowledgement failure (all three SDKs).
- Python stop() excludes the caller's own task from the drain, so a handler that
  calls stop() no longer deadlocks (and gets cancelled, skipping deregister).
- TS clears the drain timeout when tasks finish early; Swift stops heartbeats
  before the drain.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: willwashburn

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

@@ -423,6 +423,118 @@ describe('node providers', () => {
     expect(actionRows.map((a) => a.name).sort()).toEqual(['run-etl']);
   });
 
+  it('routes a node.spawn frame to the connection\'s own node capacity, bypassing any spawn shadow', async () => {
+    const ws = await createWorkspace(stack.app, 'np-node-spawn');
+    await enrollNode(ws, 'node_a', 'alpha');
+    // node_a: broker provider offers native spawn:claude capacity.
+    const broker = await attachProvider(ws.workspaceId, 'node_a', 'alpha', 'default', [{ name: 'spawn:claude', kind: 'capacity' }]);
+    // A policy provider shadows it with a spawn:claude action, and is where the
+    // handler runs: its ctx.spawnAgent must delegate to broker capacity, never
+    // re-enter its own shadow.
+    const policy = await attachProvider(ws.workspaceId, 'node_a', 'alpha', 'policy', [{ name: 'spawn:claude', kind: 'action' }]);
+    // A second node with its own spawn capacity: a node.spawn from node_a must
+    // never reach it — a node credential cannot direct a spawn at another node.
+    await enrollNode(ws, 'node_b', 'beta');
+    const otherBroker = await attachProvider(ws.workspaceId, 'node_b', 'beta', 'default', [{ name: 'spawn:claude', kind: 'capacity' }]);
+
+    await policy.handle.handleMessage(JSON.stringify({
+      v: 1,
+      id: 'spawn-1',
+      type: 'node.spawn',
+      input: { cli: 'claude', name: 'worker-1' },
+    }));
+
+    // Delegated to node_a's broker capacity...
+    expect(broker.sock.ofType('action.invoke').at(-1)).toMatchObject({ action: 'spawn:claude' });
+    // ...not back into the policy provider's shadow action...
+    expect(policy.sock.ofType('action.invoke')).toHaveLength(0);
+    // ...and never crossing to the other node.
+    expect(otherBroker.sock.ofType('action.invoke')).toHaveLength(0);
+    const reply = policy.sock.ofType('reply').at(-1) as { ok?: boolean; id?: string; data?: { invocation_id?: string; handler_node_id?: string } };
+    expect(reply).toMatchObject({ ok: true, id: 'spawn-1' });
+    expect(typeof reply.data?.invocation_id).toBe('string');
+    expect(reply.data?.handler_node_id).toBe('node_a');
+  });
+
+  // ctx.sendMessage posts through the canonical message route with a node token
+  // and a required `from`, so node-attributed posts get delivery routing,
+  // observer events, and triggers by construction (no parallel posting path).
+  async function enrollNodeWithToken(ws: { workspaceKey: string }, nodeId: string, name: string): Promise<string> {
+    const res = await stack.app.request('/v1/nodes', {
+      method: 'POST',
+      headers: { 'content-type': 'application/json', authorization: `Bearer ${ws.workspaceKey}` },
+      body: JSON.stringify({ node_id: nodeId, name, role: 'broker', capabilities: [], max_agents: 4, tags: ['test'], version: 'v0' }),
+    });
+    expect(res.status).toBe(201);
+    return (await res.json() as { data: { token: string } }).data.token;
+  }
+
+  function postMessage(token: string, channel: string, body: Record<string, unknown>) {
+    return stack.app.request(`/v1/channels/${channel}/messages`, {
+      method: 'POST',
+      headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` },
+      body: JSON.stringify(body),
+    });
+  }
+
+  it('posts a node-token message attributed to `from`, delivered to node-hosted recipients', async () => {
+    const ws = await createWorkspace(stack.app, 'nt-msg');
+    await registerAgent(stack.app, ws.workspaceKey, 'poster');
+    const nodeToken = await enrollNodeWithToken(ws, 'node_a', 'alpha');
+    const py = await attachProvider(ws.workspaceId, 'node_a', 'alpha', 'py', [{ name: 'run-etl', kind: 'action' }]);
+    // A worker agent hosted by py joins #general, so it is a delivery recipient.
+    await py.handle.handleMessage(JSON.stringify({ v: 1, id: 'agent-1', type: 'agent.register', name: 'worker', session_ref: 'pty://py/worker' }));
+    py.sock.received.length = 0;
+
+    const res = await postMessage(nodeToken, 'general', { text: 'etl finished', from: 'poster' });
+    expect(res.status).toBe(201);
+    const body = await res.json() as { data: { agent_name: string; text: string } };
+    expect(body.data).toMatchObject({ agent_name: 'poster', text: 'etl finished' });
+
+    // Poll for the async fanout instead of a fixed sleep, to avoid CI flakiness.
+    for (let i = 0; i < 50 && deliverFramesOfType(py.sock, 'message.created').length === 0; i++) {
+      await new Promise((r) => setTimeout(r, 10));
+    }
+    // Delivery routing came for free from the canonical route.
+    expect(deliverFramesOfType(py.sock, 'message.created').length).toBeGreaterThanOrEqual(1);
+  });
+
+  it('requires `from` on a node-token message', async () => {
+    const ws = await createWorkspace(stack.app, 'nt-msg-from-required');
+    const nodeToken = await enrollNodeWithToken(ws, 'node_a', 'alpha');
+    const res = await postMessage(nodeToken, 'general', { text: 'hi' });
+    expect(res.status).toBe(400);
+    expect((await res.json() as { error: { code: string } }).error.code).toBe('from_required');
+  });
+
+  it('rejects a node-token message whose `from` agent does not exist', async () => {
+    const ws = await createWorkspace(stack.app, 'nt-msg-unknown');
+    const nodeToken = await enrollNodeWithToken(ws, 'node_a', 'alpha');
+    const res = await postMessage(nodeToken, 'general', { text: 'hi', from: 'ghost' });
+    expect(res.status).toBe(404);
+    expect((await res.json() as { error: { code: string } }).error.code).toBe('agent_not_found');
+  });
+
+  it('scopes node-token `from` to the node workspace, rejecting a cross-workspace agent', async () => {
+    const other = await createWorkspace(stack.app, 'nt-msg-other-ws');
+    await registerAgent(stack.app, other.workspaceKey, 'outsider');
+
+    const ws = await createWorkspace(stack.app, 'nt-msg-scope');
+    const nodeToken = await enrollNodeWithToken(ws, 'node_a', 'alpha');
+    const res = await postMessage(nodeToken, 'general', { text: 'hi', from: 'outsider' });
+    expect(res.status).toBe(404);
+    expect((await res.json() as { error: { code: string } }).error.code).toBe('agent_not_found');
+  });
+
+  it('rejects `from` on an agent-token message', async () => {
+    const ws = await createWorkspace(stack.app, 'nt-msg-agent-from');
+    const agent = await registerAgent(stack.app, ws.workspaceKey, 'poster');
+    const other = await registerAgent(stack.app, ws.workspaceKey, 'other');
+    const res = await postMessage(agent.token, 'general', { text: 'hi', from: other.name });
+    expect(res.status).toBe(400);
+    expect((await res.json() as { error: { code: string } }).error.code).toBe('from_not_allowed');
+  });
+
   it('removes a provider through DELETE /v1/nodes/:node/providers/:name', async () => {
     const ws = await createWorkspace(stack.app, 'np-delete');
     await enrollNode(ws, 'node_a', 'alpha');

packages/engine/src/auth/tokenKind.ts

@@ -65,7 +65,7 @@ export function getAuthTokenKind(token: string): AuthTokenKind | undefined {
   return parseAuthToken(token)?.kind;
 }
 
-export function requiredTokenMessage(kind: Exclude<AuthRequire, 'any'>): string {
+export function requiredTokenMessage(kind: Exclude<AuthRequire, 'any' | 'sender'>): string {
   return tokenKindConfig[kind].requiredMessage;
 }
 
@@ -77,6 +77,9 @@ const allowedKindsByRequire = {
   // `any` intentionally means user/workspace API credentials, not transport-only
   // node credentials or scoped observer credentials.
   any: ['workspace', 'agent'],
+  // `sender` is a principal that can post a message: a workspace key, an agent
+  // (posts as itself), or a node token (posts as an explicit `from` agent).
+  sender: ['workspace', 'agent', 'node'],
 } as const satisfies Record<AuthRequire, readonly AuthTokenKind[]>;
 
 export type AuthRequirementValidation =
@@ -108,7 +111,7 @@ export function validateTokenRequirement(
     };
   }
 
-  if (require === 'any') {
+  if (require === 'any' || require === 'sender') {
     return { ok: false, code: 'unauthorized', message: 'Invalid token format' };
   }
 

packages/engine/src/engine/node.ts

@@ -30,7 +30,12 @@ import {
   removeProvider,
   upsertProvider,
 } from './nodeProvider.js';
-import { completeNodeInvocation, rescheduleInvocationsForLostNode, rescheduleNodeInvocation } from './action.js';
+import {
+  completeNodeInvocation,
+  dispatchCapacitySpawn,
+  rescheduleInvocationsForLostNode,
+  rescheduleNodeInvocation,
+} from './action.js';
 import { emitInvocationCompletionEffects } from './invocationCompletion.js';
 import type { InvocationCompletionDeps } from './invocationCompletion.js';
 import { ackDeliveriesUpToSeq, deliverPendingToNode } from './delivery.js';
@@ -1425,6 +1430,30 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg
       case 'node.deregister':
         await deregisterProvider(args.db, args.registry, args.workspaceId, args.nodeId, frameProviderName);
         return;
+      case 'node.spawn': {
+        // Handler-context `ctx.spawnAgent`: capacity-direct delegation to this
+        // connection's own node capacity executor (the broker provider).
+        // Bypasses action dispatch so a `spawn:<harness>` shadow handler cannot
+        // re-enter itself. The target is always the connection's node — a node
+        // credential cannot direct a spawn at another node.
+        const result = await dispatchCapacitySpawn(
+          args.db,
+          args.workspaceId,
+          {
+            input: message.input,
+            target_node_id: args.nodeId,
+          },
+          { nodeConnections: args.registry },
+        );
+        sendControl(args.socket, {
+          v: 1,
+          id: requestId(message),
+          type: 'reply',
+          ok: true,
+          data: result as Record<string, unknown>,
+        });
+        return;
+      }
       case 'agent.register': {
         const registered = await registerAgentViaNode(args.db, args.workspaceId, args.nodeId, frameProviderName, message);
         // The relay broker's node_control client awaits a `reply` frame keyed by

packages/engine/src/middleware/auth.ts

@@ -127,6 +127,13 @@ export const requireWorkspaceKey = makeAuthMiddleware('workspace');
 /** Accepts either a workspace key or an agent token. */
 export const requireAuth = makeAuthMiddleware('any');
 
+/**
+ * Accepts a workspace key, an agent token, or a node token — the principals that
+ * can post a message. An agent posts as itself; a node posts as an explicit
+ * `from` agent resolved within the node's workspace.
+ */
+export const requireSender = makeAuthMiddleware('sender');
+
 /** Requires an agent token (`at_live_...`). */
 export const requireAgentToken = makeAuthMiddleware('agent');
 

packages/engine/src/ports/auth.ts

@@ -6,7 +6,7 @@ export type Agent = typeof agents.$inferSelect;
 export type Node = typeof nodes.$inferSelect;
 export type ObserverToken = typeof observerTokens.$inferSelect;
 
-export type AuthRequire = 'workspace' | 'agent' | 'node' | 'observer' | 'any';
+export type AuthRequire = 'workspace' | 'agent' | 'node' | 'observer' | 'any' | 'sender';
 
 export type AuthResult =
   | { ok: true; workspace: Workspace; agent?: Agent; node?: Node; observerToken?: ObserverToken }

packages/engine/src/routes/message.ts

@@ -1,7 +1,9 @@
 import { Hono } from 'hono';
 import { z } from 'zod';
+import { and, eq } from 'drizzle-orm';
+import { agents } from '../db/schema.js';
 import type { AppEnv } from '../env.js';
-import { requireAuth, requireWorkspaceRead } from '../middleware/auth.js';
+import { requireSender, requireWorkspaceRead } from '../middleware/auth.js';
 import { rateLimit } from '../middleware/rateLimit.js';
 import { jsonIdempotentOk, parseIdempotencyKey, runIdempotent } from '../middleware/idempotency.js';
 import * as messageEngine from '../engine/message.js';
@@ -40,6 +42,9 @@ const postMessageSchema = z.object({
   data: z.record(z.string(), z.unknown()).nullable().optional(),
   content_type: z.string().optional(),
   mode: z.enum(['wait', 'steer']).default('wait'),
+  // Node-token posts attribute the message to `from` — an agent name resolved
+  // strictly within the node token's workspace. Agent-token posts must omit it.
+  from: z.string().min(1).optional(),
 });
 
 type ValidationFailure = { error: { issues: Array<{ path: PropertyKey[] }> } };
@@ -58,18 +63,19 @@ function postMessageInvalidMessage(failure: ValidationFailure) {
 // POST /v1/channels/:name/messages - post a message
 messageRoutes.post(
   '/channels/:name/messages',
-  requireAuth,
+  requireSender,
   rateLimit,
   async (c) => {
     try {
       const db = c.get('db');
       const workspace = c.get('workspace');
       const agent = c.get('agent');
+      const node = c.get('node');
       const parsed = await parseJsonBody(c, postMessageSchema, postMessageInvalidMessage);
       if (!parsed.ok) {
         return parsed.response;
       }
-      const { text, blocks, attachments, data, content_type, mode } = parsed.data;
+      const { text, blocks, attachments, data, content_type, mode, from } = parsed.data;
 
       const { key: idempotencyKey, error: idempotencyError } = parseIdempotencyKey(c.req.header('Idempotency-Key'));
       if (idempotencyError) {
@@ -84,33 +90,55 @@ messageRoutes.post(
         return jsonNotFound(c, 'channel_not_found', `Channel "${channelName}" not found`);
       }
 
-      // Determine agent ID (from agent token or body)
-      const agentId = agent?.id;
-      if (!agentId) {
-        return jsonError(c, 'agent_token_required', 'Agent token required to post messages', 403);
+      // Resolve the sender. An agent token posts as itself; a node token posts
+      // as `from`, an agent resolved strictly within the node's workspace (so a
+      // node credential can never attribute a message to another workspace).
+      let senderAgentId: string;
+      let senderAgentName: string;
+      if (agent) {
+        if (from !== undefined) {
+          return jsonError(c, 'from_not_allowed', '"from" is only valid with a node token', 400);
+        }
+        senderAgentId = agent.id;
+        senderAgentName = agent.name;
+      } else if (node) {
+        if (!from) {
+          return jsonError(c, 'from_required', 'A node token must set "from" (an agent name in the node workspace)', 400);
+        }
+        const [fromAgent] = await db
+          .select({ id: agents.id, name: agents.name })
+          .from(agents)
+          .where(and(eq(agents.workspaceId, workspace.id), eq(agents.name, from)));
+        if (!fromAgent) {
+          return jsonNotFound(c, 'agent_not_found', `Agent "${from}" not found in this workspace`);
+        }
+        senderAgentId = fromAgent.id;
+        senderAgentName = fromAgent.name;
+      } else {
+        return jsonError(c, 'agent_token_required', 'Agent or node token required to post messages', 403);
       }
 
       const mailbox = resolveMailboxConfig(c.get('engine').config, workspace.id);
       const toMessageCreatedEventData = (data: Awaited<ReturnType<typeof messageEngine.postMessage>>) => buildMessageCreatedEventData(data, {
         channelName,
-        fromName: agent?.name ?? 'unknown',
+        fromName: senderAgentName,
         mode,
       });
 
       const idempotent = await runIdempotent({
         workspaceId: workspace.id,
-        actorId: agentId,
+        actorId: senderAgentId,
         scope: `channel-message:${channel.id}`,
         key: idempotencyKey,
         status: 201,
-        fingerprint: JSON.stringify({ channelId: channel.id, text, blocks, attachments, data, content_type, mode }),
+        fingerprint: JSON.stringify({ channelId: channel.id, text, blocks, attachments, data, content_type, mode, from }),
         kv: c.get('engine').kv,
         operation: () =>
           messageEngine.postMessage(
             db,
             workspace.id,
             channel.id,
-            agentId,
+            senderAgentId,
             { text, blocks, attachments, data, content_type, mode },
             { mailbox },
           ),
@@ -146,7 +174,7 @@ messageRoutes.post(
         if (_delivery_rejections && _delivery_rejections.length > 0) {
           runInBackground(
             c,
-            notifyDeliveryRejections(c, agentId, _delivery_rejections),
+            notifyDeliveryRejections(c, senderAgentId, _delivery_rejections),
             'fanout delivery rejected',
           );
         }
@@ -170,8 +198,8 @@ messageRoutes.post(
                 id: String(publicData.id),
                 channel_id: channel.id,
                 channel_name: channelName,
-                agent_id: agentId,
-                agent_name: agent?.name,
+                agent_id: senderAgentId,
+                agent_name: senderAgentName,
                 text: String(publicData.text ?? text),
                 mentions: Array.isArray(publicData.mentions) ? publicData.mentions as string[] : [],
                 metadata: (publicData.metadata ?? null) as Record<string, unknown> | null,

packages/sdk-python/src/relay_sdk/__init__.py

@@ -10,6 +10,7 @@
     sanitize_agent_relay_distinct_id,
 )
 from .errors import RelayError
+from .node import NodeProvider, NodeRegistrationError
 from .relay import AsyncRelay, Relay
 from .ws import WsClient
 
@@ -23,6 +24,8 @@
     "AsyncHttpClient",
     "AsyncRelay",
     "HttpClient",
+    "NodeProvider",
+    "NodeRegistrationError",
     "Relay",
     "RelayError",
     "SDK_VERSION",