fix(world-vercel): retry idempotent event POSTs in-process to avoid step re-execution (#2675)

* fix(world-vercel): retry idempotent event POSTs in-process to avoid step re-execution

undici's RetryAgent never retries a POST, so a transient transport blip (UND_ERR_REQ_RETRY, ECONNRESET, socket/headers timeout, transient 5xx) when committing a step's terminal event bubbles out, the queue redelivers, and the step's user code re-executes with attempt++ even though it already ran to completion.

workflow-server makes these writes idempotent in outcome: entity handlers run before the event-log row is inserted and state transitions are conditional writes excluding terminal states, so a retry whose original landed throws before any row is written and surfaces as a 409 the SDK already handles. This adds a bounded in-process retry (new event-retry.ts) gated by a validated per-event EVENT_RETRY_ELIGIBILITY map, excluding step_started (double-increments attempt), step_retrying (appends a duplicate row), and hook_received (no server guard).

Complements #2666, which routes completion-persistence failures to queue redelivery instead of recording them as user failures; this avoids the redelivery (and re-execution) for transient blips.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(world-vercel): address review on event-POST retry

- Don't retry external cancellation: drop AbortError from the transient set (keep self-timeout TimeoutError), so a caller-requested abort isn't re-issued or stalled by the backoff budget.

- Add DEBUG-gated logging on each retry and on retry exhaustion so an in-process retry vs. a fall-through to queue redelivery is distinguishable in logs.

- Clarify docs: a landed retry surfaces as 409 for most types, but run_started/attr_set return 200 success (not a 409).

- Add createWorkflowRunEvent integration tests (events-retry.test.ts): eventType is threaded into the retry wrapper, and the 404->HookNotFoundError mapping still fires after the retry loop.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: Peter Wielander <mittgfu@gmail.com>

Author: 3 people

.changeset/retry-idempotent-event-posts.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-vercel': patch
+---
+
+Retry transient transport failures (e.g. `UND_ERR_REQ_RETRY`, `ECONNRESET`, socket timeouts, 5xx) in-process for idempotent-on-retry event POSTs, so a brief network blip after a step completes no longer re-executes the step. `step_started`, `step_retrying`, and `hook_received` are excluded as they are not safe to blindly retry.

packages/world-vercel/src/event-retry.test.ts

@@ -0,0 +1,228 @@
+import {
+  EntityConflictError,
+  RunExpiredError,
+  ThrottleError,
+  TooEarlyError,
+  WorkflowWorldError,
+} from '@workflow/errors';
+import { EventTypeSchema } from '@workflow/world';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+  EVENT_RETRY_ELIGIBILITY,
+  isRetryableEventPostError,
+  MAX_EVENT_POST_RETRIES,
+  withEventPostRetry,
+} from './event-retry.js';
+
+const transportErr = (code: string) =>
+  Object.assign(new Error(`transport ${code}`), { code });
+
+// undici `fetch` wraps low-level failures in a TypeError whose `cause` carries
+// the real code — the classifier must walk the cause chain.
+const fetchFailed = (code: string) =>
+  Object.assign(new TypeError('fetch failed'), {
+    cause: Object.assign(new Error('underlying'), { code }),
+  });
+
+describe('EVENT_RETRY_ELIGIBILITY', () => {
+  it('classifies every world event type (no gaps)', () => {
+    for (const type of EventTypeSchema.options) {
+      const policy = EVENT_RETRY_ELIGIBILITY[type];
+      expect(policy, `missing policy for ${type}`).toBeDefined();
+      expect(typeof policy.retryable).toBe('boolean');
+      expect(policy.reason.length).toBeGreaterThan(0);
+    }
+  });
+
+  it('marks idempotent-on-retry events retryable', () => {
+    for (const type of [
+      'run_created',
+      'run_started',
+      'run_completed',
+      'run_failed',
+      'run_cancelled',
+      'attr_set',
+      'step_created',
+      'step_completed',
+      'step_failed',
+      'wait_created',
+      'wait_completed',
+      'hook_created',
+      'hook_disposed',
+    ] as const) {
+      expect(EVENT_RETRY_ELIGIBILITY[type].retryable, type).toBe(true);
+    }
+  });
+
+  it('excludes events that are unsafe to blindly retry', () => {
+    // step_started double-increments attempt; step_retrying / hook_received
+    // append a duplicate event row; hook_conflict is server-originated.
+    for (const type of [
+      'step_started',
+      'step_retrying',
+      'hook_received',
+      'hook_conflict',
+    ] as const) {
+      expect(EVENT_RETRY_ELIGIBILITY[type].retryable, type).toBe(false);
+    }
+  });
+});
+
+describe('isRetryableEventPostError', () => {
+  it('retries transient 5xx', () => {
+    for (const status of [500, 502, 503, 504]) {
+      expect(
+        isRetryableEventPostError(new WorkflowWorldError('boom', { status }))
+      ).toBe(true);
+    }
+  });
+
+  it('does not retry 4xx definitive responses', () => {
+    for (const status of [400, 404]) {
+      expect(
+        isRetryableEventPostError(new WorkflowWorldError('nope', { status }))
+      ).toBe(false);
+    }
+  });
+
+  it('does not retry server-considered conflicts/terminal/too-early/throttle', () => {
+    expect(isRetryableEventPostError(new EntityConflictError('409'))).toBe(
+      false
+    );
+    expect(isRetryableEventPostError(new RunExpiredError('410'))).toBe(false);
+    expect(isRetryableEventPostError(new TooEarlyError('425'))).toBe(false);
+    expect(isRetryableEventPostError(new ThrottleError('429'))).toBe(false);
+  });
+
+  it('retries a body-parse failure (write may have landed)', () => {
+    expect(
+      isRetryableEventPostError(
+        new WorkflowWorldError('parse', { code: 'PARSE_ERROR' })
+      )
+    ).toBe(true);
+  });
+
+  it('retries raw transport errors by code', () => {
+    for (const code of [
+      'ECONNRESET',
+      'ETIMEDOUT',
+      'UND_ERR_SOCKET',
+      'UND_ERR_REQ_RETRY',
+      'UND_ERR_HEADERS_TIMEOUT',
+    ]) {
+      expect(isRetryableEventPostError(transportErr(code)), code).toBe(true);
+    }
+  });
+
+  it('retries transport errors hidden under a `fetch failed` cause', () => {
+    expect(isRetryableEventPostError(fetchFailed('UND_ERR_SOCKET'))).toBe(true);
+  });
+
+  it('retries our own timeout (TimeoutError) but not an external abort (AbortError)', () => {
+    // Self-deadline via AbortSignal.timeout → TimeoutError → ambiguous, retry.
+    expect(
+      isRetryableEventPostError(
+        Object.assign(new Error('timed out'), { name: 'TimeoutError' })
+      )
+    ).toBe(true);
+    // Caller-requested cancellation surfaces as AbortError → must NOT be retried.
+    expect(
+      isRetryableEventPostError(
+        Object.assign(new Error('aborted'), { name: 'AbortError' })
+      )
+    ).toBe(false);
+    // Also when wrapped by makeRequest as a WorkflowWorldError(cause).
+    expect(
+      isRetryableEventPostError(
+        new WorkflowWorldError('request aborted', {
+          cause: Object.assign(new Error('aborted'), { name: 'AbortError' }),
+        })
+      )
+    ).toBe(false);
+  });
+
+  it('does not retry an unclassified error', () => {
+    expect(isRetryableEventPostError(new Error('something else'))).toBe(false);
+  });
+});
+
+describe('withEventPostRetry', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it('retries a retryable event past a transient blip and returns the result', async () => {
+    let calls = 0;
+    const fn = vi.fn(async () => {
+      calls++;
+      if (calls === 1) throw transportErr('ECONNRESET');
+      return 'ok';
+    });
+
+    const p = withEventPostRetry(fn, 'step_completed');
+    await vi.runAllTimersAsync();
+
+    await expect(p).resolves.toBe('ok');
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('surfaces a 409 that appears on a retry (original landed)', async () => {
+    let calls = 0;
+    const fn = vi.fn(async () => {
+      calls++;
+      if (calls === 1) throw transportErr('ECONNRESET');
+      // The first attempt actually landed; the retry observes the conflict.
+      throw new EntityConflictError('already completed');
+    });
+
+    const p = withEventPostRetry(fn, 'step_completed').catch((e) => e);
+    await vi.runAllTimersAsync();
+
+    const err = await p;
+    expect(EntityConflictError.is(err)).toBe(true);
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('gives up after MAX_EVENT_POST_RETRIES and throws the last error', async () => {
+    const fn = vi.fn(async () => {
+      throw transportErr('ECONNRESET');
+    });
+
+    const p = withEventPostRetry(fn, 'step_completed').catch((e) => e);
+    await vi.runAllTimersAsync();
+
+    const err = await p;
+    expect((err as { code?: string }).code).toBe('ECONNRESET');
+    expect(fn).toHaveBeenCalledTimes(MAX_EVENT_POST_RETRIES + 1);
+  });
+
+  it('does not retry a definitive failure even for a retryable event', async () => {
+    const fn = vi.fn(async () => {
+      throw new WorkflowWorldError('bad request', { status: 400 });
+    });
+
+    await expect(withEventPostRetry(fn, 'step_completed')).rejects.toThrow();
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('does not retry an excluded event (step_started — protects the attempt counter)', async () => {
+    const fn = vi.fn(async () => {
+      throw transportErr('ECONNRESET');
+    });
+
+    await expect(withEventPostRetry(fn, 'step_started')).rejects.toThrow();
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('does not retry an excluded event (hook_received — avoids double-delivery)', async () => {
+    const fn = vi.fn(async () => {
+      throw transportErr('ECONNRESET');
+    });
+
+    await expect(withEventPostRetry(fn, 'hook_received')).rejects.toThrow();
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+});