@@ -338,6 +338,22 @@ async function run(
338338 return data
339339}
340340
341+ /** Source runs must finish deferred log persistence before a later run can reuse their state. */
342+ async function waitForPersistedRun ( workflowId : string , runId : string ) : Promise < void > {
343+ const deadline = performance . now ( ) + REQUEST_TIMEOUT_MS
344+ while ( performance . now ( ) < deadline ) {
345+ const [ saved ] = await sql `
346+ select status, ended_at from workflow_execution_logs
347+ where workflow_id = ${ workflowId } and execution_id = ${ runId }
348+ `
349+ if ( saved ?. status === 'completed' && saved . ended_at ) return
350+ assert . notEqual ( saved ?. status , 'failed' , `Source run ${ runId } failed while persisting` )
351+ assert . notEqual ( saved ?. status , 'cancelled' , `Source run ${ runId } was cancelled` )
352+ await sleep ( 100 )
353+ }
354+ throw new Error ( `Source run ${ runId } did not persist within ${ REQUEST_TIMEOUT_MS } ms` )
355+ }
356+
341357async function pause ( fixture : PipelineFixture , stopAfterBlockId ?: string ) {
342358 const result = v2ExecuteWorkflowDataSchema . parse (
343359 record (
@@ -525,6 +541,7 @@ try {
525541 `a full run must include Slow's ${ SLOW_MS } ms, took ${ full . durationMs } ms`
526542 )
527543 sourceRunId = full . runId
544+ await waitForPersistedRun ( pipeline . workflowId , sourceRunId )
528545 } )
529546
530547 await check (
@@ -798,6 +815,7 @@ try {
798815
799816 await check ( 'a source run from another workflow is not accepted' , async ( ) => {
800817 const foreign = await run ( otherPipeline . workflowId , { run : { source : 'manual' } } )
818+ await waitForPersistedRun ( otherPipeline . workflowId , foreign . runId )
801819 await expectBadRequest (
802820 pipeline . workflowId ,
803821 {
0 commit comments