From 93e5ce7809379924e0e4f00221b590a480fe5b3f Mon Sep 17 00:00:00 2001 From: Miguel Lezama Date: Thu, 6 Aug 2026 17:42:56 -0300 Subject: [PATCH 1/6] Security: make generic run-control mutations atomic The generic lifecycle mutations on WP_Agent_Run_Control (start_run, save_run, finish_run, request_cancel) performed a non-atomic state() -> mutate -> save_state() read-modify-write with no locking. These generic methods back workflow run-control (WP_Agent_Workflow_Runner, WP_Agent_Workflow_Request_Controller) and runtime-package runs, so a workflow request_cancel (register-agents-workflow-abilities.php) could race the runner's finish_run (class-wp-agent-workflow-runner.php) on the same store and run_id: both read the same base state and the second save silently overwrites the first, losing the cancel or the terminal status. Route all four mutations through the store's atomic read-modify-write path (mutate_state / mutate_workspace_state) via a new private mutate_run_state() helper, closing the lost-update window. This mirrors commit 8980da5, which hardened the chat layer (WP_Agent_Chat_Run_Control) to the atomic path but left these generic methods non-atomic. Stores without atomic support, and environments without a database connection (pure-PHP execution before WordPress boots), keep their historical non-atomic behavior, matching the chat layer's wpdb-availability guard. Adds tests/run-control-atomic-mutations-smoke.php, which uses a store spy to assert the four mutations take the atomic mutate path (not plain save_state) and that an interleaved concurrent write is not lost. The test fails without the fix and passes with it. Co-Authored-By: Claude Opus 4.8 (1M context) --- composer.json | 1 + src/Runtime/class-wp-agent-run-control.php | 147 ++++++++++++----- tests/run-control-atomic-mutations-smoke.php | 156 +++++++++++++++++++ 3 files changed, 265 insertions(+), 39 deletions(-) create mode 100644 tests/run-control-atomic-mutations-smoke.php diff --git a/composer.json b/composer.json index 8b864bf..2e00898 100644 --- a/composer.json +++ b/composer.json @@ -110,6 +110,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/canonical-run-lifecycle-smoke.php", "php tests/channels-smoke.php", "php tests/chat-run-control-smoke.php", diff --git a/src/Runtime/class-wp-agent-run-control.php b/src/Runtime/class-wp-agent-run-control.php index ea8860d..2e4b4d5 100644 --- a/src/Runtime/class-wp-agent-run-control.php +++ b/src/Runtime/class-wp-agent-run-control.php @@ -249,10 +249,15 @@ public static function start_run( string $store_key, string $run_id, array $run ) ); - $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 ); + self::mutate_run_state( + $store_key, + static function ( array $state ) use ( $run_id, $run ): array { + $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' => null ); + }, + $workspace + ); return self::normalize_run( $run ); } @@ -269,10 +274,14 @@ public static function save_run( string $store_key, array $run ): array { $normalized['updated_at'] = '' !== $normalized['updated_at'] ? $normalized['updated_at'] : self::now(); $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 ); + self::mutate_run_state( + $store_key, + static function ( array $state ) use ( $run_id, $normalized ): array { + $state['runs'][ $run_id ] = $normalized; + $state = self::record_event_in_state( $state, $run_id, 'run_updated', array( 'status' => $normalized['status'] ) ); + return array( 'state' => $state, 'result' => null ); + } + ); return $normalized; } @@ -286,23 +295,28 @@ public static function save_run( string $store_key, array $run ): array { * @return array|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 = $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'] ) ); + 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 +336,27 @@ public static function get_run( string $store_key, string $run_id, ?WP_Agent_Wor * @return array|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 = $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'] ) ); + 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 +460,56 @@ 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. + * + * @param callable(array{runs:array>,queues:array>>,events:array>>}):array{state:array{runs:array>,queues:array>>,events:array>>},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(); + if ( null === $workspace && $store instanceof WP_Agent_Atomic_Run_Control_Store ) { + try { + return $store->mutate_state( $store_key, $mutation ); + } catch ( \RuntimeException $error ) { + self::rethrow_if_database_available( $error ); + } + } elseif ( $workspace instanceof WP_Agent_Workspace_Scope && $store instanceof WP_Agent_Atomic_Workspace_Run_Control_Store ) { + try { + return $store->mutate_workspace_state( $store_key, $workspace, $mutation ); + } catch ( \RuntimeException $error ) { + self::rethrow_if_database_available( $error ); + } + } + + // Stores without atomic support, or environments without a database + // connection (for example pure-PHP execution before WordPress boots), + // keep the historical non-atomic read-modify-write. + $mutated = $mutation( self::state( $store_key, $workspace ) ); + self::save_state( $store_key, $mutated['state'], $workspace ); + return $mutated['result']; + } + + /** + * Rethrow an atomic-mutation failure when a real database is available. + * + * A booted WordPress always defines the wpdb class, so a genuine atomic + * failure there is never swallowed. Without wpdb the only possible failure + * is the missing database connection, which safely degrades to the + * non-atomic read-modify-write path. + */ + private static function rethrow_if_database_available( \RuntimeException $error ): void { + if ( class_exists( '\wpdb' ) ) { + throw $error; + } + } + public static function now(): string { return gmdate( 'c' ); } diff --git a/tests/run-control-atomic-mutations-smoke.php b/tests/run-control-atomic-mutations-smoke.php new file mode 100644 index 0000000..47d3469 --- /dev/null +++ b/tests/run-control-atomic-mutations-smoke.php @@ -0,0 +1,156 @@ + mutate -> save_state() sequence. + * + * A non-atomic read-modify-write leaves a lost-update window: a workflow + * request_cancel that reads state, then a runner finish_run that also reads the + * same base state, will each overwrite the other on save. Routing every + * lifecycle mutation through the store's atomic mutate_state closes that window. + * + * Run with: php tests/run-control-atomic-mutations-smoke.php + * + * @package AgentsAPI\Tests + */ + +use AgentsAPI\AI\WP_Agent_Atomic_Workspace_Run_Control_Store; +use AgentsAPI\AI\WP_Agent_Run_Control; +use AgentsAPI\Core\Workspace\WP_Agent_Workspace_Scope; + +if ( ! defined( 'ABSPATH' ) ) { + define( 'ABSPATH', __DIR__ . '/' ); +} + +$failures = array(); +$passes = 0; + +echo "run-control-atomic-mutations-smoke\n"; + +require_once __DIR__ . '/agents-api-smoke-helpers.php'; + +if ( ! class_exists( 'WP_Error' ) ) { + class WP_Error { + public function __construct( private string $code = '', private string $message = '', private array $data = array() ) {} + public function get_error_code(): string { return $this->code; } + public function get_error_message(): string { return $this->message; } + public function get_error_data(): array { return $this->data; } + } +} + +agents_api_smoke_require_module(); + +/** + * Store spy that separates the atomic mutate path from the non-atomic + * save_state path so a test can assert which one a lifecycle mutation took. + * Its mutate_state deliberately does NOT call save_state, so the two counters + * never overlap. + */ +final class Run_Control_Atomic_Spy_Store implements WP_Agent_Atomic_Workspace_Run_Control_Store { + + public int $mutate_calls = 0; + public int $mutate_workspace_calls = 0; + public int $save_calls = 0; + public int $save_workspace_calls = 0; + + /** @var callable|null Runs inside mutate_state before the mutation, to simulate a concurrent committed write. */ + public $before_mutation = null; + + /** @var array>,queues:array>>,events:array>>}> */ + private array $states = array(); + + public function get_state( string $store_key ): array { + return $this->states[ 'site:' . $store_key ] ?? $this->empty_state(); + } + + public function save_state( string $store_key, array $state ): void { + ++$this->save_calls; + $this->states[ 'site:' . $store_key ] = $state; + } + + public function mutate_state( string $store_key, callable $mutation ): mixed { + ++$this->mutate_calls; + if ( is_callable( $this->before_mutation ) ) { + ( $this->before_mutation )( $this, $store_key ); + } + $mutated = $mutation( $this->states[ 'site:' . $store_key ] ?? $this->empty_state() ); + $this->states[ 'site:' . $store_key ] = $mutated['state']; + return $mutated['result']; + } + + public function get_workspace_state( string $store_key, WP_Agent_Workspace_Scope $workspace ): array { + return $this->states[ 'workspace:' . $workspace->key() . ':' . $store_key ] ?? $this->empty_state(); + } + + public function save_workspace_state( string $store_key, WP_Agent_Workspace_Scope $workspace, array $state ): void { + ++$this->save_workspace_calls; + $this->states[ 'workspace:' . $workspace->key() . ':' . $store_key ] = $state; + } + + public function mutate_workspace_state( string $store_key, WP_Agent_Workspace_Scope $workspace, callable $mutation ): mixed { + ++$this->mutate_workspace_calls; + $key = 'workspace:' . $workspace->key() . ':' . $store_key; + $mutated = $mutation( $this->states[ $key ] ?? $this->empty_state() ); + $this->states[ $key ] = $mutated['state']; + return $mutated['result']; + } + + /** Inject a run directly into stored state, bypassing the counters. */ + public function inject_run( string $store_key, string $run_id, array $run ): void { + $state = $this->states[ 'site:' . $store_key ] ?? $this->empty_state(); + $state['runs'][ $run_id ] = $run; + $this->states[ 'site:' . $store_key ] = $state; + } + + /** @return array{runs:array>,queues:array>>,events:array>>} */ + private function empty_state(): array { + return array( 'runs' => array(), 'queues' => array(), 'events' => array() ); + } +} + +// --- Section A: site-local lifecycle mutations route through mutate_state --- +$spy = new Run_Control_Atomic_Spy_Store(); +WP_Agent_Run_Control::set_store( $spy ); + +WP_Agent_Run_Control::start_run( 'atomic-store', 'run-a' ); +WP_Agent_Run_Control::save_run( 'atomic-store', array( 'run_id' => 'run-a', 'status' => 'running' ) ); +WP_Agent_Run_Control::finish_run( 'atomic-store', 'run-a', WP_Agent_Run_Control::STATUS_COMPLETED ); +WP_Agent_Run_Control::request_cancel( 'atomic-store', 'run-a' ); + +agents_api_smoke_assert_equals( 4, $spy->mutate_calls, 'start/save/finish/cancel all route through the atomic mutate_state path', $failures, $passes ); +agents_api_smoke_assert_equals( 0, $spy->save_calls, 'lifecycle mutations never use the non-atomic save_state path', $failures, $passes ); + +// --- Section B: workspace lifecycle mutations route through mutate_workspace_state --- +$workspace_spy = new Run_Control_Atomic_Spy_Store(); +WP_Agent_Run_Control::set_store( $workspace_spy ); +$workspace = WP_Agent_Workspace_Scope::from_parts( 'site', 'atomic-ws' ); + +WP_Agent_Run_Control::start_run( 'atomic-store', 'run-w', array(), $workspace ); +WP_Agent_Run_Control::finish_run( 'atomic-store', 'run-w', WP_Agent_Run_Control::STATUS_COMPLETED, $workspace ); +WP_Agent_Run_Control::request_cancel( 'atomic-store', 'run-w', $workspace ); + +agents_api_smoke_assert_equals( 3, $workspace_spy->mutate_workspace_calls, 'workspace lifecycle mutations route through the atomic mutate_workspace_state path', $failures, $passes ); +agents_api_smoke_assert_equals( 0, $workspace_spy->save_workspace_calls, 'workspace lifecycle mutations never use the non-atomic save_workspace_state path', $failures, $passes ); + +// --- Section C: an interleaved concurrent write is not lost --- +// Simulate a second writer that commits a different run between our read and +// write. A non-atomic read-modify-write would clobber it; the atomic path reads +// current state inside the mutation and preserves it. +$race_spy = new Run_Control_Atomic_Spy_Store(); +WP_Agent_Run_Control::set_store( $race_spy ); + +WP_Agent_Run_Control::start_run( 'race-store', 'runner-run' ); +$race_spy->before_mutation = static function ( Run_Control_Atomic_Spy_Store $store, string $store_key ): void { + $store->before_mutation = null; // Fire exactly once. + $store->inject_run( $store_key, 'concurrent-run', array( 'run_id' => 'concurrent-run', 'status' => 'running' ) ); +}; + +$finished = WP_Agent_Run_Control::finish_run( 'race-store', 'runner-run', WP_Agent_Run_Control::STATUS_COMPLETED ); + +agents_api_smoke_assert_equals( 'completed', $finished['status'] ?? null, 'finish_run still commits its own terminal status', $failures, $passes ); +$concurrent = WP_Agent_Run_Control::get_run( 'race-store', 'concurrent-run' ); +agents_api_smoke_assert_equals( 'running', $concurrent['status'] ?? null, 'a concurrently committed run survives an interleaved lifecycle mutation', $failures, $passes ); +$runner = WP_Agent_Run_Control::get_run( 'race-store', 'runner-run' ); +agents_api_smoke_assert_equals( 'completed', $runner['status'] ?? null, 'the mutated run keeps its committed status alongside the concurrent run', $failures, $passes ); + +agents_api_smoke_finish( 'run-control atomic mutations', $failures, $passes ); From fbcf5409954efd940b677986096ab0b182dcab10 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 21:55:52 +0000 Subject: [PATCH 2/6] fix: make run-control lifecycle transitions atomic --- src/Runtime/class-wp-agent-run-control.php | 131 +++++++----- tests/run-control-atomic-mutations-smoke.php | 199 ++++++++++++------- 2 files changed, 212 insertions(+), 118 deletions(-) diff --git a/src/Runtime/class-wp-agent-run-control.php b/src/Runtime/class-wp-agent-run-control.php index 7c3b5e8..1b738f9 100644 --- a/src/Runtime/class-wp-agent-run-control.php +++ b/src/Runtime/class-wp-agent-run-control.php @@ -238,7 +238,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,19 +247,24 @@ 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(), ) - ); + ) ); - self::mutate_run_state( + $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 ) && ( self::is_terminal_status( $current['status'] ?? null ) || self::is_cancellation_requested( $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' => null ); + 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 ); } /** @@ -267,23 +272,37 @@ static function ( array $state ) use ( $run_id, $run ): array { * * @param string $store_key Option key used by the backing store. * @param array $run Run payload. + * @param WP_Agent_Workspace_Scope|null $workspace Explicit workspace scope. * @return array */ - 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'] ); - self::mutate_run_state( + $result = self::mutate_run_state( $store_key, static function ( array $state ) use ( $run_id, $normalized ): array { - $state['runs'][ $run_id ] = $normalized; - $state = self::record_event_in_state( $state, $run_id, 'run_updated', array( 'status' => $normalized['status'] ) ); - return array( 'state' => $state, 'result' => null ); - } + $current = $state['runs'][ $run_id ] ?? null; + 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; } /** @@ -302,13 +321,15 @@ static function ( array $state ) use ( $run_id, $status ): array { return array( 'state' => $state, 'result' => 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; + $run = $state['runs'][ $run_id ]; + 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 ); @@ -343,10 +364,13 @@ static function ( array $state ) use ( $run_id ): array { return array( 'state' => $state, 'result' => 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 = $state['runs'][ $run_id ]; + 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; @@ -467,47 +491,62 @@ public static function mutate_state( string $store_key, callable $mutation, ?WP_ * 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. + * 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>,queues:array>>,events:array>>}):array{state:array{runs:array>,queues:array>>,events:array>>},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(); - if ( null === $workspace && $store instanceof WP_Agent_Atomic_Run_Control_Store ) { - try { - return $store->mutate_state( $store_key, $mutation ); - } catch ( \RuntimeException $error ) { - self::rethrow_if_database_available( $error ); - } - } elseif ( $workspace instanceof WP_Agent_Workspace_Scope && $store instanceof WP_Agent_Atomic_Workspace_Run_Control_Store ) { - try { - return $store->mutate_workspace_state( $store_key, $workspace, $mutation ); - } catch ( \RuntimeException $error ) { - self::rethrow_if_database_available( $error ); - } + $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 ); } - // Stores without atomic support, or environments without a database - // connection (for example pure-PHP execution before WordPress boots), - // keep the historical non-atomic read-modify-write. + // 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_CANCELLED, + self::STATUS_BUDGET_EXCEEDED, + self::STATUS_STALLED, + self::STATUS_INTERRUPTED, + ), + true + ); + } + + /** @param array $run */ + private static function is_cancellation_requested( array $run ): bool { + return self::STATUS_CANCELLING === self::normalize_status( $run['status'] ?? null ) || true === ( $run['cancelled'] ?? false ); + } + /** - * Rethrow an atomic-mutation failure when a real database is available. - * - * A booted WordPress always defines the wpdb class, so a genuine atomic - * failure there is never swallowed. Without wpdb the only possible failure - * is the missing database connection, which safely degrades to the - * non-atomic read-modify-write path. + * @param array $run Run state. + * @return array */ - private static function rethrow_if_database_available( \RuntimeException $error ): void { - if ( class_exists( '\wpdb' ) ) { - throw $error; + 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 { diff --git a/tests/run-control-atomic-mutations-smoke.php b/tests/run-control-atomic-mutations-smoke.php index 47d3469..cefa228 100644 --- a/tests/run-control-atomic-mutations-smoke.php +++ b/tests/run-control-atomic-mutations-smoke.php @@ -1,13 +1,6 @@ mutate -> save_state() sequence. - * - * A non-atomic read-modify-write leaves a lost-update window: a workflow - * request_cancel that reads state, then a runner finish_run that also reads the - * same base state, will each overwrite the other on save. Routing every - * lifecycle mutation through the store's atomic mutate_state closes that window. + * Pure-PHP smoke test for atomic generic run-control lifecycle mutations. * * Run with: php tests/run-control-atomic-mutations-smoke.php * @@ -16,11 +9,10 @@ use AgentsAPI\AI\WP_Agent_Atomic_Workspace_Run_Control_Store; use AgentsAPI\AI\WP_Agent_Run_Control; +use AgentsAPI\AI\WP_Agent_Run_Control_Store_Exception; use AgentsAPI\Core\Workspace\WP_Agent_Workspace_Scope; -if ( ! defined( 'ABSPATH' ) ) { - define( 'ABSPATH', __DIR__ . '/' ); -} +defined( 'ABSPATH' ) || define( 'ABSPATH', __DIR__ . '/' ); $failures = array(); $passes = 0; @@ -40,66 +32,74 @@ public function get_error_data(): array { return $this->data; } agents_api_smoke_require_module(); -/** - * Store spy that separates the atomic mutate path from the non-atomic - * save_state path so a test can assert which one a lifecycle mutation took. - * Its mutate_state deliberately does NOT call save_state, so the two counters - * never overlap. - */ final class Run_Control_Atomic_Spy_Store implements WP_Agent_Atomic_Workspace_Run_Control_Store { public int $mutate_calls = 0; public int $mutate_workspace_calls = 0; public int $save_calls = 0; public int $save_workspace_calls = 0; - - /** @var callable|null Runs inside mutate_state before the mutation, to simulate a concurrent committed write. */ + public ?WP_Agent_Run_Control_Store_Exception $site_failure = null; + public ?WP_Agent_Run_Control_Store_Exception $workspace_failure = null; + /** @var callable|null */ public $before_mutation = null; /** @var array>,queues:array>>,events:array>>}> */ private array $states = array(); public function get_state( string $store_key ): array { - return $this->states[ 'site:' . $store_key ] ?? $this->empty_state(); + return $this->states[ $this->state_key( $store_key ) ] ?? $this->empty_state(); } public function save_state( string $store_key, array $state ): void { ++$this->save_calls; - $this->states[ 'site:' . $store_key ] = $state; + $this->states[ $this->state_key( $store_key ) ] = $state; } public function mutate_state( string $store_key, callable $mutation ): mixed { ++$this->mutate_calls; - if ( is_callable( $this->before_mutation ) ) { - ( $this->before_mutation )( $this, $store_key ); + if ( $this->site_failure instanceof WP_Agent_Run_Control_Store_Exception ) { + throw $this->site_failure; } - $mutated = $mutation( $this->states[ 'site:' . $store_key ] ?? $this->empty_state() ); - $this->states[ 'site:' . $store_key ] = $mutated['state']; - return $mutated['result']; + return $this->mutate( $store_key, $mutation ); } public function get_workspace_state( string $store_key, WP_Agent_Workspace_Scope $workspace ): array { - return $this->states[ 'workspace:' . $workspace->key() . ':' . $store_key ] ?? $this->empty_state(); + return $this->states[ $this->state_key( $store_key, $workspace ) ] ?? $this->empty_state(); } public function save_workspace_state( string $store_key, WP_Agent_Workspace_Scope $workspace, array $state ): void { ++$this->save_workspace_calls; - $this->states[ 'workspace:' . $workspace->key() . ':' . $store_key ] = $state; + $this->states[ $this->state_key( $store_key, $workspace ) ] = $state; } public function mutate_workspace_state( string $store_key, WP_Agent_Workspace_Scope $workspace, callable $mutation ): mixed { ++$this->mutate_workspace_calls; - $key = 'workspace:' . $workspace->key() . ':' . $store_key; + if ( $this->workspace_failure instanceof WP_Agent_Run_Control_Store_Exception ) { + throw $this->workspace_failure; + } + return $this->mutate( $store_key, $mutation, $workspace ); + } + + /** @param array $run */ + public function inject_run( string $store_key, string $run_id, array $run, ?WP_Agent_Workspace_Scope $workspace = null ): void { + $key = $this->state_key( $store_key, $workspace ); + $state = $this->states[ $key ] ?? $this->empty_state(); + $state['runs'][ $run_id ] = $run; + $this->states[ $key ] = $state; + } + + private function mutate( string $store_key, callable $mutation, ?WP_Agent_Workspace_Scope $workspace = null ): mixed { + if ( is_callable( $this->before_mutation ) ) { + ( $this->before_mutation )( $this, $store_key, $workspace ); + } + $key = $this->state_key( $store_key, $workspace ); $mutated = $mutation( $this->states[ $key ] ?? $this->empty_state() ); $this->states[ $key ] = $mutated['state']; return $mutated['result']; } - /** Inject a run directly into stored state, bypassing the counters. */ - public function inject_run( string $store_key, string $run_id, array $run ): void { - $state = $this->states[ 'site:' . $store_key ] ?? $this->empty_state(); - $state['runs'][ $run_id ] = $run; - $this->states[ 'site:' . $store_key ] = $state; + private function state_key( string $store_key, ?WP_Agent_Workspace_Scope $workspace = null ): string { + return null === $workspace ? 'site:' . $store_key : 'workspace:' . $workspace->key() . ':' . $store_key; } /** @return array{runs:array>,queues:array>>,events:array>>} */ @@ -108,49 +108,104 @@ private function empty_state(): array { } } -// --- Section A: site-local lifecycle mutations route through mutate_state --- -$spy = new Run_Control_Atomic_Spy_Store(); -WP_Agent_Run_Control::set_store( $spy ); +/** @return array */ +function atomic_get_run( string $store_key, string $run_id, ?WP_Agent_Workspace_Scope $workspace ): array { + return WP_Agent_Run_Control::get_run( $store_key, $run_id, $workspace ) ?? array(); +} + +function atomic_start( string $store_key, string $run_id, ?WP_Agent_Workspace_Scope $workspace ): void { + WP_Agent_Run_Control::start_run( $store_key, $run_id, array(), $workspace ); +} -WP_Agent_Run_Control::start_run( 'atomic-store', 'run-a' ); -WP_Agent_Run_Control::save_run( 'atomic-store', array( 'run_id' => 'run-a', 'status' => 'running' ) ); -WP_Agent_Run_Control::finish_run( 'atomic-store', 'run-a', WP_Agent_Run_Control::STATUS_COMPLETED ); -WP_Agent_Run_Control::request_cancel( 'atomic-store', 'run-a' ); +function atomic_save( string $store_key, string $run_id, string $status, ?WP_Agent_Workspace_Scope $workspace ): array { + return WP_Agent_Run_Control::save_run( $store_key, array( 'run_id' => $run_id, 'status' => $status ), $workspace ); +} -agents_api_smoke_assert_equals( 4, $spy->mutate_calls, 'start/save/finish/cancel all route through the atomic mutate_state path', $failures, $passes ); -agents_api_smoke_assert_equals( 0, $spy->save_calls, 'lifecycle mutations never use the non-atomic save_state path', $failures, $passes ); +$site_spy = new Run_Control_Atomic_Spy_Store(); +WP_Agent_Run_Control::set_store( $site_spy ); +atomic_start( 'atomic-site', 'run-site', null ); +atomic_save( 'atomic-site', 'run-site', WP_Agent_Run_Control::STATUS_RUNNING, null ); +WP_Agent_Run_Control::finish_run( 'atomic-site', 'run-site' ); +WP_Agent_Run_Control::request_cancel( 'atomic-site', 'run-site' ); +agents_api_smoke_assert_equals( 4, $site_spy->mutate_calls, 'all site lifecycle writes use mutate_state', $failures, $passes ); +agents_api_smoke_assert_equals( 0, $site_spy->save_calls, 'site lifecycle writes never use unlocked save_state', $failures, $passes ); -// --- Section B: workspace lifecycle mutations route through mutate_workspace_state --- $workspace_spy = new Run_Control_Atomic_Spy_Store(); +$workspace = WP_Agent_Workspace_Scope::from_parts( 'site', 'atomic-ws' ); WP_Agent_Run_Control::set_store( $workspace_spy ); -$workspace = WP_Agent_Workspace_Scope::from_parts( 'site', 'atomic-ws' ); - -WP_Agent_Run_Control::start_run( 'atomic-store', 'run-w', array(), $workspace ); -WP_Agent_Run_Control::finish_run( 'atomic-store', 'run-w', WP_Agent_Run_Control::STATUS_COMPLETED, $workspace ); -WP_Agent_Run_Control::request_cancel( 'atomic-store', 'run-w', $workspace ); - -agents_api_smoke_assert_equals( 3, $workspace_spy->mutate_workspace_calls, 'workspace lifecycle mutations route through the atomic mutate_workspace_state path', $failures, $passes ); -agents_api_smoke_assert_equals( 0, $workspace_spy->save_workspace_calls, 'workspace lifecycle mutations never use the non-atomic save_workspace_state path', $failures, $passes ); - -// --- Section C: an interleaved concurrent write is not lost --- -// Simulate a second writer that commits a different run between our read and -// write. A non-atomic read-modify-write would clobber it; the atomic path reads -// current state inside the mutation and preserves it. -$race_spy = new Run_Control_Atomic_Spy_Store(); -WP_Agent_Run_Control::set_store( $race_spy ); - -WP_Agent_Run_Control::start_run( 'race-store', 'runner-run' ); -$race_spy->before_mutation = static function ( Run_Control_Atomic_Spy_Store $store, string $store_key ): void { - $store->before_mutation = null; // Fire exactly once. - $store->inject_run( $store_key, 'concurrent-run', array( 'run_id' => 'concurrent-run', 'status' => 'running' ) ); -}; - -$finished = WP_Agent_Run_Control::finish_run( 'race-store', 'runner-run', WP_Agent_Run_Control::STATUS_COMPLETED ); +atomic_start( 'atomic-workspace', 'run-workspace', $workspace ); +atomic_save( 'atomic-workspace', 'run-workspace', WP_Agent_Run_Control::STATUS_RUNNING, $workspace ); +WP_Agent_Run_Control::finish_run( 'atomic-workspace', 'run-workspace', WP_Agent_Run_Control::STATUS_COMPLETED, $workspace ); +WP_Agent_Run_Control::request_cancel( 'atomic-workspace', 'run-workspace', $workspace ); +agents_api_smoke_assert_equals( 4, $workspace_spy->mutate_workspace_calls, 'all workspace lifecycle writes use mutate_workspace_state', $failures, $passes ); +agents_api_smoke_assert_equals( 0, $workspace_spy->save_workspace_calls, 'workspace lifecycle writes never use unlocked save_workspace_state', $failures, $passes ); + +foreach ( array( null, $workspace ) as $scope ) { + $scope_name = null === $scope ? 'site' : 'workspace'; + $race_spy = new Run_Control_Atomic_Spy_Store(); + WP_Agent_Run_Control::set_store( $race_spy ); + atomic_start( 'unrelated-race', 'runner-run', $scope ); + $race_spy->before_mutation = static function ( Run_Control_Atomic_Spy_Store $store, string $store_key, ?WP_Agent_Workspace_Scope $active_scope ): void { + $store->before_mutation = null; + $store->inject_run( $store_key, 'concurrent-run', array( 'run_id' => 'concurrent-run', 'status' => 'running' ), $active_scope ); + }; + WP_Agent_Run_Control::finish_run( 'unrelated-race', 'runner-run', WP_Agent_Run_Control::STATUS_COMPLETED, $scope ); + agents_api_smoke_assert_equals( 'running', atomic_get_run( 'unrelated-race', 'concurrent-run', $scope )['status'] ?? null, "{$scope_name} atomic mutation preserves an unrelated concurrent run", $failures, $passes ); + + $cancel_first = new Run_Control_Atomic_Spy_Store(); + WP_Agent_Run_Control::set_store( $cancel_first ); + atomic_start( 'cancel-first', 'same-run', $scope ); + WP_Agent_Run_Control::request_cancel( 'cancel-first', 'same-run', $scope ); + $cancelled = WP_Agent_Run_Control::finish_run( 'cancel-first', 'same-run', WP_Agent_Run_Control::STATUS_SUCCEEDED, $scope ); + agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_CANCELLED, $cancelled['status'] ?? null, "{$scope_name} committed cancellation wins a later successful finish", $failures, $passes ); + agents_api_smoke_assert_equals( true, $cancelled['cancelled'] ?? null, "{$scope_name} cancelled terminal state remains internally consistent", $failures, $passes ); + $saved_after_cancel = atomic_save( 'cancel-first', 'same-run', WP_Agent_Run_Control::STATUS_SUCCEEDED, $scope ); + agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_CANCELLED, $saved_after_cancel['status'] ?? null, "{$scope_name} save_run cannot replace a cancelled terminal state", $failures, $passes ); + + $finish_first = new Run_Control_Atomic_Spy_Store(); + WP_Agent_Run_Control::set_store( $finish_first ); + atomic_start( 'finish-first', 'same-run', $scope ); + $finished = WP_Agent_Run_Control::finish_run( 'finish-first', 'same-run', WP_Agent_Run_Control::STATUS_SUCCEEDED, $scope ); + $events_before_noops = count( $finish_first->get_workspace_state( 'finish-first', $workspace )['events']['same-run'] ?? array() ); + if ( null === $scope ) { + $events_before_noops = count( $finish_first->get_state( 'finish-first' )['events']['same-run'] ?? array() ); + } + WP_Agent_Run_Control::request_cancel( 'finish-first', 'same-run', $scope ); + atomic_save( 'finish-first', 'same-run', WP_Agent_Run_Control::STATUS_RUNNING, $scope ); + atomic_start( 'finish-first', 'same-run', $scope ); + $terminal = WP_Agent_Run_Control::finish_run( 'finish-first', 'same-run', WP_Agent_Run_Control::STATUS_FAILED, $scope ); + agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_SUCCEEDED, $finished['status'] ?? null, "{$scope_name} finish commits success before a later cancellation", $failures, $passes ); + agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_SUCCEEDED, $terminal['status'] ?? null, "{$scope_name} terminal state is monotonic across every lifecycle writer", $failures, $passes ); + agents_api_smoke_assert_equals( false, $terminal['cancelled'] ?? false, "{$scope_name} cancellation after terminal does not create a contradictory flag", $failures, $passes ); + $events_after_noops = count( $finish_first->get_workspace_state( 'finish-first', $workspace )['events']['same-run'] ?? array() ); + if ( null === $scope ) { + $events_after_noops = count( $finish_first->get_state( 'finish-first' )['events']['same-run'] ?? array() ); + } + agents_api_smoke_assert_equals( $events_before_noops, $events_after_noops, "{$scope_name} post-terminal lifecycle requests are event-free no-ops", $failures, $passes ); +} -agents_api_smoke_assert_equals( 'completed', $finished['status'] ?? null, 'finish_run still commits its own terminal status', $failures, $passes ); -$concurrent = WP_Agent_Run_Control::get_run( 'race-store', 'concurrent-run' ); -agents_api_smoke_assert_equals( 'running', $concurrent['status'] ?? null, 'a concurrently committed run survives an interleaved lifecycle mutation', $failures, $passes ); -$runner = WP_Agent_Run_Control::get_run( 'race-store', 'runner-run' ); -agents_api_smoke_assert_equals( 'completed', $runner['status'] ?? null, 'the mutated run keeps its committed status alongside the concurrent run', $failures, $passes ); +$failure_spy = new Run_Control_Atomic_Spy_Store(); +$failure_spy->site_failure = new WP_Agent_Run_Control_Store_Exception( 'retry site mutation' ); +WP_Agent_Run_Control::set_store( $failure_spy ); +try { + atomic_start( 'failed-site', 'run', null ); + $site_failure_propagated = false; +} catch ( WP_Agent_Run_Control_Store_Exception $error ) { + $site_failure_propagated = 'retry site mutation' === $error->getMessage(); +} +agents_api_smoke_assert_equals( true, $site_failure_propagated, 'typed site atomic failure propagates without fallback', $failures, $passes ); +agents_api_smoke_assert_equals( 0, $failure_spy->save_calls, 'site atomic failure never falls back to unlocked save_state', $failures, $passes ); + +$failure_spy->site_failure = null; +$failure_spy->workspace_failure = new WP_Agent_Run_Control_Store_Exception( 'retry workspace mutation' ); +try { + atomic_start( 'failed-workspace', 'run', $workspace ); + $workspace_failure_propagated = false; +} catch ( WP_Agent_Run_Control_Store_Exception $error ) { + $workspace_failure_propagated = 'retry workspace mutation' === $error->getMessage(); +} +agents_api_smoke_assert_equals( true, $workspace_failure_propagated, 'typed workspace atomic failure propagates without fallback', $failures, $passes ); +agents_api_smoke_assert_equals( 0, $failure_spy->save_workspace_calls, 'workspace atomic failure never falls back to unlocked save_workspace_state', $failures, $passes ); +WP_Agent_Run_Control::reset_store(); agents_api_smoke_finish( 'run-control atomic mutations', $failures, $passes ); From 620a80cb252e9e4fa83002ffe8f275454e596f15 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 22:04:35 +0000 Subject: [PATCH 3/6] fix: project authoritative run-control outcomes --- src/Runtime/class-wp-agent-run-control.php | 16 +++- .../register-runtime-package-run-ability.php | 9 ++- .../class-wp-agent-workflow-runner.php | 28 ++++++- tests/run-control-atomic-mutations-smoke.php | 32 ++++++++ tests/runtime-package-run-contract-smoke.php | 61 +++++++++++++++ tests/workflow-runner-smoke.php | 77 +++++++++++++++++++ 6 files changed, 215 insertions(+), 8 deletions(-) diff --git a/src/Runtime/class-wp-agent-run-control.php b/src/Runtime/class-wp-agent-run-control.php index 1b738f9..5d2d39a 100644 --- a/src/Runtime/class-wp-agent-run-control.php +++ b/src/Runtime/class-wp-agent-run-control.php @@ -109,7 +109,7 @@ public static function normalize_run( array $run ): array { $normalized['cancelled'] = (bool) $run['cancelled']; } - return $normalized; + return self::normalize_cancellation_state( $normalized ); } /** @@ -253,6 +253,10 @@ public static function start_run( string $store_key, string $run_id, array $run $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 ) && ( self::is_terminal_status( $current['status'] ?? null ) || self::is_cancellation_requested( $current ) ) ) { return array( 'state' => $state, 'result' => $current ); } @@ -285,6 +289,10 @@ public static function save_run( string $store_key, array $run, ?WP_Agent_Worksp $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 ); } @@ -321,7 +329,8 @@ static function ( array $state ) use ( $run_id, $status ): array { return array( 'state' => $state, 'result' => null ); } - $run = $state['runs'][ $run_id ]; + $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 ); } @@ -364,7 +373,8 @@ static function ( array $state ) use ( $run_id ): array { return array( 'state' => $state, 'result' => null ); } - $run = $state['runs'][ $run_id ]; + $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 ); } diff --git a/src/Runtime/register-runtime-package-run-ability.php b/src/Runtime/register-runtime-package-run-ability.php index 3b4e02b..359439c 100644 --- a/src/Runtime/register-runtime-package-run-ability.php +++ b/src/Runtime/register-runtime-package-run-ability.php @@ -237,7 +237,7 @@ function agents_runtime_package_run_dispatch( array $input ) { $result['run_id'] = $run_id; $normalized = WP_Agent_Runtime_Package_Run_Result::from_array( $result )->to_array(); $status = WP_Agent_Run_Control::normalize_status( $normalized['status'] ?? WP_Agent_Run_Control::STATUS_SUCCEEDED ); - WP_Agent_Run_Control::save_run( + $authoritative = WP_Agent_Run_Control::save_run( AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, array( 'run_id' => $run_id, @@ -249,6 +249,13 @@ function agents_runtime_package_run_dispatch( array $input ) { 'started_at' => WP_Agent_Run_Control::now(), ) ); + if ( WP_Agent_Run_Control::STATUS_CANCELLED === ( $authoritative['status'] ?? '' ) ) { + $normalized['status'] = WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED; + $normalized['error'] = array( + 'code' => 'cancel_requested', + 'message' => 'Runtime package run cancellation was requested.', + ); + } return $normalized; } diff --git a/src/Workflows/class-wp-agent-workflow-runner.php b/src/Workflows/class-wp-agent-workflow-runner.php index 8de2306..8dbe5e3 100644 --- a/src/Workflows/class-wp-agent-workflow-runner.php +++ b/src/Workflows/class-wp-agent-workflow-runner.php @@ -400,14 +400,16 @@ private function run_step_loop( WP_Agent_Workflow_Spec $spec, WP_Agent_Workflow_ ) ); - if ( $this->recorder ) { - $this->recorder->update( $result ); - } - WP_Agent_Run_Control::finish_run( + $authoritative = WP_Agent_Run_Control::finish_run( self::RUN_CONTROL_STORE, $result->get_run_id(), $failed ? WP_Agent_Run_Control::STATUS_FAILED : WP_Agent_Run_Control::STATUS_SUCCEEDED ); + $result = self::project_authoritative_terminal( $result, $authoritative ); + + if ( $this->recorder ) { + $this->recorder->update( $result ); + } /** * Fires when a workflow run reaches a terminal state through the step loop. @@ -517,6 +519,24 @@ private static function is_cancel_requested( string $run_id ): bool { private static function cancelled_result( WP_Agent_Workflow_Run_Result $result, array $step_records ): WP_Agent_Workflow_Run_Result { WP_Agent_Run_Control::finish_run( self::RUN_CONTROL_STORE, $result->get_run_id(), WP_Agent_Run_Control::STATUS_CANCELLED ); + return self::project_cancelled_result( $result, $step_records ); + } + + /** + * Project the serialized run-control winner into the workflow result. + * + * @param array|null $authoritative Authoritative run-control row. + */ + private static function project_authoritative_terminal( WP_Agent_Workflow_Run_Result $result, ?array $authoritative ): WP_Agent_Workflow_Run_Result { + if ( null !== $authoritative && WP_Agent_Run_Control::STATUS_CANCELLED === ( $authoritative['status'] ?? '' ) ) { + return self::project_cancelled_result( $result, $result->get_steps() ); + } + + return $result; + } + + /** @param array $step_records Steps completed before cancellation won. */ + private static function project_cancelled_result( WP_Agent_Workflow_Run_Result $result, array $step_records ): WP_Agent_Workflow_Run_Result { return $result->with( array( 'status' => WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, diff --git a/tests/run-control-atomic-mutations-smoke.php b/tests/run-control-atomic-mutations-smoke.php index cefa228..95a207c 100644 --- a/tests/run-control-atomic-mutations-smoke.php +++ b/tests/run-control-atomic-mutations-smoke.php @@ -182,6 +182,38 @@ function atomic_save( string $store_key, string $run_id, string $status, ?WP_Age $events_after_noops = count( $finish_first->get_state( 'finish-first' )['events']['same-run'] ?? array() ); } agents_api_smoke_assert_equals( $events_before_noops, $events_after_noops, "{$scope_name} post-terminal lifecycle requests are event-free no-ops", $failures, $passes ); + + foreach ( array( 'start', 'save', 'finish', 'cancel' ) as $operation ) { + $legacy_spy = new Run_Control_Atomic_Spy_Store(); + $run_id = 'legacy-' . $operation; + $legacy_spy->inject_run( + 'legacy-terminal', + $run_id, + array( + 'run_id' => $run_id, + 'status' => WP_Agent_Run_Control::STATUS_SUCCEEDED, + 'cancelled' => true, + ), + $scope + ); + WP_Agent_Run_Control::set_store( $legacy_spy ); + agents_api_smoke_assert_equals( false, atomic_get_run( 'legacy-terminal', $run_id, $scope )['cancelled'] ?? null, "{$scope_name} reads normalize a legacy succeeded/cancelled contradiction", $failures, $passes ); + + if ( 'start' === $operation ) { + atomic_start( 'legacy-terminal', $run_id, $scope ); + } elseif ( 'save' === $operation ) { + atomic_save( 'legacy-terminal', $run_id, WP_Agent_Run_Control::STATUS_RUNNING, $scope ); + } elseif ( 'finish' === $operation ) { + WP_Agent_Run_Control::finish_run( 'legacy-terminal', $run_id, WP_Agent_Run_Control::STATUS_FAILED, $scope ); + } else { + WP_Agent_Run_Control::request_cancel( 'legacy-terminal', $run_id, $scope ); + } + + $legacy_state = null === $scope ? $legacy_spy->get_state( 'legacy-terminal' ) : $legacy_spy->get_workspace_state( 'legacy-terminal', $scope ); + agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_SUCCEEDED, $legacy_state['runs'][ $run_id ]['status'] ?? null, "{$scope_name} {$operation} keeps a legacy terminal status monotonic", $failures, $passes ); + agents_api_smoke_assert_equals( false, $legacy_state['runs'][ $run_id ]['cancelled'] ?? null, "{$scope_name} {$operation} persists the healed cancellation flag before its terminal guard", $failures, $passes ); + agents_api_smoke_assert_equals( array(), $legacy_state['events'][ $run_id ] ?? array(), "{$scope_name} {$operation} heals legacy terminal state without a lifecycle event", $failures, $passes ); + } } $failure_spy = new Run_Control_Atomic_Spy_Store(); diff --git a/tests/runtime-package-run-contract-smoke.php b/tests/runtime-package-run-contract-smoke.php index 81c0bfb..6a11a47 100644 --- a/tests/runtime-package-run-contract-smoke.php +++ b/tests/runtime-package-run-contract-smoke.php @@ -220,6 +220,37 @@ function wp_get_ability( string $ability ): ?WP_Ability { use AgentsAPI\AI\WP_Agent_Runtime_Package_Run_Request; use AgentsAPI\AI\WP_Agent_Runtime_Package_Run_Result; +use AgentsAPI\AI\WP_Agent_Atomic_Run_Control_Store; +use AgentsAPI\AI\WP_Agent_Run_Control; + +final class Runtime_Package_Late_Cancel_Store implements WP_Agent_Atomic_Run_Control_Store { + public bool $cancel_before_next_mutation = false; + /** @var array>,queues:array>>,events:array>>}> */ + private array $states = array(); + + public function get_state( string $store_key ): array { + return $this->states[ $store_key ] ?? array( 'runs' => array(), 'queues' => array(), 'events' => array() ); + } + + public function save_state( string $store_key, array $state ): void { + $this->states[ $store_key ] = $state; + } + + public function mutate_state( string $store_key, callable $mutation ): mixed { + $state = $this->get_state( $store_key ); + if ( $this->cancel_before_next_mutation ) { + $this->cancel_before_next_mutation = false; + foreach ( $state['runs'] as &$run ) { + $run['status'] = WP_Agent_Run_Control::STATUS_CANCELLING; + $run['cancelled'] = true; + } + unset( $run ); + } + $mutated = $mutation( $state ); + $this->states[ $store_key ] = $mutated['state']; + return $mutated['result']; + } +} echo "\n[0] Runtime package ability resolves in normal and late Abilities API lifecycles:\n"; do_action( 'init' ); @@ -385,4 +416,34 @@ static function ( $handler, WP_Agent_Runtime_Package_Run_Request $handler_reques agents_api_smoke_assert_equals( 'succeeded', is_array( $helper_dispatch ) ? $helper_dispatch['status'] ?? '' : '', 'public helper preserves result status', $failures, $passes ); agents_api_smoke_assert_equals( 'build-site', is_array( $helper_dispatch ) ? $helper_dispatch['result']['workflow_id'] ?? '' : '', 'public helper passes workflow to handler', $failures, $passes ); +echo "\n[5] Atomic cancellation wins after the handler returns success:\n"; +$GLOBALS['__agents_api_smoke_actions']['wp_agent_runtime_package_run_handler'] = array(); +$late_cancel_store = new Runtime_Package_Late_Cancel_Store(); +WP_Agent_Run_Control::set_store( $late_cancel_store ); +add_filter( + 'wp_agent_runtime_package_run_handler', + static function () use ( $late_cancel_store ): callable { + return static function () use ( $late_cancel_store ): array { + $late_cancel_store->cancel_before_next_mutation = true; + return array( + 'status' => WP_Agent_Runtime_Package_Run_Result::STATUS_SUCCEEDED, + 'result' => array( 'candidate' => 'handler-success' ), + ); + }; + } +); +$late_cancel_dispatch = AgentsAPI\AI\agents_runtime_package_run_dispatch( + array( + 'run_id' => 'runtime-late-cancel', + 'package' => array( 'slug' => 'site-builder' ), + 'workflow' => array( 'id' => 'late-cancel' ), + ) +); +$late_cancel_run = WP_Agent_Run_Control::get_run( AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, 'runtime-late-cancel' ); +agents_api_smoke_assert_equals( WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED, is_array( $late_cancel_dispatch ) ? $late_cancel_dispatch['status'] ?? '' : '', 'dispatcher returns the authoritative cancellation winner', $failures, $passes ); +agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_CANCELLED, $late_cancel_run['status'] ?? '', 'runtime package run-control stores the cancellation winner', $failures, $passes ); +agents_api_smoke_assert_equals( 'cancel_requested', is_array( $late_cancel_dispatch ) ? $late_cancel_dispatch['error']['code'] ?? '' : '', 'dispatcher projects the canonical cancellation error', $failures, $passes ); +agents_api_smoke_assert_equals( 'handler-success', is_array( $late_cancel_dispatch ) ? $late_cancel_dispatch['result']['candidate'] ?? '' : '', 'cancellation projection preserves handler evidence', $failures, $passes ); +WP_Agent_Run_Control::reset_store(); + agents_api_smoke_finish( 'Agents API runtime package run contract', $failures, $passes ); diff --git a/tests/workflow-runner-smoke.php b/tests/workflow-runner-smoke.php index 83c7feb..74c2322 100644 --- a/tests/workflow-runner-smoke.php +++ b/tests/workflow-runner-smoke.php @@ -191,6 +191,7 @@ function smoke_assert( $expected, $actual, string $name, array &$failures, int & require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-runner.php'; use AgentsAPI\AI\WP_Agent_Run_Control; +use AgentsAPI\AI\WP_Agent_Atomic_Run_Control_Store; use AgentsAPI\AI\WP_Agent_Run_Control_Store; use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Recorder; use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Result; @@ -241,6 +242,35 @@ public function save_state( string $store_key, array $state ): void { } } +class Workflow_Late_Cancel_Store implements WP_Agent_Atomic_Run_Control_Store { + public bool $cancel_before_next_mutation = false; + /** @var array>,queues:array>>,events:array>>}> */ + private array $states = array(); + + public function get_state( string $store_key ): array { + return $this->states[ $store_key ] ?? array( 'runs' => array(), 'queues' => array(), 'events' => array() ); + } + + public function save_state( string $store_key, array $state ): void { + $this->states[ $store_key ] = $state; + } + + public function mutate_state( string $store_key, callable $mutation ): mixed { + $state = $this->get_state( $store_key ); + if ( $this->cancel_before_next_mutation ) { + $this->cancel_before_next_mutation = false; + foreach ( $state['runs'] as &$run ) { + $run['status'] = WP_Agent_Run_Control::STATUS_CANCELLING; + $run['cancelled'] = true; + } + unset( $run ); + } + $mutated = $mutation( $state ); + $this->states[ $store_key ] = $mutated['state']; + return $mutated['result']; + } +} + // ─── Happy path: 2 sequential ability steps with bindings between them ─── workflow_runner_smoke_register_ability( @@ -278,6 +308,16 @@ static function ( array $input ): array { return array( 'cancel_requested' => true ); } ); +workflow_runner_smoke_register_ability( + 'demo/arm-late-cancel', + static function (): array { + $store = $GLOBALS['__workflow_late_cancel_store'] ?? null; + if ( $store instanceof Workflow_Late_Cancel_Store ) { + $store->cancel_before_next_mutation = true; + } + return array( 'completed' => true ); + } +); workflow_runner_smoke_register_ability( 'agents/chat', static function ( array $input ): \WP_Error { @@ -353,6 +393,43 @@ static function ( array $input ): \WP_Error { smoke_assert( true, 64 === strlen( $result->get_replay_metadata()['workflow_spec_hash'] ?? '' ), 'replay metadata includes sha256 spec hash', $failures, $passes ); smoke_assert( $spec->to_array(), $result->get_replay_metadata()['workflow_spec_snapshot'] ?? array(), 'replay metadata includes workflow spec snapshot', $failures, $passes ); +// ─── Cancellation committed inside finish wins every terminal projection ─ + +$late_cancel_spec = WP_Agent_Workflow_Spec::from_array( + array( + 'id' => 'demo/late-cancel', + 'steps' => array( + array( 'id' => 'arm', 'type' => 'ability', 'ability' => 'demo/arm-late-cancel' ), + ), + ) +); +$late_cancel_store = new Workflow_Late_Cancel_Store(); +$GLOBALS['__workflow_late_cancel_store'] = $late_cancel_store; +$late_cancel_hook_result = null; +WP_Agent_Run_Control::set_store( $late_cancel_store ); +add_action( + 'wp_agent_workflow_run_completed', + static function ( WP_Agent_Workflow_Run_Result $completed, string $run_id ) use ( &$late_cancel_hook_result ): void { + if ( 'late-cancel-run' === $run_id ) { + $late_cancel_hook_result = $completed; + } + }, + 10, + 2 +); +$late_cancel_recorder = new Capture_Recorder(); +$late_cancel_result = ( new WP_Agent_Workflow_Runner( $late_cancel_recorder ) )->run( $late_cancel_spec, array(), array( 'run_id' => 'late-cancel-run' ) ); +$late_cancel_record = end( $late_cancel_recorder->writes ); +$late_cancel_stored = WP_Agent_Run_Control::get_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, 'late-cancel-run' ); + +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, $late_cancel_result->get_status(), 'late atomic cancellation replaces the runner candidate return', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, $late_cancel_record['status'] ?? '', 'late atomic cancellation reaches the recorder terminal update', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, $late_cancel_hook_result instanceof WP_Agent_Workflow_Run_Result ? $late_cancel_hook_result->get_status() : '', 'late atomic cancellation reaches the completion hook', $failures, $passes ); +smoke_assert( WP_Agent_Run_Control::STATUS_CANCELLED, $late_cancel_stored['status'] ?? '', 'late atomic cancellation is the run-control winner', $failures, $passes ); +smoke_assert( 'cancel_requested', $late_cancel_result->get_error()['code'] ?? '', 'late atomic cancellation returns the canonical cancellation error', $failures, $passes ); +WP_Agent_Run_Control::reset_store(); +unset( $GLOBALS['__workflow_late_cancel_store'] ); + // ─── Generic run-control cancellation stops before the next step ────── $cancel_spec = WP_Agent_Workflow_Spec::from_array( From a8810127c547f721f1782d32143f4ed392d9e544 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 22:19:07 +0000 Subject: [PATCH 4/6] fix: make runtime package run IDs idempotent --- src/Runtime/class-wp-agent-run-control.php | 4 +- .../register-runtime-package-run-ability.php | 77 +++++++++++-- tests/runtime-package-run-contract-smoke.php | 103 +++++++++++++++++- 3 files changed, 173 insertions(+), 11 deletions(-) diff --git a/src/Runtime/class-wp-agent-run-control.php b/src/Runtime/class-wp-agent-run-control.php index 5d2d39a..96de0cc 100644 --- a/src/Runtime/class-wp-agent-run-control.php +++ b/src/Runtime/class-wp-agent-run-control.php @@ -229,7 +229,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. @@ -257,7 +257,7 @@ static function ( array $state ) use ( $run_id, $run ): array { $current = self::normalize_cancellation_state( $current ); $state['runs'][ $run_id ] = $current; } - if ( is_array( $current ) && ( self::is_terminal_status( $current['status'] ?? null ) || self::is_cancellation_requested( $current ) ) ) { + if ( is_array( $current ) ) { return array( 'state' => $state, 'result' => $current ); } diff --git a/src/Runtime/register-runtime-package-run-ability.php b/src/Runtime/register-runtime-package-run-ability.php index 359439c..bc96ebc 100644 --- a/src/Runtime/register-runtime-package-run-ability.php +++ b/src/Runtime/register-runtime-package-run-ability.php @@ -184,16 +184,27 @@ function agents_runtime_package_run_dispatch( array $input ) { return $request; } - WP_Agent_Run_Control::start_run( + $claim_token = WP_Agent_Run_Control::generate_run_id( 'claim_' ); + $started = WP_Agent_Run_Control::start_run( AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, $run_id, array( 'metadata' => array( - 'package' => $request->get_package(), - 'workflow' => $request->get_workflow(), + 'package' => $request->get_package(), + 'workflow' => $request->get_workflow(), + '_claim_token' => $claim_token, ), ) ); + $started_metadata = is_array( $started['metadata'] ?? null ) ? agents_runtime_package_run_string_keyed_array( $started['metadata'] ) : array(); + if ( $claim_token !== agents_runtime_package_run_string( $started_metadata['_claim_token'] ?? '' ) ) { + if ( agents_runtime_package_run_is_terminal_status( $started['status'] ?? '' ) ) { + return agents_runtime_package_run_project_authoritative( $started ); + } + + do_action( 'agents_runtime_package_run_dispatch_failed', 'already_started', $input ); + return new \WP_Error( 'agents_runtime_package_run_already_started', 'The run_id has already been claimed for execution.' ); + } /** * Filters the runtime package execution handler. @@ -249,15 +260,67 @@ function agents_runtime_package_run_dispatch( array $input ) { 'started_at' => WP_Agent_Run_Control::now(), ) ); - if ( WP_Agent_Run_Control::STATUS_CANCELLED === ( $authoritative['status'] ?? '' ) ) { - $normalized['status'] = WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED; - $normalized['error'] = array( + return agents_runtime_package_run_project_authoritative( $authoritative, $normalized ); +} + +function agents_runtime_package_run_is_terminal_status( mixed $status ): bool { + return in_array( + WP_Agent_Run_Control::normalize_status( $status ), + array( + WP_Agent_Run_Control::STATUS_COMPLETED, + WP_Agent_Run_Control::STATUS_SUCCEEDED, + WP_Agent_Run_Control::STATUS_FAILED, + WP_Agent_Run_Control::STATUS_CANCELLED, + WP_Agent_Run_Control::STATUS_BUDGET_EXCEEDED, + WP_Agent_Run_Control::STATUS_STALLED, + WP_Agent_Run_Control::STATUS_INTERRUPTED, + ), + true + ); +} + +/** + * Project the authoritative run-control row into a runtime-package result. + * + * @param array $authoritative Stored run-control winner. + * @param array|null $candidate Handler result, when execution occurred. + * @return array + */ +function agents_runtime_package_run_project_authoritative( array $authoritative, ?array $candidate = null ): array { + $status = WP_Agent_Run_Control::normalize_status( $authoritative['status'] ?? '' ); + if ( WP_Agent_Run_Control::STATUS_CANCELLED === $status ) { + $runtime_status = WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED; + } elseif ( in_array( $status, array( WP_Agent_Run_Control::STATUS_FAILED, WP_Agent_Run_Control::STATUS_BUDGET_EXCEEDED, WP_Agent_Run_Control::STATUS_STALLED, WP_Agent_Run_Control::STATUS_INTERRUPTED ), true ) ) { + $runtime_status = WP_Agent_Runtime_Package_Run_Result::STATUS_FAILED; + } elseif ( in_array( $status, array( WP_Agent_Run_Control::STATUS_SUCCEEDED, WP_Agent_Run_Control::STATUS_COMPLETED ), true ) ) { + $runtime_status = WP_Agent_Runtime_Package_Run_Result::STATUS_SUCCEEDED; + } else { + $runtime_status = WP_Agent_Runtime_Package_Run_Result::STATUS_RUNNING; + } + + $projected = $candidate ?? WP_Agent_Runtime_Package_Run_Result::from_array( + array( + 'run_id' => $authoritative['run_id'] ?? '', + 'status' => $runtime_status, + 'metadata' => $authoritative['metadata'] ?? array(), + ) + )->to_array(); + $projected['run_id'] = $authoritative['run_id'] ?? ''; + $projected['status'] = $runtime_status; + + if ( WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED === $runtime_status ) { + $projected['error'] = array( 'code' => 'cancel_requested', 'message' => 'Runtime package run cancellation was requested.', ); + } elseif ( WP_Agent_Runtime_Package_Run_Result::STATUS_FAILED === $runtime_status && empty( $projected['error'] ) ) { + $projected['error'] = array( + 'code' => 'agents_runtime_package_run_failed', + 'message' => 'The runtime package run failed.', + ); } - return $normalized; + return $projected; } /** diff --git a/tests/runtime-package-run-contract-smoke.php b/tests/runtime-package-run-contract-smoke.php index 6a11a47..d7c8b16 100644 --- a/tests/runtime-package-run-contract-smoke.php +++ b/tests/runtime-package-run-contract-smoke.php @@ -225,6 +225,7 @@ function wp_get_ability( string $ability ): ?WP_Ability { final class Runtime_Package_Late_Cancel_Store implements WP_Agent_Atomic_Run_Control_Store { public bool $cancel_before_next_mutation = false; + public ?string $terminal_before_next_mutation = null; /** @var array>,queues:array>>,events:array>>}> */ private array $states = array(); @@ -238,7 +239,15 @@ public function save_state( string $store_key, array $state ): void { public function mutate_state( string $store_key, callable $mutation ): mixed { $state = $this->get_state( $store_key ); - if ( $this->cancel_before_next_mutation ) { + if ( null !== $this->terminal_before_next_mutation ) { + $status = $this->terminal_before_next_mutation; + $this->terminal_before_next_mutation = null; + foreach ( $state['runs'] as &$run ) { + $run['status'] = $status; + $run['cancelled'] = WP_Agent_Run_Control::STATUS_CANCELLED === $status; + } + unset( $run ); + } elseif ( $this->cancel_before_next_mutation ) { $this->cancel_before_next_mutation = false; foreach ( $state['runs'] as &$run ) { $run['status'] = WP_Agent_Run_Control::STATUS_CANCELLING; @@ -416,7 +425,71 @@ static function ( $handler, WP_Agent_Runtime_Package_Run_Request $handler_reques agents_api_smoke_assert_equals( 'succeeded', is_array( $helper_dispatch ) ? $helper_dispatch['status'] ?? '' : '', 'public helper preserves result status', $failures, $passes ); agents_api_smoke_assert_equals( 'build-site', is_array( $helper_dispatch ) ? $helper_dispatch['result']['workflow_id'] ?? '' : '', 'public helper passes workflow to handler', $failures, $passes ); -echo "\n[5] Atomic cancellation wins after the handler returns success:\n"; +echo "\n[5] Reused run IDs are exact-once and terminal-idempotent:\n"; +$GLOBALS['__agents_api_smoke_actions']['wp_agent_runtime_package_run_handler'] = array(); +$reuse_store = new Runtime_Package_Late_Cancel_Store(); +$reuse_effects = 0; +WP_Agent_Run_Control::set_store( $reuse_store ); +add_filter( + 'wp_agent_runtime_package_run_handler', + static function () use ( &$reuse_effects ): callable { + return static function () use ( &$reuse_effects ): array { + ++$reuse_effects; + return array( 'status' => WP_Agent_Runtime_Package_Run_Result::STATUS_SUCCEEDED ); + }; + } +); + +foreach ( + array( + WP_Agent_Run_Control::STATUS_SUCCEEDED => WP_Agent_Runtime_Package_Run_Result::STATUS_SUCCEEDED, + WP_Agent_Run_Control::STATUS_FAILED => WP_Agent_Runtime_Package_Run_Result::STATUS_FAILED, + WP_Agent_Run_Control::STATUS_CANCELLED => WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED, + ) as $stored_status => $response_status +) { + $reused_run_id = 'runtime-reused-' . $stored_status; + WP_Agent_Run_Control::save_run( + AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, + array( + 'run_id' => $reused_run_id, + 'status' => $stored_status, + 'metadata' => array( + 'package' => array( 'slug' => 'site-builder' ), + 'workflow' => array( 'id' => 'reused-terminal' ), + ), + ) + ); + $effects_before = $reuse_effects; + $reused = AgentsAPI\AI\agents_runtime_package_run_dispatch( + array( + 'run_id' => $reused_run_id, + 'package' => array( 'slug' => 'site-builder' ), + 'workflow' => array( 'id' => 'duplicate-must-not-run' ), + ) + ); + $reused_stored = WP_Agent_Run_Control::get_run( AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, $reused_run_id ); + agents_api_smoke_assert_equals( $effects_before, $reuse_effects, "reused {$stored_status} run skips duplicate handler effects", $failures, $passes ); + agents_api_smoke_assert_equals( $response_status, is_array( $reused ) ? $reused['status'] ?? '' : '', "reused {$stored_status} run returns its authoritative status", $failures, $passes ); + agents_api_smoke_assert_equals( $stored_status, $reused_stored['status'] ?? '', "reused {$stored_status} response leaves stored winner unchanged", $failures, $passes ); + agents_api_smoke_assert_equals( $reused_run_id, is_array( $reused ) ? $reused['run_id'] ?? '' : '', "reused {$stored_status} response preserves run identity", $failures, $passes ); +} + +WP_Agent_Run_Control::start_run( AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, 'runtime-reused-running' ); +$effects_before_active = $reuse_effects; +$active_reuse = AgentsAPI\AI\agents_runtime_package_run_dispatch( + array( + 'run_id' => 'runtime-reused-running', + 'package' => array( 'slug' => 'site-builder' ), + 'workflow' => array( 'id' => 'duplicate-must-not-run' ), + ) +); +$active_stored = WP_Agent_Run_Control::get_run( AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, 'runtime-reused-running' ); +agents_api_smoke_assert_equals( true, is_wp_error( $active_reuse ), 'active duplicate run is rejected before handler execution', $failures, $passes ); +agents_api_smoke_assert_equals( 'agents_runtime_package_run_already_started', is_wp_error( $active_reuse ) ? $active_reuse->get_error_code() : '', 'active duplicate uses the exact-once already-started error', $failures, $passes ); +agents_api_smoke_assert_equals( $effects_before_active, $reuse_effects, 'active duplicate run causes no handler effects', $failures, $passes ); +agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_RUNNING, $active_stored['status'] ?? '', 'active duplicate leaves the stored running owner unchanged', $failures, $passes ); + +echo "\n[6] Atomic cancellation wins after the handler returns success:\n"; $GLOBALS['__agents_api_smoke_actions']['wp_agent_runtime_package_run_handler'] = array(); $late_cancel_store = new Runtime_Package_Late_Cancel_Store(); WP_Agent_Run_Control::set_store( $late_cancel_store ); @@ -446,4 +519,30 @@ static function () use ( $late_cancel_store ): callable { agents_api_smoke_assert_equals( 'handler-success', is_array( $late_cancel_dispatch ) ? $late_cancel_dispatch['result']['candidate'] ?? '' : '', 'cancellation projection preserves handler evidence', $failures, $passes ); WP_Agent_Run_Control::reset_store(); +echo "\n[7] Post-handler projection follows non-cancellation terminal winners:\n"; +$GLOBALS['__agents_api_smoke_actions']['wp_agent_runtime_package_run_handler'] = array(); +$late_failure_store = new Runtime_Package_Late_Cancel_Store(); +WP_Agent_Run_Control::set_store( $late_failure_store ); +add_filter( + 'wp_agent_runtime_package_run_handler', + static function () use ( $late_failure_store ): callable { + return static function () use ( $late_failure_store ): array { + $late_failure_store->terminal_before_next_mutation = WP_Agent_Run_Control::STATUS_FAILED; + return array( 'status' => WP_Agent_Runtime_Package_Run_Result::STATUS_SUCCEEDED ); + }; + } +); +$late_failure_dispatch = AgentsAPI\AI\agents_runtime_package_run_dispatch( + array( + 'run_id' => 'runtime-late-failure', + 'package' => array( 'slug' => 'site-builder' ), + 'workflow' => array( 'id' => 'late-failure' ), + ) +); +$late_failure_run = WP_Agent_Run_Control::get_run( AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, 'runtime-late-failure' ); +agents_api_smoke_assert_equals( WP_Agent_Runtime_Package_Run_Result::STATUS_FAILED, is_array( $late_failure_dispatch ) ? $late_failure_dispatch['status'] ?? '' : '', 'dispatcher projects an authoritative failed winner after handler success', $failures, $passes ); +agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_FAILED, $late_failure_run['status'] ?? '', 'late failed winner remains authoritative in storage', $failures, $passes ); +agents_api_smoke_assert_equals( 'agents_runtime_package_run_failed', is_array( $late_failure_dispatch ) ? $late_failure_dispatch['error']['code'] ?? '' : '', 'late failed winner receives a stable projected error', $failures, $passes ); +WP_Agent_Run_Control::reset_store(); + agents_api_smoke_finish( 'Agents API runtime package run contract', $failures, $passes ); From 7c003247e80e9f10cb951844484040e36373705e Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 22:29:27 +0000 Subject: [PATCH 5/6] fix: centralize authoritative terminal outcomes --- .../register-runtime-package-run-ability.php | 23 ++++-- ...kflow-action-scheduler-branch-executor.php | 2 +- .../class-wp-agent-workflow-runner.php | 50 ++++++++---- tests/runtime-package-run-contract-smoke.php | 50 ++++++++++++ tests/workflow-as-branch-smoke.php | 80 +++++++++++++++++++ tests/workflow-runner-smoke.php | 32 +++++++- 6 files changed, 211 insertions(+), 26 deletions(-) diff --git a/src/Runtime/register-runtime-package-run-ability.php b/src/Runtime/register-runtime-package-run-ability.php index bc96ebc..b6bbfd8 100644 --- a/src/Runtime/register-runtime-package-run-ability.php +++ b/src/Runtime/register-runtime-package-run-ability.php @@ -218,30 +218,27 @@ function agents_runtime_package_run_dispatch( array $input ) { */ $handler = apply_filters( 'wp_agent_runtime_package_run_handler', null, $request, $input ); if ( ! is_callable( $handler ) ) { - WP_Agent_Run_Control::finish_run( AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, $run_id, WP_Agent_Run_Control::STATUS_FAILED ); do_action( 'agents_runtime_package_run_dispatch_failed', 'no_handler', $input ); - return new \WP_Error( + return agents_runtime_package_run_finalize_failure( $run_id, new \WP_Error( 'agents_runtime_package_run_no_handler', 'No agents/run-runtime-package handler is registered. Install a consumer runtime or add a callable to the wp_agent_runtime_package_run_handler filter.' - ); + ) ); } $result = call_user_func( $handler, $request, $input ); if ( is_wp_error( $result ) ) { - WP_Agent_Run_Control::finish_run( AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, $run_id, WP_Agent_Run_Control::STATUS_FAILED ); do_action( 'agents_runtime_package_run_dispatch_failed', $result->get_error_code(), $input ); - return $result; + return agents_runtime_package_run_finalize_failure( $run_id, $result ); } if ( $result instanceof WP_Agent_Runtime_Package_Run_Result ) { $result = $result->to_array(); } elseif ( ! is_array( $result ) ) { - WP_Agent_Run_Control::finish_run( AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, $run_id, WP_Agent_Run_Control::STATUS_FAILED ); do_action( 'agents_runtime_package_run_dispatch_failed', 'invalid_result', $input ); - return new \WP_Error( + return agents_runtime_package_run_finalize_failure( $run_id, new \WP_Error( 'agents_runtime_package_run_invalid_result', 'agents/run-runtime-package handlers must return an array, WP_Agent_Runtime_Package_Run_Result, or WP_Error.' - ); + ) ); } $result = agents_runtime_package_run_string_keyed_array( $result ); @@ -279,6 +276,16 @@ function agents_runtime_package_run_is_terminal_status( mixed $status ): bool { ); } +/** @return array|\WP_Error */ +function agents_runtime_package_run_finalize_failure( string $run_id, \WP_Error $failure ) { + $authoritative = WP_Agent_Run_Control::finish_run( AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, $run_id, WP_Agent_Run_Control::STATUS_FAILED ); + if ( null === $authoritative || WP_Agent_Run_Control::STATUS_FAILED === ( $authoritative['status'] ?? '' ) ) { + return $failure; + } + + return agents_runtime_package_run_project_authoritative( $authoritative ); +} + /** * Project the authoritative run-control row into a runtime-package result. * 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 c4f7b94..c94436d 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 @@ -1203,6 +1203,7 @@ static function () use ( $recorder, $run_id, $handle_id, $message, $code ) { 'metadata' => $metadata, ) ); + $terminal = WP_Agent_Workflow_Runner::authoritative_terminal_result( $terminal ); $updated = $recorder->update( $terminal ); return is_wp_error( $updated ) ? $updated : array( 'won' => true, 'terminal' => $terminal ); }, @@ -1212,7 +1213,6 @@ static function () use ( $recorder, $run_id, $handle_id, $message, $code ) { return false; } $terminal = $transition['terminal']; - \AgentsAPI\AI\WP_Agent_Run_Control::finish_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, $run_id, \AgentsAPI\AI\WP_Agent_Run_Control::STATUS_FAILED ); do_action( 'wp_agent_workflow_run_completed', $terminal, $run_id ); WP_Agent_Workflow_Branch_Store::forget_run( $run_id ); return true; diff --git a/src/Workflows/class-wp-agent-workflow-runner.php b/src/Workflows/class-wp-agent-workflow-runner.php index 8dbe5e3..eff1aad 100644 --- a/src/Workflows/class-wp-agent-workflow-runner.php +++ b/src/Workflows/class-wp-agent-workflow-runner.php @@ -185,10 +185,11 @@ public function run( WP_Agent_Workflow_Spec $spec, array $inputs = array(), arra 'ended_at' => time(), ) ); + $terminal = self::authoritative_terminal_result( $terminal ); if ( $this->recorder ) { $this->recorder->update( $terminal ); } - WP_Agent_Run_Control::finish_run( self::RUN_CONTROL_STORE, $result->get_run_id(), WP_Agent_Run_Control::STATUS_FAILED ); + do_action( 'wp_agent_workflow_run_completed', $terminal, $terminal->get_run_id() ); return $terminal; } @@ -314,10 +315,11 @@ private function run_step_loop( WP_Agent_Workflow_Spec $spec, WP_Agent_Workflow_ $step = $steps[ $step_index ]; if ( self::is_cancel_requested( $result->get_run_id() ) ) { - $result = self::cancelled_result( $result, $step_records ); + $result = self::authoritative_terminal_result( self::cancelled_result( $result, $step_records ) ); if ( $this->recorder ) { $this->recorder->update( $result ); } + do_action( 'wp_agent_workflow_run_completed', $result, $result->get_run_id() ); return $result; } @@ -325,10 +327,11 @@ private function run_step_loop( WP_Agent_Workflow_Spec $spec, WP_Agent_Workflow_ $step_records[] = $record; if ( self::is_cancel_requested( $result->get_run_id() ) ) { - $result = self::cancelled_result( $result, $step_records ); + $result = self::authoritative_terminal_result( self::cancelled_result( $result, $step_records ) ); if ( $this->recorder ) { $this->recorder->update( $result ); } + do_action( 'wp_agent_workflow_run_completed', $result, $result->get_run_id() ); return $result; } @@ -400,12 +403,7 @@ private function run_step_loop( WP_Agent_Workflow_Spec $spec, WP_Agent_Workflow_ ) ); - $authoritative = WP_Agent_Run_Control::finish_run( - self::RUN_CONTROL_STORE, - $result->get_run_id(), - $failed ? WP_Agent_Run_Control::STATUS_FAILED : WP_Agent_Run_Control::STATUS_SUCCEEDED - ); - $result = self::project_authoritative_terminal( $result, $authoritative ); + $result = self::authoritative_terminal_result( $result ); if ( $this->recorder ) { $this->recorder->update( $result ); @@ -517,20 +515,40 @@ private static function is_cancel_requested( string $run_id ): bool { * @param array $step_records Step records completed before cancellation was observed. */ private static function cancelled_result( WP_Agent_Workflow_Run_Result $result, array $step_records ): WP_Agent_Workflow_Run_Result { - WP_Agent_Run_Control::finish_run( self::RUN_CONTROL_STORE, $result->get_run_id(), WP_Agent_Run_Control::STATUS_CANCELLED ); - return self::project_cancelled_result( $result, $step_records ); } /** - * Project the serialized run-control winner into the workflow result. - * - * @param array|null $authoritative Authoritative run-control row. + * Commit and project one authoritative terminal workflow outcome. */ - private static function project_authoritative_terminal( WP_Agent_Workflow_Run_Result $result, ?array $authoritative ): WP_Agent_Workflow_Run_Result { - if ( null !== $authoritative && WP_Agent_Run_Control::STATUS_CANCELLED === ( $authoritative['status'] ?? '' ) ) { + public static function authoritative_terminal_result( WP_Agent_Workflow_Run_Result $result ): WP_Agent_Workflow_Run_Result { + $status = WP_Agent_Workflow_Run_Result::STATUS_CANCELLED === $result->get_status() + ? WP_Agent_Run_Control::STATUS_CANCELLED + : ( WP_Agent_Workflow_Run_Result::STATUS_FAILED === $result->get_status() ? WP_Agent_Run_Control::STATUS_FAILED : WP_Agent_Run_Control::STATUS_SUCCEEDED ); + $authoritative = WP_Agent_Run_Control::finish_run( self::RUN_CONTROL_STORE, $result->get_run_id(), $status ); + if ( null === $authoritative ) { + return $result; + } + + $stored_status = WP_Agent_Run_Control::normalize_status( $authoritative['status'] ?? '' ); + if ( WP_Agent_Run_Control::STATUS_CANCELLED === $stored_status ) { return self::project_cancelled_result( $result, $result->get_steps() ); } + if ( WP_Agent_Run_Control::STATUS_FAILED === $stored_status ) { + $error = $result->get_error(); + return $result->with( + array( + 'status' => WP_Agent_Workflow_Run_Result::STATUS_FAILED, + 'error' => array() !== $error ? $error : array( + 'code' => 'workflow_run_failed', + 'message' => 'The workflow run failed.', + ), + ) + ); + } + if ( in_array( $stored_status, array( WP_Agent_Run_Control::STATUS_SUCCEEDED, WP_Agent_Run_Control::STATUS_COMPLETED ), true ) ) { + return $result->with( array( 'status' => WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, 'error' => array() ) ); + } return $result; } diff --git a/tests/runtime-package-run-contract-smoke.php b/tests/runtime-package-run-contract-smoke.php index d7c8b16..85683df 100644 --- a/tests/runtime-package-run-contract-smoke.php +++ b/tests/runtime-package-run-contract-smoke.php @@ -545,4 +545,54 @@ static function () use ( $late_failure_store ): callable { agents_api_smoke_assert_equals( 'agents_runtime_package_run_failed', is_array( $late_failure_dispatch ) ? $late_failure_dispatch['error']['code'] ?? '' : '', 'late failed winner receives a stable projected error', $failures, $passes ); WP_Agent_Run_Control::reset_store(); +echo "\n[8] Every error exit projects a concurrent cancellation winner:\n"; +foreach ( array( 'no_handler', 'handler_error', 'invalid_result' ) as $error_exit ) { + $GLOBALS['__agents_api_smoke_actions']['wp_agent_runtime_package_run_handler'] = array(); + $error_store = new Runtime_Package_Late_Cancel_Store(); + WP_Agent_Run_Control::set_store( $error_store ); + if ( 'no_handler' === $error_exit ) { + add_filter( + 'wp_agent_runtime_package_run_handler', + static function ( $handler ) use ( $error_store ) { + $error_store->cancel_before_next_mutation = true; + return $handler; + } + ); + } elseif ( 'handler_error' === $error_exit ) { + add_filter( + 'wp_agent_runtime_package_run_handler', + static function () use ( $error_store ): callable { + return static function () use ( $error_store ): WP_Error { + $error_store->cancel_before_next_mutation = true; + return new WP_Error( 'runtime_handler_failed', 'handler failed' ); + }; + } + ); + } else { + add_filter( + 'wp_agent_runtime_package_run_handler', + static function () use ( $error_store ): callable { + return static function () use ( $error_store ): string { + $error_store->cancel_before_next_mutation = true; + return 'invalid'; + }; + } + ); + } + + $error_run_id = 'runtime-cancel-' . $error_exit; + $error_result = AgentsAPI\AI\agents_runtime_package_run_dispatch( + array( + 'run_id' => $error_run_id, + 'package' => array( 'slug' => 'site-builder' ), + 'workflow' => array( 'id' => $error_exit ), + ) + ); + $error_stored = WP_Agent_Run_Control::get_run( AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, $error_run_id ); + agents_api_smoke_assert_equals( WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED, is_array( $error_result ) ? $error_result['status'] ?? '' : '', "{$error_exit} exit returns the concurrent cancellation winner", $failures, $passes ); + agents_api_smoke_assert_equals( 'cancel_requested', is_array( $error_result ) ? $error_result['error']['code'] ?? '' : '', "{$error_exit} exit projects the canonical cancellation error", $failures, $passes ); + agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_CANCELLED, $error_stored['status'] ?? '', "{$error_exit} exit matches authoritative storage", $failures, $passes ); +} +WP_Agent_Run_Control::reset_store(); + agents_api_smoke_finish( 'Agents API runtime package run contract', $failures, $passes ); diff --git a/tests/workflow-as-branch-smoke.php b/tests/workflow-as-branch-smoke.php index aa0a03c..f8e1da5 100644 --- a/tests/workflow-as-branch-smoke.php +++ b/tests/workflow-as-branch-smoke.php @@ -340,6 +340,8 @@ function smoke_assert_true( $actual, string $name, array &$failures, int &$passe use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Runner; use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Spec; use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Action_Scheduler_Branch_Executor; +use AgentsAPI\AI\WP_Agent_Atomic_Run_Control_Store; +use AgentsAPI\AI\WP_Agent_Run_Control; // ── A durable, reloadable in-memory recorder ───────────────────────────────── // The frame lives in metadata._suspension inside the serialized row — there is @@ -394,6 +396,35 @@ public function fail_next_update(): void { } } +final class AS_Late_Cancel_Run_Control_Store implements WP_Agent_Atomic_Run_Control_Store { + public bool $cancel_before_next_mutation = false; + /** @var array>,queues:array>>,events:array>>}> */ + private array $states = array(); + + public function get_state( string $store_key ): array { + return $this->states[ $store_key ] ?? array( 'runs' => array(), 'queues' => array(), 'events' => array() ); + } + + public function save_state( string $store_key, array $state ): void { + $this->states[ $store_key ] = $state; + } + + public function mutate_state( string $store_key, callable $mutation ): mixed { + $state = $this->get_state( $store_key ); + if ( $this->cancel_before_next_mutation ) { + $this->cancel_before_next_mutation = false; + foreach ( $state['runs'] as &$run ) { + $run['status'] = WP_Agent_Run_Control::STATUS_CANCELLING; + $run['cancelled'] = true; + } + unset( $run ); + } + $mutated = $mutation( $state ); + $this->states[ $store_key ] = $mutated['state']; + return $mutated['result']; + } +} + // ── Abilities: aggregator + sequential consumer + a real role worker ───────── // The AS executor RUNS branches for real (unlike the Phase 1 FakeExecutor), so // the role worker must produce a real fragment the aggregate consumes. @@ -1192,6 +1223,55 @@ static function ( bool $handled ) use ( &$forget_calls8 ): bool { remove_all_filters( 'wp_agent_workflow_run_completed' ); remove_all_filters( 'wp_agent_workflow_branch_store_forget' ); +// A cancellation committed inside reconcile-recovery finish must replace the +// candidate failure before recorder publication and completion notification. +AS_Shim::reset(); +$GLOBALS['__options'] = array(); +$reconcile_cancel_recorder = new AS_Smoke_Recorder(); +$reconcile_cancel_store = new AS_Late_Cancel_Run_Control_Store(); +WP_Agent_Run_Control::set_store( $reconcile_cancel_store ); +remove_all_filters( 'wp_agent_workflow_run_recorder' ); +add_filter( 'wp_agent_workflow_run_recorder', static function () use ( $reconcile_cancel_recorder ) { return $reconcile_cancel_recorder; } ); +( new WP_Agent_Workflow_Runner( $reconcile_cancel_recorder ) )->run( as_smoke_roles_spec(), array(), array( 'run_id' => 'as-recovery-cancel' ) ); +$reconcile_cancel_branches = AS_Shim::actions_for( WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK ); +$reconcile_cancel_attempts = 0; +add_filter( + 'wp_agent_workflow_reconcile_lock', + static function ( $override, string $run_id, callable $critical ) use ( &$reconcile_cancel_attempts ) { + unset( $override ); + if ( 'as-recovery-cancel' === $run_id && $reconcile_cancel_attempts++ < 2 ) { + return new WP_Error( 'agents_reconcile_lock_unavailable', 'contended' ); + } + return $critical(); + }, + 10, + 3 +); +$reconcile_cancel_events = array(); +add_action( + 'wp_agent_workflow_run_completed', + static function ( WP_Agent_Workflow_Run_Result $result, string $run_id ) use ( &$reconcile_cancel_events ): void { + if ( 'as-recovery-cancel' === $run_id ) { + $reconcile_cancel_events[] = $result; + } + }, + 10, + 2 +); +$reconcile_cancel_store->cancel_before_next_mutation = true; +AS_Shim::$reject_hook = WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RECONCILE_HOOK; +AS_Shim::fire_with_failure_lifecycle( $reconcile_cancel_branches[0]['id'] ); +AS_Shim::$reject_hook = ''; +$reconcile_cancel_terminal = $reconcile_cancel_recorder->find( 'as-recovery-cancel' ); +$reconcile_cancel_stored = WP_Agent_Run_Control::get_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, 'as-recovery-cancel' ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, $reconcile_cancel_terminal->get_status(), 'reconcile recovery recorder receives the atomic cancellation winner', $failures, $passes ); +smoke_assert( 'cancel_requested', $reconcile_cancel_terminal->get_error()['code'] ?? '', 'reconcile recovery projects the canonical cancellation error', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, isset( $reconcile_cancel_events[0] ) ? $reconcile_cancel_events[0]->get_status() : '', 'reconcile recovery completion hook receives the cancellation winner', $failures, $passes ); +smoke_assert( WP_Agent_Run_Control::STATUS_CANCELLED, $reconcile_cancel_stored['status'] ?? '', 'reconcile recovery publication matches authoritative storage', $failures, $passes ); +remove_all_filters( 'wp_agent_workflow_reconcile_lock' ); +remove_all_filters( 'wp_agent_workflow_run_completed' ); +WP_Agent_Run_Control::reset_store(); + // A RECONCILE_HOOK action can itself fail while handing off another contended // attempt. Its failed-action lifecycle must recover the receipt directly. AS_Shim::reset(); diff --git a/tests/workflow-runner-smoke.php b/tests/workflow-runner-smoke.php index 74c2322..1a32291 100644 --- a/tests/workflow-runner-smoke.php +++ b/tests/workflow-runner-smoke.php @@ -244,6 +244,7 @@ public function save_state( string $store_key, array $state ): void { class Workflow_Late_Cancel_Store implements WP_Agent_Atomic_Run_Control_Store { public bool $cancel_before_next_mutation = false; + public int $mutations_until_cancel = 0; /** @var array>,queues:array>>,events:array>>}> */ private array $states = array(); @@ -257,7 +258,12 @@ public function save_state( string $store_key, array $state ): void { public function mutate_state( string $store_key, callable $mutation ): mixed { $state = $this->get_state( $store_key ); - if ( $this->cancel_before_next_mutation ) { + $cancel = $this->cancel_before_next_mutation; + if ( $this->mutations_until_cancel > 0 ) { + --$this->mutations_until_cancel; + $cancel = 0 === $this->mutations_until_cancel; + } + if ( $cancel ) { $this->cancel_before_next_mutation = false; foreach ( $state['runs'] as &$run ) { $run['status'] = WP_Agent_Run_Control::STATUS_CANCELLING; @@ -633,6 +639,30 @@ static function ( WP_Agent_Workflow_Run_Result $completed, string $run_id ) use smoke_assert( 'missing_required_input', $result4->get_error()['code'], 'input error has expected code', $failures, $passes ); smoke_assert( 0, count( $result4->get_steps() ), 'no steps run when input validation fails', $failures, $passes ); +$input_cancel_store = new Workflow_Late_Cancel_Store(); +$input_cancel_store->mutations_until_cancel = 2; +$input_cancel_hook = null; +WP_Agent_Run_Control::set_store( $input_cancel_store ); +add_action( + 'wp_agent_workflow_run_completed', + static function ( WP_Agent_Workflow_Run_Result $completed, string $run_id ) use ( &$input_cancel_hook ): void { + if ( 'input-cancel-run' === $run_id ) { + $input_cancel_hook = $completed; + } + }, + 10, + 2 +); +$input_cancel_recorder = new Capture_Recorder(); +$input_cancel_result = ( new WP_Agent_Workflow_Runner( $input_cancel_recorder ) )->run( $spec, array(), array( 'run_id' => 'input-cancel-run' ) ); +$input_cancel_record = end( $input_cancel_recorder->writes ); +$input_cancel_stored = WP_Agent_Run_Control::get_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, 'input-cancel-run' ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, $input_cancel_result->get_status(), 'input-validation finish returns a concurrent cancellation winner', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, $input_cancel_record['status'] ?? '', 'input-validation recorder receives the cancellation winner', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, $input_cancel_hook instanceof WP_Agent_Workflow_Run_Result ? $input_cancel_hook->get_status() : '', 'input-validation completion hook receives the cancellation winner', $failures, $passes ); +smoke_assert( WP_Agent_Run_Control::STATUS_CANCELLED, $input_cancel_stored['status'] ?? '', 'input-validation response matches authoritative storage', $failures, $passes ); +WP_Agent_Run_Control::reset_store(); + // ─── Unknown step type with no handler ─────────────────────────────── $martian = WP_Agent_Workflow_Spec::from_array( From 7e455b99491189dd778bc920469477e263bf6209 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Wed, 26 Aug 2026 22:40:08 +0000 Subject: [PATCH 6/6] fix: project every terminal run outcome --- src/Runtime/class-wp-agent-run-control.php | 3 + .../register-runtime-package-run-ability.php | 3 + .../class-wp-agent-workflow-runner.php | 97 +++++++------ tests/runtime-package-run-contract-smoke.php | 33 +++++ tests/workflow-runner-smoke.php | 129 +++++++++++++++++- 5 files changed, 221 insertions(+), 44 deletions(-) diff --git a/src/Runtime/class-wp-agent-run-control.php b/src/Runtime/class-wp-agent-run-control.php index 96de0cc..315571e 100644 --- a/src/Runtime/class-wp-agent-run-control.php +++ b/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, @@ -530,6 +532,7 @@ private static function is_terminal_status( mixed $status ): bool { self::STATUS_COMPLETED, self::STATUS_SUCCEEDED, self::STATUS_FAILED, + self::STATUS_SKIPPED, self::STATUS_CANCELLED, self::STATUS_BUDGET_EXCEEDED, self::STATUS_STALLED, diff --git a/src/Runtime/register-runtime-package-run-ability.php b/src/Runtime/register-runtime-package-run-ability.php index b6bbfd8..b605ef5 100644 --- a/src/Runtime/register-runtime-package-run-ability.php +++ b/src/Runtime/register-runtime-package-run-ability.php @@ -267,6 +267,7 @@ function agents_runtime_package_run_is_terminal_status( mixed $status ): bool { WP_Agent_Run_Control::STATUS_COMPLETED, WP_Agent_Run_Control::STATUS_SUCCEEDED, WP_Agent_Run_Control::STATUS_FAILED, + WP_Agent_Run_Control::STATUS_SKIPPED, WP_Agent_Run_Control::STATUS_CANCELLED, WP_Agent_Run_Control::STATUS_BUDGET_EXCEEDED, WP_Agent_Run_Control::STATUS_STALLED, @@ -297,6 +298,8 @@ function agents_runtime_package_run_project_authoritative( array $authoritative, $status = WP_Agent_Run_Control::normalize_status( $authoritative['status'] ?? '' ); if ( WP_Agent_Run_Control::STATUS_CANCELLED === $status ) { $runtime_status = WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED; + } elseif ( WP_Agent_Run_Control::STATUS_SKIPPED === $status ) { + $runtime_status = WP_Agent_Runtime_Package_Run_Result::STATUS_SKIPPED; } elseif ( in_array( $status, array( WP_Agent_Run_Control::STATUS_FAILED, WP_Agent_Run_Control::STATUS_BUDGET_EXCEEDED, WP_Agent_Run_Control::STATUS_STALLED, WP_Agent_Run_Control::STATUS_INTERRUPTED ), true ) ) { $runtime_status = WP_Agent_Runtime_Package_Run_Result::STATUS_FAILED; } elseif ( in_array( $status, array( WP_Agent_Run_Control::STATUS_SUCCEEDED, WP_Agent_Run_Control::STATUS_COMPLETED ), true ) ) { diff --git a/src/Workflows/class-wp-agent-workflow-runner.php b/src/Workflows/class-wp-agent-workflow-runner.php index eff1aad..fe7aba4 100644 --- a/src/Workflows/class-wp-agent-workflow-runner.php +++ b/src/Workflows/class-wp-agent-workflow-runner.php @@ -150,7 +150,7 @@ public function run( WP_Agent_Workflow_Spec $spec, array $inputs = array(), arra // Recorder unavailable on entry — return a failed result without // running steps. The caller still gets the in-memory record so // observability hooks fire; the step pipeline does not run. - return $result->with( + $terminal = $result->with( array( 'status' => WP_Agent_Workflow_Run_Result::STATUS_FAILED, 'error' => array( @@ -160,6 +160,8 @@ public function run( WP_Agent_Workflow_Spec $spec, array $inputs = array(), arra 'ended_at' => time(), ) ); + WP_Agent_Run_Control::start_run( self::RUN_CONTROL_STORE, $run_id, array( 'workflow_id' => $spec->get_id(), 'metadata' => $metadata ) ); + return $this->complete_terminal_result( $terminal, false ); } if ( '' !== $persisted ) { $result = $result->with( array( 'run_id' => $persisted ) ); @@ -185,12 +187,7 @@ public function run( WP_Agent_Workflow_Spec $spec, array $inputs = array(), arra 'ended_at' => time(), ) ); - $terminal = self::authoritative_terminal_result( $terminal ); - if ( $this->recorder ) { - $this->recorder->update( $terminal ); - } - do_action( 'wp_agent_workflow_run_completed', $terminal, $terminal->get_run_id() ); - return $terminal; + return $this->complete_terminal_result( $terminal ); } $context = new WP_Agent_Workflow_Run_Context( @@ -224,12 +221,12 @@ public function run( WP_Agent_Workflow_Spec $spec, array $inputs = array(), arra */ public function resume( string $run_id, array $options = array() ): WP_Agent_Workflow_Run_Result { if ( null === $this->recorder ) { - return self::resume_error_result( $run_id, 'workflow_resume_no_recorder', 'A recorder is required to resume a suspended run.' ); + return $this->resume_failure( null, $run_id, 'workflow_resume_no_recorder', 'A recorder is required to resume a suspended run.', false ); } $result = $this->recorder->find( $run_id ); if ( null === $result ) { - return self::resume_error_result( $run_id, 'workflow_resume_run_not_found', sprintf( 'No suspended run was found for run_id `%s`.', $run_id ) ); + return $this->resume_failure( null, $run_id, 'workflow_resume_run_not_found', sprintf( 'No suspended run was found for run_id `%s`.', $run_id ) ); } if ( ! $result->is_suspended() ) { // Idempotency guard: an already-resumed (or never-suspended) run is @@ -240,7 +237,7 @@ public function resume( string $run_id, array $options = array() ): WP_Agent_Wor $suspension = $result->get_suspension(); $spec = self::spec_from_result( $result ); if ( null === $spec ) { - return self::resume_error_result( $run_id, 'workflow_resume_spec_unavailable', 'The suspended run has no replayable spec snapshot to resume from.' ); + return $this->resume_failure( $result, $run_id, 'workflow_resume_spec_unavailable', 'The suspended run has no replayable spec snapshot to resume from.' ); } $snapshot = is_array( $suspension['context_snapshot'] ?? null ) ? $suspension['context_snapshot'] : array(); @@ -315,24 +312,14 @@ private function run_step_loop( WP_Agent_Workflow_Spec $spec, WP_Agent_Workflow_ $step = $steps[ $step_index ]; if ( self::is_cancel_requested( $result->get_run_id() ) ) { - $result = self::authoritative_terminal_result( self::cancelled_result( $result, $step_records ) ); - if ( $this->recorder ) { - $this->recorder->update( $result ); - } - do_action( 'wp_agent_workflow_run_completed', $result, $result->get_run_id() ); - return $result; + return $this->complete_terminal_result( self::cancelled_result( $result, $step_records ) ); } $record = self::string_keyed_array( $executor->execute( $step, $context ) ); $step_records[] = $record; if ( self::is_cancel_requested( $result->get_run_id() ) ) { - $result = self::authoritative_terminal_result( self::cancelled_result( $result, $step_records ) ); - if ( $this->recorder ) { - $this->recorder->update( $result ); - } - do_action( 'wp_agent_workflow_run_completed', $result, $result->get_run_id() ); - return $result; + return $this->complete_terminal_result( self::cancelled_result( $result, $step_records ) ); } // Pending / suspend gate — BEFORE the failure gate. The step asked @@ -403,30 +390,16 @@ private function run_step_loop( WP_Agent_Workflow_Spec $spec, WP_Agent_Workflow_ ) ); - $result = self::authoritative_terminal_result( $result ); + return $this->complete_terminal_result( $result ); + } - if ( $this->recorder ) { + /** Complete one terminal result after projecting the run-control winner. */ + private function complete_terminal_result( WP_Agent_Workflow_Run_Result $result, bool $update_recorder = true ): WP_Agent_Workflow_Run_Result { + $result = self::authoritative_terminal_result( $result ); + if ( $update_recorder && $this->recorder ) { $this->recorder->update( $result ); } - - /** - * Fires when a workflow run reaches a terminal state through the step loop. - * - * This is the single funnel for a run finishing — whether it ran straight - * through in one request ({@see run()}) or completed via an async resume - * after its parallel branches reconciled ({@see resume()}). It lets an - * async consumer react to completion WITHOUT block-polling the recorder: - * a consumer that dispatched an async fanout can return immediately from - * its own worker and do its finalization here instead, so it never holds a - * queue claim while waiting on the very branches it dispatched. - * - * @since 0.5.0 - * - * @param WP_Agent_Workflow_Run_Result $result The terminal run result. - * @param string $run_id The run id. - */ do_action( 'wp_agent_workflow_run_completed', $result, $result->get_run_id() ); - return $result; } @@ -506,6 +479,26 @@ private static function resume_error_result( string $run_id, string $code, strin ); } + private function resume_failure( ?WP_Agent_Workflow_Run_Result $result, string $run_id, string $code, string $message, bool $update_recorder = true ): WP_Agent_Workflow_Run_Result { + if ( null === $result ) { + $result = self::resume_error_result( $run_id, $code, $message ); + } else { + $metadata = $result->get_metadata(); + unset( $metadata['_suspension'] ); + $result = $result->with( + array( + 'status' => WP_Agent_Workflow_Run_Result::STATUS_FAILED, + 'error' => array( 'code' => $code, 'message' => $message ), + 'ended_at' => time(), + 'metadata' => $metadata, + ) + ); + } + + WP_Agent_Run_Control::start_run( self::RUN_CONTROL_STORE, $run_id, array( 'workflow_id' => $result->get_workflow_id() ) ); + return $this->complete_terminal_result( $result, $update_recorder ); + } + /** @phpstan-impure */ private static function is_cancel_requested( string $run_id ): bool { return WP_Agent_Run_Control::cancel_requested( self::RUN_CONTROL_STORE, $run_id ); @@ -546,6 +539,26 @@ public static function authoritative_terminal_result( WP_Agent_Workflow_Run_Resu ) ); } + $terminal_errors = array( + WP_Agent_Run_Control::STATUS_BUDGET_EXCEEDED => array( 'workflow_run_budget_exceeded', 'The workflow run exceeded its budget.' ), + WP_Agent_Run_Control::STATUS_STALLED => array( 'workflow_run_stalled', 'The workflow run stalled.' ), + WP_Agent_Run_Control::STATUS_INTERRUPTED => array( 'workflow_run_interrupted', 'The workflow run was interrupted.' ), + ); + if ( isset( $terminal_errors[ $stored_status ] ) ) { + return $result->with( + array( + 'status' => WP_Agent_Workflow_Run_Result::STATUS_FAILED, + 'error' => array( + 'code' => $terminal_errors[ $stored_status ][0], + 'message' => $terminal_errors[ $stored_status ][1], + ), + 'ended_at' => time(), + ) + ); + } + if ( WP_Agent_Run_Control::STATUS_SKIPPED === $stored_status ) { + return $result->with( array( 'status' => WP_Agent_Workflow_Run_Result::STATUS_SKIPPED, 'error' => array() ) ); + } if ( in_array( $stored_status, array( WP_Agent_Run_Control::STATUS_SUCCEEDED, WP_Agent_Run_Control::STATUS_COMPLETED ), true ) ) { return $result->with( array( 'status' => WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, 'error' => array() ) ); } diff --git a/tests/runtime-package-run-contract-smoke.php b/tests/runtime-package-run-contract-smoke.php index 85683df..4dabc5a 100644 --- a/tests/runtime-package-run-contract-smoke.php +++ b/tests/runtime-package-run-contract-smoke.php @@ -445,6 +445,7 @@ static function () use ( &$reuse_effects ): callable { WP_Agent_Run_Control::STATUS_SUCCEEDED => WP_Agent_Runtime_Package_Run_Result::STATUS_SUCCEEDED, WP_Agent_Run_Control::STATUS_FAILED => WP_Agent_Runtime_Package_Run_Result::STATUS_FAILED, WP_Agent_Run_Control::STATUS_CANCELLED => WP_Agent_Runtime_Package_Run_Result::STATUS_CANCELLED, + WP_Agent_Run_Control::STATUS_SKIPPED => WP_Agent_Runtime_Package_Run_Result::STATUS_SKIPPED, ) as $stored_status => $response_status ) { $reused_run_id = 'runtime-reused-' . $stored_status; @@ -595,4 +596,36 @@ static function () use ( $error_store ): callable { } WP_Agent_Run_Control::reset_store(); +echo "\n[9] Skipped handler outcomes are terminal and replay-safe:\n"; +$GLOBALS['__agents_api_smoke_actions']['wp_agent_runtime_package_run_handler'] = array(); +$skipped_store = new Runtime_Package_Late_Cancel_Store(); +$skipped_effects = 0; +WP_Agent_Run_Control::set_store( $skipped_store ); +add_filter( + 'wp_agent_runtime_package_run_handler', + static function () use ( &$skipped_effects ): callable { + return static function () use ( &$skipped_effects ): array { + ++$skipped_effects; + return array( + 'status' => WP_Agent_Runtime_Package_Run_Result::STATUS_SKIPPED, + 'result' => array( 'reason' => 'not_applicable' ), + ); + }; + } +); +$skipped_input = array( + 'run_id' => 'runtime-skipped-terminal', + 'package' => array( 'slug' => 'site-builder' ), + 'workflow' => array( 'id' => 'skip' ), +); +$skipped_first = AgentsAPI\AI\agents_runtime_package_run_dispatch( $skipped_input ); +$skipped_replay = AgentsAPI\AI\agents_runtime_package_run_dispatch( $skipped_input ); +$skipped_run = WP_Agent_Run_Control::get_run( AgentsAPI\AI\AGENTS_RUNTIME_PACKAGE_RUN_CONTROL_STORE, 'runtime-skipped-terminal' ); +agents_api_smoke_assert_equals( WP_Agent_Runtime_Package_Run_Result::STATUS_SKIPPED, is_array( $skipped_first ) ? $skipped_first['status'] ?? '' : '', 'handler skipped result remains skipped', $failures, $passes ); +agents_api_smoke_assert_equals( 'not_applicable', is_array( $skipped_first ) ? $skipped_first['result']['reason'] ?? '' : '', 'initial skipped result preserves handler payload', $failures, $passes ); +agents_api_smoke_assert_equals( WP_Agent_Run_Control::STATUS_SKIPPED, $skipped_run['status'] ?? '', 'skipped is terminal in authoritative run-control storage', $failures, $passes ); +agents_api_smoke_assert_equals( WP_Agent_Runtime_Package_Run_Result::STATUS_SKIPPED, is_array( $skipped_replay ) ? $skipped_replay['status'] ?? '' : '', 'skipped duplicate returns the stored terminal winner', $failures, $passes ); +agents_api_smoke_assert_equals( 1, $skipped_effects, 'skipped duplicate does not re-execute handler effects', $failures, $passes ); +WP_Agent_Run_Control::reset_store(); + agents_api_smoke_finish( 'Agents API runtime package run contract', $failures, $passes ); diff --git a/tests/workflow-runner-smoke.php b/tests/workflow-runner-smoke.php index 1a32291..8768915 100644 --- a/tests/workflow-runner-smoke.php +++ b/tests/workflow-runner-smoke.php @@ -245,6 +245,7 @@ public function save_state( string $store_key, array $state ): void { class Workflow_Late_Cancel_Store implements WP_Agent_Atomic_Run_Control_Store { public bool $cancel_before_next_mutation = false; public int $mutations_until_cancel = 0; + public ?string $terminal_before_next_mutation = null; /** @var array>,queues:array>>,events:array>>}> */ private array $states = array(); @@ -263,7 +264,15 @@ public function mutate_state( string $store_key, callable $mutation ): mixed { --$this->mutations_until_cancel; $cancel = 0 === $this->mutations_until_cancel; } - if ( $cancel ) { + if ( null !== $this->terminal_before_next_mutation ) { + $status = $this->terminal_before_next_mutation; + $this->terminal_before_next_mutation = null; + foreach ( $state['runs'] as &$run ) { + $run['status'] = $status; + $run['cancelled'] = false; + } + unset( $run ); + } elseif ( $cancel ) { $this->cancel_before_next_mutation = false; foreach ( $state['runs'] as &$run ) { $run['status'] = WP_Agent_Run_Control::STATUS_CANCELLING; @@ -324,6 +333,16 @@ static function (): array { return array( 'completed' => true ); } ); +workflow_runner_smoke_register_ability( + 'demo/arm-terminal-winner', + static function (): array { + $store = $GLOBALS['__workflow_terminal_store'] ?? null; + if ( $store instanceof Workflow_Late_Cancel_Store ) { + $store->terminal_before_next_mutation = (string) ( $GLOBALS['__workflow_terminal_status'] ?? '' ); + } + return array( 'completed' => true ); + } +); workflow_runner_smoke_register_ability( 'agents/chat', static function ( array $input ): \WP_Error { @@ -436,6 +455,48 @@ static function ( WP_Agent_Workflow_Run_Result $completed, string $run_id ) use WP_Agent_Run_Control::reset_store(); unset( $GLOBALS['__workflow_late_cancel_store'] ); +$terminal_winner_spec = WP_Agent_Workflow_Spec::from_array( + array( + 'id' => 'demo/generic-terminal-winner', + 'steps' => array( array( 'id' => 'arm', 'type' => 'ability', 'ability' => 'demo/arm-terminal-winner' ) ), + ) +); +foreach ( + array( + WP_Agent_Run_Control::STATUS_BUDGET_EXCEEDED => 'workflow_run_budget_exceeded', + WP_Agent_Run_Control::STATUS_STALLED => 'workflow_run_stalled', + WP_Agent_Run_Control::STATUS_INTERRUPTED => 'workflow_run_interrupted', + ) as $generic_status => $workflow_error +) { + $terminal_store = new Workflow_Late_Cancel_Store(); + $GLOBALS['__workflow_terminal_store'] = $terminal_store; + $GLOBALS['__workflow_terminal_status'] = $generic_status; + $terminal_hook = null; + $terminal_run_id = 'workflow-' . $generic_status; + WP_Agent_Run_Control::set_store( $terminal_store ); + add_action( + 'wp_agent_workflow_run_completed', + static function ( WP_Agent_Workflow_Run_Result $completed, string $run_id ) use ( &$terminal_hook, $terminal_run_id ): void { + if ( $terminal_run_id === $run_id ) { + $terminal_hook = $completed; + } + }, + 10, + 2 + ); + $terminal_recorder = new Capture_Recorder(); + $terminal_result = ( new WP_Agent_Workflow_Runner( $terminal_recorder ) )->run( $terminal_winner_spec, array(), array( 'run_id' => $terminal_run_id ) ); + $terminal_record = end( $terminal_recorder->writes ); + $terminal_stored = WP_Agent_Run_Control::get_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, $terminal_run_id ); + smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $terminal_result->get_status(), "{$generic_status} winner returns an honest failed workflow result", $failures, $passes ); + smoke_assert( $workflow_error, $terminal_result->get_error()['code'] ?? '', "{$generic_status} winner uses a stable workflow error", $failures, $passes ); + smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $terminal_record['status'] ?? '', "{$generic_status} winner reaches recorder projection", $failures, $passes ); + smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $terminal_hook instanceof WP_Agent_Workflow_Run_Result ? $terminal_hook->get_status() : '', "{$generic_status} winner reaches completion projection", $failures, $passes ); + smoke_assert( $generic_status, $terminal_stored['status'] ?? '', "{$generic_status} workflow projection matches authoritative storage", $failures, $passes ); +} +WP_Agent_Run_Control::reset_store(); +unset( $GLOBALS['__workflow_terminal_store'], $GLOBALS['__workflow_terminal_status'] ); + // ─── Generic run-control cancellation stops before the next step ────── $cancel_spec = WP_Agent_Workflow_Spec::from_array( @@ -708,12 +769,27 @@ public function find( string $run_id ): ?WP_Agent_Workflow_Run_Result { return n public function recent( array $args = array() ): array { return array(); } } +$recorder_start_completions = 0; +add_action( + 'wp_agent_workflow_run_completed', + static function ( WP_Agent_Workflow_Run_Result $completed, string $run_id ) use ( &$recorder_start_completions ): void { + unset( $completed ); + if ( 'recorder-start-fail' === $run_id ) { + ++$recorder_start_completions; + } + }, + 10, + 2 +); $recorder3 = new Failing_Start_Recorder(); -$result6 = ( new WP_Agent_Workflow_Runner( $recorder3 ) )->run( $spec, array( 'text' => 'hi' ) ); +$result6 = ( new WP_Agent_Workflow_Runner( $recorder3 ) )->run( $spec, array( 'text' => 'hi' ), array( 'run_id' => 'recorder-start-fail' ) ); +$recorder_start_stored = WP_Agent_Run_Control::get_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, 'recorder-start-fail' ); smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $result6->get_status(), 'recorder start failure => failed run', $failures, $passes ); smoke_assert( 'recorder_start_failed', $result6->get_error()['code'], 'recorder start failure has expected code', $failures, $passes ); smoke_assert( 0, count( $result6->get_steps() ), 'no steps run when recorder start fails', $failures, $passes ); smoke_assert( 0, $recorder3->update_calls, 'no update fired when start failed', $failures, $passes ); +smoke_assert( 1, $recorder_start_completions, 'recorder start failure fires completion exactly once', $failures, $passes ); +smoke_assert( WP_Agent_Run_Control::STATUS_FAILED, $recorder_start_stored['status'] ?? '', 'recorder start failure terminalizes run-control', $failures, $passes ); // ─── Input-validation failure goes through start → update lifecycle ─── @@ -739,6 +815,55 @@ public function recent( array $args = array() ): array { return array(); } smoke_assert( 'update', $tracker->events[1]['op'] ?? '', 'recorder sees update second', $failures, $passes ); smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $tracker->events[1]['status'] ?? '', 'update flips status to FAILED', $failures, $passes ); +class Resume_Failure_Recorder implements WP_Agent_Workflow_Run_Recorder { + public int $updates = 0; + public function __construct( private WP_Agent_Workflow_Run_Result $result ) {} + public function start( WP_Agent_Workflow_Run_Result $result ) { unset( $result ); return ''; } + public function update( WP_Agent_Workflow_Run_Result $result ) { ++$this->updates; $this->result = $result; return true; } + public function find( string $run_id ): ?WP_Agent_Workflow_Run_Result { return $run_id === $this->result->get_run_id() ? $this->result : null; } + public function recent( array $args = array() ): array { unset( $args ); return array( $this->result ); } +} + +$resume_source = new WP_Agent_Workflow_Run_Result( + 'resume-spec-missing', + 'demo/missing-replay', + WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED, + array(), + array(), + array(), + array(), + time(), + 0, + array( '_suspension' => array( 'step_index' => 0 ) ), + array(), + array() +); +$resume_recorder = new Resume_Failure_Recorder( $resume_source ); +$resume_completions = 0; +WP_Agent_Run_Control::start_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, 'resume-spec-missing', array( 'workflow_id' => 'demo/missing-replay' ) ); +add_action( + 'wp_agent_workflow_run_completed', + static function ( WP_Agent_Workflow_Run_Result $completed, string $run_id ) use ( &$resume_completions ): void { + unset( $completed ); + if ( 'resume-spec-missing' === $run_id ) { + ++$resume_completions; + } + }, + 10, + 2 +); +$resume_runner = new WP_Agent_Workflow_Runner( $resume_recorder ); +$resume_failed = $resume_runner->resume( 'resume-spec-missing' ); +$resume_stored = WP_Agent_Run_Control::get_run( WP_Agent_Workflow_Runner::RUN_CONTROL_STORE, 'resume-spec-missing' ); +$resume_replay = $resume_runner->resume( 'resume-spec-missing' ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $resume_failed->get_status(), 'missing resume spec returns terminal failure', $failures, $passes ); +smoke_assert( 'workflow_resume_spec_unavailable', $resume_failed->get_error()['code'] ?? '', 'missing resume spec keeps checked failure code', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $resume_recorder->find( 'resume-spec-missing' )->get_status(), 'missing resume spec updates recorder terminal state', $failures, $passes ); +smoke_assert( WP_Agent_Run_Control::STATUS_FAILED, $resume_stored['status'] ?? '', 'missing resume spec terminalizes run-control', $failures, $passes ); +smoke_assert( 1, $resume_completions, 'missing resume spec publishes completion exactly once', $failures, $passes ); +smoke_assert( 1, $resume_recorder->updates, 'duplicate failed resume does not update recorder again', $failures, $passes ); +smoke_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $resume_replay->get_status(), 'duplicate failed resume is idempotent', $failures, $passes ); + // ─── foreach step iterates over bound arrays with scoped vars ──────── $foreach_spec = WP_Agent_Workflow_Spec::from_array(