diff --git a/.changeset/analytics-attribute-filters.md b/.changeset/analytics-attribute-filters.md new file mode 100644 index 0000000000..110baf642a --- /dev/null +++ b/.changeset/analytics-attribute-filters.md @@ -0,0 +1,6 @@ +--- +'@workflow/world': minor +'@workflow/world-vercel': minor +--- + +Add `analytics.attributes.list()` for attribute key discovery, and an `attributes` key=value filter to `analytics.runs.list()`. diff --git a/docs/content/docs/v5/api-reference/index.mdx b/docs/content/docs/v5/api-reference/index.mdx index 76dc71cb24..7aa633ce56 100644 --- a/docs/content/docs/v5/api-reference/index.mdx +++ b/docs/content/docs/v5/api-reference/index.mdx @@ -18,7 +18,7 @@ All the functions and primitives that come with Workflow SDK by package. API reference for runtime functions from the `workflow/api` package. - Runtime functions for resolving the World instance and the low-level World SDK. + Runtime functions for resolving the World instance and the low-level World SDK, including storage and analytics queries. Utilities to hydrate step I/O, parse display names, and decrypt workflow data. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/index.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/index.mdx index 042e181c11..5111d655d9 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/index.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/index.mdx @@ -16,7 +16,7 @@ The runtime package provides low-level access to the workflow runtime — resolv Async: resolve the World instance for storage, queuing, and streaming backends. - Low-level API for inspecting runs, steps, events, hooks, streams, and queues. + Low-level API for inspecting runs, steps, events, hooks, streams, and queues, plus metadata-only analytics with attribute search. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/analytics.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/world/analytics.mdx new file mode 100644 index 0000000000..5dfaefc71f --- /dev/null +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/analytics.mdx @@ -0,0 +1,139 @@ +--- +title: Analytics +description: Metadata-only read APIs for runs, steps, events, hooks, waits, and attributes, backed by the observability pipeline. +type: reference +summary: "Interfaces: world.analytics.runs, .attributes, .steps, .events, .hooks, .waits. Metadata-only listings with plan-based lookback windows; filter runs by attribute key=value." +prerequisites: + - /docs/api-reference/workflow-runtime/get-world +related: + - /docs/api-reference/workflow-runtime/world/storage + - /docs/observability/attributes +keywords: + - world.analytics + - analytics.runs + - analytics.attributes + - attribute filter + - listValues + - lookback window + - observability-upgrade-required + - pageInfo + - metadata-only +--- + +`world.analytics` is an optional, read-only namespace for observability surfaces — dashboards, CLIs, and admin tools that list large numbers of runs without touching payload data. + +It differs from [Storage](/docs/api-reference/workflow-runtime/world/storage) in three ways: + +- **Metadata only.** Results never include run input/output, step data, or hook tokens. There is no `resolveData` option. +- **Served from the observability pipeline.** On Vercel, queries hit a ClickHouse-backed analytics store instead of the runtime database, so large listings do not compete with workflow execution. Data is ingested asynchronously and may trail the live state by a few seconds. +- **Plan-bounded lookback.** Listings only cover the observability retention window for your plan (up to 30 days). Every page includes a `pageInfo` block describing the current window, and requesting an older window fails with an `observability-upgrade-required` error. + +The namespace is optional — worlds that don't implement it (such as the local development world) leave it `undefined`, so feature-detect before use: + +```typescript lineNumbers +import { getWorld } from "workflow/runtime"; + +const world = await getWorld(); +if (world.analytics) { // [!code highlight] + const page = await world.analytics.runs.list(); +} +``` + +--- + +## analytics.runs + +### runs.list() + +List runs with metadata, current status, and attributes. Without an explicit time window, the listing defaults to the trailing 24 hours; pass `startTime`/`endTime` to reach older runs within your plan window. + +```typescript lineNumbers +const page = await world.analytics.runs.list({ + workflowName: "orderWorkflow", + status: "failed", + attributes: { source: "checkout" }, // [!code highlight] + pagination: { limit: 50, sortOrder: "desc" }, +}); +``` + +| Parameter | Type | Description | +|-----------|------|-------------| +| `params.workflowName` | `string` | Filter to one workflow | +| `params.status` | `string` | `pending`, `running`, `completed`, `failed`, or `cancelled` | +| `params.startTime` / `params.endTime` | `string` | ISO 8601 window; must be provided together | +| `params.attributes` | `Record` | Only return runs whose latest attributes match every pair (up to 8) | +| `params.pagination` | `PaginationOptions` | Cursor pagination | + +**Returns:** `PaginatedResponse` — each run includes `runId`, `status`, `workflowName`, `deploymentId`, `attributes`, and lifecycle timestamps. + +Attribute matching is latest-write-wins: a run whose attribute moved from `"v1"` to `"v2"` no longer matches `{ key: "v1" }`. Reserved `$`-prefixed keys may be used in filters even though user code cannot write them. + +### runs.get() + +Fetch one run by ID. Point lookups search the full plan window, not just the trailing 24 hours. + +```typescript lineNumbers +const run = await world.analytics.runs.get(runId); +``` + +--- + +## analytics.attributes + +Discover which [attributes](/docs/observability/attributes) exist on your runs — for example to build filter dropdowns over arbitrary user-defined keys. + +### attributes.list() + +List the distinct attribute keys observed on runs in the window, ordered alphabetically. + +```typescript lineNumbers +const page = await world.analytics.attributes.list({ // [!code highlight] + workflowName: "orderWorkflow", +}); +for (const { key, runCount, lastSeenAt } of page.data) { + console.log(key, runCount, lastSeenAt); +} +``` + +| Parameter | Type | Description | +|-----------|------|-------------| +| `params.workflowName` | `string` | Only count runs of one workflow | +| `params.startTime` / `params.endTime` | `string` | ISO 8601 window; must be provided together | +| `params.pagination` | `PaginationOptions` | Cursor pagination | + +**Returns:** `PaginatedResponse` — `{ key, runCount, firstSeenAt, lastSeenAt }` + +--- + +## analytics.steps, analytics.events, analytics.hooks, analytics.waits + +Run-scoped listings mirroring their [Storage](/docs/api-reference/workflow-runtime/world/storage) counterparts, minus payload data: + +```typescript lineNumbers +const steps = await world.analytics.steps.list({ runId }); +const events = await world.analytics.events.list({ runId, eventType: "step_failed" }); +const related = await world.analytics.events.listByCorrelationId({ correlationId }); +const hooks = await world.analytics.hooks.list({ runId }); +const waits = await world.analytics.waits.list({ runId, status: "waiting" }); +``` + +Each namespace also has a `get()` for point lookups (`steps.get(runId, stepId)`, `events.get(runId, eventId)`, `hooks.get(hookId)`, `waits.get(runId, waitId)`). Hook listings never include the hook token — resolve it separately through the runtime APIs if you need to deliver a payload. + +--- + +## Lookback windows and pageInfo + +Every paginated response carries `pageInfo` describing the window the query was allowed to scan: + +{/* @skip-typecheck: shape illustration, not runnable code */} +```typescript +{ + currentLookbackDays: 2, // what your plan allows today + maxLookbackDays: 30, // ceiling with Observability Plus + currentWindowStart: Date, + maxWindowStart: Date, + upgradeAvailable: true, // more history exists behind the plan gate +} +``` + +Requests for a window older than `currentWindowStart` fail with an `observability-upgrade-required` error; windows older than `maxWindowStart` return not-found. Use `pageInfo` to size date pickers and to decide whether to surface an upgrade prompt. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/index.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/world/index.mdx index 5685640d58..aa9d3f4f5e 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/world/index.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/index.mdx @@ -1,6 +1,6 @@ --- title: World SDK -description: Low-level API for inspecting and managing workflow runs, steps, events, hooks, streams, and queues. +description: Low-level API for inspecting and managing workflow runs, steps, events, hooks, streams, and queues, with an analytics namespace for metadata listings and attribute search. type: overview summary: Access workflow infrastructure via await getWorld() for building observability dashboards, admin tools, and custom integrations. prerequisites: @@ -14,7 +14,7 @@ keywords: - workflow management --- -The World SDK provides direct access to workflow infrastructure — runs, steps, events, hooks, streams, and queues. Use it to build observability dashboards, admin panels, debugging tools, and custom workflow management logic. +The World SDK provides direct access to workflow infrastructure — runs, steps, events, hooks, streams, and queues — plus a metadata-only [Analytics](/docs/api-reference/workflow-runtime/world/analytics) namespace with attribute discovery and filtering. Use it to build observability dashboards, admin panels, debugging tools, and custom workflow management logic. ```typescript lineNumbers import { getWorld } from "workflow/runtime"; @@ -28,6 +28,9 @@ const world = await getWorld(); // [!code highlight] Query runs, steps, hooks, and the underlying event log. + + Metadata-only listings with attribute discovery and filtering, built for observability dashboards. + Read, write, and manage real-time data streams for workflow runs. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/meta.json b/docs/content/docs/v5/api-reference/workflow-runtime/world/meta.json index e7508bb033..1b07f13346 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/world/meta.json +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/meta.json @@ -1,4 +1,4 @@ { "title": "World SDK", - "pages": ["storage", "streams", "queue"] + "pages": ["storage", "analytics", "streams", "queue"] } diff --git a/docs/content/docs/v5/observability/attributes.mdx b/docs/content/docs/v5/observability/attributes.mdx index 7e7607561f..d91b4e2d9c 100644 --- a/docs/content/docs/v5/observability/attributes.mdx +++ b/docs/content/docs/v5/observability/attributes.mdx @@ -11,8 +11,7 @@ related: - /docs/api-reference/workflow-errors/workflow-world-error --- -[`setAttributes`](/docs/api-reference/workflow/set-attributes) attaches plaintext string metadata to the current workflow run. These attributes are displayed in observability CLI/UI. -In the future, you'll be able to search and filter runs by attributes. +[`setAttributes`](/docs/api-reference/workflow/set-attributes) attaches plaintext string metadata to the current workflow run. These attributes are displayed in observability CLI/UI, and can be used to search and filter runs through the [Analytics API](/docs/api-reference/workflow-runtime/world/analytics). You can also seed any attributes directly when starting a run: @@ -72,10 +71,33 @@ Expanding an `attr_set` event — in the run sidebar or the Events tab — shows ![Expanded attr_set events showing changes and the writer](/screenshots/attributes/run-details-attr-set-events.png) +## Searching and filtering by attributes + +The [Analytics API](/docs/api-reference/workflow-runtime/world/analytics) can discover which attribute keys exist and filter run listings by them. The `analytics` namespace is optional on `World` — feature-detect it before use; it is absent on local, Postgres, and other custom Worlds: + +```typescript lineNumbers +import { getWorld } from "workflow/runtime"; + +const world = await getWorld(); +if (!world.analytics) { + throw new Error("This World does not support analytics queries"); // [!code highlight] +} + +// Which attribute keys exist, and on how many runs? +const keys = await world.analytics.attributes.list(); + +// List runs whose latest attributes match every pair +const stuck = await world.analytics.runs.list({ + attributes: { phase: "received" }, // [!code highlight] +}); +``` + +Matching is latest-write-wins: once the run above writes `phase: "complete"`, it stops matching `phase: "received"`. + ## Behavior - Attributes require a World implementing spec version 4 or later. - Writes from workflow and step bodies append native `attr_set` events and immediately materialize `run.attributes`. - Storage errors surface rather than being silently ignored: transient errors on workflow-body writes are retried, and a write the World rejects as invalid (for example, exceeding the per-run attribute cap across multiple calls) fails the run with the validation error. - Step-body storage errors throw from `setAttributes` like any other step-side network write. Catch the error inside the step if the attribute is best-effort. -- Reading and querying attributes is not available yet. A query API is planned. +- Reading and querying: each run's current attributes are returned on the run objects from the [Storage](/docs/api-reference/workflow-runtime/world/storage) and [Analytics](/docs/api-reference/workflow-runtime/world/analytics) APIs, and the Analytics API supports discovering attribute keys and filtering run listings by key=value pairs (see [Searching and filtering by attributes](#searching-and-filtering-by-attributes)). On Worlds without the optional `analytics` namespace, attributes are readable on run objects but not searchable. diff --git a/packages/docs-typecheck/src/docs-globals.d.ts b/packages/docs-typecheck/src/docs-globals.d.ts index 72be638ae2..ba5d020ade 100644 --- a/packages/docs-typecheck/src/docs-globals.d.ts +++ b/packages/docs-typecheck/src/docs-globals.d.ts @@ -231,6 +231,34 @@ declare global { name: string ) => Promise<{ tailIndex: number; done: boolean }>; }; + // Metadata-only analytics namespace (optional on World; liberal here + // so reference snippets can call it without a feature-detect guard) + analytics: { + runs: { + get: (...args: any[]) => Promise; + list: (...args: any[]) => Promise; + }; + attributes: { + list: (...args: any[]) => Promise; + }; + steps: { + get: (...args: any[]) => Promise; + list: (...args: any[]) => Promise; + }; + events: { + get: (...args: any[]) => Promise; + list: (...args: any[]) => Promise; + listByCorrelationId: (...args: any[]) => Promise; + }; + hooks: { + get: (...args: any[]) => Promise; + list: (...args: any[]) => Promise; + }; + waits: { + get: (...args: any[]) => Promise; + list: (...args: any[]) => Promise; + }; + }; // Queue methods live directly on world (Queue interface) getDeploymentId: (...args: any[]) => Promise; queue: (...args: any[]) => Promise; @@ -243,6 +271,7 @@ declare global { const streamName: string; const hookId: string; const eventId: string; + const correlationId: string; const cursor: string | undefined; const name: string; const chunk: any; diff --git a/packages/world-vercel/src/analytics.test.ts b/packages/world-vercel/src/analytics.test.ts new file mode 100644 index 0000000000..6888fa2f62 --- /dev/null +++ b/packages/world-vercel/src/analytics.test.ts @@ -0,0 +1,66 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const state = vi.hoisted(() => ({ + makeRequest: vi.fn(), +})); + +vi.mock('./utils.js', () => ({ + makeRequest: state.makeRequest, +})); + +const { createAnalytics } = await import('./analytics.js'); + +describe('createAnalytics attributes', () => { + beforeEach(() => { + state.makeRequest.mockReset(); + state.makeRequest.mockResolvedValue({ + data: [], + cursor: null, + hasMore: false, + }); + }); + + it('serializes attribute filters on runs.list as a JSON query param', async () => { + const analytics = createAnalytics(); + await analytics.runs.list({ + attributes: { team: 'growth', '$eve.type': 'session' }, + }); + + const { endpoint } = state.makeRequest.mock.calls[0][0]; + expect(endpoint).toBe( + `/v2/analytics/runs?attributes=${encodeURIComponent( + JSON.stringify({ team: 'growth', '$eve.type': 'session' }) + )}` + ); + }); + + it('omits the attributes param when the filter object is empty', async () => { + const analytics = createAnalytics(); + await analytics.runs.list({ attributes: {} }); + + const { endpoint } = state.makeRequest.mock.calls[0][0]; + expect(endpoint).toBe('/v2/analytics/runs'); + }); + + it('lists attribute keys with filters and pagination', async () => { + const analytics = createAnalytics(); + await analytics.attributes.list({ + workflowName: 'daily-report', + startTime: '2026-06-20T00:00:00.000Z', + endTime: '2026-06-21T00:00:00.000Z', + pagination: { limit: 25, cursor: 'abc', sortOrder: 'asc' }, + }); + + const { endpoint } = state.makeRequest.mock.calls[0][0]; + const url = new URL(endpoint, 'https://example.test'); + expect(url.pathname).toBe('/v2/analytics/attributes'); + expect(Object.fromEntries(url.searchParams)).toEqual({ + workflowName: 'daily-report', + startTime: '2026-06-20T00:00:00.000Z', + endTime: '2026-06-21T00:00:00.000Z', + limit: '25', + cursor: 'abc', + sortOrder: 'asc', + }); + }); +}); diff --git a/packages/world-vercel/src/analytics.ts b/packages/world-vercel/src/analytics.ts index b896b194f8..92d3b13fb9 100644 --- a/packages/world-vercel/src/analytics.ts +++ b/packages/world-vercel/src/analytics.ts @@ -1,7 +1,9 @@ import { type Analytics, + AnalyticsAttributeKeySchema, AnalyticsEventSchema, AnalyticsHookSchema, + type AnalyticsListAttributesParams, AnalyticsRunSchema, AnalyticsStepSchema, AnalyticsWaitSchema, @@ -25,6 +27,20 @@ function createQueryString(params: URLSearchParams): string { return query ? `?${query}` : ''; } +function appendAttributeListParams( + searchParams: URLSearchParams, + params: AnalyticsListAttributesParams +): void { + if (params.workflowName) { + searchParams.set('workflowName', params.workflowName); + } + if (params.startTime && params.endTime) { + searchParams.set('startTime', params.startTime); + searchParams.set('endTime', params.endTime); + } + appendPagination(searchParams, params.pagination); +} + export function createAnalytics(config?: APIConfig): Analytics { return { runs: { @@ -47,6 +63,12 @@ export function createAnalytics(config?: APIConfig): Analytics { searchParams.set('startTime', params.startTime); searchParams.set('endTime', params.endTime); } + if (params.attributes && Object.keys(params.attributes).length > 0) { + // JSON-encoded rather than repeated key=value pairs: attribute + // keys and values are arbitrary user strings that may themselves + // contain `=` or `,`. + searchParams.set('attributes', JSON.stringify(params.attributes)); + } appendPagination(searchParams, params.pagination); return makeRequest({ @@ -56,6 +78,18 @@ export function createAnalytics(config?: APIConfig): Analytics { }); }, }, + attributes: { + list(params = {}) { + const searchParams = new URLSearchParams(); + appendAttributeListParams(searchParams, params); + + return makeRequest({ + endpoint: `/v2/analytics/attributes${createQueryString(searchParams)}`, + config, + schema: PaginatedResponseSchema(AnalyticsAttributeKeySchema), + }); + }, + }, steps: { get(runId, stepId) { return makeRequest({ diff --git a/packages/world/src/analytics.ts b/packages/world/src/analytics.ts index 5c50066334..7aa800b562 100644 --- a/packages/world/src/analytics.ts +++ b/packages/world/src/analytics.ts @@ -94,11 +94,19 @@ export const AnalyticsWaitSchema = z.object({ workflowEncryptionEnabled: NullableBooleanSchema, }); +export const AnalyticsAttributeKeySchema = z.object({ + key: z.string(), + runCount: z.coerce.number(), + firstSeenAt: z.coerce.date(), + lastSeenAt: z.coerce.date(), +}); + export type AnalyticsRun = z.infer; export type AnalyticsStep = z.infer; export type AnalyticsEvent = z.infer; export type AnalyticsHook = z.infer; export type AnalyticsWait = z.infer; +export type AnalyticsAttributeKey = z.infer; export interface AnalyticsListRunsParams { workflowName?: string; @@ -113,6 +121,27 @@ export interface AnalyticsListRunsParams { */ startTime?: string; endTime?: string; + /** + * Restrict the listing to runs whose latest attribute snapshot matches + * every provided key=value pair (up to 8 pairs). Matching is + * latest-write-wins: a run whose attribute moved from `v1` to `v2` no + * longer matches `v1`. Reserved `$`-prefixed keys may be used in filters + * even though user writes to that namespace are rejected. + */ + attributes?: Record; + pagination?: PaginationOptions; +} + +export interface AnalyticsListAttributesParams { + workflowName?: string; + /** + * Bound the listing to attribute writes between `startTime` and `endTime` + * (ISO 8601 timestamps). Both must be provided together. Requesting a + * window older than the plan's observability lookback fails with + * `observability-upgrade-required`. + */ + startTime?: string; + endTime?: string; pagination?: PaginationOptions; } @@ -148,6 +177,16 @@ export interface Analytics { params?: AnalyticsListRunsParams ): Promise>; }; + attributes: { + /** + * List the distinct attribute keys observed on runs in the window, + * with run counts and first/last seen timestamps. Ordered + * alphabetically by key. + */ + list( + params?: AnalyticsListAttributesParams + ): Promise>; + }; steps: { get(runId: string, stepId: string): Promise; list( diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index cd30180766..5c3b658f14 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -1,5 +1,6 @@ export type * from './analytics.js'; export { + AnalyticsAttributeKeySchema, AnalyticsEventSchema, AnalyticsHookSchema, AnalyticsRunSchema,