From e96a6c560ab53c665ca6d1a16d0291eb82444a22 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 13:59:26 +0000 Subject: [PATCH 1/2] fix: make workflow fanout admission atomic --- ...kflow-action-scheduler-branch-executor.php | 107 ++++++++++++++++-- .../class-wp-agent-workflow-branch-store.php | 71 ++++++++++++ stubs/action-scheduler-classes.php | 3 + tests/workflow-async-branch-payload-smoke.php | 82 +++++++++++++- 4 files changed, 248 insertions(+), 15 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..314ceb0 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 @@ -154,19 +154,27 @@ public function id(): string { public function dispatch( array $branches, array $context ) { $context_run_id = self::string_value( $context['_workflow_run_id'] ?? '' ); $context_step_id = self::string_value( $context['_workflow_step_id'] ?? '' ); + $run_id_for_ctx = '' !== $context_run_id ? $context_run_id : self::first_branch_run_id( $branches ); + $admission_token = WP_Agent_Workflow_Branch_Store::begin_admission( $run_id_for_ctx ); + if ( '' === $admission_token ) { + return new \WP_Error( + 'workflow_branch_dispatch_admission_failed', + sprintf( 'Could not persist the async branch admission fence for run `%s`.', $run_id_for_ctx ) + ); + } // The shared immutable context is identical for every branch, so store it // ONCE per run and reference it from each branch — never duplicate a // multi-KB context into N branch payloads. Its ref rides in every branch's // AS args (a short option name), and the branch action re-seats it into the // descriptor's branch_vars.context on rehydrate. - $run_id_for_ctx = '' !== $context_run_id ? $context_run_id : self::first_branch_run_id( $branches ); $shared_context = is_array( $context['shared_context'] ?? null ) ? self::string_keyed_array( $context['shared_context'] ) : array(); $context_ref = '' !== $run_id_for_ctx ? WP_Agent_Workflow_Branch_Store::put_shared_context( $run_id_for_ctx, $shared_context ) : ''; - $handles = array(); + $handles = array(); + $enqueued_actions = array(); foreach ( $branches as $index => $branch ) { $key = self::string_value( $branch['key'] ?? (string) $index ); @@ -201,10 +209,11 @@ public function dispatch( array $branches, array $context ) { $store_ref = WP_Agent_Workflow_Branch_Store::put_branch( $run_id, $handle_id, $descriptor ); $payload = array( - 'run_id' => $run_id, - 'handle_id' => $handle_id, - 'store_ref' => $store_ref, - 'context_ref' => $context_ref, + 'run_id' => $run_id, + 'handle_id' => $handle_id, + 'store_ref' => $store_ref, + 'context_ref' => $context_ref, + 'admission_token' => $admission_token, ); $group = self::group_for_run( $run_id ); @@ -217,6 +226,8 @@ public function dispatch( array $branches, array $context ) { // dispatch instead — clean up what we already stored so no orphan rows // linger, and surface a descriptive WP_Error. if ( $action_id <= 0 ) { + WP_Agent_Workflow_Branch_Store::reject_admission( $admission_token ); + self::cancel_enqueued_actions( $enqueued_actions ); if ( '' !== $run_id ) { WP_Agent_Workflow_Branch_Store::forget_run( $run_id ); } @@ -230,6 +241,12 @@ public function dispatch( array $branches, array $context ) { ); } + $enqueued_actions[] = array( + 'id' => $action_id, + 'payload' => $payload, + 'group' => $group, + ); + $handles[] = array( 'id' => $handle_id, 'key' => $key, @@ -240,6 +257,16 @@ public function dispatch( array $branches, array $context ) { ); } + if ( ! WP_Agent_Workflow_Branch_Store::admit( $admission_token ) ) { + WP_Agent_Workflow_Branch_Store::reject_admission( $admission_token ); + self::cancel_enqueued_actions( $enqueued_actions ); + WP_Agent_Workflow_Branch_Store::forget_run( $run_id_for_ctx ); + return new \WP_Error( + 'workflow_branch_dispatch_admission_failed', + sprintf( 'Could not commit async branch admission for run `%s`.', $run_id_for_ctx ) + ); + } + // Every branch is now durably enqueued as a claimed AS action. Fire one // concurrent async-runner request per branch just enqueued so N distinct // worker processes each claim ONE branch in the same window — turning the @@ -736,19 +763,29 @@ public function collect( array $handles ): array { * * @since 0.5.0 * - * @param array $payload Action payload: { run_id, handle_id, store_ref, context_ref }. + * @param array $payload Action payload: { run_id, handle_id, store_ref, context_ref, admission_token }. * @return void */ public static function run_branch_action( array $payload ): void { - $run_id = self::string_value( $payload['run_id'] ?? '' ); - $handle_id = self::string_value( $payload['handle_id'] ?? '' ); - $store_ref = self::string_value( $payload['store_ref'] ?? '' ); - $context_ref = self::string_value( $payload['context_ref'] ?? '' ); + $run_id = self::string_value( $payload['run_id'] ?? '' ); + $handle_id = self::string_value( $payload['handle_id'] ?? '' ); + $store_ref = self::string_value( $payload['store_ref'] ?? '' ); + $context_ref = self::string_value( $payload['context_ref'] ?? '' ); + $admission_token = self::string_value( $payload['admission_token'] ?? '' ); if ( '' === $run_id || '' === $handle_id ) { return; } + $admission_status = WP_Agent_Workflow_Branch_Store::admission_status( $admission_token ); + if ( 'pending' === $admission_status ) { + self::defer_pending_branch( $payload, $run_id ); + return; + } + if ( 'admitted' !== $admission_status ) { + return; + } + // Rehydrate the full self-contained descriptor from the branch store using // the lightweight ref the AS args carried. The store re-seats the run-scoped // shared context into branch_vars.context, so the branch runs against the @@ -1051,6 +1088,54 @@ private static function enqueue_async_action( string $hook, array $args, string } } + /** + * Cancel every action inserted before a sibling enqueue failed. + * + * The rejected admission fence closes the running-worker race; exact-argument + * unscheduling removes pending originals and any callback deferred while the + * generation was still pending. + * + * @param array,group:string}> $actions Enqueued actions. + */ + private static function cancel_enqueued_actions( array $actions ): void { + foreach ( $actions as $action ) { + if ( function_exists( 'as_unschedule_all_actions' ) ) { + as_unschedule_all_actions( self::BRANCH_HOOK, array( $action['payload'] ), $action['group'] ); + } + + if ( class_exists( '\ActionScheduler_Store' ) ) { + try { + \ActionScheduler_Store::instance()->cancel_action( $action['id'] ); + } catch ( \Throwable $error ) { + unset( $error ); + } + } + } + } + + /** + * Requeue a branch claimed before its dispatch generation is committed. + * + * @param array $payload Branch payload. + */ + private static function defer_pending_branch( array $payload, string $run_id ): void { + $group = self::group_for_run( $run_id ); + if ( function_exists( 'as_schedule_single_action' ) ) { + try { + $action_id = (int) as_schedule_single_action( time() + 1, self::BRANCH_HOOK, array( $payload ), $group ); + if ( $action_id > 0 ) { + return; + } + } catch ( \Throwable $error ) { + unset( $error ); + } + } + + if ( self::enqueue_async_action( self::BRANCH_HOOK, array( $payload ), $group ) <= 0 ) { + throw new \RuntimeException( sprintf( 'Could not defer branch for run `%s` while fan-out admission was pending.', $run_id ) ); + } + } + /** * Derive a stable identity for one exact suspension generation. * diff --git a/src/Workflows/class-wp-agent-workflow-branch-store.php b/src/Workflows/class-wp-agent-workflow-branch-store.php index 60266fe..fb96b93 100644 --- a/src/Workflows/class-wp-agent-workflow-branch-store.php +++ b/src/Workflows/class-wp-agent-workflow-branch-store.php @@ -82,6 +82,11 @@ final class WP_Agent_Workflow_Branch_Store { */ private const INDEX_PREFIX = 'agents_wf_branch_index_'; + /** + * Option-name prefix for dispatch admission fences. + */ + private const ADMISSION_PREFIX = 'agents_wf_branch_admission_'; + /** * Payload time-to-live (seconds). After this a row belonging to a run that * never resolved is treated as expired and returns nothing on read. Generous @@ -92,6 +97,72 @@ final class WP_Agent_Workflow_Branch_Store { */ private const TTL_SECONDS = 7200; + /** + * Begin a fenced fan-out admission generation. + * + * Branch callbacks may be claimed while their siblings are still being + * enqueued, so they must not begin effects until this generation is admitted. + * + * @param string $run_id Run being admitted. + * @return string Opaque admission token, or an empty string on persistence failure. + */ + public static function begin_admission( string $run_id ): string { + if ( '' === $run_id ) { + return ''; + } + + $token = self::ADMISSION_PREFIX . md5( $run_id . ':' . uniqid( '', true ) ); + self::write_row( + $token, + array( + 'run_id' => $run_id, + 'status' => 'pending', + 'expires' => time() + self::TTL_SECONDS, + ) + ); + self::index_ref( $run_id, $token ); + + return 'pending' === self::admission_status( $token ) ? $token : ''; + } + + /** + * Open an admission fence after every sibling is durably enqueued. + */ + public static function admit( string $token ): bool { + $row = self::read_row( $token ); + if ( null === $row || 'pending' !== ( $row['status'] ?? '' ) ) { + return false; + } + + $row['status'] = 'admitted'; + self::write_row( $token, $row ); + return 'admitted' === self::admission_status( $token ); + } + + /** + * Close an admission fence before compensating a partial enqueue. + */ + public static function reject_admission( string $token ): void { + $row = self::read_row( $token ); + if ( null === $row ) { + return; + } + + $row['status'] = 'rejected'; + self::write_row( $token, $row ); + } + + /** + * Read an admission generation's state. A missing or expired fence is closed. + * + * @return string pending, admitted, or rejected. + */ + public static function admission_status( string $token ): string { + $row = '' !== $token ? self::read_row( $token ) : null; + $status = is_array( $row ) && is_string( $row['status'] ?? null ) ? $row['status'] : ''; + return in_array( $status, array( 'pending', 'admitted' ), true ) ? $status : 'rejected'; + } + /** * Persist one branch descriptor under a per-(run_id, handle_id) key and * return the opaque store ref the AS args carry. The descriptor stored here diff --git a/stubs/action-scheduler-classes.php b/stubs/action-scheduler-classes.php index 5f77756..efabe06 100644 --- a/stubs/action-scheduler-classes.php +++ b/stubs/action-scheduler-classes.php @@ -52,6 +52,9 @@ public function stake_claim( int $max_actions = 10, ?\DateTime $before_date = nu } public function release_claim( ActionScheduler_ActionClaim $claim ): void {} + + /** @param int|string $action_id Action id. */ + public function cancel_action( $action_id ): void {} } /** diff --git a/tests/workflow-async-branch-payload-smoke.php b/tests/workflow-async-branch-payload-smoke.php index a6dbf11..bec1485 100644 --- a/tests/workflow-async-branch-payload-smoke.php +++ b/tests/workflow-async-branch-payload-smoke.php @@ -169,15 +169,29 @@ final class AS_Limit_Shim { private static int $seq = 0; /** When true, every enqueue "fails" the way AS's runner swallows — throws. */ public static bool $force_fail = false; + public static int $fail_on_attempt = 0; + private static int $enqueue_attempt = 0; + /** @var array */ + public static array $cancelled = array(); public static function reset(): void { - self::$queue = array(); - self::$claimed = array(); - self::$seq = 0; - self::$force_fail = false; + self::$queue = array(); + self::$claimed = array(); + self::$seq = 0; + self::$force_fail = false; + self::$fail_on_attempt = 0; + self::$enqueue_attempt = 0; + self::$cancelled = array(); } public static function enqueue( string $hook, array $args, string $group ): int { + ++self::$enqueue_attempt; + if ( self::$fail_on_attempt === self::$enqueue_attempt ) { + if ( isset( self::$queue[0] ) ) { + self::fire( self::$queue[0]['id'] ); + } + throw new \RuntimeException( 'Later enqueue failed (simulated).' ); + } if ( self::$force_fail ) { throw new \RuntimeException( 'ActionScheduler queue is unavailable (simulated).' ); } @@ -223,6 +237,9 @@ static function ( array $action ) use ( $hook ): bool { } public static function fire( int $id ): bool { + if ( ! empty( self::$cancelled[ $id ] ) ) { + return false; + } if ( ! self::claim( $id ) ) { return false; } @@ -234,6 +251,25 @@ public static function fire( int $id ): bool { } return false; } + + public static function cancel( string $hook, array $args, string $group ): void { + foreach ( self::$queue as $action ) { + if ( $hook === $action['hook'] && $group === $action['group'] && $args === $action['args'] ) { + self::$cancelled[ $action['id'] ] = true; + } + } + } + + public static function runnable_count(): int { + return count( + array_filter( + self::$queue, + static function ( array $action ): bool { + return empty( self::$cancelled[ $action['id'] ] ) && empty( self::$claimed[ $action['id'] ] ); + } + ) + ); + } } function wp_json_encode_shim( $value ): string { @@ -246,6 +282,17 @@ function as_enqueue_async_action( string $hook, array $args = array(), string $g return AS_Limit_Shim::enqueue( $hook, $args, $group ); } } +if ( ! function_exists( 'as_schedule_single_action' ) ) { + function as_schedule_single_action( int $timestamp, string $hook, array $args = array(), string $group = '' ) { + unset( $timestamp ); + return AS_Limit_Shim::enqueue( $hook, $args, $group ); + } +} +if ( ! function_exists( 'as_unschedule_all_actions' ) ) { + function as_unschedule_all_actions( string $hook, ?array $args = array(), string $group = '' ): void { + AS_Limit_Shim::cancel( $hook, is_array( $args ) ? $args : array(), $group ); + } +} function smoke_assert( $expected, $actual, string $name, array &$failures, int &$passes ): void { if ( $expected === $actual ) { @@ -325,9 +372,11 @@ function payload_register_ability( string $name, \Closure $handler ): void { payload_register_ability( 'demo/role-worker', static function ( array $input ): array { + ++$GLOBALS['__branch_effects']; return array( 'fragment' => strtoupper( (string) ( $input['label'] ?? 'X' ) ) ); } ); +$GLOBALS['__branch_effects'] = 0; payload_register_ability( 'demo/aggregate', static function ( array $input ): array { @@ -502,5 +551,30 @@ function payload_roles_spec(): WP_Agent_Workflow_Spec { } smoke_assert( 0, $leftover2, 'bug2: failed dispatch cleaned up its stored rows (no orphans)', $failures, $passes ); +// A later enqueue failure fences a worker racing compensation, cancels every +// pending sibling/retry, and only then releases branch payload storage. +AS_Limit_Shim::reset(); +$GLOBALS['__options'] = array(); +$GLOBALS['__branch_effects'] = 0; +AS_Limit_Shim::$fail_on_attempt = 2; +$partial = $executor->dispatch( + array( + array( 'key' => 'first', 'run_id' => 'pay-partial', 'step_id' => 'scatter', 'required' => true, 'steps' => array( array( 'id' => 'first', 'type' => 'ability', 'ability' => 'demo/role-worker', 'args' => array( 'label' => 'first' ) ) ), 'branch_vars' => array( 'context' => array() ) ), + array( 'key' => 'second', 'run_id' => 'pay-partial', 'step_id' => 'scatter', 'required' => true, 'steps' => array( array( 'id' => 'second', 'type' => 'ability', 'ability' => 'demo/role-worker', 'args' => array( 'label' => 'second' ) ) ), 'branch_vars' => array( 'context' => array() ) ), + ), + array( '_workflow_run_id' => 'pay-partial', '_workflow_step_id' => 'scatter', 'shared_context' => array() ) +); +smoke_assert( 'workflow_branch_dispatch_enqueue_failed', is_wp_error( $partial ) ? $partial->get_error_code() : '', 'atomic admission: later enqueue failure fails the whole dispatch', $failures, $passes ); +smoke_assert( 0, $GLOBALS['__branch_effects'], 'atomic admission: a worker racing compensation begins no branch effects', $failures, $passes ); +smoke_assert( 0, AS_Limit_Shim::runnable_count(), 'atomic admission: no earlier sibling or deferred retry remains runnable', $failures, $passes ); + +$leftover3 = 0; +foreach ( array_keys( $GLOBALS['__options'] ) as $opt ) { + if ( str_starts_with( (string) $opt, 'agents_wf_branch_' ) ) { + ++$leftover3; + } +} +smoke_assert( 0, $leftover3, 'atomic admission: fenced compensation cleans branch-store rows', $failures, $passes ); + echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; exit( count( $failures ) > 0 ? 1 : 0 ); From 0f8875990387ce5ba1bc0d19c58dbf9720e27071 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 17:10:50 +0000 Subject: [PATCH 2/2] fix: preserve queued workflow branches on upgrade --- ...kflow-action-scheduler-branch-executor.php | 57 +++++++++++------ .../class-wp-agent-workflow-branch-store.php | 52 +++++++++++---- tests/workflow-as-branch-smoke.php | 21 +++++++ tests/workflow-async-branch-payload-smoke.php | 63 +++++++++++++++---- 4 files changed, 151 insertions(+), 42 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 314ceb0..2118706 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 @@ -777,13 +777,18 @@ public static function run_branch_action( array $payload ): void { return; } - $admission_status = WP_Agent_Workflow_Branch_Store::admission_status( $admission_token ); - if ( 'pending' === $admission_status ) { - self::defer_pending_branch( $payload, $run_id ); - return; - } - if ( 'admitted' !== $admission_status ) { - return; + // Payloads queued before admission fencing shipped have no token and must + // retain their original execution behavior across an upgrade. + if ( array_key_exists( 'admission_token', $payload ) ) { + $admission_status = WP_Agent_Workflow_Branch_Store::admission_status( $admission_token ); + if ( 'pending' === $admission_status ) { + self::defer_pending_branch( $payload, $run_id, $admission_token ); + return; + } + if ( 'admitted' !== $admission_status ) { + self::cancel_matching_actions( $payload, $run_id ); + return; + } } // Rehydrate the full self-contained descriptor from the branch store using @@ -1113,27 +1118,43 @@ private static function cancel_enqueued_actions( array $actions ): void { } } + /** + * Remove pending copies of one exact branch payload. + * + * @param array $payload Branch payload. + */ + private static function cancel_matching_actions( array $payload, string $run_id ): void { + if ( function_exists( 'as_unschedule_all_actions' ) ) { + as_unschedule_all_actions( self::BRANCH_HOOK, array( $payload ), self::group_for_run( $run_id ) ); + } + } + /** * Requeue a branch claimed before its dispatch generation is committed. * * @param array $payload Branch payload. */ - private static function defer_pending_branch( array $payload, string $run_id ): void { + private static function defer_pending_branch( array $payload, string $run_id, string $admission_token ): void { $group = self::group_for_run( $run_id ); - if ( function_exists( 'as_schedule_single_action' ) ) { - try { - $action_id = (int) as_schedule_single_action( time() + 1, self::BRANCH_HOOK, array( $payload ), $group ); - if ( $action_id > 0 ) { - return; - } - } catch ( \Throwable $error ) { - unset( $error ); - } + if ( ! function_exists( 'as_schedule_single_action' ) ) { + throw new \RuntimeException( sprintf( 'Could not defer branch for run `%s`: delayed Action Scheduler enqueue is unavailable.', $run_id ) ); } - if ( self::enqueue_async_action( self::BRANCH_HOOK, array( $payload ), $group ) <= 0 ) { + try { + $action_id = (int) as_schedule_single_action( time() + 1, self::BRANCH_HOOK, array( $payload ), $group ); + } catch ( \Throwable $error ) { + unset( $error ); + $action_id = 0; + } + if ( $action_id <= 0 ) { throw new \RuntimeException( sprintf( 'Could not defer branch for run `%s` while fan-out admission was pending.', $run_id ) ); } + + // Compensation may have scanned while this worker was scheduling its copy. + // Re-check after enqueue so rejection closes that final insertion race. + if ( 'rejected' === WP_Agent_Workflow_Branch_Store::admission_status( $admission_token ) ) { + self::cancel_matching_actions( $payload, $run_id ); + } } /** diff --git a/src/Workflows/class-wp-agent-workflow-branch-store.php b/src/Workflows/class-wp-agent-workflow-branch-store.php index fb96b93..ba2c690 100644 --- a/src/Workflows/class-wp-agent-workflow-branch-store.php +++ b/src/Workflows/class-wp-agent-workflow-branch-store.php @@ -107,35 +107,42 @@ final class WP_Agent_Workflow_Branch_Store { * @return string Opaque admission token, or an empty string on persistence failure. */ public static function begin_admission( string $run_id ): string { - if ( '' === $run_id ) { + if ( '' === $run_id || ! function_exists( 'delete_option' ) ) { return ''; } - $token = self::ADMISSION_PREFIX . md5( $run_id . ':' . uniqid( '', true ) ); + $ref = self::ADMISSION_PREFIX . md5( $run_id ); + $token = $ref . ':' . md5( uniqid( $run_id . ':', true ) ); self::write_row( - $token, + $ref, array( 'run_id' => $run_id, + 'token' => $token, 'status' => 'pending', 'expires' => time() + self::TTL_SECONDS, ) ); - self::index_ref( $run_id, $token ); - return 'pending' === self::admission_status( $token ) ? $token : ''; + if ( 'pending' === self::admission_status( $token ) ) { + return $token; + } + + delete_option( $ref ); + return ''; } /** * Open an admission fence after every sibling is durably enqueued. */ public static function admit( string $token ): bool { - $row = self::read_row( $token ); - if ( null === $row || 'pending' !== ( $row['status'] ?? '' ) ) { + $ref = self::admission_ref( $token ); + $row = '' !== $ref ? self::read_row( $ref ) : null; + if ( null === $row || ! hash_equals( $token, is_string( $row['token'] ?? null ) ? $row['token'] : '' ) || 'pending' !== ( $row['status'] ?? '' ) ) { return false; } $row['status'] = 'admitted'; - self::write_row( $token, $row ); + self::write_row( $ref, $row ); return 'admitted' === self::admission_status( $token ); } @@ -143,13 +150,14 @@ public static function admit( string $token ): bool { * Close an admission fence before compensating a partial enqueue. */ public static function reject_admission( string $token ): void { - $row = self::read_row( $token ); - if ( null === $row ) { + $ref = self::admission_ref( $token ); + $row = '' !== $ref ? self::read_row( $ref ) : null; + if ( null === $row || ! hash_equals( $token, is_string( $row['token'] ?? null ) ? $row['token'] : '' ) ) { return; } $row['status'] = 'rejected'; - self::write_row( $token, $row ); + self::write_row( $ref, $row ); } /** @@ -158,8 +166,12 @@ public static function reject_admission( string $token ): void { * @return string pending, admitted, or rejected. */ public static function admission_status( string $token ): string { - $row = '' !== $token ? self::read_row( $token ) : null; - $status = is_array( $row ) && is_string( $row['status'] ?? null ) ? $row['status'] : ''; + $ref = self::admission_ref( $token ); + $row = '' !== $ref ? self::read_row( $ref ) : null; + if ( null === $row || ! hash_equals( $token, is_string( $row['token'] ?? null ) ? $row['token'] : '' ) ) { + return 'rejected'; + } + $status = is_string( $row['status'] ?? null ) ? $row['status'] : ''; return in_array( $status, array( 'pending', 'admitted' ), true ) ? $status : 'rejected'; } @@ -271,6 +283,11 @@ public static function get_branch( string $store_ref, string $context_ref ): ?ar * @return void */ public static function forget_run( string $run_id ): void { + // Admission has a deterministic per-run ref, so its cleanup never depends + // on the mutable branch index or a consumer-owned payload store. + if ( function_exists( 'delete_option' ) ) { + delete_option( self::ADMISSION_PREFIX . md5( $run_id ) ); + } if ( self::filtered_forget_run( $run_id ) ) { return; } @@ -290,6 +307,15 @@ public static function forget_run( string $run_id ): void { delete_option( self::CONTEXT_PREFIX . md5( $run_id ) ); } + /** + * Resolve the deterministic option ref embedded in an admission token. + */ + private static function admission_ref( string $token ): string { + $separator = strrpos( $token, ':' ); + $ref = false !== $separator ? substr( $token, 0, $separator ) : ''; + return str_starts_with( $ref, self::ADMISSION_PREFIX ) ? $ref : ''; + } + /** * Write one option row (no autoload — these are transient runtime payloads). * diff --git a/tests/workflow-as-branch-smoke.php b/tests/workflow-as-branch-smoke.php index cf07fb8..fc06cc9 100644 --- a/tests/workflow-as-branch-smoke.php +++ b/tests/workflow-as-branch-smoke.php @@ -411,6 +411,27 @@ function as_smoke_roles_spec(): WP_Agent_Workflow_Spec { $selected = apply_filters( 'wp_agent_workflow_step_executor', null, array( 'id' => 'scatter', 'type' => 'parallel' ), array() ); smoke_assert_true( $selected instanceof WP_Agent_Workflow_Action_Scheduler_Branch_Executor, 'selection: AS present → AS branch executor selected', $failures, $passes ); +// A runtime exposing async enqueue without delayed enqueue cannot safely retry +// a branch claimed while admission is pending. It must fail closed rather than +// amplify the queue with immediate self-requeues. +$pending_token = \AgentsAPI\AI\Workflows\WP_Agent_Workflow_Branch_Store::begin_admission( 'as-no-delay' ); +$pending_error = ''; +$pending_queue_before = count( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK ) ); +try { + WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_branch_action( + array( + 'run_id' => 'as-no-delay', + 'handle_id' => 'as-no-delay:scatter:first:0', + 'admission_token' => $pending_token, + ) + ); +} catch ( \RuntimeException $error ) { + $pending_error = $error->getMessage(); +} +smoke_assert_true( str_contains( $pending_error, 'delayed Action Scheduler enqueue is unavailable' ), 'pending admission: unavailable delayed enqueue fails closed without immediate queue amplification', $failures, $passes ); +smoke_assert( $pending_queue_before, count( AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK ) ), 'pending admission: unavailable delayed enqueue adds no immediate retry action', $failures, $passes ); +\AgentsAPI\AI\Workflows\WP_Agent_Workflow_Branch_Store::forget_run( 'as-no-delay' ); + $tables_before = $recorder->tables(); $run = ( new WP_Agent_Workflow_Runner( $recorder ) )->run( as_smoke_roles_spec(), array(), array( 'run_id' => 'as-A' ) ); diff --git a/tests/workflow-async-branch-payload-smoke.php b/tests/workflow-async-branch-payload-smoke.php index bec1485..e9a44a6 100644 --- a/tests/workflow-async-branch-payload-smoke.php +++ b/tests/workflow-async-branch-payload-smoke.php @@ -170,18 +170,20 @@ final class AS_Limit_Shim { /** When true, every enqueue "fails" the way AS's runner swallows — throws. */ public static bool $force_fail = false; public static int $fail_on_attempt = 0; + public static bool $reject_after_delayed_enqueue = false; private static int $enqueue_attempt = 0; /** @var array */ public static array $cancelled = array(); public static function reset(): void { - self::$queue = array(); - self::$claimed = array(); - self::$seq = 0; - self::$force_fail = false; - self::$fail_on_attempt = 0; - self::$enqueue_attempt = 0; - self::$cancelled = array(); + self::$queue = array(); + self::$claimed = array(); + self::$seq = 0; + self::$force_fail = false; + self::$fail_on_attempt = 0; + self::$reject_after_delayed_enqueue = false; + self::$enqueue_attempt = 0; + self::$cancelled = array(); } public static function enqueue( string $hook, array $args, string $group ): int { @@ -285,7 +287,12 @@ function as_enqueue_async_action( string $hook, array $args = array(), string $g if ( ! function_exists( 'as_schedule_single_action' ) ) { function as_schedule_single_action( int $timestamp, string $hook, array $args = array(), string $group = '' ) { unset( $timestamp ); - return AS_Limit_Shim::enqueue( $hook, $args, $group ); + $action_id = AS_Limit_Shim::enqueue( $hook, $args, $group ); + if ( AS_Limit_Shim::$reject_after_delayed_enqueue ) { + $payload = is_array( $args[0] ?? null ) ? $args[0] : array(); + \AgentsAPI\AI\Workflows\WP_Agent_Workflow_Branch_Store::reject_admission( (string) ( $payload['admission_token'] ?? '' ) ); + } + return $action_id; } } if ( ! function_exists( 'as_unschedule_all_actions' ) ) { @@ -551,12 +558,41 @@ function payload_roles_spec(): WP_Agent_Workflow_Spec { } smoke_assert( 0, $leftover2, 'bug2: failed dispatch cleaned up its stored rows (no orphans)', $failures, $passes ); +// A payload queued by the previous release has no admission token. It must +// retain its pre-upgrade execution behavior instead of being silently fenced. +AS_Limit_Shim::reset(); +$GLOBALS['__options'] = array(); +$GLOBALS['__branch_effects'] = 0; +$legacy_ref = WP_Agent_Workflow_Branch_Store::put_branch( + 'pay-legacy', + 'pay-legacy:scatter:first:0', + array( + 'run_id' => 'pay-legacy', + 'step_id' => 'scatter', + 'handle_id' => 'pay-legacy:scatter:first:0', + 'key' => 'first', + 'required' => true, + 'steps' => array( array( 'id' => 'first', 'type' => 'ability', 'ability' => 'demo/role-worker', 'args' => array( 'label' => 'legacy' ) ) ), + 'branch_vars' => array(), + ) +); +WP_Agent_Workflow_Action_Scheduler_Branch_Executor::run_branch_action( + array( + 'run_id' => 'pay-legacy', + 'handle_id' => 'pay-legacy:scatter:first:0', + 'store_ref' => $legacy_ref, + ) +); +smoke_assert( 1, $GLOBALS['__branch_effects'], 'upgrade compatibility: a legacy payload without admission_token still executes', $failures, $passes ); +WP_Agent_Workflow_Branch_Store::forget_run( 'pay-legacy' ); + // A later enqueue failure fences a worker racing compensation, cancels every // pending sibling/retry, and only then releases branch payload storage. AS_Limit_Shim::reset(); -$GLOBALS['__options'] = array(); -$GLOBALS['__branch_effects'] = 0; -AS_Limit_Shim::$fail_on_attempt = 2; +$GLOBALS['__options'] = array(); +$GLOBALS['__branch_effects'] = 0; +AS_Limit_Shim::$fail_on_attempt = 2; +AS_Limit_Shim::$reject_after_delayed_enqueue = true; $partial = $executor->dispatch( array( array( 'key' => 'first', 'run_id' => 'pay-partial', 'step_id' => 'scatter', 'required' => true, 'steps' => array( array( 'id' => 'first', 'type' => 'ability', 'ability' => 'demo/role-worker', 'args' => array( 'label' => 'first' ) ) ), 'branch_vars' => array( 'context' => array() ) ), @@ -576,5 +612,10 @@ function payload_roles_spec(): WP_Agent_Workflow_Spec { } smoke_assert( 0, $leftover3, 'atomic admission: fenced compensation cleans branch-store rows', $failures, $passes ); +$cleanup_token = WP_Agent_Workflow_Branch_Store::begin_admission( 'pay-cleanup' ); +smoke_assert( 'pending', WP_Agent_Workflow_Branch_Store::admission_status( $cleanup_token ), 'admission storage: persisted generation is readable before cleanup', $failures, $passes ); +WP_Agent_Workflow_Branch_Store::forget_run( 'pay-cleanup' ); +smoke_assert( 'rejected', WP_Agent_Workflow_Branch_Store::admission_status( $cleanup_token ), 'admission storage: deterministic cleanup does not depend on branch indexing', $failures, $passes ); + echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; exit( count( $failures ) > 0 ? 1 : 0 );