From 459e34b7f048e28d8923d4c133449b1f5a9a862f Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:19:32 -0700 Subject: [PATCH] feat(world): expose capabilities in health checks --- .changeset/capabilities-health-schema.md | 7 ++ packages/core/src/runtime/helpers.test.ts | 65 ++++++++--- packages/core/src/runtime/helpers.ts | 102 +++++------------- packages/world/src/capabilities.ts | 43 ++++++++ packages/world/src/index.ts | 2 + packages/world/src/interfaces.ts | 125 +--------------------- 6 files changed, 133 insertions(+), 211 deletions(-) create mode 100644 .changeset/capabilities-health-schema.md create mode 100644 packages/world/src/capabilities.ts diff --git a/.changeset/capabilities-health-schema.md b/.changeset/capabilities-health-schema.md new file mode 100644 index 0000000000..87815b1dce --- /dev/null +++ b/.changeset/capabilities-health-schema.md @@ -0,0 +1,7 @@ +--- +'@workflow/core': minor +'@workflow/world': minor +'workflow': minor +--- + +Export a Zod-backed World capabilities schema and include World capabilities in health-check responses. diff --git a/packages/core/src/runtime/helpers.test.ts b/packages/core/src/runtime/helpers.test.ts index 4623f006a4..615908f548 100644 --- a/packages/core/src/runtime/helpers.test.ts +++ b/packages/core/src/runtime/helpers.test.ts @@ -238,6 +238,37 @@ describe('healthCheck response parsing', () => { expect(result.workflowCoreVersion).toBe('5.0.0-beta.7'); }); + it('surfaces valid World capabilities', async () => { + const world = makeWorldWithResponse( + JSON.stringify({ + healthy: true, + capabilities: { + hookRetention: { active: true }, + futureCapability: true, + }, + }) + ); + + await expect(healthCheck(world, { timeout: 1000 })).resolves.toMatchObject({ + healthy: true, + capabilities: { hookRetention: { active: true } }, + }); + }); + + it('rejects invalid World capabilities', async () => { + const world = makeWorldWithResponse( + JSON.stringify({ + healthy: true, + capabilities: { hookRetention: { active: 'yes' } }, + }) + ); + + const result = await healthCheck(world, { timeout: 20 }); + + expect(result.healthy).toBe(false); + expect(result.error).toMatch(/timed out/); + }); + it('omits workflowCoreVersion when the response does not include the field', async () => { // Independent of specVersion — the field is omitted by any responder // running an older `@workflow/core` that predates the addition of @@ -259,9 +290,7 @@ describe('healthCheck response parsing', () => { expect(result.workflowCoreVersion).toBeUndefined(); }); - it('omits workflowCoreVersion when the field is the wrong type', async () => { - // Defensive: the parser only accepts strings. Anything else is dropped - // rather than surfaced as garbage. + it('rejects workflowCoreVersion with the wrong type', async () => { const world = makeWorldWithResponse( JSON.stringify({ healthy: true, @@ -272,10 +301,10 @@ describe('healthCheck response parsing', () => { }) ); - const result = await healthCheck(world, { timeout: 1000 }); + const result = await healthCheck(world, { timeout: 20 }); - expect(result.healthy).toBe(true); - expect(result.workflowCoreVersion).toBeUndefined(); + expect(result.healthy).toBe(false); + expect(result.error).toMatch(/timed out/); }); it('surfaces hookResumeInputVersion from the target so the caller stamps the consumer value', async () => { @@ -319,9 +348,7 @@ describe('healthCheck response parsing', () => { expect(result.hookResumeInputVersion).toBeUndefined(); }); - it('omits hookResumeInputVersion when the field is the wrong type', async () => { - // Defensive: only a number is accepted; anything else is dropped rather - // than surfaced as a bogus capability. + it('rejects hookResumeInputVersion with the wrong type', async () => { const world = makeWorldWithResponse( JSON.stringify({ healthy: true, @@ -332,10 +359,10 @@ describe('healthCheck response parsing', () => { }) ); - const result = await healthCheck(world, { timeout: 1000 }); + const result = await healthCheck(world, { timeout: 20 }); - expect(result.healthy).toBe(true); - expect(result.hookResumeInputVersion).toBeUndefined(); + expect(result.healthy).toBe(false); + expect(result.error).toMatch(/timed out/); }); it('returns healthy with no fields for non-JSON plain-text responses', async () => { @@ -1214,6 +1241,20 @@ describe('health check run public key', () => { expect(writtenResponse(write).healthy).toBe(true); }); + it('returns the target World capabilities', async () => { + const { getWorldLazy } = await import('./get-world-lazy.js'); + const { world, write } = responderWorld(undefined); + world.capabilities = { hookRetention: { active: true } }; + vi.mocked(getWorldLazy).mockReturnValue(world as any); + + await handleHealthCheckMessage({ + __healthCheck: true, + correlationId: 'corr_capabilities', + }); + + expect(writtenResponse(write).capabilities).toEqual(world.capabilities); + }); + it('omits the key when encryption is not configured', async () => { const { getWorldLazy } = await import('./get-world-lazy.js'); const { world, write } = responderWorld(undefined); diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index 7c8d1cfcee..0331b66bc4 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -23,8 +23,10 @@ import { SPEC_VERSION_CURRENT, SPEC_VERSION_LEGACY, ulidToDate, + WorldCapabilitiesSchema, } from '@workflow/world'; import { monotonicFactory } from 'ulid'; +import { z } from 'zod'; import { runtimeLogger } from '../logger.js'; import { bytesToBase64, deriveRunKeyPair } from '../sealed-box.js'; import { @@ -76,47 +78,29 @@ function getHealthCheckStreamName(correlationId: string): string { return `__health_check__${correlationId}`; } -/** - * Result of a health check operation. - */ -export interface HealthCheckResult { - healthy: boolean; +export const HealthCheckResponseSchema = z.object({ + healthy: z.boolean(), + /** Spec version of the responding deployment. */ + specVersion: z.number().optional(), + /** `@workflow/core` version of the responding deployment. */ + workflowCoreVersion: z.string().optional(), + /** The target run's X25519 public key, encoded as base64. */ + encryptionPublicKey: z.string().optional(), + /** The responding deployment's hook-resume input protocol version. */ + hookResumeInputVersion: z.number().optional(), + /** Optional features supported by the responding deployment's World. */ + capabilities: WorldCapabilitiesSchema.optional(), +}); + +type HealthCheckResponse = z.infer; + +/** Result of a health check operation. */ +export type HealthCheckResult = HealthCheckResponse & { /** Error message if health check failed */ error?: string; /** Latency if the health check was successful */ latencyMs?: number; - /** Spec version of the responding deployment */ - specVersion?: number; - /** - * `@workflow/core` version of the responding deployment, used for - * capability detection (see `getRunCapabilities`). Omitted when the - * responding deployment did not provide the field as a string — - * for example, an older `@workflow/core` that predates this field, - * or a non-JSON plain-text health response. - */ - workflowCoreVersion?: string; - /** - * The target run's X25519 public key (base64), returned only when the probe - * carried a `runId` and the responding deployment has encryption enabled. - * - * Lets a cross-deployment `start()` seal the workflow arguments using a - * response it was already waiting on, instead of making a separate - * key-lookup request. - */ - encryptionPublicKey?: string; - /** - * The responding deployment's `HOOK_RESUME_INPUT_VERSION` — the protocol - * version at which the *consumer* (queue-message target) re-ensures the - * `hook_received` event from `hookInput` on replay. A cross-deployment - * `start()` stamps the *target's* value (not the caller's) into the new - * run's `executionContext.hookResumeInputVersion` so that `resumeHook()` - * only takes the parallel path when the deployment that will actually - * consume the queue message is known to honor `hookInput`. Omitted when the - * responding deployment predates this field (an older consumer that ignores - * `hookInput`), which fails the gate closed. - */ - hookResumeInputVersion?: number; -} +}; /** * Checks if the given message is a health check payload. @@ -194,6 +178,7 @@ export async function handleHealthCheckMessage( // the *consumer's* hook-resume protocol version — exactly what a // cross-deployment caller needs to gate its parallel resume path on. hookResumeInputVersion: HOOK_RESUME_INPUT_VERSION, + capabilities: world.capabilities, ...(encryptionPublicKey ? { encryptionPublicKey } : {}), timestamp: Date.now(), }); @@ -301,12 +286,9 @@ async function readStreamWithTimeout( * Parse and validate a health check response from stream chunks. * Returns the parsed response or null if invalid. */ -function parseHealthCheckResponse(chunks: Uint8Array[]): { - healthy: boolean; - specVersion?: number; - workflowCoreVersion?: string; - encryptionPublicKey?: string; -} | null { +function parseHealthCheckResponse( + chunks: Uint8Array[] +): HealthCheckResponse | null { if (chunks.length === 0) return null; const totalLength = chunks.reduce((sum, chunk) => sum + chunk.length, 0); @@ -331,38 +313,8 @@ function parseHealthCheckResponse(chunks: Uint8Array[]): { return null; } - if ( - typeof response !== 'object' || - response === null || - !('healthy' in response) || - typeof (response as { healthy: unknown }).healthy !== 'boolean' - ) { - return null; - } - - const r = response as Record; - const parsed: { - healthy: boolean; - specVersion?: number; - workflowCoreVersion?: string; - encryptionPublicKey?: string; - hookResumeInputVersion?: number; - } = { - healthy: r.healthy as boolean, - }; - if (typeof r.specVersion === 'number') { - parsed.specVersion = r.specVersion; - } - if (typeof r.workflowCoreVersion === 'string') { - parsed.workflowCoreVersion = r.workflowCoreVersion; - } - if (typeof r.encryptionPublicKey === 'string') { - parsed.encryptionPublicKey = r.encryptionPublicKey; - } - if (typeof r.hookResumeInputVersion === 'number') { - parsed.hookResumeInputVersion = r.hookResumeInputVersion; - } - return parsed; + const result = HealthCheckResponseSchema.safeParse(response); + return result.success ? result.data : null; } export async function healthCheck( diff --git a/packages/world/src/capabilities.ts b/packages/world/src/capabilities.ts new file mode 100644 index 0000000000..fa7a115cd1 --- /dev/null +++ b/packages/world/src/capabilities.ts @@ -0,0 +1,43 @@ +import { z } from 'zod'; + +/** + * Optional features a World implementation supports. Missing capabilities are + * unsupported so runtime behavior always fails closed. + */ +export const WorldCapabilitiesSchema = z.object({ + /** + * Supports `experimental_minRetention` for Hooks. Missing or inactive means + * the runtime rejects retained Hooks before registration. + */ + hookRetention: z.object({ active: z.boolean() }).optional(), + + /** + * Enforces the event count and update-time preconditions on event creation. + * Worlds that accept but ignore either field must leave this unset. + */ + preconditionGuard: z.boolean().optional(), + + /** Supports `maxConcurrency`-limited queue consumption. */ + maxConcurrency: z.boolean().optional(), + + /** + * Deduplicates concurrent `hook_received` writes sharing a + * `(runId, resumeId)` and returns the canonical event to every caller. + */ + hookResumeDedup: z.boolean().optional(), + + /** + * Uses atomic, immutable deployments with strict run affinity. Worlds with + * synthetic or version-tagged deployment ids must leave this unset. + */ + deploymentAffinity: z.boolean().optional(), + + /** + * Allocates dense, slot-numbered event IDs for new runs. Event creation + * advances past occupied slots and returns the skipped events to the writer. + * Existing runs keep their original event ID scheme. + */ + slotEventIds: z.boolean().optional(), +}); + +export type WorldCapabilities = z.infer; diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index 50ce1b9a0a..bc0234703d 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -23,6 +23,8 @@ export { validateAttributeKey, validateAttributeValue, } from './attributes.js'; +export type * from './capabilities.js'; +export { WorldCapabilitiesSchema } from './capabilities.js'; export { _resetEnvWarnCacheForTests, type EnvNumberOptions, diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index 06a49b973e..0d5936540e 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -3,6 +3,7 @@ import type { AttributeChange, ExperimentalSetAttributesResult, } from './attributes.js'; +import type { WorldCapabilities } from './capabilities.js'; import type { CreateEventParams, CreateEventRequest, @@ -317,130 +318,6 @@ export interface Storage { }; } -/** - * Optional feature capabilities a World implementation declares so the core - * runtime can enable optimizations that depend on backend behavior, instead - * of inferring support from environment variables alone. Every capability - * defaults to "unsupported" when absent — runtime fast paths that rely on - * one must fail closed (keep their conservative behavior) unless the World - * explicitly declares it. - */ -export interface WorldCapabilities { - /** - * Supports `experimental_minRetention` for Hooks. Missing or inactive means - * the runtime rejects retained Hooks before registration. - */ - hookRetention?: { - active: boolean; - }; - - /** - * The World enforces the optimistic-concurrency precondition guard: an - * event creation carrying a `stateUpdatedAt` snapshot is rejected with a - * `PreconditionFailedError` (412) when a newer out-of-band event (e.g. a - * received hook) was recorded after that snapshot. Worlds that accept but - * ignore `stateUpdatedAt` must leave this unset so runtime optimizations - * that rely on the 412 fence (see `WORKFLOW_PRECONDITION_GUARD`) are not - * enabled without an actual fence behind them. - * - * A World declaring this should honour the whole snapshot the runtime sends, - * not just the watermark: `stateUpdatedAt`, `stateEventCount` (the count - * fence, which catches an event missing at or below the watermark — the case - * the watermark provably cannot see) and, optionally, `stateCursor` (return - * the missing events on the 412 to save the client a reload). See - * `CreateEventParams` for each field's contract. The runtime does not branch - * on which halves are implemented; a World that ignores the count simply - * fences less. - */ - preconditionGuard?: boolean; - - /** - * The World's queue supports `maxConcurrency`-limited consumption — in - * particular the per-run flow topics consumed with `maxConcurrency: 1` - * that `WORKFLOW_SEQUENTIAL_REPLAYS=1` uses to serialize a run's - * orchestrator invocations. Worlds whose queue has no concurrency-limit - * concept must leave this unset. - * - * Note this declares queue *support*, not deployed configuration: the - * serialization also requires the build-time half (a flow trigger emitted - * with `maxConcurrency: 1`), which a runtime process cannot verify today. - * The core runtime therefore does not yet take any fast path from this - * capability alone — it exists so a future build-verified signal can be - * combined with it (and so Worlds document the contract explicitly). - */ - maxConcurrency?: boolean; - - /** - * The World's `events.create` deduplicates concurrent `hook_received` writes - * that carry the same `(runId, resumeId)` — collapsing them onto a single - * committed event and returning the canonical one to every caller. This is - * the backend half of `resumeHook()`'s parallel fast path: the producer's - * direct write and the queue consumer's re-ensure both write the same - * `resumeId`, and exactly one event must survive or the run replays a - * duplicated `hook_received`. - * - * The core runtime fails closed on this: the parallel path is taken ONLY - * when the World declares `hookResumeDedup === true` AND the target run's - * deployment can re-ensure from `hookInput` (see the execution-context - * marker `hookResumeInputVersion`). A World that accepts a `resumeId` but - * does not enforce the `(runId, resumeId)` constraint must leave this unset - * so the runtime keeps the sequential single-writer path. - * - * Enabled statically for `world-local` (filesystem sidecar claim keyed on - * `(runId, resumeId)`; the adapter and its backend ship together, so a static - * capability can never drift from the backend). `world-vercel` deliberately - * leaves this UNSET and instead attests support per-lookup via the - * server-computed, response-only `Hook.resumeCapabilities.hookResumeDedupVersion` - * (see `HookResumeCapabilitiesSchema`), so a server rollback or kill switch - * degrades new resumes to the sequential path immediately without redeploying - * the adapter. `world-postgres` leaves it unset for now and stays sequential. - * - * The resume gate treats EITHER signal as backend support (see - * `resume-hook.ts`): this static capability OR a current - * `resumeCapabilities.hookResumeDedupVersion` on the by-token hook. - */ - hookResumeDedup?: boolean; - - /** - * Deployments are atomic and immutable: a deployment id names one fixed - * build for its whole lifetime, so a run pinned to one may only execute - * there. Worlds that declare this get the runtime's deployment-affinity - * guard, which re-routes a misrouted delivery to the run's own deployment - * and ultimately fails the run with `DEPLOYMENT_MISMATCH`. - * - * Worlds whose deployment id is synthetic or version-tagged (e.g. - * `dpl_local@`, which legitimately differs across SDK versions - * within one logical environment) must leave this unset: there a - * "mismatch" is not a real cross-deployment delivery, and guarding would - * fail ordinary runs after a version bump. - */ - deploymentAffinity?: boolean; - - /** - * The World allocates **slot-numbered** event ids: `evnt_` plus the event's - * dense, 1-based position in its run's log, zero-padded to 26 characters - * (see `slot-identity.ts`). Two guarantees come with it, and the runtime - * relies on both: - * - * - **Density.** A run's slots are contiguous from 1, so the number of - * events a reader holds *is* the position of the last one. That is what - * makes {@link CreateEventParams.eventCount} a complete statement of the - * writer's snapshot, where the `stateUpdatedAt` / `stateEventCount` - * watermark pair could only approximate it. - * - **Bump and report.** A create never fails because its requested slot is - * taken. The World advances to the next free slot, commits there, and - * returns the events occupying the slots it skipped over on the success - * response (see {@link EventResult.events}). The writer learns its - * snapshot was stale without the write being rejected. - * - * A run's scheme is pinned by the run, not by this flag: it is readable off - * the shape of the run's own first event id, so a World that turns slots on - * keeps replaying its existing ULID-numbered runs unchanged. The capability - * only says what *new* runs get. - */ - slotEventIds?: boolean; -} - /** * The "World" interface represents how Workflows are able to communicate with the outside world. */