From d32cef7d657ddca10c777eac039a93f27ada9e35 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Sun, 12 Jul 2026 13:06:10 -0400 Subject: [PATCH] feat: add workflow run await service --- agents-api.php | 1 + composer.json | 1 + .../class-wp-agent-workflow-run-awaiter.php | 123 ++++++++++++++ tests/bootstrap-smoke.php | 1 + tests/workflow-run-awaiter-smoke.php | 157 ++++++++++++++++++ 5 files changed, 283 insertions(+) create mode 100644 src/Workflows/class-wp-agent-workflow-run-awaiter.php create mode 100644 tests/workflow-run-awaiter-smoke.php diff --git a/agents-api.php b/agents-api.php index 7ed4196..a665bcc 100644 --- a/agents-api.php +++ b/agents-api.php @@ -251,6 +251,7 @@ require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-branch-store.php'; require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php'; require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-scoped-drain.php'; +require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-run-awaiter.php'; require_once AGENTS_API_PATH . 'src/Workflows/register-workflows.php'; require_once AGENTS_API_PATH . 'src/Workflows/register-workflow-step-executor.php'; require_once AGENTS_API_PATH . 'src/Workflows/register-agents-workflow-abilities.php'; diff --git a/composer.json b/composer.json index ac65c2e..4fb1c37 100644 --- a/composer.json +++ b/composer.json @@ -121,6 +121,7 @@ "php tests/workflow-as-branch-smoke.php", "php tests/workflow-branch-concurrency-gate-smoke.php", "php tests/workflow-scoped-drain-smoke.php", + "php tests/workflow-run-awaiter-smoke.php", "php tests/workflow-async-branch-payload-smoke.php", "php tests/workflow-async-loopback-target-smoke.php", "php tests/workflow-async-runner-dispatch-result-smoke.php", diff --git a/src/Workflows/class-wp-agent-workflow-run-awaiter.php b/src/Workflows/class-wp-agent-workflow-run-awaiter.php new file mode 100644 index 0000000..44494ad --- /dev/null +++ b/src/Workflows/class-wp-agent-workflow-run-awaiter.php @@ -0,0 +1,123 @@ +drain = $drain ?? new WP_Agent_Workflow_Scoped_Drain(); + } + + /** + * Await the current state of a workflow run within the supplied drain budget. + * + * The recorder is always caller-owned. A suspended run after a budget stop is a + * successful, reconnectable response; drain refusal is preserved in `drain`. + * + * @since 0.5.2 + * @param string $run_id Workflow run id. + * @param WP_Agent_Workflow_Run_Recorder $recorder Caller-owned run recorder. + * @param array $options Scoped drain options. + * @return array{schema:string,run_id:string,status:string,terminal:bool,reconnectable:bool,result:?array,drain:array}|\WP_Error + */ + public function await( string $run_id, WP_Agent_Workflow_Run_Recorder $recorder, array $options = array() ) { + $current = $recorder->find( $run_id ); + if ( null === $current ) { + return new \WP_Error( 'agents_workflow_run_not_found', 'No workflow run was found for the requested run_id.' ); + } + + if ( $current->is_suspended() ) { + $awaited = $this->drain_suspended_run( $run_id, $recorder, $options ); + $current = $awaited['result']; + if ( null === $current ) { + return new \WP_Error( 'agents_workflow_run_not_found', 'No workflow run was found for the requested run_id.' ); + } + $stats = $awaited['stats']; + } else { + $stats = $this->zero_work_stats( $current->get_status(), $options ); + } + + $terminal = $this->is_terminal( $current ); + + return array( + 'schema' => self::SCHEMA, + 'run_id' => $current->get_run_id(), + 'status' => $current->get_status(), + 'terminal' => $terminal, + 'reconnectable' => ! $terminal, + 'result' => $terminal ? $current->to_run_result_envelope()->to_array() : null, + 'drain' => $stats, + ); + } + + /** + * Test seam that delegates to the production scoped drain. + * + * @param string $run_id Workflow run id. + * @param WP_Agent_Workflow_Run_Recorder $recorder Run recorder. + * @param array $options Drain options. + * @return array{result:?WP_Agent_Workflow_Run_Result,stats:array} + */ + protected function drain_suspended_run( string $run_id, WP_Agent_Workflow_Run_Recorder $recorder, array $options ): array { + return $this->drain->drain_suspended_run( $run_id, $recorder, $options ); + } + + private function is_terminal( WP_Agent_Workflow_Run_Result $result ): bool { + return in_array( + $result->get_status(), + array( + WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, + WP_Agent_Workflow_Run_Result::STATUS_FAILED, + WP_Agent_Workflow_Run_Result::STATUS_SKIPPED, + WP_Agent_Workflow_Run_Result::STATUS_CANCELLED, + ), + true + ); + } + + /** + * Build the scoped-drain stats shape without touching the queue. + * + * @param array $options Await options. + * @return array + */ + private function zero_work_stats( string $status, array $options ): array { + $hooks = isset( $options['hooks'] ) && is_array( $options['hooks'] ) + ? array_values( array_filter( $options['hooks'], 'is_string' ) ) + : WP_Agent_Workflow_Scoped_Drain::default_hooks(); + $hooks = empty( $hooks ) ? WP_Agent_Workflow_Scoped_Drain::default_hooks() : $hooks; + + return array( + 'batches' => 0, + 'actions_processed' => 0, + 'completions' => 0, + 'failures' => 0, + 'remaining_pending' => 0, + 'total_pending' => 0, + 'warnings' => 0, + 'stop_reason' => 'terminal_status', + 'terminal_state' => $status, + 'hooks' => implode( ',', $hooks ), + 'group' => isset( $options['group'] ) && is_scalar( $options['group'] ) && '' !== (string) $options['group'] ? (string) $options['group'] : WP_Agent_Workflow_Scoped_Drain::default_group(), + 'available' => WP_Agent_Workflow_Scoped_Drain::is_available(), + ); + } +} diff --git a/tests/bootstrap-smoke.php b/tests/bootstrap-smoke.php index d904dfb..b7df772 100644 --- a/tests/bootstrap-smoke.php +++ b/tests/bootstrap-smoke.php @@ -129,6 +129,7 @@ agents_api_smoke_assert_equals( true, interface_exists( 'AgentsAPI\AI\Channels\WP_Agent_Bridge_Store' ), 'WP_Agent_Bridge_Store contract is available', $failures, $passes ); agents_api_smoke_assert_equals( true, class_exists( 'AgentsAPI\AI\Channels\WP_Agent_Option_Bridge_Store' ), 'WP_Agent_Option_Bridge_Store implementation is available', $failures, $passes ); agents_api_smoke_assert_equals( true, class_exists( 'AgentsAPI\AI\Channels\WP_Agent_Bridge' ), 'WP_Agent_Bridge facade is available', $failures, $passes ); +agents_api_smoke_assert_equals( true, class_exists( 'AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Awaiter' ), 'WP_Agent_Workflow_Run_Awaiter service is available', $failures, $passes ); foreach ( $namespace_map as $source_class => $target_class ) { agents_api_smoke_assert_equals( true, class_exists( $target_class ) || interface_exists( $target_class ), $target_class . ' contract is available', $failures, $passes ); agents_api_smoke_assert_equals( false, class_exists( $source_class, false ) || interface_exists( $source_class, false ), $source_class . ' compatibility alias is not loaded', $failures, $passes ); diff --git a/tests/workflow-run-awaiter-smoke.php b/tests/workflow-run-awaiter-smoke.php new file mode 100644 index 0000000..f27baa2 --- /dev/null +++ b/tests/workflow-run-awaiter-smoke.php @@ -0,0 +1,157 @@ +code; + } + public function get_error_message(): string { + return $this->message; + } + } +} + +require_once __DIR__ . '/../src/Runtime/class-wp-agent-run-result-envelope.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-result.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-recorder.php'; +require_once __DIR__ . '/../src/Workflows/interface-wp-agent-workflow-branch-executor.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-action-scheduler-bridge.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-scoped-drain.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-awaiter.php'; + +use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Awaiter; +use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Recorder; +use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Result; + +final class Await_Smoke_Recorder implements WP_Agent_Workflow_Run_Recorder { + /** @var array */ + public array $runs = array(); + public int $finds = 0; + + public function start( WP_Agent_Workflow_Run_Result $result ) { + $this->runs[ $result->get_run_id() ] = $result; + return $result->get_run_id(); + } + public function update( WP_Agent_Workflow_Run_Result $result ) { + $this->runs[ $result->get_run_id() ] = $result; + return true; + } + public function find( string $run_id ): ?WP_Agent_Workflow_Run_Result { + ++$this->finds; + return $this->runs[ $run_id ] ?? null; + } + public function recent( array $args = array() ): array { + unset( $args ); + return array_values( $this->runs ); + } +} + +final class Await_Smoke_Awaiter extends WP_Agent_Workflow_Run_Awaiter { + public int $drains = 0; + public string $next_status = WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED; + public string $stop_reason = 'terminal_status'; + + public function __construct() {} + + protected function drain_suspended_run( string $run_id, WP_Agent_Workflow_Run_Recorder $recorder, array $options ): array { + ++$this->drains; + $current = $recorder->find( $run_id ); + if ( null !== $current && WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED !== $this->next_status ) { + $recorder->update( $current->with( array( 'status' => $this->next_status ) ) ); + } + $result = $recorder->find( $run_id ); + return array( + 'result' => $result, + 'stats' => array( + 'batches' => isset( $options['limit'] ) ? 1 : 2, 'actions_processed' => isset( $options['limit'] ) ? 1 : 2, + 'completions' => 2, 'failures' => 0, 'remaining_pending' => WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED === $this->next_status ? 1 : 0, + 'total_pending' => 1, 'warnings' => 0, 'stop_reason' => $this->stop_reason, + 'terminal_state' => WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED === $this->next_status ? '' : $this->next_status, + 'hooks' => 'branch,resume', 'group' => 'workflow', 'available' => true, + ), + ); + } +} + +function await_result( string $id, string $status, array $error = array() ): WP_Agent_Workflow_Run_Result { + return new WP_Agent_Workflow_Run_Result( $id, 'workflow', $status, array(), array( 'answer' => 42 ), array(), $error, 1, 2, array() ); +} + +$failures = array(); +$passes = 0; +function await_assert( $expected, $actual, string $name ): void { + global $failures, $passes; + if ( $expected === $actual ) { + ++$passes; + echo " PASS {$name}\n"; + return; + } + $failures[] = $name; + echo " FAIL {$name}\n expected: " . var_export( $expected, true ) . "\n actual: " . var_export( $actual, true ) . "\n"; +} + +echo "workflow-run-awaiter-smoke\n"; +$recorder = new Await_Smoke_Recorder(); +$awaiter = new Await_Smoke_Awaiter(); + +$missing = $awaiter->await( 'missing', $recorder ); +await_assert( true, $missing instanceof WP_Error, 'missing run returns WP_Error' ); +await_assert( 'agents_workflow_run_not_found', $missing->get_error_code(), 'missing run uses generic workflow error code' ); +await_assert( 0, $awaiter->drains, 'missing run performs no drain' ); + +$recorder->start( await_result( 'succeeded', WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED ) ); +$succeeded = $awaiter->await( 'succeeded', $recorder ); +await_assert( WP_Agent_Workflow_Run_Awaiter::SCHEMA, $succeeded['schema'], 'response has versioned await schema' ); +await_assert( true, $succeeded['terminal'], 'already succeeded is terminal' ); +await_assert( false, $succeeded['reconnectable'], 'already succeeded is not reconnectable' ); +await_assert( 0, $succeeded['drain']['actions_processed'], 'already succeeded performs zero drain work' ); +await_assert( 'agents-api/run-result/v1', $succeeded['result']['schema'], 'terminal result uses canonical generic envelope' ); + +$recorder->start( await_result( 'suspended-success', WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED ) ); +$before_finds = $recorder->finds; +$completed = $awaiter->await( 'suspended-success', $recorder ); +await_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $completed['status'], 'suspended run can complete through drain seam' ); +await_assert( true, $completed['terminal'], 'completed suspended run is terminal' ); +await_assert( true, $recorder->finds - $before_finds >= 3, 'suspended await re-reads recorder live and afterward' ); + +$recorder->start( await_result( 'limited', WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED ) ); +$awaiter->next_status = WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED; +$awaiter->stop_reason = 'limit'; +$limited = $awaiter->await( 'limited', $recorder, array( 'limit' => 1 ) ); +await_assert( false, $limited['terminal'], 'limit-exhausted suspended run is nonterminal' ); +await_assert( true, $limited['reconnectable'], 'limit-exhausted suspended run is reconnectable' ); +await_assert( 'limit', $limited['drain']['stop_reason'], 'limit stop is preserved' ); +await_assert( null, $limited['result'], 'nonterminal state has no terminal result' ); + +$awaiter->stop_reason = 'time_limit'; +$budgeted = $awaiter->await( 'limited', $recorder, array( 'time_limit_ms' => 1 ) ); +await_assert( true, $budgeted['reconnectable'], 'budget-exhausted suspended run is reconnectable' ); +await_assert( 'time_limit', $budgeted['drain']['stop_reason'], 'budget stop is preserved' ); + +$recorder->start( await_result( 'failed', WP_Agent_Workflow_Run_Result::STATUS_FAILED, array( 'code' => 'step_failed', 'message' => 'Nope' ) ) ); +$failed = $awaiter->await( 'failed', $recorder ); +await_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $failed['status'], 'failed terminal remains failed' ); +await_assert( 'step_failed', $failed['result']['error']['code'], 'failed terminal preserves canonical error' ); + +$awaiter->stop_reason = 'refused_reentrant'; +$refused = $awaiter->await( 'limited', $recorder ); +await_assert( 'refused_reentrant', $refused['drain']['stop_reason'], 'reentrant refusal is preserved without fallback' ); +await_assert( true, $refused['reconnectable'], 'refused suspended run remains reconnectable' ); + +$source = (string) file_get_contents( __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-awaiter.php' ); +await_assert( false, str_contains( $source, 'wp_agent_workflow_run_recorder' ), 'awaiter has no ambient recorder resolver' ); +await_assert( false, str_contains( $source, 'implements WP_Agent_Workflow_Run_Recorder' ), 'awaiter adds no storage implementation' ); + +echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; +exit( count( $failures ) > 0 ? 1 : 0 );