fix: make workflow fanout admission atomic

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,34 @@ public function collect( array $handles ): array {
 	 *
 	 * @since 0.5.0
 	 *
-	 * @param array<mixed> $payload Action payload: { run_id, handle_id, store_ref, context_ref }.
+	 * @param array<mixed> $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;
 		}
 
+		// 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
 		// 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 +1093,70 @@ 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<int,array{id:int,payload:array<mixed>,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 );
+				}
+			}
+		}
+	}
+
+	/**
+	 * Remove pending copies of one exact branch payload.
+	 *
+	 * @param array<mixed> $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<mixed> $payload Branch payload.
+	 */
+	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' ) ) {
+			throw new \RuntimeException( sprintf( 'Could not defer branch for run `%s`: delayed Action Scheduler enqueue is unavailable.', $run_id ) );
+		}
+
+		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 );
+		}
+	}
+
 	/**
 	 * Derive a stable identity for one exact suspension generation.
 	 *

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,84 @@ 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 || ! function_exists( 'delete_option' ) ) {
+			return '';
+		}
+
+		$ref   = self::ADMISSION_PREFIX . md5( $run_id );
+		$token = $ref . ':' . md5( uniqid( $run_id . ':', true ) );
+		self::write_row(
+			$ref,
+			array(
+				'run_id'  => $run_id,
+				'token'   => $token,
+				'status'  => 'pending',
+				'expires' => time() + self::TTL_SECONDS,
+			)
+		);
+
+		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 {
+		$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( $ref, $row );
+		return 'admitted' === self::admission_status( $token );
+	}
+
+	/**
+	 * Close an admission fence before compensating a partial enqueue.
+	 */
+	public static function reject_admission( string $token ): void {
+		$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( $ref, $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 {
+		$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';
+	}
+
 	/**
 	 * 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
@@ -200,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;
 		}
@@ -219,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).
 	 *

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 {}
 }
 
 /**

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' ) );