fix(engine): deliver ephemeral node events (reactions, receipts, presence) to http_push nodes

README.md

@@ -439,6 +439,14 @@ 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 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.

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<void>, 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<typeof vi.spyOn>, type: string): Array<Record<string, unknown>> {
+    return fetchMock.mock.calls
+      .map((call) => {
+        try { return JSON.parse((call[1] as RequestInit).body as string) as Record<string, unknown>; }
+        catch { return null; }
+      })
+      .filter((body): body is Record<string, unknown> => !!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' });
+    });
+  });
+});

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<typeof vi.spyOn>,
+    eventType = 'message.created',
+  ): Array<[string, RequestInit]> {
+    return fetchMock.mock.calls.filter((call) => {
+      const headers = (call[1] as RequestInit | undefined)?.headers as Record<string, string> | 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<string, string> | 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<string, string>;
@@ -218,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');
@@ -263,20 +294,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 +326,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 +355,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 +664,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 +689,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 +725,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 +741,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 +763,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 +773,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<Response>((resolve) => {
@@ -765,12 +794,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<Response>((resolve) => {
-        releaseFetch = resolve;
-      }));
+      .mockImplementation((async (_url: unknown, init?: RequestInit) => {
+        const headers = init?.headers as Record<string, string> | undefined;
+        if (headers?.['X-Relaycast-Event'] === 'message.created') {
+          return new Promise<Response>((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 +828,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({