From 8aa15700f9a573c43402cc3aec63ee0d0a3fa39c Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 4 Aug 2026 13:56:52 -0700 Subject: [PATCH] Send the run id on correlation-id event reads #3280 made runId required on ListEventsByCorrelationIdParams, but world-vercel could only apply the scope after the fact: it selected by correlation id on the wire and filtered the returned page by run. That depends on the backend having happened to return the run's rows in the page it answered with. Put runId on the request. The backend reads the run's own partition and answers for that run, so the page comes back scoped. The client-side filter stays for backends that predate the parameter and still answer across runs. Co-Authored-By: Claude Opus 5 --- .changeset/correlation-id-run-scope-wire.md | 5 ++ packages/world-vercel/src/events-v4.test.ts | 66 +++++++++++++++++++++ packages/world-vercel/src/events-v4.ts | 14 ++++- packages/world-vercel/src/events.test.ts | 63 ++++++++++++++++++++ packages/world-vercel/src/events.ts | 15 +++-- 5 files changed, 156 insertions(+), 7 deletions(-) create mode 100644 .changeset/correlation-id-run-scope-wire.md diff --git a/.changeset/correlation-id-run-scope-wire.md b/.changeset/correlation-id-run-scope-wire.md new file mode 100644 index 0000000000..1d07522443 --- /dev/null +++ b/.changeset/correlation-id-run-scope-wire.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-vercel': patch +--- + +Send the run id with correlation-id event lookups so the backend can scope them to a single run. diff --git a/packages/world-vercel/src/events-v4.test.ts b/packages/world-vercel/src/events-v4.test.ts index 181c678fec..6fd4bc81d6 100644 --- a/packages/world-vercel/src/events-v4.test.ts +++ b/packages/world-vercel/src/events-v4.test.ts @@ -12,6 +12,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { splitEventDataForV4 } from './events.js'; import { createWorkflowRunEventV4, + getEventsByCorrelationIdV4, getEventV4, getWorkflowRunEventsV4, throwForErrorResponse, @@ -265,6 +266,71 @@ describe('getWorkflowRunEventsV4 over HTTP', () => { }); }); +/** + * A correlation id names a step, hook or wait within *its* run, so the same + * one appears in every slot-numbered run (`step_…001` is each run's first + * step). The run id has to reach the backend for it to answer for one run. + */ +describe('getEventsByCorrelationIdV4 over HTTP', () => { + it('sends the run id alongside the correlation id', async () => { + const origin = + WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; + const agent = new MockAgent(); + agent.disableNetConnect(); + + const frames = Buffer.concat([ + encodeFrame( + { + eventId: 'evnt_1', + runId: 'wrun_1', + eventType: 'step_created', + correlationId: 'step_001', + createdAt: '2026-06-10T00:00:00.000Z', + eventData: {}, + }, + new Uint8Array(0) + ), + encodeFrame({ _end: 1, hasMore: false }, new Uint8Array(0)), + ]); + + // undici consults the matcher more than once per request (raw path and a + // query-sorted normalization of it), so assert on the parsed query of + // whatever it offered rather than on call counts or string equality. + const requestedPaths: string[] = []; + agent + .get(origin) + .intercept({ + path: (path) => { + requestedPaths.push(path); + return path.startsWith('/api/v4/events?'); + }, + method: 'GET', + }) + .reply(200, frames, { + headers: { 'content-type': V4_FRAME_CONTENT_TYPE }, + }); + + const result = await getEventsByCorrelationIdV4( + 'step_001', + 'wrun_1', + { limit: 10 }, + { token: 'test-token', dispatcher: agent } + ); + + expect(requestedPaths.length).toBeGreaterThan(0); + for (const path of requestedPaths) { + const query = new URL(path, origin).searchParams; + expect(query.get('correlationId')).toBe('step_001'); + expect(query.get('runId')).toBe('wrun_1'); + expect(query.get('limit')).toBe('10'); + } + + expect(result.events).toHaveLength(1); + expect(result.events[0].event.runId).toBe('wrun_1'); + agent.assertNoPendingInterceptors(); + }); +}); + /** * getEventV4 returns after the first frame. The early return must cancel the * response body (releasing its undici socket) without corrupting the returned diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index f26187f856..c3a6b47bc2 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -797,21 +797,29 @@ export async function getWorkflowRunEventsV4( } /** - * GET /api/v4/events?correlationId=... + * GET /api/v4/events?correlationId=...&runId=... * - * Same frame stream as getWorkflowRunEventsV4 but selected by - * correlationId (GSI) instead of runId. Used by the storage adapter's + * Same frame stream as getWorkflowRunEventsV4 but selected by correlation id + * instead of run id alone. Used by the storage adapter's * `events.listByCorrelationId` path — the v3 client used * `/v2/events?correlationId=...` for the equivalent query. + * + * `runId` scopes the lookup. A correlation id names a step, hook or wait + * within *its* run, so the same one can appear in many runs; sending the run + * is what lets the backend answer for one. A backend that predates the + * parameter ignores it and answers across runs, so the caller still filters + * the page by run id. */ export async function getEventsByCorrelationIdV4( correlationId: string, + runId: string, params: ListEventsV4Params = {}, config?: APIConfig ): Promise { const { baseUrl, headers } = await getHttpConfig(config); const sp = new URLSearchParams(); sp.set('correlationId', correlationId); + sp.set('runId', runId); appendListParams(sp, params); const url = `${baseUrl}/v4/events?${sp.toString()}`; return consumeListFrameStream( diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index 908588f944..fff1b7874e 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -1223,3 +1223,66 @@ describe('getWorkflowRunEvents hasMore mapping', () => { expect(result.cursor).toBe('cursor-2'); }); }); + +/** + * A correlation id names a step, hook or wait within *its* run: every + * slot-numbered run numbers its own steps, so `step_…001` belongs to all of + * them. The run id goes out on the request so the backend can answer for one + * run, and the page is filtered again on arrival because a backend predating + * that parameter answers across runs. + */ +describe('getWorkflowRunEvents by correlation id is scoped to the run', () => { + it('sends runId and drops any foreign-run event a legacy backend returns', async () => { + const agent = mockAgent(); + const event = (runId: string, eventId: string) => + encodeFrame( + { + eventId, + runId, + eventType: 'step_created', + correlationId: 'step_001', + createdAt: '2026-06-10T00:00:00.000Z', + eventData: {}, + }, + new Uint8Array(0) + ); + // What the pre-scope backend returns for this correlation id: the step of + // the run we asked about plus the identically numbered step of another. + const frames = Buffer.concat([ + event('wrun_1', 'evnt_1'), + event('wrun_2', 'evnt_2'), + encodeFrame( + { _end: 1, next: 'eid:evnt_2', hasMore: true }, + new Uint8Array(0) + ), + ]); + + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/events', + method: 'GET', + query: { + correlationId: 'step_001', + runId: 'wrun_1', + remoteRefBehavior: 'resolve', + }, + }) + .reply(200, frames, { + headers: { 'content-type': V4_FRAME_CONTENT_TYPE }, + }); + + const result = await getWorkflowRunEvents( + { correlationId: 'step_001', runId: 'wrun_1' }, + { token: 'test-token', dispatcher: agent } + ); + + // The interceptor only fires if runId reached the query string. + agent.assertNoPendingInterceptors(); + expect(result.data.map((e) => e.eventId)).toEqual(['evnt_1']); + // Pagination stays the backend's: a page that filters down to nothing is + // still followed by the next one. + expect(result.hasMore).toBe(true); + expect(result.cursor).toBe('eid:evnt_2'); + }); +}); diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 75750040d3..679d278410 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -551,7 +551,12 @@ export async function getWorkflowRunEvents( }; const result = await ('correlationId' in params - ? getEventsByCorrelationIdV4(params.correlationId, wirePagination, config) + ? getEventsByCorrelationIdV4( + params.correlationId, + params.runId, + wirePagination, + config + ) : getWorkflowRunEventsV4(params.runId, wirePagination, config)); const events = result.events.map((listed) => @@ -560,9 +565,11 @@ export async function getWorkflowRunEvents( // A correlation id is unique per run, not globally — a slot-numbered run // numbers its own steps, so `step_…001` names the first step of every such - // run. The backend selects by correlation id alone, so the run scope is - // applied here. `hasMore`/`cursor` stay the backend's, so a page that - // filters down to nothing is still followed by the next one. + // run. The run id goes out on the request above, and a backend that + // understands it answers for that run alone. One that predates the parameter + // selects by correlation id and spans runs, so the scope is re-applied here. + // `hasMore`/`cursor` stay the backend's, so a page that filters down to + // nothing is still followed by the next one. const runScoped = 'correlationId' in params ? events.filter((event) => event.runId === params.runId)