Attributes MVP (experimental and write-only)

.changeset/attributes-mvp-plan.md

@@ -0,0 +1,10 @@
+---
+'@workflow/core': patch
+'@workflow/world': patch
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+'@workflow/world-vercel': patch
+'workflow': patch
+---
+
+Add `experimental_setAttributes()` workflow-level helper for attaching string key/value metadata to a workflow run, surfaced as `run.attributes`

docs/content/docs/v4/cookbook/common-patterns/timeouts.mdx

@@ -68,6 +68,9 @@ export async function waitForApproval(requestId: string) {
     throw new Error("Approval request expired after 7 days");
   }
 
+  // You may see warnings like `Workflow run completed with 1 uncommitted operations` in your
+  // logs when the workflow completes. This is expected behavior.
+
   return result.approved;
 }
 ```

packages/core/e2e/e2e.test.ts

@@ -3430,4 +3430,117 @@ describe('e2e', () => {
       expect(returnValue.reason).toBe('Test complete');
     }
   );
+
+  // ==========================================================================
+  // experimental_setAttributes (experimental MVP)
+  // ==========================================================================
+
+  describe('experimental_setAttributes', () => {
+    test(
+      'experimentalSetAttributesWorkflow: workflow-body calls dispatch through the step bridge and merge correctly',
+      { timeout: 30_000 },
+      async () => {
+        const run = await start(
+          await e2e('experimentalSetAttributesWorkflow'),
+          [7]
+        );
+        const output = await run.returnValue;
+        expect(output).toBe(21);
+
+        const world = await getWorld();
+        const persisted = await world.runs.get(run.runId);
+
+        // First call sets {phase: 'init', source: 'workflow-body'}; second
+        // overwrites phase; third unsets source via undefined → null.
+        expect(persisted?.attributes).toEqual({ phase: 'done' });
+        expect(persisted?.attributes ?? {}).not.toHaveProperty('source');
+
+        // Dispatch is via a real step — verify at least one
+        // `step_created`/`step_completed` pair for the `__builtin_set_attributes`
+        // step exists on the run's event log.
+        const { data: events } = await world.events.list({ runId: run.runId });
+        const attrStepEvents = events.filter(
+          (e) =>
+            (e.eventType === 'step_created' ||
+              e.eventType === 'step_completed') &&
+            typeof (e.eventData as { stepName?: string } | undefined)
+              ?.stepName === 'string' &&
+            (e.eventData as { stepName: string }).stepName.includes(
+              '__builtin_set_attributes'
+            )
+        );
+        expect(attrStepEvents.length).toBeGreaterThanOrEqual(2);
+      }
+    );
+
+    // TODO(attributes): un-skip once the platform supports executing
+    // step bodies queued by `drainPendingQueueItems`. Today the step
+    // worker calls `executeStep` → `world.events.create('step_started')`,
+    // which the server rejects with `RunExpiredError` (HTTP 410) once
+    // the run has transitioned to a terminal state. Drain commits the
+    // `step_created` event and enqueues the message, but by the time
+    // the queue worker picks it up `run_completed` has landed and the
+    // worker skips the step ("Workflow run X has already completed,
+    // skipping step Y" in step-executor.ts).
+    //
+    // The fire-and-forget pattern itself works for `void` calls placed
+    // before any later `await` on a runtime primitive (the suspension
+    // queues the step before the run terminates) — see the awaited
+    // workflow-body test above for that coverage. What's broken is
+    // specifically "last void immediately before return". Either the
+    // platform needs to keep accepting `step_started` for steps the
+    // workflow itself queued at drain time, or attribute writes need
+    // a non-step dispatch path (planned for the full V1 attributes
+    // feature where attr_set is a first-class event type).
+    test.todo(
+      'fire-and-forget: void experimental_setAttributes lands without awaiting'
+    );
+
+    test(
+      'Promise.all of disjoint-key writes: every key lands',
+      { timeout: 30_000 },
+      async () => {
+        const run = await start(
+          await e2e('experimentalSetAttributesParallelWorkflow'),
+          []
+        );
+        const output = await run.returnValue;
+        expect(output).toBe('done');
+
+        const world = await getWorld();
+        const persisted = await world.runs.get(run.runId);
+
+        // Disjoint-key writes never collide, so all three keys must
+        // land regardless of dispatch ordering at the world.
+        expect(persisted?.attributes).toEqual({ a: '1', b: '2', c: '3' });
+      }
+    );
+
+    test(
+      'workflow throws after awaited setAttributes: attribute still persists on the failed run',
+      { timeout: 30_000 },
+      async () => {
+        const run = await start(
+          await e2e('experimentalSetAttributesThrowsAfterWorkflow'),
+          []
+        );
+        // The workflow throws — `returnValue` rejects.
+        await expect(run.returnValue).rejects.toThrow(/intentional failure/);
+
+        const world = await getWorld();
+        const persisted = await world.runs.get(run.runId);
+
+        expect(persisted?.status).toBe('failed');
+        // The attribute was awaited and therefore landed before the
+        // throw. The `run_failed` lifecycle write must preserve the
+        // attribute snapshot — the per-run file lock guarantees the
+        // lifecycle handler reads the fresh value off disk inside its
+        // critical section before writing the failed state back.
+        expect(persisted?.attributes).toEqual({
+          phase: 'about-to-fail',
+          reason: 'intentional',
+        });
+      }
+    );
+  });
 });

packages/core/src/index.ts

@@ -25,6 +25,7 @@ export {
   type WebhookOptions,
 } from './create-hook.js';
 export { defineHook, type TypedHook } from './define-hook.js';
+export { experimental_setAttributes } from './set-attributes.js';
 export { sleep } from './sleep.js';
 export {
   getStepMetadata,

packages/core/src/runtime/helpers.ts

@@ -354,7 +354,7 @@ export async function loadWorkflowRunEvents(
         // Preserve the last non-null cursor across pages. A World may
         // legitimately return `{ data: [], cursor: null, hasMore: false }`
         // on a trailing empty page — for example when the previous page's
-        // underlying DynamoDB query hit `Limit` exactly and returned a
+        // underlying DB query hit the limit exactly and returned a
         // `LastEvaluatedKey` "just in case". Overwriting with that null
         // would lose the position past the last real event we loaded and
         // force the runtime into the "no cursor after initial load" full-

packages/core/src/set-attributes.test.ts

@@ -0,0 +1,19 @@
+import { FatalError } from '@workflow/errors';
+import { describe, expect, it } from 'vitest';
+import { experimental_setAttributes } from './set-attributes.js';
+
+describe('experimental_setAttributes (host-side stub)', () => {
+  // The host-side `experimental_setAttributes` is the fallback resolved when callers
+  // are NOT in the workflow VM. The real implementation lives in
+  // `workflow/set-attributes.ts` and is selected via the `workflow`
+  // package-exports condition. Reaching this file from a step body or
+  // plain host code is unsupported and must surface a clear error.
+  it('throws FatalError telling the user experimental_setAttributes is workflow-body only', async () => {
+    await expect(experimental_setAttributes({ phase: 'init' })).rejects.toBeInstanceOf(
+      FatalError
+    );
+    await expect(experimental_setAttributes({ phase: 'init' })).rejects.toThrow(
+      /workflow.*function/i
+    );
+  });
+});

packages/core/src/set-attributes.ts

@@ -0,0 +1,25 @@
+import { FatalError } from '@workflow/errors';
+import type { ExperimentalSetAttributesOptions } from './workflow/set-attributes.js';
+
+export type { ExperimentalSetAttributesOptions };
+
+/**
+ * Host-side stub for `experimental_setAttributes`. The real
+ * implementation lives in `./workflow/set-attributes.ts` and is
+ * selected by the `workflow` package-exports condition when the
+ * workflow VM bundle is resolved.
+ *
+ * Reaching this stub means the function was called outside a workflow
+ * body — most likely from a `'use step'` function or plain host code.
+ * That isn't supported in the MVP: attribute mutations must be
+ * event-sourced through the workflow runtime so they survive replay.
+ */
+export async function experimental_setAttributes(
+  _attrs: Record<string, string | undefined>,
+  _options?: ExperimentalSetAttributesOptions
+): Promise<void> {
+  throw new FatalError(
+    "experimental_setAttributes() must be called from a 'use workflow' function. " +
+      'Calling it from a step body or plain host code is not supported.'
+  );
+}

packages/core/src/workflow.ts

@@ -6,7 +6,6 @@ import {
   WorkflowRuntimeError,
 } from '@workflow/errors';
 import { withResolvers } from '@workflow/utils';
-import { getPortLazy } from './runtime/get-port-lazy.js';
 import { parseWorkflowName } from '@workflow/utils/parse-name';
 import type { Event, WorkflowRun } from '@workflow/world';
 import * as nanoid from 'nanoid';
@@ -16,9 +15,10 @@ import { EventConsumerResult, EventsConsumer } from './events-consumer.js';
 import type { QueueItem } from './global.js';
 import { ENOTSUP, WorkflowSuspension } from './global.js';
 import { runtimeLogger } from './logger.js';
+import type { WorkflowOrchestratorContext } from './private.js';
+import { getPortLazy } from './runtime/get-port-lazy.js';
 import { handleSuspension } from './runtime/suspension-handler.js';
 import { getWorld } from './runtime/world.js';
-import type { WorkflowOrchestratorContext } from './private.js';
 import {
   dehydrateWorkflowReturnValue,
   hydrateWorkflowArguments,
@@ -36,12 +36,12 @@ import * as Attribute from './telemetry/semantic-conventions.js';
 import { trace } from './telemetry.js';
 import { getWorkflowRunStreamId } from './util.js';
 import { createContext } from './vm/index.js';
-import type { WorkflowMetadata } from './workflow/get-workflow-metadata.js';
-import { WORKFLOW_CONTEXT_SYMBOL } from './workflow/get-workflow-metadata.js';
 import {
   createAbortSignalStatics,
   createCreateAbortController,
 } from './workflow/abort-controller.js';
+import type { WorkflowMetadata } from './workflow/get-workflow-metadata.js';
+import { WORKFLOW_CONTEXT_SYMBOL } from './workflow/get-workflow-metadata.js';
 import { createCreateHook } from './workflow/hook.js';
 import { createSleep } from './workflow/sleep.js';
 
@@ -60,6 +60,15 @@ import { createSleep } from './workflow/sleep.js';
  * the abort hook is created but never resumed and the cancellation never
  * reaches the running step.
  *
+ * NOTE: drain only commits the `*_created` events; it does NOT enqueue step
+ * bodies for execution. The platform's step worker rejects `step_started`
+ * for runs that have already transitioned to terminal (`RunExpiredError`),
+ * so a step queued here would be skipped anyway. Fire-and-forget step calls
+ * with side effects therefore work only when followed by some later `await`
+ * on a runtime primitive that triggers a real suspension (the normal
+ * runtime loop in `runtime.ts` queues the step there). A `void` placed
+ * immediately before `return` is not reliably executed.
+ *
  * Drain failures are swallowed: the workflow's own outcome (the user's return
  * value or thrown error) is the source of truth; secondary cleanup that fails
  * shouldn't change the run's terminal state.

packages/core/src/workflow/index.ts

@@ -10,6 +10,7 @@ export {
   type RetryableErrorOptions,
 } from '@workflow/errors';
 export type { Hook, HookOptions } from '../create-hook.js';
+export { experimental_setAttributes } from './set-attributes.js';
 export { sleep } from '../sleep.js';
 export { createHook, createWebhook } from './create-hook.js';
 export { defineHook } from './define-hook.js';

packages/core/src/workflow/set-attributes.test.ts

@@ -0,0 +1,109 @@
+import { FatalError } from '@workflow/errors';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { WORKFLOW_USE_STEP } from '../symbols.js';
+import { experimental_setAttributes } from './set-attributes.js';
+
+describe('workflow.experimental_setAttributes', () => {
+  const dispatchCalls: Array<{
+    stepName: string;
+    changes: Array<{ key: string; value: string | null }>;
+    options: { allowReservedAttributes?: boolean } | undefined;
+  }> = [];
+
+  beforeEach(() => {
+    dispatchCalls.length = 0;
+    (globalThis as Record<symbol, unknown>)[WORKFLOW_USE_STEP] = vi.fn(
+      (stepName: string) =>
+        async (
+          changes: Array<{ key: string; value: string | null }>,
+          options?: { allowReservedAttributes?: boolean }
+        ) => {
+          dispatchCalls.push({ stepName, changes, options });
+        }
+    );
+  });
+
+  afterEach(() => {
+    delete (globalThis as Record<symbol, unknown>)[WORKFLOW_USE_STEP];
+  });
+
+  it('dispatches normalized changes through __builtin_set_attributes', async () => {
+    await experimental_setAttributes({ phase: 'init', orderId: 'ord_1' });
+    expect(dispatchCalls).toEqual([
+      {
+        stepName: '__builtin_set_attributes',
+        changes: [
+          { key: 'phase', value: 'init' },
+          { key: 'orderId', value: 'ord_1' },
+        ],
+        options: {},
+      },
+    ]);
+  });
+
+  it('translates undefined values into null (unset semantics)', async () => {
+    await experimental_setAttributes({ phase: 'done', stale: undefined });
+    expect(dispatchCalls).toEqual([
+      {
+        stepName: '__builtin_set_attributes',
+        changes: [
+          { key: 'phase', value: 'done' },
+          { key: 'stale', value: null },
+        ],
+        options: {},
+      },
+    ]);
+  });
+
+  it('is a no-op for an empty record (no dispatch)', async () => {
+    await experimental_setAttributes({});
+    expect(dispatchCalls).toHaveLength(0);
+  });
+
+  it('throws FatalError when the workflow runtime has not initialized useStep', async () => {
+    delete (globalThis as Record<symbol, unknown>)[WORKFLOW_USE_STEP];
+    await expect(
+      experimental_setAttributes({ phase: 'init' })
+    ).rejects.toBeInstanceOf(FatalError);
+  });
+
+  it('throws FatalError for reserved-prefix keys before any dispatch', async () => {
+    await expect(
+      experimental_setAttributes({ $sys: 'x' })
+    ).rejects.toBeInstanceOf(FatalError);
+    expect(dispatchCalls).toHaveLength(0);
+  });
+
+  it('dispatches reserved-prefix keys when allowReservedAttributes opt-in is set, and forwards the flag to the step', async () => {
+    await experimental_setAttributes(
+      { '$framework.kind': 'agent' },
+      { allowReservedAttributes: true }
+    );
+    expect(dispatchCalls).toEqual([
+      {
+        stepName: '__builtin_set_attributes',
+        changes: [{ key: '$framework.kind', value: 'agent' }],
+        options: { allowReservedAttributes: true },
+      },
+    ]);
+  });
+
+  it('still rejects reserved-prefix keys when allowReservedAttributes is explicitly false', async () => {
+    await expect(
+      experimental_setAttributes(
+        { '$framework.kind': 'agent' },
+        { allowReservedAttributes: false }
+      )
+    ).rejects.toBeInstanceOf(FatalError);
+    expect(dispatchCalls).toHaveLength(0);
+  });
+
+  it('throws FatalError when called with a non-object', async () => {
+    await expect(
+      experimental_setAttributes(null as unknown as Record<string, string>)
+    ).rejects.toBeInstanceOf(FatalError);
+    await expect(
+      experimental_setAttributes([] as unknown as Record<string, string>)
+    ).rejects.toBeInstanceOf(FatalError);
+  });
+});