From 17d4ce225309d83a434cb2e4b1a34e2b647b4e04 Mon Sep 17 00:00:00 2001 From: Karthik Kalyan <105607645+karthikscale3@users.noreply.github.com> Date: Tue, 7 Jul 2026 11:44:57 -0700 Subject: [PATCH 1/8] cli: read list views from world.analytics when available (#2648) * Add workflow analytics world APIs * cli: read list views from world.analytics when available inspect list views (runs, steps, events, hooks, sleeps) now read from the optional world.analytics namespace when the active backend provides one, falling back to the runtime storage APIs otherwise. Payload and detail views are unchanged. Deprecate --with-data for list views; payloads are viewable per-resource via 'inspect '. * cli: keep hook listing on the runtime storage API The analytics read path omits ownerId (and the secret hook token), so routing hook listing through it silently drops the ownerId column. Keep inspect hooks on the runtime APIs, consistent with the web observability UI. Runs, steps, events, and sleeps continue to use the analytics read path when available. Co-Authored-By: Claude Opus 4.8 * Handle analytics access metadata in CLI * test(cli): preserve analytics pageInfo in json output * fix(cli): paginate analytics sleeps output * fix(cli): correct deprecation message flag name to --withData The list-view deprecation warning referenced '--with-data', but the actual oclif flag is '--withData' (with '-d' alias); '--with-data' errors with "Nonexistent flag". Fix the warning text, the doc comment, and the changeset to reference the real flag name. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(cli): preserve inspect json array output * fix(cli): fall back when analytics lists are empty --------- Co-authored-by: Claude Opus 4.8 --- .changeset/cli-analytics-list-reads.md | 5 + packages/cli/src/commands/cancel.ts | 41 +- packages/cli/src/commands/inspect.ts | 11 +- packages/cli/src/commands/start.ts | 11 +- packages/cli/src/lib/inspect/errors.ts | 38 ++ packages/cli/src/lib/inspect/output.test.ts | 462 ++++++++++++++++++- packages/cli/src/lib/inspect/output.ts | 476 +++++++++++++------- packages/cli/src/lib/inspect/pagination.ts | 46 +- packages/cli/src/lib/inspect/run.ts | 28 +- 9 files changed, 921 insertions(+), 197 deletions(-) create mode 100644 .changeset/cli-analytics-list-reads.md create mode 100644 packages/cli/src/lib/inspect/errors.ts diff --git a/.changeset/cli-analytics-list-reads.md b/.changeset/cli-analytics-list-reads.md new file mode 100644 index 0000000000..4419539bdf --- /dev/null +++ b/.changeset/cli-analytics-list-reads.md @@ -0,0 +1,5 @@ +--- +'@workflow/cli': patch +--- + +`inspect` list views (runs, steps, events, sleeps) now read from the optional `world.analytics` namespace when the backend provides one, falling back to the runtime storage APIs otherwise. Hook listing stays on the runtime storage APIs (the analytics rows omit `ownerId` and the hook token). Payload/detail views are unchanged. The `--withData` flag is deprecated for list views; use `workflow inspect ` to view payloads for a single resource. diff --git a/packages/cli/src/commands/cancel.ts b/packages/cli/src/commands/cancel.ts index 210ef2f257..6186bd190b 100644 --- a/packages/cli/src/commands/cancel.ts +++ b/packages/cli/src/commands/cancel.ts @@ -2,10 +2,15 @@ import readline from 'node:readline'; import { Args, Flags } from '@oclif/core'; import { cancelRun } from '@workflow/core/runtime'; import { parseWorkflowName } from '@workflow/utils/parse-name'; +import type { WorkflowRun } from '@workflow/world'; import chalk from 'chalk'; import Table from 'easy-table'; import { BaseCommand } from '../base.js'; import { LOGGING_CONFIG, logger } from '../lib/config/log.js'; +import { + getObservabilityUpgradeRequiredMessage, + isObservabilityUpgradeRequiredError, +} from '../lib/inspect/errors.js'; import { cliFlags } from '../lib/inspect/flags.js'; import { setupCliWorld } from '../lib/inspect/setup.js'; @@ -23,7 +28,10 @@ export default class Cancel extends BaseCommand { ]; async catch(error: any) { - if (LOGGING_CONFIG.VERBOSE_MODE) { + if (isObservabilityUpgradeRequiredError(error)) { + logger.error(getObservabilityUpgradeRequiredMessage()); + process.exit(1); + } else if (LOGGING_CONFIG.VERBOSE_MODE) { console.error(error); } throw error; @@ -99,13 +107,30 @@ export default class Cancel extends BaseCommand { process.exit(1); } - // Fetch matching runs - const runs = await world.runs.list({ - status: flags.status as any, - workflowName: flags.workflowName, - pagination: { limit: flags.limit || 50 }, - resolveData: 'none', - }); + // Fetch matching runs. Only metadata is needed to display and cancel, so + // prefer the analytics read path when the backend provides one. + const status = flags.status as WorkflowRun['status'] | undefined; + const runList = world.analytics + ? await world.analytics.runs.list({ + status, + workflowName: flags.workflowName, + pagination: { limit: flags.limit || 50 }, + }) + : await world.runs.list({ + status, + workflowName: flags.workflowName, + pagination: { limit: flags.limit || 50 }, + resolveData: 'none', + }); + const runs = { + data: runList.data.map((run) => ({ + runId: run.runId, + workflowName: run.workflowName, + status: run.status, + startedAt: run.startedAt, + })), + hasMore: runList.hasMore, + }; if (runs.data.length === 0) { logger.warn('No matching runs found.'); diff --git a/packages/cli/src/commands/inspect.ts b/packages/cli/src/commands/inspect.ts index 40c8b9ffe4..34b8c81632 100644 --- a/packages/cli/src/commands/inspect.ts +++ b/packages/cli/src/commands/inspect.ts @@ -3,6 +3,10 @@ import { VERCEL_403_ERROR_MESSAGE } from '@workflow/errors'; import { BaseCommand } from '../base.js'; import { LOGGING_CONFIG, logger } from '../lib/config/log.js'; import type { InspectCLIOptions } from '../lib/config/types.js'; +import { + getObservabilityUpgradeRequiredMessage, + isObservabilityUpgradeRequiredError, +} from '../lib/inspect/errors.js'; import { cliFlags, urlFlag } from '../lib/inspect/flags.js'; import { listEvents, @@ -35,7 +39,9 @@ export default class Inspect extends BaseCommand { async catch(error: any) { // Check if this is a 403 error from the Vercel backend - if (error?.status === 403) { + if (isObservabilityUpgradeRequiredError(error)) { + logger.error(getObservabilityUpgradeRequiredMessage()); + } else if (error?.status === 403) { const message = VERCEL_403_ERROR_MESSAGE; logger.error(message); } else if (LOGGING_CONFIG.VERBOSE_MODE) { @@ -124,7 +130,8 @@ export default class Inspect extends BaseCommand { helpLabel: '--status', }), withData: Flags.boolean({ - description: 'include full input/output data in list views', + description: + 'include full input/output data in list views (deprecated for list views — use `inspect ` to view payloads)', required: false, char: 'd', default: false, diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index 3433db505d..be5d33105d 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -1,6 +1,10 @@ import { Args } from '@oclif/core'; import { BaseCommand } from '../base.js'; -import { LOGGING_CONFIG } from '../lib/config/log.js'; +import { LOGGING_CONFIG, logger } from '../lib/config/log.js'; +import { + getObservabilityUpgradeRequiredMessage, + isObservabilityUpgradeRequiredError, +} from '../lib/inspect/errors.js'; import { cliFlags } from '../lib/inspect/flags.js'; import { startRun } from '../lib/inspect/run.js'; import { setupCliWorld } from '../lib/inspect/setup.js'; @@ -16,7 +20,10 @@ export default class Start extends BaseCommand { ]; async catch(error: any) { - if (LOGGING_CONFIG.VERBOSE_MODE) { + if (isObservabilityUpgradeRequiredError(error)) { + logger.error(getObservabilityUpgradeRequiredMessage()); + process.exit(1); + } else if (LOGGING_CONFIG.VERBOSE_MODE) { console.error(error); } throw error; diff --git a/packages/cli/src/lib/inspect/errors.ts b/packages/cli/src/lib/inspect/errors.ts new file mode 100644 index 0000000000..920fc64bc6 --- /dev/null +++ b/packages/cli/src/lib/inspect/errors.ts @@ -0,0 +1,38 @@ +const OBSERVABILITY_UPGRADE_REQUIRED_CODE = 'observability-upgrade-required'; +const OBSERVABILITY_UPGRADE_REQUIRED_MESSAGE = + 'This workflow observability data is outside your current plan window. Upgrade Observability Plus to view up to 30 days of workflow data.'; + +const extractErrorCode = (err: Record): string | undefined => { + if (err.code && typeof err.code === 'string') { + return err.code; + } + + if (err.body && typeof err.body === 'object') { + const body = err.body as Record; + if (body.code && typeof body.code === 'string') { + return body.code; + } + if (body.error && typeof body.error === 'string') { + return body.error; + } + } + + return undefined; +}; + +export const isObservabilityUpgradeRequiredError = ( + error: unknown +): boolean => { + if (!error || typeof error !== 'object') { + return false; + } + + const err = error as Record; + return ( + err.status === 402 && + extractErrorCode(err) === OBSERVABILITY_UPGRADE_REQUIRED_CODE + ); +}; + +export const getObservabilityUpgradeRequiredMessage = (): string => + OBSERVABILITY_UPGRADE_REQUIRED_MESSAGE; diff --git a/packages/cli/src/lib/inspect/output.test.ts b/packages/cli/src/lib/inspect/output.test.ts index 191ac77105..ab8adecd9a 100644 --- a/packages/cli/src/lib/inspect/output.test.ts +++ b/packages/cli/src/lib/inspect/output.test.ts @@ -1,6 +1,26 @@ -import { describe, expect, it } from 'vitest'; -import type { WorkflowRun } from '@workflow/world'; -import { formatTableValue, hasExpiredData } from './output.js'; +import type { + AnalyticsEvent, + AnalyticsRun, + AnalyticsStep, + AnalyticsWait, + Event, + Step, + WorkflowRun, + World, +} from '@workflow/world'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + getObservabilityUpgradeRequiredMessage, + isObservabilityUpgradeRequiredError, +} from './errors.js'; +import { + formatTableValue, + hasExpiredData, + listEvents, + listRuns, + listSleeps, + listSteps, +} from './output.js'; const makeRun = (overrides: Partial = {}): WorkflowRun => ({ @@ -21,6 +41,10 @@ const makeRun = (overrides: Partial = {}): WorkflowRun => ...overrides, }) as unknown as WorkflowRun; +afterEach(() => { + vi.restoreAllMocks(); +}); + describe('hasExpiredData', () => { it('returns false when expiredAt is undefined', () => { expect(hasExpiredData(makeRun({ expiredAt: undefined }))).toBe(false); @@ -58,3 +82,435 @@ describe('formatTableValue expired data handling', () => { expect(String(result)).not.toContain('expired'); }); }); + +describe('isObservabilityUpgradeRequiredError', () => { + it('detects workflow analytics 402 errors by top-level code', () => { + expect( + isObservabilityUpgradeRequiredError({ + status: 402, + code: 'observability-upgrade-required', + }) + ).toBe(true); + }); + + it('detects workflow analytics 402 errors by response body error', () => { + expect( + isObservabilityUpgradeRequiredError({ + status: 402, + body: { error: 'observability-upgrade-required' }, + }) + ).toBe(true); + }); + + it('does not treat 404s as upgrade prompts', () => { + expect( + isObservabilityUpgradeRequiredError({ + status: 404, + code: 'observability-upgrade-required', + }) + ).toBe(false); + }); + + it('uses an upgrade prompt message', () => { + expect(getObservabilityUpgradeRequiredMessage()).toContain( + 'Upgrade Observability Plus' + ); + }); +}); + +describe('listRuns', () => { + it('preserves analytics page metadata in JSON output', async () => { + const run = { + runId: 'run-1', + status: 'running', + deploymentId: 'dep-1', + workflowName: 'workflow//./src/workflows/test//myWorkflow', + specVersion: 2, + attributes: {}, + createdAt: new Date('2026-06-30T00:00:00.000Z'), + updatedAt: new Date('2026-06-30T00:00:00.000Z'), + startedAt: new Date('2026-06-30T00:00:01.000Z'), + completedAt: null, + errorCode: null, + workflowCoreVersion: null, + workflowEncryptionEnabled: false, + } satisfies AnalyticsRun; + const pageInfo = { + currentLookbackDays: 2, + maxLookbackDays: 30, + currentWindowStart: new Date('2026-06-28T00:00:00.000Z'), + maxWindowStart: new Date('2026-06-01T00:00:00.000Z'), + upgradeAvailable: true, + }; + const world = { + analytics: { + runs: { + list: vi.fn().mockResolvedValue({ + data: [run], + cursor: null, + hasMore: false, + pageInfo, + }), + }, + }, + } as unknown as World; + const write = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true); + + await listRuns(world, { json: true }); + + expect(world.analytics?.runs.list).toHaveBeenCalledWith({ + workflowName: undefined, + status: undefined, + pagination: { + sortOrder: 'desc', + cursor: undefined, + limit: 20, + }, + }); + expect(write).toHaveBeenCalledTimes(1); + expect(JSON.parse(String(write.mock.calls[0][0]))).toEqual({ + data: [ + { + ...run, + createdAt: '2026-06-30T00:00:00.000Z', + updatedAt: '2026-06-30T00:00:00.000Z', + startedAt: '2026-06-30T00:00:01.000Z', + }, + ], + cursor: null, + hasMore: false, + pageInfo: { + currentLookbackDays: 2, + maxLookbackDays: 30, + currentWindowStart: '2026-06-28T00:00:00.000Z', + maxWindowStart: '2026-06-01T00:00:00.000Z', + upgradeAvailable: true, + }, + }); + }); +}); + +describe('listSteps', () => { + it('passes cursors and preserves the JSON array output', async () => { + const step = { + runId: 'run-1', + stepId: 'step-1', + stepName: 'doWork', + status: 'completed', + attempt: 1, + createdAt: new Date('2026-06-30T00:00:00.000Z'), + updatedAt: new Date('2026-06-30T00:00:02.000Z'), + startedAt: new Date('2026-06-30T00:00:01.000Z'), + completedAt: new Date('2026-06-30T00:00:02.000Z'), + retryAfter: null, + errorCode: null, + workflowCoreVersion: null, + workflowEncryptionEnabled: false, + } satisfies AnalyticsStep; + const world = { + analytics: { + steps: { + list: vi.fn().mockResolvedValue({ + data: [step], + cursor: 'next-step-cursor', + hasMore: true, + }), + }, + }, + } as unknown as World; + const write = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true); + + await listSteps(world, { + json: true, + runId: 'run-1', + cursor: 'step-cursor', + limit: 1, + }); + + expect(world.analytics?.steps.list).toHaveBeenCalledWith({ + runId: 'run-1', + pagination: { + sortOrder: 'desc', + cursor: 'step-cursor', + limit: 1, + }, + }); + expect(JSON.parse(String(write.mock.calls[0][0]))).toEqual([ + { + ...step, + createdAt: '2026-06-30T00:00:00.000Z', + updatedAt: '2026-06-30T00:00:02.000Z', + startedAt: '2026-06-30T00:00:01.000Z', + completedAt: '2026-06-30T00:00:02.000Z', + }, + ]); + }); + + it('falls back to storage when the first analytics page is empty', async () => { + const step = { + runId: 'run-1', + stepId: 'step-1', + stepName: 'step//./src/workflows/test//doWork', + status: 'completed', + attempt: 1, + input: undefined, + output: undefined, + createdAt: new Date('2026-06-30T00:00:00.000Z'), + updatedAt: new Date('2026-06-30T00:00:02.000Z'), + startedAt: new Date('2026-06-30T00:00:01.000Z'), + completedAt: new Date('2026-06-30T00:00:02.000Z'), + } satisfies Step; + const world = { + analytics: { + steps: { + list: vi.fn().mockResolvedValue({ + data: [], + cursor: null, + hasMore: false, + }), + }, + }, + steps: { + list: vi.fn().mockResolvedValue({ + data: [step], + cursor: null, + hasMore: false, + }), + }, + } as unknown as World; + const write = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true); + + await listSteps(world, { json: true, runId: 'run-1' }); + + expect(world.analytics?.steps.list).toHaveBeenCalled(); + expect(world.steps.list).toHaveBeenCalledWith({ + runId: 'run-1', + pagination: { + sortOrder: 'desc', + cursor: undefined, + limit: 20, + }, + resolveData: 'none', + }); + expect(JSON.parse(String(write.mock.calls[0][0]))).toEqual([ + { + ...step, + createdAt: '2026-06-30T00:00:00.000Z', + updatedAt: '2026-06-30T00:00:02.000Z', + startedAt: '2026-06-30T00:00:01.000Z', + completedAt: '2026-06-30T00:00:02.000Z', + }, + ]); + }); +}); + +describe('listEvents', () => { + it('passes cursors and preserves the JSON array output', async () => { + const event = { + runId: 'run-1', + eventId: 'event-1', + eventType: 'step_completed', + correlationId: 'step-1', + entityId: 'step-1', + stepName: 'doWork', + workflowName: 'workflow//./src/workflows/test//myWorkflow', + deploymentId: 'dep-1', + specVersion: 2, + runCreatedAt: new Date('2026-06-30T00:00:00.000Z'), + createdAt: new Date('2026-06-30T00:00:02.000Z'), + region: null, + vercelId: null, + requestId: null, + resumeAt: null, + retryAfter: null, + errorCode: null, + workflowCoreVersion: null, + isWebhook: false, + isSystem: false, + workflowEncryptionEnabled: false, + } satisfies AnalyticsEvent; + const world = { + analytics: { + events: { + list: vi.fn().mockResolvedValue({ + data: [event], + cursor: 'next-event-cursor', + hasMore: true, + }), + }, + }, + } as unknown as World; + const write = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true); + + await listEvents(world, { + json: true, + runId: 'run-1', + cursor: 'event-cursor', + limit: 1, + }); + + expect(world.analytics?.events.list).toHaveBeenCalledWith({ + runId: 'run-1', + correlationId: undefined, + pagination: { + sortOrder: 'desc', + cursor: 'event-cursor', + limit: 1, + }, + }); + expect(JSON.parse(String(write.mock.calls[0][0]))).toEqual([ + { + ...event, + runCreatedAt: '2026-06-30T00:00:00.000Z', + createdAt: '2026-06-30T00:00:02.000Z', + }, + ]); + }); + + it('falls back to storage when the first analytics page is empty', async () => { + const event = { + runId: 'run-1', + eventId: 'event-1', + eventType: 'step_completed', + correlationId: 'step-1', + eventData: { + stepName: 'doWork', + result: undefined, + }, + createdAt: new Date('2026-06-30T00:00:02.000Z'), + } as unknown as Event; + const world = { + analytics: { + events: { + list: vi.fn().mockResolvedValue({ + data: [], + cursor: null, + hasMore: false, + }), + }, + }, + events: { + list: vi.fn().mockResolvedValue({ + data: [event], + cursor: null, + hasMore: false, + }), + }, + } as unknown as World; + const write = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true); + + await listEvents(world, { json: true, runId: 'run-1' }); + + expect(world.analytics?.events.list).toHaveBeenCalled(); + expect(world.events.list).toHaveBeenCalledWith({ + runId: 'run-1', + pagination: { + sortOrder: 'desc', + cursor: undefined, + limit: 20, + }, + resolveData: 'none', + }); + expect(JSON.parse(String(write.mock.calls[0][0]))).toEqual([ + { + ...event, + createdAt: '2026-06-30T00:00:02.000Z', + }, + ]); + }); +}); + +describe('listSleeps', () => { + const wait = { + runId: 'run-1', + waitId: 'wait-1', + status: 'waiting', + resumeAt: new Date('2026-06-30T00:01:00.000Z'), + createdAt: new Date('2026-06-30T00:00:00.000Z'), + updatedAt: new Date('2026-06-30T00:00:00.000Z'), + completedAt: null, + workflowCoreVersion: null, + workflowEncryptionEnabled: false, + } satisfies AnalyticsWait; + const pageInfo = { + currentLookbackDays: 2, + maxLookbackDays: 30, + currentWindowStart: new Date('2026-06-28T00:00:00.000Z'), + maxWindowStart: new Date('2026-06-01T00:00:00.000Z'), + upgradeAvailable: true, + }; + + it('passes cursors and preserves the JSON array output through analytics', async () => { + const world = { + analytics: { + waits: { + list: vi.fn().mockResolvedValue({ + data: [wait], + cursor: 'next-wait-cursor', + hasMore: true, + pageInfo, + }), + }, + }, + } as unknown as World; + const write = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true); + + await listSleeps(world, { + json: true, + runId: 'run-1', + cursor: 'wait-cursor', + limit: 1, + }); + + expect(world.analytics?.waits.list).toHaveBeenCalledWith({ + runId: 'run-1', + pagination: { + sortOrder: 'desc', + cursor: 'wait-cursor', + limit: 1, + }, + }); + expect(JSON.parse(String(write.mock.calls[0][0]))).toEqual([ + { + ...wait, + resumeAt: '2026-06-30T00:01:00.000Z', + createdAt: '2026-06-30T00:00:00.000Z', + updatedAt: '2026-06-30T00:00:00.000Z', + }, + ]); + }); + + it('surfaces the observability upgrade hint in table mode', async () => { + const world = { + analytics: { + waits: { + list: vi.fn().mockResolvedValue({ + data: [wait], + cursor: null, + hasMore: false, + pageInfo, + }), + }, + }, + } as unknown as World; + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); + + await listSleeps(world, { runId: 'run-1' }); + + expect(log.mock.calls.flat().join('\n')).toContain( + 'Upgrade Observability Plus' + ); + }); +}); diff --git a/packages/cli/src/lib/inspect/output.ts b/packages/cli/src/lib/inspect/output.ts index 8202a6f264..c5d9a6da38 100644 --- a/packages/cli/src/lib/inspect/output.ts +++ b/packages/cli/src/lib/inspect/output.ts @@ -6,23 +6,17 @@ import { } from '@workflow/core/serialization'; import { VERCEL_403_ERROR_MESSAGE } from '@workflow/errors'; import { parseStepName, parseWorkflowName } from '@workflow/utils/parse-name'; -import type { - Event, - Hook, - ListEventsParams, - PaginationOptions, - Step, - StepWithoutData, - WorkflowRun, - WorkflowRunWithoutData, - World, -} from '@workflow/world'; +import type { Event, Hook, Step, WorkflowRun, World } from '@workflow/world'; import chalk from 'chalk'; import { formatDistance } from 'date-fns'; import Table from 'easy-table'; import { logger } from '../config/log.js'; import type { InspectCLIOptions } from '../config/types.js'; +import { + getObservabilityUpgradeRequiredMessage, + isObservabilityUpgradeRequiredError, +} from './errors.js'; import { type EncryptionKeyResolver, hydrateResourceIO, @@ -53,7 +47,11 @@ function createResolver(world: World, decrypt: boolean): EncryptionKeyResolver { }; } -import { setupListPagination } from './pagination.js'; +import { + type AnalyticsPageInfo, + type PageData, + setupListPagination, +} from './pagination.js'; import { streamToConsole } from './stream.js'; import { formatISODate, @@ -126,15 +124,13 @@ const WAIT_LISTED_PROPS: (keyof Sleep)[] = [ 'completedAt', ]; -const STATUS_COLORS: Record< - WorkflowRun['status'] | Step['status'], - (value: string) => string -> = { +const STATUS_COLORS: Record string> = { running: chalk.blue, completed: chalk.green, failed: chalk.red, cancelled: chalk.strikethrough.yellow, pending: chalk.blue, + waiting: chalk.blue, }; const isStreamId = (value: string) => { @@ -143,12 +139,13 @@ const isStreamId = (value: string) => { const showStatusLegend = () => { logger.log('\nStatus Legend:'); - const statuses: Array = [ + const statuses = [ 'running', 'completed', 'failed', 'cancelled', 'pending', + 'waiting', ]; const legendItems = statuses.map((status) => { @@ -199,12 +196,28 @@ const extractErrorMessage = ( return undefined; }; +const getPageInfo = (result: unknown): AnalyticsPageInfo | undefined => { + if ( + typeof result !== 'object' || + result === null || + !('pageInfo' in result) + ) { + return undefined; + } + return (result as { pageInfo?: AnalyticsPageInfo }).pageInfo; +}; + const handleApiError = (error: unknown, backend?: string): boolean => { // First check for Vercel access errors if (checkAndHandleVercelAccessError(error, backend)) { return true; } + if (isObservabilityUpgradeRequiredError(error)) { + logger.error(getObservabilityUpgradeRequiredMessage()); + return true; + } + // Handle other HTTP errors if (error && typeof error === 'object') { const err = error as Record; @@ -237,6 +250,7 @@ const getStatusText = (status: number): string => { const statusTexts: Record = { 400: 'Bad Request', 401: 'Unauthorized', + 402: 'Payment Required', 403: 'Forbidden', 404: 'Not Found', 429: 'Too Many Requests', @@ -331,7 +345,8 @@ export const formatTableValue = ( if (prop === 'status') { const status = value as WorkflowRun['status'] | Step['status']; - const colorFunc = STATUS_COLORS[status]; + const colorFunc = + STATUS_COLORS[status] ?? ((formatted: string) => formatted); const formattedStatus = displaySettings?.abbreviateStatus ? formatStatusAbbrev(status, true) : status; @@ -436,6 +451,15 @@ const showJson = (data: unknown) => { process.stdout.write(`${json}\n`); }; +const showJsonPage = (page: PageData) => { + showJson({ + data: page.data, + cursor: page.cursor, + hasMore: page.hasMore, + ...(page.pageInfo ? { pageInfo: page.pageInfo } : {}), + }); +}; + /** * In tables, we want to show a shorter timestamp, YYYY-MM-DD HH:MM:SS */ @@ -563,6 +587,22 @@ const inlineFormatIO = (io: T, topLevel: boolean = true): string => { return value; }; +/** + * List views surface metadata only. When the active backend exposes the + * optional `world.analytics` read namespace we read list pages from it + * (metadata-only — no payload resolution); otherwise we fall back to the + * runtime storage APIs with `resolveData: 'none'`. + * + * `--withData` forces the runtime path so payloads can be resolved into the + * table. It is deprecated for list views: view payloads per-resource with + * `workflow inspect `. + */ +const warnWithDataDeprecatedForList = (singular: string): void => { + logger.warn( + `'--withData' is deprecated for list views and will be removed in a future release. Use 'workflow inspect ${singular} ' to view payloads for a single ${singular}.` + ); +}; + export const listRuns = async (world: World, opts: InspectCLIOptions = {}) => { const resolveKey = createResolver(world, opts?.decrypt ?? false); @@ -572,7 +612,13 @@ export const listRuns = async (world: World, opts: InspectCLIOptions = {}) => { ); } + if (opts.withData) { + warnWithDataDeprecatedForList('run'); + } + + const useAnalytics = !opts.withData && Boolean(world.analytics); const resolveData = opts.withData ? 'all' : 'none'; + const status = opts.status as WorkflowRun['status'] | undefined; // Determine which props to show based on withData flag const props = opts.withData @@ -581,23 +627,49 @@ export const listRuns = async (world: World, opts: InspectCLIOptions = {}) => { (prop) => !WORKFLOW_RUN_IO_PROPS.includes(prop) ); + const fetchRunsPage = async ( + cursor: string | undefined + ): Promise>> => { + const pagination = { + sortOrder: opts.sort || 'desc', + cursor, + limit: opts.limit || DEFAULT_PAGE_SIZE, + }; + if (useAnalytics && world.analytics) { + const runs = await world.analytics.runs.list({ + workflowName: opts.workflowName, + status, + pagination, + }); + return { + data: runs.data as unknown as Record[], + cursor: runs.cursor, + hasMore: runs.hasMore, + pageInfo: getPageInfo(runs), + }; + } + const runs = await world.runs.list({ + workflowName: opts.workflowName, + status, + pagination, + resolveData, + }); + const data = await Promise.all( + runs.data.map((r) => hydrateResourceIO(r, resolveKey)) + ); + return { + data: data as unknown as Record[], + cursor: runs.cursor, + hasMore: runs.hasMore, + pageInfo: getPageInfo(runs), + }; + }; + // For JSON output, just fetch once and return if (opts.json) { try { - const runs = await world.runs.list({ - workflowName: opts.workflowName, - status: opts.status as any, - pagination: { - sortOrder: opts.sort || 'desc', - cursor: opts.cursor, - limit: opts.limit || DEFAULT_PAGE_SIZE, - }, - resolveData, - }); - const runsWithHydratedIO = await Promise.all( - runs.data.map((r) => hydrateResourceIO(r, resolveKey)) - ); - showJson({ ...runs, data: runsWithHydratedIO }); + const page = await fetchRunsPage(opts.cursor); + showJsonPage(page); return; } catch (error) { if (handleApiError(error, opts.backend)) { @@ -607,26 +679,12 @@ export const listRuns = async (world: World, opts: InspectCLIOptions = {}) => { } } - await setupListPagination({ + await setupListPagination>({ initialCursor: opts.cursor, interactive: opts.interactive, fetchPage: async (cursor) => { try { - const runs = await world.runs.list({ - workflowName: opts.workflowName, - status: opts.status as any, - pagination: { - sortOrder: opts.sort || 'desc', - cursor, - limit: opts.limit || DEFAULT_PAGE_SIZE, - }, - resolveData, - }); - return { - data: runs.data, - cursor: runs.cursor, - hasMore: runs.hasMore, - }; + return await fetchRunsPage(cursor); } catch (error) { if (handleApiError(error, opts.backend)) { process.exit(1); @@ -635,10 +693,7 @@ export const listRuns = async (world: World, opts: InspectCLIOptions = {}) => { } }, displayPage: async (runs) => { - const runsWithHydratedIO = await Promise.all( - runs.map((r) => hydrateResourceIO(r, resolveKey)) - ); - logger.log(showTable(runsWithHydratedIO, props, opts)); + logger.log(showTable(runs, props, opts)); }, }); }; @@ -726,6 +781,11 @@ export const listSteps = async ( return; } + if (opts.withData) { + warnWithDataDeprecatedForList('step'); + } + + const useAnalytics = !opts.withData && Boolean(world.analytics); const resolveData = opts.withData ? 'all' : 'none'; // Determine which props to show based on withData flag @@ -733,23 +793,51 @@ export const listSteps = async ( ? STEP_LISTED_PROPS : STEP_LISTED_PROPS.filter((prop) => !STEP_IO_PROPS.includes(prop)); + const fetchStepsPage = async ( + cursor: string | undefined + ): Promise>> => { + logger.debug(`Fetching steps for run ${runId}`); + const pagination = { + sortOrder: opts.sort || 'desc', + cursor, + limit: opts.limit || DEFAULT_PAGE_SIZE, + }; + if (useAnalytics && world.analytics) { + const steps = await world.analytics.steps.list({ runId, pagination }); + const page = { + data: steps.data as unknown as Record[], + cursor: steps.cursor, + hasMore: steps.hasMore, + pageInfo: getPageInfo(steps), + }; + if (cursor || page.data.length > 0 || page.hasMore) { + return page; + } + logger.debug( + `No analytics steps found for run ${runId}; falling back to storage` + ); + } + const stepChunks = await world.steps.list({ + runId, + pagination, + resolveData, + }); + const data = await Promise.all( + stepChunks.data.map((s) => hydrateResourceIO(s, resolveKey)) + ); + return { + data: data as unknown as Record[], + cursor: stepChunks.cursor, + hasMore: stepChunks.hasMore, + pageInfo: getPageInfo(stepChunks), + }; + }; + // For JSON output, just fetch once and return if (opts.json) { - logger.debug(`Fetching steps for run ${runId}`); try { - const stepChunks = await world.steps.list({ - runId, - pagination: { - sortOrder: opts.sort || 'desc', - cursor: opts.cursor, - limit: opts.limit || DEFAULT_PAGE_SIZE, - }, - resolveData, - }); - const stepsWithHydratedIO = await Promise.all( - stepChunks.data.map((s) => hydrateResourceIO(s, resolveKey)) - ); - showJson(stepsWithHydratedIO); + const page = await fetchStepsPage(opts.cursor); + showJson(page.data); return; } catch (error) { if (handleApiError(error, opts.backend)) { @@ -759,26 +847,12 @@ export const listSteps = async ( } } - await setupListPagination({ + await setupListPagination>({ initialCursor: opts.cursor, interactive: opts.interactive, fetchPage: async (cursor) => { - logger.debug(`Fetching steps for run ${runId}`); try { - const stepChunks = await world.steps.list({ - runId, - pagination: { - sortOrder: opts.sort || 'desc', - cursor, - limit: opts.limit || DEFAULT_PAGE_SIZE, - }, - resolveData, - }); - return { - data: stepChunks.data, - cursor: stepChunks.cursor, - hasMore: stepChunks.hasMore, - }; + return await fetchStepsPage(cursor); } catch (error) { if (handleApiError(error, opts.backend)) { process.exit(1); @@ -787,10 +861,7 @@ export const listSteps = async ( } }, displayPage: async (steps) => { - const stepsWithHydratedIO = await Promise.all( - steps.map((s) => hydrateResourceIO(s, resolveKey)) - ); - logger.log(showTable(stepsWithHydratedIO, props, opts)); + logger.log(showTable(steps, props, opts)); showInspectInfoBox('step'); }, }); @@ -969,46 +1040,70 @@ export const listEvents = async ( return; } + if (opts.withData) { + warnWithDataDeprecatedForList('event'); + } + + const useAnalytics = !opts.withData && Boolean(world.analytics); const correlationIdFilter = opts.hookId || opts.stepId; - const params: ListEventsParams = { - runId, - pagination: { - sortOrder: opts.sort || 'desc', - cursor: opts.cursor, - limit: opts.limit || DEFAULT_PAGE_SIZE, - }, - resolveData: opts.withData ? 'all' : 'none', - }; - const listCall = async (pagination: PaginationOptions) => { - const result = await world.events.list({ - ...params, - pagination: { ...params.pagination, ...pagination }, - }); - if (correlationIdFilter) { - return { - ...result, - data: result.data.filter( - (e) => e.correlationId === correlationIdFilter - ), - }; - } - return result; - }; // Determine which props to show based on withData flag const props = opts.withData ? EVENT_LISTED_PROPS : EVENT_LISTED_PROPS.filter((prop) => !EVENT_IO_PROPS.includes(prop)); + const fetchEventsPage = async ( + cursor: string | undefined + ): Promise>> => { + logger.debug(`Fetching events for run ${runId}`); + const pagination = { + sortOrder: opts.sort || 'desc', + cursor, + limit: opts.limit || DEFAULT_PAGE_SIZE, + }; + if (useAnalytics && world.analytics) { + const events = await world.analytics.events.list({ + runId, + correlationId: correlationIdFilter, + pagination, + }); + const page = { + data: events.data as unknown as Record[], + cursor: events.cursor, + hasMore: events.hasMore, + pageInfo: getPageInfo(events), + }; + if (cursor || page.data.length > 0 || page.hasMore) { + return page; + } + logger.debug( + `No analytics events found for run ${runId}; falling back to storage` + ); + } + const result = await world.events.list({ + runId, + pagination, + resolveData: opts.withData ? 'all' : 'none', + }); + const filtered = correlationIdFilter + ? result.data.filter((e) => e.correlationId === correlationIdFilter) + : result.data; + const data = await Promise.all( + filtered.map((e) => hydrateResourceIO(e, resolveKey)) + ); + return { + data: data as unknown as Record[], + cursor: result.cursor, + hasMore: result.hasMore, + pageInfo: getPageInfo(result), + }; + }; + // For JSON output, just fetch once and return if (opts.json) { - logger.debug(`Fetching events for run ${runId}`); try { - const events = await listCall({}); - const hydratedEvents = await Promise.all( - events.data.map((e) => hydrateResourceIO(e, resolveKey)) - ); - showJson(hydratedEvents); + const page = await fetchEventsPage(opts.cursor); + showJson(page.data); return; } catch (error) { if (handleApiError(error, opts.backend)) { @@ -1018,18 +1113,12 @@ export const listEvents = async ( } } - await setupListPagination({ + await setupListPagination>({ initialCursor: opts.cursor, interactive: opts.interactive, fetchPage: async (cursor) => { - logger.debug(`Fetching events for run ${runId}`); try { - const events = await listCall({ cursor }); - return { - data: events.data, - cursor: events.cursor, - hasMore: events.hasMore, - }; + return await fetchEventsPage(cursor); } catch (error) { if (handleApiError(error, opts.backend)) { process.exit(1); @@ -1038,10 +1127,7 @@ export const listEvents = async ( } }, displayPage: async (events) => { - const hydratedEvents = await Promise.all( - events.map((e) => hydrateResourceIO(e, resolveKey)) - ); - logger.log(showTable(hydratedEvents, props, opts)); + logger.log(showTable(events, props, opts)); showInspectInfoBox('event'); }, }); @@ -1063,28 +1149,39 @@ export const listHooks = async (world: World, opts: InspectCLIOptions = {}) => { const runId = opts.runId; const resolveData = opts.withData ? 'all' : 'none'; - - // For JSON output, just fetch once and return - if (opts.json) { + // Hooks always read from the runtime storage API. Unlike runs/steps/events, + // the analytics read path omits `ownerId` (and the secret hook token used by + // resume/copy-token surfaces), so hook listing stays on the canonical APIs. + const fetchHooksPage = async ( + cursor: string | undefined + ): Promise>> => { if (!runId) { logger.debug('Fetching all hooks'); } else { logger.debug(`Fetching hooks for run ${runId}`); } + const pagination = { + sortOrder: opts.sort || 'desc', + cursor, + limit: opts.limit || DEFAULT_PAGE_SIZE, + }; + const hooks = await world.hooks.list({ runId, pagination, resolveData }); + const data = await Promise.all( + hooks.data.map((h) => hydrateResourceIO(h, resolveKey)) + ); + return { + data: data as unknown as Record[], + cursor: hooks.cursor, + hasMore: hooks.hasMore, + pageInfo: getPageInfo(hooks), + }; + }; + + // For JSON output, just fetch once and return + if (opts.json) { try { - const hooks = await world.hooks.list({ - runId, - pagination: { - sortOrder: opts.sort || 'desc', - cursor: opts.cursor, - limit: opts.limit || DEFAULT_PAGE_SIZE, - }, - resolveData, - }); - const hydratedHooks = await Promise.all( - hooks.data.map((h) => hydrateResourceIO(h, resolveKey)) - ); - showJson({ ...hooks, data: hydratedHooks }); + const page = await fetchHooksPage(opts.cursor); + showJsonPage(page); return; } catch (error) { if (handleApiError(error, opts.backend)) { @@ -1095,30 +1192,12 @@ export const listHooks = async (world: World, opts: InspectCLIOptions = {}) => { } // Setup pagination with new mechanism - await setupListPagination({ + await setupListPagination>({ initialCursor: opts.cursor, interactive: opts.interactive, fetchPage: async (cursor) => { - if (!runId) { - logger.debug('Fetching all hooks'); - } else { - logger.debug(`Fetching hooks for run ${runId}`); - } try { - const hooks = await world.hooks.list({ - runId, - pagination: { - sortOrder: opts.sort || 'desc', - cursor, - limit: opts.limit || DEFAULT_PAGE_SIZE, - }, - resolveData, - }); - return { - data: hooks.data, - cursor: hooks.cursor, - hasMore: hooks.hasMore, - }; + return await fetchHooksPage(cursor); } catch (error) { if (handleApiError(error, opts.backend)) { process.exit(1); @@ -1127,10 +1206,7 @@ export const listHooks = async (world: World, opts: InspectCLIOptions = {}) => { } }, displayPage: async (hooks) => { - const hydratedHooks = await Promise.all( - hooks.map((h) => hydrateResourceIO(h, resolveKey)) - ); - logger.log(showTable(hydratedHooks, HOOK_LISTED_PROPS, opts)); + logger.log(showTable(hooks, HOOK_LISTED_PROPS, opts)); showInspectInfoBox('hook'); }, }); @@ -1172,6 +1248,57 @@ export const showHook = async ( } }; +/** + * List waits/sleeps via the analytics read path. The analytics wait shape is + * keyed by `waitId`/`status` rather than the `correlationId`/`eventId` columns + * reconstructed from the event stream in the fallback path. + */ +const listSleepsViaAnalytics = async ( + analytics: NonNullable, + opts: InspectCLIOptions +): Promise => { + const fetchSleepsPage = async ( + cursor: string | undefined + ): Promise>> => { + const waits = await analytics.waits.list({ + runId: opts.runId as string, + pagination: { + sortOrder: opts.sort || 'desc', + cursor, + limit: opts.limit || DEFAULT_PAGE_SIZE, + }, + }); + + return { + data: waits.data as unknown as Record[], + cursor: waits.cursor, + hasMore: waits.hasMore, + pageInfo: getPageInfo(waits), + }; + }; + + if (opts.json) { + const page = await fetchSleepsPage(opts.cursor); + showJson(page.data); + return; + } + + await setupListPagination>({ + initialCursor: opts.cursor, + interactive: opts.interactive, + fetchPage: fetchSleepsPage, + displayPage: async (waits) => { + logger.log( + showTable( + waits, + ['waitId', 'status', 'createdAt', 'resumeAt', 'completedAt'], + { ...opts, disableRelativeDates: true } + ) + ); + }, + }); +}; + export const listSleeps = async ( world: World, opts: InspectCLIOptions = {} @@ -1197,6 +1324,19 @@ export const listSleeps = async ( logger.warn('`withData` flag is ignored when listing sleeps'); } + // Prefer the analytics read path for wait/sleep listing when available. + if (world.analytics) { + try { + await listSleepsViaAnalytics(world.analytics, opts); + return; + } catch (error) { + if (handleApiError(error, opts.backend)) { + process.exit(1); + } + throw error; + } + } + try { // Fetch all events for the run with resolveData='all' to get wait eventData const events = await world.events.list({ diff --git a/packages/cli/src/lib/inspect/pagination.ts b/packages/cli/src/lib/inspect/pagination.ts index 89c04299b6..c24f17f26a 100644 --- a/packages/cli/src/lib/inspect/pagination.ts +++ b/packages/cli/src/lib/inspect/pagination.ts @@ -118,8 +118,39 @@ export interface PageData { data: T[]; cursor: string | null | undefined; hasMore: boolean; + pageInfo?: AnalyticsPageInfo; } +export interface AnalyticsPageInfo { + currentLookbackDays: number; + maxLookbackDays: number; + currentWindowStart: Date | string; + maxWindowStart: Date | string; + upgradeAvailable: boolean; +} + +export const getPageInfoUpgradeHint = ( + pageInfo: AnalyticsPageInfo | undefined +): string | undefined => { + if ( + !pageInfo?.upgradeAvailable || + pageInfo.currentLookbackDays >= pageInfo.maxLookbackDays + ) { + return undefined; + } + + return `Showing the last ${pageInfo.currentLookbackDays} days of workflow observability data. Upgrade Observability Plus to query up to ${pageInfo.maxLookbackDays} days.`; +}; + +const showPageInfoUpgradeHint = ( + pageInfo: AnalyticsPageInfo | undefined +): void => { + const hint = getPageInfoUpgradeHint(pageInfo); + if (hint) { + logger.info(hint); + } +}; + /** * Generic pagination handler for list operations */ @@ -137,7 +168,11 @@ export interface ListPaginationOptions { /** * Display function that renders the current page */ - displayPage: (data: TData[], pageIndex: number) => void | Promise; + displayPage: ( + data: TData[], + pageIndex: number, + page: PageData + ) => void | Promise; /** * Optional callback for when fetching starts @@ -185,7 +220,8 @@ export async function setupListPagination( console.clear(); } - await displayPage(page.data, index); + await displayPage(page.data, index, page); + showPageInfoUpgradeHint(page.pageInfo); return page; }; @@ -227,7 +263,8 @@ export async function setupListPagination( console.clear(); } - await displayPage(cachedPage.data, pageIndex); + await displayPage(cachedPage.data, pageIndex, cachedPage); + showPageInfoUpgradeHint(cachedPage.pageInfo); paginationState.cursor = cachedPage.cursor; paginationState.hasMore = cachedPage.hasMore; @@ -276,7 +313,8 @@ export async function setupListPagination( console.clear(); } - await displayPage(prevPage.data, pageIndex); + await displayPage(prevPage.data, pageIndex, prevPage); + showPageInfoUpgradeHint(prevPage.pageInfo); paginationState.cursor = prevPage.cursor; paginationState.hasMore = prevPage.hasMore; diff --git a/packages/cli/src/lib/inspect/run.ts b/packages/cli/src/lib/inspect/run.ts index 335bb9544d..e1f8cc893f 100644 --- a/packages/cli/src/lib/inspect/run.ts +++ b/packages/cli/src/lib/inspect/run.ts @@ -1,6 +1,6 @@ import { start } from '@workflow/core/runtime'; import { healthCheck } from '@workflow/core/runtime/helpers'; -import type { WorkflowRun, World } from '@workflow/world'; +import type { World } from '@workflow/world'; import { logger } from '../config/log.js'; interface CLICreateOpts { @@ -34,20 +34,28 @@ export const startRun = async ( } }); - let run: WorkflowRun | undefined; + // Only deployment/spec metadata is needed to start a new run, so this + // accepts either a full run record or an analytics (metadata-only) row. + let run: + | { deploymentId: string; specVersion?: number; workflowName: string } + | undefined; // If the workflowNameOrRunId is a run ID, get the run if (workflowNameOrRunId.startsWith('wrun_')) { run = await world.runs.get(workflowNameOrRunId); } else { // Get the first run for that name, hopefully the newest deployment, - // but can't guarantee that. - const runList = await world.runs.list({ - workflowName: workflowNameOrRunId, - pagination: { - sortOrder: 'desc', - limit: 1, - }, - }); + // but can't guarantee that. This is metadata only, so prefer the + // analytics read path when the backend provides one. + const runList = world.analytics + ? await world.analytics.runs.list({ + workflowName: workflowNameOrRunId, + pagination: { sortOrder: 'desc', limit: 1 }, + }) + : await world.runs.list({ + workflowName: workflowNameOrRunId, + pagination: { sortOrder: 'desc', limit: 1 }, + resolveData: 'none', + }); run = runList.data[0]; } From 1518c4860822986436711ec14b1b5b16192fefb5 Mon Sep 17 00:00:00 2001 From: Karthik Kalyan <105607645+karthikscale3@users.noreply.github.com> Date: Tue, 7 Jul 2026 11:45:07 -0700 Subject: [PATCH 2/8] web: read observability list views from world.analytics when available (#2647) * Add workflow analytics world APIs * web: read observability list views from world.analytics when available Route the runs/steps/events/hooks list server actions through the optional world.analytics namespace when the backend provides one, falling back to the runtime storage APIs otherwise. Events listing only uses the analytics path when no payload data is requested. Detail/get actions, streams, and mutations are unchanged. * web: keep events and hooks list reads on the runtime storage API The Events tab and trace viewer derive step names and wait resumeAt from resolved event payloads, and the hooks table needs the secret token and ownerId for its resume/copy-token actions. The metadata-only analytics rows do not carry these, so only the runs and steps list views use world.analytics. Co-Authored-By: Claude Opus 4.8 * web: read events list from world.analytics with a runtime-shape remap The events list/trace consumers read only top-level eventType, correlationId, and createdAt; event payloads are loaded lazily per event via fetchEvent(..., 'all') on the runtime path. Map the flat analytics event rows into the runtime Event shape (reconstructing eventData.stepName) so fetchEvents and fetchEventsByCorrelationId can use the analytics read path when available. Hooks remain on the runtime path (they need the secret token + ownerId). Co-Authored-By: Claude Opus 4.8 * Handle analytics access metadata in web --------- Co-authored-by: Claude Opus 4.8 --- .changeset/web-analytics-list-reads.md | 9 ++ packages/web/app/components/hooks-table.tsx | 15 +- .../web/app/components/run-detail-view.tsx | 8 +- packages/web/app/components/runs-table.tsx | 9 +- .../client/hooks/use-paginated-list.test.ts | 31 +++- .../lib/client/hooks/use-paginated-list.ts | 18 ++- .../app/lib/client/workflow-errors.test.ts | 30 +++- .../web/app/lib/client/workflow-errors.ts | 27 ++++ packages/web/app/lib/types.ts | 9 ++ packages/web/app/lib/workflow-api-client.ts | 1 + .../workflow-server-actions.server.test.ts | 71 +++++++++ .../server/workflow-server-actions.server.ts | 150 ++++++++++++++++-- 12 files changed, 346 insertions(+), 32 deletions(-) create mode 100644 .changeset/web-analytics-list-reads.md create mode 100644 packages/web/app/server/workflow-server-actions.server.test.ts diff --git a/.changeset/web-analytics-list-reads.md b/.changeset/web-analytics-list-reads.md new file mode 100644 index 0000000000..748e70cac0 --- /dev/null +++ b/.changeset/web-analytics-list-reads.md @@ -0,0 +1,9 @@ +--- +'@workflow/web': patch +--- + +The runs, steps, and events observability list views now read from the +metadata-only `world.analytics` namespace when the configured backend provides +one, and fall back to the runtime storage APIs otherwise. Event payloads are +still loaded lazily per event on the runtime path. Hooks listing, detail views, +payload resolution, streams, and mutations are unchanged. diff --git a/packages/web/app/components/hooks-table.tsx b/packages/web/app/components/hooks-table.tsx index b8dcd36fa3..069689b57e 100644 --- a/packages/web/app/components/hooks-table.tsx +++ b/packages/web/app/components/hooks-table.tsx @@ -3,7 +3,7 @@ import { ResolveHookDropdownItem, useHookActions, } from '@workflow/web-shared'; -import type { Event, Hook } from '@workflow/world'; +import type { Hook } from '@workflow/world'; import { AlertCircle, ChevronLeft, @@ -35,16 +35,17 @@ import { TooltipContent, TooltipTrigger, } from '~/components/ui/tooltip'; -import { CopyableText } from './display-utils/copyable-text'; -import { RelativeTime } from './display-utils/relative-time'; -import { TableSkeleton } from './display-utils/table-skeleton'; +import { fetchEvents } from '~/lib/rpc-client'; +import type { EnvMap } from '~/lib/types'; import { getErrorMessage, + getErrorTitle, resumeHook, useWorkflowHooks, } from '~/lib/workflow-api-client'; -import type { EnvMap } from '~/lib/types'; -import { fetchEvents } from '~/lib/rpc-client'; +import { CopyableText } from './display-utils/copyable-text'; +import { RelativeTime } from './display-utils/relative-time'; +import { TableSkeleton } from './display-utils/table-skeleton'; interface HooksTableProps { runId?: string; @@ -272,7 +273,7 @@ export function HooksTable({ {error ? ( - Error loading hooks + {getErrorTitle(error, 'Error loading hooks')} {getErrorMessage(error)} ) : !loading && (!hooks || hooks.length === 0) ? ( diff --git a/packages/web/app/components/run-detail-view.tsx b/packages/web/app/components/run-detail-view.tsx index d9094b6b34..06f603f9e8 100644 --- a/packages/web/app/components/run-detail-view.tsx +++ b/packages/web/app/components/run-detail-view.tsx @@ -56,6 +56,8 @@ import type { EnvMap } from '~/lib/types'; import { cancelRun, fetchSpanDetailResource, + getErrorMessage, + getErrorTitle, recreateRun, resumeHook, unwrapServerActionResult, @@ -507,8 +509,10 @@ export function RunDetailView({ return ( - Error loading workflow run - {error.message} + + {getErrorTitle(error, 'Error loading workflow run')} + + {getErrorMessage(error)} ); } diff --git a/packages/web/app/components/runs-table.tsx b/packages/web/app/components/runs-table.tsx index e56c96e7ab..4670c01fc5 100644 --- a/packages/web/app/components/runs-table.tsx +++ b/packages/web/app/components/runs-table.tsx @@ -44,15 +44,16 @@ import { TooltipTrigger, } from '~/components/ui/tooltip'; import { useTableSelection } from '~/lib/hooks/use-table-selection'; +import { fetchEvents, fetchRun } from '~/lib/rpc-client'; +import type { EnvMap } from '~/lib/types'; import { cancelRun, getErrorMessage, + getErrorTitle, reenqueueRun, useWorkflowRuns, } from '~/lib/workflow-api-client'; import { useServerConfig } from '~/lib/world-config-context'; -import type { EnvMap } from '~/lib/types'; -import { fetchEvents, fetchRun } from '~/lib/rpc-client'; import { CopyableText } from './display-utils/copyable-text'; import { RelativeTime } from './display-utils/relative-time'; import { SelectionBar } from './display-utils/selection-bar'; @@ -646,7 +647,9 @@ export function RunsTable({ onRunClick }: RunsTableProps) {
- Error loading runs + + {getErrorTitle(error, 'Error loading runs')} + {getErrorMessage(error)} diff --git a/packages/web/app/lib/client/hooks/use-paginated-list.test.ts b/packages/web/app/lib/client/hooks/use-paginated-list.test.ts index 346404cb18..2ef23d8ea4 100644 --- a/packages/web/app/lib/client/hooks/use-paginated-list.test.ts +++ b/packages/web/app/lib/client/hooks/use-paginated-list.test.ts @@ -13,6 +13,7 @@ vi.mock('~/lib/rpc-client', () => ({ import type { Hook, WorkflowRun } from '@workflow/world'; import { fetchHooks, fetchRuns } from '~/lib/rpc-client'; +import type { AnalyticsPageInfo } from '~/lib/types'; // ─── Fixtures ────────────────────────────────────────────────────────────── @@ -45,12 +46,28 @@ const HOOK: Hook = { environment: 'development', }; +const PAGE_INFO: AnalyticsPageInfo = { + currentLookbackDays: 2, + maxLookbackDays: 30, + currentWindowStart: new Date('2026-06-28T00:00:00.000Z'), + maxWindowStart: new Date('2026-06-01T00:00:00.000Z'), + upgradeAvailable: true, +}; + /** Resolved PaginatedResult for usePaginatedList's fetchFn */ -function page(data: T[], opts: { cursor?: string; hasMore?: boolean } = {}) { +function page( + data: T[], + opts: { + cursor?: string; + hasMore?: boolean; + pageInfo?: AnalyticsPageInfo; + } = {} +) { return Promise.resolve({ data, cursor: opts.cursor, hasMore: opts.hasMore ?? false, + pageInfo: opts.pageInfo, }); } @@ -90,6 +107,18 @@ describe('usePaginatedList', () => { expect(result.current.error).toBeNull(); }); + it('preserves analytics page metadata from the current page', async () => { + const fetchFn = vi + .fn() + .mockReturnValue(page(['a'], { pageInfo: PAGE_INFO })); + const { result } = renderHook(() => usePaginatedList(fetchFn)); + + await waitFor(() => expect(result.current.data.isLoading).toBe(false)); + + expect(result.current.analyticsPageInfo).toEqual(PAGE_INFO); + expect(result.current.data.pageInfo).toEqual(PAGE_INFO); + }); + it('shows error when fetch throws', async () => { const fetchFn = vi.fn().mockRejectedValue(new Error('network error')); const { result } = renderHook(() => usePaginatedList(fetchFn)); diff --git a/packages/web/app/lib/client/hooks/use-paginated-list.ts b/packages/web/app/lib/client/hooks/use-paginated-list.ts index dc3be59e11..f5e05bf168 100644 --- a/packages/web/app/lib/client/hooks/use-paginated-list.ts +++ b/packages/web/app/lib/client/hooks/use-paginated-list.ts @@ -5,13 +5,14 @@ import { WorkflowWebAPIError, } from '~/lib/client/workflow-errors'; import { fetchHooks, fetchRuns } from '~/lib/rpc-client'; -import type { EnvMap, PaginatedResult } from '~/lib/types'; +import type { AnalyticsPageInfo, EnvMap, PaginatedResult } from '~/lib/types'; import { getPaginationDisplay } from '~/lib/utils'; export interface PageResult { data: T[] | null; isLoading: boolean; error: Error | null; + pageInfo?: AnalyticsPageInfo; } export interface PaginatedList { @@ -28,6 +29,7 @@ export interface PaginatedList { reload: () => void; refresh: () => void; pageInfo: string; + analyticsPageInfo?: AnalyticsPageInfo; } /** @@ -54,7 +56,15 @@ export function usePaginatedList( // Cache for fetched pages - key is cursor (or 'initial' for first page) const pageCache = useRef< - Map + Map< + string, + { + data: T[]; + cursor?: string; + hasMore: boolean; + pageInfo?: AnalyticsPageInfo; + } + > >(new Map()); const fetchPage = useCallback( @@ -82,6 +92,7 @@ export function usePaginatedList( data: cached.data, isLoading: false, error: null, + pageInfo: cached.pageInfo, }); return newMap; }); @@ -99,6 +110,7 @@ export function usePaginatedList( data: result.data, cursor: result.cursor, hasMore: result.hasMore, + pageInfo: result.pageInfo, }); setAllPageResults((prev) => { @@ -107,6 +119,7 @@ export function usePaginatedList( data: result.data, isLoading: false, error: null, + pageInfo: result.pageInfo, }); return newMap; }); @@ -233,6 +246,7 @@ export function usePaginatedList( reload, refresh, pageInfo, + analyticsPageInfo: currentPageResult.pageInfo, }; } diff --git a/packages/web/app/lib/client/workflow-errors.test.ts b/packages/web/app/lib/client/workflow-errors.test.ts index 9ecb23b062..73bb1c84e7 100644 --- a/packages/web/app/lib/client/workflow-errors.test.ts +++ b/packages/web/app/lib/client/workflow-errors.test.ts @@ -1,5 +1,12 @@ import { describe, expect, it } from 'vitest'; -import { WorkflowWebAPIError, unwrapOrThrow } from './workflow-errors'; +import { + createWorkflowAPIError, + getErrorMessage, + getErrorTitle, + isObservabilityUpgradeRequired, + unwrapOrThrow, + WorkflowWebAPIError, +} from './workflow-errors'; describe('unwrapOrThrow', () => { it('returns data on success', async () => { @@ -40,4 +47,25 @@ describe('unwrapOrThrow', () => { expect(err).toBeInstanceOf(WorkflowWebAPIError); expect(err.message).toBe('network error'); }); + + it('formats observability plan cap errors as upgrade prompts', () => { + const error = createWorkflowAPIError({ + message: 'upgrade required', + layer: 'API', + request: { + operation: 'fetchRun', + params: { runId: 'run_123' }, + status: 402, + code: 'observability-upgrade-required', + }, + }); + + expect(isObservabilityUpgradeRequired(error)).toBe(true); + expect(getErrorTitle(error, 'Error loading run')).toBe( + 'Upgrade Observability Plus' + ); + expect(getErrorMessage(error)).toBe( + 'This workflow observability data is outside your current plan window. Upgrade Observability Plus to view up to 30 days of workflow data.' + ); + }); }); diff --git a/packages/web/app/lib/client/workflow-errors.ts b/packages/web/app/lib/client/workflow-errors.ts index 2230492102..a5ed9fc51d 100644 --- a/packages/web/app/lib/client/workflow-errors.ts +++ b/packages/web/app/lib/client/workflow-errors.ts @@ -1,6 +1,11 @@ import { VERCEL_403_ERROR_MESSAGE } from '@workflow/errors'; import type { ServerActionError } from '~/lib/types'; +const OBSERVABILITY_UPGRADE_REQUIRED_CODE = 'observability-upgrade-required'; +const OBSERVABILITY_UPGRADE_REQUIRED_MESSAGE = + 'This workflow observability data is outside your current plan window. Upgrade Observability Plus to view up to 30 days of workflow data.'; +const OBSERVABILITY_UPGRADE_REQUIRED_TITLE = 'Upgrade Observability Plus'; + /** * Error instance for API and server-side errors. * `error.message` will be a user-facing error message, to be displayed in UI. @@ -49,11 +54,33 @@ export function createWorkflowAPIError( }); } +export function isObservabilityUpgradeRequired(error: unknown): boolean { + return ( + error instanceof WorkflowWebAPIError && + error.request?.status === 402 && + error.request?.code === OBSERVABILITY_UPGRADE_REQUIRED_CODE + ); +} + +export function getErrorTitle( + error: Error | WorkflowWebAPIError, + fallback: string +): string { + if (isObservabilityUpgradeRequired(error)) { + return OBSERVABILITY_UPGRADE_REQUIRED_TITLE; + } + return fallback; +} + /** * Gets a user-facing error message from an error object. * Handles both WorkflowWebAPIError and regular Error instances. */ export const getErrorMessage = (error: Error | WorkflowWebAPIError): string => { + if (isObservabilityUpgradeRequired(error)) { + return OBSERVABILITY_UPGRADE_REQUIRED_MESSAGE; + } + if ('layer' in error && error.layer) { if (error instanceof WorkflowWebAPIError) { if (error.request?.status === 403) { diff --git a/packages/web/app/lib/types.ts b/packages/web/app/lib/types.ts index 9624283a72..a7231b6328 100644 --- a/packages/web/app/lib/types.ts +++ b/packages/web/app/lib/types.ts @@ -62,10 +62,19 @@ export type ServerActionResult = | { success: true; data: T } | { success: false; error: ServerActionError }; +export interface AnalyticsPageInfo { + currentLookbackDays: number; + maxLookbackDays: number; + currentWindowStart: Date | string; + maxWindowStart: Date | string; + upgradeAvailable: boolean; +} + export interface PaginatedResult { data: T[]; cursor?: string; hasMore: boolean; + pageInfo?: AnalyticsPageInfo; } export interface StopSleepResult { diff --git a/packages/web/app/lib/workflow-api-client.ts b/packages/web/app/lib/workflow-api-client.ts index 41b2b0c07c..0b034daa64 100644 --- a/packages/web/app/lib/workflow-api-client.ts +++ b/packages/web/app/lib/workflow-api-client.ts @@ -35,6 +35,7 @@ export { } from './client/workflow-actions'; export { getErrorMessage, + getErrorTitle, unwrapServerActionResult, WorkflowWebAPIError, } from './client/workflow-errors'; diff --git a/packages/web/app/server/workflow-server-actions.server.test.ts b/packages/web/app/server/workflow-server-actions.server.test.ts new file mode 100644 index 0000000000..29f4a1d964 --- /dev/null +++ b/packages/web/app/server/workflow-server-actions.server.test.ts @@ -0,0 +1,71 @@ +import type { AnalyticsEvent } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { analyticsEventToEvent } from './workflow-server-actions.server'; + +const makeAnalyticsEvent = ( + overrides: Partial +): AnalyticsEvent => ({ + runId: 'run-1', + eventId: 'event-1', + eventType: 'wait_created', + correlationId: 'wait-1', + entityId: 'wait-1', + stepName: null, + workflowName: 'workflow//./src/workflows/test//myWorkflow', + deploymentId: 'dep-1', + specVersion: 2, + runCreatedAt: new Date('2026-06-30T00:00:00.000Z'), + createdAt: new Date('2026-06-30T00:00:01.000Z'), + region: null, + vercelId: null, + requestId: null, + resumeAt: null, + retryAfter: null, + errorCode: null, + workflowCoreVersion: null, + isWebhook: null, + isSystem: null, + workflowEncryptionEnabled: false, + ...overrides, +}); + +describe('analyticsEventToEvent', () => { + it('preserves wait resumeAt metadata in eventData', () => { + const resumeAt = new Date('2026-06-30T00:05:00.000Z'); + const event = analyticsEventToEvent( + makeAnalyticsEvent({ + resumeAt, + }) + ); + + expect(event).toMatchObject({ + runId: 'run-1', + eventId: 'event-1', + eventType: 'wait_created', + correlationId: 'wait-1', + eventData: { resumeAt }, + }); + }); + + it('preserves stepName and retryAfter metadata in eventData', () => { + const retryAfter = new Date('2026-06-30T00:10:00.000Z'); + const event = analyticsEventToEvent( + makeAnalyticsEvent({ + eventType: 'step_retrying', + correlationId: 'step-1', + entityId: 'step-1', + stepName: 'step//./src/workflows/test//doWork', + retryAfter, + }) + ); + + expect(event).toMatchObject({ + eventType: 'step_retrying', + correlationId: 'step-1', + eventData: { + stepName: 'step//./src/workflows/test//doWork', + retryAfter, + }, + }); + }); +}); diff --git a/packages/web/app/server/workflow-server-actions.server.ts b/packages/web/app/server/workflow-server-actions.server.ts index 3bdfe975fb..f1e1c5ec9f 100644 --- a/packages/web/app/server/workflow-server-actions.server.ts +++ b/packages/web/app/server/workflow-server-actions.server.ts @@ -16,9 +16,10 @@ import { } from '@workflow/core/runtime/helpers'; import { resumeHook as resumeHookRuntime } from '@workflow/core/runtime/resume-hook'; -import { WorkflowWorldError, WorkflowRunNotFoundError } from '@workflow/errors'; +import { WorkflowRunNotFoundError, WorkflowWorldError } from '@workflow/errors'; import { findWorkflowDataDir } from '@workflow/utils/check-data-dir'; import type { + AnalyticsEvent, Event, Hook, Step, @@ -26,7 +27,7 @@ import type { WorkflowRunStatus, World, } from '@workflow/world'; -import { type APIConfig, createVercelWorld } from '@workflow/world-vercel'; +import { createVercelWorld } from '@workflow/world-vercel'; /** * Environment variable map for world configuration. @@ -357,6 +358,7 @@ export interface PaginatedResult { data: T[]; cursor?: string; hasMore: boolean; + pageInfo?: AnalyticsPageInfo; } /** @@ -383,6 +385,29 @@ export type ServerActionResult = | { success: true; data: T } | { success: false; error: ServerActionError }; +interface AnalyticsPageInfo { + currentLookbackDays: number; + maxLookbackDays: number; + currentWindowStart: Date; + maxWindowStart: Date; + upgradeAvailable: boolean; +} + +const OBSERVABILITY_UPGRADE_REQUIRED_CODE = 'observability-upgrade-required'; +const OBSERVABILITY_UPGRADE_REQUIRED_MESSAGE = + 'This workflow observability data is outside your current plan window. Upgrade Observability Plus to view up to 30 days of workflow data.'; + +function getPageInfo(result: unknown): AnalyticsPageInfo | undefined { + if ( + typeof result !== 'object' || + result === null || + !('pageInfo' in result) + ) { + return undefined; + } + return (result as { pageInfo?: AnalyticsPageInfo }).pageInfo; +} + /** * Cache for World instances. * @@ -472,7 +497,7 @@ function createServerActionError( console.error(`[web-api] ${operation} error:`, err); } errorResponse = { - message: getUserFacingErrorMessage(err, error.status), + message: getUserFacingErrorMessage(err, error.status, error.code), layer: 'API', cause: err.stack || err.message, request: { @@ -511,11 +536,19 @@ function createServerActionError( /** * Converts an error into a user-facing message */ -function getUserFacingErrorMessage(error: Error, status?: number): string { +function getUserFacingErrorMessage( + error: Error, + status?: number, + code?: string +): string { if (!status) { return `Error creating response: ${error.message}`; } + if (status === 402 && code === OBSERVABILITY_UPGRADE_REQUIRED_CODE) { + return OBSERVABILITY_UPGRADE_REQUIRED_MESSAGE; + } + // Check for common error patterns if (status === 403 || status === 401) { return 'Access denied. Please check your credentials and permissions.'; @@ -571,16 +604,25 @@ export async function fetchRuns( } = params; try { const world = await getWorldFromEnv(worldEnv); - const result = await world.runs.list({ - ...(workflowName ? { workflowName } : {}), - ...(status ? { status: status } : {}), - pagination: { cursor, limit, sortOrder }, - resolveData: 'none', - }); + // Prefer the metadata-only analytics read path when the backend provides + // one; fall back to the runtime storage API otherwise. + const result = world.analytics + ? await world.analytics.runs.list({ + ...(workflowName ? { workflowName } : {}), + ...(status ? { status } : {}), + pagination: { cursor, limit, sortOrder }, + }) + : await world.runs.list({ + ...(workflowName ? { workflowName } : {}), + ...(status ? { status } : {}), + pagination: { cursor, limit, sortOrder }, + resolveData: 'none', + }); return createResponse({ data: result.data as unknown as WorkflowRun[], cursor: result.cursor ?? undefined, hasMore: result.hasMore, + pageInfo: getPageInfo(result), }); } catch (error) { return createServerActionError>( @@ -626,16 +668,24 @@ export async function fetchSteps( const { cursor, sortOrder = 'asc', limit = 100 } = params; try { const world = await getWorldFromEnv(worldEnv); - const result = await world.steps.list({ - runId, - pagination: { cursor, limit, sortOrder }, - resolveData: 'none', - }); + // Prefer the metadata-only analytics read path when the backend provides + // one; fall back to the runtime storage API otherwise. + const result = world.analytics + ? await world.analytics.steps.list({ + runId, + pagination: { cursor, limit, sortOrder }, + }) + : await world.steps.list({ + runId, + pagination: { cursor, limit, sortOrder }, + resolveData: 'none', + }); return createResponse({ // StepWithoutData has undefined input/output, but after hydration the structure is compatible data: result.data as unknown as Step[], cursor: result.cursor ?? undefined, hasMore: result.hasMore, + pageInfo: getPageInfo(result), }); } catch (error) { return createServerActionError>( @@ -671,6 +721,35 @@ export async function fetchStep( } } +/** + * Adapt a metadata-only analytics event row to the runtime `Event` shape the + * observability UI consumes. The analytics row carries event metadata as flat + * top-level fields, whereas the runtime `Event` nests step metadata under + * `eventData`. We reconstruct that nested shape (matching what the runtime list + * returns with `resolveData: 'none'`), carrying `stepName` when present. Payload + * fields (input/output/result/error/payload) are intentionally omitted — they + * are loaded lazily per event via `fetchEvent(..., 'all')` on the runtime path. + */ +export function analyticsEventToEvent(event: AnalyticsEvent): Event { + const eventData = { + ...(event.stepName ? { stepName: event.stepName } : {}), + ...(event.resumeAt ? { resumeAt: event.resumeAt } : {}), + ...(event.retryAfter ? { retryAfter: event.retryAfter } : {}), + }; + const base = { + runId: event.runId, + eventId: event.eventId, + eventType: event.eventType, + createdAt: event.createdAt, + ...(event.correlationId ? { correlationId: event.correlationId } : {}), + ...(event.specVersion !== undefined + ? { specVersion: event.specVersion } + : {}), + ...(Object.keys(eventData).length > 0 ? { eventData } : {}), + }; + return base as unknown as Event; +} + /** * Fetch paginated list of events for a run */ @@ -687,6 +766,22 @@ export async function fetchEvents( const { cursor, sortOrder = 'asc', limit = 1000, withData = false } = params; try { const world = await getWorldFromEnv(worldEnv); + // Prefer the metadata-only analytics read path when the backend provides one + // and no payloads are requested. Event payloads are loaded lazily per event + // via `fetchEvent(..., 'all')`, so the list never needs them; the analytics + // rows are mapped to the runtime `Event` shape the UI consumes. + if (world.analytics && !withData) { + const result = await world.analytics.events.list({ + runId, + pagination: { cursor, limit, sortOrder }, + }); + return createResponse({ + data: result.data.map(analyticsEventToEvent), + cursor: result.cursor ?? undefined, + hasMore: result.hasMore, + pageInfo: getPageInfo(result), + }); + } const result = await world.events.list({ runId, pagination: { cursor, limit, sortOrder }, @@ -696,6 +791,7 @@ export async function fetchEvents( data: result.data as unknown as Event[], cursor: result.cursor ?? undefined, hasMore: result.hasMore, + pageInfo: getPageInfo(result), }); } catch (error) { return createServerActionError>( @@ -747,6 +843,22 @@ export async function fetchEventsByCorrelationId( const { cursor, sortOrder = 'asc', limit = 100, withData = false } = params; try { const world = await getWorldFromEnv(worldEnv); + // Prefer the metadata-only analytics read path when the backend provides one + // and no payloads are requested (payloads are loaded lazily per event via + // `fetchEvent(..., 'all')`); analytics rows are mapped to the runtime `Event` + // shape the UI consumes. + if (world.analytics && !withData) { + const result = await world.analytics.events.listByCorrelationId({ + correlationId, + pagination: { cursor, limit, sortOrder }, + }); + return createResponse({ + data: result.data.map(analyticsEventToEvent), + cursor: result.cursor ?? undefined, + hasMore: result.hasMore, + pageInfo: getPageInfo(result), + }); + } const result = await world.events.listByCorrelationId({ correlationId, pagination: { cursor, limit, sortOrder }, @@ -756,6 +868,7 @@ export async function fetchEventsByCorrelationId( data: result.data as unknown as Event[], cursor: result.cursor ?? undefined, hasMore: result.hasMore, + pageInfo: getPageInfo(result), }); } catch (error) { return createServerActionError>( @@ -784,15 +897,20 @@ export async function fetchHooks( const { runId, cursor, sortOrder = 'desc', limit = 10 } = params; try { const world = await getWorldFromEnv(worldEnv); + // Always use the runtime storage API for hooks. Unlike runs/steps, the + // hooks table needs the secret `token` (used for the resume action and the + // copy-token affordance) and `ownerId`, which the metadata-only analytics + // hook rows do not carry. const result = await world.hooks.list({ ...(runId ? { runId } : {}), pagination: { cursor, limit, sortOrder }, resolveData: 'none', }); return createResponse({ - data: result.data as Hook[], + data: result.data as unknown as Hook[], cursor: result.cursor ?? undefined, hasMore: result.hasMore, + pageInfo: getPageInfo(result), }); } catch (error) { return createServerActionError>( From ae51f45166298e855885330b2f1d4e1f8d729faa Mon Sep 17 00:00:00 2001 From: Karthik Kalyan <105607645+karthikscale3@users.noreply.github.com> Date: Tue, 7 Jul 2026 12:06:41 -0700 Subject: [PATCH 3/8] web: list hooks from analytics, fetch token on demand (#2652) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Add workflow analytics world APIs * web: read observability list views from world.analytics when available Route the runs/steps/events/hooks list server actions through the optional world.analytics namespace when the backend provides one, falling back to the runtime storage APIs otherwise. Events listing only uses the analytics path when no payload data is requested. Detail/get actions, streams, and mutations are unchanged. * web: keep events and hooks list reads on the runtime storage API The Events tab and trace viewer derive step names and wait resumeAt from resolved event payloads, and the hooks table needs the secret token and ownerId for its resume/copy-token actions. The metadata-only analytics rows do not carry these, so only the runs and steps list views use world.analytics. Co-Authored-By: Claude Opus 4.8 * web: read events list from world.analytics with a runtime-shape remap The events list/trace consumers read only top-level eventType, correlationId, and createdAt; event payloads are loaded lazily per event via fetchEvent(..., 'all') on the runtime path. Map the flat analytics event rows into the runtime Event shape (reconstructing eventData.stepName) so fetchEvents and fetchEventsByCorrelationId can use the analytics read path when available. Hooks remain on the runtime path (they need the secret token + ownerId). Co-Authored-By: Claude Opus 4.8 * web: list hooks from world.analytics, fetch token on demand The hooks list now reads from the metadata-only world.analytics namespace when the backend provides one (falling back to the runtime storage APIs otherwise). A hook's secret token is no longer shipped in list rows — it is fetched one hook at a time via world.hooks.get only when the user copies the token or resumes the hook, keeping the secret out of bulk list responses. Adds a fetchHookToken server action + RPC, a HookListItem type (Hook without token), and a lazy HookTokenCell for the copy-token affordance. Co-Authored-By: Claude Opus 4.8 * Handle analytics access metadata in web --------- Co-authored-by: Claude Opus 4.8 --- .changeset/web-hook-token-lazy.md | 5 ++ .../display-utils/hook-token-cell.tsx | 80 +++++++++++++++++++ packages/web/app/components/hooks-table.tsx | 27 ++++--- .../lib/client/hooks/use-paginated-list.ts | 11 ++- .../web/app/lib/client/workflow-actions.ts | 18 +++++ packages/web/app/lib/rpc-client.ts | 12 ++- packages/web/app/lib/types.ts | 14 ++++ packages/web/app/lib/workflow-api-client.ts | 1 + packages/web/app/routes/api.rpc.tsx | 3 + .../workflow-server-actions.server.test.ts | 35 +++++++- .../server/workflow-server-actions.server.ts | 57 +++++++++++-- 11 files changed, 241 insertions(+), 22 deletions(-) create mode 100644 .changeset/web-hook-token-lazy.md create mode 100644 packages/web/app/components/display-utils/hook-token-cell.tsx diff --git a/.changeset/web-hook-token-lazy.md b/.changeset/web-hook-token-lazy.md new file mode 100644 index 0000000000..3943a2b0de --- /dev/null +++ b/.changeset/web-hook-token-lazy.md @@ -0,0 +1,5 @@ +--- +'@workflow/web': patch +--- + +The hooks list now reads from the metadata-only `world.analytics` namespace when the backend provides one (falling back to the runtime storage APIs otherwise). A hook's secret token is no longer included in list rows — it is fetched one hook at a time via `world.hooks.get` only when the user copies the token or resumes the hook. diff --git a/packages/web/app/components/display-utils/hook-token-cell.tsx b/packages/web/app/components/display-utils/hook-token-cell.tsx new file mode 100644 index 0000000000..e3d2ef4ae1 --- /dev/null +++ b/packages/web/app/components/display-utils/hook-token-cell.tsx @@ -0,0 +1,80 @@ +import { Check, Copy, Loader2 } from 'lucide-react'; +import { useState } from 'react'; +import { + Tooltip, + TooltipContent, + TooltipTrigger, +} from '~/components/ui/tooltip'; +import type { EnvMap } from '~/lib/types'; +import { fetchHookToken } from '~/lib/workflow-api-client'; + +interface HookTokenCellProps { + env: EnvMap; + runId: string; + hookId: string; +} + +/** + * Renders a hook's secret token as a masked, copy-on-demand cell. + * + * The token is not present in hook list rows — it is fetched one hook at a + * time (via `fetchHookToken`) only when the user clicks to copy it. This keeps + * the secret out of bulk list responses while preserving the copy affordance. + */ +export function HookTokenCell({ env, runId, hookId }: HookTokenCellProps) { + const [status, setStatus] = useState<'idle' | 'loading' | 'copied'>('idle'); + const [error, setError] = useState(null); + + const handleCopy = async (e: React.MouseEvent) => { + e.stopPropagation(); + if (status === 'loading') return; + setStatus('loading'); + setError(null); + try { + const token = await fetchHookToken(env, runId, hookId); + await navigator.clipboard.writeText(token); + setStatus('copied'); + setTimeout(() => setStatus('idle'), 2000); + } catch (err) { + console.error('Failed to copy hook token:', err); + setError('Failed to copy token'); + setStatus('idle'); + } + }; + + const tooltip = error + ? error + : status === 'copied' + ? 'Copied!' + : status === 'loading' + ? 'Fetching token…' + : 'Copy token'; + + return ( + + •••••••••••• + + + + + +

{tooltip}

+
+
+
+ ); +} diff --git a/packages/web/app/components/hooks-table.tsx b/packages/web/app/components/hooks-table.tsx index 069689b57e..be17f060cb 100644 --- a/packages/web/app/components/hooks-table.tsx +++ b/packages/web/app/components/hooks-table.tsx @@ -36,14 +36,16 @@ import { TooltipTrigger, } from '~/components/ui/tooltip'; import { fetchEvents } from '~/lib/rpc-client'; -import type { EnvMap } from '~/lib/types'; +import type { EnvMap, HookListItem } from '~/lib/types'; import { + fetchHookToken, getErrorMessage, getErrorTitle, resumeHook, useWorkflowHooks, } from '~/lib/workflow-api-client'; import { CopyableText } from './display-utils/copyable-text'; +import { HookTokenCell } from './display-utils/hook-token-cell'; import { RelativeTime } from './display-utils/relative-time'; import { TableSkeleton } from './display-utils/table-skeleton'; @@ -95,7 +97,10 @@ export function HooksTable({ // Hook actions for resolve functionality const hookActions = useHookActions({ onResolve: async (hook, payload) => { - await resumeHook(env, hook.token, payload); + // List rows are metadata-only; fetch the secret token on demand just + // before resuming, keyed by the hook's run/hook id. + const token = await fetchHookToken(env, hook.runId, hook.hookId); + await resumeHook(env, token, payload); }, callbacks: { onSuccess: refresh, @@ -202,7 +207,7 @@ export function HooksTable({ }, [hooks, env, runId]); // Render invocation count for a hook - const renderInvocationCount = (hook: Hook) => { + const renderInvocationCount = (hook: HookListItem) => { const data = invocationData.get(hook.hookId); if (!data || data.loading) { @@ -329,11 +334,11 @@ export function HooksTable({ - - - •••••••••••• - - + {hook.createdAt ? ( @@ -359,7 +364,11 @@ export function HooksTable({ { @@ -285,7 +290,7 @@ export function useWorkflowHooks( limit?: number; sortOrder?: 'asc' | 'desc'; } -): PaginatedList { +): PaginatedList { const { runId, limit = 10, sortOrder = 'desc' } = params; const fetchFn = useCallback( diff --git a/packages/web/app/lib/client/workflow-actions.ts b/packages/web/app/lib/client/workflow-actions.ts index 9531241d9c..f607d3b028 100644 --- a/packages/web/app/lib/client/workflow-actions.ts +++ b/packages/web/app/lib/client/workflow-actions.ts @@ -1,5 +1,6 @@ import { cancelRun as cancelRunServerAction, + fetchHookToken as fetchHookTokenServerAction, recreateRun as recreateRunServerAction, reenqueueRun as reenqueueRunServerAction, resumeHook as resumeHookServerAction, @@ -47,3 +48,20 @@ export async function resumeHook( ): Promise { return unwrapOrThrow(resumeHookServerAction(env, token, payload)); } + +/** + * Fetch a single hook's secret token on demand. + * + * Used by the hooks list, where rows are metadata-only: the token is fetched + * one hook at a time when the user reveals/copies it or resumes the hook. + */ +export async function fetchHookToken( + env: EnvMap, + runId: string, + hookId: string +): Promise { + const { token } = await unwrapOrThrow( + fetchHookTokenServerAction(env, runId, hookId) + ); + return token; +} diff --git a/packages/web/app/lib/rpc-client.ts b/packages/web/app/lib/rpc-client.ts index 2357faff3a..99641fb5f8 100644 --- a/packages/web/app/lib/rpc-client.ts +++ b/packages/web/app/lib/rpc-client.ts @@ -18,6 +18,8 @@ import type { EnvMap, HealthCheckEndpoint, HealthCheckResult, + HookListItem, + HookTokenResult, PaginatedResult, ResumeHookResult, ServerActionResult, @@ -149,10 +151,18 @@ export async function fetchHooks( sortOrder?: 'asc' | 'desc'; limit?: number; } -): Promise>> { +): Promise>> { return rpc('fetchHooks', { worldEnv, params }); } +export async function fetchHookToken( + worldEnv: EnvMap, + runId: string, + hookId: string +): Promise> { + return rpc('fetchHookToken', { worldEnv, runId, hookId }); +} + export async function fetchHook( worldEnv: EnvMap, hookId: string, diff --git a/packages/web/app/lib/types.ts b/packages/web/app/lib/types.ts index a7231b6328..2eacb2904c 100644 --- a/packages/web/app/lib/types.ts +++ b/packages/web/app/lib/types.ts @@ -3,6 +3,8 @@ * This file should NOT import any server-only modules. */ +import type { Hook } from '@workflow/world'; + /** * Public configuration info that is safe to send to the client. * @@ -91,6 +93,18 @@ export interface ResumeHookResult { runId: string; } +/** + * A hook as shown in list views. The secret `token` is omitted: list rows are + * metadata-only, and the token is fetched on demand per hook (see + * `fetchHookToken`) for the copy-token and resume affordances. + */ +export type HookListItem = Omit; + +/** Result of a lazy per-hook token fetch. */ +export interface HookTokenResult { + token: string; +} + export type { HealthCheckEndpoint, HealthCheckResult, diff --git a/packages/web/app/lib/workflow-api-client.ts b/packages/web/app/lib/workflow-api-client.ts index 0b034daa64..89a93b2382 100644 --- a/packages/web/app/lib/workflow-api-client.ts +++ b/packages/web/app/lib/workflow-api-client.ts @@ -28,6 +28,7 @@ export { useWorkflowTraceViewerData } from './client/hooks/use-trace-viewer'; export { useWorkflowStreams } from './client/hooks/use-workflow-streams'; export { cancelRun, + fetchHookToken, recreateRun, reenqueueRun, resumeHook, diff --git a/packages/web/app/routes/api.rpc.tsx b/packages/web/app/routes/api.rpc.tsx index a3824ab447..4bca070745 100644 --- a/packages/web/app/routes/api.rpc.tsx +++ b/packages/web/app/routes/api.rpc.tsx @@ -14,6 +14,7 @@ import { fetchEventsByCorrelationId, fetchHook, fetchHooks, + fetchHookToken, fetchRun, fetchRuns, fetchStep, @@ -49,6 +50,8 @@ const handlers = { p.params ?? {} ), fetchHooks: (p: any) => fetchHooks(p.worldEnv ?? {}, p.params ?? {}), + fetchHookToken: (p: any) => + fetchHookToken(p.worldEnv ?? {}, p.runId, p.hookId), fetchHook: (p: any) => fetchHook(p.worldEnv ?? {}, p.hookId, p.resolveData), cancelRun: (p: any) => cancelRun(p.worldEnv ?? {}, p.runId), recreateRun: (p: any) => diff --git a/packages/web/app/server/workflow-server-actions.server.test.ts b/packages/web/app/server/workflow-server-actions.server.test.ts index 29f4a1d964..b6fbf9cc39 100644 --- a/packages/web/app/server/workflow-server-actions.server.test.ts +++ b/packages/web/app/server/workflow-server-actions.server.test.ts @@ -1,6 +1,9 @@ -import type { AnalyticsEvent } from '@workflow/world'; +import type { AnalyticsEvent, Hook } from '@workflow/world'; import { describe, expect, it } from 'vitest'; -import { analyticsEventToEvent } from './workflow-server-actions.server'; +import { + analyticsEventToEvent, + hookToListItem, +} from './workflow-server-actions.server'; const makeAnalyticsEvent = ( overrides: Partial @@ -69,3 +72,31 @@ describe('analyticsEventToEvent', () => { }); }); }); + +describe('hookToListItem', () => { + it('strips the secret token from runtime hook rows', () => { + const hook: Hook = { + hookId: 'hook-1', + runId: 'run-1', + token: 'secret-token', + ownerId: 'owner-1', + projectId: 'project-1', + environment: 'production', + createdAt: new Date('2026-06-30T00:00:00.000Z'), + specVersion: 2, + }; + + const listItem = hookToListItem(hook); + + expect(listItem).toEqual({ + hookId: 'hook-1', + runId: 'run-1', + ownerId: 'owner-1', + projectId: 'project-1', + environment: 'production', + createdAt: new Date('2026-06-30T00:00:00.000Z'), + specVersion: 2, + }); + expect('token' in listItem).toBe(false); + }); +}); diff --git a/packages/web/app/server/workflow-server-actions.server.ts b/packages/web/app/server/workflow-server-actions.server.ts index f1e1c5ec9f..9262e5f66f 100644 --- a/packages/web/app/server/workflow-server-actions.server.ts +++ b/packages/web/app/server/workflow-server-actions.server.ts @@ -28,6 +28,7 @@ import type { World, } from '@workflow/world'; import { createVercelWorld } from '@workflow/world-vercel'; +import type { HookListItem, HookTokenResult } from '~/lib/types'; /** * Environment variable map for world configuration. @@ -885,6 +886,11 @@ export async function fetchEventsByCorrelationId( /** * Fetch paginated list of hooks */ +export function hookToListItem(hook: Hook): HookListItem { + const { token: _token, ...hookListItem } = hook; + return hookListItem; +} + export async function fetchHooks( worldEnv: EnvMap, params: { @@ -893,27 +899,40 @@ export async function fetchHooks( sortOrder?: 'asc' | 'desc'; limit?: number; } -): Promise>> { +): Promise>> { const { runId, cursor, sortOrder = 'desc', limit = 10 } = params; try { const world = await getWorldFromEnv(worldEnv); - // Always use the runtime storage API for hooks. Unlike runs/steps, the - // hooks table needs the secret `token` (used for the resume action and the - // copy-token affordance) and `ownerId`, which the metadata-only analytics - // hook rows do not carry. + // Prefer the metadata-only analytics read path when the backend provides + // one and the run scope required by analytics is present. The hook list is + // metadata only — the secret `token` is fetched on demand per hook via + // `fetchHookToken` for the copy-token and resume affordances — so the list + // never carries it. + if (world.analytics && runId) { + const result = await world.analytics.hooks.list({ + runId, + pagination: { cursor, limit, sortOrder }, + }); + return createResponse({ + data: result.data as unknown as HookListItem[], + cursor: result.cursor ?? undefined, + hasMore: result.hasMore, + pageInfo: getPageInfo(result), + }); + } const result = await world.hooks.list({ ...(runId ? { runId } : {}), pagination: { cursor, limit, sortOrder }, resolveData: 'none', }); return createResponse({ - data: result.data as unknown as Hook[], + data: result.data.map(hookToListItem), cursor: result.cursor ?? undefined, hasMore: result.hasMore, pageInfo: getPageInfo(result), }); } catch (error) { - return createServerActionError>( + return createServerActionError>( error, 'world.hooks.list', params @@ -921,6 +940,30 @@ export async function fetchHooks( } } +/** + * Fetch a single hook's secret token on demand. + * + * Hook list rows are metadata-only; the token is read from the runtime storage + * API one hook at a time, only when the user reveals/copies it or resumes the + * hook. This keeps the secret out of bulk list responses. + */ +export async function fetchHookToken( + worldEnv: EnvMap, + runId: string, + hookId: string +): Promise> { + try { + const world = await getWorldFromEnv(worldEnv); + const hook = await world.hooks.get(hookId, { resolveData: 'none' }); + return createResponse({ token: hook.token }); + } catch (error) { + return createServerActionError(error, 'world.hooks.get', { + runId, + hookId, + }); + } +} + /** * Fetch a single hook with full data */ From 7637196cf0f605ce62243bf8c7762a26153dcd36 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 7 Jul 2026 14:13:15 -0700 Subject: [PATCH 4/8] Fix hook token reuse after dispose() (same-run and cross-run) (#2779) --- .changeset/hook-token-claim-release.md | 5 + .changeset/hook-token-reuse-after-dispose.md | 5 + packages/core/e2e/e2e.test.ts | 50 ++++- .../src/runtime/suspension-handler.test.ts | 112 ++++++++++ .../core/src/runtime/suspension-handler.ts | 201 ++++++++++-------- packages/world-local/src/storage.test.ts | 81 ++++++- .../world-local/src/storage/events-storage.ts | 161 +++++++++++--- packages/world-local/src/storage/helpers.ts | 44 +++- .../world-local/src/storage/hooks-storage.ts | 18 +- workbench/example/workflows/99_e2e.ts | 33 +++ .../vitest/test/hook-token-reuse.test.ts | 55 +++++ .../vitest/workflows/hook-token-reuse.ts | 43 ++++ 12 files changed, 685 insertions(+), 123 deletions(-) create mode 100644 .changeset/hook-token-claim-release.md create mode 100644 .changeset/hook-token-reuse-after-dispose.md create mode 100644 workbench/vitest/test/hook-token-reuse.test.ts create mode 100644 workbench/vitest/workflows/hook-token-reuse.ts diff --git a/.changeset/hook-token-claim-release.md b/.changeset/hook-token-claim-release.md new file mode 100644 index 0000000000..c645a95b35 --- /dev/null +++ b/.changeset/hook-token-claim-release.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-local': patch +--- + +Fix hook token claims of disposed hooks and finished runs not being released to the next claimant diff --git a/.changeset/hook-token-reuse-after-dispose.md b/.changeset/hook-token-reuse-after-dispose.md new file mode 100644 index 0000000000..f81f94555a --- /dev/null +++ b/.changeset/hook-token-reuse-after-dispose.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Fix `createHook()` conflicting with the run's own disposed hook when a token is reused after `dispose()` within the same run diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 6db2fca8c7..3dc0e27e6c 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -148,9 +148,15 @@ async function waitForHook( timeoutMs?: number; intervalMs?: number; runId?: string; + excludeHookId?: string; } = {} ): Promise>> { - const { timeoutMs = 30_000, intervalMs = 250, runId } = options; + const { + timeoutMs = 30_000, + intervalMs = 250, + runId, + excludeHookId, + } = options; const deadline = Date.now() + timeoutMs; let lastError: unknown = new Error( `waitForHook(${token}) timed out before any attempt` @@ -166,6 +172,12 @@ async function waitForHook( lastError = new Error( `waitForHook(${token}) saw runId=${hook.runId}, expected ${runId}` ); + } else if (excludeHookId && hook.hookId === excludeHookId) { + // Same-run token-reuse tests wait for the NEXT hook on the token; + // keep polling while the lookup still resolves to the prior hook. + lastError = new Error( + `waitForHook(${token}) still sees hookId=${hook.hookId}` + ); } else { return hook; } @@ -2305,6 +2317,42 @@ describe('e2e', () => { } ); + // Regression test for #2777: recreating a hook with the same token after + // dispose() within a single run must not conflict with the run's own + // disposed hook. + test( + 'hookTokenReuseLoopWorkflow - same run recreates a hook with the same token after dispose()', + { timeout: 90_000 }, + async () => { + const token = Math.random().toString(36).slice(2); + const rounds = 3; + + const run = await start(await e2e('hookTokenReuseLoopWorkflow'), [ + token, + rounds, + ]); + + let previousHookId: string | undefined; + for (let round = 0; round < rounds; round++) { + const hook = await waitForHook(token, { + runId: run.runId, + excludeHookId: previousHookId, + }); + previousHookId = hook.hookId; + await resumeHook(hook, { message: `round-${round}` }); + } + + const result = await run.returnValue; + expect(result).toEqual({ + received: ['round-0', 'round-1', 'round-2'], + conflictRound: null, + }); + + const { json: runData } = await cliInspectJson(`runs ${run.runId}`); + expect(runData.status).toBe('completed'); + } + ); + test( 'stepFunctionPassingWorkflow - step function references can be passed as arguments (without closure vars)', { timeout: 60_000 }, diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index 9acf9b609d..9557ea6d84 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -247,4 +247,116 @@ describe('handleSuspension', () => { expect(result.hasAwaitedHookCreation).toBe(false); expect(result.timeoutSeconds).toBeUndefined(); }); + + // Regression test for #2777: a dispose() of an earlier hook must be + // flushed before a later same-token hook's creation is validated, or the + // new hook records a spurious hook_conflict against the run's own + // disposed hook. + it('flushes a prior hook disposal before validating a same-token recreation', async () => { + const eventsCreate = vi.fn(async (_runId, event) => ({ event })); + const world = createWorld(eventsCreate); + const pending = new Map([ + [ + 'hook_old', + { + type: 'hook' as const, + correlationId: 'hook_old', + token: 'reused-token', + hasCreatedEvent: true, + disposed: true, + }, + ], + [ + 'hook_new', + { + type: 'hook' as const, + correlationId: 'hook_new', + token: 'reused-token', + hasConflictAwaiter: true, + }, + ], + ]); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + const hookCalls = eventsCreate.mock.calls.map(([, event]) => ({ + eventType: event.eventType, + correlationId: event.correlationId, + })); + expect(hookCalls).toEqual([ + { eventType: 'hook_disposed', correlationId: 'hook_old' }, + { eventType: 'hook_created', correlationId: 'hook_new' }, + ]); + expect(result.hasHookConflict).toBe(false); + expect(result.hasAwaitedHookCreation).toBe(true); + }); + + it('creates a hook before disposing it when both happen within one suspension', async () => { + const eventsCreate = vi.fn(async (_runId, event) => ({ event })); + const world = createWorld(eventsCreate); + const pending = new Map([ + [ + 'hook_ephemeral', + { + type: 'hook' as const, + correlationId: 'hook_ephemeral', + token: 'ephemeral-token', + disposed: true, + }, + ], + ]); + + await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + const hookCalls = eventsCreate.mock.calls.map(([, event]) => ({ + eventType: event.eventType, + correlationId: event.correlationId, + })); + expect(hookCalls).toEqual([ + { eventType: 'hook_created', correlationId: 'hook_ephemeral' }, + { eventType: 'hook_disposed', correlationId: 'hook_ephemeral' }, + ]); + }); + + it('does not dispose a hook whose creation conflicted', async () => { + const eventsCreate = vi.fn(async (_runId, event) => { + if (event.eventType === 'hook_created') { + return { event: { eventType: 'hook_conflict' } }; + } + return { event }; + }); + const world = createWorld(eventsCreate); + const pending = new Map([ + [ + 'hook_contended', + { + type: 'hook' as const, + correlationId: 'hook_contended', + token: 'contended-token', + disposed: true, + }, + ], + ]); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(pending, globalThis), + world, + run, + }); + + expect(result.hasHookConflict).toBe(true); + expect( + eventsCreate.mock.calls.some( + ([, event]) => event.eventType === 'hook_disposed' + ) + ).toBe(false); + }); }); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index bca1453693..b0b212b994 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -216,11 +216,30 @@ export async function handleSuspension({ (item): item is AttributeInvocationQueueItem => item.type === 'attribute' ); - // Split hooks by what actions they need const hooksNeedingCreation = allHookItems.filter( (item) => !item.hasCreatedEvent ); - const hooksNeedingDisposal = allHookItems.filter((item) => item.disposed); + + // Group hook items that need work by token, preserving queue-insertion + // (workflow code) order within each token. Operations on one token must + // apply in code order: a dispose() of an earlier hook releases the token + // before a later same-token hook's creation is validated — otherwise the + // new hook records a spurious hook_conflict against the run's own + // disposed hook — while a hook created and disposed within the same + // suspension is still created before it is disposed. Different tokens + // have no claim interaction, so token groups are processed in parallel. + const hookItemsByToken = new Map(); + for (const item of allHookItems) { + if (item.hasCreatedEvent && !item.disposed) { + continue; // already committed and still live — nothing to do + } + const group = hookItemsByToken.get(item.token); + if (group) { + group.push(item); + } else { + hookItemsByToken.set(item.token, [item]); + } + } // Resolve encryption key for this run const rawKey = await world.getEncryptionKeyForRun?.(run); @@ -231,107 +250,105 @@ export async function handleSuspension({ const compression = (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION; - // Build and process hook_created events (same as V1) - const hookEvents = await Promise.all( - hooksNeedingCreation.map(async (queueItem) => { - const hookMetadata: SerializedData | undefined = - typeof queueItem.metadata === 'undefined' - ? undefined - : ((await dehydrateStepArguments( - queueItem.metadata, - runId, - encryptionKey, - suspension.globalThis, - false, - compression - )) as SerializedData); - return { - queueItem, - hookEvent: { - eventType: 'hook_created' as const, - specVersion: SPEC_VERSION_CURRENT, + async function disposeHook( + queueItem: HookInvocationQueueItem + ): Promise { + const hookDisposedEvent: CreateEventRequest = { + eventType: 'hook_disposed' as const, + specVersion: SPEC_VERSION_CURRENT, + correlationId: queueItem.correlationId, + eventData: { + token: queueItem.token, + }, + }; + try { + await world.events.create(runId, hookDisposedEvent, { requestId }); + } catch (err) { + if (EntityConflictError.is(err)) { + // Hook was already disposed by a concurrent invocation — safe to skip + runtimeLogger.info( + 'Hook already disposed, skipping duplicate disposal', + { + workflowRunId: runId, + correlationId: queueItem.correlationId, + message: err.message, + } + ); + } else if (RunExpiredError.is(err)) { + runtimeLogger.info( + 'Workflow run already completed, skipping hook disposal', + { + workflowRunId: runId, + correlationId: queueItem.correlationId, + message: err.message, + } + ); + } else if (HookNotFoundError.is(err)) { + // Hook may have already been disposed or never created + runtimeLogger.info('Hook not found for disposal, continuing', { + workflowRunId: runId, correlationId: queueItem.correlationId, - eventData: { - token: queueItem.token, - metadata: hookMetadata, - isWebhook: queueItem.isWebhook ?? false, - ...(queueItem.isSystem && { isSystem: true }), - }, - }, - }; - }) - ); + message: err.message, + }); + } else { + throw err; + } + } + } // Process hooks first to prevent race conditions with webhook receivers. - // All hook creations run in parallel. // Track any hook conflicts that occur — these are returned to the caller // so the V2 handler can re-invoke immediately. let hasHookConflict = false; let hasAwaitedHookCreation = false; - if (hookEvents.length > 0) { - await ensureRunReady(); - const results = await Promise.all( - hookEvents.map(({ hookEvent, queueItem }) => - createHookEvent({ - world, - runId, - hookEvent, - queueItem, - requestId, - }) - ) - ); - hasHookConflict = results.some((result) => result.hasHookConflict); - hasAwaitedHookCreation = results.some( - (result) => result.hasAwaitedHookCreation - ); - } - - // Process hook disposals — these release hook tokens for reuse by other workflows. - if (hooksNeedingDisposal.length > 0) { + if (hookItemsByToken.size > 0) { await ensureRunReady(); await Promise.all( - hooksNeedingDisposal.map(async (queueItem) => { - const hookDisposedEvent: CreateEventRequest = { - eventType: 'hook_disposed' as const, - specVersion: SPEC_VERSION_CURRENT, - correlationId: queueItem.correlationId, - eventData: { - token: queueItem.token, - }, - }; - try { - await world.events.create(runId, hookDisposedEvent, { requestId }); - } catch (err) { - if (EntityConflictError.is(err)) { - // Hook was already disposed by a concurrent invocation — safe to skip - runtimeLogger.info( - 'Hook already disposed, skipping duplicate disposal', - { - workflowRunId: runId, - correlationId: queueItem.correlationId, - message: err.message, - } - ); - } else if (RunExpiredError.is(err)) { - runtimeLogger.info( - 'Workflow run already completed, skipping hook disposal', - { - workflowRunId: runId, - correlationId: queueItem.correlationId, - message: err.message, - } - ); - } else if (HookNotFoundError.is(err)) { - // Hook may have already been disposed or never created - runtimeLogger.info('Hook not found for disposal, continuing', { - workflowRunId: runId, + [...hookItemsByToken.values()].map(async (items) => { + for (const queueItem of items) { + let creationConflicted = false; + + if (!queueItem.hasCreatedEvent) { + const hookMetadata: SerializedData | undefined = + typeof queueItem.metadata === 'undefined' + ? undefined + : ((await dehydrateStepArguments( + queueItem.metadata, + runId, + encryptionKey, + suspension.globalThis, + false, + compression + )) as SerializedData); + const hookEvent: CreateEventRequest = { + eventType: 'hook_created' as const, + specVersion: SPEC_VERSION_CURRENT, correlationId: queueItem.correlationId, - message: err.message, + eventData: { + token: queueItem.token, + metadata: hookMetadata, + isWebhook: queueItem.isWebhook ?? false, + ...(queueItem.isSystem && { isSystem: true }), + }, + }; + const result = await createHookEvent({ + world, + runId, + hookEvent, + queueItem, + requestId, }); - } else { - throw err; + hasHookConflict ||= result.hasHookConflict; + hasAwaitedHookCreation ||= result.hasAwaitedHookCreation; + creationConflicted = result.hasHookConflict; + } + + // Dispose after creation for hooks born and disposed within this + // batch. A hook whose creation conflicted was never created, so + // there is nothing to dispose. + if (queueItem.disposed && !creationConflicted) { + await disposeHook(queueItem); } } }) @@ -654,7 +671,7 @@ export async function handleSuspension({ hasHookConflict, hasAwaitedHookCreation, hasAttributeEvents: attributeItems.length > 0, - hasHookEvents: hookEvents.length > 0, + hasHookEvents: hooksNeedingCreation.length > 0, }; } diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index 2126187baf..e5d8c1f214 100644 --- a/packages/world-local/src/storage.test.ts +++ b/packages/world-local/src/storage.test.ts @@ -7,7 +7,7 @@ import { SPEC_VERSION_CURRENT, stripEventDataRefs } from '@workflow/world'; import { monotonicFactory } from 'ulid'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { writeJSON } from './fs.js'; -import { hashToken } from './storage/helpers.js'; +import { hashToken, hookDisposeLockPath } from './storage/helpers.js'; import { createStorage } from './storage.js'; import { completeWait, @@ -2197,6 +2197,85 @@ describe('Storage', () => { expect(hook2.hookId).toBe('hook_2'); }); + // Regression test for #2778: a claim whose owning run is terminal can + // never become live again — a new claimant must treat it as vacant + // instead of recording a hook_conflict against a finished run. The + // claim file is rewritten manually to simulate the window where the + // run-completion cleanup has not deleted it yet (or crashed). + it('should treat a token claim owned by a terminal run as vacant', async () => { + const token = 'terminal-owner-token'; + + await createHook(storage, testRunId, { + hookId: 'hook_1', + token, + }); + await updateRun(storage, testRunId, 'run_started'); + await updateRun(storage, testRunId, 'run_completed', { + output: new Uint8Array(), + }); + + // Simulate the stale claim surviving the terminal-run cleanup + await fs.writeFile( + path.join(testDir, 'hooks', 'tokens', `${hashToken(token)}.json`), + JSON.stringify({ + token, + hookId: 'hook_1', + runId: testRunId, + eventId: 'evnt_00000000000000000000000000', + }) + ); + + const run2 = await createRun(storage, { + deploymentId: 'deployment-456', + workflowName: 'another-workflow', + input: new Uint8Array(), + }); + + const result = await storage.events.create(run2.runId, { + eventType: 'hook_created', + correlationId: 'hook_2', + eventData: { token }, + }); + + expect(result.event.eventType).toBe('hook_created'); + expect(result.hook?.runId).toBe(run2.runId); + }); + + // Regression test for #2778: the hook_disposed handler commits the + // disposal (dispose lock) before deleting the token claim, so a new + // claimant can observe the claim of a hook whose disposal is already + // committed. It must treat that claim as vacant — and the event-log + // rebuild must not resurrect the half-torn-down hook. + it('should treat a token claim of a dispose-committed hook as vacant', async () => { + const token = 'disposed-owner-token'; + + await createHook(storage, testRunId, { + hookId: 'hook_1', + token, + }); + + // Simulate a disposer that crashed right after committing the + // dispose lock, before deleting the claim and hook entity. + const lockPath = hookDisposeLockPath(testDir, 'hook_1'); + await fs.mkdir(path.dirname(lockPath), { recursive: true }); + await fs.writeFile(lockPath, ''); + + const run2 = await createRun(storage, { + deploymentId: 'deployment-456', + workflowName: 'another-workflow', + input: new Uint8Array(), + }); + + const result = await storage.events.create(run2.runId, { + eventType: 'hook_created', + correlationId: 'hook_2', + eventData: { token }, + }); + + expect(result.event.eventType).toBe('hook_created'); + expect(result.hook?.runId).toBe(run2.runId); + }); + it('should enforce token uniqueness across different runs within the same project', async () => { // Create a second run const run2 = await createRun(storage, { diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index d2130b4f2b..eb0e396464 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -57,7 +57,9 @@ import { stripEventDataRefs } from './filters.js'; import { getObjectCreatedAt, hashToken, + hookDisposeLockPath, hookRecoveryMarkerPath, + isHookDisposalCommitted, monotonicUlid, } from './helpers.js'; import { @@ -144,6 +146,46 @@ async function readHookTokenClaim( } } +/** + * Whether a token claim held by another `(runId, hookId)` can never become + * live again and may therefore be released by a new claimant: + * + * - the claimed hook's disposal is committed (its dispose lock exists — + * the durable release of the claim file just hasn't landed yet, or was + * lost to a crash between the lock write and the claim delete), or + * - the owning run is terminal (the run-completion cleanup releases all + * of the run's claims, but a claimant can observe the claim in the + * window before that cleanup lands — or the cleanup crashed), or + * - the owning run does not exist (a claim can only be written during a + * suspension of an existing run, so an ownerless claim is debris). + * + * A claim from a mid-creation writer is never releasable: its owning run + * exists and is non-terminal, and its dispose lock does not exist. + */ +async function isHookTokenClaimReleasable( + basedir: string, + claim: z.infer, + tag?: string +): Promise { + if ( + claim.hookId && + (await isHookDisposalCommitted(basedir, claim.hookId, tag)) + ) { + return true; + } + const owningRun = await readJSONWithFallback( + basedir, + 'runs', + claim.runId, + WorkflowRunSchema, + tag + ); + if (!owningRun) { + return true; + } + return isTerminalWorkflowRunStatus(owningRun.status); +} + async function readHookRecoveryMarker( markerPath: string ): Promise | null> { @@ -1563,28 +1605,95 @@ export function createEventsStorage( 'tokens', `${hashToken(hookData.token)}.json` ); - // When the claim is absent, the event log is the only durable source - // that can distinguish a first hook from a crash-lost token cache. - if (!(await readHookTokenClaim(constraintPath))) { - await rebuildLiveHookByTokenFromEventLog( - basedir, - hookData.token, - tag - ); - } // Persist `eventId` in the claim so concurrent / cross- // process retries can converge on a single canonical // `hook_created` event path. See the recovery comment // below. - const tokenClaimed = await writeExclusive( - constraintPath, - JSON.stringify({ - token: hookData.token, - hookId: data.correlationId, - runId: effectiveRunId, - eventId, - }) - ); + const claimContent = JSON.stringify({ + token: hookData.token, + hookId: data.correlationId, + runId: effectiveRunId, + eventId, + }); + + // Bounded claim loop. A same-token claim can be observed in the + // short window where its release is already committed but the + // claim file itself is still on disk (or was just deleted): + // + // - the claim vanished between the exclusive-create attempt + // and the ownership read → retry immediately; + // - the claim is held by a disposed hook or a terminal/missing + // run (`isHookTokenClaimReleasable`) → wait briefly for the + // in-flight releaser (the hook_disposed handler or the + // run-completion cleanup) to delete it, then reclaim; if it + // never does (the releaser crashed after committing its + // dispose lock / terminal status), release the stale claim + // ourselves and reclaim. + // + // Without this, a run claiming a token right after the previous + // owner disposed it records a durable, spurious hook_conflict + // (issue #2778). A genuinely live claim exits the loop on the + // first iteration and falls through to the dedup / conflict + // handling below. + let tokenClaimed = false; + let existingClaim: z.infer | null = null; + let releasableObservations = 0; + for (let attempt = 0; attempt < 10; attempt++) { + // When the claim is absent, the event log is the only durable + // source that can distinguish a first hook from a crash-lost + // token cache. + if (!(await readHookTokenClaim(constraintPath))) { + await rebuildLiveHookByTokenFromEventLog( + basedir, + hookData.token, + tag + ); + } + tokenClaimed = await writeExclusive(constraintPath, claimContent); + if (tokenClaimed) break; + + existingClaim = await readHookTokenClaim(constraintPath); + if (!existingClaim) { + continue; + } + if ( + existingClaim.runId === effectiveRunId && + existingClaim.hookId === data.correlationId + ) { + // Our own prior claim — dedup-recovery handling below. + break; + } + if ( + !(await isHookTokenClaimReleasable(basedir, existingClaim, tag)) + ) { + // Genuinely live claim — conflict handling below. + break; + } + releasableObservations++; + if (releasableObservations < 3) { + // Give the in-flight releaser a moment to delete the claim. + await new Promise((resolve) => setTimeout(resolve, 10)); + } else { + // The releaser is not coming. Release the stale claim and + // the hook entity it points at, mirroring the hook_disposed + // cleanup. + await deleteJSON(constraintPath); + if (existingClaim.hookId) { + await deleteJSON( + taggedPath(basedir, 'hooks', existingClaim.hookId, tag) + ); + await deleteJSON( + hookRecoveryMarkerPath( + basedir, + hookData.token, + existingClaim.runId, + existingClaim.hookId + ) + ); + } + } + existingClaim = null; + } // Recovery shape: the durable record of a successful hook // creation is the `hook_created` event in the event log. The @@ -1624,8 +1733,6 @@ export function createEventsStorage( let writeHookEntityWithOverwrite = false; if (!tokenClaimed) { - const existingClaim = await readHookTokenClaim(constraintPath); - if ( existingClaim?.runId === effectiveRunId && existingClaim.hookId === data.correlationId @@ -1814,14 +1921,14 @@ export function createEventsStorage( // hook_disposed: Deletes hook entity, rejects duplicates. // Uses writeExclusive on a lock file to atomically prevent concurrent // invocations from both disposing the same hook (TOCTOU race). - const hookLockName = tag - ? `${data.correlationId}.disposed.${tag}` - : `${data.correlationId}.disposed`; - const lockPath = resolveWithinBase( + // The lock doubles as the durable disposal marker consulted by the + // hook_created claim path and the event-log rebuild (see + // `isHookDisposalCommitted`), which is why it must be written + // before any of the destructive deletes below. + const lockPath = hookDisposeLockPath( basedir, - '.locks', - 'hooks', - hookLockName + data.correlationId, + tag ); const claimed = await writeExclusive(lockPath, ''); if (!claimed) { diff --git a/packages/world-local/src/storage/helpers.ts b/packages/world-local/src/storage/helpers.ts index 23156df992..95b27fa4a7 100644 --- a/packages/world-local/src/storage/helpers.ts +++ b/packages/world-local/src/storage/helpers.ts @@ -1,7 +1,8 @@ import { createHash } from 'node:crypto'; +import fs from 'node:fs/promises'; import path from 'node:path'; import { monotonicFactory } from 'ulid'; -import { stripTag, ulidToDate } from '../fs.js'; +import { resolveWithinBase, stripTag, ulidToDate } from '../fs.js'; /** * Hash a hook token to produce a filesystem-safe constraint filename. @@ -10,6 +11,47 @@ export function hashToken(token: string): string { return createHash('sha256').update(token).digest('hex'); } +/** + * Path of the exclusive-create lock file that commits a hook's disposal. + * The `hook_disposed` handler writes this lock BEFORE deleting the token + * claim and hook entity (and before appending the event to the log), so + * its existence is the earliest durable evidence that the hook can never + * be live again. + */ +export function hookDisposeLockPath( + basedir: string, + hookId: string, + tag?: string +): string { + const name = tag ? `${hookId}.disposed.${tag}` : `${hookId}.disposed`; + return resolveWithinBase(basedir, '.locks', 'hooks', name); +} + +/** + * Whether a hook's disposal has been committed (its dispose lock exists). + * Mirrors event visibility for tagged worlds: an untagged lock is visible + * to every tag, a tagged lock only to its own tag. + */ +export async function isHookDisposalCommitted( + basedir: string, + hookId: string, + tag?: string +): Promise { + const candidates = [hookDisposeLockPath(basedir, hookId)]; + if (tag) { + candidates.push(hookDisposeLockPath(basedir, hookId, tag)); + } + for (const lockPath of candidates) { + try { + await fs.access(lockPath); + return true; + } catch { + // lock not present at this path + } + } + return false; +} + /** * Compute the path of the recovery-marker sidecar for a specific * `(token, runId, hookId)` triple. Identity is encoded in the diff --git a/packages/world-local/src/storage/hooks-storage.ts b/packages/world-local/src/storage/hooks-storage.ts index ad94af25ae..3c449b5bfe 100644 --- a/packages/world-local/src/storage/hooks-storage.ts +++ b/packages/world-local/src/storage/hooks-storage.ts @@ -32,7 +32,11 @@ import { writeExclusive, } from '../fs.js'; import { filterHookData } from './filters.js'; -import { hashToken, hookRecoveryMarkerPath } from './helpers.js'; +import { + hashToken, + hookRecoveryMarkerPath, + isHookDisposalCommitted, +} from './helpers.js'; function isVisibleToTag(fileId: string, tag: string | undefined): boolean { return tag ? isUntagged(fileId) || hasTag(fileId, tag) : isUntagged(fileId); @@ -147,6 +151,18 @@ async function findLiveHookCreatedEvent( return null; } + // A committed disposal (dispose lock on disk) closes the hook even when + // its `hook_disposed` event has not landed in the log yet — the disposer + // writes the lock, releases the token claim and hook entity, and only + // then appends the event. Rebuilding the caches from the log in that + // window would resurrect a claim for a hook that is being torn down. + if ( + liveEvent && + (await isHookDisposalCommitted(basedir, liveEvent.correlationId, tag)) + ) { + return null; + } + return liveEvent; } diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index 304578eebb..5750636a82 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -911,6 +911,39 @@ export async function hookDisposeTestWorkflow( ////////////////////////////////////////////////////////// +/** + * Workflow for testing same-run hook token recreation: each iteration + * recreates a hook with the same token after disposing the previous one. + * + * Regression workflow for issue #2777 — the disposal must be flushed + * before the next hook's creation is validated, otherwise the second + * round records a spurious conflict against the run's own disposed hook. + */ +export async function hookTokenReuseLoopWorkflow( + token: string, + rounds: number +) { + 'use workflow'; + + const received: string[] = []; + for (let round = 0; round < rounds; round++) { + const hook = createHook<{ message: string }>({ token }); + + const conflict = await hook.getConflict(); + if (conflict) { + return { received, conflictRound: round }; + } + + const payload = await hook; + received.push(payload.message); + hook.dispose(); + } + + return { received, conflictRound: null }; +} + +////////////////////////////////////////////////////////// + export async function stepFunctionPassingWorkflow() { 'use workflow'; // Pass a step function reference to another step (without closure vars) diff --git a/workbench/vitest/test/hook-token-reuse.test.ts b/workbench/vitest/test/hook-token-reuse.test.ts new file mode 100644 index 0000000000..966f4fea9b --- /dev/null +++ b/workbench/vitest/test/hook-token-reuse.test.ts @@ -0,0 +1,55 @@ +import { waitForHook } from '@workflow/vitest'; +import { describe, expect, it } from 'vitest'; +import type { Run } from 'workflow/api'; +import { resumeHook, start } from 'workflow/api'; +import { + claimTokenOnceWorkflow, + reuseHookTokenWorkflow, +} from '../workflows/hook-token-reuse.js'; + +describe('hook token reuse after dispose', () => { + // Issue #2777: dispose() followed by createHook() with the same token in + // the same run must not conflict with the run's own disposed hook. + it('same run can recreate a hook with the same token after dispose()', async () => { + const token = `same-run-reuse-${Math.random().toString(36).slice(2)}`; + const rounds = 3; + const run = await start(reuseHookTokenWorkflow, [token, rounds]); + + for (let round = 0; round < rounds; round++) { + const settled = await Promise.race([ + waitForHook(run, { token }).then(() => 'hook' as const), + run.returnValue.then((value) => ({ value })), + ]); + expect(settled, `round ${round} should register a hook`).toBe('hook'); + await resumeHook(token, { n: round }); + } + + await expect(run.returnValue).resolves.toBe('ok'); + }, 60_000); + + // Issue #2778: when run A disposes its hook and completes, run B claiming + // the same token immediately afterwards must not conflict against run A. + it('next run can claim the token right after the previous run disposed it', async () => { + const token = `handoff-reuse-${Math.random().toString(36).slice(2)}`; + const rounds = 5; + const runs: Run[] = []; + + for (let round = 0; round < rounds; round++) { + const run = await start(claimTokenOnceWorkflow, [token]); + runs.push(run); + + const settled = await Promise.race([ + waitForHook(run, { token }).then(() => 'hook' as const), + run.returnValue.then((value) => ({ value })), + ]); + expect(settled, `round ${round} should register a hook`).toBe('hook'); + + // Resume the hook and immediately start the next claimant without + // waiting for this run to settle (the "fast handoff" timing). + await resumeHook(token, { n: round }); + } + + const results = await Promise.all(runs.map((run) => run.returnValue)); + expect(results).toEqual(Array(rounds).fill('ok')); + }, 60_000); +}); diff --git a/workbench/vitest/workflows/hook-token-reuse.ts b/workbench/vitest/workflows/hook-token-reuse.ts new file mode 100644 index 0000000000..e5c669493a --- /dev/null +++ b/workbench/vitest/workflows/hook-token-reuse.ts @@ -0,0 +1,43 @@ +import { createHook } from 'workflow'; + +/** + * Regression workflow for issue #2777: recreating a hook with the same + * token after dispose() within a single run must not self-conflict. + */ +export async function reuseHookTokenWorkflow(token: string, rounds: number) { + 'use workflow'; + + for (let round = 0; round < rounds; round++) { + const hook = createHook<{ n: number }>({ token }); + + const conflict = await hook.getConflict(); + if (conflict !== null) { + return `conflict-round-${round}:${conflict.runId}`; + } + + await hook; + hook.dispose(); + } + + return 'ok'; +} + +/** + * Regression workflow for issue #2778: a run that claims a token, receives + * one payload, disposes, and completes must release the token claim so the + * next run can immediately claim the same token. + */ +export async function claimTokenOnceWorkflow(token: string) { + 'use workflow'; + + const hook = createHook<{ n: number }>({ token }); + + const conflict = await hook.getConflict(); + if (conflict !== null) { + return `conflict:${conflict.runId}`; + } + + await hook; + hook.dispose(); + return 'ok'; +} From 1c07d7d29aba4fac884ab94c8df65dcb2a262a21 Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:09:15 -0700 Subject: [PATCH 5/8] fix(workbench): add @repo/* tsconfig alias to nitro-v2 (#2787) The workbench workflows are shared across apps (nitro-v2/workflows symlinks into files shared with workbench/example), and the shared 99_e2e.ts workflow imports @repo/lib/steps/paths-alias-test. nitro-v2's tsconfig was missing the @repo/* path alias that nitro-v3 already has, so building nitro-v2 with the Vercel preset failed at esbuild resolution: ../example/workflows/99_e2e.ts: ERROR: Could not resolve "@repo/lib/steps/paths-alias-test" Mirror nitro-v3's alias. Verified NITRO_PRESET=vercel pnpm build now succeeds (was failing on main before this change). --- workbench/nitro-v2/tsconfig.json | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/workbench/nitro-v2/tsconfig.json b/workbench/nitro-v2/tsconfig.json index 6eaef8b8de..39c2f1ea68 100644 --- a/workbench/nitro-v2/tsconfig.json +++ b/workbench/nitro-v2/tsconfig.json @@ -14,7 +14,8 @@ "jsx": "preserve", "incremental": true, "paths": { - "@/*": ["./*"] + "@/*": ["./*"], + "@repo/*": ["../../*"] }, "plugins": [ { From cdfac39e07e2d15c95f7ab9ff8d244b8677931bd Mon Sep 17 00:00:00 2001 From: Karthik Kalyan <105607645+karthikscale3@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:47:51 -0700 Subject: [PATCH 6/8] web: construct worlds explicitly instead of via static-injection stub (#2804) * web: construct worlds explicitly instead of via static-injection stub Since #2752, createWorld() from @workflow/core/runtime is a stub that throws unless a framework build plugin aliases it to a world package. The CLI was migrated to construct worlds directly, but @workflow/web still called the stub, so 'workflow web' crashed with 'Workflow target world was not statically injected' for local and postgres backends (the vercel path was unaffected since it constructs the world directly). Mirror the CLI: import the local world statically and resolve any other configured world package from the inspected project's directory. Co-Authored-By: Claude Fable 5 * web: fall back to default.createWorld for CJS world packages Review feedback: import() of a require-resolved CJS entry relies on cjs-module-lexer to surface named exports; fall back to mod.default.createWorld when detection fails. Mirrored in the CLI in #2806. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- .../web-local-world-static-injection.md | 5 ++ .../server/workflow-server-actions.server.ts | 50 ++++++++++++++++++- packages/web/package.json | 1 + pnpm-lock.yaml | 3 ++ 4 files changed, 57 insertions(+), 2 deletions(-) create mode 100644 .changeset/web-local-world-static-injection.md diff --git a/.changeset/web-local-world-static-injection.md b/.changeset/web-local-world-static-injection.md new file mode 100644 index 0000000000..9354021448 --- /dev/null +++ b/.changeset/web-local-world-static-injection.md @@ -0,0 +1,5 @@ +--- +'@workflow/web': patch +--- + +Fix `workflow web` for local and postgres backends after static world-target injection. The web server now constructs the local world directly and resolves other world packages from the inspected project, instead of calling the `createWorld()` static-injection stub (which throws when no build plugin aliased it). diff --git a/packages/web/app/server/workflow-server-actions.server.ts b/packages/web/app/server/workflow-server-actions.server.ts index 9262e5f66f..a7e0a30a50 100644 --- a/packages/web/app/server/workflow-server-actions.server.ts +++ b/packages/web/app/server/workflow-server-actions.server.ts @@ -6,9 +6,10 @@ */ import fs from 'node:fs/promises'; +import { createRequire } from 'node:module'; import path from 'node:path'; +import { pathToFileURL } from 'node:url'; import * as workflowRunHelpers from '@workflow/core/runtime'; -import { createWorld } from '@workflow/core/runtime'; import { type HealthCheckEndpoint, type HealthCheckResult, @@ -27,6 +28,7 @@ import type { WorkflowRunStatus, World, } from '@workflow/world'; +import { createWorld as createLocalWorld } from '@workflow/world-local'; import { createVercelWorld } from '@workflow/world-vercel'; import type { HookListItem, HookTokenResult } from '~/lib/types'; @@ -473,11 +475,55 @@ async function getWorldFromEnv(userEnvMap: EnvMap): Promise { return cachedWorld; } - const world = await createWorld(); + const world = await createWorldForBackend(backendId); worldCache.set(cacheKey, world); return world; } +/** + * Construct the world for the configured backend explicitly. + * + * `createWorld()` from `@workflow/core/runtime` is a static-injection stub: + * framework build plugins alias it to the selected world package when the + * user's app is built. The web UI selects its backend at runtime via env, + * so it must construct worlds directly (mirroring the CLI's world setup). + * The local world is a direct dependency; other world packages (e.g. + * `@workflow/world-postgres` or community worlds) are resolved from the + * inspected project's directory. + */ +async function createWorldForBackend(backendId: string): Promise { + if (backendId === 'local' || backendId === '@workflow/world-local') { + return createLocalWorld(); + } + + const cwd = getObservabilityCwd(); + let worldPath: string; + try { + worldPath = createRequire(path.join(cwd, 'package.json')).resolve( + backendId, + { paths: [cwd] } + ); + } catch { + throw new Error( + `Could not resolve workflow backend package "${backendId}" from "${cwd}". ` + + 'Make sure the package is installed in the inspected project.' + ); + } + const mod = (await import(pathToFileURL(worldPath).href)) as { + createWorld?: () => World | Promise; + default?: { createWorld?: () => World | Promise }; + }; + // Fall back to default.createWorld for CJS packages whose named exports + // aren't statically detectable by cjs-module-lexer. + const createWorldFn = mod.createWorld ?? mod.default?.createWorld; + if (typeof createWorldFn !== 'function') { + throw new Error( + `Workflow backend package "${backendId}" does not export createWorld().` + ); + } + return createWorldFn(); +} + /** * Creates a structured error object from a caught error */ diff --git a/packages/web/package.json b/packages/web/package.json index a366a94b5a..3c019166a3 100644 --- a/packages/web/package.json +++ b/packages/web/package.json @@ -36,6 +36,7 @@ "test": "vitest run" }, "dependencies": { + "@workflow/world-local": "workspace:*", "express": "^5.2.1" }, "devDependencies": { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 89aa0af7a6..daf57cdcb4 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1037,6 +1037,9 @@ importers: packages/web: dependencies: + '@workflow/world-local': + specifier: workspace:* + version: link:../world-local express: specifier: ^5.2.1 version: 5.2.1 From 54f46f976da8d8d5a646bceab60cfab7f0ae47e9 Mon Sep 17 00:00:00 2001 From: Karthik Kalyan <105607645+karthikscale3@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:47:58 -0700 Subject: [PATCH 7/8] cli: restore dynamic world loading for community backends (#2806) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * cli: restore dynamic world loading for community backends Since #2752 moved world selection to static injection, setupCliWorld only constructed the vercel, local, and postgres worlds explicitly and threw 'Unsupported workflow backend' for anything else — breaking community worlds (e.g. @workflow-worlds/turso) that previously loaded through the dynamic createWorld() in @workflow/core/runtime. Generalize the postgres-only dynamic path: any backend other than vercel/local is now resolved from the user's project directory and loaded via its createWorld() export, matching @workflow/web's world construction (#2804), with clear errors when the package is missing or does not export createWorld(). Co-Authored-By: Claude Fable 5 * cli: fall back to default.createWorld for CJS world packages Review feedback on #2804: import() of a require-resolved CJS entry relies on cjs-module-lexer to surface named exports; fall back to mod.default.createWorld when detection fails. Co-Authored-By: Claude Fable 5 * Update .changeset/cli-generic-world-backends.md Signed-off-by: Peter Wielander --------- Signed-off-by: Peter Wielander Co-authored-by: Claude Fable 5 Co-authored-by: Peter Wielander --- .changeset/cli-generic-world-backends.md | 5 +++ packages/cli/src/lib/inspect/setup.ts | 46 +++++++++++++++++------- 2 files changed, 38 insertions(+), 13 deletions(-) create mode 100644 .changeset/cli-generic-world-backends.md diff --git a/.changeset/cli-generic-world-backends.md b/.changeset/cli-generic-world-backends.md new file mode 100644 index 0000000000..c5106ce6dc --- /dev/null +++ b/.changeset/cli-generic-world-backends.md @@ -0,0 +1,5 @@ +--- +'@workflow/cli': patch +--- + +Ensure the CLI can resolve community world packages if not statically injected. Fixes "Unsupported workflow backend" error. diff --git a/packages/cli/src/lib/inspect/setup.ts b/packages/cli/src/lib/inspect/setup.ts index b263e23a73..924fc37bd6 100644 --- a/packages/cli/src/lib/inspect/setup.ts +++ b/packages/cli/src/lib/inspect/setup.ts @@ -131,17 +131,13 @@ export const setupCliWorld = async ( teamId: vercelEnvVars.teamId, }, }); - } else if (flags.backend === '@workflow/world-postgres') { - world = await createPostgresCliWorld(); } else if ( flags.backend === 'local' || flags.backend === '@workflow/world-local' ) { world = createLocalWorld(); } else { - throw new Error( - `Unsupported workflow backend for CLI world initialization: ${flags.backend}` - ); + world = await createDynamicCliWorld(flags.backend); } // Store in the global cache so BaseCommand.finally() can find and close it. @@ -149,13 +145,37 @@ export const setupCliWorld = async ( return world; }; -async function createPostgresCliWorld(): Promise { - const postgresWorldPackage = '@workflow/world-postgres'; - const postgresWorldPath = createRequire( - join(process.cwd(), 'package.json') - ).resolve(postgresWorldPackage, { paths: [process.cwd()] }); - const mod = (await import(pathToFileURL(postgresWorldPath).href)) as { - createWorld: () => World | Promise; +/** + * Construct a world from its package (e.g. `@workflow/world-postgres` or a + * community world), resolved from the user's project directory. The vercel + * and local worlds are direct dependencies and constructed explicitly above; + * every other backend is loaded dynamically so third-party worlds keep + * working without being dependencies of the CLI. + */ +async function createDynamicCliWorld(backendId: string): Promise { + const cwd = process.cwd(); + let worldPath: string; + try { + worldPath = createRequire(join(cwd, 'package.json')).resolve(backendId, { + paths: [cwd], + }); + } catch { + throw new Error( + `Could not resolve workflow backend package "${backendId}" from "${cwd}". ` + + 'Make sure the package is installed in your project.' + ); + } + const mod = (await import(pathToFileURL(worldPath).href)) as { + createWorld?: () => World | Promise; + default?: { createWorld?: () => World | Promise }; }; - return mod.createWorld(); + // Fall back to default.createWorld for CJS packages whose named exports + // aren't statically detectable by cjs-module-lexer. + const createWorldFn = mod.createWorld ?? mod.default?.createWorld; + if (typeof createWorldFn !== 'function') { + throw new Error( + `Workflow backend package "${backendId}" does not export createWorld().` + ); + } + return createWorldFn(); } From 2ca34ac69c5c201ef85a61fe3a10cc75ca3c22c4 Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Tue, 7 Jul 2026 16:07:13 -0700 Subject: [PATCH 8/8] fix(sveltekit): production server crash from bundled TypeScript compiler (#2799) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(sveltekit): patch server chunks with rollup-renamed __filename bindings The adapter-node chunk patch skipped any chunk matching /\b(const|let|var)\s+__(file|dir)name\b/ — but $ is not a regex word character, so rollup-renamed declarations like `__filename$1` (produced when adapter-node re-bundles the intermediate server output and our banner's declaration collides) satisfied the check. Chunks that declared only a renamed binding while a bundled CJS dependency referenced the bare `__filename` were skipped, and the production server crashed at boot (observed on main with the TypeScript compiler bundled via cosmiconfig through @workflow/world-postgres). Anchor both regexes with (?![\w$]) so renamed identifiers no longer match. Verified: the sveltekit workbench production server now boots and serves health checks (with and without a base path), and queue deliveries from start() succeed. Signed-off-by: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> * fix(sveltekit): keep the TypeScript compiler out of the server bundle Since the world-target injection change, the sveltekit workbench's hooks.server.ts imports @workflow/world-postgres, whose dependency chain (graphile-worker -> cosmiconfig) reaches cosmiconfig's TS-config loader. At runtime that loader's require('typescript') is lazy and never fires, but SvelteKit bundles the whole chain into the server and rollup's CJS conversion hoists it into an eager top-level evaluation — executing the entire TypeScript compiler at boot and crashing the server ("__filename is not defined" inside the bundled compiler). Alias 'typescript' to a stub module in the SvelteKit plugin, following the existing pg-native pattern. Server output shrinks from 36MB to 11MB and boots cleanly. Verified: sveltekit workbench production build boots, serves flow?__health (200), and start() runs execute with clean queue deliveries. The chunk-patch regex fix from the previous commit stays as hardening for any other CJS dependency that references __filename. Signed-off-by: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> * Update .changeset/fix-sveltekit-filename-chunk-patch.md Signed-off-by: Peter Wielander --------- Signed-off-by: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Signed-off-by: Peter Wielander Co-authored-by: Peter Wielander --- .changeset/fix-sveltekit-filename-chunk-patch.md | 6 ++++++ packages/builders/src/index.ts | 1 + .../builders/src/optional-typescript-alias.ts | 5 +++++ packages/builders/src/optional-typescript.ts | 9 +++++++++ packages/sveltekit/src/plugin.ts | 16 ++++++++++++++-- 5 files changed, 35 insertions(+), 2 deletions(-) create mode 100644 .changeset/fix-sveltekit-filename-chunk-patch.md create mode 100644 packages/builders/src/optional-typescript-alias.ts create mode 100644 packages/builders/src/optional-typescript.ts diff --git a/.changeset/fix-sveltekit-filename-chunk-patch.md b/.changeset/fix-sveltekit-filename-chunk-patch.md new file mode 100644 index 0000000000..fd2b574b63 --- /dev/null +++ b/.changeset/fix-sveltekit-filename-chunk-patch.md @@ -0,0 +1,6 @@ +--- +"@workflow/sveltekit": patch +"@workflow/builders": patch +--- + +Fix SvelteKit production server crash at boot if a world package pulls cosmiconfig into the server bundle. diff --git a/packages/builders/src/index.ts b/packages/builders/src/index.ts index 209bb7a15d..7ea1905dcf 100644 --- a/packages/builders/src/index.ts +++ b/packages/builders/src/index.ts @@ -33,6 +33,7 @@ export { } from './node-compat-banner.js'; export { createNodeModuleErrorPlugin } from './node-module-esbuild-plugin.js'; export { WORKFLOW_OPTIONAL_PG_NATIVE_ALIAS } from './optional-pg-native-alias.js'; +export { WORKFLOW_OPTIONAL_TYPESCRIPT_ALIAS } from './optional-typescript-alias.js'; export { createPseudoPackagePlugin, PSEUDO_PACKAGES, diff --git a/packages/builders/src/optional-typescript-alias.ts b/packages/builders/src/optional-typescript-alias.ts new file mode 100644 index 0000000000..7ab7547ce1 --- /dev/null +++ b/packages/builders/src/optional-typescript-alias.ts @@ -0,0 +1,5 @@ +import { fileURLToPath } from 'node:url'; + +export const WORKFLOW_OPTIONAL_TYPESCRIPT_ALIAS = fileURLToPath( + new URL('./optional-typescript.js', import.meta.url) +); diff --git a/packages/builders/src/optional-typescript.ts b/packages/builders/src/optional-typescript.ts new file mode 100644 index 0000000000..b8261b2ea8 --- /dev/null +++ b/packages/builders/src/optional-typescript.ts @@ -0,0 +1,9 @@ +// Stub aliased in place of the `typescript` package in framework server +// bundles. It is only reachable through cosmiconfig's TS-config loader +// (via world packages -> graphile-worker), where `require('typescript')` +// is lazy and never fires at runtime — but bundling converts it into an +// eager top-level evaluation, pulling the entire compiler into the server +// output and executing it at boot. +const typescript = {}; + +export default typescript; diff --git a/packages/sveltekit/src/plugin.ts b/packages/sveltekit/src/plugin.ts index 4d6a00a329..b8f9ad1bb8 100644 --- a/packages/sveltekit/src/plugin.ts +++ b/packages/sveltekit/src/plugin.ts @@ -7,6 +7,7 @@ import { WORKFLOW_NODE_COMPAT_BANNER, WORKFLOW_NODE_FILENAME_BANNER, WORKFLOW_OPTIONAL_PG_NATIVE_ALIAS, + WORKFLOW_OPTIONAL_TYPESCRIPT_ALIAS, WORKFLOW_WORLD_TARGET_MODULE, } from '@workflow/builders'; import { workflowTransformPlugin } from '@workflow/rollup'; @@ -48,6 +49,7 @@ export function workflowPlugin(options: WorkflowPluginOptions = {}): Plugin[] { alias: { [WORKFLOW_WORLD_TARGET_MODULE]: workflowTargetWorldAlias, 'pg-native': WORKFLOW_OPTIONAL_PG_NATIVE_ALIAS, + typescript: WORKFLOW_OPTIONAL_TYPESCRIPT_ALIAS, }, }, }; @@ -143,9 +145,19 @@ function patchAdapterNodeServerChunks(cwd: string): void { // that already declares either identifier (const/let/var, e.g. a // CJS-interop shim) would produce a duplicate top-level declaration // and crash the server with a SyntaxError at startup. - const referencesFilename = /\b__(?:file|dir)name\b/.test(source); + // + // The `(?![\w$])` lookaheads matter: when adapter-node re-bundles the + // intermediate output, rollup renames colliding declarations to e.g. + // `__filename$1`. A bare `\b` boundary matches those ($ is not a word + // character), which made this skip chunks that declare only a RENAMED + // binding while still referencing the bare `__filename` (observed with + // the bundled `typescript` compiler pulled in via cosmiconfig — the + // server crashed at boot with "__filename is not defined"). + const referencesFilename = /(?