fix: reconcile recurring Action Scheduler actions safely

inc/Engine/AI/System/SystemAgentServiceProvider.php

@@ -59,7 +59,10 @@ public function __construct() {
 		$this->registerBuiltInSchedules();
 		$this->initializeRegistry();
 		$this->registerActionSchedulerHooks();
+		// Bootstrap immediately after AS initializes, then let AS's native daily
+		// recurring-ensure action repair schedules that are later interrupted.
 		add_action( 'action_scheduler_init', array( $this, 'manageRecurringTaskSchedules' ) );
+		add_action( 'action_scheduler_ensure_recurring_actions', array( $this, 'manageRecurringTaskSchedules' ) );
 		WakeBriefingTask::registerStalenessGuard();
 	}
 

inc/Engine/Tasks/RecurringScheduler.php

@@ -228,14 +228,17 @@ public static function ensureSchedule(
 				: 0;
 			$first_run_time = time() + $stagger_offset;
 		}
+		// Action Scheduler unique actions are keyed by hook and group, not args.
+		// Preserve independent schedules that share a hook but have distinct args.
+		$unique = empty( $args );
 
 		if ( empty( $options['force_reschedule'] ) && self::hasMatchingRecurringAction( $hook, $args, $group, (int) $interval_seconds ) ) {
 			// Self-healing dedup: if a previous call already created MORE THAN
 			// ONE matching pending chain for this signature, collapse them to a
 			// single chain now instead of preserving the duplicates.
 			if ( self::countMatchingPendingActions( $hook, $args, $group ) > 1 ) {
 				self::unschedule( $hook, $args, $group );
-				as_schedule_recurring_action( $first_run_time, $interval_seconds, $hook, $args, $group );
+				as_schedule_recurring_action( $first_run_time, $interval_seconds, $hook, $args, $group, $unique );
 
 				if ( ! self::isScheduled( $hook, $args, $group ) ) {
 					return self::error(
@@ -263,9 +266,13 @@ public static function ensureSchedule(
 			);
 		}
 
-		self::unschedule( $hook, $args, $group );
+		if ( self::getPendingAction( $hook, $args, $group ) ) {
+			self::unschedule( $hook, $args, $group );
+		}
 
-		as_schedule_recurring_action( $first_run_time, $interval_seconds, $hook, $args, $group );
+		// A reconciliation can race another request during activation, upgrades,
+		// or cron. Let Action Scheduler atomically preserve the first chain.
+		as_schedule_recurring_action( $first_run_time, $interval_seconds, $hook, $args, $group, $unique );
 
 		// Verify persistence. AS can silently drop actions when its tables
 		// aren't ready (e.g. CLI context during plugin activation).
@@ -671,12 +678,16 @@ private static function scheduleCron( string $hook, array $args, string $cron_ex
 			);
 		}
 
+		// Action Scheduler unique actions are keyed by hook and group, not args.
+		// Preserve independent schedules that share a hook but have distinct args.
+		$unique = empty( $args );
+
 		if ( ! $force_reschedule && self::hasMatchingCronAction( $hook, $args, $group, $cron_expression ) ) {
 			// Self-healing dedup: collapse an already-duplicated signature to a
 			// single chain instead of preserving the duplicates.
 			if ( self::countMatchingPendingActions( $hook, $args, $group ) > 1 ) {
 				self::unschedule( $hook, $args, $group );
-				$action_id = as_schedule_cron_action( time(), $cron_expression, $hook, $args, $group );
+				$action_id = as_schedule_cron_action( time(), $cron_expression, $hook, $args, $group, $unique );
 
 				if ( ! self::isScheduled( $hook, $args, $group ) ) {
 					return self::error(
@@ -703,9 +714,11 @@ private static function scheduleCron( string $hook, array $args, string $cron_ex
 			);
 		}
 
-		self::unschedule( $hook, $args, $group );
+		if ( self::getPendingAction( $hook, $args, $group ) ) {
+			self::unschedule( $hook, $args, $group );
+		}
 
-		$action_id = as_schedule_cron_action( time(), $cron_expression, $hook, $args, $group );
+		$action_id = as_schedule_cron_action( time(), $cron_expression, $hook, $args, $group, $unique );
 
 		if ( ! self::isScheduled( $hook, $args, $group ) ) {
 			return self::error(

tests/abilities-image-template-load-order-smoke.php

@@ -7,9 +7,9 @@
  *
  * Two kinds of assertions:
  *
- * 1. Source-string assertions — `data-machine.php` must call
- *    `ImageTemplateAbilities::ensure_registered()` UNCONDITIONALLY at
- *    file include time, NOT only inside the gated runtime function.
+ * 1. Source-string assertions — `data-machine.php` must declare
+ *    `ImageTemplateAbilities::ensure_registered()` in the lightweight
+ *    ability manifest registered at file include time.
  *
  * 2. Behavioral assertions — `ImageTemplateAbilities::ensure_registered()`
  *    must handle all three timing states defensively, matching the
@@ -50,22 +50,20 @@
 }
 
 $assert(
-	'data-machine.php calls ImageTemplateAbilities::ensure_registered() unconditionally at file load',
-	str_contains( $bootstrap, 'Register `datamachine/render-image-template` and' )
-		&& str_contains( $bootstrap, "require_once __DIR__ . '/inc/Abilities/Media/ImageTemplateAbilities.php';\n\\DataMachine\\Abilities\\Media\\ImageTemplateAbilities::ensure_registered();" )
+	'data-machine.php declares ImageTemplateAbilities::ensure_registered() in the lightweight manifest',
+	str_contains( $bootstrap, "'file'   => __DIR__ . '/inc/Abilities/Media/ImageTemplateAbilities.php'," )
+		&& str_contains( $bootstrap, "'class'  => \\DataMachine\\Abilities\\Media\\ImageTemplateAbilities::class," )
+		&& str_contains( $bootstrap, "'method' => 'ensure_registered'," )
 );
 
 $assert(
-	'unconditional call site is OUTSIDE datamachine_run_datamachine_plugin()',
-	// The image-template registration must sit at file scope, after the
-	// unconditional `AbilityCategories::ensure_registered()` call (which other
-	// unconditional ability registrations — e.g. AgentAbilities — may follow).
+	'lightweight manifest registration is unconditional at file load',
 	(bool) preg_match(
 		'/^\\\\DataMachine\\\\Abilities\\\\AbilityCategories::ensure_registered\(\);\s*\n/m',
 		$bootstrap
 	)
 		&& (bool) preg_match(
-			'/^\/\*\*\s*\n\s*\*\s*Register `datamachine\/render-image-template`/m',
+			'/^\\\\DataMachine\\\\Abilities\\\\AbilityManifest::register\( datamachine_lightweight_ability_manifest\(\) \);\s*$/m',
 			$bootstrap
 		)
 );
@@ -180,7 +178,9 @@ function wp_register_ability( $name, $args ) {
 	}
 }
 
-// Stub PermissionHelper which the ability definitions reference.
+// Load or stub every collaborator used while building ability definitions.
+require_once $plugin_root . '/inc/Abilities/AbilityRegistration.php';
+
 if ( ! class_exists( 'DataMachine\\Abilities\\PermissionHelper' ) ) {
 	eval(
 		'namespace DataMachine\\Abilities;
@@ -190,6 +190,16 @@ public static function can_manage(): bool { return true; }
 	);
 }
 
+$assert(
+	'bootstrap loads AbilityRegistration used to build registration definitions',
+	class_exists( 'DataMachine\\Abilities\\AbilityRegistration', false )
+);
+
+$assert(
+	'bootstrap loads PermissionHelper referenced by registration definitions',
+	class_exists( 'DataMachine\\Abilities\\PermissionHelper', false )
+);
+
 require_once $plugin_root . '/inc/Abilities/Media/ImageTemplateAbilities.php';
 
 $reset = static function (): void {
@@ -214,6 +224,8 @@ public static function can_manage(): bool { return true; }
 	'state 1: when doing_action fires, abilities register immediately via wp_register_ability()',
 	isset( $GLOBALS['datamachine_2290_state']->registered['datamachine/render-image-template'] )
 		&& isset( $GLOBALS['datamachine_2290_state']->registered['datamachine/list-image-templates'] )
+		&& $GLOBALS['datamachine_2290_state']->registered['datamachine/render-image-template']['execute_callback'] instanceof Closure
+		&& $GLOBALS['datamachine_2290_state']->registered['datamachine/list-image-templates']['execute_callback'] instanceof Closure
 		&& empty( $GLOBALS['datamachine_2290_state']->hooked )
 );
 

tests/recurring-scheduler-idempotency-smoke.php

@@ -147,6 +147,9 @@ public function get_schedule(): DmRecurringSchedulerFakeSchedule {
 $GLOBALS['datamachine_rs_scheduled_recurring'] = 0;
 $GLOBALS['datamachine_rs_scheduled_cron']      = 0;
 $GLOBALS['datamachine_rs_unscheduled']         = 0;
+$GLOBALS['datamachine_rs_recurring_unique']    = array();
+$GLOBALS['datamachine_rs_cron_unique']         = array();
+$GLOBALS['datamachine_rs_schedule_race']       = false;
 
 function datamachine_rs_key( string $hook, array $args, string $group ): string {
 	return $group . '|' . $hook . '|' . serialize( $args );
@@ -177,6 +180,9 @@ function datamachine_rs_reset(): void {
 	$GLOBALS['datamachine_rs_scheduled_recurring'] = 0;
 	$GLOBALS['datamachine_rs_scheduled_cron']      = 0;
 	$GLOBALS['datamachine_rs_unscheduled']         = 0;
+	$GLOBALS['datamachine_rs_recurring_unique']    = array();
+	$GLOBALS['datamachine_rs_cron_unique']         = array();
+	$GLOBALS['datamachine_rs_schedule_race']       = false;
 	ActionScheduler::$initialized         = true;
 }
 
@@ -220,14 +226,24 @@ function as_schedule_single_action( int $timestamp, string $hook, array $args =
 	return 101;
 }
 
-function as_schedule_recurring_action( int $timestamp, int $interval, string $hook, array $args = array(), string $group = '' ): int {
+function as_schedule_recurring_action( int $timestamp, int $interval, string $hook, array $args = array(), string $group = '', bool $unique = false ): int {
 	++$GLOBALS['datamachine_rs_scheduled_recurring'];
+	$GLOBALS['datamachine_rs_recurring_unique'][] = $unique;
+	if ( $GLOBALS['datamachine_rs_schedule_race'] ) {
+		$GLOBALS['datamachine_rs_schedule_race'] = false;
+		datamachine_rs_seed_action( $hook, $args, $group, new DmRecurringSchedulerFakeSchedule( true, $interval, $timestamp ) );
+		return 0;
+	}
+	if ( $unique && datamachine_rs_pending_count( $hook, $args, $group ) > 0 ) {
+		return 0;
+	}
 	datamachine_rs_seed_action( $hook, $args, $group, new DmRecurringSchedulerFakeSchedule( true, $interval, $timestamp ) );
 	return 202;
 }
 
-function as_schedule_cron_action( int $timestamp, string $expression, string $hook, array $args = array(), string $group = '' ): int {
+function as_schedule_cron_action( int $timestamp, string $expression, string $hook, array $args = array(), string $group = '', bool $unique = false ): int {
 	++$GLOBALS['datamachine_rs_scheduled_cron'];
+	$GLOBALS['datamachine_rs_cron_unique'][] = $unique;
 	datamachine_rs_seed_action( $hook, $args, $group, new DmRecurringSchedulerFakeSchedule( true, $expression, $timestamp ) );
 	return 303;
 }
@@ -371,4 +387,32 @@ function datamachine_assert_schedule_result( $result ): array {
 datamachine_assert( 1 === datamachine_rs_counter( 'datamachine_rs_scheduled_cron' ), 'exactly one cron chain recreated' );
 datamachine_assert( 1 === datamachine_rs_pending_count( 'datamachine_cron', array(), RecurringScheduler::GROUP ), 'cron collapsed to exactly one pending action' );
 
+echo "\n[14] concurrent recurring reconciliation creates exactly one unique chain (#2892)\n";
+datamachine_rs_reset();
+$GLOBALS['datamachine_rs_schedule_race'] = true;
+$result                                  = datamachine_assert_schedule_result( RecurringScheduler::ensureSchedule( 'datamachine_recurring_retention_as_actions', array(), 'hourly' ) );
+datamachine_assert( 1 === datamachine_rs_pending_count( 'datamachine_recurring_retention_as_actions', array(), RecurringScheduler::GROUP ), 'concurrent reconciliation preserves exactly one pending chain' );
+datamachine_assert( array( true ) === $GLOBALS['datamachine_rs_recurring_unique'], 'recurring reconciliation requests Action Scheduler uniqueness' );
+datamachine_assert( 0 === datamachine_rs_counter( 'datamachine_rs_unscheduled' ), 'initial reconciliation does not cancel a concurrent healthy chain' );
+
+echo "\n[15] all recurring and cron replacement paths request Action Scheduler uniqueness (#2892)\n";
+datamachine_rs_reset();
+datamachine_assert_schedule_result( RecurringScheduler::ensureSchedule( 'datamachine_recurring_retention_as_actions', array(), 'hourly' ) );
+datamachine_assert( array( true ) === $GLOBALS['datamachine_rs_recurring_unique'], 'missing enabled recurrence is restored as a unique action' );
+datamachine_rs_reset();
+datamachine_assert_schedule_result( RecurringScheduler::ensureSchedule( 'datamachine_cron', array(), 'cron', array( 'cron_expression' => '0 0 * * *' ) ) );
+datamachine_assert( array( true ) === $GLOBALS['datamachine_rs_cron_unique'], 'cron replacement is also unique' );
+
+echo "\n[16] Action Scheduler native recurring-ensure hook repairs interrupted schedules (#2892)\n";
+$provider_source = file_get_contents( __DIR__ . '/../inc/Engine/AI/System/SystemAgentServiceProvider.php' ) ?: '';
+datamachine_assert( str_contains( $provider_source, "add_action( 'action_scheduler_ensure_recurring_actions', array( \$this, 'manageRecurringTaskSchedules' ) );" ), 'provider registers reconciliation on Action Scheduler native recurring-ensure hook' );
+
+echo "\n[17] distinct argument tuples sharing a hook remain independently schedulable\n";
+datamachine_rs_reset();
+datamachine_assert_schedule_result( RecurringScheduler::ensureSchedule( 'datamachine_per_flow', array( 'flow_id' => 1 ), 'hourly' ) );
+datamachine_assert_schedule_result( RecurringScheduler::ensureSchedule( 'datamachine_per_flow', array( 'flow_id' => 2 ), 'hourly' ) );
+datamachine_assert( array( false, false ) === $GLOBALS['datamachine_rs_recurring_unique'], 'argument-bearing schedules do not use hook/group-only Action Scheduler uniqueness' );
+datamachine_assert( 1 === datamachine_rs_pending_count( 'datamachine_per_flow', array( 'flow_id' => 1 ), RecurringScheduler::GROUP ), 'first argument tuple has its own pending chain' );
+datamachine_assert( 1 === datamachine_rs_pending_count( 'datamachine_per_flow', array( 'flow_id' => 2 ), RecurringScheduler::GROUP ), 'second argument tuple has its own pending chain' );
+
 echo "\nAll recurring scheduler idempotency assertions passed.\n";

tests/retention-action-scheduler-batching-smoke.php

@@ -332,6 +332,47 @@ private function matching_logs( string $cutoff, ?string $hook ): array {
 		);
 		++$aid;
 	}
+	$expired_recurring_action_id = $aid;
+	$expired_recurring_log_id    = $lid;
+	$fake_wpdb->actions[ $aid ]  = array(
+		'action_id'        => $aid,
+		'hook'             => 'datamachine_recurring_retention_as_actions',
+		'status'           => 'canceled',
+		'last_attempt_gmt' => $eight_day,
+	);
+	$fake_wpdb->logs[ $lid ] = array(
+		'log_id'    => $lid,
+		'action_id' => $aid,
+	);
+	++$aid;
+	++$lid;
+
+	$recent_recurring_action_id = $aid;
+	$recent_recurring_log_id    = $lid;
+	$fake_wpdb->actions[ $aid ] = array(
+		'action_id'        => $aid,
+		'hook'             => 'datamachine_recurring_retention_as_actions',
+		'status'           => 'canceled',
+		'last_attempt_gmt' => $fresh,
+	);
+	$fake_wpdb->logs[ $lid ] = array(
+		'log_id'    => $lid,
+		'action_id' => $aid,
+	);
+	++$aid;
+	++$lid;
+
+	$non_datamachine_log_id    = $lid;
+	$fake_wpdb->actions[ $aid ] = array(
+		'action_id'        => $aid,
+		'hook'             => 'woocommerce_background_task',
+		'status'           => 'pending',
+		'last_attempt_gmt' => $eight_day,
+	);
+	$fake_wpdb->logs[ $lid ] = array(
+		'log_id'    => $lid,
+		'action_id' => $aid,
+	);
 
 	$GLOBALS['wpdb'] = $fake_wpdb;
 
@@ -347,8 +388,8 @@ private function matching_logs( string $cutoff, ?string $hook ): array {
 			&& 0 === $fake_wpdb->delete_queries
 	);
 	assert_batching(
-		'count covers per-hook + global eligible rows (2500 logs + 2530 actions)',
-		5030 === $count_total,
+		'count includes expired logs attached to canceled Data Machine recurrences (2501 logs + 2531 actions)',
+		5032 === $count_total,
 		"got {$count_total}"
 	);
 
@@ -402,13 +443,24 @@ private function matching_logs( string $cutoff, ?string $hook ): array {
 	);
 	assert_batching(
 		'result reports per-table deletion counts + batch metadata',
-		2530 === $result['actions_deleted']
-			&& 2500 === $result['logs_deleted']
+		2531 === $result['actions_deleted']
+			&& 2501 === $result['logs_deleted']
 			&& 1000 === $result['batch_size']
 			&& isset( $result['iterations'] )
 			&& false === $result['hit_limit'],
 		"actions={$result['actions_deleted']} logs={$result['logs_deleted']}"
 	);
+	assert_batching(
+		'expired canceled Data Machine recurrence and its log are deleted',
+		! isset( $fake_wpdb->actions[ $expired_recurring_action_id ] )
+			&& ! isset( $fake_wpdb->logs[ $expired_recurring_log_id ] )
+	);
+	assert_batching(
+		'recent canceled and non-Data-Machine logs are preserved',
+		isset( $fake_wpdb->actions[ $recent_recurring_action_id ] )
+			&& isset( $fake_wpdb->logs[ $recent_recurring_log_id ] )
+			&& isset( $fake_wpdb->logs[ $non_datamachine_log_id ] )
+	);
 
 	// OPTIMIZE is opt-in: default off => no OPTIMIZE call.
 	assert_batching(