From da0a297ee21cf09365b538a405074817325f938e Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 14:00:12 +0000 Subject: [PATCH 1/5] fix: fence workflow reconciliation effects --- ...class-wp-agent-workflow-reconcile-lock.php | 21 +- .../register-reconcile-workflow-branch.php | 195 +++++++++++++----- tests/workflow-reconcile-race-smoke.php | 74 ++++++- 3 files changed, 226 insertions(+), 64 deletions(-) diff --git a/src/Workflows/class-wp-agent-workflow-reconcile-lock.php b/src/Workflows/class-wp-agent-workflow-reconcile-lock.php index 21f3600..56e7c91 100644 --- a/src/Workflows/class-wp-agent-workflow-reconcile-lock.php +++ b/src/Workflows/class-wp-agent-workflow-reconcile-lock.php @@ -1,10 +1,10 @@ resume( $run_id ); } /** - * The reconcile critical section, run under the per-run lock. Loads the run FRESH - * (inside the lock), merges the branch, decides all-terminal, and aggregates + - * dispatches resume when it was the last branch. Extracted from - * {@see agents_reconcile_workflow_branch()} so the lock wraps exactly the - * load-modify-decide window and nothing more. + * The reconcile state transition run under the per-run lock. It loads the run, + * merges one branch, and claims aggregation for the suspension generation when + * all branches are terminal. Aggregation itself runs after this function returns + * and the lock has been released. * * @since 0.5.0 * @@ -156,7 +185,7 @@ static function () use ( $recorder, $run_id, $handle_id, $branch_result ) { * @param string $run_id Suspended run id. * @param string $handle_id The completed branch's handle id. * @param array $branch_result BranchResult. - * @return WP_Agent_Workflow_Run_Result|\WP_Error + * @return WP_Agent_Workflow_Run_Result|array{claim_token:string,generation:string,step_index:int,aggregate:array,branch_results:array,required_failed:bool}|\WP_Error */ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $handle_id, array $branch_result ) { $result = $recorder->find( $run_id ); @@ -251,6 +280,23 @@ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder return $result; } + // A durable claim, rather than the expiring option lock, owns unbounded + // aggregation and resume dispatch. A duplicate reconciler for this exact + // suspension generation observes the claim and cannot execute either effect. + if ( is_array( $suspension['reconcile_claim'] ?? null ) ) { + return $result; + } + + $generation = agents_workflow_suspension_generation( $suspension ); + $claim_token = agents_workflow_reconcile_claim_token(); + $suspension['reconcile_claim'] = array( + 'token' => $claim_token, + 'generation' => $generation, + ); + $metadata['_suspension'] = $suspension; + $result = $result->with( array( 'metadata' => $metadata ) ); + $recorder->update( $result ); + // All branches terminal. Was a REQUIRED branch failed? A required-branch // failure fails the parallel step, which re-enters the failure path on // resume (mirrors the sync `run_role_branch()` required-branch rule). @@ -259,40 +305,91 @@ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder $step_index = is_numeric( $suspension['step_index'] ?? null ) ? (int) $suspension['step_index'] : 0; $aggregate = is_array( $suspension['aggregate'] ?? null ) ? \AgentsAPI\AI\WP_Agent_Run_Control::string_keyed_array( $suspension['aggregate'] ) : array(); - $handlers = agents_workflow_resolve_step_handlers(); + return array( + 'claim_token' => $claim_token, + 'generation' => $generation, + 'step_index' => $step_index, + 'aggregate' => $aggregate, + 'branch_results' => $branch_results, + 'required_failed' => $required_failed, + ); +} - if ( $required_failed ) { - $step_output = new \WP_Error( 'workflow_parallel_required_branch_failed', 'A required parallel branch failed during out-of-band execution.' ); - } else { - $step_output = WP_Agent_Workflow_Runner::aggregate_branch_results( $aggregate, $branch_results, $handlers ); +/** + * Commit an aggregate result only while its generation-bound claim still owns + * the suspended run. The fresh read is the fence that prevents an expired former + * option-lock holder from overwriting newer recorder state. + * + * @since 0.5.0 + * + * @param WP_Agent_Workflow_Run_Recorder $recorder Resolved recorder. + * @param string $run_id Suspended run id. + * @param string $claim_token Claim token elected before aggregation. + * @param string $generation Suspension generation identity. + * @param int $step_index Suspended parallel step index. + * @param array|\WP_Error $step_output Aggregated output or failure. + * @return WP_Agent_Workflow_Run_Result|array{result:WP_Agent_Workflow_Run_Result}|\WP_Error + */ +function agents_workflow_commit_reconcile_claim( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $claim_token, string $generation, int $step_index, $step_output ) { + $result = $recorder->find( $run_id ); + if ( null === $result || ! $result->is_suspended() ) { + return null === $result + ? new \WP_Error( 'agents_reconcile_workflow_branch_not_found', sprintf( 'No suspended run was found for run_id `%s`.', $run_id ) ) + : $result; + } + + $suspension = $result->get_suspension(); + $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); + if ( + ! hash_equals( $claim_token, agents_workflow_string( $claim['token'] ?? '' ) ) || + ! hash_equals( $generation, agents_workflow_string( $claim['generation'] ?? '' ) ) || + ! hash_equals( $generation, agents_workflow_suspension_generation( $suspension ) ) + ) { + return $result; } - // Splice the parallel step's final output (or failure) into its record so - // resume sees a terminal step and downstream `${steps..output}` - // bindings resolve against the aggregated result. The run is still - // SUSPENDED at this point (the frame is intact); resume() is what clears it. $result = agents_workflow_splice_step_output( $result, $step_index, $step_output ); $recorder->update( $result ); + return array( 'result' => $result ); +} - // The "all terminal → resume" transition is the ONE place two branches - // finishing near-simultaneously in separate processes can race. Rather than - // hand-roll a cross-process lock (unsafe / forbidden), the transition is - // pluggable: the owning executor may DEFER resume to an atomically-claimed - // out-of-band action so exactly one resume runs. The default (Phase 1, and - // any synchronous / in-process executor) resumes inline right here. - // - // A deferring handler enqueues its claimed resume action and returns true; - // reconcile then returns the aggregate-spliced-but-still-SUSPENDED run. The - // deferred handler, when its claimed action fires, re-checks the run is - // still SUSPENDED and calls resume() exactly once (§3.4, §4.3). - if ( agents_workflow_defer_resume( $run_id, $result ) ) { - return $result; +/** + * Derive the stable identity of one suspension generation. + * + * @param array $suspension Suspension frame. + */ +function agents_workflow_suspension_generation( array $suspension ): string { + $handles = is_array( $suspension['handles'] ?? null ) ? $suspension['handles'] : array(); + $ids = array(); + foreach ( $handles as $handle ) { + if ( is_array( $handle ) ) { + $ids[] = agents_workflow_string( $handle['id'] ?? '' ); + } } - // Resume the step loop from step_index + 1 inline. resume() clears the - // suspension frame and continues (or fails) from the aggregated output. - $runner = agents_workflow_resolve_runner( $recorder ); - return $runner->resume( $run_id ); + return hash( + 'sha256', + serialize( + array( + 'step_index' => is_numeric( $suspension['step_index'] ?? null ) ? (int) $suspension['step_index'] : 0, + 'step_id' => agents_workflow_string( $suspension['step_id'] ?? '' ), + 'handles' => $ids, + ) + ) + ); +} + +/** Mint an opaque owner token for one reconcile claim. */ +function agents_workflow_reconcile_claim_token(): string { + if ( function_exists( 'wp_generate_uuid4' ) ) { + return wp_generate_uuid4(); + } + try { + return bin2hex( random_bytes( 16 ) ); + } catch ( \Throwable $error ) { + unset( $error ); + return uniqid( 'reconcile_', true ); + } } /** diff --git a/tests/workflow-reconcile-race-smoke.php b/tests/workflow-reconcile-race-smoke.php index c36ce5c..f87152b 100644 --- a/tests/workflow-reconcile-race-smoke.php +++ b/tests/workflow-reconcile-race-smoke.php @@ -236,10 +236,14 @@ public function update( WP_Agent_Workflow_Run_Result $result ) { return true; } public function find( string $run_id ): ?WP_Agent_Workflow_Run_Result { - // Serve the stale pre-merge snapshot only in the unlocked window; a held - // reconcile lock means a real second process would have blocked and read - // fresh, so honor that ordering here too. - if ( null !== $this->frozen && isset( $this->frozen[ $run_id ] ) && ! self::reconcile_lock_held( $run_id ) ) { + // Serve the stale pre-merge snapshot only before serialization or durable + // generation ownership establishes the fresh state ordering. + if ( + null !== $this->frozen && + isset( $this->frozen[ $run_id ] ) && + ! self::reconcile_lock_held( $run_id ) && + ! $this->reconcile_claim_held( $run_id ) + ) { return WP_Agent_Workflow_Run_Result::from_array( $this->frozen[ $run_id ] ); } return isset( $this->rows[ $run_id ] ) @@ -265,6 +269,13 @@ private static function reconcile_lock_held( string $run_id ): bool { $option = 'agents_wf_reconcile_lock_' . md5( $run_id ); return array_key_exists( $option, $GLOBALS['__options'] ); } + + /** Whether the current suspension generation has elected its effect owner. */ + private function reconcile_claim_held( string $run_id ): bool { + $metadata = $this->rows[ $run_id ]['metadata'] ?? array(); + $suspension = is_array( $metadata ) && is_array( $metadata['_suspension'] ?? null ) ? $metadata['_suspension'] : array(); + return is_array( $suspension['reconcile_claim'] ?? null ); + } } function race_register_ability( string $name, \Closure $handler ): void { @@ -280,6 +291,10 @@ static function ( array $input ): array { race_register_ability( 'demo/aggregate', static function ( array $input ): array { + ++$GLOBALS['__aggregate_calls']; + if ( is_callable( $GLOBALS['__during_aggregate'] ?? null ) ) { + call_user_func( $GLOBALS['__during_aggregate'] ); + } // Fuse ALL sibling fragments; a lost completion shows up as a blank slot. return array( 'final_bundle' => 'FUSED[' . (string) ( $input['a'] ?? '' ) . '|' . (string) ( $input['b'] ?? '' ) . '|' . (string) ( $input['c'] ?? '' ) . ']' ); } @@ -409,6 +424,8 @@ function race_execute_branch( array $descriptor ): array { // ═════════════════════════════════════════════════════════════════════════════ $GLOBALS['__options'] = array(); +$GLOBALS['__aggregate_calls'] = 0; +$GLOBALS['__during_aggregate'] = null; $recorder = new Race_Recorder(); remove_all_filters( 'wp_agent_workflow_run_recorder' ); remove_all_filters( 'wp_agent_workflow_step_executor' ); @@ -420,6 +437,13 @@ function race_execute_branch( array $descriptor ): array { // the bug is the completed[] accounting, not the resume-dedup guard: even with a // perfectly working resume path, a lost completion means resume is never reached. $GLOBALS['__resume_dispatch_calls'] = 0; +add_filter( + 'wp_agent_workflow_resume_dispatch', + static function ( bool $deferred ): bool { + ++$GLOBALS['__resume_dispatch_calls']; + return $deferred; + } +); $run = ( new WP_Agent_Workflow_Runner( $recorder ) )->run( race_roles_spec(), array(), array( 'run_id' => 'race-1' ) ); smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED, $run->get_status(), 'race: run SUSPENDED after dispatch', $failures, $passes ); @@ -451,6 +475,48 @@ function race_execute_branch( array $descriptor ): array { $bundle = $final->get_output()['steps']['scatter']['final']['final_bundle'] ?? ''; smoke_assert( 'FUSED[A|B|C]', $bundle, 'race: aggregate fused ALL branch outputs — no completion lost', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'race: aggregator executes exactly once', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__resume_dispatch_calls'], 'race: resume dispatch executes exactly once', $failures, $passes ); + +// ── TTL OVERRUN: aggregation runs outside the short option lock and remains +// generation-fenced after more than the old 60-second TTL. The callback installs +// an expired lock row (representing the former holder after 61 seconds), then a +// duplicate reconciler reclaims it. The durable claim must keep that reconciler +// from aggregating or dispatching a competing resume. +$GLOBALS['__options'] = array(); +$GLOBALS['__aggregate_calls'] = 0; +$GLOBALS['__resume_dispatch_calls'] = 0; +$recorder = new Race_Recorder(); +remove_all_filters( 'wp_agent_workflow_run_recorder' ); +add_filter( 'wp_agent_workflow_run_recorder', static function () use ( $recorder ) { return $recorder; } ); + +$run = ( new WP_Agent_Workflow_Runner( $recorder ) )->run( race_roles_spec(), array(), array( 'run_id' => 'ttl-1' ) ); +$descriptors = Race_Executor::$dispatched; +foreach ( array( 0, 1 ) as $index ) { + $branch = race_execute_branch( $descriptors[ $index ] ); + agents_reconcile_workflow_branch( 'ttl-1', (string) $descriptors[ $index ]['handle_id'], $branch ); +} +$last_result = race_execute_branch( $descriptors[2] ); +$GLOBALS['__aggregation_started_unlocked'] = false; +$GLOBALS['__during_aggregate'] = static function () use ( $descriptors, $last_result ): void { + $GLOBALS['__during_aggregate'] = null; + $option = 'agents_wf_reconcile_lock_' . md5( 'ttl-1' ); + $GLOBALS['__aggregation_started_unlocked'] = ! array_key_exists( $option, $GLOBALS['__options'] ); + + // Advance the lock boundary beyond the former 60-second TTL without sleeping. + $GLOBALS['__options'][ $option ] = array( + 'token' => 'expired-former-holder', + 'expires' => time() - 1, + ); + agents_reconcile_workflow_branch( 'ttl-1', (string) $descriptors[2]['handle_id'], $last_result ); +}; + +agents_reconcile_workflow_branch( 'ttl-1', (string) $descriptors[2]['handle_id'], $last_result ); +$final = $recorder->find( 'ttl-1' ); +smoke_assert( true, $GLOBALS['__aggregation_started_unlocked'], 'ttl: aggregation starts after the short option lock is released', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'ttl: one aggregator executes after a competing reconciler reclaims the expired lock', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__resume_dispatch_calls'], 'ttl: one resume dispatches for the claimed suspension generation', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $final->get_status(), 'ttl: claimed owner commits and resumes successfully', $failures, $passes ); echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; exit( count( $failures ) > 0 ? 1 : 0 ); From 2cf5d19d0966b4655799da480a2bbf6492380ebd Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 17:16:02 +0000 Subject: [PATCH 2/5] fix: recover workflow reconcile continuations --- .../register-reconcile-workflow-branch.php | 358 +++++++++++++----- tests/workflow-reconcile-race-smoke.php | 139 +++++++ 2 files changed, 401 insertions(+), 96 deletions(-) diff --git a/src/Workflows/register-reconcile-workflow-branch.php b/src/Workflows/register-reconcile-workflow-branch.php index 379c880..4a914b1 100644 --- a/src/Workflows/register-reconcile-workflow-branch.php +++ b/src/Workflows/register-reconcile-workflow-branch.php @@ -23,6 +23,9 @@ const AGENTS_RECONCILE_WORKFLOW_BRANCH_ABILITY = 'agents/reconcile-workflow-branch'; +/** Effect-owner lease used only to decide when an ambiguous aggregation must fail closed. */ +const AGENTS_RECONCILE_CLAIM_TTL_SECONDS = 60; + add_action( 'wp_abilities_api_init', static function (): void { @@ -138,6 +141,22 @@ static function () use ( $recorder, $run_id, $handle_id, $branch_result ) { if ( is_wp_error( $transition ) || $transition instanceof WP_Agent_Workflow_Run_Result ) { return $transition; } + if ( 'resume' === $transition['action'] ) { + return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $transition['result'] ); + } + + $transition = agents_workflow_reconcile_with_lock( + $run_id, + static function () use ( $recorder, $run_id, $transition ) { + return agents_workflow_begin_reconcile_aggregation( $recorder, $run_id, $transition['generation'] ); + } + ); + if ( is_wp_error( $transition ) || $transition instanceof WP_Agent_Workflow_Run_Result ) { + return $transition; + } + if ( 'resume' === $transition['action'] ) { + return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $transition['result'] ); + } $required_failed = ! empty( $transition['required_failed'] ); if ( $required_failed ) { @@ -153,7 +172,7 @@ static function () use ( $recorder, $run_id, $transition, $step_output ) { return agents_workflow_commit_reconcile_claim( $recorder, $run_id, - $transition['claim_token'], + $transition['owner_token'], $transition['generation'], $transition['step_index'], $step_output @@ -164,13 +183,7 @@ static function () use ( $recorder, $run_id, $transition, $step_output ) { return $commit; } - $result = $commit['result']; - if ( agents_workflow_defer_resume( $run_id, $result ) ) { - return $result; - } - - $runner = agents_workflow_resolve_runner( $recorder ); - return $runner->resume( $run_id ); + return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $commit['result'] ); } /** @@ -185,7 +198,7 @@ static function () use ( $recorder, $run_id, $transition, $step_output ) { * @param string $run_id Suspended run id. * @param string $handle_id The completed branch's handle id. * @param array $branch_result BranchResult. - * @return WP_Agent_Workflow_Run_Result|array{claim_token:string,generation:string,step_index:int,aggregate:array,branch_results:array,required_failed:bool}|\WP_Error + * @return WP_Agent_Workflow_Run_Result|array{action:'begin',generation:string}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error */ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $handle_id, array $branch_result ) { $result = $recorder->find( $run_id ); @@ -204,10 +217,8 @@ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder /** @var array $completed */ $completed = is_array( $suspension['completed'] ?? null ) ? \AgentsAPI\AI\WP_Agent_Run_Control::string_keyed_array( $suspension['completed'] ) : array(); - // Idempotency: a handle already recorded terminal does not re-merge or - // re-resume. Return the current run untouched. - if ( isset( $completed[ $handle_id ] ) ) { - return $result; + if ( is_array( $suspension['reconcile_claim'] ?? null ) ) { + return agents_workflow_advance_reconcile_continuation_locked( $recorder, $result ); } // Bind the completion to server-stored suspension state: only a handle id @@ -215,104 +226,84 @@ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder // caller-asserted handle id that is not among the stored handles is rejected // (fail closed) so a forged/unknown id cannot inflate the completed[] // accounting and prematurely trip the all-terminal gate below. - $stored_handle = null; - foreach ( $handles as $handle ) { - if ( is_array( $handle ) && agents_workflow_string( $handle['id'] ?? '' ) === $handle_id ) { - $stored_handle = $handle; - break; + if ( ! isset( $completed[ $handle_id ] ) ) { + $stored_handle = null; + foreach ( $handles as $handle ) { + if ( is_array( $handle ) && agents_workflow_string( $handle['id'] ?? '' ) === $handle_id ) { + $stored_handle = $handle; + break; + } + } + if ( null === $stored_handle ) { + return new \WP_Error( + 'agents_reconcile_workflow_branch_unknown_handle', + sprintf( 'Handle id `%s` is not a known suspended branch of run `%s`.', $handle_id, $run_id ) + ); } - } - if ( null === $stored_handle ) { - return new \WP_Error( - 'agents_reconcile_workflow_branch_unknown_handle', - sprintf( 'Handle id `%s` is not a known suspended branch of run `%s`.', $handle_id, $run_id ) - ); - } // Bind the completion to the handle's stored key too: the caller may not // remap its output onto a different branch's aggregate key. An empty stored // key preserves compatibility with frames that did not stamp one; otherwise // the stored key remains authoritative when the caller omits it. - $stored_key = agents_workflow_string( $stored_handle['key'] ?? '' ); - $asserted_key = agents_workflow_string( $branch_result['key'] ?? '' ); - if ( '' !== $stored_key && '' !== $asserted_key && $stored_key !== $asserted_key ) { - return new \WP_Error( - 'agents_reconcile_workflow_branch_key_mismatch', - sprintf( 'branch_result key `%s` does not match the stored key `%s` for handle `%s`.', $asserted_key, $stored_key, $handle_id ) - ); - } + $stored_key = agents_workflow_string( $stored_handle['key'] ?? '' ); + $asserted_key = agents_workflow_string( $branch_result['key'] ?? '' ); + if ( '' !== $stored_key && '' !== $asserted_key && $stored_key !== $asserted_key ) { + return new \WP_Error( + 'agents_reconcile_workflow_branch_key_mismatch', + sprintf( 'branch_result key `%s` does not match the stored key `%s` for handle `%s`.', $asserted_key, $stored_key, $handle_id ) + ); + } - $status = agents_workflow_string( $branch_result['status'] ?? '' ); - if ( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED !== $status && WP_Agent_Workflow_Run_Result::STATUS_FAILED !== $status ) { - return new \WP_Error( - 'agents_reconcile_workflow_branch_invalid_status', - sprintf( 'Branch result status `%s` is not terminal for handle `%s`.', $status, $handle_id ) - ); - } + $status = agents_workflow_string( $branch_result['status'] ?? '' ); + if ( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED !== $status && WP_Agent_Workflow_Run_Result::STATUS_FAILED !== $status ) { + return new \WP_Error( + 'agents_reconcile_workflow_branch_invalid_status', + sprintf( 'Branch result status `%s` is not terminal for handle `%s`.', $status, $handle_id ) + ); + } - $completed[ $handle_id ] = array( - 'key' => '' !== $stored_key ? $stored_key : $asserted_key, - 'status' => $status, - 'output' => $branch_result['output'] ?? null, - 'steps' => is_array( $branch_result['steps'] ?? null ) ? $branch_result['steps'] : array(), - 'error' => is_array( $branch_result['error'] ?? null ) ? $branch_result['error'] : null, - 'item' => $branch_result['item'] ?? null, - ); + $completed[ $handle_id ] = array( + 'key' => '' !== $stored_key ? $stored_key : $asserted_key, + 'status' => $status, + 'output' => $branch_result['output'] ?? null, + 'steps' => is_array( $branch_result['steps'] ?? null ) ? $branch_result['steps'] : array(), + 'error' => is_array( $branch_result['error'] ?? null ) ? $branch_result['error'] : null, + 'item' => $branch_result['item'] ?? null, + ); // Flip the matching handle's status. - foreach ( $handles as $index => $handle ) { - if ( is_array( $handle ) && agents_workflow_string( $handle['id'] ?? '' ) === $handle_id ) { - $handle['status'] = $status; - $handles[ $index ] = $handle; + foreach ( $handles as $index => $handle ) { + if ( is_array( $handle ) && agents_workflow_string( $handle['id'] ?? '' ) === $handle_id ) { + $handle['status'] = $status; + $handles[ $index ] = $handle; + } } } $suspension['handles'] = $handles; $suspension['completed'] = $completed; - $metadata = $result->get_metadata(); - $metadata['_suspension'] = $suspension; - $result = $result->with( array( 'metadata' => $metadata ) ); - $recorder->update( $result ); - - // Not all terminal yet → wait for more reconcile calls. - if ( count( $completed ) < count( $handles ) ) { - return $result; - } - - // A durable claim, rather than the expiring option lock, owns unbounded - // aggregation and resume dispatch. A duplicate reconciler for this exact - // suspension generation observes the claim and cannot execute either effect. - if ( is_array( $suspension['reconcile_claim'] ?? null ) ) { - return $result; + if ( count( $completed ) >= count( $handles ) ) { + $suspension['reconcile_claim'] = array( + 'phase' => 'pending', + 'generation' => agents_workflow_suspension_generation( $suspension ), + ); } - $generation = agents_workflow_suspension_generation( $suspension ); - $claim_token = agents_workflow_reconcile_claim_token(); - $suspension['reconcile_claim'] = array( - 'token' => $claim_token, - 'generation' => $generation, - ); + $metadata = $result->get_metadata(); $metadata['_suspension'] = $suspension; - $result = $result->with( array( 'metadata' => $metadata ) ); - $recorder->update( $result ); - - // All branches terminal. Was a REQUIRED branch failed? A required-branch - // failure fails the parallel step, which re-enters the failure path on - // resume (mirrors the sync `run_role_branch()` required-branch rule). - $branch_results = agents_workflow_branch_results_by_key( $completed ); - $required_failed = agents_workflow_required_branch_failed( $suspension, $completed ); + $result = $result->with( array( 'metadata' => $metadata ) ); + $updated = agents_workflow_update_reconcile_state( $recorder, $result, 'record branch completion and continuation claim' ); + if ( is_wp_error( $updated ) ) { + return $updated; + } - $step_index = is_numeric( $suspension['step_index'] ?? null ) ? (int) $suspension['step_index'] : 0; - $aggregate = is_array( $suspension['aggregate'] ?? null ) ? \AgentsAPI\AI\WP_Agent_Run_Control::string_keyed_array( $suspension['aggregate'] ) : array(); - return array( - 'claim_token' => $claim_token, - 'generation' => $generation, - 'step_index' => $step_index, - 'aggregate' => $aggregate, - 'branch_results' => $branch_results, - 'required_failed' => $required_failed, - ); + return count( $completed ) < count( $handles ) + ? $result + : array( + 'action' => 'begin', + 'generation' => agents_workflow_suspension_generation( $suspension ), + ); } /** @@ -324,11 +315,11 @@ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder * * @param WP_Agent_Workflow_Run_Recorder $recorder Resolved recorder. * @param string $run_id Suspended run id. - * @param string $claim_token Claim token elected before aggregation. + * @param string $claim_token Effect owner token elected before aggregation. * @param string $generation Suspension generation identity. * @param int $step_index Suspended parallel step index. * @param array|\WP_Error $step_output Aggregated output or failure. - * @return WP_Agent_Workflow_Run_Result|array{result:WP_Agent_Workflow_Run_Result}|\WP_Error + * @return WP_Agent_Workflow_Run_Result|array{result:WP_Agent_Workflow_Run_Result}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error */ function agents_workflow_commit_reconcile_claim( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $claim_token, string $generation, int $step_index, $step_output ) { $result = $recorder->find( $run_id ); @@ -341,18 +332,193 @@ function agents_workflow_commit_reconcile_claim( WP_Agent_Workflow_Run_Recorder $suspension = $result->get_suspension(); $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); if ( - ! hash_equals( $claim_token, agents_workflow_string( $claim['token'] ?? '' ) ) || + 'aggregating' !== agents_workflow_string( $claim['phase'] ?? '' ) || + ! hash_equals( $claim_token, agents_workflow_string( $claim['owner_token'] ?? '' ) ) || ! hash_equals( $generation, agents_workflow_string( $claim['generation'] ?? '' ) ) || ! hash_equals( $generation, agents_workflow_suspension_generation( $suspension ) ) ) { - return $result; + if ( 'pending' === agents_workflow_string( $claim['phase'] ?? '' ) ) { + return $result; + } + $continuation = agents_workflow_advance_reconcile_continuation_locked( $recorder, $result ); + if ( is_array( $continuation ) && 'begin' === $continuation['action'] ) { + return $result; + } + return $continuation; } + $suspension['reconcile_claim'] = array( + 'phase' => 'committed', + 'generation' => $generation, + ); + $metadata = $result->get_metadata(); + $metadata['_suspension'] = $suspension; + $result = $result->with( array( 'metadata' => $metadata ) ); $result = agents_workflow_splice_step_output( $result, $step_index, $step_output ); - $recorder->update( $result ); + $updated = agents_workflow_update_reconcile_state( $recorder, $result, 'commit aggregate output' ); + if ( is_wp_error( $updated ) ) { + return $updated; + } return array( 'result' => $result ); } +/** + * Persist the effect-start boundary for a pending continuation. A process lost + * before this transition is safe to retry; after it, aggregation is never rerun. + * + * @param WP_Agent_Workflow_Run_Recorder $recorder Resolved recorder. + * @param string $run_id Suspended run id. + * @param string $generation Suspension generation identity. + * @return WP_Agent_Workflow_Run_Result|array{action:'aggregate',owner_token:string,generation:string,step_index:int,aggregate:array,branch_results:array,required_failed:bool}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error + */ +function agents_workflow_begin_reconcile_aggregation( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $generation ) { + $result = $recorder->find( $run_id ); + if ( null === $result ) { + return new \WP_Error( 'agents_reconcile_workflow_branch_not_found', sprintf( 'No suspended run was found for run_id `%s`.', $run_id ) ); + } + if ( ! $result->is_suspended() ) { + return $result; + } + + $suspension = $result->get_suspension(); + $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); + if ( + 'pending' !== agents_workflow_string( $claim['phase'] ?? '' ) || + ! hash_equals( $generation, agents_workflow_string( $claim['generation'] ?? '' ) ) || + ! hash_equals( $generation, agents_workflow_suspension_generation( $suspension ) ) + ) { + if ( 'pending' === agents_workflow_string( $claim['phase'] ?? '' ) ) { + return $result; + } + $continuation = agents_workflow_advance_reconcile_continuation_locked( $recorder, $result ); + if ( is_array( $continuation ) && 'begin' === $continuation['action'] ) { + return $result; + } + return $continuation; + } + + $owner_token = agents_workflow_reconcile_claim_token(); + $suspension['reconcile_claim'] = array( + 'phase' => 'aggregating', + 'generation' => $generation, + 'owner_token' => $owner_token, + 'expires' => time() + AGENTS_RECONCILE_CLAIM_TTL_SECONDS, + ); + $metadata = $result->get_metadata(); + $metadata['_suspension'] = $suspension; + $result = $result->with( array( 'metadata' => $metadata ) ); + $updated = agents_workflow_update_reconcile_state( $recorder, $result, 'mark aggregation effects as started' ); + if ( is_wp_error( $updated ) ) { + return $updated; + } + + $completed = is_array( $suspension['completed'] ?? null ) ? \AgentsAPI\AI\WP_Agent_Run_Control::string_keyed_array( $suspension['completed'] ) : array(); + return array( + 'action' => 'aggregate', + 'owner_token' => $owner_token, + 'generation' => $generation, + 'step_index' => is_numeric( $suspension['step_index'] ?? null ) ? (int) $suspension['step_index'] : 0, + 'aggregate' => is_array( $suspension['aggregate'] ?? null ) ? \AgentsAPI\AI\WP_Agent_Run_Control::string_keyed_array( $suspension['aggregate'] ) : array(), + 'branch_results' => agents_workflow_branch_results_by_key( $completed ), + 'required_failed' => agents_workflow_required_branch_failed( $suspension, $completed ), + ); +} + +/** + * Advance an already-persisted continuation without relying on a new branch + * completion. Active effects fail retryably; stale ambiguous effects are fenced + * and durably converted to an honest terminal workflow failure. + * + * @return array{action:'begin',generation:string}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error + */ +function agents_workflow_advance_reconcile_continuation_locked( WP_Agent_Workflow_Run_Recorder $recorder, WP_Agent_Workflow_Run_Result $result ) { + $suspension = $result->get_suspension(); + $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); + $phase = agents_workflow_string( $claim['phase'] ?? '' ); + $generation = agents_workflow_string( $claim['generation'] ?? '' ); + if ( '' === $generation || ! hash_equals( $generation, agents_workflow_suspension_generation( $suspension ) ) ) { + return agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_reconcile_claim_invalid', 'The persisted reconcile continuation does not match the suspended workflow generation.' ); + } + if ( 'pending' === $phase ) { + return array( + 'action' => 'begin', + 'generation' => $generation, + ); + } + if ( 'committed' === $phase ) { + return array( + 'action' => 'resume', + 'result' => $result, + ); + } + if ( 'aggregating' === $phase ) { + $expires = is_numeric( $claim['expires'] ?? null ) ? (int) $claim['expires'] : 0; + if ( $expires > time() ) { + return new \WP_Error( 'agents_reconcile_lock_unavailable', 'Aggregation is already owned for this suspension generation; retry the persisted reconcile continuation.' ); + } + return agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_aggregation_outcome_uncertain', 'The aggregation owner was lost after external effects may have begun; the aggregate was not rerun.' ); + } + + return agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_reconcile_claim_invalid', 'The persisted reconcile continuation phase is invalid.' ); +} + +/** + * Persist an honest failed aggregate and make the continuation resumable. + * + * @return array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error + */ +function agents_workflow_terminalize_reconcile_continuation( WP_Agent_Workflow_Run_Recorder $recorder, WP_Agent_Workflow_Run_Result $result, string $code, string $message ) { + $suspension = $result->get_suspension(); + $generation = agents_workflow_suspension_generation( $suspension ); + $suspension['reconcile_claim'] = array( + 'phase' => 'committed', + 'generation' => $generation, + ); + $metadata = $result->get_metadata(); + $metadata['_suspension'] = $suspension; + $result = $result->with( array( 'metadata' => $metadata ) ); + $step_index = is_numeric( $suspension['step_index'] ?? null ) ? (int) $suspension['step_index'] : 0; + $result = agents_workflow_splice_step_output( $result, $step_index, new \WP_Error( $code, $message ) ); + $updated = agents_workflow_update_reconcile_state( $recorder, $result, 'terminalize an ambiguous reconcile continuation' ); + if ( is_wp_error( $updated ) ) { + return $updated; + } + return array( + 'action' => 'resume', + 'result' => $result, + ); +} + +/** + * Normalize recorder write uncertainty to PR #534's persisted retry contract. + * + * @return true|\WP_Error + */ +function agents_workflow_update_reconcile_state( WP_Agent_Workflow_Run_Recorder $recorder, WP_Agent_Workflow_Run_Result $result, string $transition ) { + try { + $updated = $recorder->update( $result ); + } catch ( \Throwable $error ) { + return new \WP_Error( 'agents_reconcile_lock_unavailable', sprintf( 'Could not %s; retry the persisted reconcile continuation.', $transition ), array( 'cause' => $error->getMessage() ) ); + } + if ( is_wp_error( $updated ) ) { + return new \WP_Error( 'agents_reconcile_lock_unavailable', sprintf( 'Could not %s; retry the persisted reconcile continuation.', $transition ), array( 'cause' => $updated->get_error_code() ) ); + } + return true; +} + +/** + * Resume a durably committed continuation without rerunning aggregation. + * + * @return WP_Agent_Workflow_Run_Result + */ +function agents_workflow_resume_reconcile_continuation( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, WP_Agent_Workflow_Run_Result $result ) { + if ( agents_workflow_defer_resume( $run_id, $result ) ) { + return $result; + } + $runner = agents_workflow_resolve_runner( $recorder ); + return $runner->resume( $run_id ); +} + /** * Derive the stable identity of one suspension generation. * diff --git a/tests/workflow-reconcile-race-smoke.php b/tests/workflow-reconcile-race-smoke.php index f87152b..009bda3 100644 --- a/tests/workflow-reconcile-race-smoke.php +++ b/tests/workflow-reconcile-race-smoke.php @@ -227,11 +227,17 @@ final class Race_Recorder implements WP_Agent_Workflow_Run_Recorder { /** @var array>|null */ private ?array $frozen = null; + private bool $fail_next_update = false; + public function start( WP_Agent_Workflow_Run_Result $result ) { $this->rows[ $result->get_run_id() ] = $result->to_array(); return $result->get_run_id(); } public function update( WP_Agent_Workflow_Run_Result $result ) { + if ( $this->fail_next_update ) { + $this->fail_next_update = false; + return new WP_Error( 'race_recorder_write_failed', 'Injected recorder update failure.' ); + } $this->rows[ $result->get_run_id() ] = $result->to_array(); return true; } @@ -264,6 +270,18 @@ public function unfreeze_reads(): void { $this->frozen = null; } + public function fail_next_update(): void { + $this->fail_next_update = true; + } + + public function expire_reconcile_claim( string $run_id ): void { + $this->rows[ $run_id ]['metadata']['_suspension']['reconcile_claim']['expires'] = time() - 1; + } + + public function reconcile_phase( string $run_id ): string { + return (string) ( $this->rows[ $run_id ]['metadata']['_suspension']['reconcile_claim']['phase'] ?? '' ); + } + /** Whether the built-in add_option() reconcile lock row exists for the run. */ private static function reconcile_lock_held( string $run_id ): bool { $option = 'agents_wf_reconcile_lock_' . md5( $run_id ); @@ -415,6 +433,24 @@ function race_execute_branch( array $descriptor ): array { return array( 'key' => $key, 'status' => WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, 'output' => $run['last'], 'steps' => $run['steps'], 'error' => null ); } +/** Build an all-but-last reconciled run for continuation failure tests. */ +function race_prepare_continuation_run( string $run_id ): array { + $GLOBALS['__options'] = array(); + $GLOBALS['__aggregate_calls'] = 0; + $GLOBALS['__resume_dispatch_calls'] = 0; + $GLOBALS['__during_aggregate'] = null; + $recorder = new Race_Recorder(); + remove_all_filters( 'wp_agent_workflow_run_recorder' ); + add_filter( 'wp_agent_workflow_run_recorder', static function () use ( $recorder ) { return $recorder; } ); + ( new WP_Agent_Workflow_Runner( $recorder ) )->run( race_roles_spec(), array(), array( 'run_id' => $run_id ) ); + $descriptors = Race_Executor::$dispatched; + foreach ( array( 0, 1 ) as $index ) { + $branch = race_execute_branch( $descriptors[ $index ] ); + agents_reconcile_workflow_branch( $run_id, (string) $descriptors[ $index ]['handle_id'], $branch ); + } + return array( $recorder, $descriptors, race_execute_branch( $descriptors[2] ) ); +} + // ═════════════════════════════════════════════════════════════════════════════ // THE RACE: two sibling branches reconcile CONCURRENTLY (both read the frame // before either writes). Under the buggy code the later write clobbers the @@ -518,5 +554,108 @@ static function ( bool $deferred ): bool { smoke_assert( 1, $GLOBALS['__resume_dispatch_calls'], 'ttl: one resume dispatches for the claimed suspension generation', $failures, $passes ); smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $final->get_status(), 'ttl: claimed owner commits and resumes successfully', $failures, $passes ); +// CLAIM WRITE FAILURE: aggregation cannot start until the pending continuation +// and final completion are durable. A persisted-result retry can safely repeat +// the transition because no external aggregate effect has begun. +list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'claim-write' ); +$recorder->fail_next_update(); +$failed_claim = agents_reconcile_workflow_branch( 'claim-write', (string) $descriptors[2]['handle_id'], $last_result ); +smoke_assert( 'agents_reconcile_lock_unavailable', is_wp_error( $failed_claim ) ? $failed_claim->get_error_code() : '', 'claim write: recorder failure returns PR #534 retry contract', $failures, $passes ); +smoke_assert( '', $recorder->reconcile_phase( 'claim-write' ), 'claim write: failed update installs no phantom continuation', $failures, $passes ); +smoke_assert( 0, $GLOBALS['__aggregate_calls'], 'claim write: aggregate effect does not run without durable ownership', $failures, $passes ); +$retried_claim = agents_reconcile_workflow_branch( 'claim-write', (string) $descriptors[2]['handle_id'], $last_result ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $retried_claim->get_status(), 'claim write: persisted-result retry safely completes the run', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'claim write: retry executes aggregator exactly once', $failures, $passes ); + +// AGGREGATE COMMIT WRITE FAILURE: effects have run, so a retry must never rerun +// them. Once the abandoned owner is fenced stale, continuation persists an +// honest uncertain-outcome failure and resumes the run to terminal. +list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'commit-write' ); +$GLOBALS['__during_aggregate'] = static function () use ( $recorder ): void { + $GLOBALS['__during_aggregate'] = null; + $recorder->fail_next_update(); +}; +$failed_commit = agents_reconcile_workflow_branch( 'commit-write', (string) $descriptors[2]['handle_id'], $last_result ); +smoke_assert( 'agents_reconcile_lock_unavailable', is_wp_error( $failed_commit ) ? $failed_commit->get_error_code() : '', 'commit write: recorder failure requests persisted reconcile retry', $failures, $passes ); +smoke_assert( 'aggregating', $recorder->reconcile_phase( 'commit-write' ), 'commit write: durable effect-start phase remains authoritative', $failures, $passes ); +$recorder->expire_reconcile_claim( 'commit-write' ); +$commit_recovery = agents_reconcile_workflow_branch( 'commit-write', (string) $descriptors[2]['handle_id'], $last_result ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $commit_recovery->get_status(), 'commit write: stale ambiguous owner terminalizes honestly', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'commit write: external aggregator is never rerun after uncertainty', $failures, $passes ); +smoke_assert( 'workflow_parallel_aggregation_outcome_uncertain', $commit_recovery->get_error()['code'] ?? '', 'commit write: terminal failure names uncertain aggregate outcome', $failures, $passes ); + +// SECOND LOCK CONTENTION: the original aggregate result cannot be committed, and +// PR #534 receives the same retryable contract. Redelivery advances the persisted +// aggregating phase instead of stopping at the completed-handle guard. +list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'commit-lock' ); +$lock_calls = 0; +add_filter( + 'wp_agent_workflow_reconcile_lock', + static function ( $override, string $run_id, callable $critical ) use ( &$lock_calls ) { + unset( $override, $run_id ); + ++$lock_calls; + return 3 === $lock_calls ? new WP_Error( 'agents_reconcile_lock_unavailable', 'Injected second-phase contention.' ) : $critical(); + }, + 10, + 3 +); +$contended_commit = agents_reconcile_workflow_branch( 'commit-lock', (string) $descriptors[2]['handle_id'], $last_result ); +remove_all_filters( 'wp_agent_workflow_reconcile_lock' ); +smoke_assert( 'agents_reconcile_lock_unavailable', is_wp_error( $contended_commit ) ? $contended_commit->get_error_code() : '', 'commit lock: second-phase contention requests reconcile-only retry', $failures, $passes ); +smoke_assert( 'aggregating', $recorder->reconcile_phase( 'commit-lock' ), 'commit lock: continuation records that effects may have begun', $failures, $passes ); +$recorder->expire_reconcile_claim( 'commit-lock' ); +$lock_recovery = agents_reconcile_workflow_branch( 'commit-lock', (string) $descriptors[2]['handle_id'], $last_result ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $lock_recovery->get_status(), 'commit lock: completed-handle retry advances to terminal recovery', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'commit lock: continuation never repeats aggregator effects', $failures, $passes ); + +// PROCESS LOSS BEFORE AGGREGATION: stop after persisting `pending`, then let a +// duplicate completed-result delivery start and finish the safe continuation. +list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'lost-before' ); +$pending = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'lost-before', + static function () use ( $recorder, $descriptors, $last_result ) { + return \AgentsAPI\AI\Workflows\agents_reconcile_workflow_branch_locked( $recorder, 'lost-before', (string) $descriptors[2]['handle_id'], $last_result ); + } +); +smoke_assert( 'begin', $pending['action'] ?? '', 'lost before aggregate: pending continuation persists before effects', $failures, $passes ); +smoke_assert( 'pending', $recorder->reconcile_phase( 'lost-before' ), 'lost before aggregate: durable phase proves effects have not begun', $failures, $passes ); +$before_recovery = agents_reconcile_workflow_branch( 'lost-before', (string) $descriptors[2]['handle_id'], $last_result ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $before_recovery->get_status(), 'lost before aggregate: duplicate delivery safely continues pending work', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'lost before aggregate: recovery executes aggregator once', $failures, $passes ); + +// PROCESS LOSS AFTER COMMIT BEFORE RESUME: drive through the durable commit but +// omit dispatch. A duplicate reconcile observes `committed`, skips aggregation, +// and resumes from the recorded output. +list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'lost-after' ); +$pending = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'lost-after', + static function () use ( $recorder, $descriptors, $last_result ) { + return \AgentsAPI\AI\Workflows\agents_reconcile_workflow_branch_locked( $recorder, 'lost-after', (string) $descriptors[2]['handle_id'], $last_result ); + } +); +$aggregation = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'lost-after', + static function () use ( $recorder, $pending ) { + return \AgentsAPI\AI\Workflows\agents_workflow_begin_reconcile_aggregation( $recorder, 'lost-after', $pending['generation'] ); + } +); +$aggregate_output = WP_Agent_Workflow_Runner::aggregate_branch_results( + $aggregation['aggregate'], + $aggregation['branch_results'], + \AgentsAPI\AI\Workflows\agents_workflow_resolve_step_handlers() +); +$committed = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'lost-after', + static function () use ( $recorder, $aggregation, $aggregate_output ) { + return \AgentsAPI\AI\Workflows\agents_workflow_commit_reconcile_claim( $recorder, 'lost-after', $aggregation['owner_token'], $aggregation['generation'], $aggregation['step_index'], $aggregate_output ); + } +); +smoke_assert( true, isset( $committed['result'] ), 'lost after commit: aggregate output is durable before resume dispatch', $failures, $passes ); +smoke_assert( 'committed', $recorder->reconcile_phase( 'lost-after' ), 'lost after commit: durable phase is resumable', $failures, $passes ); +$after_recovery = agents_reconcile_workflow_branch( 'lost-after', (string) $descriptors[2]['handle_id'], $last_result ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $after_recovery->get_status(), 'lost after commit: duplicate delivery resumes durable aggregate', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'lost after commit: recovery does not rerun aggregator', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__resume_dispatch_calls'], 'lost after commit: recovery dispatches resume once', $failures, $passes ); + echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; exit( count( $failures ) > 0 ? 1 : 0 ); From 5344bf2f17b8dfe7d3fbcc20d1e30091fee897b3 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 17:38:03 +0000 Subject: [PATCH 3/5] fix: move aggregation to durable continuation --- ...kflow-action-scheduler-branch-executor.php | 85 +++++- .../class-wp-agent-workflow-scoped-drain.php | 1 + .../register-reconcile-workflow-branch.php | 277 ++++++++++++------ .../register-workflow-branch-executor.php | 47 ++- stubs/action-scheduler-classes.php | 5 + tests/workflow-as-branch-smoke.php | 172 ++++++++++- tests/workflow-async-branch-payload-smoke.php | 3 + tests/workflow-reconcile-race-smoke.php | 103 ------- tests/workflow-request-controller-smoke.php | 2 +- tests/workflow-scoped-drain-smoke.php | 5 +- 10 files changed, 491 insertions(+), 209 deletions(-) diff --git a/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php b/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php index 5d4c5d5..b48aa81 100644 --- a/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php +++ b/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php @@ -57,6 +57,9 @@ final class WP_Agent_Workflow_Action_Scheduler_Branch_Executor implements WP_Age */ public const BRANCH_HOOK = 'wp_agent_workflow_branch_run'; + /** Durable, atomically claimed aggregate continuation hook. */ + public const AGGREGATE_HOOK = 'wp_agent_workflow_run_aggregate'; + /** * The resume action hook. When a reconcile observes all branches terminal it * enqueues ONE action under this hook rather than resuming inline; AS claims @@ -782,6 +785,84 @@ public static function run_branch_action( array $payload ): void { self::reconcile_branch_result( $payload, $branch_result ); } + /** Enqueue the unique aggregate continuation for one suspension generation. */ + public static function enqueue_aggregate_action( string $run_id, string $generation, string $owner_token, bool $recover_failure = false ): int { + return self::enqueue_async_action( + self::AGGREGATE_HOOK, + array( + array( + 'run_id' => $run_id, + 'generation' => $generation, + 'owner_token' => $owner_token, + 'recover_failure' => $recover_failure, + ), + ), + self::group_for_run( $run_id ), + true + ); + } + + /** + * Run the claimed aggregate action and fail loudly into AS lifecycle hooks. + * + * @param array $payload Aggregate action payload. + */ + public static function run_aggregate_action( array $payload ): void { + $run_id = self::string_value( $payload['run_id'] ?? '' ); + $generation = self::string_value( $payload['generation'] ?? '' ); + $owner_token = self::string_value( $payload['owner_token'] ?? '' ); + if ( '' === $run_id || '' === $generation || '' === $owner_token ) { + return; + } + $recorder = agents_workflow_resolve_recorder(); + if ( null === $recorder ) { + throw new \RuntimeException( 'A recorder is required to run an aggregate continuation.' ); + } + $result = ! empty( $payload['recover_failure'] ) + ? agents_workflow_fail_aggregate_continuation( $recorder, $run_id, $generation, $owner_token, true ) + : agents_workflow_run_aggregate_continuation( $recorder, $run_id, $generation, $owner_token ); + if ( is_wp_error( $result ) ) { + throw new \RuntimeException( $result->get_error_message() ); + } + } + + /** + * Apply AS failed-action recovery to a known aggregate payload. + * + * @param array $payload Aggregate action payload. + */ + public static function run_aggregate_action_failure( array $payload ): void { + $run_id = self::string_value( $payload['run_id'] ?? '' ); + $generation = self::string_value( $payload['generation'] ?? '' ); + $owner_token = self::string_value( $payload['owner_token'] ?? '' ); + $recorder = agents_workflow_resolve_recorder(); + if ( '' === $run_id || '' === $generation || '' === $owner_token || null === $recorder ) { + return; + } + $result = agents_workflow_fail_aggregate_continuation( $recorder, $run_id, $generation, $owner_token ); + if ( is_wp_error( $result ) ) { + self::enqueue_aggregate_action( $run_id, $generation, $owner_token, true ); + } + } + + /** Resolve and recover an aggregate action reported failed by Action Scheduler. */ + public static function handle_failed_action( int $action_id ): void { + if ( $action_id <= 0 || ! class_exists( 'ActionScheduler_Store' ) ) { + return; + } + try { + $action = \ActionScheduler_Store::instance()->fetch_action( $action_id ); + if ( ! method_exists( $action, 'get_hook' ) || self::AGGREGATE_HOOK !== $action->get_hook() || ! method_exists( $action, 'get_args' ) ) { + return; + } + $args = $action->get_args(); + $payload = is_array( $args ) && is_array( $args[0] ?? null ) ? $args[0] : array(); + self::run_aggregate_action_failure( $payload ); + } catch ( \Throwable $error ) { + unset( $error ); + } + } + /** * Reconcile a completed branch, re-enqueuing it when lock contention prevents * the result from being recorded. Other errors are authoritative and are not @@ -1034,14 +1115,14 @@ public static function run_resume_action( array $payload ): void { * @param string $group Action group. * @return int Action id, or 0 when the enqueue failed (threw or returned no id). */ - private static function enqueue_async_action( string $hook, array $args, string $group ): int { + private static function enqueue_async_action( string $hook, array $args, string $group, bool $unique = false ): int { if ( ! function_exists( 'as_enqueue_async_action' ) ) { return 0; } try { // AS returns the new action id (a positive int) on success. dispatch() // treats a non-positive return as a hard failure. - return (int) as_enqueue_async_action( $hook, $args, $group ); + return (int) as_enqueue_async_action( $hook, $args, $group, $unique ); } catch ( \Throwable $error ) { // AS rejected the enqueue (e.g. args too long / queue unavailable). // Normalize to 0 so dispatch() surfaces a clean WP_Error rather than diff --git a/src/Workflows/class-wp-agent-workflow-scoped-drain.php b/src/Workflows/class-wp-agent-workflow-scoped-drain.php index 1ec140f..041102c 100644 --- a/src/Workflows/class-wp-agent-workflow-scoped-drain.php +++ b/src/Workflows/class-wp-agent-workflow-scoped-drain.php @@ -103,6 +103,7 @@ public static function is_available(): bool { public static function default_hooks(): array { return array( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK, + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK, WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK, ); } diff --git a/src/Workflows/register-reconcile-workflow-branch.php b/src/Workflows/register-reconcile-workflow-branch.php index 4a914b1..149ce0a 100644 --- a/src/Workflows/register-reconcile-workflow-branch.php +++ b/src/Workflows/register-reconcile-workflow-branch.php @@ -23,9 +23,6 @@ const AGENTS_RECONCILE_WORKFLOW_BRANCH_ABILITY = 'agents/reconcile-workflow-branch'; -/** Effect-owner lease used only to decide when an ambiguous aggregation must fail closed. */ -const AGENTS_RECONCILE_CLAIM_TTL_SECONDS = 60; - add_action( 'wp_abilities_api_init', static function (): void { @@ -103,9 +100,9 @@ function agents_reconcile_workflow_branch_ability( array $input ) { * 2. Merge the branch result into `metadata._suspension.completed[handle_id]` * and flip that handle's status. Persist. * 3. If NOT all handles terminal → return the still-suspended run. - * 4. If all terminal → claim this suspension generation under the lock, release - * the lock, run the aggregate plan, then reacquire the lock to fence its - * recorder commit before resuming from `step_index + 1`. + * 4. If all terminal → enqueue the executor's durable aggregate continuation + * and persist its generation-bound owner. The claimed continuation records + * `running`, aggregates outside the lock, commits, then dispatches resume. * * CONCURRENCY. Steps 1–3 are a read-modify-write on the shared per-run frame's * `completed[]` map. When N branches finish CONCURRENTLY in N separate processes @@ -114,10 +111,11 @@ function agents_reconcile_workflow_branch_ability( array $input ) { * merge, the frame never reaches all-terminal, and the run hangs SUSPENDED * forever (observed in a real MySQL fanout A/B). Short recorder transitions run * under a per-run cross-process lock ({@see agents_workflow_reconcile_with_lock()}). - * Once all branches are terminal, a durable claim in the suspension frame elects - * exactly one aggregation/resume owner for that generation. Aggregation runs - * outside the expiring lock, and its result is committed only after a fresh read - * proves the same claim still owns the same generation. + * For Action Scheduler runs, a dedicated atomically claimed action owns aggregate + * execution and its failed-action lifecycle fences crashes. Duplicate reconcile + * delivery observes `queued` or `running` and cannot execute aggregation. The + * aggregate result is committed only after a fresh read proves the action token + * still owns the same suspension generation. * * @since 0.5.0 * @@ -145,19 +143,6 @@ static function () use ( $recorder, $run_id, $handle_id, $branch_result ) { return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $transition['result'] ); } - $transition = agents_workflow_reconcile_with_lock( - $run_id, - static function () use ( $recorder, $run_id, $transition ) { - return agents_workflow_begin_reconcile_aggregation( $recorder, $run_id, $transition['generation'] ); - } - ); - if ( is_wp_error( $transition ) || $transition instanceof WP_Agent_Workflow_Run_Result ) { - return $transition; - } - if ( 'resume' === $transition['action'] ) { - return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $transition['result'] ); - } - $required_failed = ! empty( $transition['required_failed'] ); if ( $required_failed ) { $step_output = new \WP_Error( 'workflow_parallel_required_branch_failed', 'A required parallel branch failed during out-of-band execution.' ); @@ -198,7 +183,7 @@ static function () use ( $recorder, $run_id, $transition, $step_output ) { * @param string $run_id Suspended run id. * @param string $handle_id The completed branch's handle id. * @param array $branch_result BranchResult. - * @return WP_Agent_Workflow_Run_Result|array{action:'begin',generation:string}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error + * @return WP_Agent_Workflow_Run_Result|array{action:'aggregate',owner_token:string,generation:string,step_index:int,aggregate:array,branch_results:array,required_failed:bool}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error */ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $handle_id, array $branch_result ) { $result = $recorder->find( $run_id ); @@ -283,11 +268,40 @@ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder $suspension['handles'] = $handles; $suspension['completed'] = $completed; + $transition = null; if ( count( $completed ) >= count( $handles ) ) { - $suspension['reconcile_claim'] = array( - 'phase' => 'pending', - 'generation' => agents_workflow_suspension_generation( $suspension ), - ); + $generation = agents_workflow_suspension_generation( $suspension ); + $owner_token = agents_workflow_reconcile_claim_token(); + $action_id = agents_workflow_dispatch_aggregate_continuation( $run_id, $suspension, $generation, $owner_token ); + if ( is_int( $action_id ) && $action_id > 0 ) { + $suspension['reconcile_claim'] = array( + 'phase' => 'queued', + 'generation' => $generation, + 'owner_token' => $owner_token, + 'action_id' => $action_id, + ); + } elseif ( is_int( $action_id ) ) { + $suspension['reconcile_claim'] = array( + 'phase' => 'committed', + 'generation' => $generation, + ); + $failed_metadata = $result->get_metadata(); + $failed_metadata['_suspension'] = $suspension; + $result = agents_workflow_splice_step_output( + $result->with( array( 'metadata' => $failed_metadata ) ), + is_numeric( $suspension['step_index'] ?? null ) ? (int) $suspension['step_index'] : 0, + new \WP_Error( 'workflow_parallel_aggregation_dispatch_failed', 'The durable aggregate continuation could not be enqueued.' ) + ); + $suspension = $result->get_suspension(); + $transition = array( 'action' => 'resume' ); + } else { + $suspension['reconcile_claim'] = array( + 'phase' => 'running', + 'generation' => $generation, + 'owner_token' => $owner_token, + ); + $transition = agents_workflow_reconcile_aggregate_transition( $suspension, $owner_token, $generation ); + } } $metadata = $result->get_metadata(); @@ -298,12 +312,16 @@ function agents_reconcile_workflow_branch_locked( WP_Agent_Workflow_Run_Recorder return $updated; } - return count( $completed ) < count( $handles ) - ? $result - : array( - 'action' => 'begin', - 'generation' => agents_workflow_suspension_generation( $suspension ), + if ( null === $transition ) { + return $result; + } + if ( 'resume' === $transition['action'] ) { + return array( + 'action' => 'resume', + 'result' => $result, ); + } + return $transition; } /** @@ -332,24 +350,18 @@ function agents_workflow_commit_reconcile_claim( WP_Agent_Workflow_Run_Recorder $suspension = $result->get_suspension(); $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); if ( - 'aggregating' !== agents_workflow_string( $claim['phase'] ?? '' ) || + 'running' !== agents_workflow_string( $claim['phase'] ?? '' ) || ! hash_equals( $claim_token, agents_workflow_string( $claim['owner_token'] ?? '' ) ) || ! hash_equals( $generation, agents_workflow_string( $claim['generation'] ?? '' ) ) || ! hash_equals( $generation, agents_workflow_suspension_generation( $suspension ) ) ) { - if ( 'pending' === agents_workflow_string( $claim['phase'] ?? '' ) ) { - return $result; - } - $continuation = agents_workflow_advance_reconcile_continuation_locked( $recorder, $result ); - if ( is_array( $continuation ) && 'begin' === $continuation['action'] ) { - return $result; - } - return $continuation; + return agents_workflow_advance_reconcile_continuation_locked( $recorder, $result ); } $suspension['reconcile_claim'] = array( - 'phase' => 'committed', - 'generation' => $generation, + 'phase' => 'committed', + 'generation' => $generation, + 'owner_token' => $claim_token, ); $metadata = $result->get_metadata(); $metadata['_suspension'] = $suspension; @@ -363,15 +375,51 @@ function agents_workflow_commit_reconcile_claim( WP_Agent_Workflow_Run_Recorder } /** - * Persist the effect-start boundary for a pending continuation. A process lost - * before this transition is safe to retry; after it, aggregation is never rerun. + * Run one durable aggregate continuation action. The action atomically moves its + * matching `queued` generation to `running` before external effects, commits the + * aggregate under the short lock, then dispatches resume. * * @param WP_Agent_Workflow_Run_Recorder $recorder Resolved recorder. * @param string $run_id Suspended run id. - * @param string $generation Suspension generation identity. + * @param string $generation Suspension generation identity. + * @param string $owner_token Aggregate action owner token. + * @return WP_Agent_Workflow_Run_Result|\WP_Error + */ +function agents_workflow_run_aggregate_continuation( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $generation, string $owner_token ) { + $transition = agents_workflow_reconcile_with_lock( + $run_id, + static function () use ( $recorder, $run_id, $generation, $owner_token ) { + return agents_workflow_begin_aggregate_action_locked( $recorder, $run_id, $generation, $owner_token ); + } + ); + if ( is_wp_error( $transition ) || $transition instanceof WP_Agent_Workflow_Run_Result ) { + return $transition; + } + if ( 'resume' === $transition['action'] ) { + return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $transition['result'] ); + } + + $step_output = ! empty( $transition['required_failed'] ) + ? new \WP_Error( 'workflow_parallel_required_branch_failed', 'A required parallel branch failed during out-of-band execution.' ) + : WP_Agent_Workflow_Runner::aggregate_branch_results( $transition['aggregate'], $transition['branch_results'], agents_workflow_resolve_step_handlers() ); + $commit = agents_workflow_reconcile_with_lock( + $run_id, + static function () use ( $recorder, $run_id, $transition, $step_output ) { + return agents_workflow_commit_reconcile_claim( $recorder, $run_id, $transition['owner_token'], $transition['generation'], $transition['step_index'], $step_output ); + } + ); + if ( is_wp_error( $commit ) || $commit instanceof WP_Agent_Workflow_Run_Result ) { + return $commit; + } + return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $commit['result'] ); +} + +/** + * Begin a claimed aggregate action under the short lock. + * * @return WP_Agent_Workflow_Run_Result|array{action:'aggregate',owner_token:string,generation:string,step_index:int,aggregate:array,branch_results:array,required_failed:bool}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error */ -function agents_workflow_begin_reconcile_aggregation( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $generation ) { +function agents_workflow_begin_aggregate_action_locked( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $generation, string $owner_token ) { $result = $recorder->find( $run_id ); if ( null === $result ) { return new \WP_Error( 'agents_reconcile_workflow_branch_not_found', sprintf( 'No suspended run was found for run_id `%s`.', $run_id ) ); @@ -379,39 +427,42 @@ function agents_workflow_begin_reconcile_aggregation( WP_Agent_Workflow_Run_Reco if ( ! $result->is_suspended() ) { return $result; } - $suspension = $result->get_suspension(); $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); + if ( 'committed' === agents_workflow_string( $claim['phase'] ?? '' ) && hash_equals( $generation, agents_workflow_string( $claim['generation'] ?? '' ) ) ) { + return array( + 'action' => 'resume', + 'result' => $result, + ); + } if ( - 'pending' !== agents_workflow_string( $claim['phase'] ?? '' ) || + 'queued' !== agents_workflow_string( $claim['phase'] ?? '' ) || ! hash_equals( $generation, agents_workflow_string( $claim['generation'] ?? '' ) ) || + ! hash_equals( $owner_token, agents_workflow_string( $claim['owner_token'] ?? '' ) ) || ! hash_equals( $generation, agents_workflow_suspension_generation( $suspension ) ) ) { - if ( 'pending' === agents_workflow_string( $claim['phase'] ?? '' ) ) { - return $result; - } - $continuation = agents_workflow_advance_reconcile_continuation_locked( $recorder, $result ); - if ( is_array( $continuation ) && 'begin' === $continuation['action'] ) { - return $result; - } - return $continuation; + return $result; } - $owner_token = agents_workflow_reconcile_claim_token(); - $suspension['reconcile_claim'] = array( - 'phase' => 'aggregating', - 'generation' => $generation, - 'owner_token' => $owner_token, - 'expires' => time() + AGENTS_RECONCILE_CLAIM_TTL_SECONDS, - ); + $claim['phase'] = 'running'; + $suspension['reconcile_claim'] = $claim; $metadata = $result->get_metadata(); $metadata['_suspension'] = $suspension; $result = $result->with( array( 'metadata' => $metadata ) ); - $updated = agents_workflow_update_reconcile_state( $recorder, $result, 'mark aggregation effects as started' ); + $updated = agents_workflow_update_reconcile_state( $recorder, $result, 'mark the aggregate action as running' ); if ( is_wp_error( $updated ) ) { return $updated; } + return agents_workflow_reconcile_aggregate_transition( $suspension, $owner_token, $generation ); +} +/** + * Return the aggregate inputs carried by one claimed suspension generation. + * + * @param array $suspension Suspension frame. + * @return array{action:'aggregate',owner_token:string,generation:string,step_index:int,aggregate:array,branch_results:array,required_failed:bool} + */ +function agents_workflow_reconcile_aggregate_transition( array $suspension, string $owner_token, string $generation ): array { $completed = is_array( $suspension['completed'] ?? null ) ? \AgentsAPI\AI\WP_Agent_Run_Control::string_keyed_array( $suspension['completed'] ) : array(); return array( 'action' => 'aggregate', @@ -425,41 +476,71 @@ function agents_workflow_begin_reconcile_aggregation( WP_Agent_Workflow_Run_Reco } /** - * Advance an already-persisted continuation without relying on a new branch - * completion. Active effects fail retryably; stale ambiguous effects are fenced - * and durably converted to an honest terminal workflow failure. + * Duplicate reconciles never take over queued or running aggregate actions. * - * @return array{action:'begin',generation:string}|array{action:'resume',result:WP_Agent_Workflow_Run_Result}|\WP_Error + * @return WP_Agent_Workflow_Run_Result|array{action:'resume',result:WP_Agent_Workflow_Run_Result} */ function agents_workflow_advance_reconcile_continuation_locked( WP_Agent_Workflow_Run_Recorder $recorder, WP_Agent_Workflow_Run_Result $result ) { - $suspension = $result->get_suspension(); - $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); - $phase = agents_workflow_string( $claim['phase'] ?? '' ); - $generation = agents_workflow_string( $claim['generation'] ?? '' ); - if ( '' === $generation || ! hash_equals( $generation, agents_workflow_suspension_generation( $suspension ) ) ) { - return agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_reconcile_claim_invalid', 'The persisted reconcile continuation does not match the suspended workflow generation.' ); - } - if ( 'pending' === $phase ) { - return array( - 'action' => 'begin', - 'generation' => $generation, - ); - } - if ( 'committed' === $phase ) { - return array( - 'action' => 'resume', - 'result' => $result, - ); + unset( $recorder ); + $claim = $result->get_suspension()['reconcile_claim'] ?? array(); + if ( is_array( $claim ) && 'committed' === agents_workflow_string( $claim['phase'] ?? '' ) ) { + return array( 'action' => 'resume', 'result' => $result ); } - if ( 'aggregating' === $phase ) { - $expires = is_numeric( $claim['expires'] ?? null ) ? (int) $claim['expires'] : 0; - if ( $expires > time() ) { - return new \WP_Error( 'agents_reconcile_lock_unavailable', 'Aggregation is already owned for this suspension generation; retry the persisted reconcile continuation.' ); + return $result; +} + +/** + * Dispatch one executor-owned durable aggregate action, or null for inline fallback. + * + * @param array $suspension Suspension frame. + */ +function agents_workflow_dispatch_aggregate_continuation( string $run_id, array $suspension, string $generation, string $owner_token, bool $recover_failure = false ): ?int { + $executor_id = agents_workflow_string( $suspension['executor_id'] ?? '' ); + $action_id = apply_filters( 'wp_agent_workflow_aggregate_dispatch', null, $run_id, $executor_id, $generation, $owner_token, $recover_failure ); + return is_int( $action_id ) ? $action_id : null; +} + +/** + * Handle an aggregate action failure through its persisted phase. + * + * @return WP_Agent_Workflow_Run_Result|\WP_Error|null + */ +function agents_workflow_fail_aggregate_continuation( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, string $generation, string $owner_token, bool $from_durable_action = false ) { + $transition = agents_workflow_reconcile_with_lock( + $run_id, + static function () use ( $recorder, $run_id, $generation, $owner_token ) { + $result = $recorder->find( $run_id ); + if ( null === $result || ! $result->is_suspended() ) { + return $result; + } + $claim = $result->get_suspension()['reconcile_claim'] ?? array(); + if ( ! is_array( $claim ) || ! hash_equals( $generation, agents_workflow_string( $claim['generation'] ?? '' ) ) || ! hash_equals( $owner_token, agents_workflow_string( $claim['owner_token'] ?? '' ) ) ) { + return $result; + } + if ( 'queued' === agents_workflow_string( $claim['phase'] ?? '' ) ) { + $action_id = agents_workflow_dispatch_aggregate_continuation( $run_id, $result->get_suspension(), $generation, $owner_token ); + return is_int( $action_id ) && $action_id > 0 + ? $result + : agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_aggregation_dispatch_failed', 'The aggregate continuation failed before effects began and could not be re-enqueued.' ); + } + if ( 'running' === agents_workflow_string( $claim['phase'] ?? '' ) ) { + return agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_aggregation_outcome_uncertain', 'The aggregate action failed after external effects may have begun; the aggregate was not rerun.' ); + } + return 'committed' === agents_workflow_string( $claim['phase'] ?? '' ) + ? array( 'action' => 'resume', 'result' => $result ) + : $result; } - return agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_aggregation_outcome_uncertain', 'The aggregation owner was lost after external effects may have begun; the aggregate was not rerun.' ); + ); + if ( is_array( $transition ) ) { + if ( ! $from_durable_action ) { + $action_id = agents_workflow_dispatch_aggregate_continuation( $run_id, $transition['result']->get_suspension(), $generation, $owner_token, true ); + if ( is_int( $action_id ) && $action_id > 0 ) { + return $transition['result']; + } + } + return agents_workflow_resume_reconcile_continuation( $recorder, $run_id, $transition['result'] ); } - - return agents_workflow_terminalize_reconcile_continuation( $recorder, $result, 'workflow_parallel_reconcile_claim_invalid', 'The persisted reconcile continuation phase is invalid.' ); + return $transition; } /** @@ -470,9 +551,11 @@ function agents_workflow_advance_reconcile_continuation_locked( WP_Agent_Workflo function agents_workflow_terminalize_reconcile_continuation( WP_Agent_Workflow_Run_Recorder $recorder, WP_Agent_Workflow_Run_Result $result, string $code, string $message ) { $suspension = $result->get_suspension(); $generation = agents_workflow_suspension_generation( $suspension ); + $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); $suspension['reconcile_claim'] = array( - 'phase' => 'committed', - 'generation' => $generation, + 'phase' => 'committed', + 'generation' => $generation, + 'owner_token' => agents_workflow_string( $claim['owner_token'] ?? '' ), ); $metadata = $result->get_metadata(); $metadata['_suspension'] = $suspension; diff --git a/src/Workflows/register-workflow-branch-executor.php b/src/Workflows/register-workflow-branch-executor.php index a3c84c2..fcc2cb1 100644 --- a/src/Workflows/register-workflow-branch-executor.php +++ b/src/Workflows/register-workflow-branch-executor.php @@ -18,7 +18,10 @@ * rehydrates a branch from its payload, runs it through the SHARED * `run_branch_steps()`, and drives the REAL reconcile. * - * 3. Registers the resume action callback ({@see RESUME_HOOK}) and the + * 3. Registers the aggregate continuation callback ({@see AGGREGATE_HOOK}) and + * its Action Scheduler failed-action recovery hooks. + * + * 4. Registers the resume action callback ({@see RESUME_HOOK}) and the * deferred-resume seam ({@see wp_agent_workflow_resume_dispatch}) so the * "all branches terminal → resume" transition is performed as ONE * atomically-claimed AS action instead of resuming inline. AS's claim is @@ -79,6 +82,15 @@ static function ( $payload = array() ): void { ); // 3a. Resume action: AS claimed it exactly once → re-check SUSPENDED → resume. +add_action( + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK, + static function ( $payload = array() ): void { + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_aggregate_action( is_array( $payload ) ? $payload : array() ); + }, + 10, + 1 +); + add_action( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK, /** @@ -91,6 +103,39 @@ static function ( $payload = array() ): void { 1 ); +// Durable aggregate dispatch seam. Returning an action id tells reconcile that +// aggregation is owned by Action Scheduler; null preserves non-AS inline paths. +add_filter( + 'wp_agent_workflow_aggregate_dispatch', + static function ( $action_id, $run_id, $executor_id, $generation, $owner_token, $recover_failure ) { + if ( is_int( $action_id ) || WP_Agent_Workflow_Action_Scheduler_Branch_Executor::ID !== $executor_id ) { + return $action_id; + } + return WP_Agent_Workflow_Action_Scheduler_Branch_Executor::enqueue_aggregate_action( + is_string( $run_id ) ? $run_id : '', + is_string( $generation ) ? $generation : '', + is_string( $owner_token ) ? $owner_token : '', + (bool) $recover_failure + ); + }, + 10, + 6 +); + +// Action Scheduler reports thrown execution errors, fatal shutdowns, and reaped +// abandoned actions through separate lifecycle hooks. All three converge on the +// same phase-aware aggregate recovery callback. +foreach ( array( 'action_scheduler_failed_execution', 'action_scheduler_unexpected_shutdown', 'action_scheduler_failed_action' ) as $failure_hook ) { + add_action( + $failure_hook, + static function ( $action_id ): void { + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::handle_failed_action( is_numeric( $action_id ) ? (int) $action_id : 0 ); + }, + 20, + 1 + ); +} + // 3b. Deferred-resume seam: enqueue a claimed RESUME action for AS-owned runs // instead of resuming inline in the reconcile request. add_filter( diff --git a/stubs/action-scheduler-classes.php b/stubs/action-scheduler-classes.php index 5f77756..653950b 100644 --- a/stubs/action-scheduler-classes.php +++ b/stubs/action-scheduler-classes.php @@ -52,6 +52,11 @@ public function stake_claim( int $max_actions = 10, ?\DateTime $before_date = nu } public function release_claim( ActionScheduler_ActionClaim $claim ): void {} + + public function fetch_action( int $action_id ): object { + unset( $action_id ); + return new \stdClass(); + } } /** diff --git a/tests/workflow-as-branch-smoke.php b/tests/workflow-as-branch-smoke.php index cf07fb8..16afb5a 100644 --- a/tests/workflow-as-branch-smoke.php +++ b/tests/workflow-as-branch-smoke.php @@ -166,10 +166,17 @@ public static function reset(): void { self::$reject_hook = ''; } - public static function enqueue( string $hook, array $args, string $group ): int { + public static function enqueue( string $hook, array $args, string $group, bool $unique = false ): int { if ( '' !== self::$reject_hook && self::$reject_hook === $hook ) { return 0; } + if ( $unique ) { + foreach ( self::$queue as $action ) { + if ( $hook === $action['hook'] && $args === $action['args'] && $group === $action['group'] && empty( self::$claimed[ $action['id'] ] ) ) { + return $action['id']; + } + } + } $id = ++self::$seq; self::$queue[] = array( 'id' => $id, @@ -221,8 +228,14 @@ public static function fire( int $id ): bool { } if ( ! function_exists( 'as_enqueue_async_action' ) ) { - function as_enqueue_async_action( string $hook, array $args = array(), string $group = '' ) { - return AS_Shim::enqueue( $hook, $args, $group ); + function as_enqueue_async_action( string $hook, array $args = array(), string $group = '', bool $unique = false ) { + return AS_Shim::enqueue( $hook, $args, $group, $unique ); + } +} + +function as_smoke_fire_aggregate_actions(): void { + foreach ( AS_Shim::actions_for( \AgentsAPI\AI\Workflows\WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ) as $action ) { + AS_Shim::fire( $action['id'] ); } } @@ -278,12 +291,17 @@ function smoke_assert_true( $actual, string $name, array &$failures, int &$passe final class AS_Smoke_Recorder implements WP_Agent_Workflow_Run_Recorder { /** @var array> */ public array $rows = array(); + private bool $fail_next_update = false; public function start( WP_Agent_Workflow_Run_Result $result ) { $this->rows[ $result->get_run_id() ] = $result->to_array(); return $result->get_run_id(); } public function update( WP_Agent_Workflow_Run_Result $result ) { + if ( $this->fail_next_update ) { + $this->fail_next_update = false; + return new WP_Error( 'as_recorder_write_failed', 'Injected recorder update failure.' ); + } $this->rows[ $result->get_run_id() ] = $result->to_array(); return true; } @@ -310,6 +328,10 @@ public function recent( array $args = array() ): array { public function tables(): array { return array( 'workflow_runs' ); } + + public function fail_next_update(): void { + $this->fail_next_update = true; + } } // ── Abilities: aggregator + sequential consumer + a real role worker ───────── @@ -329,6 +351,10 @@ static function ( array $input ): array { as_smoke_register_ability( 'demo/aggregate', static function ( array $input ): array { + $GLOBALS['__as_aggregate_effects'] = (int) ( $GLOBALS['__as_aggregate_effects'] ?? 0 ) + 1; + if ( is_callable( $GLOBALS['__as_during_aggregate'] ?? null ) ) { + call_user_func( $GLOBALS['__as_during_aggregate'] ); + } return array( 'final_bundle' => 'FUSED[' . (string) ( $input['headline'] ?? '' ) . '|' . (string) ( $input['body'] ?? '' ) . ']' ); } ); @@ -503,6 +529,7 @@ static function ( $override, string $run_id, callable $critical ) use ( &$lock_a smoke_assert( $resume_before, count( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ) ), 'AS path: no resume enqueued before all branches terminal', $failures, $passes ); AS_Shim::fire( $branch_actions[1]['id'] ); +as_smoke_fire_aggregate_actions(); // Resume was DEFERRED to a claimed action — the run is still suspended until the // RESUME action fires (this is the whole point: not inline). @@ -559,6 +586,7 @@ static function ( $override, string $run_id, callable $critical ) use ( &$lock_a $payload2 = $branch_actions2[1]['args'][0] ?? array(); $last_handle_id = (string) ( $payload2['handle_id'] ?? '' ); AS_Shim::fire( $branch_actions2[1]['id'] ); // last branch → reconcile all-terminal → enqueues resume #1 +as_smoke_fire_aggregate_actions(); $resume_actions2 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); smoke_assert( 1, count( $resume_actions2 ), 'race: last branch enqueued resume #1', $failures, $passes ); @@ -615,6 +643,7 @@ static function ( $override, string $run_id, callable $critical ) use ( &$lock_a foreach ( $branch_actions3 as $action ) { AS_Shim::fire( $action['id'] ); } +as_smoke_fire_aggregate_actions(); $resume_actions3 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); foreach ( $resume_actions3 as $action ) { AS_Shim::fire( $action['id'] ); @@ -675,6 +704,7 @@ function as_smoke_failing_spec(): WP_Agent_Workflow_Spec { foreach ( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK ) as $action ) { AS_Shim::fire( $action['id'] ); } +as_smoke_fire_aggregate_actions(); foreach ( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ) as $action ) { AS_Shim::fire( $action['id'] ); } @@ -715,6 +745,7 @@ function as_smoke_two_fanout_spec(): WP_Agent_Workflow_Spec { $group5 = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::group_for_run( 'as-two' ); $branches5 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK ); AS_Shim::fire( $branches5[0]['id'] ); +as_smoke_fire_aggregate_actions(); $resumes5 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); as_enqueue_async_action( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK, $resumes5[0]['args'], $resumes5[0]['group'] ); AS_Shim::fire( $resumes5[0]['id'] ); @@ -730,11 +761,146 @@ function as_smoke_two_fanout_spec(): WP_Agent_Workflow_Spec { smoke_assert( $group5, $branches5[1]['group'] ?? '', 'multi-fanout: second fan-out stays in the original isolated group', $failures, $passes ); AS_Shim::fire( $branches5[1]['id'] ); +as_smoke_fire_aggregate_actions(); $resumes5 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); AS_Shim::fire( $resumes5[2]['id'] ); $final5 = $recorder5->find( 'as-two' ); smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $final5->get_status(), 'multi-fanout: second resume reaches terminal success', $failures, $passes ); smoke_assert( $group5, WP_Agent_Workflow_Action_Scheduler_Branch_Executor::group_for_run( 'as-two' ), 'multi-fanout: terminal run retains the same deterministic group identity', $failures, $passes ); +/** Prepare a run with all branches reconciled and its aggregate action queued. */ +function as_smoke_prepare_aggregate_run( string $run_id ): array { + AS_Shim::reset(); + $GLOBALS['__as_aggregate_effects'] = 0; + $GLOBALS['__as_during_aggregate'] = null; + $recorder = new AS_Smoke_Recorder(); + remove_all_filters( 'wp_agent_workflow_run_recorder' ); + add_filter( 'wp_agent_workflow_run_recorder', static function () use ( $recorder ) { return $recorder; } ); + ( new WP_Agent_Workflow_Runner( $recorder ) )->run( as_smoke_roles_spec(), array(), array( 'run_id' => $run_id ) ); + foreach ( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK ) as $action ) { + AS_Shim::fire( $action['id'] ); + } + return array( $recorder, AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK )[0] ); +} + +function as_smoke_fire_resume_actions(): void { + foreach ( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ) as $action ) { + if ( ! empty( $action['args'][0]['recover_failure'] ) ) { + AS_Shim::fire( $action['id'] ); + } + } + foreach ( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ) as $action ) { + AS_Shim::fire( $action['id'] ); + } +} + +// QUEUED DURABILITY + HEALTHY LONG AGGREGATION. Reconcile returns with one durable +// aggregate action and no elapsed-time lease. Delayed execution remains healthy. +list( $continuation_recorder, $aggregate_action ) = as_smoke_prepare_aggregate_run( 'as-continuation' ); +$queued_claim = $continuation_recorder->find( 'as-continuation' )->get_suspension()['reconcile_claim'] ?? array(); +smoke_assert( 'queued', $queued_claim['phase'] ?? '', 'aggregate continuation: all-terminal reconcile persists queued phase', $failures, $passes ); +smoke_assert( 1, count( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ) ), 'aggregate continuation: exactly one durable aggregate action is pending', $failures, $passes ); +smoke_assert( false, array_key_exists( 'expires', $queued_claim ), 'aggregate continuation: ownership has no fixed 60-second expiry', $failures, $passes ); +AS_Shim::fire( $aggregate_action['id'] ); +as_smoke_fire_resume_actions(); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $continuation_recorder->find( 'as-continuation' )->get_status(), 'aggregate continuation: delayed healthy action completes normally', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__as_aggregate_effects'], 'aggregate continuation: healthy delayed aggregator executes once', $failures, $passes ); + +// DUPLICATE ACTION DELIVERY. Even distinct AS actions carrying the same owner +// payload cannot both transition queued -> running. +list( $duplicate_recorder, $aggregate_action ) = as_smoke_prepare_aggregate_run( 'as-duplicate-aggregate' ); +$duplicate_id = AS_Shim::enqueue( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK, $aggregate_action['args'], $aggregate_action['group'] ); +AS_Shim::fire( $aggregate_action['id'] ); +AS_Shim::fire( $duplicate_id ); +as_smoke_fire_resume_actions(); +smoke_assert( 1, $GLOBALS['__as_aggregate_effects'], 'aggregate duplicate: external effects execute at most once', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $duplicate_recorder->find( 'as-duplicate-aggregate' )->get_status(), 'aggregate duplicate: duplicate delivery preserves successful outcome', $failures, $passes ); + +// ACTION CRASH AFTER EFFECTS MAY BEGIN. The AS failure callback fences `running`, +// persists an uncertain failure, and resumes without rerunning the aggregator. +list( $crash_recorder, $aggregate_action ) = as_smoke_prepare_aggregate_run( 'as-aggregate-crash' ); +$payload = $aggregate_action['args'][0]; +$begun = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'as-aggregate-crash', + static function () use ( $crash_recorder, $payload ) { + return \AgentsAPI\AI\Workflows\agents_workflow_begin_aggregate_action_locked( $crash_recorder, 'as-aggregate-crash', $payload['generation'], $payload['owner_token'] ); + } +); +$GLOBALS['__as_aggregate_effects'] = 1; +WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_aggregate_action_failure( $payload ); +as_smoke_fire_resume_actions(); +$crash_final = $crash_recorder->find( 'as-aggregate-crash' ); +smoke_assert( 'aggregate', $begun['action'] ?? '', 'aggregate crash: action durably marks running before effects', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $crash_final->get_status(), 'aggregate crash: failed-action lifecycle terminalizes the run', $failures, $passes ); +smoke_assert( 'workflow_parallel_aggregation_outcome_uncertain', $crash_final->get_error()['code'] ?? '', 'aggregate crash: failure reports uncertain external effects', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__as_aggregate_effects'], 'aggregate crash: failed action never reruns effects', $failures, $passes ); + +// COMMITTED BEFORE RESUME CRASH. The failure callback observes durable output and +// only dispatches resume; aggregation is not repeated. +list( $committed_recorder, $aggregate_action ) = as_smoke_prepare_aggregate_run( 'as-committed-crash' ); +$payload = $aggregate_action['args'][0]; +$aggregate_transition = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'as-committed-crash', + static function () use ( $committed_recorder, $payload ) { + return \AgentsAPI\AI\Workflows\agents_workflow_begin_aggregate_action_locked( $committed_recorder, 'as-committed-crash', $payload['generation'], $payload['owner_token'] ); + } +); +$aggregate_output = WP_Agent_Workflow_Runner::aggregate_branch_results( $aggregate_transition['aggregate'], $aggregate_transition['branch_results'], \AgentsAPI\AI\Workflows\agents_workflow_resolve_step_handlers() ); +\AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'as-committed-crash', + static function () use ( $committed_recorder, $aggregate_transition, $aggregate_output ) { + return \AgentsAPI\AI\Workflows\agents_workflow_commit_reconcile_claim( $committed_recorder, 'as-committed-crash', $aggregate_transition['owner_token'], $aggregate_transition['generation'], $aggregate_transition['step_index'], $aggregate_output ); + } +); +WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_aggregate_action_failure( $payload ); +as_smoke_fire_resume_actions(); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $committed_recorder->find( 'as-committed-crash' )->get_status(), 'aggregate committed crash: failure lifecycle resumes durable output', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__as_aggregate_effects'], 'aggregate committed crash: recovery skips aggregator execution', $failures, $passes ); + +// RECORDER FAILURES. A failed queued -> running write starts no effects and the +// failed-action callback re-enqueues. A failed commit leaves `running`, so the +// callback terminalizes uncertain without rerunning effects. +list( $write_recorder, $aggregate_action ) = as_smoke_prepare_aggregate_run( 'as-running-write' ); +$write_recorder->fail_next_update(); +try { + AS_Shim::fire( $aggregate_action['id'] ); +} catch ( \RuntimeException $error ) { + unset( $error ); +} +WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_aggregate_action_failure( $aggregate_action['args'][0] ); +smoke_assert( 0, $GLOBALS['__as_aggregate_effects'], 'aggregate running write: effects do not start before durable running phase', $failures, $passes ); +smoke_assert( 2, count( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ) ), 'aggregate running write: failure callback re-enqueues safe queued work', $failures, $passes ); + +list( $commit_recorder, $aggregate_action ) = as_smoke_prepare_aggregate_run( 'as-commit-write' ); +$GLOBALS['__as_during_aggregate'] = static function () use ( $commit_recorder ): void { + $GLOBALS['__as_during_aggregate'] = null; + $commit_recorder->fail_next_update(); +}; +try { + AS_Shim::fire( $aggregate_action['id'] ); +} catch ( \RuntimeException $error ) { + unset( $error ); +} +WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_aggregate_action_failure( $aggregate_action['args'][0] ); +as_smoke_fire_resume_actions(); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $commit_recorder->find( 'as-commit-write' )->get_status(), 'aggregate commit write: failure callback terminalizes uncertain state', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__as_aggregate_effects'], 'aggregate commit write: recorder uncertainty never repeats effects', $failures, $passes ); + +list( $recovery_recorder, $aggregate_action ) = as_smoke_prepare_aggregate_run( 'as-failure-write' ); +$payload = $aggregate_action['args'][0]; +\AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( + 'as-failure-write', + static function () use ( $recovery_recorder, $payload ) { + return \AgentsAPI\AI\Workflows\agents_workflow_begin_aggregate_action_locked( $recovery_recorder, 'as-failure-write', $payload['generation'], $payload['owner_token'] ); + } +); +$recovery_recorder->fail_next_update(); +WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_aggregate_action_failure( $payload ); +$aggregate_actions = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ); +smoke_assert( true, ! empty( $aggregate_actions[1]['args'][0]['recover_failure'] ), 'aggregate failure write: recorder outage enqueues durable recovery action', $failures, $passes ); +AS_Shim::fire( $aggregate_actions[1]['id'] ); +as_smoke_fire_resume_actions(); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $recovery_recorder->find( 'as-failure-write' )->get_status(), 'aggregate failure write: recovery action eventually terminalizes', $failures, $passes ); + echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; exit( count( $failures ) > 0 ? 1 : 0 ); diff --git a/tests/workflow-async-branch-payload-smoke.php b/tests/workflow-async-branch-payload-smoke.php index a6dbf11..ae1e972 100644 --- a/tests/workflow-async-branch-payload-smoke.php +++ b/tests/workflow-async-branch-payload-smoke.php @@ -450,6 +450,9 @@ function payload_roles_spec(): WP_Agent_Workflow_Spec { foreach ( $branch_actions as $action ) { AS_Limit_Shim::fire( $action['id'] ); } +foreach ( AS_Limit_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ) as $action ) { + AS_Limit_Shim::fire( $action['id'] ); +} foreach ( AS_Limit_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ) as $action ) { AS_Limit_Shim::fire( $action['id'] ); } diff --git a/tests/workflow-reconcile-race-smoke.php b/tests/workflow-reconcile-race-smoke.php index 009bda3..7d7c5c5 100644 --- a/tests/workflow-reconcile-race-smoke.php +++ b/tests/workflow-reconcile-race-smoke.php @@ -554,108 +554,5 @@ static function ( bool $deferred ): bool { smoke_assert( 1, $GLOBALS['__resume_dispatch_calls'], 'ttl: one resume dispatches for the claimed suspension generation', $failures, $passes ); smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $final->get_status(), 'ttl: claimed owner commits and resumes successfully', $failures, $passes ); -// CLAIM WRITE FAILURE: aggregation cannot start until the pending continuation -// and final completion are durable. A persisted-result retry can safely repeat -// the transition because no external aggregate effect has begun. -list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'claim-write' ); -$recorder->fail_next_update(); -$failed_claim = agents_reconcile_workflow_branch( 'claim-write', (string) $descriptors[2]['handle_id'], $last_result ); -smoke_assert( 'agents_reconcile_lock_unavailable', is_wp_error( $failed_claim ) ? $failed_claim->get_error_code() : '', 'claim write: recorder failure returns PR #534 retry contract', $failures, $passes ); -smoke_assert( '', $recorder->reconcile_phase( 'claim-write' ), 'claim write: failed update installs no phantom continuation', $failures, $passes ); -smoke_assert( 0, $GLOBALS['__aggregate_calls'], 'claim write: aggregate effect does not run without durable ownership', $failures, $passes ); -$retried_claim = agents_reconcile_workflow_branch( 'claim-write', (string) $descriptors[2]['handle_id'], $last_result ); -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $retried_claim->get_status(), 'claim write: persisted-result retry safely completes the run', $failures, $passes ); -smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'claim write: retry executes aggregator exactly once', $failures, $passes ); - -// AGGREGATE COMMIT WRITE FAILURE: effects have run, so a retry must never rerun -// them. Once the abandoned owner is fenced stale, continuation persists an -// honest uncertain-outcome failure and resumes the run to terminal. -list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'commit-write' ); -$GLOBALS['__during_aggregate'] = static function () use ( $recorder ): void { - $GLOBALS['__during_aggregate'] = null; - $recorder->fail_next_update(); -}; -$failed_commit = agents_reconcile_workflow_branch( 'commit-write', (string) $descriptors[2]['handle_id'], $last_result ); -smoke_assert( 'agents_reconcile_lock_unavailable', is_wp_error( $failed_commit ) ? $failed_commit->get_error_code() : '', 'commit write: recorder failure requests persisted reconcile retry', $failures, $passes ); -smoke_assert( 'aggregating', $recorder->reconcile_phase( 'commit-write' ), 'commit write: durable effect-start phase remains authoritative', $failures, $passes ); -$recorder->expire_reconcile_claim( 'commit-write' ); -$commit_recovery = agents_reconcile_workflow_branch( 'commit-write', (string) $descriptors[2]['handle_id'], $last_result ); -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $commit_recovery->get_status(), 'commit write: stale ambiguous owner terminalizes honestly', $failures, $passes ); -smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'commit write: external aggregator is never rerun after uncertainty', $failures, $passes ); -smoke_assert( 'workflow_parallel_aggregation_outcome_uncertain', $commit_recovery->get_error()['code'] ?? '', 'commit write: terminal failure names uncertain aggregate outcome', $failures, $passes ); - -// SECOND LOCK CONTENTION: the original aggregate result cannot be committed, and -// PR #534 receives the same retryable contract. Redelivery advances the persisted -// aggregating phase instead of stopping at the completed-handle guard. -list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'commit-lock' ); -$lock_calls = 0; -add_filter( - 'wp_agent_workflow_reconcile_lock', - static function ( $override, string $run_id, callable $critical ) use ( &$lock_calls ) { - unset( $override, $run_id ); - ++$lock_calls; - return 3 === $lock_calls ? new WP_Error( 'agents_reconcile_lock_unavailable', 'Injected second-phase contention.' ) : $critical(); - }, - 10, - 3 -); -$contended_commit = agents_reconcile_workflow_branch( 'commit-lock', (string) $descriptors[2]['handle_id'], $last_result ); -remove_all_filters( 'wp_agent_workflow_reconcile_lock' ); -smoke_assert( 'agents_reconcile_lock_unavailable', is_wp_error( $contended_commit ) ? $contended_commit->get_error_code() : '', 'commit lock: second-phase contention requests reconcile-only retry', $failures, $passes ); -smoke_assert( 'aggregating', $recorder->reconcile_phase( 'commit-lock' ), 'commit lock: continuation records that effects may have begun', $failures, $passes ); -$recorder->expire_reconcile_claim( 'commit-lock' ); -$lock_recovery = agents_reconcile_workflow_branch( 'commit-lock', (string) $descriptors[2]['handle_id'], $last_result ); -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $lock_recovery->get_status(), 'commit lock: completed-handle retry advances to terminal recovery', $failures, $passes ); -smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'commit lock: continuation never repeats aggregator effects', $failures, $passes ); - -// PROCESS LOSS BEFORE AGGREGATION: stop after persisting `pending`, then let a -// duplicate completed-result delivery start and finish the safe continuation. -list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'lost-before' ); -$pending = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( - 'lost-before', - static function () use ( $recorder, $descriptors, $last_result ) { - return \AgentsAPI\AI\Workflows\agents_reconcile_workflow_branch_locked( $recorder, 'lost-before', (string) $descriptors[2]['handle_id'], $last_result ); - } -); -smoke_assert( 'begin', $pending['action'] ?? '', 'lost before aggregate: pending continuation persists before effects', $failures, $passes ); -smoke_assert( 'pending', $recorder->reconcile_phase( 'lost-before' ), 'lost before aggregate: durable phase proves effects have not begun', $failures, $passes ); -$before_recovery = agents_reconcile_workflow_branch( 'lost-before', (string) $descriptors[2]['handle_id'], $last_result ); -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $before_recovery->get_status(), 'lost before aggregate: duplicate delivery safely continues pending work', $failures, $passes ); -smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'lost before aggregate: recovery executes aggregator once', $failures, $passes ); - -// PROCESS LOSS AFTER COMMIT BEFORE RESUME: drive through the durable commit but -// omit dispatch. A duplicate reconcile observes `committed`, skips aggregation, -// and resumes from the recorded output. -list( $recorder, $descriptors, $last_result ) = race_prepare_continuation_run( 'lost-after' ); -$pending = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( - 'lost-after', - static function () use ( $recorder, $descriptors, $last_result ) { - return \AgentsAPI\AI\Workflows\agents_reconcile_workflow_branch_locked( $recorder, 'lost-after', (string) $descriptors[2]['handle_id'], $last_result ); - } -); -$aggregation = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( - 'lost-after', - static function () use ( $recorder, $pending ) { - return \AgentsAPI\AI\Workflows\agents_workflow_begin_reconcile_aggregation( $recorder, 'lost-after', $pending['generation'] ); - } -); -$aggregate_output = WP_Agent_Workflow_Runner::aggregate_branch_results( - $aggregation['aggregate'], - $aggregation['branch_results'], - \AgentsAPI\AI\Workflows\agents_workflow_resolve_step_handlers() -); -$committed = \AgentsAPI\AI\Workflows\agents_workflow_reconcile_with_lock( - 'lost-after', - static function () use ( $recorder, $aggregation, $aggregate_output ) { - return \AgentsAPI\AI\Workflows\agents_workflow_commit_reconcile_claim( $recorder, 'lost-after', $aggregation['owner_token'], $aggregation['generation'], $aggregation['step_index'], $aggregate_output ); - } -); -smoke_assert( true, isset( $committed['result'] ), 'lost after commit: aggregate output is durable before resume dispatch', $failures, $passes ); -smoke_assert( 'committed', $recorder->reconcile_phase( 'lost-after' ), 'lost after commit: durable phase is resumable', $failures, $passes ); -$after_recovery = agents_reconcile_workflow_branch( 'lost-after', (string) $descriptors[2]['handle_id'], $last_result ); -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $after_recovery->get_status(), 'lost after commit: duplicate delivery resumes durable aggregate', $failures, $passes ); -smoke_assert( 1, $GLOBALS['__aggregate_calls'], 'lost after commit: recovery does not rerun aggregator', $failures, $passes ); -smoke_assert( 1, $GLOBALS['__resume_dispatch_calls'], 'lost after commit: recovery dispatches resume once', $failures, $passes ); - echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; exit( count( $failures ) > 0 ? 1 : 0 ); diff --git a/tests/workflow-request-controller-smoke.php b/tests/workflow-request-controller-smoke.php index 19ea297..ea43cbf 100644 --- a/tests/workflow-request-controller-smoke.php +++ b/tests/workflow-request-controller-smoke.php @@ -132,7 +132,7 @@ function controller_assert( bool $ok, string $name ): void { global $fails, $pas controller_assert( in_array( 'failed', $cleaned, true ) && in_array( 'cancelled', $cleaned, true ), 'failed and cancelled runs release terminal cleanup' ); controller_assert( array() === ( $controller->get( 'one' )['lease'] ?? null ), 'terminal lease is cleared' ); $one_group_cleanup = array_filter( $GLOBALS['controller_unscheduled'], static function ( array $call ) use ( $one ): bool { return 'agents-api-run-' . md5( $one['run_id'] ) === $call[2]; } ); -controller_assert( 2 === count( $one_group_cleanup ), 'terminal cleanup removes only the run-scoped branch and resume actions' ); +controller_assert( 3 === count( $one_group_cleanup ), 'terminal cleanup removes only the run-scoped branch, aggregate, and resume actions' ); controller_assert( array() === array_filter( $one_group_cleanup, static fn ( array $call ): bool => null !== $call[1] ), 'terminal cleanup matches every argument shape in the run-scoped group' ); $state = WP_Agent_Run_Control::state( 'controller-test' ); $state['runs']['two']['lease'] = array( 'token' => 'other-worker', 'worker_id' => 'worker-a', 'expires_at' => time() + 60 ); diff --git a/tests/workflow-scoped-drain-smoke.php b/tests/workflow-scoped-drain-smoke.php index 48b2b8b..6fefcdd 100644 --- a/tests/workflow-scoped-drain-smoke.php +++ b/tests/workflow-scoped-drain-smoke.php @@ -315,15 +315,16 @@ function as_get_datetime_object( ?string $date_string = null, string $timezone = use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Scoped_Drain; $branch_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK; +$aggregate_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK; $resume_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK; $group = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::GROUP; // The drain's default scope must be the executor's hooks + group (read, never // hardcoded), so this and the executor can never drift. smoke_assert( - array( $branch_hook, $resume_hook ), + array( $branch_hook, $aggregate_hook, $resume_hook ), WP_Agent_Workflow_Scoped_Drain::default_hooks(), - 'default_hooks() = executor BRANCH_HOOK + RESUME_HOOK', + 'default_hooks() includes branch, aggregate, and resume continuations', $failures, $passes ); From b6083d13ad50eb09853b685021c0ecddb0a7e7e4 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 19:17:06 +0000 Subject: [PATCH 4/5] fix: deduplicate workflow resume dispatch --- ...kflow-action-scheduler-branch-executor.php | 79 ++++++++++++++++--- .../register-reconcile-workflow-branch.php | 25 ++++-- tests/workflow-as-branch-smoke.php | 73 +++++++++-------- 3 files changed, 129 insertions(+), 48 deletions(-) diff --git a/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php b/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php index c71e969..c4f7b94 100644 --- a/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php +++ b/src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php @@ -1413,20 +1413,41 @@ public static function maybe_defer_resume( bool $deferred, string $run_id, strin $suspension = is_object( $result ) && method_exists( $result, 'get_suspension' ) ? self::string_keyed_array( (array) $result->get_suspension() ) : array(); - $action_id = self::enqueue_async_action( - self::RESUME_HOOK, + $args = array( array( - array( - 'run_id' => $run_id, - 'suspension_id' => self::suspension_id( $suspension ), - ), + 'run_id' => $run_id, + 'suspension_id' => self::suspension_id( $suspension ), ), - self::group_for_run( $run_id ) ); + $group = self::group_for_run( $run_id ); + if ( self::has_scheduled_action( self::RESUME_HOOK, $args, $group ) ) { + return true; + } - // If durable enqueue fails, return false so reconcile resumes inline rather - // than stranding a suspended run with no resume action. - return $action_id > 0; + $unique = self::supports_unique_enqueue(); + $query = self::supports_scheduled_action_query(); + if ( ! $unique && ! $query ) { + // The caller holds the per-run lock and will perform the inline fallback + // there. Do not enqueue a non-unique action that could race it. + return false; + } + $action_id = self::enqueue_async_action( + self::RESUME_HOOK, + $args, + $group, + $unique + ); + + // A unique duplicate returns 0. It is still successful deferral when the + // identical pending/running action is durably discoverable. + if ( $action_id > 0 || self::has_scheduled_action( self::RESUME_HOOK, $args, $group ) ) { + return true; + } + + // Unique enqueue without a query helper cannot distinguish duplicate 0 + // from failure. Conservatively remain deferred rather than race an existing + // action with inline resume. + return $unique && ! $query; } /** @@ -1524,6 +1545,44 @@ private static function enqueue_async_action( string $hook, array $args, string } } + /** Whether this Action Scheduler version exposes the unique enqueue argument. */ + private static function supports_unique_enqueue(): bool { + if ( ! function_exists( 'as_enqueue_async_action' ) ) { + return false; + } + try { + return ( new \ReflectionFunction( 'as_enqueue_async_action' ) )->getNumberOfParameters() >= 4; + } catch ( \ReflectionException $error ) { + unset( $error ); + return false; + } + } + + /** Whether this Action Scheduler version can query an identical action. */ + private static function supports_scheduled_action_query(): bool { + return function_exists( 'as_has_scheduled_action' ) || function_exists( 'as_next_scheduled_action' ); + } + + /** + * Whether an identical pending/running action already owns this continuation. + * + * @phpstan-impure Action Scheduler state may change after an enqueue attempt. + * @param array $args Action arguments. + */ + private static function has_scheduled_action( string $hook, array $args, string $group ): bool { + try { + if ( function_exists( 'as_has_scheduled_action' ) ) { + return (bool) as_has_scheduled_action( $hook, $args, $group ); + } + if ( function_exists( 'as_next_scheduled_action' ) ) { + return false !== as_next_scheduled_action( $hook, $args, $group ); + } + } catch ( \Throwable $error ) { + unset( $error ); + } + return false; + } + /** * Cancel every action inserted before a sibling enqueue failed. * diff --git a/src/Workflows/register-reconcile-workflow-branch.php b/src/Workflows/register-reconcile-workflow-branch.php index 149ce0a..20e40cd 100644 --- a/src/Workflows/register-reconcile-workflow-branch.php +++ b/src/Workflows/register-reconcile-workflow-branch.php @@ -592,14 +592,29 @@ function agents_workflow_update_reconcile_state( WP_Agent_Workflow_Run_Recorder /** * Resume a durably committed continuation without rerunning aggregation. * - * @return WP_Agent_Workflow_Run_Result + * @return WP_Agent_Workflow_Run_Result|\WP_Error */ function agents_workflow_resume_reconcile_continuation( WP_Agent_Workflow_Run_Recorder $recorder, string $run_id, WP_Agent_Workflow_Run_Result $result ) { - if ( agents_workflow_defer_resume( $run_id, $result ) ) { - return $result; + $dispatch = agents_workflow_reconcile_with_lock( + $run_id, + static function () use ( $recorder, $run_id, $result ) { + $current = $recorder->find( $run_id ); + if ( null === $current || ! $current->is_suspended() ) { + return null !== $current ? $current : $result; + } + if ( agents_workflow_defer_resume( $run_id, $current ) ) { + return $current; + } + + // Legacy/unavailable unique-action surfaces fall back inline while the + // per-run lock is still held, so duplicate dispatchers cannot overlap. + return agents_workflow_resolve_runner( $recorder )->resume( $run_id ); + } + ); + if ( is_wp_error( $dispatch ) ) { + return $dispatch; } - $runner = agents_workflow_resolve_runner( $recorder ); - return $runner->resume( $run_id ); + return $dispatch; } /** diff --git a/tests/workflow-as-branch-smoke.php b/tests/workflow-as-branch-smoke.php index 74b382b..aa0a03c 100644 --- a/tests/workflow-as-branch-smoke.php +++ b/tests/workflow-as-branch-smoke.php @@ -275,6 +275,21 @@ function as_enqueue_async_action( string $hook, array $args = array(), string $g return AS_Shim::enqueue( $hook, $args, $group, $unique ); } } +if ( ! function_exists( 'as_has_scheduled_action' ) ) { + function as_has_scheduled_action( string $hook, ?array $args = null, string $group = '' ): bool { + foreach ( AS_Shim::$queue as $action ) { + if ( + $hook === $action['hook'] && + ( null === $args || $args === $action['args'] ) && + $group === $action['group'] && + empty( AS_Shim::$claimed[ $action['id'] ] ) + ) { + return true; + } + } + return false; + } +} function as_smoke_fire_aggregate_actions(): void { foreach ( AS_Shim::actions_for( \AgentsAPI\AI\Workflows\WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ) as $action ) { @@ -408,6 +423,7 @@ static function ( array $input ): array { as_smoke_register_ability( 'demo/consume', static function ( array $input ): array { + $GLOBALS['__consume_effects'] = (int) ( $GLOBALS['__consume_effects'] ?? 0 ) + 1; return array( 'consumed' => 'GOT:' . (string) ( $input['bundle'] ?? '' ) ); } ); @@ -664,15 +680,8 @@ static function ( $result, string $run_id, string $handle_id, array $branch_resu // Fire the first branch normally. AS_Shim::fire( $branch_actions2[0]['id'] ); -// Now simulate TWO processes both finishing the LAST branch "at once". We drive -// the reconcile for the last branch directly TWICE from a frame state where the -// last handle is still outstanding — but the second call is a genuine duplicate. -// The real guard we prove: even if TWO resume actions are enqueued, AS's claim + -// the SUSPENDED re-check make exactly one resume effective. -// -// To create two enqueued RESUME actions we reconcile the last branch, then -// hand-enqueue a SECOND identical resume (as a lagging duplicate process would), -// mirroring "N branches each enqueue a resume action" from the design. +// Complete aggregation, then invoke two duplicate dispatchers before any resume +// worker executes. Both observe the same committed, still-SUSPENDED generation. $payload2 = $branch_actions2[1]['args'][0] ?? array(); $last_handle_id = (string) ( $payload2['handle_id'] ?? '' ); AS_Shim::fire( $branch_actions2[1]['id'] ); // last branch → reconcile all-terminal → enqueues resume #1 @@ -680,34 +689,32 @@ static function ( $result, string $run_id, string $handle_id, array $branch_resu $resume_actions2 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); smoke_assert( 1, count( $resume_actions2 ), 'race: last branch enqueued resume #1', $failures, $passes ); - -// A second, lagging finisher for the SAME run enqueues resume #2 (the race: -// both observed all-terminal before either resumed). Enqueue it directly to -// model the second process, then drive BOTH resume actions through AS's claim. -$resume_id_2 = AS_Shim::enqueue( - WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK, - array( array( 'run_id' => 'as-race' ) ), - WP_Agent_Workflow_Action_Scheduler_Branch_Executor::GROUP +$GLOBALS['__consume_effects'] = 0; +$race_completions = 0; +add_action( + 'wp_agent_workflow_run_completed', + static function ( $result, string $run_id ) use ( &$race_completions ): void { + unset( $result ); + if ( 'as-race' === $run_id ) { + ++$race_completions; + } + }, + 10, + 2 ); +$committed2 = $recorder2->find( 'as-race' ); +\AgentsAPI\AI\Workflows\agents_workflow_resume_reconcile_continuation( $recorder2, 'as-race', $committed2 ); +\AgentsAPI\AI\Workflows\agents_workflow_resume_reconcile_continuation( $recorder2, 'as-race', $committed2 ); $resume_actions2 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); -smoke_assert( 2, count( $resume_actions2 ), 'race: two RESUME actions are enqueued (simultaneous finish)', $failures, $passes ); - -// Drive AS's claim: fire both. Exactly one claims-and-runs the effective resume; -// the other is either a claimed no-op OR runs against an already-resumed run and -// bails on the SUSPENDED re-check. Count how many actually resumed the run. -$fired_first = AS_Shim::fire( $resume_actions2[0]['id'] ); -$status_after_first = $recorder2->find( 'as-race' )->get_status(); -$fired_second = AS_Shim::fire( $resume_actions2[1]['id'] ); -$status_after_second = $recorder2->find( 'as-race' )->get_status(); - -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $status_after_first, 'race: first claimed resume runs the run to SUCCEEDED', $failures, $passes ); -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $status_after_second, 'race: run stays SUCCEEDED after the second resume (no corruption / no double-run)', $failures, $passes ); - -// The second resume must be a NO-OP: its handler re-checked SUSPENDED and bailed -// (the run already resumed). We prove exactly-once by asserting the sequential -// `after` step ran exactly once with the correct output. +smoke_assert( 1, count( $resume_actions2 ), 'race: simultaneous duplicate dispatchers retain one unique resume action', $failures, $passes ); +AS_Shim::fire( $resume_actions2[0]['id'] ); +$status_after_resume = $recorder2->find( 'as-race' )->get_status(); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $status_after_resume, 'race: unique claimed resume reaches success', $failures, $passes ); +smoke_assert( 1, $GLOBALS['__consume_effects'], 'race: downstream resume effect executes exactly once', $failures, $passes ); +smoke_assert( 1, $race_completions, 'race: completion hook fires exactly once', $failures, $passes ); $race_out = $recorder2->find( 'as-race' )->get_output()['steps'] ?? array(); smoke_assert( 'GOT:FUSED[HEAD|BODY]', $race_out['after']['consumed'] ?? '', 'race: exactly-once resume — sequential step ran once with the aggregated output', $failures, $passes ); +remove_all_filters( 'wp_agent_workflow_run_completed' ); // ═════════════════════════════════════════════════════════════════════════════ // 3. CRASH-RESUME DURABILITY From caf048d4de18a28024a608c6ad0771bea7f0376e Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 19:23:49 +0000 Subject: [PATCH 5/5] fix: clean inline workflow resume state --- src/Workflows/register-reconcile-workflow-branch.php | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/Workflows/register-reconcile-workflow-branch.php b/src/Workflows/register-reconcile-workflow-branch.php index 20e40cd..c4a5685 100644 --- a/src/Workflows/register-reconcile-workflow-branch.php +++ b/src/Workflows/register-reconcile-workflow-branch.php @@ -608,7 +608,11 @@ static function () use ( $recorder, $run_id, $result ) { // Legacy/unavailable unique-action surfaces fall back inline while the // per-run lock is still held, so duplicate dispatchers cannot overlap. - return agents_workflow_resolve_runner( $recorder )->resume( $run_id ); + $resumed = agents_workflow_resolve_runner( $recorder )->resume( $run_id ); + if ( ! $resumed->is_suspended() && class_exists( WP_Agent_Workflow_Branch_Store::class ) ) { + WP_Agent_Workflow_Branch_Store::forget_run( $run_id ); + } + return $resumed; } ); if ( is_wp_error( $dispatch ) ) {