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/.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/.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/.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/.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/.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/.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/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/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]; } 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(); } 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/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 = /(?('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 b8dcd36fa3..be17f060cb 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,19 @@ 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, HookListItem } from '~/lib/types'; import { + fetchHookToken, 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 { HookTokenCell } from './display-utils/hook-token-cell'; +import { RelativeTime } from './display-utils/relative-time'; +import { TableSkeleton } from './display-utils/table-skeleton'; interface HooksTableProps { runId?: string; @@ -94,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, @@ -201,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) { @@ -272,7 +278,7 @@ export function HooksTable({ {error ? ( - Error loading hooks + {getErrorTitle(error, 'Error loading hooks')} {getErrorMessage(error)} ) : !loading && (!hooks || hooks.length === 0) ? ( @@ -328,11 +334,11 @@ export function HooksTable({ - - - •••••••••••• - - + {hook.createdAt ? ( @@ -358,7 +364,11 @@ export function HooksTable({ - 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..80c44061f1 100644 --- a/packages/web/app/lib/client/hooks/use-paginated-list.ts +++ b/packages/web/app/lib/client/hooks/use-paginated-list.ts @@ -1,17 +1,23 @@ -import type { Hook, WorkflowRun, WorkflowRunStatus } from '@workflow/world'; +import type { WorkflowRun, WorkflowRunStatus } from '@workflow/world'; import { useCallback, useEffect, useRef, useState } from 'react'; import { unwrapOrThrow, WorkflowWebAPIError, } from '~/lib/client/workflow-errors'; import { fetchHooks, fetchRuns } from '~/lib/rpc-client'; -import type { EnvMap, PaginatedResult } from '~/lib/types'; +import type { + AnalyticsPageInfo, + EnvMap, + HookListItem, + 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 +34,7 @@ export interface PaginatedList { reload: () => void; refresh: () => void; pageInfo: string; + analyticsPageInfo?: AnalyticsPageInfo; } /** @@ -54,7 +61,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 +97,7 @@ export function usePaginatedList( data: cached.data, isLoading: false, error: null, + pageInfo: cached.pageInfo, }); return newMap; }); @@ -99,6 +115,7 @@ export function usePaginatedList( data: result.data, cursor: result.cursor, hasMore: result.hasMore, + pageInfo: result.pageInfo, }); setAllPageResults((prev) => { @@ -107,6 +124,7 @@ export function usePaginatedList( data: result.data, isLoading: false, error: null, + pageInfo: result.pageInfo, }); return newMap; }); @@ -233,6 +251,7 @@ export function usePaginatedList( reload, refresh, pageInfo, + analyticsPageInfo: currentPageResult.pageInfo, }; } @@ -271,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/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/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 9624283a72..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. * @@ -62,10 +64,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 { @@ -82,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 41b2b0c07c..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, @@ -35,6 +36,7 @@ export { } from './client/workflow-actions'; export { getErrorMessage, + getErrorTitle, unwrapServerActionResult, WorkflowWebAPIError, } from './client/workflow-errors'; 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 new file mode 100644 index 0000000000..b6fbf9cc39 --- /dev/null +++ b/packages/web/app/server/workflow-server-actions.server.test.ts @@ -0,0 +1,102 @@ +import type { AnalyticsEvent, Hook } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { + analyticsEventToEvent, + hookToListItem, +} 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, + }, + }); + }); +}); + +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 3bdfe975fb..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, @@ -16,9 +17,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 +28,9 @@ import type { WorkflowRunStatus, World, } from '@workflow/world'; -import { type APIConfig, createVercelWorld } from '@workflow/world-vercel'; +import { createWorld as createLocalWorld } from '@workflow/world-local'; +import { createVercelWorld } from '@workflow/world-vercel'; +import type { HookListItem, HookTokenResult } from '~/lib/types'; /** * Environment variable map for world configuration. @@ -357,6 +361,7 @@ export interface PaginatedResult { data: T[]; cursor?: string; hasMore: boolean; + pageInfo?: AnalyticsPageInfo; } /** @@ -383,6 +388,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. * @@ -447,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 */ @@ -472,7 +544,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 +583,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 +651,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 +715,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 +768,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 +813,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 +838,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 +890,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 +915,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>( @@ -772,6 +932,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: { @@ -780,22 +945,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); + // 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 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 @@ -803,6 +986,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 */ 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/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/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 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/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": [ { 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'; +}