Fix reconcile lock expiry during aggregation

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<mixed> $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<mixed> $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<mixed> $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.
 	 *

src/Workflows/class-wp-agent-workflow-reconcile-lock.php

@@ -1,10 +1,10 @@
 <?php
 /**
- * Cross-process serialization lock for the branch-reconcile critical section.
+ * Cross-process serialization lock for branch-reconcile state transitions.
  *
  * WHY THIS EXISTS. {@see agents_reconcile_workflow_branch()} merges one finished
  * branch into the suspended run's `metadata._suspension.completed[]` map, then
- * decides whether EVERY branch is now terminal (and, if so, aggregates + resumes).
+ * decides whether EVERY branch is now terminal (and, if so, claims aggregation).
  * That merge is a read-modify-write on shared per-run state:
  *
  *     $frame     = load();          // read completed[]
@@ -22,8 +22,8 @@
  *
  * AS's atomic action-claim already de-duplicates the RESUME action, but it does
  * NOT guard THIS write — a different write, on the frame, not the resume action.
- * This lock closes that gap by serializing the reconcile critical section per
- * run so each reconcile reads the frame AFTER the previous one committed.
+ * This lock closes that gap by serializing each short reconcile state transition
+ * per run so each reconcile reads the frame AFTER the previous one committed.
  *
  * TABLE-FREE. Under the substrate's no-new-tables constraint the lock uses
  * `add_option()` as the atomic compare-and-set — `add_option()` performs an
@@ -49,10 +49,10 @@
  * Default `add_option()`-CAS per-run reconcile lock.
  *
  * A held lock stores an expiry so a crashed holder's lock is reclaimable after
- * its TTL (a process that dies mid-critical-section must not strand every future
- * reconcile for the run). Acquisition blocks with bounded retries because a
- * reconcile critical section is short (an in-memory merge + a recorder write),
- * so a waiter reliably wins within a few spins rather than dropping the branch.
+ * its TTL (a process that dies mid-transition must not strand every future
+ * reconcile for the run). Unbounded aggregation and resume work must never run
+ * under this lock; the suspension frame carries the generation-bound ownership
+ * claim that fences those operations.
  */
 final class WP_Agent_Workflow_Reconcile_Lock {
 
@@ -66,9 +66,8 @@ final class WP_Agent_Workflow_Reconcile_Lock {
 
 	/**
 	 * Lock time-to-live (seconds). After this a stale lock (crashed holder) is
-	 * reclaimable. Generous relative to a reconcile's real duration so a healthy
-	 * holder is never evicted mid-section, yet short enough that a crash does not
-	 * strand the run for long.
+	 * reclaimable. Reconcile holds the lock only around recorder state transitions,
+	 * never while running an aggregator or resumed workflow steps.
 	 *
 	 * @since 0.5.0
 	 */

src/Workflows/class-wp-agent-workflow-scoped-drain.php

@@ -104,6 +104,7 @@ public static function default_hooks(): array {
 		return array(
 			WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK,
 			WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RECONCILE_HOOK,
+			WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK,
 			WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK,
 		);
 	}