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>

Author: willwashburn

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

@@ -9,7 +9,7 @@ import {
   contextUpdatesOfType,
   type TestStack,
 } from './harness.js';
-import { actions, agents, messages, nodeProviders, nodes, workspaceEvents } from '../../db/schema.js';
+import { actions, agents, nodeProviders, nodes } from '../../db/schema.js';
 
 type Cap = { name: string; kind?: string; global?: boolean; queue?: boolean };
 
@@ -456,76 +456,80 @@ describe('node providers', () => {
     expect(reply.data?.handler_node_id).toBe('node_a');
   });
 
-  it('posts a channel message from a node.message frame and fans out to observers', async () => {
-    const ws = await createWorkspace(stack.app, 'np-node-message');
-    await registerAgent(stack.app, ws.workspaceKey, 'reporter');
-    await enrollNode(ws, 'node_a', 'alpha');
-    const py = await attachProvider(ws.workspaceId, 'node_a', 'alpha', 'py', [{ name: 'run-etl', kind: 'action' }]);
+  // 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;
+  }
 
-    await py.handle.handleMessage(JSON.stringify({
-      v: 1,
-      id: 'msg-1',
-      type: 'node.message',
-      to: 'general',
-      from: 'reporter',
-      text: 'etl finished',
-    }));
+  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),
+    });
+  }
 
-    const reply = py.sock.ofType('reply').at(-1) as { ok?: boolean; id?: string; data?: { id?: string; text?: string } };
-    expect(reply).toMatchObject({ ok: true, id: 'msg-1' });
-    expect(reply.data?.text).toBe('etl finished');
+  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 rows = await stack.runtime.handle.db
-      .select({ body: messages.body, agentId: messages.agentId })
-      .from(messages)
-      .where(eq(messages.workspaceId, ws.workspaceId));
-    expect(rows.map((r) => r.body)).toContain('etl finished');
-
-    const events = await stack.runtime.handle.db
-      .select({ type: workspaceEvents.type })
-      .from(workspaceEvents)
-      .where(and(eq(workspaceEvents.workspaceId, ws.workspaceId), eq(workspaceEvents.type, 'message.created')));
-    expect(events.length).toBeGreaterThanOrEqual(1);
-  });
+    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' });
 
-  it('rejects a node.message from an unknown agent or to an unknown channel', async () => {
-    const ws = await createWorkspace(stack.app, 'np-node-message-errors');
-    await registerAgent(stack.app, ws.workspaceKey, 'reporter');
-    await enrollNode(ws, 'node_a', 'alpha');
-    const py = await attachProvider(ws.workspaceId, 'node_a', 'alpha', 'py', [{ name: 'run-etl', kind: 'action' }]);
+    await new Promise((r) => setTimeout(r, 60));
+    // Delivery routing came for free from the canonical route.
+    expect(deliverFramesOfType(py.sock, 'message.created').length).toBeGreaterThanOrEqual(1);
+  });
 
-    await py.handle.handleMessage(JSON.stringify({
-      v: 1, id: 'bad-agent', type: 'node.message', to: 'general', from: 'ghost', text: 'hi',
-    }));
-    expect(py.sock.ofType('error').at(-1)).toMatchObject({ id: 'bad-agent', code: 'agent_not_found' });
+  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');
+  });
 
-    await py.handle.handleMessage(JSON.stringify({
-      v: 1, id: 'bad-channel', type: 'node.message', to: 'nowhere', from: 'reporter', text: 'hi',
-    }));
-    expect(py.sock.ofType('error').at(-1)).toMatchObject({ id: 'bad-channel', code: 'channel_not_found' });
+  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.message attribution to the node workspace, rejecting a cross-workspace from', async () => {
-    // The `from` agent resolves only within the authenticated node's workspace,
-    // so a node token cannot post as an agent that exists in another workspace.
-    const other = await createWorkspace(stack.app, 'np-msg-other-ws');
+  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, 'np-msg-scope');
-    await enrollNode(ws, 'node_a', 'alpha');
-    const py = await attachProvider(ws.workspaceId, 'node_a', 'alpha', 'py', [{ name: 'run-etl', kind: 'action' }]);
+    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');
+  });
 
-    await py.handle.handleMessage(JSON.stringify({
-      v: 1, id: 'cross-ws', type: 'node.message', to: 'general', from: 'outsider', text: 'hi',
-    }));
-    expect(py.sock.ofType('error').at(-1)).toMatchObject({ id: 'cross-ws', code: 'agent_not_found' });
-
-    // No message leaked into the other workspace under the outsider's name.
-    const leaked = await stack.runtime.handle.db
-      .select({ id: messages.id })
-      .from(messages)
-      .where(eq(messages.workspaceId, other.workspaceId));
-    expect(leaked).toHaveLength(0);
+  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 () => {

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

@@ -39,11 +39,6 @@ import {
 import { emitInvocationCompletionEffects } from './invocationCompletion.js';
 import type { InvocationCompletionDeps } from './invocationCompletion.js';
 import { ackDeliveriesUpToSeq, deliverPendingToNode } from './delivery.js';
-import { getChannel } from './channel.js';
-import { postMessage } from './message.js';
-import { buildMessageCreatedEventData } from './deliveryWire.js';
-import { transformForClient } from './wsTransform.js';
-import { appendAndPublishWorkspaceEvent } from './workspaceEvents.js';
 
 type Db = ReturnType<typeof getDb>;
 type NodeRow = typeof nodes.$inferSelect;
@@ -1459,58 +1454,6 @@ export async function handleNodeControlMessage(args: HandleNodeControlMessageArg
         });
         return;
       }
-      case 'node.message': {
-        // Handler-context `ctx.sendMessage`: post a channel message attributed
-        // to `from` (an agent resolved by name in the workspace).
-        const channel = await getChannel(args.db, args.workspaceId, message.to);
-        if (!channel) {
-          throw codedError(`Channel "${message.to}" not found`, 'channel_not_found', 404);
-        }
-        const [fromAgent] = await args.db
-          .select({ id: agents.id, name: agents.name })
-          .from(agents)
-          .where(and(eq(agents.workspaceId, args.workspaceId), eq(agents.name, message.from)));
-        if (!fromAgent) {
-          throw codedError(`Agent "${message.from}" not found`, 'agent_not_found', 404);
-        }
-        const posted = await postMessage(args.db, args.workspaceId, channel.id, fromAgent.id, {
-          text: message.text,
-          ...(message.data ? { data: message.data } : {}),
-          ...(message.mode ? { mode: message.mode } : {}),
-        });
-        // Fan out to workspace observers, mirroring the message route's publish.
-        if (args.completionDeps) {
-          const eventData = buildMessageCreatedEventData(posted as unknown as Record<string, unknown>, {
-            channelName: message.to,
-            fromName: fromAgent.name,
-            ...(message.mode ? { mode: message.mode } : {}),
-          });
-          const payload = transformForClient({
-            type: 'message.created',
-            workspace_id: args.workspaceId,
-            channel_id: channel.id,
-            data: eventData,
-            timestamp: new Date().toISOString(),
-          });
-          await appendAndPublishWorkspaceEvent(
-            { db: args.completionDeps.db, realtime: args.completionDeps.realtime },
-            args.workspaceId,
-            { type: 'message.created', channelId: channel.id, payload },
-          );
-        }
-        const { _deliveries, _delivery_rejections, ...publicMessage } = posted as typeof posted & {
-          _deliveries?: unknown;
-          _delivery_rejections?: unknown;
-        };
-        sendControl(args.socket, {
-          v: 1,
-          id: requestId(message),
-          type: 'reply',
-          ok: true,
-          data: publicMessage 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,