[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
NathanColosimo merged 5 commits into from
Aug 31, 2026
Merged

[core] Trace fresh workflow replay phases #3797

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
44897b7
Trace fresh workflow replay phases
NathanColosimo Aug 14, 2026
2f4d8d4
refactor(core): simplify workflow script cache API
NathanColosimo Aug 15, 2026
1bd9c04
Fix retained workflow tracing
NathanColosimo Aug 18, 2026
532197a
fix(core): keep tracing failure-safe
NathanColosimo Aug 20, 2026
2f331e9
Trace replay event loading
NathanColosimo Aug 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/trace-replay-phases.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.
13 changes: 11 additions & 2 deletions packages/core/src/runtime-trace-mode.test.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
const getWorldSpan = exporter
.getFinishedSpans()
.find((s) => s.name === 'workflow.route.get_world');

return {
workflowSpan,
routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
).toBe(false);

const replayLoadSpan = exporter
.getFinishedSpans()
.find((finished) => finished.name === 'workflow.replay.load');
expect(replayLoadSpan?.parentSpanId).toBe(
workflowSpan?.spanContext().spanId
);
expect(replayLoadSpan?.attributes).toMatchObject({
'workflow.replay.load.source': 'run_started',
'workflow.events.count': 0,
});

// Queue-delivered invocation spans use the CONSUMER kind, matching
// queue-delivered step.execute spans.
expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down
59 changes: 40 additions & 19 deletions packages/core/src/runtime.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
return result;
};

const traceReplayLoad = <T extends { events?: Event[] }>(
source: Attribute.WorkflowReplayLoadSource,
load: () => Promise<T>
): Promise<T> =>
trace('workflow.replay.load', async (loadSpan) => {
loadSpan?.setAttributes({
...Attribute.WorkflowRunId(runId),
...Attribute.WorkflowReplayLoadSource(source),
});
const result = await load();
loadSpan?.setAttributes(
Attribute.WorkflowEventsCount(
result.events?.length ?? 0
)
);
return result;
});

/**
* The slot snapshot for a write issued from this loop: how
* much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
span?.addEvent('workflow.hook_received.create.start', {
'workflow.hook_received.preload_events': true,
});
const result = await createEvent(
{
eventType: 'hook_received',
specVersion: SPEC_VERSION_CURRENT,
correlationId: hookResumeInput.hookId,
eventData: {
token: hookResumeInput.token,
payload: hookResumeInput.payload,
const result = await traceReplayLoad('hook_preload', () =>
createEvent(
{
eventType: 'hook_received',
specVersion: SPEC_VERSION_CURRENT,
correlationId: hookResumeInput.hookId,
eventData: {
token: hookResumeInput.token,
payload: hookResumeInput.payload,
},
},
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
)
);
hookEnsured = true;
// Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
span?.addEvent('workflow.run_started.create.start', {
'workflow.run_started.skip_preload': false,
});
const result = await createEvent(runStartedEvent, {
requestId,
});
const result = await traceReplayLoad(
'run_started',
() => createEvent(runStartedEvent, { requestId })
);
workflowRun = result.run;
maxEventsLimit = clampMaxEvents(result.maxEvents);
// Anchors RSFS, see the declaration above.
Expand Down
184 changes: 92 additions & 92 deletions packages/core/src/runtime/helpers.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
afterCursor?: string
): Promise<LoadedEventLog> {
const incremental = afterCursor !== undefined;
return trace(
incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
async (span) => {
span?.setAttributes({
...Attribute.WorkflowRunId(runId),
});

const loadedEvents: Event[] = [];
const loadedEventIds = new Set<string>();
const requestedCursors = new Set<string>();
let cursor: string | null = afterCursor ?? null;
let hasMore = true;
let pagesLoaded = 0;
let retriedWithoutCursor = false;

const world = await getWorldLazy();
const loadStart = Date.now();
while (hasMore) {
// TODO: we're currently loading all the data with resolveRef behavior. We need to update this
// to lazyload the data from the world instead so that we can optimize and make the event log loading
// much faster and memory efficient
const pageStart = Date.now();
const requestedCursor = cursor;
recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

let response: Awaited<ReturnType<typeof world.events.list>>;
try {
response = await world.events.list({
runId,
pagination: {
sortOrder: 'asc',
cursor: requestedCursor ?? undefined,
},
});
} catch (error) {
if (
shouldRetryWithoutEventCursor(
error,
requestedCursor,
retriedWithoutCursor
)
) {
runtimeLogger.warn(
'Event cursor was rejected; retrying with a full event reload.',
{ workflowRunId: runId }
);
loadedEvents.length = 0;
loadedEventIds.clear();
requestedCursors.clear();
cursor = null;
retriedWithoutCursor = true;
continue;
}
throw error;
}

appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
hasMore = response.hasMore;
assertEventPaginationProgress(
return trace('workflow.replay.load', async (span) => {
span?.setAttributes({
...Attribute.WorkflowRunId(runId),
...Attribute.WorkflowReplayLoadSource(
incremental ? 'events_list_incremental' : 'events_list'
),
});

const loadedEvents: Event[] = [];
const loadedEventIds = new Set<string>();
const requestedCursors = new Set<string>();
let cursor: string | null = afterCursor ?? null;
let hasMore = true;
let pagesLoaded = 0;
let retriedWithoutCursor = false;

const world = await getWorldLazy();
const loadStart = Date.now();
while (hasMore) {
// TODO: we're currently loading all the data with resolveRef behavior. We need to update this
// to lazyload the data from the world instead so that we can optimize and make the event log loading
// much faster and memory efficient
const pageStart = Date.now();
const requestedCursor = cursor;
recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

let response: Awaited<ReturnType<typeof world.events.list>>;
try {
response = await world.events.list({
runId,
hasMore,
response.cursor,
requestedCursors
);
// 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 DB query hit the limit exactly and returned a
// precautionary `LastEvaluatedKey`. 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-
// reload fallback on every subsequent replay iteration.
cursor = response.cursor ?? cursor;
pagesLoaded++;

runtimeLogger.debug('Loaded event page', {
workflowRunId: runId,
incremental,
page: pagesLoaded,
pageEvents: response.data.length,
totalEvents: loadedEvents.length,
hasMore,
pageMs: Date.now() - pageStart,
pagination: {
sortOrder: 'asc',
cursor: requestedCursor ?? undefined,
},
});
} catch (error) {
if (
shouldRetryWithoutEventCursor(
error,
requestedCursor,
retriedWithoutCursor
)
) {
runtimeLogger.warn(
'Event cursor was rejected; retrying with a full event reload.',
{ workflowRunId: runId }
);
loadedEvents.length = 0;
loadedEventIds.clear();
requestedCursors.clear();
cursor = null;
retriedWithoutCursor = true;
continue;
}
throw error;
}

runtimeLogger.debug('Event load complete', {
appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
hasMore = response.hasMore;
assertEventPaginationProgress(
runId,
hasMore,
response.cursor,
requestedCursors
);
// 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 DB query hit the limit exactly and returned a
// precautionary `LastEvaluatedKey`. 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-
// reload fallback on every subsequent replay iteration.
cursor = response.cursor ?? cursor;
pagesLoaded++;

runtimeLogger.debug('Loaded event page', {
workflowRunId: runId,
incremental,
page: pagesLoaded,
pageEvents: response.data.length,
totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
});
}

span?.setAttributes({
...Attribute.WorkflowEventsCount(loadedEvents.length),
...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
});
runtimeLogger.debug('Event load complete', {
workflowRunId: runId,
incremental,
totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
});

return { events: loadedEvents, cursor };
}
);
span?.setAttributes({
...Attribute.WorkflowEventsCount(loadedEvents.length),
...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
});

return { events: loadedEvents, cursor };
});
}

/**
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

NathanColosimo
merged 5 commits into

from

Aug 31, 2026
 Merged


[core] Trace fresh workflow replay phases
#3797
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
44897b7

Trace fresh workflow replay phases


NathanColosimo Aug 14, 2026
2f4d8d4

refactor(core): simplify workflow script cache API


NathanColosimo Aug 15, 2026
1bd9c04

Fix retained workflow tracing


NathanColosimo Aug 18, 2026
532197a

fix(core): keep tracing failure-safe


NathanColosimo Aug 20, 2026
2f331e9

Trace replay event loading


NathanColosimo Aug 25, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


5 changes: 5 additions & 0 deletions
.changeset/trace-replay-phases.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

13 changes: 11 additions & 2 deletions
packages/core/src/runtime-trace-mode.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
 const getWorldSpan = exporter
 .getFinishedSpans()
 .find((s) => s.name === 'workflow.route.get_world');

 return {
 workflowSpan,
 routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
 runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
 ).toBe(false);

 const replayLoadSpan = exporter
 .getFinishedSpans()
 .find((finished) => finished.name === 'workflow.replay.load');
 expect(replayLoadSpan?.parentSpanId).toBe(
 workflowSpan?.spanContext().spanId
 );
 expect(replayLoadSpan?.attributes).toMatchObject({
 'workflow.replay.load.source': 'run_started',
 'workflow.events.count': 0,
 });

 // Queue-delivered invocation spans use the CONSUMER kind, matching
 // queue-delivered step.execute spans.
 expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down

59 changes: 40 additions & 19 deletions
packages/core/src/runtime.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
 return result;
 };

 const traceReplayLoad = <T extends { events?: Event[] }>(
 source: Attribute.WorkflowReplayLoadSource,
 load: () => Promise<T>
 ): Promise<T> =>
 trace('workflow.replay.load', async (loadSpan) => {
 loadSpan?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(source),
 });
 const result = await load();
 loadSpan?.setAttributes(
 Attribute.WorkflowEventsCount(
 result.events?.length ?? 0
 )
 );
 return result;
 });

 /**
 * The slot snapshot for a write issued from this loop: how
 * much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.hook_received.create.start', {
 'workflow.hook_received.preload_events': true,
 });
 const result = await createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 const result = await traceReplayLoad('hook_preload', () =>
 createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 },
 },
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
 {
 requestId,
 occurredAt,
 resumeId: hookResumeInput.resumeId,
 resumePayloadDigest: hookResumeInput.payloadDigest,
 preloadEvents: true,
}
)
 );
 hookEnsured = true;
 // Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.run_started.create.start', {
 'workflow.run_started.skip_preload': false,
 });
 const result = await createEvent(runStartedEvent, {
 requestId,
 });
 const result = await traceReplayLoad(
 'run_started',
 () => createEvent(runStartedEvent, { requestId })
 );
 workflowRun = result.run;
 maxEventsLimit = clampMaxEvents(result.maxEvents);
 // Anchors RSFS, see the declaration above.
Expand Down

184 changes: 92 additions & 92 deletions
packages/core/src/runtime/helpers.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
 afterCursor?: string
): Promise<LoadedEventLog> {
 const incremental = afterCursor !== undefined;
 return trace(
 incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
 async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 return trace('workflow.replay.load', async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(
 incremental ? 'events_list_incremental' : 'events_list'
 ),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
 hasMore,
 pageMs: Date.now() - pageStart,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 runtimeLogger.debug('Event load complete', {
 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
 });
 }

 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });
 runtimeLogger.debug('Event load complete', {
 workflowRunId: runId,
 incremental,
 totalEvents: loadedEvents.length,
 pagesLoaded,
 totalMs: Date.now() - loadStart,
 });

 return { events: loadedEvents, cursor };
 }
 );
 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });

 return { events: loadedEvents, cursor };
 });
}

/**
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

NathanColosimo
merged 5 commits into

from

Aug 31, 2026
 Merged


[core] Trace fresh workflow replay phases
#3797
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
44897b7

Trace fresh workflow replay phases


NathanColosimo Aug 14, 2026
2f4d8d4

refactor(core): simplify workflow script cache API


NathanColosimo Aug 15, 2026
1bd9c04

Fix retained workflow tracing


NathanColosimo Aug 18, 2026
532197a

fix(core): keep tracing failure-safe


NathanColosimo Aug 20, 2026
2f331e9

Trace replay event loading


NathanColosimo Aug 25, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


5 changes: 5 additions & 0 deletions
.changeset/trace-replay-phases.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

13 changes: 11 additions & 2 deletions
packages/core/src/runtime-trace-mode.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
 const getWorldSpan = exporter
 .getFinishedSpans()
 .find((s) => s.name === 'workflow.route.get_world');

 return {
 workflowSpan,
 routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
 runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
 ).toBe(false);

 const replayLoadSpan = exporter
 .getFinishedSpans()
 .find((finished) => finished.name === 'workflow.replay.load');
 expect(replayLoadSpan?.parentSpanId).toBe(
 workflowSpan?.spanContext().spanId
 );
 expect(replayLoadSpan?.attributes).toMatchObject({
 'workflow.replay.load.source': 'run_started',
 'workflow.events.count': 0,
 });

 // Queue-delivered invocation spans use the CONSUMER kind, matching
 // queue-delivered step.execute spans.
 expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down

59 changes: 40 additions & 19 deletions
packages/core/src/runtime.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
 return result;
 };

 const traceReplayLoad = <T extends { events?: Event[] }>(
 source: Attribute.WorkflowReplayLoadSource,
 load: () => Promise<T>
 ): Promise<T> =>
 trace('workflow.replay.load', async (loadSpan) => {
 loadSpan?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(source),
 });
 const result = await load();
 loadSpan?.setAttributes(
 Attribute.WorkflowEventsCount(
 result.events?.length ?? 0
 )
 );
 return result;
 });

 /**
 * The slot snapshot for a write issued from this loop: how
 * much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.hook_received.create.start', {
 'workflow.hook_received.preload_events': true,
 });
 const result = await createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 const result = await traceReplayLoad('hook_preload', () =>
 createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 },
 },
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
 {
 requestId,
 occurredAt,
 resumeId: hookResumeInput.resumeId,
 resumePayloadDigest: hookResumeInput.payloadDigest,
 preloadEvents: true,
}
)
 );
 hookEnsured = true;
 // Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.run_started.create.start', {
 'workflow.run_started.skip_preload': false,
 });
 const result = await createEvent(runStartedEvent, {
 requestId,
 });
 const result = await traceReplayLoad(
 'run_started',
 () => createEvent(runStartedEvent, { requestId })
 );
 workflowRun = result.run;
 maxEventsLimit = clampMaxEvents(result.maxEvents);
 // Anchors RSFS, see the declaration above.
Expand Down

184 changes: 92 additions & 92 deletions
packages/core/src/runtime/helpers.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
 afterCursor?: string
): Promise<LoadedEventLog> {
 const incremental = afterCursor !== undefined;
 return trace(
 incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
 async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 return trace('workflow.replay.load', async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(
 incremental ? 'events_list_incremental' : 'events_list'
 ),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
 hasMore,
 pageMs: Date.now() - pageStart,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 runtimeLogger.debug('Event load complete', {
 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
 });
 }

 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });
 runtimeLogger.debug('Event load complete', {
 workflowRunId: runId,
 incremental,
 totalEvents: loadedEvents.length,
 pagesLoaded,
 totalMs: Date.now() - loadStart,
 });

 return { events: loadedEvents, cursor };
 }
 );
 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });

 return { events: loadedEvents, cursor };
 });
}

/**
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

NathanColosimo
merged 5 commits into

from

Aug 31, 2026
 Merged


[core] Trace fresh workflow replay phases
#3797
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
44897b7

Trace fresh workflow replay phases


NathanColosimo Aug 14, 2026
2f4d8d4

refactor(core): simplify workflow script cache API


NathanColosimo Aug 15, 2026
1bd9c04

Fix retained workflow tracing


NathanColosimo Aug 18, 2026
532197a

fix(core): keep tracing failure-safe


NathanColosimo Aug 20, 2026
2f331e9

Trace replay event loading


NathanColosimo Aug 25, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


5 changes: 5 additions & 0 deletions
.changeset/trace-replay-phases.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

13 changes: 11 additions & 2 deletions
packages/core/src/runtime-trace-mode.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
 const getWorldSpan = exporter
 .getFinishedSpans()
 .find((s) => s.name === 'workflow.route.get_world');

 return {
 workflowSpan,
 routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
 runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
 ).toBe(false);

 const replayLoadSpan = exporter
 .getFinishedSpans()
 .find((finished) => finished.name === 'workflow.replay.load');
 expect(replayLoadSpan?.parentSpanId).toBe(
 workflowSpan?.spanContext().spanId
 );
 expect(replayLoadSpan?.attributes).toMatchObject({
 'workflow.replay.load.source': 'run_started',
 'workflow.events.count': 0,
 });

 // Queue-delivered invocation spans use the CONSUMER kind, matching
 // queue-delivered step.execute spans.
 expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down

59 changes: 40 additions & 19 deletions
packages/core/src/runtime.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
 return result;
 };

 const traceReplayLoad = <T extends { events?: Event[] }>(
 source: Attribute.WorkflowReplayLoadSource,
 load: () => Promise<T>
 ): Promise<T> =>
 trace('workflow.replay.load', async (loadSpan) => {
 loadSpan?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(source),
 });
 const result = await load();
 loadSpan?.setAttributes(
 Attribute.WorkflowEventsCount(
 result.events?.length ?? 0
 )
 );
 return result;
 });

 /**
 * The slot snapshot for a write issued from this loop: how
 * much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.hook_received.create.start', {
 'workflow.hook_received.preload_events': true,
 });
 const result = await createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 const result = await traceReplayLoad('hook_preload', () =>
 createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 },
 },
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
 {
 requestId,
 occurredAt,
 resumeId: hookResumeInput.resumeId,
 resumePayloadDigest: hookResumeInput.payloadDigest,
 preloadEvents: true,
}
)
 );
 hookEnsured = true;
 // Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.run_started.create.start', {
 'workflow.run_started.skip_preload': false,
 });
 const result = await createEvent(runStartedEvent, {
 requestId,
 });
 const result = await traceReplayLoad(
 'run_started',
 () => createEvent(runStartedEvent, { requestId })
 );
 workflowRun = result.run;
 maxEventsLimit = clampMaxEvents(result.maxEvents);
 // Anchors RSFS, see the declaration above.
Expand Down

184 changes: 92 additions & 92 deletions
packages/core/src/runtime/helpers.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
 afterCursor?: string
): Promise<LoadedEventLog> {
 const incremental = afterCursor !== undefined;
 return trace(
 incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
 async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 return trace('workflow.replay.load', async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(
 incremental ? 'events_list_incremental' : 'events_list'
 ),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
 hasMore,
 pageMs: Date.now() - pageStart,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 runtimeLogger.debug('Event load complete', {
 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
 });
 }

 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });
 runtimeLogger.debug('Event load complete', {
 workflowRunId: runId,
 incremental,
 totalEvents: loadedEvents.length,
 pagesLoaded,
 totalMs: Date.now() - loadStart,
 });

 return { events: loadedEvents, cursor };
 }
 );
 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });

 return { events: loadedEvents, cursor };
 });
}

/**
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

NathanColosimo
merged 5 commits into

from

Aug 31, 2026
 Merged


[core] Trace fresh workflow replay phases
#3797
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
44897b7

Trace fresh workflow replay phases


NathanColosimo Aug 14, 2026
2f4d8d4

refactor(core): simplify workflow script cache API


NathanColosimo Aug 15, 2026
1bd9c04

Fix retained workflow tracing


NathanColosimo Aug 18, 2026
532197a

fix(core): keep tracing failure-safe


NathanColosimo Aug 20, 2026
2f331e9

Trace replay event loading


NathanColosimo Aug 25, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


5 changes: 5 additions & 0 deletions
.changeset/trace-replay-phases.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

13 changes: 11 additions & 2 deletions
packages/core/src/runtime-trace-mode.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
 const getWorldSpan = exporter
 .getFinishedSpans()
 .find((s) => s.name === 'workflow.route.get_world');

 return {
 workflowSpan,
 routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
 runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
 ).toBe(false);

 const replayLoadSpan = exporter
 .getFinishedSpans()
 .find((finished) => finished.name === 'workflow.replay.load');
 expect(replayLoadSpan?.parentSpanId).toBe(
 workflowSpan?.spanContext().spanId
 );
 expect(replayLoadSpan?.attributes).toMatchObject({
 'workflow.replay.load.source': 'run_started',
 'workflow.events.count': 0,
 });

 // Queue-delivered invocation spans use the CONSUMER kind, matching
 // queue-delivered step.execute spans.
 expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down

59 changes: 40 additions & 19 deletions
packages/core/src/runtime.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
 return result;
 };

 const traceReplayLoad = <T extends { events?: Event[] }>(
 source: Attribute.WorkflowReplayLoadSource,
 load: () => Promise<T>
 ): Promise<T> =>
 trace('workflow.replay.load', async (loadSpan) => {
 loadSpan?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(source),
 });
 const result = await load();
 loadSpan?.setAttributes(
 Attribute.WorkflowEventsCount(
 result.events?.length ?? 0
 )
 );
 return result;
 });

 /**
 * The slot snapshot for a write issued from this loop: how
 * much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.hook_received.create.start', {
 'workflow.hook_received.preload_events': true,
 });
 const result = await createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 const result = await traceReplayLoad('hook_preload', () =>
 createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 },
 },
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
 {
 requestId,
 occurredAt,
 resumeId: hookResumeInput.resumeId,
 resumePayloadDigest: hookResumeInput.payloadDigest,
 preloadEvents: true,
}
)
 );
 hookEnsured = true;
 // Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.run_started.create.start', {
 'workflow.run_started.skip_preload': false,
 });
 const result = await createEvent(runStartedEvent, {
 requestId,
 });
 const result = await traceReplayLoad(
 'run_started',
 () => createEvent(runStartedEvent, { requestId })
 );
 workflowRun = result.run;
 maxEventsLimit = clampMaxEvents(result.maxEvents);
 // Anchors RSFS, see the declaration above.
Expand Down

184 changes: 92 additions & 92 deletions
packages/core/src/runtime/helpers.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
 afterCursor?: string
): Promise<LoadedEventLog> {
 const incremental = afterCursor !== undefined;
 return trace(
 incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
 async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 return trace('workflow.replay.load', async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(
 incremental ? 'events_list_incremental' : 'events_list'
 ),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
 hasMore,
 pageMs: Date.now() - pageStart,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 runtimeLogger.debug('Event load complete', {
 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
 });
 }

 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });
 runtimeLogger.debug('Event load complete', {
 workflowRunId: runId,
 incremental,
 totalEvents: loadedEvents.length,
 pagesLoaded,
 totalMs: Date.now() - loadStart,
 });

 return { events: loadedEvents, cursor };
 }
 );
 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });

 return { events: loadedEvents, cursor };
 });
}

/**
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

NathanColosimo
merged 5 commits into

from

Aug 31, 2026
 Merged


[core] Trace fresh workflow replay phases
#3797
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
44897b7

Trace fresh workflow replay phases


NathanColosimo Aug 14, 2026
2f4d8d4

refactor(core): simplify workflow script cache API


NathanColosimo Aug 15, 2026
1bd9c04

Fix retained workflow tracing


NathanColosimo Aug 18, 2026
532197a

fix(core): keep tracing failure-safe


NathanColosimo Aug 20, 2026
2f331e9

Trace replay event loading


NathanColosimo Aug 25, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


5 changes: 5 additions & 0 deletions
.changeset/trace-replay-phases.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

13 changes: 11 additions & 2 deletions
packages/core/src/runtime-trace-mode.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
 const getWorldSpan = exporter
 .getFinishedSpans()
 .find((s) => s.name === 'workflow.route.get_world');

 return {
 workflowSpan,
 routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
 runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
 ).toBe(false);

 const replayLoadSpan = exporter
 .getFinishedSpans()
 .find((finished) => finished.name === 'workflow.replay.load');
 expect(replayLoadSpan?.parentSpanId).toBe(
 workflowSpan?.spanContext().spanId
 );
 expect(replayLoadSpan?.attributes).toMatchObject({
 'workflow.replay.load.source': 'run_started',
 'workflow.events.count': 0,
 });

 // Queue-delivered invocation spans use the CONSUMER kind, matching
 // queue-delivered step.execute spans.
 expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down

59 changes: 40 additions & 19 deletions
packages/core/src/runtime.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
 return result;
 };

 const traceReplayLoad = <T extends { events?: Event[] }>(
 source: Attribute.WorkflowReplayLoadSource,
 load: () => Promise<T>
 ): Promise<T> =>
 trace('workflow.replay.load', async (loadSpan) => {
 loadSpan?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(source),
 });
 const result = await load();
 loadSpan?.setAttributes(
 Attribute.WorkflowEventsCount(
 result.events?.length ?? 0
 )
 );
 return result;
 });

 /**
 * The slot snapshot for a write issued from this loop: how
 * much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.hook_received.create.start', {
 'workflow.hook_received.preload_events': true,
 });
 const result = await createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 const result = await traceReplayLoad('hook_preload', () =>
 createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 },
 },
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
 {
 requestId,
 occurredAt,
 resumeId: hookResumeInput.resumeId,
 resumePayloadDigest: hookResumeInput.payloadDigest,
 preloadEvents: true,
}
)
 );
 hookEnsured = true;
 // Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.run_started.create.start', {
 'workflow.run_started.skip_preload': false,
 });
 const result = await createEvent(runStartedEvent, {
 requestId,
 });
 const result = await traceReplayLoad(
 'run_started',
 () => createEvent(runStartedEvent, { requestId })
 );
 workflowRun = result.run;
 maxEventsLimit = clampMaxEvents(result.maxEvents);
 // Anchors RSFS, see the declaration above.
Expand Down

184 changes: 92 additions & 92 deletions
packages/core/src/runtime/helpers.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
 afterCursor?: string
): Promise<LoadedEventLog> {
 const incremental = afterCursor !== undefined;
 return trace(
 incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
 async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 return trace('workflow.replay.load', async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(
 incremental ? 'events_list_incremental' : 'events_list'
 ),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
 hasMore,
 pageMs: Date.now() - pageStart,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 runtimeLogger.debug('Event load complete', {
 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
 });
 }

 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });
 runtimeLogger.debug('Event load complete', {
 workflowRunId: runId,
 incremental,
 totalEvents: loadedEvents.length,
 pagesLoaded,
 totalMs: Date.now() - loadStart,
 });

 return { events: loadedEvents, cursor };
 }
 );
 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });

 return { events: loadedEvents, cursor };
 });
}

/**
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

NathanColosimo
merged 5 commits into

from

Aug 31, 2026
 Merged


[core] Trace fresh workflow replay phases
#3797
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
44897b7

Trace fresh workflow replay phases


NathanColosimo Aug 14, 2026
2f4d8d4

refactor(core): simplify workflow script cache API


NathanColosimo Aug 15, 2026
1bd9c04

Fix retained workflow tracing


NathanColosimo Aug 18, 2026
532197a

fix(core): keep tracing failure-safe


NathanColosimo Aug 20, 2026
2f331e9

Trace replay event loading


NathanColosimo Aug 25, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


5 changes: 5 additions & 0 deletions
.changeset/trace-replay-phases.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

13 changes: 11 additions & 2 deletions
packages/core/src/runtime-trace-mode.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
 const getWorldSpan = exporter
 .getFinishedSpans()
 .find((s) => s.name === 'workflow.route.get_world');

 return {
 workflowSpan,
 routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
 runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
 ).toBe(false);

 const replayLoadSpan = exporter
 .getFinishedSpans()
 .find((finished) => finished.name === 'workflow.replay.load');
 expect(replayLoadSpan?.parentSpanId).toBe(
 workflowSpan?.spanContext().spanId
 );
 expect(replayLoadSpan?.attributes).toMatchObject({
 'workflow.replay.load.source': 'run_started',
 'workflow.events.count': 0,
 });

 // Queue-delivered invocation spans use the CONSUMER kind, matching
 // queue-delivered step.execute spans.
 expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down

59 changes: 40 additions & 19 deletions
packages/core/src/runtime.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
 return result;
 };

 const traceReplayLoad = <T extends { events?: Event[] }>(
 source: Attribute.WorkflowReplayLoadSource,
 load: () => Promise<T>
 ): Promise<T> =>
 trace('workflow.replay.load', async (loadSpan) => {
 loadSpan?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(source),
 });
 const result = await load();
 loadSpan?.setAttributes(
 Attribute.WorkflowEventsCount(
 result.events?.length ?? 0
 )
 );
 return result;
 });

 /**
 * The slot snapshot for a write issued from this loop: how
 * much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.hook_received.create.start', {
 'workflow.hook_received.preload_events': true,
 });
 const result = await createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 const result = await traceReplayLoad('hook_preload', () =>
 createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 },
 },
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
 {
 requestId,
 occurredAt,
 resumeId: hookResumeInput.resumeId,
 resumePayloadDigest: hookResumeInput.payloadDigest,
 preloadEvents: true,
}
)
 );
 hookEnsured = true;
 // Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.run_started.create.start', {
 'workflow.run_started.skip_preload': false,
 });
 const result = await createEvent(runStartedEvent, {
 requestId,
 });
 const result = await traceReplayLoad(
 'run_started',
 () => createEvent(runStartedEvent, { requestId })
 );
 workflowRun = result.run;
 maxEventsLimit = clampMaxEvents(result.maxEvents);
 // Anchors RSFS, see the declaration above.
Expand Down

184 changes: 92 additions & 92 deletions
packages/core/src/runtime/helpers.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
 afterCursor?: string
): Promise<LoadedEventLog> {
 const incremental = afterCursor !== undefined;
 return trace(
 incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
 async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 return trace('workflow.replay.load', async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(
 incremental ? 'events_list_incremental' : 'events_list'
 ),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
 hasMore,
 pageMs: Date.now() - pageStart,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 runtimeLogger.debug('Event load complete', {
 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
 });
 }

 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });
 runtimeLogger.debug('Event load complete', {
 workflowRunId: runId,
 incremental,
 totalEvents: loadedEvents.length,
 pagesLoaded,
 totalMs: Date.now() - loadStart,
 });

 return { events: loadedEvents, cursor };
 }
 );
 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });

 return { events: loadedEvents, cursor };
 });
}

/**
Expand Down
Loading
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
[core] Trace fresh workflow replay phases#3797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

NathanColosimo
merged 5 commits into

from

Aug 31, 2026
 Merged


[core] Trace fresh workflow replay phases
#3797
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
44897b7

Trace fresh workflow replay phases


NathanColosimo Aug 14, 2026
2f4d8d4

refactor(core): simplify workflow script cache API


NathanColosimo Aug 15, 2026
1bd9c04

Fix retained workflow tracing


NathanColosimo Aug 18, 2026
532197a

fix(core): keep tracing failure-safe


NathanColosimo Aug 20, 2026
2f331e9

Trace replay event loading


NathanColosimo Aug 25, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


5 changes: 5 additions & 0 deletions
.changeset/trace-replay-phases.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Trace event loading, workflow VM creation, bundle compilation and evaluation, input hydration, and replay execution.

13 changes: 11 additions & 2 deletions
packages/core/src/runtime-trace-mode.test.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,6 @@ async function driveHandler(opts: {
 const getWorldSpan = exporter
 .getFinishedSpans()
 .find((s) => s.name === 'workflow.route.get_world');

 return {
 workflowSpan,
 routeSpan,
Expand DownExpand Up@@ -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.
Expand DownExpand Up@@ -316,6 +314,17 @@ describe('workflowEntrypoint trace modes', () => {
 runStartedCreateEvent?.attributes['workflow.run_started.skip_preload']
 ).toBe(false);

 const replayLoadSpan = exporter
 .getFinishedSpans()
 .find((finished) => finished.name === 'workflow.replay.load');
 expect(replayLoadSpan?.parentSpanId).toBe(
 workflowSpan?.spanContext().spanId
 );
 expect(replayLoadSpan?.attributes).toMatchObject({
 'workflow.replay.load.source': 'run_started',
 'workflow.events.count': 0,
 });

 // Queue-delivered invocation spans use the CONSUMER kind, matching
 // queue-delivered step.execute spans.
 expect(workflowSpan?.kind).toBe(SpanKind.CONSUMER);
Expand Down

59 changes: 40 additions & 19 deletions
packages/core/src/runtime.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -969,6 +969,24 @@ export function workflowEntrypoint(
 return result;
 };

 const traceReplayLoad = <T extends { events?: Event[] }>(
 source: Attribute.WorkflowReplayLoadSource,
 load: () => Promise<T>
 ): Promise<T> =>
 trace('workflow.replay.load', async (loadSpan) => {
 loadSpan?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(source),
 });
 const result = await load();
 loadSpan?.setAttributes(
 Attribute.WorkflowEventsCount(
 result.events?.length ?? 0
 )
 );
 return result;
 });

 /**
 * The slot snapshot for a write issued from this loop: how
 * much of the run's log the decision behind it was made
Expand DownExpand Up@@ -2034,23 +2052,25 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.hook_received.create.start', {
 'workflow.hook_received.preload_events': true,
 });
 const result = await createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 const result = await traceReplayLoad('hook_preload', () =>
 createEvent(
 {
 eventType: 'hook_received',
 specVersion: SPEC_VERSION_CURRENT,
 correlationId: hookResumeInput.hookId,
 eventData: {
 token: hookResumeInput.token,
 payload: hookResumeInput.payload,
 },
 },
},
{
requestId,
occurredAt,
resumeId: hookResumeInput.resumeId,
resumePayloadDigest: hookResumeInput.payloadDigest,
preloadEvents: true,
}
 {
 requestId,
 occurredAt,
 resumeId: hookResumeInput.resumeId,
 resumePayloadDigest: hookResumeInput.payloadDigest,
 preloadEvents: true,
}
)
 );
 hookEnsured = true;
 // Note: unlike the re-ensure below, this hoisted write
Expand DownExpand Up@@ -2324,9 +2344,10 @@ export function workflowEntrypoint(
 span?.addEvent('workflow.run_started.create.start', {
 'workflow.run_started.skip_preload': false,
 });
 const result = await createEvent(runStartedEvent, {
 requestId,
 });
 const result = await traceReplayLoad(
 'run_started',
 () => createEvent(runStartedEvent, { requestId })
 );
 workflowRun = result.run;
 maxEventsLimit = clampMaxEvents(result.maxEvents);
 // Anchors RSFS, see the declaration above.
Expand Down

184 changes: 92 additions & 92 deletions
packages/core/src/runtime/helpers.ts

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -597,108 +597,108 @@ export async function loadWorkflowRunEvents(
 afterCursor?: string
): Promise<LoadedEventLog> {
 const incremental = afterCursor !== undefined;
 return trace(
 incremental ? 'workflow.loadNewEvents' : 'workflow.loadEvents',
 async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 return trace('workflow.replay.load', async (span) => {
 span?.setAttributes({
 ...Attribute.WorkflowRunId(runId),
 ...Attribute.WorkflowReplayLoadSource(
 incremental ? 'events_list_incremental' : 'events_list'
 ),
 });

 const loadedEvents: Event[] = [];
 const loadedEventIds = new Set<string>();
 const requestedCursors = new Set<string>();
 let cursor: string | null = afterCursor ?? null;
 let hasMore = true;
 let pagesLoaded = 0;
 let retriedWithoutCursor = false;

 const world = await getWorldLazy();
 const loadStart = Date.now();
 while (hasMore) {
 // TODO: we're currently loading all the data with resolveRef behavior. We need to update this
 // to lazyload the data from the world instead so that we can optimize and make the event log loading
 // much faster and memory efficient
 const pageStart = Date.now();
 const requestedCursor = cursor;
 recordRequestedEventCursor(runId, requestedCursor, requestedCursors);

 let response: Awaited<ReturnType<typeof world.events.list>>;
 try {
 response = await world.events.list({
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
 hasMore,
 pageMs: Date.now() - pageStart,
 pagination: {
 sortOrder: 'asc',
 cursor: requestedCursor ?? undefined,
 },
 });
 } catch (error) {
 if (
 shouldRetryWithoutEventCursor(
 error,
 requestedCursor,
 retriedWithoutCursor
 )
 ) {
 runtimeLogger.warn(
 'Event cursor was rejected; retrying with a full event reload.',
 { workflowRunId: runId }
 );
 loadedEvents.length = 0;
 loadedEventIds.clear();
 requestedCursors.clear();
 cursor = null;
 retriedWithoutCursor = true;
 continue;
 }
 throw error;
 }

 runtimeLogger.debug('Event load complete', {
 appendUniqueEvents(loadedEvents, response.data, loadedEventIds);
 hasMore = response.hasMore;
 assertEventPaginationProgress(
 runId,
 hasMore,
 response.cursor,
 requestedCursors
 );
 // 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 DB query hit the limit exactly and returned a
 // precautionary `LastEvaluatedKey`. 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-
 // reload fallback on every subsequent replay iteration.
 cursor = response.cursor ?? cursor;
 pagesLoaded++;

 runtimeLogger.debug('Loaded event page', {
 workflowRunId: runId,
 incremental,
 page: pagesLoaded,
 pageEvents: response.data.length,
 totalEvents: loadedEvents.length,
pagesLoaded,
totalMs: Date.now() - loadStart,
hasMore,
pageMs: Date.now() - pageStart,
 });
 }

 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });
 runtimeLogger.debug('Event load complete', {
 workflowRunId: runId,
 incremental,
 totalEvents: loadedEvents.length,
 pagesLoaded,
 totalMs: Date.now() - loadStart,
 });

 return { events: loadedEvents, cursor };
 }
 );
 span?.setAttributes({
 ...Attribute.WorkflowEventsCount(loadedEvents.length),
 ...Attribute.WorkflowEventsPagesLoaded(pagesLoaded),
 });

 return { events: loadedEvents, cursor };
 });
}

/**
Expand Down
Loading
Loading