Add bounded step claim attribution to SDK spans

.changeset/quiet-steps-claim-telemetry.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'@workflow/world-vercel': patch
+---
+
+Add bounded trace attributes identifying lazy step-start claim strategies and ownership stamps.

packages/core/src/runtime/step-executor.ts

@@ -697,6 +697,17 @@ export async function executeStep(
         (params.forceOptimisticStart === true &&
           !isOptimisticInlineStartExplicitlyDisabled()));
 
+    // Keep this on the enclosing step span from the beginning so a losing
+    // claim retains the strategy after the 409 is reconciled as `skipped`.
+    // Bare background starts and owned recovery deliberately have no value.
+    if (params.preclaimedStart) {
+      span?.setAttributes(Attribute.StepStartStrategy('batch_preclaimed'));
+    } else if (params.lazyStepInput !== undefined) {
+      span?.setAttributes(
+        Attribute.StepStartStrategy(optimisticStart ? 'optimistic' : 'awaited')
+      );
+    }
+
     let step: StartedStep;
     // Params for the `step_started` create on either path below. The slot
     // snapshot is not spread here: `createEvent` attaches it to every write,

packages/core/src/telemetry/semantic-conventions.ts

@@ -342,6 +342,16 @@ export const StepLatencyOptimizations = SemanticConvention<string[]>(
   'step.latency_optimizations'
 );
 
+/**
+ * How the step's initial `step_started` claim was made. Only present for
+ * inline create claims; ordinary background starts and owned recovery remain
+ * unlabeled.
+ */
+export type StepStartStrategy = 'awaited' | 'optimistic' | 'batch_preclaimed';
+export const StepStartStrategy = SemanticConvention<StepStartStrategy>(
+  'workflow.step_start.strategy'
+);
+
 /** Whether the step was skipped during execution */
 export const StepSkipped = SemanticConvention<boolean>('step.skipped');
 

packages/world-vercel/src/events-v4.ts

@@ -74,6 +74,8 @@ import {
   WorkflowClientVersion,
   WorkflowEventsTransport,
   WorkflowEventType,
+  WorkflowStepStartMode,
+  WorkflowStepStartOwnerStamped,
   WorkflowWsRequestId,
   WorkflowWsUrl,
 } from './telemetry.js';
@@ -113,7 +115,7 @@ async function fetchV4(
   init: { method: string; headers: Headers; body?: Uint8Array },
   config: APIConfig | undefined,
   opName: string,
-  attributes?: Record<string, string | number | string[]>
+  attributes?: Record<string, string | number | boolean | string[]>
 ): Promise<Response> {
   const dispatcher = getEventsDispatcher(config);
   const response = await instrumentedFetch({
@@ -786,6 +788,20 @@ async function postWorkflowRunEventV4(
       ...WorkflowEventsTransport('http'),
       ...WorkflowEventType(input.eventType),
       ...WorkflowClientVersion(`@workflow/world-vercel/${version}`),
+      ...(input.eventType === 'step_started'
+        ? {
+            ...WorkflowStepStartMode(
+              input.payload === undefined
+                ? input.ownerMessageId !== undefined
+                  ? 'single_owned_recovery'
+                  : 'single_bare'
+                : 'single_lazy_create_claim'
+            ),
+            ...WorkflowStepStartOwnerStamped(
+              input.ownerMessageId !== undefined
+            ),
+          }
+        : {}),
       ...(input.stso !== undefined ? StepStsoMs(input.stso) : {}),
       ...(input.optimizations !== undefined
         ? StepLatencyOptimizations(input.optimizations)
@@ -1041,6 +1057,18 @@ export async function createWorkflowRunEventsBatchV4(
     {
       ...WorkflowEventsTransport('http'),
       'workflow.batch.bytes': body.byteLength,
+      ...(input.events.some((event) => event.eventType === 'step_started')
+        ? {
+            ...WorkflowStepStartMode(
+              input.events.some((event) => event.eventType === 'step_created')
+                ? 'batch_create_claim'
+                : 'batch_bare'
+            ),
+            ...WorkflowStepStartOwnerStamped(
+              input.events.some((event) => event.ownerMessageId !== undefined)
+            ),
+          }
+        : {}),
     }
   );
 
@@ -1253,6 +1281,20 @@ async function postEventFrameOverWs(
         ...(input.optimizations !== undefined
           ? StepLatencyOptimizations(input.optimizations)
           : {}),
+        ...(input.eventType === 'step_started'
+          ? {
+              ...WorkflowStepStartMode(
+                input.payload === undefined
+                  ? input.ownerMessageId !== undefined
+                    ? 'single_owned_recovery'
+                    : 'single_bare'
+                  : 'single_lazy_create_claim'
+              ),
+              ...WorkflowStepStartOwnerStamped(
+                input.ownerMessageId !== undefined
+              ),
+            }
+          : {}),
         ...NetworkProtocolName('websocket'),
         ...WorkflowWsUrl(wsUrl),
       },

packages/world-vercel/src/telemetry.ts

@@ -307,6 +307,22 @@ export const WorkflowEventType = SemanticConvention<string>(
   'workflow.event.type'
 );
 
+/** Server-side classification of a step_started write. */
+export type WorkflowStepStartMode =
+  | 'single_lazy_create_claim'
+  | 'single_owned_recovery'
+  | 'single_bare'
+  | 'batch_create_claim'
+  | 'batch_bare';
+export const WorkflowStepStartMode = SemanticConvention<WorkflowStepStartMode>(
+  'workflow.step_start.mode'
+);
+
+/** Whether a step_started write carries an inline ownership stamp. */
+export const WorkflowStepStartOwnerStamped = SemanticConvention<boolean>(
+  'workflow.step_start.owner_stamped'
+);
+
 /** Version of the Workflow client package issuing the request. */
 export const WorkflowClientVersion = SemanticConvention<string>(
   'workflow.client.version'

packages/world-vercel/src/ws-transport-spans.test.ts

@@ -242,7 +242,56 @@ const writeSpan = (): ReadableSpan => {
   return spans[0];
 };
 
+const startedBody = (eventId = 'evnt_1') =>
+  new Uint8Array(
+    encode({
+      event: {
+        eventId,
+        runId: 'wrun_1',
+        createdAt: CREATED_AT,
+        eventType: 'step_started',
+        specVersion: 2,
+        correlationId: 'step_1',
+        eventData: { stepName: 'step' },
+      },
+      step: {
+        runId: 'wrun_1',
+        stepId: 'step_1',
+        stepName: 'step',
+        status: 'running',
+        attempt: 1,
+        createdAt: CREATED_AT,
+        updatedAt: CREATED_AT,
+        startedAt: CREATED_AT,
+      },
+    })
+  );
+
 describe('per-write client span', () => {
+  it('attributes a stamped lazy step claim', async () => {
+    await withOpenChannel(() => ({
+      status: 201,
+      body: startedBody(),
+    }));
+
+    await createWorkflowRunEventV4(
+      {
+        ...input,
+        eventType: 'step_started',
+        payload: new Uint8Array([1]),
+        ownerMessageId: 'msg_1',
+        stepName: 'step',
+      },
+      { token: 'test-token' }
+    );
+
+    const span = writeSpan();
+    expect(span.attributes['workflow.step_start.mode']).toBe(
+      'single_lazy_create_claim'
+    );
+    expect(span.attributes['workflow.step_start.owner_stamped']).toBe(true);
+  });
+
   it('emits one `http POST` CLIENT span per event write', async () => {
     await withOpenChannel();