[core] Trace fresh workflow replay phases

.changeset/trace-replay-phases.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Trace workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

packages/core/src/runtime-trace-mode.test.ts

@@ -201,7 +201,6 @@ async function driveHandler(opts: {
   const getWorldSpan = exporter
     .getFinishedSpans()
     .find((s) => s.name === 'workflow.route.get_world');
-
   return {
     workflowSpan,
     routeSpan,
@@ -287,7 +286,6 @@ describe('workflowEntrypoint trace modes', () => {
     );
     expect(getWorldSpan).toBeDefined();
     expect(getWorldSpan?.parentSpanId).toBe(routeSpan?.spanContext().spanId);
-
     expect(workflowSpan).toBeDefined();
     // Child of the local /flow route span — same trace, so one
     // invocation is a single bounded trace rather than a new root.

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

@@ -82,6 +82,11 @@ export const WorkflowExecutionMode = SemanticConvention<'replay' | 'retained'>(
   'workflow.execution.mode'
 );
 
+/** Whether every script needed for workflow bundle evaluation was cached. */
+export const WorkflowBundleCompileCacheHit = SemanticConvention<boolean>(
+  'workflow.bundle.compile.cache_hit'
+);
+
 /**
  * Events the replay walked past that no consumer claimed, still held when the
  * replay stopped.

packages/core/src/vm/script-cache.test.ts

@@ -1,10 +1,9 @@
-import { runInContext } from 'node:vm';
+import { type Context, runInContext } from 'node:vm';
 import { afterEach, describe, expect, it } from 'vitest';
 import { createContext } from './index.js';
 import {
   clearWorkflowScriptCache,
   getCachedWorkflowScript,
-  runCachedWorkflowScript,
   workflowScriptCacheSize,
 } from './script-cache.js';
 
@@ -37,35 +36,52 @@ function buildBundle(marker: string, workflowCount = 12): string {
   return `globalThis.__private_workflows = new Map();\n${defs.join('\n')}\n`;
 }
 
+function getScript(code: string, filename: string) {
+  return getCachedWorkflowScript(code, filename).script;
+}
+
+function runScript(code: string, filename: string, context: Context) {
+  return getScript(code, filename).runInContext(context);
+}
+
 describe('script-cache', () => {
   afterEach(() => {
     clearWorkflowScriptCache();
   });
 
   it('returns the same compiled Script for identical (code, filename)', () => {
-    const a = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts');
-    const b = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts');
+    const a = getScript(SAMPLE_BUNDLE, 'workflows/a.ts');
+    const b = getScript(SAMPLE_BUNDLE, 'workflows/a.ts');
     expect(a).toBe(b);
   });
 
+  it('reports whether compilation was served from cache', () => {
+    const first = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts');
+    const second = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts');
+
+    expect(first.cacheHit).toBe(false);
+    expect(second.cacheHit).toBe(true);
+    expect(second.script).toBe(first.script);
+  });
+
   it('returns distinct Scripts for the same code under different filenames', () => {
-    const a = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts');
-    const b = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/b.ts');
+    const a = getScript(SAMPLE_BUNDLE, 'workflows/a.ts');
+    const b = getScript(SAMPLE_BUNDLE, 'workflows/b.ts');
     expect(a).not.toBe(b);
   });
 
   it('returns distinct Scripts for different code under the same filename', () => {
-    const a = getCachedWorkflowScript('1 + 1', 'workflows/a.ts');
-    const b = getCachedWorkflowScript('2 + 2', 'workflows/a.ts');
+    const a = getScript('1 + 1', 'workflows/a.ts');
+    const b = getScript('2 + 2', 'workflows/a.ts');
     expect(a).not.toBe(b);
   });
 
   it('produces a byte-identical workflow result vs. uncached runInContext', async () => {
     // Cached path: run the bundle then look up the workflow, mirroring
     // runWorkflow's two-step evaluation.
     const { context: cachedCtx } = createContext({ seed, fixedTimestamp });
-    runCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts', cachedCtx);
-    const cachedFn = runCachedWorkflowScript(
+    runScript(SAMPLE_BUNDLE, 'workflows/a.ts', cachedCtx);
+    const cachedFn = runScript(
       `globalThis.__private_workflows?.get('my/workflow')`,
       'workflows/a.ts',
       cachedCtx
@@ -90,16 +106,14 @@ describe('script-cache', () => {
   });
 
   it('reuses the compiled Script across multiple runs against fresh contexts', async () => {
-    const script = getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts');
+    const script = getScript(SAMPLE_BUNDLE, 'workflows/a.ts');
 
     const results: string[] = [];
     for (let i = 0; i < 3; i++) {
       const { context } = createContext({ seed, fixedTimestamp });
-      runCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts', context);
+      runScript(SAMPLE_BUNDLE, 'workflows/a.ts', context);
       // The same cached Script object is used every iteration.
-      expect(getCachedWorkflowScript(SAMPLE_BUNDLE, 'workflows/a.ts')).toBe(
-        script
-      );
+      expect(getScript(SAMPLE_BUNDLE, 'workflows/a.ts')).toBe(script);
       const fn = runInContext(
         `globalThis.__private_workflows?.get('my/workflow')`,
         context
@@ -119,7 +133,7 @@ describe('script-cache', () => {
     const editCount = 100;
     const filename = 'workflows/a.ts';
     for (let i = 0; i < editCount; i++) {
-      getCachedWorkflowScript(buildBundle(`edit-${i}`), filename);
+      getScript(buildBundle(`edit-${i}`), filename);
     }
 
     const size = workflowScriptCacheSize();
@@ -130,9 +144,7 @@ describe('script-cache', () => {
     // The cache still serves correctly after heavy churn: the most-recently
     // inserted bundle is retained and repeated lookups return the same Script.
     const latest = buildBundle(`edit-${editCount - 1}`);
-    expect(getCachedWorkflowScript(latest, filename)).toBe(
-      getCachedWorkflowScript(latest, filename)
-    );
+    expect(getScript(latest, filename)).toBe(getScript(latest, filename));
   });
 
   it('keeps the most-recently-used bundle and evicts the stale one', () => {
@@ -141,18 +153,18 @@ describe('script-cache', () => {
     // unrelated bundles churn through. LRU must NOT evict the bundle we keep
     // using, even though it was inserted first.
     const hot = buildBundle('hot');
-    const hotScript = getCachedWorkflowScript(hot, filename);
+    const hotScript = getScript(hot, filename);
 
     for (let i = 0; i < 50; i++) {
-      getCachedWorkflowScript(buildBundle(`cold-${i}`), filename);
+      getScript(buildBundle(`cold-${i}`), filename);
       // Re-access the hot bundle so it stays most-recently-used.
-      expect(getCachedWorkflowScript(hot, filename)).toBe(hotScript);
+      expect(getScript(hot, filename)).toBe(hotScript);
     }
 
     // After all that churn the hot bundle is still the *same* cached Script —
     // proving LRU recency (touch-on-access), not mere insertion order, governs
     // eviction.
-    expect(getCachedWorkflowScript(hot, filename)).toBe(hotScript);
+    expect(getScript(hot, filename)).toBe(hotScript);
   });
 
   it('never returns the wrong Script across realistic multi-workflow bundles', async () => {
@@ -165,10 +177,10 @@ describe('script-cache', () => {
     const fileA = 'workflows/a.ts';
     const fileB = 'workflows/b.ts';
 
-    const xa = getCachedWorkflowScript(bundleX, fileA);
-    const xb = getCachedWorkflowScript(bundleX, fileB);
-    const ya = getCachedWorkflowScript(bundleY, fileA);
-    const yb = getCachedWorkflowScript(bundleY, fileB);
+    const xa = getScript(bundleX, fileA);
+    const xb = getScript(bundleX, fileB);
+    const ya = getScript(bundleY, fileA);
+    const yb = getScript(bundleY, fileB);
 
     // All four (code, filename) combinations are distinct Script objects.
     const scripts = [xa, xb, ya, yb];
@@ -179,20 +191,20 @@ describe('script-cache', () => {
     }
 
     // Same (code, filename) is stable across lookups.
-    expect(getCachedWorkflowScript(bundleX, fileA)).toBe(xa);
-    expect(getCachedWorkflowScript(bundleY, fileB)).toBe(yb);
+    expect(getScript(bundleX, fileA)).toBe(xa);
+    expect(getScript(bundleY, fileB)).toBe(yb);
 
     // Running each bundle yields its OWN marker, confirming no cross-wiring.
     const { context: ctxX } = createContext({ seed, fixedTimestamp });
-    runCachedWorkflowScript(bundleX, fileA, ctxX);
+    runScript(bundleX, fileA, ctxX);
     const fnX = runInContext(
       `globalThis.__private_workflows?.get('app/workflow-3')`,
       ctxX
     ) as (n: string) => Promise<string>;
     expect(await fnX('z')).toContain('bundle-X:3:z');
 
     const { context: ctxY } = createContext({ seed, fixedTimestamp });
-    runCachedWorkflowScript(bundleY, fileA, ctxY);
+    runScript(bundleY, fileA, ctxY);
     const fnY = runInContext(
       `globalThis.__private_workflows?.get('app/workflow-3')`,
       ctxY

packages/core/src/vm/script-cache.ts

@@ -1,4 +1,4 @@
-import { type Context, Script } from 'node:vm';
+import { Script } from 'node:vm';
 
 /**
  * Module-level cache of compiled workflow-bundle `vm.Script` objects.
@@ -101,7 +101,7 @@ function touchBundle(code: string): Map<string, Script> | undefined {
 export function getCachedWorkflowScript(
   code: string,
   filename: string
-): Script {
+): { script: Script; cacheHit: boolean } {
   let byFilename = touchBundle(code);
   if (byFilename === undefined) {
     byFilename = new Map<string, Script>();
@@ -117,23 +117,12 @@ export function getCachedWorkflowScript(
     }
   }
   let script = byFilename.get(filename);
+  const cacheHit = script !== undefined;
   if (script === undefined) {
     script = new Script(code, { filename });
     byFilename.set(filename, script);
   }
-  return script;
-}
-
-/**
- * Runs the cached workflow-bundle `Script` against `context`. Compiles and
- * caches the `Script` on first use for the given `(code, filename)`.
- */
-export function runCachedWorkflowScript(
-  code: string,
-  filename: string,
-  context: Context
-): unknown {
-  return getCachedWorkflowScript(code, filename).runInContext(context);
+  return { script, cacheHit };
 }
 
 /**

packages/core/src/workflow-tracing.test.ts

@@ -0,0 +1,133 @@
+import { context, trace as otelTrace } from '@opentelemetry/api';
+import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks';
+import {
+  BasicTracerProvider,
+  InMemorySpanExporter,
+  SimpleSpanProcessor,
+} from '@opentelemetry/sdk-trace-base';
+import type { WorkflowRun } from '@workflow/world';
+import {
+  afterAll,
+  afterEach,
+  beforeAll,
+  beforeEach,
+  describe,
+  expect,
+  it,
+} from 'vitest';
+import { dehydrateWorkflowArguments } from './serialization.js';
+import { clearWorkflowScriptCache } from './vm/script-cache.js';
+import { runWorkflow } from './workflow.js';
+
+const exporter = new InMemorySpanExporter();
+const provider = new BasicTracerProvider();
+const contextManager = new AsyncLocalStorageContextManager();
+
+beforeAll(() => {
+  provider.addSpanProcessor(new SimpleSpanProcessor(exporter));
+  contextManager.enable();
+  context.setGlobalContextManager(contextManager);
+  otelTrace.setGlobalTracerProvider(provider);
+});
+
+afterAll(async () => {
+  await provider.shutdown();
+  context.disable();
+  otelTrace.disable();
+});
+
+beforeEach(() => {
+  clearWorkflowScriptCache();
+});
+
+afterEach(() => {
+  exporter.reset();
+});
+
+async function makeRun(): Promise<WorkflowRun> {
+  const runId = 'wrun_trace_replay';
+  return {
+    runId,
+    workflowName: 'workflow',
+    status: 'running',
+    input: await dehydrateWorkflowArguments(['hello'], runId, undefined, []),
+    createdAt: new Date('2024-01-01T00:00:00.000Z'),
+    updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+    startedAt: new Date('2024-01-01T00:00:00.000Z'),
+    deploymentId: 'test-deployment',
+  };
+}
+
+const workflowCode = `
+async function workflow(value) { return value; }
+globalThis.__private_workflows = new Map();
+globalThis.__private_workflows.set('workflow', workflow);
+`;
+
+describe('fresh replay tracing', () => {
+  it('breaks workflow.run into blocking replay phases', async () => {
+    const run = await makeRun();
+    await runWorkflow(workflowCode, run, [], undefined);
+
+    const spans = exporter.getFinishedSpans();
+    const workflowRun = spans.find(
+      (span) => span.name === 'workflow.run workflow'
+    );
+    expect(workflowRun).toBeDefined();
+
+    const childNames = spans
+      .filter((span) => span.parentSpanId === workflowRun?.spanContext().spanId)
+      .map((span) => span.name);
+    expect(childNames).toEqual(
+      expect.arrayContaining([
+        'workflow.vm.create_context',
+        'workflow.bundle.compile',
+        'workflow.bundle.evaluate',
+        'workflow.input.hydrate',
+        'workflow.replay.execute',
+      ])
+    );
+  });
+
+  it('marks bundle compilation cache hits on later fresh replays', async () => {
+    const run = await makeRun();
+    await runWorkflow(workflowCode, run, [], undefined);
+    await runWorkflow(workflowCode, run, [], undefined);
+
+    const compileSpans = exporter
+      .getFinishedSpans()
+      .filter((span) => span.name === 'workflow.bundle.compile');
+    expect(compileSpans).toHaveLength(2);
+    expect(
+      compileSpans.map(
+        (span) => span.attributes['workflow.bundle.compile.cache_hit']
+      )
+    ).toEqual([false, true]);
+  });
+
+  it('reports a bundle hit when only a different workflow lookup compiles', async () => {
+    const firstName = 'workflow//./workflows/shared//first';
+    const secondName = 'workflow//./workflows/shared//second';
+    const sharedBundle = `
+async function first(value) { return value; }
+async function second(value) { return value; }
+globalThis.__private_workflows = new Map();
+globalThis.__private_workflows.set(${JSON.stringify(firstName)}, first);
+globalThis.__private_workflows.set(${JSON.stringify(secondName)}, second);
+`;
+    const firstRun = { ...(await makeRun()), workflowName: firstName };
+    const secondRun = { ...(await makeRun()), workflowName: secondName };
+
+    await runWorkflow(sharedBundle, firstRun, [], undefined);
+    await runWorkflow(sharedBundle, secondRun, [], undefined);
+
+    const compileSpans = exporter
+      .getFinishedSpans()
+      .filter((span) => span.name === 'workflow.bundle.compile');
+    expect(
+      compileSpans.map(
+        (span) => span.attributes['workflow.bundle.compile.cache_hit']
+      )
+    ).toEqual([false, true]);
+  });
+});