diff --git a/.changeset/resume-partial-replay-streams.md b/.changeset/resume-partial-replay-streams.md
new file mode 100644
index 0000000000..9f5d35e640
--- /dev/null
+++ b/.changeset/resume-partial-replay-streams.md
@@ -0,0 +1,6 @@
+---
+"@workflow/world": patch
+"@workflow/world-vercel": patch
+---
+
+Resume interrupted or partial replay event streams after their last validated event and expose decoded events to streaming consumers.
diff --git a/packages/world-vercel/src/events-v4.test.ts b/packages/world-vercel/src/events-v4.test.ts
index 911e1a57e3..a9e966efe1 100644
--- a/packages/world-vercel/src/events-v4.test.ts
+++ b/packages/world-vercel/src/events-v4.test.ts
@@ -29,6 +29,13 @@ import {
 import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js';
 
 const CREATED_AT = '2026-06-10T00:00:00.000Z';
+const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
+
+function mockAgent() {
+  const agent = new MockAgent();
+  agent.disableNetConnect();
+  return agent;
+}
 
 function createEventBody(
   event: AnyEventRequest,
@@ -264,10 +271,7 @@ describe('throwForErrorResponse', () => {
  */
 describe('getWorkflowRunEventsV4 over HTTP', () => {
   it('parses a frame stream fetched via a custom dispatcher', async () => {
-    const origin =
-      WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
-    const agent = new MockAgent();
-    agent.disableNetConnect();
+    const agent = mockAgent();
 
     const body = new TextEncoder().encode('payload-bytes');
     const frames = Buffer.concat([
@@ -292,9 +296,9 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
     ]);
 
     agent
-      .get(origin)
+      .get(ORIGIN)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true',
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(200, frames, {
@@ -302,13 +306,12 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
       });
 
     const result = await getWorkflowRunEventsV4(
-      'wrun_1',
-      {},
+      { runId: 'wrun_1' },
       { token: 'test-token', dispatcher: agent }
     );
 
-    expect(result.events).toHaveLength(1);
-    expect(result.events[0]).toMatchObject({
+    expect(result.data).toHaveLength(1);
+    expect(result.data[0]).toMatchObject({
       eventId: 'evnt_1',
       eventData: { input: body },
     });
@@ -316,19 +319,62 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
     agent.assertNoPendingInterceptors();
   });
 
+  it('does not resume past a GET observer failure', async () => {
+    const agent = mockAgent();
+
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
+        method: 'GET',
+      })
+      .reply(
+        200,
+        Buffer.concat([
+          encodeFrame(
+            {
+              eventId: 'evnt_1',
+              runId: 'wrun_1',
+              eventType: 'run_started',
+              createdAt: CREATED_AT,
+            },
+            new Uint8Array()
+          ),
+          encodeFrame(
+            { _end: 1, next: 'eid:evnt_1', hasMore: false },
+            new Uint8Array()
+          ),
+        ]),
+        { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } }
+      );
+
+    const observerError = new WorkflowWorldError('observer failed', {
+      code: 'TRANSPORT',
+    });
+    await expect(
+      getWorkflowRunEventsV4(
+        {
+          runId: 'wrun_1',
+          onEvent: () => {
+            throw observerError;
+          },
+        },
+        { token: 'test-token', dispatcher: agent }
+      )
+    ).rejects.toBe(observerError);
+    agent.assertNoPendingInterceptors();
+  });
+
   it.each([
     ['an unknown event type', { eventType: 'future_event', eventData: {} }],
     ['invalid event metadata', { eventType: 'run_created', eventData: {} }],
   ])('rejects %s', async (_description, meta) => {
-    const origin =
-      WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
-    const agent = new MockAgent();
-    agent.disableNetConnect();
+    const agent = mockAgent();
 
     agent
-      .get(origin)
+      .get(ORIGIN)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true',
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(
@@ -342,8 +388,7 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
 
     await expect(
       getWorkflowRunEventsV4(
-        'wrun_1',
-        {},
+        { runId: 'wrun_1' },
         { token: 'test-token', dispatcher: agent }
       )
     ).rejects.toThrow();
@@ -351,10 +396,7 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
   });
 
   it('captures an explicit hasMore from the sentinel, independent of next', async () => {
-    const origin =
-      WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
-    const agent = new MockAgent();
-    agent.disableNetConnect();
+    const agent = mockAgent();
 
     // The regression shape: a final page still carries a trailing `next`
     // cursor (incremental-load resume point) but hasMore is false.
@@ -380,9 +422,9 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
     ]);
 
     agent
-      .get(origin)
+      .get(ORIGIN)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true',
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(200, frames, {
@@ -390,8 +432,7 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
       });
 
     const result = await getWorkflowRunEventsV4(
-      'wrun_1',
-      {},
+      { runId: 'wrun_1' },
       { token: 'test-token', dispatcher: agent }
     );
 
@@ -400,10 +441,7 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
   });
 
   it('rejects an end frame without hasMore', async () => {
-    const origin =
-      WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
-    const agent = new MockAgent();
-    agent.disableNetConnect();
+    const agent = mockAgent();
 
     const frames = encodeFrame(
       { _end: 1, next: 'cursor-2' },
@@ -411,9 +449,9 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
     );
 
     agent
-      .get(origin)
+      .get(ORIGIN)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true',
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(200, frames, {
@@ -422,18 +460,14 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
 
     await expect(
       getWorkflowRunEventsV4(
-        'wrun_1',
-        {},
+        { runId: 'wrun_1' },
         { token: 'test-token', dispatcher: agent }
       )
     ).rejects.toThrow();
   });
 
   it('throws when the stream ends without the end sentinel (truncated response)', async () => {
-    const origin =
-      WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
-    const agent = new MockAgent();
-    agent.disableNetConnect();
+    const agent = mockAgent();
 
     // A complete event frame but NO `{_end: 1}` sentinel — what a response
     // truncated on a frame boundary looks like. Returning this as a
@@ -454,9 +488,9 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
     );
 
     agent
-      .get(origin)
+      .get(ORIGIN)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?limit=500',
+        path: '/api/v4/runs/wrun_1/events?limit=500&remoteRefBehavior=resolve',
         method: 'GET',
       })
       .reply(200, frames, {
@@ -465,23 +499,19 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
 
     await expect(
       getWorkflowRunEventsV4(
-        'wrun_1',
-        { limit: 500 },
+        { runId: 'wrun_1', pagination: { limit: 500 } },
         { token: 'test-token', dispatcher: agent }
       )
     ).rejects.toThrow(/end-of-stream sentinel/);
   });
 
   it('resumes a truncated full stream after its last accepted event', async () => {
-    const origin =
-      WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
-    const agent = new MockAgent();
-    agent.disableNetConnect();
+    const agent = mockAgent();
 
     agent
-      .get(origin)
+      .get(ORIGIN)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true',
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(
@@ -503,9 +533,9 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
         { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } }
       );
     agent
-      .get(origin)
+      .get(ORIGIN)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true&cursor=eid%3Aevnt_1',
+        path: '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_1&remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(
@@ -529,12 +559,11 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
       );
 
     const result = await getWorkflowRunEventsV4(
-      'wrun_1',
-      {},
+      { runId: 'wrun_1' },
       { token: 'test-token', dispatcher: agent }
     );
 
-    expect(result.events.map((event) => event.eventId)).toEqual([
+    expect(result.data.map((event) => event.eventId)).toEqual([
       'evnt_1',
       'evnt_2',
     ]);
@@ -542,6 +571,74 @@ describe('getWorkflowRunEventsV4 over HTTP', () => {
     expect(result.hasMore).toBe(false);
     agent.assertNoPendingInterceptors();
   });
+
+  it('stops after three partial-stream recovery retries', async () => {
+    const agent = mockAgent();
+    const paths = [
+      '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
+      '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_1&remoteRefBehavior=resolve&returnAll=true',
+      '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_2&remoteRefBehavior=resolve&returnAll=true',
+      '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_3&remoteRefBehavior=resolve&returnAll=true',
+    ];
+
+    for (const [index, path] of paths.entries()) {
+      agent
+        .get(ORIGIN)
+        .intercept({ path, method: 'GET' })
+        .reply(
+          200,
+          encodeFrame(
+            {
+              eventId: `evnt_${index + 1}`,
+              runId: 'wrun_1',
+              eventType: 'run_created',
+              createdAt: CREATED_AT,
+              eventData: {
+                deploymentId: 'dpl_1',
+                workflowName: 'workflow',
+                input: null,
+              },
+            },
+            new Uint8Array()
+          ),
+          { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } }
+        );
+    }
+
+    await expect(
+      getWorkflowRunEventsV4(
+        { runId: 'wrun_1' },
+        { token: 'test-token', dispatcher: agent }
+      )
+    ).rejects.toThrow(/end-of-stream sentinel/);
+    agent.assertNoPendingInterceptors();
+  });
+
+  it('rejects an unexpected clean pagination response', async () => {
+    const agent = mockAgent();
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
+        method: 'GET',
+      })
+      .reply(
+        200,
+        encodeFrame(
+          { _end: 1, next: 'eid:evnt_1', hasMore: true },
+          new Uint8Array()
+        ),
+        { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } }
+      );
+
+    await expect(
+      getWorkflowRunEventsV4(
+        { runId: 'wrun_1' },
+        { token: 'test-token', dispatcher: agent }
+      )
+    ).rejects.toThrow(/returnAll response was unexpectedly paginated/);
+    agent.assertNoPendingInterceptors();
+  });
 });
 
 /**
@@ -589,9 +686,11 @@ describe('getEventsByCorrelationIdV4 over HTTP', () => {
       });
 
     const result = await getEventsByCorrelationIdV4(
-      'step_001',
-      'wrun_1',
-      { limit: 10 },
+      {
+        correlationId: 'step_001',
+        runId: 'wrun_1',
+        pagination: { limit: 10 },
+      },
       { token: 'test-token', dispatcher: agent }
     );
 
@@ -603,8 +702,8 @@ describe('getEventsByCorrelationIdV4 over HTTP', () => {
       expect(query.get('limit')).toBe('10');
     }
 
-    expect(result.events).toHaveLength(1);
-    expect(result.events[0].runId).toBe('wrun_1');
+    expect(result.data).toHaveLength(1);
+    expect(result.data[0].runId).toBe('wrun_1');
     agent.assertNoPendingInterceptors();
   });
 });
@@ -686,7 +785,7 @@ describe('v4 transport uses global fetch (observability)', () => {
     agent
       .get(origin)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true',
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(200, encodeFrame({ _end: 1, hasMore: false }, new Uint8Array(0)), {
@@ -698,8 +797,7 @@ describe('v4 transport uses global fetch (observability)', () => {
     const fetchSpy = vi.spyOn(globalThis, 'fetch');
 
     await getWorkflowRunEventsV4(
-      'wrun_1',
-      {},
+      { runId: 'wrun_1' },
       { token: 'test-token', dispatcher: agent }
     );
 
@@ -891,6 +989,228 @@ describe('createWorkflowRunEventV4 over HTTP', () => {
     agent.assertNoPendingInterceptors();
   });
 
+  it('continues a truncated run_started stream after its last event', async () => {
+    const agent = mockAgent();
+    const observed: string[] = [];
+
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events/run_started',
+        method: 'POST',
+        headers: { accept: V4_FRAME_CONTENT_TYPE },
+      })
+      .reply(
+        200,
+        encodeFrame(
+          {
+            eventId: 'evnt_1',
+            runId: 'wrun_1',
+            eventType: 'run_created',
+            createdAt: CREATED_AT,
+            eventData: {
+              deploymentId: 'dpl_1',
+              workflowName: 'workflow',
+              input: null,
+            },
+          },
+          new Uint8Array()
+        ),
+        {
+          headers: {
+            'content-type': V4_FRAME_CONTENT_TYPE,
+            'x-wf-max-events': '10000',
+          },
+        }
+      );
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_1&remoteRefBehavior=resolve&returnAll=true',
+        method: 'GET',
+      })
+      .reply(
+        200,
+        Buffer.concat([
+          encodeFrame(
+            {
+              eventId: 'evnt_2',
+              runId: 'wrun_1',
+              eventType: 'run_started',
+              createdAt: CREATED_AT,
+            },
+            new Uint8Array()
+          ),
+          encodeFrame(
+            { _end: 1, next: 'eid:evnt_2', hasMore: false },
+            new Uint8Array()
+          ),
+        ]),
+        { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } }
+      );
+
+    const result = await createWorkflowRunStartedEventV4(
+      { runId: 'wrun_1', specVersion: 5 },
+      { token: 'test-token', dispatcher: agent },
+      (event) => observed.push(event.eventId)
+    );
+
+    expect(result.events.map((event) => event.eventId)).toEqual([
+      'evnt_1',
+      'evnt_2',
+    ]);
+    expect(result.hasMore).toBe(false);
+    expect(observed).toEqual(['evnt_1', 'evnt_2']);
+    agent.assertNoPendingInterceptors();
+  });
+
+  it('preserves the POST cursor when truncation recovery returns an empty suffix', async () => {
+    const agent = mockAgent();
+
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events/run_started',
+        method: 'POST',
+        headers: { accept: V4_FRAME_CONTENT_TYPE },
+      })
+      .reply(
+        200,
+        Buffer.concat([
+          encodeFrame(
+            {
+              eventId: 'evnt_1',
+              runId: 'wrun_1',
+              eventType: 'run_created',
+              createdAt: CREATED_AT,
+              eventData: {
+                deploymentId: 'dpl_1',
+                workflowName: 'workflow',
+                input: null,
+              },
+            },
+            new Uint8Array()
+          ),
+          encodeFrame(
+            {
+              eventId: 'evnt_2',
+              runId: 'wrun_1',
+              eventType: 'run_started',
+              createdAt: CREATED_AT,
+            },
+            new Uint8Array()
+          ),
+        ]),
+        {
+          headers: {
+            'content-type': V4_FRAME_CONTENT_TYPE,
+            'x-wf-max-events': '10000',
+          },
+        }
+      );
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_2&remoteRefBehavior=resolve&returnAll=true',
+        method: 'GET',
+      })
+      .reply(200, encodeFrame({ _end: 1, hasMore: false }, new Uint8Array()), {
+        headers: { 'content-type': V4_FRAME_CONTENT_TYPE },
+      });
+
+    const result = await createWorkflowRunStartedEventV4(
+      { runId: 'wrun_1', specVersion: 5 },
+      { token: 'test-token', dispatcher: agent }
+    );
+
+    expect(result.events.map((event) => event.eventId)).toEqual([
+      'evnt_1',
+      'evnt_2',
+    ]);
+    expect(result.cursor).toBe('eid:evnt_2');
+    expect(result.hasMore).toBe(false);
+    agent.assertNoPendingInterceptors();
+  });
+
+  it('continues a graceful partial run_started stream from its sentinel cursor', async () => {
+    const agent = mockAgent();
+
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events/run_started',
+        method: 'POST',
+        headers: { accept: V4_FRAME_CONTENT_TYPE },
+      })
+      .reply(
+        200,
+        Buffer.concat([
+          encodeFrame(
+            {
+              eventId: 'evnt_1',
+              runId: 'wrun_1',
+              eventType: 'run_created',
+              createdAt: CREATED_AT,
+              eventData: {
+                deploymentId: 'dpl_1',
+                workflowName: 'workflow',
+                input: null,
+              },
+            },
+            new Uint8Array()
+          ),
+          encodeFrame(
+            { _end: 1, next: 'eid:evnt_1', hasMore: true },
+            new Uint8Array()
+          ),
+        ]),
+        {
+          headers: {
+            'content-type': V4_FRAME_CONTENT_TYPE,
+            'x-wf-max-events': '10000',
+          },
+        }
+      );
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_1&remoteRefBehavior=resolve&returnAll=true',
+        method: 'GET',
+      })
+      .reply(
+        200,
+        Buffer.concat([
+          encodeFrame(
+            {
+              eventId: 'evnt_2',
+              runId: 'wrun_1',
+              eventType: 'run_started',
+              createdAt: CREATED_AT,
+            },
+            new Uint8Array()
+          ),
+          encodeFrame(
+            { _end: 1, next: 'eid:evnt_2', hasMore: false },
+            new Uint8Array()
+          ),
+        ]),
+        { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } }
+      );
+
+    const result = await createWorkflowRunStartedEventV4(
+      { runId: 'wrun_1', specVersion: 5 },
+      { token: 'test-token', dispatcher: agent }
+    );
+
+    expect(result.events.map((event) => event.eventId)).toEqual([
+      'evnt_1',
+      'evnt_2',
+    ]);
+    expect(result.cursor).toBe('eid:evnt_2');
+    expect(result.hasMore).toBe(false);
+    agent.assertNoPendingInterceptors();
+  });
+
   it('requires the event-stream response requested by run_started', async () => {
     const origin =
       WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com';
@@ -1501,10 +1821,9 @@ describe('v4 POST frame meta forwards every field the splitter produces', () =>
 
 /**
  * The recycler in http-client only sees transport failures the v4 client
- * reports to it. This covers that wiring end to end: a `fetch()` that rejects
- * the way a wedged HTTP/2 session does must retire the shared events pool once
- * the failures reach the threshold. Without the `onTransportOutcome` hook in
- * `fetchV4` the recycler is never told anything and the pool lives forever.
+ * reports to it. A streamed response resolves `fetch()` as soon as headers
+ * arrive, before its body can fail, so the body consumer must own the success
+ * report or that early success erases every later stream failure.
  */
 describe('v4 transport reports failures to the events recycler', () => {
   // There is only an undici pool to retire while the adapter owns one:
@@ -1528,8 +1847,17 @@ describe('v4 transport reports failures to the events recycler', () => {
       }),
     });
 
-  it('rebuilds the shared pool after repeated stream timeouts', async () => {
-    vi.spyOn(globalThis, 'fetch').mockRejectedValue(wedgedSessionError());
+  it('rebuilds the shared pool after repeated response-body timeouts', async () => {
+    vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
+      const body = new ReadableStream({
+        start(controller) {
+          controller.error(wedgedSessionError());
+        },
+      });
+      return new Response(body, {
+        headers: { 'content-type': V4_FRAME_CONTENT_TYPE },
+      });
+    });
 
     // No `dispatcher` in the config: the request must resolve the shared one,
     // which is what the recycler owns.
@@ -1537,7 +1865,7 @@ describe('v4 transport reports failures to the events recycler', () => {
 
     for (let i = 0; i < EVENTS_RECYCLE_AFTER_CONSECUTIVE_FAILURES; i++) {
       await expect(
-        getWorkflowRunEventsV4('wrun_1', {}, { token: 'test-token' })
+        getWorkflowRunEventsV4({ runId: 'wrun_1' }, { token: 'test-token' })
       ).rejects.toThrow();
       // Still the same pool until the threshold is reached.
       if (i < EVENTS_RECYCLE_AFTER_CONSECUTIVE_FAILURES - 1) {
diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts
index a3586f95de..f2ef4a3a30 100644
--- a/packages/world-vercel/src/events-v4.ts
+++ b/packages/world-vercel/src/events-v4.ts
@@ -31,7 +31,9 @@ import {
   EventTypeSchema,
   getEventDataPayloadField,
   HookSchema,
-  type PaginationOptions,
+  type ListEventsByCorrelationIdParams,
+  type ListEventsParams,
+  type PaginatedResponse,
   StructuredErrorSchema,
   WaitSchema,
   WorkflowRunSchema,
@@ -98,15 +100,23 @@ import { isWsEventsTransportEnabled } from './ws-transport-enabled.js';
  * for a large run can legitimately take a while to drain — a whole-request
  * deadline would abort it mid-stream.
  */
+interface V4Response {
+  response: Response;
+  reportTransportOutcome(error?: unknown): void;
+}
+
 async function fetchV4(
   url: string,
   init: { method: string; headers: Headers; body?: Uint8Array },
   config: APIConfig | undefined,
   opName: string,
+  transportSuccess: 'headers' | 'body',
   attributes?: Record<string, string | number | string[]>
-): Promise<Response> {
+): Promise<V4Response> {
   const dispatcher = getEventsDispatcher(config);
-  return instrumentedFetch({
+  const reportTransportOutcome = (error?: unknown) =>
+    noteEventsTransportOutcome(dispatcher, error);
+  const response = await instrumentedFetch({
     method: init.method,
     url,
     headers: init.headers,
@@ -117,8 +127,8 @@ async function fetchV4(
     // request builds a fresh one. undici keeps a black-holed HTTP/2 session in
     // service indefinitely, so without this every request routed onto it fails
     // until the compute instance is recycled — see noteEventsTransportOutcome.
-    onTransportOutcome: (error) =>
-      noteEventsTransportOutcome(dispatcher, error),
+    onTransportOutcome: reportTransportOutcome,
+    deferTransportSuccess: transportSuccess === 'body',
     timeoutMs: null,
     logLabel: opName,
     // Read the body as bytes, not text: a CBOR error body (the fence 412
@@ -132,6 +142,7 @@ async function fetchV4(
         url
       ),
   });
+  return { response, reportTransportOutcome };
 }
 
 const EVENT_ID_HEADER = 'x-wf-event-id';
@@ -710,6 +721,7 @@ async function postWorkflowRunEventV4(
     { method: 'POST', headers, body: frame },
     config,
     'createEvent',
+    responseType === 'event-stream' ? 'body' : 'headers',
     {
       ...WorkflowEventsTransport('http'),
       ...WorkflowEventType(input.eventType),
@@ -739,7 +751,11 @@ export async function createWorkflowRunEventV4<T extends EventType>(
     if (reply) return decodeCreateEventResponse(reply, input.eventType);
   }
 
-  const response = await postWorkflowRunEventV4(input, 'materialized', config);
+  const { response } = await postWorkflowRunEventV4(
+    input,
+    'materialized',
+    config
+  );
 
   const contentType = response.headers.get('content-type');
   if (contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) {
@@ -779,18 +795,22 @@ async function decodeCreateEventResponse<T extends EventType>(
 
 export async function createWorkflowRunStartedEventV4(
   input: CreateEventV4InputBase,
-  config?: APIConfig
+  config?: APIConfig,
+  onEvent?: (event: Event) => void
 ) {
-  const response = await postWorkflowRunEventV4(
+  const stream = await postWorkflowRunEventV4(
     { ...input, eventType: 'run_started' },
     'event-stream',
     config
   );
-  const events: Event[] = [];
-  const page = await consumeEventFrameStream(response, 'createEvent', events);
-  assert(page.cursor, 'v4 createEvent: event stream missing cursor');
+  const replay = await consumeReplayLogResponse(
+    stream,
+    { runId: input.runId, onEvent },
+    config
+  );
+  assert(replay.cursor, 'v4 createEvent: event stream missing cursor');
   const maxEvents = MaxEventsHeaderSchema.safeParse(
-    response.headers.get(MAX_EVENTS_HEADER)
+    stream.response.headers.get(MAX_EVENTS_HEADER)
   );
   if (!maxEvents.success) {
     throw new WorkflowWorldError('v4 createEvent: invalid max-events header', {
@@ -799,7 +819,7 @@ export async function createWorkflowRunStartedEventV4(
     });
   }
 
-  return { events, ...page, maxEvents: maxEvents.data };
+  return { ...replay, maxEvents: maxEvents.data };
 }
 
 /** One event of a v4 batch POST, index-aligned with the response results. */
@@ -889,11 +909,12 @@ export async function createWorkflowRunEventsBatchV4(
   }
 
   const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/batch`;
-  const response = await fetchV4(
+  const { response } = await fetchV4(
     url,
     { method: 'POST', headers, body },
     config,
     'createEventBatch',
+    'headers',
     {
       ...WorkflowEventsTransport('http'),
       ...WorkflowEventType(input.events[0].eventType),
@@ -1190,13 +1211,19 @@ async function postEventFrameOverWs(
   );
 }
 
+export interface ReplayLogResult {
+  events: Event[];
+  cursor: string | null;
+  hasMore: boolean;
+}
+
 /**
  * Result of a `hook_received` POST that opted into the replay-log preload,
  * discriminated on `kind` (keyed on the response content type).
  */
 export type HookReceivedPreloadV4Result =
   /** The server streamed the replay log back as v4 frames. */
-  | (ListEventsV4Result & {
+  | (ReplayLogResult & {
       kind: 'stream';
       /**
        * The canonical event this write created or converged on (the resume
@@ -1225,37 +1252,42 @@ export type HookReceivedPreloadV4Result =
  * A server that supports the lazy-hook replay stream answers the consumer's
  * idempotent re-ensure with the run's complete replay log as v4 frames —
  * the same event-frame sequence LIST uses, ending with the `_end` sentinel.
- * A truncated stream (EOF without the sentinel) throws; the write is
- * deduplicated by the server's `(runId, resumeId)` constraint, so retrying
- * the whole request is safe and converges on the same canonical event.
+ * A truncated stream resumes with a GET after its last validated event. If it
+ * fails before producing any event, the outer POST retry remains safe because
+ * the server deduplicates `(runId, resumeId)` and returns the canonical event.
  */
 export async function createHookReceivedPreloadEventV4(
   input: CreateEventV4InputBase,
-  config?: APIConfig
+  config?: APIConfig,
+  onEvent?: (event: Event) => void
 ): Promise<HookReceivedPreloadV4Result> {
-  const response = await postWorkflowRunEventV4(
+  const stream = await postWorkflowRunEventV4(
     { ...input, eventType: 'hook_received' },
     'event-stream',
     config
   );
 
+  const { response } = stream;
   const contentType = response.headers.get('content-type');
   if (!contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) {
+    stream.reportTransportOutcome();
     return {
       kind: 'materialized',
       result: await decodeCreateEventResponse(response, 'hook_received'),
     };
   }
 
-  const events: Event[] = [];
-  const page = await consumeEventFrameStream(response, 'createEvent', events);
+  const replay = await consumeReplayLogResponse(
+    stream,
+    { runId: input.runId, onEvent },
+    config
+  );
   const maxEvents = MaxEventsHeaderSchema.safeParse(
     response.headers.get(MAX_EVENTS_HEADER)
   );
   return {
     kind: 'stream',
-    events,
-    ...page,
+    ...replay,
     canonicalEventId: response.headers.get(EVENT_ID_HEADER) ?? undefined,
     maxEvents: maxEvents.success ? maxEvents.data : undefined,
   };
@@ -1288,14 +1320,17 @@ export async function getEventV4(
   const url =
     `${baseUrl}/v4/runs/${encodeURIComponent(runId)}/events/${encodeURIComponent(eventId)}` +
     `?remoteRefBehavior=${remoteRefBehavior}`;
-  const response = await fetchV4(
+  const stream = await fetchV4(
     url,
     { method: 'GET', headers },
     config,
-    'getEvent'
+    'getEvent',
+    'body'
   );
+  const { response } = stream;
   const contentType = response.headers.get('content-type');
   if (!contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) {
+    stream.reportTransportOutcome();
     throw new Error(
       `v4 getEvent: expected ${V4_FRAME_CONTENT_TYPE}, got ${contentType ?? '(none)'}`
     );
@@ -1311,85 +1346,152 @@ export async function getEventV4(
 
   // GET emits a single frame (no sentinel); decodeFrames returns at EOF
   // after yielding it.
-  for await (const frame of decodeFrames(chunks)) {
-    return decodeEventFrame(frame);
+  try {
+    for await (const frame of decodeFrames(chunks)) {
+      stream.reportTransportOutcome();
+      return decodeEventFrame(frame);
+    }
+    stream.reportTransportOutcome();
+  } catch (error) {
+    stream.reportTransportOutcome(error);
+    throw error;
   }
   throw new Error(`v4 getEvent: empty frame stream for ${eventId}`);
 }
 
-export interface ListEventsV4Params extends PaginationOptions {
-  /**
-   * Whether the backend resolves payload bytes into each frame body.
-   * `resolve` (default) streams the bytes; `lazy` emits empty-body frames
-   * (the ref descriptor stays in the frame meta) — for metadata-only
-   * listings that would otherwise download every payload just to discard
-   * it.
-   */
-  remoteRefBehavior?: 'resolve' | 'lazy';
+class PartialEventStreamError extends WorkflowWorldError {
+  constructor(message: string, cause?: unknown) {
+    super(message, { code: 'TRANSPORT', cause });
+  }
 }
 
-export interface ListEventsV4Result {
-  events: Event[];
-  /** Trailing event-log cursor, or null when the stream contained no events. */
-  cursor: string | null;
-  /** Explicit "another page of results exists" flag from the sentinel. */
-  hasMore: boolean;
-}
+const MAX_PARTIAL_EVENT_STREAM_RETRIES = 3;
+
+type EventFrameStreamResult =
+  | {
+      kind: 'complete';
+      cursor: string | null;
+      hasMore: boolean;
+    }
+  | {
+      kind: 'partial';
+      error: PartialEventStreamError;
+    };
 
 async function consumeEventFrameStream(
-  response: Response,
+  stream: V4Response,
   opName: string,
-  events: Event[]
-): Promise<Pick<ListEventsV4Result, 'cursor' | 'hasMore'>> {
+  events: Event[],
+  onEvent?: (event: Event) => void
+): Promise<EventFrameStreamResult> {
+  const { response } = stream;
   const contentType = response.headers.get('content-type');
   if (!contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) {
+    stream.reportTransportOutcome();
     throw new Error(
       `v4 ${opName}: expected ${V4_FRAME_CONTENT_TYPE}, got ${contentType ?? '(none)'}`
     );
   }
 
   const chunks = response.body as unknown as AsyncIterable<Uint8Array>;
+  const frames = decodeFrames(chunks);
 
-  for await (const frame of decodeFrames(chunks)) {
-    if (frame.meta._end === 1) {
-      const end = EventStreamEndSchema.parse(frame.meta);
-      return { cursor: end.next ?? null, hasMore: end.hasMore };
-    }
-    if (Object.keys(frame.meta).some((key) => key.startsWith('_'))) {
-      throw new Error(`v4 ${opName}: unexpected control frame`);
+  try {
+    while (true) {
+      let next: IteratorResult<DecodedFrame>;
+      try {
+        next = await frames.next();
+      } catch (cause) {
+        const error = new PartialEventStreamError(
+          `v4 ${opName}: event frame stream failed after ${events.length} events`,
+          cause
+        );
+        stream.reportTransportOutcome(error);
+        return { kind: 'partial', error };
+      }
+      if (next.done) break;
+
+      try {
+        const frame = next.value;
+        if (frame.meta._end === 1) {
+          const end = EventStreamEndSchema.parse(frame.meta);
+          stream.reportTransportOutcome();
+          return {
+            kind: 'complete',
+            cursor: end.next ?? null,
+            hasMore: end.hasMore,
+          };
+        }
+        if (Object.keys(frame.meta).some((key) => key.startsWith('_'))) {
+          throw new Error(`v4 ${opName}: unexpected control frame`);
+        }
+        const event = decodeEventFrame(frame);
+        events.push(event);
+        onEvent?.(event);
+      } catch (error) {
+        stream.reportTransportOutcome();
+        throw error;
+      }
     }
-    events.push(decodeEventFrame(frame));
+  } finally {
+    void frames.return(undefined);
   }
 
-  throw new Error(
+  const error = new PartialEventStreamError(
     `v4 ${opName}: frame stream ended without the end-of-stream sentinel ` +
       `(${events.length} events read) — truncated response?`
   );
+  stream.reportTransportOutcome(error);
+  return { kind: 'partial', error };
 }
 
 /**
- * Drive a v4 frame-stream list response into an in-memory page. Used by
- * both the by-runId and by-correlationId list endpoints — the wire
- * shape is identical, only the URL differs.
- *
- * `headers` come from the caller's single getHttpConfig resolution (the
- * same call that produced the baseUrl in `url`) so each LIST resolves
- * auth exactly once.
+ * Finish a replay-log response, reusing every validated prefix. A graceful
+ * `hasMore` sentinel and an interrupted body now converge on the same GET
+ * continuation instead of making run_started download its accepted prefix
+ * again.
  */
-async function consumeListFrameStream(
-  url: string,
-  headers: Headers,
-  config: APIConfig | undefined,
-  opName: string,
-  events: Event[]
-): Promise<Pick<ListEventsV4Result, 'cursor' | 'hasMore'>> {
-  const response = await fetchV4(
-    url,
-    { method: 'GET', headers },
-    config,
-    opName
+async function consumeReplayLogResponse(
+  stream: V4Response,
+  params: Pick<ListEventsParams, 'runId' | 'onEvent'>,
+  config?: APIConfig
+): Promise<ReplayLogResult> {
+  const events: Event[] = [];
+  let page: { cursor: string | null; hasMore: boolean };
+  const consumed = await consumeEventFrameStream(
+    stream,
+    'createEvent',
+    events,
+    params.onEvent
+  );
+  if (consumed.kind === 'partial') {
+    const lastEvent = events.at(-1);
+    if (!lastEvent) throw consumed.error;
+    page = { cursor: `eid:${lastEvent.eventId}`, hasMore: true };
+  } else {
+    page = consumed;
+  }
+
+  if (!page.hasMore) return { events, ...page };
+  assert(page.cursor, 'v4 createEvent: partial event stream missing cursor');
+  const suffix = await getWorkflowRunEventsV4(
+    {
+      runId: params.runId,
+      pagination: { cursor: page.cursor },
+      onEvent: params.onEvent,
+    },
+    config
   );
-  return consumeEventFrameStream(response, opName, events);
+  events.push(...suffix.data);
+  assert(
+    suffix.data.length === 0 || suffix.cursor,
+    'v4 createEvent: non-empty continuation missing cursor'
+  );
+  return {
+    events,
+    cursor: suffix.cursor ?? page.cursor,
+    hasMore: suffix.hasMore,
+  };
 }
 
 /**
@@ -1397,20 +1499,29 @@ async function consumeListFrameStream(
  * Shared by the runId and correlationId list query builders so both send
  * `remoteRefBehavior` identically.
  */
-function appendListParams(sp: URLSearchParams, params: ListEventsV4Params) {
-  if (params.cursor) sp.set('cursor', params.cursor);
-  if (params.limit !== undefined) sp.set('limit', String(params.limit));
-  if (params.sortOrder) sp.set('sortOrder', params.sortOrder);
-  if (params.remoteRefBehavior) {
-    sp.set('remoteRefBehavior', params.remoteRefBehavior);
-  }
+function appendListParams(
+  sp: URLSearchParams,
+  params: ListEventsParams | ListEventsByCorrelationIdParams,
+  cursor: string | null
+) {
+  const { limit, sortOrder } = params.pagination ?? {};
+  if (cursor) sp.set('cursor', cursor);
+  if (limit !== undefined) sp.set('limit', String(limit));
+  if (sortOrder) sp.set('sortOrder', sortOrder);
+  sp.set(
+    'remoteRefBehavior',
+    params.resolveData === 'none' ? 'lazy' : 'resolve'
+  );
 }
 
-function paginationToQuery(params: ListEventsV4Params): string {
+function paginationToQuery(
+  params: ListEventsParams,
+  cursor: string | null
+): string {
   const sp = new URLSearchParams();
   // The World API uses an omitted limit for a complete event log.
-  if (params.limit === undefined) sp.set('returnAll', 'true');
-  appendListParams(sp, params);
+  if (params.pagination?.limit === undefined) sp.set('returnAll', 'true');
+  appendListParams(sp, params, cursor);
   return `?${sp.toString()}`;
 }
 
@@ -1425,38 +1536,51 @@ function paginationToQuery(params: ListEventsV4Params): string {
  * after its last validated event instead of downloading accepted frames again.
  */
 export async function getWorkflowRunEventsV4(
-  runId: string,
-  params: ListEventsV4Params = {},
+  params: ListEventsParams,
   config?: APIConfig
-): Promise<ListEventsV4Result> {
+): Promise<PaginatedResponse<Event>> {
   const { baseUrl, headers } = await getHttpConfig(config);
   const events: Event[] = [];
-  let cursor = params.cursor;
+  let cursor = params.pagination?.cursor ?? null;
 
-  while (true) {
+  for (let partialRetries = 0; ; partialRetries++) {
     const url =
-      `${baseUrl}/v4/runs/${encodeURIComponent(runId)}/events` +
-      paginationToQuery({ ...params, cursor });
-    try {
-      const page = await consumeListFrameStream(
-        url,
-        headers,
-        config,
-        'listEvents',
-        events
-      );
-      return { events, ...page };
-    } catch (error) {
+      `${baseUrl}/v4/runs/${encodeURIComponent(params.runId)}/events` +
+      paginationToQuery(params, cursor);
+    const stream = await fetchV4(
+      url,
+      { method: 'GET', headers },
+      config,
+      'listEvents',
+      'body'
+    );
+    const result = await consumeEventFrameStream(
+      stream,
+      'listEvents',
+      events,
+      params.onEvent
+    );
+    if (result.kind === 'partial') {
       const lastEvent = events.at(-1);
       if (
-        params.limit !== undefined ||
+        partialRetries === MAX_PARTIAL_EVENT_STREAM_RETRIES ||
+        params.pagination?.limit !== undefined ||
         !lastEvent ||
         `eid:${lastEvent.eventId}` === cursor
       ) {
-        throw error;
+        throw result.error;
       }
       cursor = `eid:${lastEvent.eventId}`;
+      continue;
+    }
+
+    if (params.pagination?.limit === undefined && result.hasMore) {
+      throw new WorkflowWorldError(
+        `v4 listEvents: returnAll response was unexpectedly paginated for run ${params.runId}`,
+        { code: 'SCHEMA_VALIDATION' }
+      );
     }
+    return { data: events, cursor: result.cursor, hasMore: result.hasMore };
   }
 }
 
@@ -1475,24 +1599,28 @@ export async function getWorkflowRunEventsV4(
  * the page by run id.
  */
 export async function getEventsByCorrelationIdV4(
-  correlationId: string,
-  runId: string,
-  params: ListEventsV4Params = {},
+  params: ListEventsByCorrelationIdParams,
   config?: APIConfig
-): Promise<ListEventsV4Result> {
+): Promise<PaginatedResponse<Event>> {
   const { baseUrl, headers } = await getHttpConfig(config);
   const sp = new URLSearchParams();
-  sp.set('correlationId', correlationId);
-  sp.set('runId', runId);
-  appendListParams(sp, params);
+  sp.set('correlationId', params.correlationId);
+  sp.set('runId', params.runId);
+  appendListParams(sp, params, params.pagination?.cursor ?? null);
   const url = `${baseUrl}/v4/events?${sp.toString()}`;
   const events: Event[] = [];
-  const page = await consumeListFrameStream(
+  const stream = await fetchV4(
     url,
-    headers,
+    { method: 'GET', headers },
     config,
     'listEventsByCorrelationId',
+    'body'
+  );
+  const result = await consumeEventFrameStream(
+    stream,
+    'listEventsByCorrelationId',
     events
   );
-  return { events, ...page };
+  if (result.kind === 'partial') throw result.error;
+  return { data: events, cursor: result.cursor, hasMore: result.hasMore };
 }
diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts
index 5fea9675a4..87ba906319 100644
--- a/packages/world-vercel/src/events.test.ts
+++ b/packages/world-vercel/src/events.test.ts
@@ -1,13 +1,15 @@
 import { Buffer } from 'node:buffer';
 import { gzipSync } from 'node:zlib';
+import { WorkflowWorldError } from '@workflow/errors';
 import type { AnyEventRequest, CreateEventParams } from '@workflow/world';
 import { decode, encode } from 'cbor-x';
 import { ulid } from 'ulid';
 import { MockAgent } from 'undici';
-import { describe, expect, it } from 'vitest';
+import { describe, expect, it, vi } from 'vitest';
 import {
   createWorkflowRunEvent,
   getWorkflowRunEvents,
+  getWorkflowRunEventsByCorrelationId,
   splitEventDataForV4,
 } from './events.js';
 import { encodeFrame, V4_FRAME_CONTENT_TYPE } from './frames.js';
@@ -373,6 +375,44 @@ describe('createWorkflowRunEvent result contract', () => {
     ).rejects.toMatchObject(error);
     agent.assertNoPendingInterceptors();
   });
+
+  it('does not retry an observer failure that looks like a transport error', async () => {
+    const agent = mockAgent();
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events/run_started',
+        method: 'POST',
+      })
+      .reply(200, runStartedResponse(), {
+        headers: {
+          'content-type': V4_FRAME_CONTENT_TYPE,
+          'x-wf-event-id': 'evnt_1',
+          'x-wf-run-id': 'wrun_1',
+          'x-wf-created-at': STARTED_AT.toISOString(),
+          'x-wf-max-events': '10000',
+        },
+      });
+
+    const error = new WorkflowWorldError('observer failed', {
+      code: 'TRANSPORT',
+    });
+    const onEvent = vi.fn(() => {
+      throw error;
+    });
+
+    await expect(
+      createWorkflowRunEvent(
+        'wrun_1',
+        { eventType: 'run_started', specVersion: 2 } as AnyEventRequest,
+        { onEvent },
+        { token: 'test-token', dispatcher: agent }
+      )
+    ).rejects.toBe(error);
+
+    expect(onEvent).toHaveBeenCalledOnce();
+    agent.assertNoPendingInterceptors();
+  });
 });
 
 /** POSTs a v4 step_started with `params` and returns the decoded frame meta. */
@@ -1393,7 +1433,14 @@ describe('getWorkflowRunEvents legacy structured-error compatibility', () => {
  * fallback preserves their (correct, if slower) behavior.
  */
 describe('getWorkflowRunEvents hasMore mapping', () => {
-  function mockListResponse(agent: MockAgent, sentinelMeta: object) {
+  function mockListResponse(
+    agent: MockAgent,
+    sentinelMeta: object,
+    query: Record<string, string> = {
+      returnAll: 'true',
+      remoteRefBehavior: 'resolve',
+    }
+  ) {
     const frames = Buffer.concat([
       encodeFrame(
         {
@@ -1412,9 +1459,7 @@ describe('getWorkflowRunEvents hasMore mapping', () => {
       .intercept({
         path: '/api/v4/runs/wrun_1/events',
         method: 'GET',
-        // These tests omit the limit and use the default resolveData
-        // ('all' → resolve); match both translated query params.
-        query: { returnAll: 'true', remoteRefBehavior: 'resolve' },
+        query,
       })
       .reply(200, frames, {
         headers: { 'content-type': V4_FRAME_CONTENT_TYPE },
@@ -1439,10 +1484,14 @@ describe('getWorkflowRunEvents hasMore mapping', () => {
 
   it('maps an explicit hasMore:true through', async () => {
     const agent = mockAgent();
-    mockListResponse(agent, { _end: 1, next: 'cursor-2', hasMore: true });
+    mockListResponse(
+      agent,
+      { _end: 1, next: 'cursor-2', hasMore: true },
+      { limit: '500', remoteRefBehavior: 'resolve' }
+    );
 
     const result = await getWorkflowRunEvents(
-      { runId: 'wrun_1' },
+      { runId: 'wrun_1', pagination: { limit: 500 } },
       { token: 'test-token', dispatcher: agent }
     );
 
@@ -1511,7 +1560,7 @@ describe('getWorkflowRunEvents by correlation id is scoped to the run', () => {
         headers: { 'content-type': V4_FRAME_CONTENT_TYPE },
       });
 
-    const result = await getWorkflowRunEvents(
+    const result = await getWorkflowRunEventsByCorrelationId(
       { correlationId: 'step_001', runId: 'wrun_1' },
       { token: 'test-token', dispatcher: agent }
     );
@@ -1842,7 +1891,7 @@ describe('createWorkflowRunEvent hook_received replay preload', () => {
     agent.assertNoPendingInterceptors();
   });
 
-  it('rejects a truncated preload stream (no end sentinel)', async () => {
+  it('continues a truncated preload stream after its last event', async () => {
     const agent = mockAgent();
     agent
       .get(ORIGIN)
@@ -1866,15 +1915,43 @@ describe('createWorkflowRunEvent hook_received replay preload', () => {
           },
           PAYLOAD
         ),
+        {
+          headers: {
+            'content-type': V4_FRAME_CONTENT_TYPE,
+            'x-wf-event-id': 'evnt_4',
+          },
+        }
+      );
+
+    agent
+      .get(ORIGIN)
+      .intercept({
+        path: '/api/v4/runs/wrun_1/events?cursor=eid%3Aevnt_4&remoteRefBehavior=resolve&returnAll=true',
+        method: 'GET',
+      })
+      .reply(
+        200,
+        encodeFrame(
+          { _end: 1, next: 'eid:evnt_4', hasMore: false },
+          new Uint8Array()
+        ),
         { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } }
       );
 
-    await expect(
-      createWorkflowRunEvent('wrun_1', hookReceivedRequest(), preloadParams, {
+    const result = await createWorkflowRunEvent(
+      'wrun_1',
+      hookReceivedRequest(),
+      preloadParams,
+      {
         token: 'test-token',
         dispatcher: agent,
-      })
-    ).rejects.toThrow(/end-of-stream sentinel/);
+      }
+    );
+
+    expect(result.event?.eventId).toBe('evnt_4');
+    expect(result.events).toHaveLength(1);
+    expect(result.cursor).toBe('eid:evnt_4');
+    expect(result.hasMore).toBe(false);
     agent.assertNoPendingInterceptors();
   });
 
diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts
index 1d0279229d..e4f10b6ae0 100644
--- a/packages/world-vercel/src/events.ts
+++ b/packages/world-vercel/src/events.ts
@@ -31,6 +31,7 @@
  * the v3 path.
  */
 
+import assert from 'node:assert/strict';
 import { HookNotFoundError, WorkflowWorldError } from '@workflow/errors';
 import {
   type AnyEventRequest,
@@ -62,15 +63,10 @@ import {
   getEventsByCorrelationIdV4,
   getEventV4,
   getWorkflowRunEventsV4,
-  type ListEventsV4Params,
 } from './events-v4.js';
 import { decode as decodeRunId } from './run-id/index.js';
 import { cancelWorkflowRunV1, createWorkflowRunV1 } from './runs.js';
-import {
-  type APIConfig,
-  DEFAULT_RESOLVE_DATA_OPTION,
-  makeRequest,
-} from './utils.js';
+import { type APIConfig, makeRequest } from './utils.js';
 
 function validateWorkflowRunIdTimestamp(id: string): string | null {
   const raw = id.startsWith('wrun_') ? id.slice('wrun_'.length) : id;
@@ -436,26 +432,17 @@ export async function getEvent(
 }
 
 export async function getWorkflowRunEvents(
-  params: ListEventsParams | ListEventsByCorrelationIdParams,
+  params: ListEventsParams,
   config?: APIConfig
 ): Promise<PaginatedResponse<Event>> {
-  const { pagination, resolveData = DEFAULT_RESOLVE_DATA_OPTION } = params;
-  // `resolveData: 'none'` leaves payload refs unresolved, so the backend can
-  // skip reading and streaming their contents. The validated lazy descriptors
-  // remain on the returned events.
-  const listParams: ListEventsV4Params = {
-    ...pagination,
-    remoteRefBehavior: resolveData === 'none' ? 'lazy' : 'resolve',
-  };
+  return getWorkflowRunEventsV4(params, config);
+}
 
-  const result = await ('correlationId' in params
-    ? getEventsByCorrelationIdV4(
-        params.correlationId,
-        params.runId,
-        listParams,
-        config
-      )
-    : getWorkflowRunEventsV4(params.runId, listParams, config));
+export async function getWorkflowRunEventsByCorrelationId(
+  params: ListEventsByCorrelationIdParams,
+  config?: APIConfig
+): Promise<PaginatedResponse<Event>> {
+  const result = await getEventsByCorrelationIdV4(params, config);
 
   // 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
@@ -464,10 +451,7 @@ export async function getWorkflowRunEvents(
   // `hasMore`/`cursor` stay the backend's, so a page that filters down to
   // nothing is still followed by the next one.
   return {
-    data:
-      'correlationId' in params
-        ? result.events.filter((event) => event.runId === params.runId)
-        : result.events,
+    data: result.data.filter((event) => event.runId === params.runId),
     // The cursor is present even on the final page because it is also the
     // incremental-load resume point. `hasMore` is the pagination signal.
     cursor: result.cursor,
@@ -582,12 +566,33 @@ export async function createWorkflowRunEventBatch(
   };
 }
 
+class EventObserverError extends Error {
+  constructor(readonly error: unknown) {
+    super('event observer failed');
+  }
+}
+
 export async function createWorkflowRunEvent<T extends AnyEventRequest>(
   id: string | null,
   data: T,
   params?: CreateEventParams,
   config?: APIConfig
 ): Promise<EventResult<T['eventType']>> {
+  const onEvent = params?.onEvent;
+  const requestParams =
+    onEvent === undefined
+      ? params
+      : {
+          ...params,
+          onEvent(event: Event) {
+            try {
+              onEvent(event);
+            } catch (error) {
+              throw new EventObserverError(error);
+            }
+          },
+        };
+
   try {
     // Retry transient transport failures (UND_ERR_REQ_RETRY, ECONNRESET,
     // socket/headers timeouts, transient 5xx) in-process for event types that
@@ -598,7 +603,7 @@ export async function createWorkflowRunEvent<T extends AnyEventRequest>(
     // types (step_started, step_retrying, hook_received) run once. See
     // ./event-retry for the validated per-event classification.
     const result = await withEventPostRetry(
-      () => createWorkflowRunEventInner(id, data, params, config),
+      () => createWorkflowRunEventInner(id, data, requestParams, config),
       data.eventType,
       {
         // The atomic lazy-resume shape is deduplicated server-side by the
@@ -630,6 +635,7 @@ export async function createWorkflowRunEvent<T extends AnyEventRequest>(
     }
     return result as EventResult<T['eventType']>;
   } catch (err) {
+    if (err instanceof EventObserverError) throw err.error;
     // 404 on hook_disposed / hook_received → already-disposed hook.
     if (
       isHookEventRequiringExistence(data.eventType) &&
@@ -744,7 +750,11 @@ async function createWorkflowRunEventInner(
   };
 
   if (data.eventType === 'run_started' && !params?.skipPreload) {
-    const result = await createWorkflowRunStartedEventV4(input, config);
+    const result = await createWorkflowRunStartedEventV4(
+      input,
+      config,
+      params?.onEvent
+    );
     const runCreated = result.events.find(
       (event) => event.eventType === 'run_created'
     );
@@ -761,32 +771,12 @@ async function createWorkflowRunEventInner(
         'v4 createEvent: run_started stream is missing run_started'
       );
     }
-
-    let attributes = runCreated.eventData.attributes ?? {};
-    let updatedAt = runStarted.createdAt;
-    for (const event of result.events) {
-      if (event.eventType === 'attr_set') {
-        attributes = applyAttributeChanges(attributes, event.eventData.changes);
-        updatedAt = event.createdAt;
-      }
-    }
+    const run = reconstructRunFromReplayEvents(result.events);
+    assert(run);
 
     return {
       event: runStarted,
-      run: {
-        runId: runCreated.runId,
-        status: 'running',
-        deploymentId: runCreated.eventData.deploymentId,
-        workflowName: runCreated.eventData.workflowName,
-        specVersion: runCreated.specVersion,
-        executionContext: runCreated.eventData.executionContext,
-        input: runCreated.eventData.input,
-        attributes,
-        encryptionPublicKey: runCreated.eventData.encryptionPublicKey,
-        startedAt: runStarted.createdAt,
-        createdAt: runCreated.createdAt,
-        updatedAt,
-      },
+      run,
       events: result.events,
       cursor: result.cursor,
       hasMore: result.hasMore,
@@ -811,7 +801,8 @@ async function createWorkflowRunEventInner(
     // an S3-backed hook payload the runtime would discard anyway.
     const outcome = await createHookReceivedPreloadEventV4(
       { ...input, remoteRefBehavior: 'lazy' },
-      config
+      config,
+      params.onEvent
     );
     if (outcome.kind === 'materialized') {
       // Older server (or optimization declined): the write still succeeded
diff --git a/packages/world-vercel/src/http-core.ts b/packages/world-vercel/src/http-core.ts
index b8122bb7fd..c0729fbeb1 100644
--- a/packages/world-vercel/src/http-core.ts
+++ b/packages/world-vercel/src/http-core.ts
@@ -487,6 +487,8 @@ export interface InstrumentedFetchOptions extends HttpClientSpanOptions {
    * connections stop delivering (see noteEventsTransportOutcome).
    */
   onTransportOutcome?: (error?: unknown) => void;
+  /** Let a streaming body consumer report success after it finishes. */
+  deferTransportSuccess?: boolean;
 }
 
 /**
@@ -520,6 +522,7 @@ export async function instrumentedFetch(
     attributes,
     durationAttribute,
     onTransportOutcome,
+    deferTransportSuccess = false,
   } = opts;
   const label = logLabel ?? url;
 
@@ -601,7 +604,9 @@ export async function instrumentedFetch(
         throw error;
       }
       const ms = Date.now() - start;
-      onTransportOutcome?.();
+      if (!deferTransportSuccess || !response.ok) {
+        onTransportOutcome?.();
+      }
 
       httpLog(method, label, response, ms);
       recordClientSpanStatus(span, response.status);
diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts
index 186c759805..72abe6c57b 100644
--- a/packages/world-vercel/src/storage.ts
+++ b/packages/world-vercel/src/storage.ts
@@ -8,6 +8,7 @@ import {
   createWorkflowRunEventBatch,
   getEvent,
   getWorkflowRunEvents,
+  getWorkflowRunEventsByCorrelationId,
 } from './events.js';
 import { getHook, getHookByToken, listHooks } from './hooks.js';
 import { instrumentObject } from './instrumentObject.js';
@@ -53,7 +54,8 @@ export function createStorage(config?: APIConfig): Storage {
         createWorkflowRunEventBatch(runId, events, params, config),
       get: (runId, eventId, params) => getEvent(runId, eventId, params, config),
       list: (params) => getWorkflowRunEvents(params, config),
-      listByCorrelationId: (params) => getWorkflowRunEvents(params, config),
+      listByCorrelationId: (params) =>
+        getWorkflowRunEventsByCorrelationId(params, config),
     },
     hooks: {
       get: (hookId, params) => getHook(hookId, params, config),
diff --git a/packages/world-vercel/src/trace-propagation.test.ts b/packages/world-vercel/src/trace-propagation.test.ts
index b22b2d0ec6..a8155c6f5d 100644
--- a/packages/world-vercel/src/trace-propagation.test.ts
+++ b/packages/world-vercel/src/trace-propagation.test.ts
@@ -166,7 +166,7 @@ describe('v4 event requests (fetchV4) trace propagation', () => {
     agent
       .get(origin)
       .intercept({
-        path: '/api/v4/runs/wrun_1/events?returnAll=true',
+        path: '/api/v4/runs/wrun_1/events?remoteRefBehavior=resolve&returnAll=true',
         method: 'GET',
       })
       .reply(200, encodeFrame({ _end: 1, hasMore: false }, new Uint8Array(0)), {
@@ -184,8 +184,7 @@ describe('v4 event requests (fetchV4) trace propagation', () => {
       traceId = span.spanContext().traceId;
       spanId = span.spanContext().spanId;
       await getWorkflowRunEventsV4(
-        'wrun_1',
-        {},
+        { runId: 'wrun_1' },
         { token: 'test-token', dispatcher: agent }
       );
       span.end();
diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts
index 6f0c55ba6a..9ac9226fe0 100644
--- a/packages/world/src/events.ts
+++ b/packages/world/src/events.ts
@@ -941,6 +941,11 @@ export interface CreateEventParams {
    * `resumeHook()` must not set it.
    */
   preloadEvents?: true;
+  /**
+   * Observe replay-preload events as their frames are decoded. This is a
+   * client-side delivery hook only; it is never serialized to a backend.
+   */
+  onEvent?: (event: Event) => void;
 }
 
 /**
@@ -1113,6 +1118,11 @@ export interface ListEventsParams {
   /** Omit `limit` to return every remaining event. */
   pagination?: PaginationOptions;
   resolveData?: ResolveData;
+  /**
+   * Observe events as a streaming World decodes them. The callback runs
+   * synchronously and therefore applies response-stream backpressure.
+   */
+  onEvent?: (event: Event) => void;
 }
 
 export interface ListEventsByCorrelationIdParams {
@@ -1126,6 +1136,7 @@ export interface ListEventsByCorrelationIdParams {
    * event id alone is not.
    */
   runId: string;
+  /** Omit `limit` to return every remaining event. */
   pagination?: PaginationOptions;
   resolveData?: ResolveData;
 }