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)