Make generic run-control mutations atomic (workflow cancel/finish race)

composer.json

@@ -117,6 +117,7 @@
       "php tests/conversation-loop-budgets-smoke.php",
       "php tests/runtime-package-run-contract-smoke.php",
       "php tests/run-control-normalization-smoke.php",
+      "php tests/run-control-atomic-mutations-smoke.php",
       "php tests/run-control-option-store-failures-smoke.php",
       "php tests/canonical-run-lifecycle-smoke.php",
       "php tests/channels-smoke.php",

src/Runtime/class-wp-agent-run-control.php

@@ -23,6 +23,7 @@ class WP_Agent_Run_Control {
 	public const STATUS_COMPLETED            = 'completed';
 	public const STATUS_SUCCEEDED            = 'succeeded';
 	public const STATUS_FAILED               = 'failed';
+	public const STATUS_SKIPPED              = 'skipped';
 	public const STATUS_RUNTIME_TOOL_PENDING = 'runtime_tool_pending';
 	public const STATUS_APPROVAL_REQUIRED    = 'approval_required';
 	public const STATUS_BUDGET_EXCEEDED      = 'budget_exceeded';
@@ -41,6 +42,7 @@ public static function statuses(): array {
 			self::STATUS_COMPLETED,
 			self::STATUS_SUCCEEDED,
 			self::STATUS_FAILED,
+			self::STATUS_SKIPPED,
 			self::STATUS_RUNTIME_TOOL_PENDING,
 			self::STATUS_APPROVAL_REQUIRED,
 			self::STATUS_BUDGET_EXCEEDED,
@@ -109,7 +111,7 @@ public static function normalize_run( array $run ): array {
 			$normalized['cancelled'] = (bool) $run['cancelled'];
 		}
 
-		return $normalized;
+		return self::normalize_cancellation_state( $normalized );
 	}
 
 	/**
@@ -229,7 +231,7 @@ public static function redacted_observer_payload( array $payload ): array {
 	}
 
 	/**
-	 * Start or update an addressable run in the selected store.
+	 * Start an addressable run in the selected store.
 	 *
 	 * @param string              $store_key Option key used by the backing store.
 	 * @param string              $run_id    Run ID.
@@ -238,7 +240,7 @@ public static function redacted_observer_payload( array $payload ): array {
 	 */
 	public static function start_run( string $store_key, string $run_id, array $run = array(), ?WP_Agent_Workspace_Scope $workspace = null ): array {
 		$now = self::now();
-		$run = array_merge(
+		$run = self::normalize_cancellation_state( array_merge(
 			$run,
 			array(
 				'run_id'     => $run_id,
@@ -247,34 +249,70 @@ public static function start_run( string $store_key, string $run_id, array $run
 				'updated_at' => $now,
 				'metadata'   => isset( $run['metadata'] ) && is_array( $run['metadata'] ) ? $run['metadata'] : array(),
 			)
-		);
+		) );
 
-		$state                    = self::state( $store_key, $workspace );
-		$state['runs'][ $run_id ] = $run;
-		$state                    = self::record_event_in_state( $state, $run_id, 'run_started', array( 'status' => self::STATUS_RUNNING ) );
-		self::save_state( $store_key, $state, $workspace );
+		$result = self::mutate_run_state(
+			$store_key,
+			static function ( array $state ) use ( $run_id, $run ): array {
+				$current = $state['runs'][ $run_id ] ?? null;
+				if ( is_array( $current ) ) {
+					$current                    = self::normalize_cancellation_state( $current );
+					$state['runs'][ $run_id ] = $current;
+				}
+				if ( is_array( $current ) ) {
+					return array( 'state' => $state, 'result' => $current );
+				}
+
+				$state['runs'][ $run_id ] = $run;
+				$state                    = self::record_event_in_state( $state, $run_id, 'run_started', array( 'status' => self::STATUS_RUNNING ) );
+				return array( 'state' => $state, 'result' => $run );
+			},
+			$workspace
+		);
 
-		return self::normalize_run( $run );
+		return is_array( $result ) ? self::normalize_run( self::string_keyed_array( $result ) ) : self::normalize_run( $run );
 	}
 
 	/**
 	 * Store a normalized run result.
 	 *
 	 * @param string              $store_key Option key used by the backing store.
 	 * @param array<string,mixed> $run       Run payload.
+	 * @param WP_Agent_Workspace_Scope|null $workspace Explicit workspace scope.
 	 * @return array<string,mixed>
 	 */
-	public static function save_run( string $store_key, array $run ): array {
+	public static function save_run( string $store_key, array $run, ?WP_Agent_Workspace_Scope $workspace = null ): array {
 		$normalized               = self::normalize_run( $run );
 		$normalized['updated_at'] = '' !== $normalized['updated_at'] ? $normalized['updated_at'] : self::now();
+		$normalized               = self::normalize_cancellation_state( $normalized );
 		$run_id                   = self::string_value( $normalized['run_id'] );
 
-		$state                    = self::state( $store_key );
-		$state['runs'][ $run_id ] = $normalized;
-		$state                    = self::record_event_in_state( $state, $run_id, 'run_updated', array( 'status' => $normalized['status'] ) );
-		self::save_state( $store_key, $state );
+		$result = self::mutate_run_state(
+			$store_key,
+			static function ( array $state ) use ( $run_id, $normalized ): array {
+				$current = $state['runs'][ $run_id ] ?? null;
+				if ( is_array( $current ) ) {
+					$current                    = self::normalize_cancellation_state( $current );
+					$state['runs'][ $run_id ] = $current;
+				}
+				if ( is_array( $current ) && self::is_terminal_status( $current['status'] ?? null ) ) {
+					return array( 'state' => $state, 'result' => $current );
+				}
+
+				$next = $normalized;
+				if ( is_array( $current ) && self::is_cancellation_requested( $current ) ) {
+					$next['status']    = self::is_terminal_status( $normalized['status'] ?? null ) ? self::STATUS_CANCELLED : self::STATUS_CANCELLING;
+					$next['cancelled'] = true;
+				}
+
+				$state['runs'][ $run_id ] = $next;
+				$state                    = self::record_event_in_state( $state, $run_id, 'run_updated', array( 'status' => $next['status'] ) );
+				return array( 'state' => $state, 'result' => $next );
+			},
+			$workspace
+		);
 
-		return $normalized;
+		return is_array( $result ) ? self::normalize_run( self::string_keyed_array( $result ) ) : $normalized;
 	}
 
 	/**
@@ -286,23 +324,31 @@ public static function save_run( string $store_key, array $run ): array {
 	 * @return array<string,mixed>|null
 	 */
 	public static function finish_run( string $store_key, string $run_id, string $status = self::STATUS_COMPLETED, ?WP_Agent_Workspace_Scope $workspace = null ): ?array {
-		$state = self::state( $store_key, $workspace );
-		if ( ! isset( $state['runs'][ $run_id ] ) ) {
-			return null;
-		}
-
-		$run               = $state['runs'][ $run_id ];
-		$run['status']     = self::normalize_status( $status );
-		$run['updated_at'] = self::now();
-		if ( self::STATUS_CANCELLED === $run['status'] ) {
-			$run['cancelled'] = true;
-		}
-
-		$state['runs'][ $run_id ] = $run;
-		$state                    = self::record_event_in_state( $state, $run_id, 'run_finished', array( 'status' => $run['status'] ) );
-		self::save_state( $store_key, $state, $workspace );
+		$result = self::mutate_run_state(
+			$store_key,
+			static function ( array $state ) use ( $run_id, $status ): array {
+				if ( ! isset( $state['runs'][ $run_id ] ) ) {
+					return array( 'state' => $state, 'result' => null );
+				}
+
+				$run                      = self::normalize_cancellation_state( $state['runs'][ $run_id ] );
+				$state['runs'][ $run_id ] = $run;
+				if ( self::is_terminal_status( $run['status'] ?? null ) ) {
+					return array( 'state' => $state, 'result' => $run );
+				}
+
+				$run['status']     = self::is_cancellation_requested( $run ) ? self::STATUS_CANCELLED : self::normalize_status( $status );
+				$run['updated_at'] = self::now();
+				$run               = self::normalize_cancellation_state( $run );
+
+				$state['runs'][ $run_id ] = $run;
+				$state                    = self::record_event_in_state( $state, $run_id, 'run_finished', array( 'status' => $run['status'] ) );
+				return array( 'state' => $state, 'result' => $run );
+			},
+			$workspace
+		);
 
-		return self::normalize_run( $run );
+		return is_array( $result ) ? self::normalize_run( self::string_keyed_array( $result ) ) : null;
 	}
 
 	/**
@@ -322,22 +368,31 @@ public static function get_run( string $store_key, string $run_id, ?WP_Agent_Wor
 	 * @return array<string,mixed>|null
 	 */
 	public static function request_cancel( string $store_key, string $run_id, ?WP_Agent_Workspace_Scope $workspace = null ): ?array {
-		$state = self::state( $store_key, $workspace );
-		if ( ! isset( $state['runs'][ $run_id ] ) ) {
-			return null;
-		}
-
-		$run               = $state['runs'][ $run_id ];
-		$terminal          = in_array( self::normalize_status( $run['status'] ?? '' ), array( self::STATUS_COMPLETED, self::STATUS_SUCCEEDED, self::STATUS_FAILED, self::STATUS_CANCELLED, self::STATUS_BUDGET_EXCEEDED, self::STATUS_STALLED, self::STATUS_INTERRUPTED ), true );
-		$run['status']     = $terminal ? self::normalize_status( $run['status'] ?? '' ) : self::STATUS_CANCELLING;
-		$run['cancelled']  = ! $terminal;
-		$run['updated_at'] = self::now();
-
-		$state['runs'][ $run_id ] = $run;
-		$state                    = self::record_event_in_state( $state, $run_id, 'cancel_requested', array( 'status' => $run['status'] ) );
-		self::save_state( $store_key, $state, $workspace );
+		$result = self::mutate_run_state(
+			$store_key,
+			static function ( array $state ) use ( $run_id ): array {
+				if ( ! isset( $state['runs'][ $run_id ] ) ) {
+					return array( 'state' => $state, 'result' => null );
+				}
+
+				$run                      = self::normalize_cancellation_state( $state['runs'][ $run_id ] );
+				$state['runs'][ $run_id ] = $run;
+				if ( self::is_terminal_status( $run['status'] ?? null ) ) {
+					return array( 'state' => $state, 'result' => $run );
+				}
+
+				$run['status']     = self::STATUS_CANCELLING;
+				$run['cancelled']  = true;
+				$run['updated_at'] = self::now();
+
+				$state['runs'][ $run_id ] = $run;
+				$state                    = self::record_event_in_state( $state, $run_id, 'cancel_requested', array( 'status' => $run['status'] ) );
+				return array( 'state' => $state, 'result' => $run );
+			},
+			$workspace
+		);
 
-		return self::normalize_run( $run );
+		return is_array( $result ) ? self::normalize_run( self::string_keyed_array( $result ) ) : null;
 	}
 
 	public static function cancel_requested( string $store_key, string $run_id ): bool {
@@ -441,6 +496,72 @@ public static function mutate_state( string $store_key, callable $mutation, ?WP_
 		return $store->mutate_workspace_state( $store_key, $workspace, $mutation );
 	}
 
+	/**
+	 * Serialize a lifecycle read-modify-write through the registered store.
+	 *
+	 * Routes generic lifecycle mutations through the store's atomic
+	 * read-modify-write path so concurrent mutations on the same run_id (for
+	 * example a workflow cancel racing its runner's finish) cannot lose updates.
+	 * Stores that do not advertise the atomic capability keep their historical
+	 * non-atomic read-modify-write behavior. Atomic-store failures always
+	 * propagate so callers can retry without an unlocked fallback write.
+	 *
+	 * @param callable(array{runs:array<string,array<string,mixed>>,queues:array<string,array<int,array<string,mixed>>>,events:array<string,array<int,array<string,mixed>>>}):array{state:array{runs:array<string,array<string,mixed>>,queues:array<string,array<int,array<string,mixed>>>,events:array<string,array<int,array<string,mixed>>>},result:mixed} $mutation State mutation.
+	 * @return mixed Mutation result.
+	 */
+	private static function mutate_run_state( string $store_key, callable $mutation, ?WP_Agent_Workspace_Scope $workspace = null ): mixed {
+		$store = self::store();
+		$default_store_without_wordpress = $store instanceof WP_Agent_Option_Run_Control_Store && ! class_exists( '\wpdb' );
+		if ( null === $workspace && $store instanceof WP_Agent_Atomic_Run_Control_Store && ! $default_store_without_wordpress ) {
+			return $store->mutate_state( $store_key, $mutation );
+		} elseif ( $workspace instanceof WP_Agent_Workspace_Scope && $store instanceof WP_Agent_Atomic_Workspace_Run_Control_Store && ! $default_store_without_wordpress ) {
+			return $store->mutate_workspace_state( $store_key, $workspace, $mutation );
+		}
+
+		// Custom non-atomic stores and the default store before WordPress boots keep
+		// their historical mutation path. Attempted atomic mutations never fall back.
+		$mutated = $mutation( self::state( $store_key, $workspace ) );
+		self::save_state( $store_key, $mutated['state'], $workspace );
+		return $mutated['result'];
+	}
+
+	private static function is_terminal_status( mixed $status ): bool {
+		return in_array(
+			self::normalize_status( $status ),
+			array(
+				self::STATUS_COMPLETED,
+				self::STATUS_SUCCEEDED,
+				self::STATUS_FAILED,
+				self::STATUS_SKIPPED,
+				self::STATUS_CANCELLED,
+				self::STATUS_BUDGET_EXCEEDED,
+				self::STATUS_STALLED,
+				self::STATUS_INTERRUPTED,
+			),
+			true
+		);
+	}
+
+	/** @param array<string,mixed> $run */
+	private static function is_cancellation_requested( array $run ): bool {
+		return self::STATUS_CANCELLING === self::normalize_status( $run['status'] ?? null ) || true === ( $run['cancelled'] ?? false );
+	}
+
+	/**
+	 * @param array<string,mixed> $run Run state.
+	 * @return array<string,mixed>
+	 */
+	private static function normalize_cancellation_state( array $run ): array {
+		$status = self::normalize_status( $run['status'] ?? null );
+		if ( in_array( $status, array( self::STATUS_CANCELLING, self::STATUS_CANCELLED ), true ) ) {
+			$run['cancelled'] = true;
+		} elseif ( isset( $run['cancelled'] ) ) {
+			$run['cancelled'] = false;
+		}
+
+		return $run;
+	}
+
 	public static function now(): string {
 		return gmdate( 'c' );
 	}