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 0664d8a..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 @@ -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'; + /** * Reconcile-only retry hook. Its payload references a persisted terminal * BranchResult, so this callback never executes branch steps. @@ -459,6 +462,27 @@ public static function reconcile_inflight_count(): int { return is_array( $ids ) ? count( $ids ) : 0; } + /** Count durable aggregate continuations still pending or in progress. */ + public static function aggregate_inflight_count(): int { + if ( ! function_exists( 'as_get_scheduled_actions' ) || ! class_exists( '\ActionScheduler_Store' ) ) { + return 0; + } + + $ids = as_get_scheduled_actions( + array( + 'hook' => self::AGGREGATE_HOOK, + 'status' => array( + \ActionScheduler_Store::STATUS_PENDING, + \ActionScheduler_Store::STATUS_RUNNING, + ), + 'per_page' => self::MAX_BRANCH_CONCURRENCY, + ), + 'ids' + ); + + return is_array( $ids ) ? count( $ids ) : 0; + } + /** * Trigger Action Scheduler's async-request queue runner: fire N CONCURRENT * loopback HTTP requests to admin-ajax.php so AS spawns N separate native-PHP @@ -877,6 +901,84 @@ public static function run_branch_action( array $payload ): void { self::reconcile_branch_action_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 ( self::AGGREGATE_HOOK !== $action->get_hook() ) { + return; + } + $args = $action->get_args(); + $payload = is_array( $args[0] ?? null ) ? $args[0] : array(); + self::run_aggregate_action_failure( $payload ); + } catch ( \Throwable $error ) { + unset( $error ); + } + } + /** * Reconcile a completed branch result directly from memory. Only lock * contention persists a retry receipt, minimizing post-terminal writes. @@ -1311,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; + } + + $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; + } - // 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 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; } /** @@ -1405,14 +1528,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 @@ -1422,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/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 @@ $branch_result BranchResult. - * @return 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 ); @@ -175,10 +202,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 @@ -186,113 +211,453 @@ 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 ) + ); + } + + $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; + } + } } - $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 ) + $suspension['handles'] = $handles; + $suspension['completed'] = $completed; + + $transition = null; + if ( count( $completed ) >= count( $handles ) ) { + $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(); + $metadata['_suspension'] = $suspension; + $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; + } + + if ( null === $transition ) { + return $result; + } + if ( 'resume' === $transition['action'] ) { + return array( + 'action' => 'resume', + 'result' => $result, ); } + return $transition; +} - $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, +/** + * 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 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}|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 ); + 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 ( + '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 ) ) + ) { + return agents_workflow_advance_reconcile_continuation_locked( $recorder, $result ); + } + + $suspension['reconcile_claim'] = array( + 'phase' => 'committed', + 'generation' => $generation, + 'owner_token' => $claim_token, ); + $metadata = $result->get_metadata(); + $metadata['_suspension'] = $suspension; + $result = $result->with( array( 'metadata' => $metadata ) ); + $result = agents_workflow_splice_step_output( $result, $step_index, $step_output ); + $updated = agents_workflow_update_reconcile_state( $recorder, $result, 'commit aggregate output' ); + if ( is_wp_error( $updated ) ) { + return $updated; + } + return array( 'result' => $result ); +} - // 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; +/** + * 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 $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'] ); } - $suspension['handles'] = $handles; - $suspension['completed'] = $completed; - - $metadata = $result->get_metadata(); - $metadata['_suspension'] = $suspension; - $result = $result->with( array( 'metadata' => $metadata ) ); - $recorder->update( $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'] ); +} - // Not all terminal yet → wait for more reconcile calls. - if ( count( $completed ) < count( $handles ) ) { +/** + * 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_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 ) ); + } + 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 ( + '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 ) ) + ) { return $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 ); + $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 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', + '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 ), + ); +} + +/** + * Duplicate reconciles never take over queued or running aggregate actions. + * + * @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 ) { + 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 ); + } + 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; + } + ); + 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 $transition; +} + +/** + * 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 ); + $claim = is_array( $suspension['reconcile_claim'] ?? null ) ? $suspension['reconcile_claim'] : array(); + $suspension['reconcile_claim'] = array( + 'phase' => 'committed', + 'generation' => $generation, + 'owner_token' => agents_workflow_string( $claim['owner_token'] ?? '' ), + ); + $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; - $aggregate = is_array( $suspension['aggregate'] ?? null ) ? \AgentsAPI\AI\WP_Agent_Run_Control::string_keyed_array( $suspension['aggregate'] ) : array(); - $handlers = agents_workflow_resolve_step_handlers(); + $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, + ); +} - 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 ); +/** + * 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; +} - // 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 ); - - // 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; +/** + * Resume a durably committed continuation without rerunning aggregation. + * + * @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 ) { + $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. + $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 ) ) { + return $dispatch; } + return $dispatch; +} - // 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 ); +/** + * 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'] ?? '' ); + } + } + + 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/src/Workflows/register-workflow-branch-executor.php b/src/Workflows/register-workflow-branch-executor.php index a3954f7..e5132ca 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 @@ -87,6 +90,7 @@ static function ( $payload = array() ): void { static function ( $action_id, $error = null ): void { if ( is_int( $action_id ) || is_string( $action_id ) ) { WP_Agent_Workflow_Action_Scheduler_Branch_Executor::recover_failed_action( $action_id, $error ); + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::handle_failed_action( is_numeric( $action_id ) ? (int) $action_id : 0 ); } }, 20, @@ -97,6 +101,7 @@ static function ( $action_id, $error = null ): void { static function ( $action_id, $error = null ): void { if ( is_int( $action_id ) || is_string( $action_id ) ) { WP_Agent_Workflow_Action_Scheduler_Branch_Executor::recover_failed_action( $action_id, $error ); + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::handle_failed_action( is_numeric( $action_id ) ? (int) $action_id : 0 ); } }, 20, @@ -117,6 +122,15 @@ static function ( $payload = array() ): void { ); // 3b. 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, /** @@ -129,6 +143,38 @@ 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 +); + +// Fatal shutdown is separate from AS's failed execution/timeout hooks above. +add_action( + 'action_scheduler_unexpected_shutdown', + static function ( $action_id, $error = null ): void { + if ( is_int( $action_id ) || is_string( $action_id ) ) { + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::recover_failed_action( $action_id, $error ); + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::handle_failed_action( is_numeric( $action_id ) ? (int) $action_id : 0 ); + } + }, + 20, + 2 +); + // 3c. Deferred-resume seam: enqueue a claimed RESUME action for AS-owned runs // instead of resuming inline in the reconcile request. add_filter( @@ -272,7 +318,8 @@ static function ( $batches ) { $branches = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::branch_inflight_count(); $resumes = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::resume_inflight_count(); $reconciles = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::reconcile_inflight_count(); - if ( $branches < 1 && $resumes < 1 && $reconciles < 1 ) { + $aggregates = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::aggregate_inflight_count(); + if ( $branches < 1 && $resumes < 1 && $reconciles < 1 && $aggregates < 1 ) { return $incoming; } @@ -292,8 +339,9 @@ static function ( $batches ) { // each bounded count lifts the ceiling so the WP-Cron runner can claim it. $resume_headroom = min( $resumes, $max ); $reconcile_headroom = min( $reconciles, $max ); + $aggregate_headroom = min( $aggregates, $max ); - return $branch_ceiling + $resume_headroom + $reconcile_headroom; + return $branch_ceiling + $resume_headroom + $reconcile_headroom + $aggregate_headroom; }, 100 ); @@ -308,7 +356,10 @@ static function ( $batch_size ) { // Pin to 1 while branches are in flight (pending or in-progress) so each worker // claims exactly one branch; otherwise pass the incoming value through so // ordinary AS throughput (25) is untouched. - if ( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::branch_inflight_count() > 0 ) { + if ( + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::branch_inflight_count() > 0 || + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::aggregate_inflight_count() > 0 + ) { return 1; } return is_numeric( $batch_size ) ? (int) $batch_size : 25; diff --git a/tests/workflow-as-branch-smoke.php b/tests/workflow-as-branch-smoke.php index 2157129..aa0a03c 100644 --- a/tests/workflow-as-branch-smoke.php +++ b/tests/workflow-as-branch-smoke.php @@ -170,10 +170,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, @@ -264,8 +271,29 @@ public function fetch_action( $action_id ): AS_Shim_Action { } 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 ); + } +} +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 ) { + AS_Shim::fire( $action['id'] ); } } @@ -321,6 +349,7 @@ 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 int $updates = 0; public function start( WP_Agent_Workflow_Run_Result $result ) { @@ -329,6 +358,10 @@ public function start( WP_Agent_Workflow_Run_Result $result ) { } public function update( WP_Agent_Workflow_Run_Result $result ) { ++$this->updates; + 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; } @@ -355,6 +388,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 ───────── @@ -376,12 +413,17 @@ 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'] ?? '' ) . ']' ); } ); 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'] ?? '' ) ); } ); @@ -593,6 +635,7 @@ static function ( $result, string $run_id, string $handle_id, array $branch_resu 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). @@ -637,49 +680,41 @@ 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 +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 ); - -// 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 @@ -705,6 +740,7 @@ static function ( $result, string $run_id, string $handle_id, array $branch_resu 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'] ); @@ -784,6 +820,7 @@ function as_smoke_single_branch_spec( string $label ): 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'] ); } @@ -824,6 +861,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'] ); @@ -839,12 +877,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 ); // A contended result is persisted only after the in-memory reconcile attempt. If // that receipt write fails, the action fails loudly and queues no unreadable retry. AS_Shim::reset(); @@ -924,6 +1096,7 @@ static function ( $override, string $run_id, callable $critical ) use ( &$edge_l smoke_assert( 'workflow_branch_descriptor_missing', $edge_receipt['branch_result']['error']['code'] ?? '', $descriptor_state . ' descriptor: receipt durably carries terminal missing-descriptor failure', $failures, $passes ); AS_Shim::fire( $edge_retries[0]['id'] ); smoke_assert( false, array_key_exists( $edge_store_ref, $GLOBALS['__options'] ), $descriptor_state . ' descriptor: successful reconcile deletes exact receipt-only row', $failures, $passes ); + as_smoke_fire_aggregate_actions(); $edge_resumes = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); AS_Shim::fire( $edge_resumes[0]['id'] ); $edge_final = $edge_recorder->find( $edge_run_id ); @@ -1080,6 +1253,7 @@ static function ( $override, string $run_id, callable $critical ) use ( &$transi ); $transition_effect_before9 = (int) ( $GLOBALS['__role_worker_effects']['transition'] ?? 0 ); AS_Shim::fire( $branches9[0]['id'] ); +as_smoke_fire_aggregate_actions(); $resumes9 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); AS_Shim::fire( $resumes9[0]['id'] ); $transition_final9 = $recorder9->find( 'as-transition' ); @@ -1163,6 +1337,7 @@ static function ( $result, string $run_id ) use ( &$tokenless_completions ): voi ); $tokenless_effect_before = (int) ( $GLOBALS['__role_worker_effects']['tokenless'] ?? 0 ); AS_Shim::fire( $tokenless_branches[0]['id'] ); +as_smoke_fire_aggregate_actions(); $tokenless_resumes = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); AS_Shim::fire( $tokenless_resumes[0]['id'] ); $tokenless_final = $recorder_tokenless->find( 'as-tokenless' ); @@ -1231,6 +1406,7 @@ static function ( $override, string $run_id, callable $critical ) use ( &$distin AS_Shim::$reject_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RECONCILE_HOOK; AS_Shim::fire_with_failure_lifecycle( $branches11[0]['id'] ); AS_Shim::$reject_hook = ''; +as_smoke_fire_aggregate_actions(); $resumes11 = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); AS_Shim::fire( $resumes11[0]['id'] ); $distinct_final11 = $recorder11->find( 'as-distinct' ); @@ -1268,11 +1444,12 @@ static function ( $result, string $run_id ) use ( &$late_failure_completions ): 2 ); AS_Shim::fire( $late_failure_branches[0]['id'] ); -$late_failure_resumes = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); do_action( 'action_scheduler_failed_execution', $late_failure_branches[0]['id'], new RuntimeException( 'late timeout callback' ), 'test' ); $late_failure_mid = $recorder_late_failure->find( 'as-late-failure' ); -smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED, $late_failure_mid->get_status(), 'late failed callback: reconciled handle preserves suspended run for queued resume', $failures, $passes ); -smoke_assert( 1, count( $late_failure_resumes ), 'late failed callback: queued resume remains authoritative', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED, $late_failure_mid->get_status(), 'late failed callback: reconciled handle preserves suspended run for queued aggregate', $failures, $passes ); +smoke_assert( 1, count( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK ) ), 'late failed callback: queued aggregate remains authoritative', $failures, $passes ); +as_smoke_fire_aggregate_actions(); +$late_failure_resumes = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK ); AS_Shim::fire( $late_failure_resumes[0]['id'] ); $late_failure_final = $recorder_late_failure->find( 'as-late-failure' ); smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $late_failure_final->get_status(), 'late failed callback: queued resume reaches success', $failures, $passes ); @@ -1362,6 +1539,5 @@ static function ( bool $handled ) use ( &$legacy_failure_cleanups ): bool { smoke_assert( 1, $legacy_failure_cleanups, 'duplicate failure race: run-scoped cleanup executes once', $failures, $passes ); remove_all_filters( 'wp_agent_workflow_run_completed' ); remove_all_filters( 'wp_agent_workflow_branch_store_forget' ); - 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 515cf6b..7379420 100644 --- a/tests/workflow-async-branch-payload-smoke.php +++ b/tests/workflow-async-branch-payload-smoke.php @@ -510,6 +510,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-branch-concurrency-gate-smoke.php b/tests/workflow-branch-concurrency-gate-smoke.php index a33c332..59e2bbe 100644 --- a/tests/workflow-branch-concurrency-gate-smoke.php +++ b/tests/workflow-branch-concurrency-gate-smoke.php @@ -166,6 +166,7 @@ function smoke_assert( $expected, $actual, string $name, array &$failures, int & $branch_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK; $reconcile_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RECONCILE_HOOK; +$aggregate_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK; $resume_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK; $max = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::MAX_BRANCH_CONCURRENCY; @@ -345,5 +346,27 @@ function smoke_assert( $expected, $actual, string $name, array &$failures, int & smoke_assert( $max, WP_Agent_Workflow_Action_Scheduler_Branch_Executor::reconcile_inflight_count(), 'reconcile headroom count is bounded by MAX_BRANCH_CONCURRENCY', $failures, $passes ); smoke_assert( $max + 1, $concurrent_batches(), 'reconcile headroom raise remains bounded', $failures, $passes ); +// Durable aggregate actions are long-running effects: they need additive claim +// headroom and batch-size 1, but both counts remain bounded. +AS_Query_Shim::reset(); +AS_Query_Shim::add( 'unrelated_long_action', ActionScheduler_Store::STATUS_RUNNING, 1 ); +AS_Query_Shim::add( $aggregate_hook, ActionScheduler_Store::STATUS_PENDING, 1 ); +smoke_assert( 1, WP_Agent_Workflow_Action_Scheduler_Branch_Executor::aggregate_inflight_count(), 'aggregate starvation: pending continuation counts as in flight', $failures, $passes ); +smoke_assert( 2, $concurrent_batches(), 'aggregate starvation: ceiling adds one continuation slot', $failures, $passes ); +smoke_assert( 1, $batch_size(), 'aggregate continuation pins batch size while effects run', $failures, $passes ); + +AS_Query_Shim::reset(); +AS_Query_Shim::add( $aggregate_hook, ActionScheduler_Store::STATUS_PENDING, $max + 5 ); +AS_Query_Shim::add( $aggregate_hook, ActionScheduler_Store::STATUS_RUNNING, $max + 5 ); +smoke_assert( $max, WP_Agent_Workflow_Action_Scheduler_Branch_Executor::aggregate_inflight_count(), 'aggregate headroom count is bounded by MAX_BRANCH_CONCURRENCY', $failures, $passes ); +smoke_assert( $max + 1, $concurrent_batches(), 'aggregate headroom raise remains bounded', $failures, $passes ); + +AS_Query_Shim::reset(); +AS_Query_Shim::add( $branch_hook, ActionScheduler_Store::STATUS_RUNNING, 2 ); +AS_Query_Shim::add( $reconcile_hook, ActionScheduler_Store::STATUS_PENDING, 1 ); +AS_Query_Shim::add( $aggregate_hook, ActionScheduler_Store::STATUS_PENDING, 1 ); +AS_Query_Shim::add( $resume_hook, ActionScheduler_Store::STATUS_PENDING, 1 ); +smoke_assert( 5, $concurrent_batches(), 'combined headroom: branch, reconcile, aggregate, and resume slots are additive', $failures, $passes ); + echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; exit( count( $failures ) > 0 ? 1 : 0 ); diff --git a/tests/workflow-reconcile-race-smoke.php b/tests/workflow-reconcile-race-smoke.php index c36ce5c..7d7c5c5 100644 --- a/tests/workflow-reconcile-race-smoke.php +++ b/tests/workflow-reconcile-race-smoke.php @@ -227,19 +227,29 @@ 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; } 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 ] ) @@ -260,11 +270,30 @@ 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 ); 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 +309,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'] ?? '' ) . ']' ); } @@ -400,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 @@ -409,6 +460,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 +473,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 +511,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 ); diff --git a/tests/workflow-request-controller-smoke.php b/tests/workflow-request-controller-smoke.php index 7476702..7965b8f 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( 3 === count( $one_group_cleanup ), 'terminal cleanup removes only the run-scoped branch, reconcile, and resume actions' ); +controller_assert( 4 === count( $one_group_cleanup ), 'terminal cleanup removes only the run-scoped branch, reconcile, 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 4102ff3..f3b7f42 100644 --- a/tests/workflow-scoped-drain-smoke.php +++ b/tests/workflow-scoped-drain-smoke.php @@ -316,15 +316,16 @@ function as_get_datetime_object( ?string $date_string = null, string $timezone = $branch_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK; $reconcile_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RECONCILE_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, $reconcile_hook, $resume_hook ), + array( $branch_hook, $reconcile_hook, $aggregate_hook, $resume_hook ), WP_Agent_Workflow_Scoped_Drain::default_hooks(), - 'default_hooks() includes executor branch, reconcile, and resume hooks', + 'default_hooks() includes branch, reconcile, aggregate, and resume hooks', $failures, $passes );