From b2c08d72054e9a1d0198da91af33c852f8b73730 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Tue, 30 Jun 2026 06:32:27 -0700 Subject: [PATCH 1/2] fix(engine): deliver ephemeral node events to http_push nodes Reactions, read receipts, and presence/context updates were pushed only to WebSocket nodes: `nodeDeliver` and `nodeContext` hard-filtered recipients to `ws`/`fleet_ws`/`direct_ws`, so an agent bound to a `kind: "http_push"` node received durable messages but silently missed every ephemeral event. Extract the http_push auth/HMAC header builder into a shared `httpPushDispatch` module (reused by the durable dispatch path) and add a best-effort `postEphemeralEventToHttpPushNode` that mirrors the fire-and-forget semantics of a WS frame (no delivery row, ack, or retry). Route http_push recipients through it from both node-event paths, threading `environment` for SSRF parity with durable dispatch (defaults to strict when omitted). Adds a regression test proving message.reacted / message.read / agent.status.* now POST to an http_push receiver, and updates the durable-delivery contract tests to be event-type-aware (presence POSTs legitimately fire during them now). Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 5 + .../httpPushEphemeralEvents.test.ts | 157 ++++++++++++++++++ .../conformance/nodeDeliveryContracts.test.ts | 97 +++++++---- packages/engine/src/adapters/node/index.ts | 2 +- .../engine/src/engine/httpPushDispatch.ts | 104 ++++++++++++ .../engine/src/engine/invocationCompletion.ts | 3 +- packages/engine/src/engine/nodeContext.ts | 49 +++++- packages/engine/src/engine/nodeDeliver.ts | 80 ++++++--- packages/engine/src/routes/action.ts | 1 + packages/engine/src/routes/agent.ts | 2 + packages/engine/src/routes/deliveryRouting.ts | 34 +--- packages/engine/src/routes/fanout.ts | 2 + packages/engine/src/routes/reaction.ts | 2 + packages/engine/src/routes/receipt.ts | 1 + 14 files changed, 448 insertions(+), 91 deletions(-) create mode 100644 packages/engine/src/__tests__/conformance/httpPushEphemeralEvents.test.ts create mode 100644 packages/engine/src/engine/httpPushDispatch.ts diff --git a/README.md b/README.md index 26e66db5..a6970777 100644 --- a/README.md +++ b/README.md @@ -439,6 +439,11 @@ on any 2xx HTTP response, and `response` acks when the response body declares an Manual HTTP receivers ack by calling `/v1/deliveries/:id/ack` with the bound agent's token, so pure webhook endpoints should use `on_2xx` or `response` unless they can securely hold that token. +HTTP push nodes also receive the ephemeral channel/workspace events a WebSocket node +gets — reactions (`message.reacted`), read receipts (`message.read`), and presence / +status updates — as best-effort POSTs to the same delivery URL (same auth/signing, +no delivery row or ack). Each carries a `type` plus the event `data`; receivers that +only want durable messages can filter on the `X-Relaycast-Event` header. Queue/cron-backed deployments must call `sweepDueHttpPushDeliveries` from a scheduled handler to retry queued HTTP push deliveries whose `next_attempt_at` is due; the Node self-host adapter runs that sweep on its local maintenance timer. diff --git a/packages/engine/src/__tests__/conformance/httpPushEphemeralEvents.test.ts b/packages/engine/src/__tests__/conformance/httpPushEphemeralEvents.test.ts new file mode 100644 index 00000000..ca6c57c1 --- /dev/null +++ b/packages/engine/src/__tests__/conformance/httpPushEphemeralEvents.test.ts @@ -0,0 +1,157 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + makeNodeStack, + createWorkspace, + registerAgent, + type TestStack, +} from './harness.js'; + +/** + * Regression: an http_push node must receive the ephemeral channel/workspace + * events a WebSocket node gets — reactions, read receipts, and presence/status + * updates. These previously went through WS-only delivery paths + * (`nodeDeliver`/`nodeContext` hard-filtered to ws/fleet_ws/direct_ws) and were + * silently dropped for http_push receivers. + */ +describe('http_push ephemeral event delivery', () => { + let stack: TestStack; + + beforeEach(() => { + stack = makeNodeStack({ ttlMs: 60_000 }); + }); + + afterEach(() => { + vi.restoreAllMocks(); + stack.close(); + }); + + async function createHttpNode(workspaceKey: string, name: string) { + const res = await stack.app.request('/v1/nodes', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${workspaceKey}` }, + body: JSON.stringify({ + name, + kind: 'http_push', + delivery: { url: 'https://receiver.example.test/relaycast', ack_mode: 'manual', auth: { type: 'none' } }, + }), + }); + expect(res.status).toBe(201); + return (await res.json()) as { data: { name: string } }; + } + + async function bindAgent(workspaceKey: string, nodeName: string, agentName: string) { + const res = await stack.app.request(`/v1/nodes/${nodeName}/agents`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${workspaceKey}` }, + body: JSON.stringify({ agent_name: agentName }), + }); + expect(res.status).toBe(201); + } + + async function waitFor(assertion: () => void | Promise, timeoutMs = 1000) { + const started = Date.now(); + let lastError: unknown; + while (Date.now() - started < timeoutMs) { + try { await assertion(); return; } catch (err) { lastError = err; await new Promise((r) => setTimeout(r, 10)); } + } + throw lastError; + } + + function postsOfType(fetchMock: ReturnType, type: string): Array> { + return fetchMock.mock.calls + .map((call) => { + try { return JSON.parse((call[1] as RequestInit).body as string) as Record; } + catch { return null; } + }) + .filter((body): body is Record => !!body && body.type === type); + } + + it('POSTs message.reacted to an http_push node', async () => { + const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('', { status: 202 })); + const ws = await createWorkspace(stack.app, 'react-http'); + const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); + await registerAgent(stack.app, ws.workspaceKey, 'bob'); + const node = await createHttpNode(ws.workspaceKey, 'react-node'); + await bindAgent(ws.workspaceKey, node.data.name, 'bob'); + + const post = await stack.app.request('/v1/channels/general/messages', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${alice.token}` }, + body: JSON.stringify({ text: 'react to me' }), + }); + expect(post.status).toBe(201); + const messageId = ((await post.json()) as { data: { id: string } }).data.id; + await waitFor(() => expect(postsOfType(fetchMock, 'message.created')).toHaveLength(1)); + + const react = await stack.app.request(`/v1/messages/${messageId}/reactions`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${alice.token}` }, + body: JSON.stringify({ emoji: '👍', channel_id: 'general' }), + }); + expect(react.status).toBe(201); + + await waitFor(() => { + const reacted = postsOfType(fetchMock, 'message.reacted'); + expect(reacted).toHaveLength(1); + expect(reacted[0].data).toMatchObject({ emoji: '👍', agent_name: 'alice', action: 'added' }); + }); + }); + + it('POSTs message.read to an http_push node', async () => { + const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('', { status: 202 })); + const ws = await createWorkspace(stack.app, 'read-http'); + const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); + await registerAgent(stack.app, ws.workspaceKey, 'bob'); + const node = await createHttpNode(ws.workspaceKey, 'read-node'); + await bindAgent(ws.workspaceKey, node.data.name, 'bob'); + + const post = await stack.app.request('/v1/channels/general/messages', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${alice.token}` }, + body: JSON.stringify({ text: 'read me' }), + }); + expect(post.status).toBe(201); + const messageId = ((await post.json()) as { data: { id: string } }).data.id; + await waitFor(() => expect(postsOfType(fetchMock, 'message.created')).toHaveLength(1)); + + // alice (a channel member) reads the message; bob's http_push node should + // receive the read receipt for the channel. + const read = await stack.app.request(`/v1/messages/${messageId}/read`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${alice.token}` }, + body: JSON.stringify({}), + }); + expect(read.status).toBe(200); + + await waitFor(() => { + const reads = postsOfType(fetchMock, 'message.read'); + expect(reads.length).toBeGreaterThanOrEqual(1); + expect(reads[0].data).toMatchObject({ agent_name: 'alice' }); + }); + }); + + it('POSTs agent presence/status updates to an http_push node', async () => { + const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('', { status: 202 })); + const ws = await createWorkspace(stack.app, 'presence-http'); + const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); + await registerAgent(stack.app, ws.workspaceKey, 'bob'); + const node = await createHttpNode(ws.workspaceKey, 'presence-node'); + await bindAgent(ws.workspaceKey, node.data.name, 'bob'); + + // alice changes her status; bob (bound to the http_push node) should be + // notified via a presence context POST. + const patch = await stack.app.request('/v1/agents/alice', { + method: 'PATCH', + headers: { 'content-type': 'application/json', authorization: `Bearer ${ws.workspaceKey}` }, + body: JSON.stringify({ status: 'idle' }), + }); + expect(patch.status).toBe(200); + + await waitFor(() => { + const presence = postsOfType(fetchMock, 'agent.status.idle'); + expect(presence).toHaveLength(1); + expect(presence[0]).toMatchObject({ topic: 'presence' }); + expect(presence[0].data).toMatchObject({ agent_name: 'alice', status: 'idle' }); + }); + }); +}); diff --git a/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts b/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts index 662a833b..c45b7eb1 100644 --- a/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts +++ b/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts @@ -104,6 +104,35 @@ describe('node delivery contracts', () => { throw lastError; } + // http_push nodes also receive ephemeral events (presence/status, reactions, + // receipts) as POSTs. Durable-message contract tests care only about the + // message-delivery POSTs, so filter by the event header to ignore that noise. + function deliveryPosts( + fetchMock: ReturnType, + eventType = 'message.created', + ): Array<[string, RequestInit]> { + return fetchMock.mock.calls.filter((call) => { + const headers = (call[1] as RequestInit | undefined)?.headers as Record | undefined; + return headers?.['X-Relaycast-Event'] === eventType; + }) as Array<[string, RequestInit]>; + } + + // Type-aware fetch mock: message-delivery POSTs consume `messageResponses` in + // order (last entry repeats); every other POST (presence/reaction/receipt) + // resolves 202 so ephemeral noise never steals a scripted message response. + function mockMessageDeliveryFetch(messageResponses: Array<() => Response>) { + let index = 0; + return vi.spyOn(globalThis, 'fetch').mockImplementation((async (_url: unknown, init?: RequestInit) => { + const headers = init?.headers as Record | undefined; + if (headers?.['X-Relaycast-Event'] === 'message.created') { + const make = messageResponses[Math.min(index, messageResponses.length - 1)]; + index += 1; + return make(); + } + return new Response('', { status: 202 }); + }) as typeof globalThis.fetch); + } + it('dispatches to an http_push node with custom HMAC headers and manual ack semantics', async () => { const fetchMock = vi .spyOn(globalThis, 'fetch') @@ -129,8 +158,8 @@ describe('node delivery contracts', () => { }); expect(post.status).toBe(201); const messageId = ((await post.json()) as { data: { id: string } }).data.id; - await waitForAssertion(() => expect(fetchMock).toHaveBeenCalledTimes(1)); - const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + await waitForAssertion(() => expect(deliveryPosts(fetchMock)).toHaveLength(1)); + const [url, init] = deliveryPosts(fetchMock)[0]; expect(url).toBe('https://receiver.example.test/relaycast'); expect(init.redirect).toBe('error'); const headers = init.headers as Record; @@ -263,20 +292,16 @@ describe('node delivery contracts', () => { }); it('keeps response-mode http_push deliveries queued when 2xx omits an ack signal', async () => { - const fetchMock = vi - .spyOn(globalThis, 'fetch') - .mockResolvedValueOnce( - new Response(JSON.stringify({ ok: true }), { - status: 200, - headers: { 'content-type': 'application/json' }, - }), - ) - .mockResolvedValueOnce( - new Response(JSON.stringify({ ack: true }), { - status: 200, - headers: { 'content-type': 'application/json' }, - }), - ); + const fetchMock = mockMessageDeliveryFetch([ + () => new Response(JSON.stringify({ ok: true }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + () => new Response(JSON.stringify({ ack: true }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + ]); const ws = await createWorkspace(stack.app, 'http-node-response-no-ack'); const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob'); @@ -299,7 +324,7 @@ describe('node delivery contracts', () => { expect(post.status).toBe(201); await waitForAssertion(async () => { - expect(fetchMock).toHaveBeenCalledTimes(1); + expect(deliveryPosts(fetchMock)).toHaveLength(1); const queued = await stack.app.request('/v1/deliveries', { headers: { authorization: `Bearer ${bob.token}` }, }); @@ -328,7 +353,7 @@ describe('node delivery contracts', () => { expect(swept).toBe(1); await waitForAssertion(async () => { - expect(fetchMock).toHaveBeenCalledTimes(2); + expect(deliveryPosts(fetchMock)).toHaveLength(2); const acked = await stack.app.request('/v1/deliveries?status=acked', { headers: { authorization: `Bearer ${bob.token}` }, }); @@ -637,10 +662,10 @@ describe('node delivery contracts', () => { }); it('redrives failed http_push deliveries when their retry time is due', async () => { - const fetchMock = vi - .spyOn(globalThis, 'fetch') - .mockResolvedValueOnce(new Response('', { status: 503 })) - .mockResolvedValueOnce(new Response('', { status: 202 })); + const fetchMock = mockMessageDeliveryFetch([ + () => new Response('', { status: 503 }), + () => new Response('', { status: 202 }), + ]); const ws = await createWorkspace(stack.app, 'http-node-redrive'); const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob'); @@ -662,7 +687,7 @@ describe('node delivery contracts', () => { expect(post.status).toBe(201); await waitForAssertion(async () => { - expect(fetchMock).toHaveBeenCalledTimes(1); + expect(deliveryPosts(fetchMock)).toHaveLength(1); const queued = await stack.app.request('/v1/deliveries', { headers: { authorization: `Bearer ${bob.token}` }, }); @@ -698,7 +723,7 @@ describe('node delivery contracts', () => { expect(swept).toBe(1); await waitForAssertion(async () => { - expect(fetchMock).toHaveBeenCalledTimes(2); + expect(deliveryPosts(fetchMock)).toHaveLength(2); const queued = await stack.app.request('/v1/deliveries', { headers: { authorization: `Bearer ${bob.token}` }, }); @@ -714,9 +739,9 @@ describe('node delivery contracts', () => { }); it('claims a due http_push delivery only once across overlapping redrive sweeps', async () => { - const fetchMock = vi - .spyOn(globalThis, 'fetch') - .mockResolvedValueOnce(new Response('', { status: 503 })); + const fetchMock = mockMessageDeliveryFetch([ + () => new Response('', { status: 503 }), + ]); const ws = await createWorkspace(stack.app, 'http-node-redrive-claim'); const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob'); @@ -736,7 +761,7 @@ describe('node delivery contracts', () => { body: JSON.stringify({ text: 'retry once' }), }); expect(post.status).toBe(201); - await waitForAssertion(() => expect(fetchMock).toHaveBeenCalledTimes(1)); + await waitForAssertion(() => expect(deliveryPosts(fetchMock)).toHaveLength(1)); await stack.runtime.deps.db .update(deliveries) @@ -746,6 +771,8 @@ describe('node delivery contracts', () => { eq(deliveries.agentId, bob.agentId), )); + // The redrive sweep only POSTs the durable message; no agent activity fires + // ephemeral events here, so a plain hang-all mock isolates the claim check. let releaseFetch: ((response: Response) => void) | undefined; fetchMock.mockReset(); fetchMock.mockImplementation(() => new Promise((resolve) => { @@ -765,12 +792,18 @@ describe('node delivery contracts', () => { }); it('does not let a slow http_push receiver block self-connected recipients', async () => { + // Only the durable message POST hangs; ephemeral event POSTs (presence) + // resolve immediately so the single hung message delivery stays isolated. let releaseFetch: ((response: Response) => void) | undefined; const fetchMock = vi .spyOn(globalThis, 'fetch') - .mockImplementation(() => new Promise((resolve) => { - releaseFetch = resolve; - })); + .mockImplementation((async (_url: unknown, init?: RequestInit) => { + const headers = init?.headers as Record | undefined; + if (headers?.['X-Relaycast-Event'] === 'message.created') { + return new Promise((resolve) => { releaseFetch = resolve; }); + } + return new Response('', { status: 202 }); + }) as typeof globalThis.fetch); const ws = await createWorkspace(stack.app, 'http-node-slow-isolation'); const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob'); @@ -793,7 +826,7 @@ describe('node delivery contracts', () => { }); expect(post.status).toBe(201); - await waitForAssertion(() => expect(fetchMock).toHaveBeenCalledTimes(1)); + await waitForAssertion(() => expect(deliveryPosts(fetchMock)).toHaveLength(1)); await waitForAssertion(async () => { expect(deliverFramesOfType(carolSock, 'message.created')).toEqual([ expect.objectContaining({ diff --git a/packages/engine/src/adapters/node/index.ts b/packages/engine/src/adapters/node/index.ts index 7da330e0..e0128e6b 100644 --- a/packages/engine/src/adapters/node/index.ts +++ b/packages/engine/src/adapters/node/index.ts @@ -108,7 +108,7 @@ export function createNodeRuntime(options: NodeRuntimeOptions): NodeRuntime { const eventType = typeof event.type === 'string' ? event.type : null; if (!subjectAgentId || !eventType) return; await sendNodePresenceContext( - { db, nodeConnections: realtime, realtime, workspaceId }, + { db, nodeConnections: realtime, realtime, workspaceId, environment: options.config?.environment }, { subjectAgentId, event: eventType, diff --git a/packages/engine/src/engine/httpPushDispatch.ts b/packages/engine/src/engine/httpPushDispatch.ts new file mode 100644 index 00000000..93141720 --- /dev/null +++ b/packages/engine/src/engine/httpPushDispatch.ts @@ -0,0 +1,104 @@ +import { hmacSha256Hex } from '../lib/crypto.js'; +import { isSafeExternalUrl } from '../lib/ssrf.js'; + +function publicHeaders(headers: Record | undefined): Record { + if (!headers) return {}; + const result: Record = {}; + for (const [name, value] of Object.entries(headers)) { + if (typeof value === 'string') result[name] = value; + } + return result; +} + +/** + * Build the auth + signing headers for an http_push POST given the node's + * delivery config and the exact body + timestamp being sent. Shared by durable + * message dispatch (`deliveryRouting`) and ephemeral node-event dispatch so both + * honor the same bearer / static-header / HMAC contract. + */ +export async function buildHttpPushHeaders( + config: Record, + eventType: string, + deliveryId: string | null, + body: string, + timestamp: string, +): Promise> { + const auth = config.auth && typeof config.auth === 'object' && !Array.isArray(config.auth) + ? config.auth as Record + : { type: 'none' }; + + const headers: Record = { + 'Content-Type': 'application/json', + 'X-Relaycast-Event': eventType, + }; + if (deliveryId) headers['X-Relaycast-Delivery'] = deliveryId; + + if (auth.type === 'bearer' && typeof auth.token === 'string') { + headers.Authorization = `Bearer ${auth.token}`; + } else if (auth.type === 'static_headers') { + Object.assign(headers, publicHeaders(auth.headers as Record | undefined)); + } else if (auth.type === 'hmac_sha256' && typeof auth.secret === 'string') { + const timestampHeader = typeof auth.timestamp_header === 'string' ? auth.timestamp_header : 'X-Relaycast-Timestamp'; + const signatureHeader = typeof auth.signature_header === 'string' ? auth.signature_header : 'X-Relaycast-Signature'; + const prefix = typeof auth.prefix === 'string' ? auth.prefix : 'sha256='; + const signedPayload = auth.signed_payload === 'body' ? body : `${timestamp}.${body}`; + headers[timestampHeader] = timestamp; + headers[signatureHeader] = `${prefix}${await hmacSha256Hex(signedPayload, auth.secret)}`; + } + + return headers; +} + +/** `true` outside the test environment — keeps SSRF hardening on by default. */ +export function strictHttpPushDispatch(environment: string | undefined): boolean { + return environment !== 'test'; +} + +export interface EphemeralNodeEvent { + workspaceId: string; + eventType: string; + eventData: Record; + /** Top-level body fields beyond `type`/`workspace_id`/`timestamp`/`data`. */ + extra?: Record; +} + +/** + * Best-effort POST of an ephemeral node event (reaction, read receipt, presence + * or context update) to an http_push node's receiver. Unlike durable message + * dispatch there is no delivery row, ack, or retry — it mirrors the fire-and- + * forget semantics of pushing a frame over a WebSocket node socket. Returns + * whether the receiver acknowledged with a 2xx. + */ +export async function postEphemeralEventToHttpPushNode(args: { + deliveryConfig: Record | null | undefined; + strict: boolean; + event: EphemeralNodeEvent; +}): Promise { + const config = args.deliveryConfig ?? {}; + const url = typeof config.url === 'string' ? config.url : null; + if (!url) return false; + if (!isSafeExternalUrl(url, { strict: args.strict })) return false; + + const timestamp = new Date().toISOString(); + const body = JSON.stringify({ + type: args.event.eventType, + workspace_id: args.event.workspaceId, + ...(args.event.extra ?? {}), + timestamp, + data: args.event.eventData, + }); + const headers = await buildHttpPushHeaders(config, args.event.eventType, null, body, timestamp); + + try { + const response = await globalThis.fetch(url, { + method: 'POST', + headers, + body, + redirect: 'error', + signal: AbortSignal.timeout(10_000), + }); + return response.ok; + } catch { + return false; + } +} diff --git a/packages/engine/src/engine/invocationCompletion.ts b/packages/engine/src/engine/invocationCompletion.ts index a6f54aa6..fda09ade 100644 --- a/packages/engine/src/engine/invocationCompletion.ts +++ b/packages/engine/src/engine/invocationCompletion.ts @@ -3,7 +3,7 @@ import { enqueueEvent } from './eventQueue.js'; import type { EngineDeps } from '../ports/index.js'; import { sendNodeDeliveriesToAgents } from './nodeDeliver.js'; -export type InvocationCompletionDeps = Pick; +export type InvocationCompletionDeps = Pick; type CompletionResult = { invocation_id: string; @@ -46,6 +46,7 @@ export async function emitInvocationCompletionEffects( db: deps.db, nodeConnections: deps.nodeConnections, workspaceId, + environment: deps.config?.environment, }, { agentIds: [result.caller_id], event: eventType, diff --git a/packages/engine/src/engine/nodeContext.ts b/packages/engine/src/engine/nodeContext.ts index 1c08851d..9b919ed9 100644 --- a/packages/engine/src/engine/nodeContext.ts +++ b/packages/engine/src/engine/nodeContext.ts @@ -3,6 +3,7 @@ import type { EngineDb } from '../ports/database.js'; import type { NodeConnectionRegistry, RealtimeBus } from '../ports/realtime.js'; import { agents, agentNodeBindings, channelMembers, nodes } from '../db/schema.js'; import { toFleetWireJson } from './deliveryWire.js'; +import { postEphemeralEventToHttpPushNode, strictHttpPushDispatch } from './httpPushDispatch.js'; type NodeContextTopic = 'presence' | 'channel' | 'thread' | 'agent'; @@ -11,6 +12,8 @@ type NodeContextDeps = { nodeConnections: NodeConnectionRegistry; realtime: RealtimeBus; workspaceId: string; + /** Defaults to strict SSRF hardening when omitted (production-safe). */ + environment?: string; }; type ScopedNodeRow = { @@ -19,8 +22,13 @@ type ScopedNodeRow = { nodeKind: string; nodeRole: string; deliveryAdapter: string | null; + deliveryConfig: Record | null; }; +// Node kinds eligible for context updates: WebSocket nodes receive a pushed +// `context.update` frame; http_push nodes receive a best-effort POST. +const CONTEXT_NODE_KINDS = ['ws', 'fleet_ws', 'direct_ws', 'http_push'] as const; + function normalizeDeliveryAdapter(adapter: string | null | undefined, nodeKind: string | null | undefined): string | null { if (adapter === 'fleet.ws.v1' || adapter === 'direct.ws.v1') return 'ws.node.v1'; if (adapter) return adapter; @@ -28,13 +36,22 @@ function normalizeDeliveryAdapter(adapter: string | null | undefined, nodeKind: return null; } -function groupByNode(rows: ScopedNodeRow[]): Map { - const grouped = new Map(); +type GroupedNode = { + nodeKind: string; + nodeRole: string; + deliveryAdapter: string | null; + deliveryConfig: Record | null; + agentIds: string[]; +}; + +function groupByNode(rows: ScopedNodeRow[]): Map { + const grouped = new Map(); for (const row of rows) { const existing = grouped.get(row.nodeId) ?? { nodeKind: row.nodeKind, nodeRole: row.nodeRole, deliveryAdapter: row.deliveryAdapter, + deliveryConfig: row.deliveryConfig, agentIds: [], }; existing.agentIds.push(row.agentId); @@ -71,6 +88,25 @@ async function sendContextToRows( ); continue; } + if (group.nodeKind === 'http_push') { + tasks.push( + postEphemeralEventToHttpPushNode({ + deliveryConfig: group.deliveryConfig, + strict: strictHttpPushDispatch(deps.environment), + event: { + workspaceId: deps.workspaceId, + eventType: message.event, + eventData: message.data, + extra: { + topic: message.topic, + channel_id: message.channelId ?? null, + agent_ids: agentIds, + }, + }, + }), + ); + continue; + } console.warn('[node.context] unsupported node kind for context update', { workspace_id: deps.workspaceId, node_id: nodeId, @@ -101,6 +137,7 @@ export async function sendNodeContextForChannel( nodeKind: nodes.kind, nodeRole: nodes.role, deliveryAdapter: nodes.deliveryAdapter, + deliveryConfig: nodes.deliveryConfig, }) .from(channelMembers) .innerJoin(agentNodeBindings, and( @@ -120,7 +157,7 @@ export async function sendNodeContextForChannel( )) .where(and( eq(channelMembers.channelId, args.channelId), - inArray(nodes.kind, ['ws', 'fleet_ws', 'direct_ws']), + inArray(nodes.kind, CONTEXT_NODE_KINDS), )); await sendContextToRows(deps, rows, { @@ -146,6 +183,7 @@ export async function sendNodePresenceContext( nodeKind: nodes.kind, nodeRole: nodes.role, deliveryAdapter: nodes.deliveryAdapter, + deliveryConfig: nodes.deliveryConfig, }) .from(agentNodeBindings) .innerJoin(agents, and( @@ -161,7 +199,7 @@ export async function sendNodePresenceContext( .where(and( eq(agentNodeBindings.workspaceId, deps.workspaceId), eq(agentNodeBindings.status, 'active'), - inArray(nodes.kind, ['ws', 'fleet_ws', 'direct_ws']), + inArray(nodes.kind, CONTEXT_NODE_KINDS), )); await sendContextToRows(deps, rows, { @@ -189,6 +227,7 @@ export async function sendNodeContextToAgents( nodeKind: nodes.kind, nodeRole: nodes.role, deliveryAdapter: nodes.deliveryAdapter, + deliveryConfig: nodes.deliveryConfig, }) .from(agentNodeBindings) .innerJoin(agents, and( @@ -205,7 +244,7 @@ export async function sendNodeContextToAgents( eq(agentNodeBindings.workspaceId, deps.workspaceId), eq(agentNodeBindings.status, 'active'), inArray(agentNodeBindings.agentId, uniqueAgentIds), - inArray(nodes.kind, ['ws', 'fleet_ws', 'direct_ws']), + inArray(nodes.kind, CONTEXT_NODE_KINDS), )); await sendContextToRows(deps, rows, { diff --git a/packages/engine/src/engine/nodeDeliver.ts b/packages/engine/src/engine/nodeDeliver.ts index a434d447..3fed9fcd 100644 --- a/packages/engine/src/engine/nodeDeliver.ts +++ b/packages/engine/src/engine/nodeDeliver.ts @@ -3,11 +3,14 @@ import { agents, agentNodeBindings, channelMembers, dmConversations, dmParticipa import type { EngineDb } from '../ports/database.js'; import type { NodeConnectionRegistry } from '../ports/realtime.js'; import { buildDeliverFrame, buildDeliverPayload } from './deliveryWire.js'; +import { postEphemeralEventToHttpPushNode, strictHttpPushDispatch } from './httpPushDispatch.js'; type NodeDeliverDeps = { db: EngineDb; nodeConnections: NodeConnectionRegistry; workspaceId: string; + /** Defaults to strict SSRF hardening when omitted (production-safe). */ + environment?: string; }; type NodeDeliverRecipient = { @@ -15,9 +18,13 @@ type NodeDeliverRecipient = { agentName: string; nodeId: string; nodeKind: string; + deliveryConfig: Record | null; }; const WS_NODE_KINDS = ['ws', 'fleet_ws', 'direct_ws'] as const; +// Node kinds eligible for ephemeral event delivery: WebSocket nodes receive a +// pushed frame; http_push nodes receive a best-effort POST to their receiver. +const EVENTED_NODE_KINDS = [...WS_NODE_KINDS, 'http_push'] as const; function eventDeliveryId(event: string, eventKey: string, agentId: string): string { const normalizedEvent = event.replace(/[^a-zA-Z0-9]+/g, '_').replace(/^_+|_+$/g, '') || 'event'; @@ -38,6 +45,7 @@ async function channelRecipients(deps: NodeDeliverDeps, channelId: string): Prom agentName: agents.name, nodeId: agentNodeBindings.nodeId, nodeKind: nodes.kind, + deliveryConfig: nodes.deliveryConfig, }) .from(dmParticipants) .innerJoin(agents, and( @@ -58,7 +66,7 @@ async function channelRecipients(deps: NodeDeliverDeps, channelId: string): Prom isNull(dmParticipants.leftAt), eq(agents.locationType, 'via_node'), eq(agents.locationNodeId, agentNodeBindings.nodeId), - inArray(nodes.kind, WS_NODE_KINDS), + inArray(nodes.kind, EVENTED_NODE_KINDS), )); } @@ -68,6 +76,7 @@ async function channelRecipients(deps: NodeDeliverDeps, channelId: string): Prom agentName: agents.name, nodeId: agentNodeBindings.nodeId, nodeKind: nodes.kind, + deliveryConfig: nodes.deliveryConfig, }) .from(channelMembers) .innerJoin(agents, and( @@ -87,10 +96,44 @@ async function channelRecipients(deps: NodeDeliverDeps, channelId: string): Prom eq(channelMembers.channelId, channelId), eq(agents.locationType, 'via_node'), eq(agents.locationNodeId, agentNodeBindings.nodeId), - inArray(nodes.kind, WS_NODE_KINDS), + inArray(nodes.kind, EVENTED_NODE_KINDS), )); } +// Deliver one ephemeral event to a node recipient: WebSocket nodes get a pushed +// `deliver` frame, http_push nodes get a best-effort POST to their receiver. +function deliverEventToRecipient( + deps: NodeDeliverDeps, + recipient: NodeDeliverRecipient, + args: { event: string; eventKey: string; data: Record; messageId: string }, +): Promise { + if (recipient.nodeKind === 'http_push') { + return postEphemeralEventToHttpPushNode({ + deliveryConfig: recipient.deliveryConfig, + strict: strictHttpPushDispatch(deps.environment), + event: { + workspaceId: deps.workspaceId, + eventType: args.event, + eventData: args.data, + extra: { + message_id: args.messageId, + agent_id: recipient.agentId, + agent_name: recipient.agentName, + }, + }, + }); + } + return deps.nodeConnections.sendToNode(deps.workspaceId, recipient.nodeId, buildDeliverFrame({ + delivery_id: eventDeliveryId(args.event, args.eventKey, recipient.agentId), + agent_id: recipient.agentId, + agent: recipient.agentName, + msg_id: args.messageId, + seq: 0, + mode: 'wait', + payload: buildDeliverPayload(args.event, args.data), + })); +} + export async function sendNodeDeliveriesForChannel( deps: NodeDeliverDeps, args: { @@ -103,15 +146,12 @@ export async function sendNodeDeliveriesForChannel( ): Promise { const recipients = await channelRecipients(deps, args.channelId); const tasks = recipients.map((recipient) => - deps.nodeConnections.sendToNode(deps.workspaceId, recipient.nodeId, buildDeliverFrame({ - delivery_id: eventDeliveryId(args.event, args.eventKey, recipient.agentId), - agent_id: recipient.agentId, - agent: recipient.agentName, - msg_id: args.messageId, - seq: 0, - mode: 'wait', - payload: buildDeliverPayload(args.event, args.data), - })), + deliverEventToRecipient(deps, recipient, { + event: args.event, + eventKey: args.eventKey, + data: args.data, + messageId: args.messageId, + }), ); await Promise.allSettled(tasks); } @@ -135,6 +175,7 @@ export async function sendNodeDeliveriesToAgents( agentName: agents.name, nodeId: agentNodeBindings.nodeId, nodeKind: nodes.kind, + deliveryConfig: nodes.deliveryConfig, }) .from(agents) .innerJoin(agentNodeBindings, and( @@ -151,19 +192,16 @@ export async function sendNodeDeliveriesToAgents( inArray(agents.id, unique), eq(agents.locationType, 'via_node'), eq(agents.locationNodeId, agentNodeBindings.nodeId), - inArray(nodes.kind, WS_NODE_KINDS), + inArray(nodes.kind, EVENTED_NODE_KINDS), )); const tasks = recipients.map((recipient) => - deps.nodeConnections.sendToNode(deps.workspaceId, recipient.nodeId, buildDeliverFrame({ - delivery_id: eventDeliveryId(args.event, args.eventKey, recipient.agentId), - agent_id: recipient.agentId, - agent: recipient.agentName, - msg_id: args.messageId ?? args.eventKey, - seq: 0, - mode: 'wait', - payload: buildDeliverPayload(args.event, args.data), - })), + deliverEventToRecipient(deps, recipient, { + event: args.event, + eventKey: args.eventKey, + data: args.data, + messageId: args.messageId ?? args.eventKey, + }), ); await Promise.allSettled(tasks); } diff --git a/packages/engine/src/routes/action.ts b/packages/engine/src/routes/action.ts index 1ef00b19..7ce68de1 100644 --- a/packages/engine/src/routes/action.ts +++ b/packages/engine/src/routes/action.ts @@ -211,6 +211,7 @@ actionRoutes.post('/actions/:name/invoke', requireAuth, rateLimit, async (c) => { db: c.get('db'), nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, workspaceId: workspace.id, }, { diff --git a/packages/engine/src/routes/agent.ts b/packages/engine/src/routes/agent.ts index 5f1b38bc..7215f5db 100644 --- a/packages/engine/src/routes/agent.ts +++ b/packages/engine/src/routes/agent.ts @@ -117,6 +117,7 @@ async function fanoutAgentStatus(c: Parameters[0], agent { db: c.get('db'), nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, realtime: c.get('engine').realtime, workspaceId: c.get('workspace').id, }, @@ -517,6 +518,7 @@ agentRoutes.post( { db, nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, realtime: c.get('engine').realtime, workspaceId: workspace.id, }, diff --git a/packages/engine/src/routes/deliveryRouting.ts b/packages/engine/src/routes/deliveryRouting.ts index e1651519..5ff1dbf4 100644 --- a/packages/engine/src/routes/deliveryRouting.ts +++ b/packages/engine/src/routes/deliveryRouting.ts @@ -8,7 +8,7 @@ import type { DeliveryRejectionRecord, } from '../engine/deliveryWrites.js'; import { agents, agentNodeBindings, deliveries as deliveryRows, nodes } from '../db/schema.js'; -import { hmacSha256Hex } from '../lib/crypto.js'; +import { buildHttpPushHeaders } from '../engine/httpPushDispatch.js'; import { isSafeExternalUrl } from '../lib/ssrf.js'; import { transformForClient, type WsEvent } from '../engine/wsTransform.js'; import type { EngineDb, EngineDeps } from '../ports/index.js'; @@ -78,6 +78,7 @@ async function fanoutToAgentsForContext( nodeConnections: ctx.engine.nodeConnections, realtime: ctx.engine.realtime, workspaceId: ctx.workspaceId, + environment: ctx.engine.config?.environment, }, { agentIds: unique, @@ -206,15 +207,6 @@ function normalizeDeliveryAdapter(adapter: string | null | undefined, nodeKind: return null; } -function publicHeaders(headers: Record | undefined): Record { - if (!headers) return {}; - const result: Record = {}; - for (const [name, value] of Object.entries(headers)) { - if (typeof value === 'string') result[name] = value; - } - return result; -} - async function recordHttpPushRetry( ctx: RoutingContext, deliveryId: string, @@ -271,27 +263,7 @@ async function dispatchHttpPush(args: { data: args.eventData, }); - const auth = config.auth && typeof config.auth === 'object' && !Array.isArray(config.auth) - ? config.auth as Record - : { type: 'none' }; - - const headers: Record = { - 'Content-Type': 'application/json', - 'X-Relaycast-Event': args.eventType, - 'X-Relaycast-Delivery': args.delivery.id, - }; - if (auth.type === 'bearer' && typeof auth.token === 'string') { - headers.Authorization = `Bearer ${auth.token}`; - } else if (auth.type === 'static_headers') { - Object.assign(headers, publicHeaders(auth.headers as Record | undefined)); - } else if (auth.type === 'hmac_sha256' && typeof auth.secret === 'string') { - const timestampHeader = typeof auth.timestamp_header === 'string' ? auth.timestamp_header : 'X-Relaycast-Timestamp'; - const signatureHeader = typeof auth.signature_header === 'string' ? auth.signature_header : 'X-Relaycast-Signature'; - const prefix = typeof auth.prefix === 'string' ? auth.prefix : 'sha256='; - const signedPayload = auth.signed_payload === 'body' ? body : `${timestamp}.${body}`; - headers[timestampHeader] = timestamp; - headers[signatureHeader] = `${prefix}${await hmacSha256Hex(signedPayload, auth.secret)}`; - } + const headers = await buildHttpPushHeaders(config, args.eventType, args.delivery.id, body, timestamp); try { const claimConditions = [ diff --git a/packages/engine/src/routes/fanout.ts b/packages/engine/src/routes/fanout.ts index 6a27dec7..24808031 100644 --- a/packages/engine/src/routes/fanout.ts +++ b/packages/engine/src/routes/fanout.ts @@ -94,6 +94,7 @@ export async function fanoutToChannel( { db: c.get('db'), nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, realtime: c.get('engine').realtime, workspaceId: ws, }, @@ -136,6 +137,7 @@ export async function fanoutToAgents( { db: c.get('db'), nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, realtime: c.get('engine').realtime, workspaceId, }, diff --git a/packages/engine/src/routes/reaction.ts b/packages/engine/src/routes/reaction.ts index 94bb4d96..9fee5244 100644 --- a/packages/engine/src/routes/reaction.ts +++ b/packages/engine/src/routes/reaction.ts @@ -94,6 +94,7 @@ reactionRoutes.post( { db, nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, workspaceId: workspace.id, }, { @@ -184,6 +185,7 @@ reactionRoutes.delete( { db, nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, workspaceId: workspace.id, }, { diff --git a/packages/engine/src/routes/receipt.ts b/packages/engine/src/routes/receipt.ts index 55a32025..ae1317cf 100644 --- a/packages/engine/src/routes/receipt.ts +++ b/packages/engine/src/routes/receipt.ts @@ -54,6 +54,7 @@ receiptRoutes.post( { db, nodeConnections: c.get('engine').nodeConnections, + environment: c.get('engine').config?.environment, workspaceId: workspace.id, }, { From 815fcbb93fb5bbf2d06ec13699770f6172f0a597 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Tue, 30 Jun 2026 08:56:33 -0700 Subject: [PATCH 2/2] fix(engine): harden http_push ephemeral dispatch per review - Detach the adapter presence-context POST so a slow/black-holed http_push receiver can't stall presence broadcasts or offline sweeps (Codex). - Cancel the fetch response body on ephemeral POSTs to release the connection instead of leaking it under load (Gemini). - Build headers inside the try/retry boundary in both the ephemeral and durable dispatch paths so a signing failure is best-effort / recorded as a retry rather than rejecting uncaught (CodeRabbit). - Stop operator static_headers from overriding Relaycast protocol headers (Content-Type, X-Relaycast-Event, X-Relaycast-Delivery) (CodeRabbit). - Spread `extra` before canonical fields so it can't override type/workspace_id; guard publicHeaders against non-object input; document the full payload shape. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 7 ++-- .../conformance/nodeDeliveryContracts.test.ts | 8 +++-- packages/engine/src/adapters/node/index.ts | 7 ++-- .../engine/src/engine/httpPushDispatch.ts | 36 +++++++++++++------ packages/engine/src/routes/deliveryRouting.ts | 5 +-- 5 files changed, 44 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index a6970777..b55e2853 100644 --- a/README.md +++ b/README.md @@ -442,8 +442,11 @@ securely hold that token. HTTP push nodes also receive the ephemeral channel/workspace events a WebSocket node gets — reactions (`message.reacted`), read receipts (`message.read`), and presence / status updates — as best-effort POSTs to the same delivery URL (same auth/signing, -no delivery row or ack). Each carries a `type` plus the event `data`; receivers that -only want durable messages can filter on the `X-Relaycast-Event` header. +no delivery row or ack). Each body is snake_case with `type`, `workspace_id`, +`timestamp`, and the event `data`, plus event-specific identifiers: `message_id` / +`agent_id` / `agent_name` for reactions and receipts, or `topic` / `channel_id` / +`agent_ids` for presence and context updates. Receivers that only want durable +messages can filter on the `X-Relaycast-Event` header. Queue/cron-backed deployments must call `sweepDueHttpPushDeliveries` from a scheduled handler to retry queued HTTP push deliveries whose `next_attempt_at` is due; the Node self-host adapter runs that sweep on its local maintenance timer. diff --git a/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts b/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts index c45b7eb1..fef24c97 100644 --- a/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts +++ b/packages/engine/src/__tests__/conformance/nodeDeliveryContracts.test.ts @@ -247,12 +247,14 @@ describe('node delivery contracts', () => { }); it('acks an http_push delivery when the node contract uses response body ack', async () => { - vi.spyOn(globalThis, 'fetch').mockResolvedValue( - new Response(JSON.stringify({ ack: true }), { + // A fresh Response per call: the message dispatch reads the JSON body while + // ephemeral presence POSTs cancel theirs, so they must not share one object. + mockMessageDeliveryFetch([ + () => new Response(JSON.stringify({ ack: true }), { status: 200, headers: { 'content-type': 'application/json' }, }), - ); + ]); const ws = await createWorkspace(stack.app, 'http-node-response-ack'); const alice = await registerAgent(stack.app, ws.workspaceKey, 'alice'); const bob = await registerAgent(stack.app, ws.workspaceKey, 'bob'); diff --git a/packages/engine/src/adapters/node/index.ts b/packages/engine/src/adapters/node/index.ts index e0128e6b..840a2f47 100644 --- a/packages/engine/src/adapters/node/index.ts +++ b/packages/engine/src/adapters/node/index.ts @@ -107,7 +107,10 @@ export function createNodeRuntime(options: NodeRuntimeOptions): NodeRuntime { const subjectAgentId = typeof event.subject_agent_id === 'string' ? event.subject_agent_id : null; const eventType = typeof event.type === 'string' ? event.type : null; if (!subjectAgentId || !eventType) return; - await sendNodePresenceContext( + // Fire-and-forget: presence broadcasts/offline sweeps await this handler, + // so a slow or black-holed http_push receiver must not stall them. WS + // context sends are in-memory; the http_push POST runs detached. + void sendNodePresenceContext( { db, nodeConnections: realtime, realtime, workspaceId, environment: options.config?.environment }, { subjectAgentId, @@ -118,7 +121,7 @@ export function createNodeRuntime(options: NodeRuntimeOptions): NodeRuntime { status: event.status, }, }, - ); + ).catch(() => {}); }, }); const rateLimiter = new InProcessRateLimiter(); diff --git a/packages/engine/src/engine/httpPushDispatch.ts b/packages/engine/src/engine/httpPushDispatch.ts index 93141720..1b982892 100644 --- a/packages/engine/src/engine/httpPushDispatch.ts +++ b/packages/engine/src/engine/httpPushDispatch.ts @@ -1,11 +1,22 @@ import { hmacSha256Hex } from '../lib/crypto.js'; import { isSafeExternalUrl } from '../lib/ssrf.js'; -function publicHeaders(headers: Record | undefined): Record { - if (!headers) return {}; +// Relaycast-controlled headers must always win over operator-supplied +// static_headers so a node config can't mask the event type or delivery id. +const RELAYCAST_CONTROLLED_HEADERS = new Set([ + 'content-type', + 'x-relaycast-event', + 'x-relaycast-delivery', +]); + +function publicHeaders( + headers: Record | undefined, + reserved: Set = new Set(), +): Record { + if (!headers || typeof headers !== 'object' || Array.isArray(headers)) return {}; const result: Record = {}; for (const [name, value] of Object.entries(headers)) { - if (typeof value === 'string') result[name] = value; + if (typeof value === 'string' && !reserved.has(name.toLowerCase())) result[name] = value; } return result; } @@ -14,7 +25,8 @@ function publicHeaders(headers: Record | undefined): Record, @@ -36,7 +48,8 @@ export async function buildHttpPushHeaders( if (auth.type === 'bearer' && typeof auth.token === 'string') { headers.Authorization = `Bearer ${auth.token}`; } else if (auth.type === 'static_headers') { - Object.assign(headers, publicHeaders(auth.headers as Record | undefined)); + // Merge operator headers without clobbering the Relaycast protocol headers. + Object.assign(headers, publicHeaders(auth.headers as Record | undefined, RELAYCAST_CONTROLLED_HEADERS)); } else if (auth.type === 'hmac_sha256' && typeof auth.secret === 'string') { const timestampHeader = typeof auth.timestamp_header === 'string' ? auth.timestamp_header : 'X-Relaycast-Timestamp'; const signatureHeader = typeof auth.signature_header === 'string' ? auth.signature_header : 'X-Relaycast-Signature'; @@ -58,7 +71,7 @@ export interface EphemeralNodeEvent { workspaceId: string; eventType: string; eventData: Record; - /** Top-level body fields beyond `type`/`workspace_id`/`timestamp`/`data`. */ + /** Supplemental top-level body fields; cannot override the canonical fields. */ extra?: Record; } @@ -66,8 +79,8 @@ export interface EphemeralNodeEvent { * Best-effort POST of an ephemeral node event (reaction, read receipt, presence * or context update) to an http_push node's receiver. Unlike durable message * dispatch there is no delivery row, ack, or retry — it mirrors the fire-and- - * forget semantics of pushing a frame over a WebSocket node socket. Returns - * whether the receiver acknowledged with a 2xx. + * forget semantics of pushing a frame over a WebSocket node socket. Never + * throws; returns whether the receiver acknowledged with a 2xx. */ export async function postEphemeralEventToHttpPushNode(args: { deliveryConfig: Record | null | undefined; @@ -80,16 +93,17 @@ export async function postEphemeralEventToHttpPushNode(args: { if (!isSafeExternalUrl(url, { strict: args.strict })) return false; const timestamp = new Date().toISOString(); + // Spread `extra` first so the canonical event fields always win. const body = JSON.stringify({ + ...(args.event.extra ?? {}), type: args.event.eventType, workspace_id: args.event.workspaceId, - ...(args.event.extra ?? {}), timestamp, data: args.event.eventData, }); - const headers = await buildHttpPushHeaders(config, args.event.eventType, null, body, timestamp); try { + const headers = await buildHttpPushHeaders(config, args.event.eventType, null, body, timestamp); const response = await globalThis.fetch(url, { method: 'POST', headers, @@ -97,6 +111,8 @@ export async function postEphemeralEventToHttpPushNode(args: { redirect: 'error', signal: AbortSignal.timeout(10_000), }); + // We only inspect status; release the connection instead of leaking the body. + await response.body?.cancel().catch(() => {}); return response.ok; } catch { return false; diff --git a/packages/engine/src/routes/deliveryRouting.ts b/packages/engine/src/routes/deliveryRouting.ts index 5ff1dbf4..b0737339 100644 --- a/packages/engine/src/routes/deliveryRouting.ts +++ b/packages/engine/src/routes/deliveryRouting.ts @@ -263,8 +263,6 @@ async function dispatchHttpPush(args: { data: args.eventData, }); - const headers = await buildHttpPushHeaders(config, args.eventType, args.delivery.id, body, timestamp); - try { const claimConditions = [ eq(deliveryRows.workspaceId, args.ctx.workspaceId), @@ -287,6 +285,9 @@ async function dispatchHttpPush(args: { .returning({ id: deliveryRows.id }); if (started.length === 0) return 'failed'; + // Build headers inside the claim/retry boundary so a signing failure is + // recorded as a retryable dispatch error rather than rejecting uncaught. + const headers = await buildHttpPushHeaders(config, args.eventType, args.delivery.id, body, timestamp); const response = await globalThis.fetch(url, { method: 'POST', headers,