Skip to content

Commit 5addf59

Browse files
chubes4chubes4
authored
feat: add workflow run await service (#422)
1 parent 14c0603 commit 5addf59

5 files changed

Lines changed: 283 additions & 0 deletions

File tree

‎agents-api.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -251,6 +251,7 @@
251251
require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-branch-store.php';
252252
require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php';
253253
require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-scoped-drain.php';
254+
require_once AGENTS_API_PATH . 'src/Workflows/class-wp-agent-workflow-run-awaiter.php';
254255
require_once AGENTS_API_PATH . 'src/Workflows/register-workflows.php';
255256
require_once AGENTS_API_PATH . 'src/Workflows/register-workflow-step-executor.php';
256257
require_once AGENTS_API_PATH . 'src/Workflows/register-agents-workflow-abilities.php';

‎composer.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,7 @@
121121
"php tests/workflow-as-branch-smoke.php",
122122
"php tests/workflow-branch-concurrency-gate-smoke.php",
123123
"php tests/workflow-scoped-drain-smoke.php",
124+
"php tests/workflow-run-awaiter-smoke.php",
124125
"php tests/workflow-async-branch-payload-smoke.php",
125126
"php tests/workflow-async-loopback-target-smoke.php",
126127
"php tests/workflow-async-runner-dispatch-result-smoke.php",

‎src/Workflows/class-wp-agent-workflow-run-awaiter.php‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
<?php
2+
/**
3+
* Storage-agnostic foreground await service for persisted workflow runs.
4+
*
5+
* @package AgentsAPI
6+
* @since 0.5.2
7+
*/
8+
9+
namespace AgentsAPI\AI\Workflows;
10+
11+
defined( 'ABSPATH' ) || exit;
12+
13+
/**
14+
* Reload a workflow run and, when suspended, drive its scoped scheduler work.
15+
*/
16+
class WP_Agent_Workflow_Run_Awaiter {
17+
18+
public const SCHEMA = 'agents-api/workflow-run-await/v1';
19+
private WP_Agent_Workflow_Scoped_Drain $drain;
20+
21+
/**
22+
* @since 0.5.2
23+
* @param WP_Agent_Workflow_Scoped_Drain|null $drain Scoped foreground drain.
24+
*/
25+
public function __construct( ?WP_Agent_Workflow_Scoped_Drain $drain = null ) {
26+
$this->drain = $drain ?? new WP_Agent_Workflow_Scoped_Drain();
27+
}
28+
29+
/**
30+
* Await the current state of a workflow run within the supplied drain budget.
31+
*
32+
* The recorder is always caller-owned. A suspended run after a budget stop is a
33+
* successful, reconnectable response; drain refusal is preserved in `drain`.
34+
*
35+
* @since 0.5.2
36+
* @param string $run_id Workflow run id.
37+
* @param WP_Agent_Workflow_Run_Recorder $recorder Caller-owned run recorder.
38+
* @param array<string,mixed> $options Scoped drain options.
39+
* @return array{schema:string,run_id:string,status:string,terminal:bool,reconnectable:bool,result:?array<string,mixed>,drain:array<string,int|string|bool>}|\WP_Error
40+
*/
41+
public function await( string $run_id, WP_Agent_Workflow_Run_Recorder $recorder, array $options = array() ) {
42+
$current = $recorder->find( $run_id );
43+
if ( null === $current ) {
44+
return new \WP_Error( 'agents_workflow_run_not_found', 'No workflow run was found for the requested run_id.' );
45+
}
46+
47+
if ( $current->is_suspended() ) {
48+
$awaited = $this->drain_suspended_run( $run_id, $recorder, $options );
49+
$current = $awaited['result'];
50+
if ( null === $current ) {
51+
return new \WP_Error( 'agents_workflow_run_not_found', 'No workflow run was found for the requested run_id.' );
52+
}
53+
$stats = $awaited['stats'];
54+
} else {
55+
$stats = $this->zero_work_stats( $current->get_status(), $options );
56+
}
57+
58+
$terminal = $this->is_terminal( $current );
59+
60+
return array(
61+
'schema' => self::SCHEMA,
62+
'run_id' => $current->get_run_id(),
63+
'status' => $current->get_status(),
64+
'terminal' => $terminal,
65+
'reconnectable' => ! $terminal,
66+
'result' => $terminal ? $current->to_run_result_envelope()->to_array() : null,
67+
'drain' => $stats,
68+
);
69+
}
70+
71+
/**
72+
* Test seam that delegates to the production scoped drain.
73+
*
74+
* @param string $run_id Workflow run id.
75+
* @param WP_Agent_Workflow_Run_Recorder $recorder Run recorder.
76+
* @param array<string,mixed> $options Drain options.
77+
* @return array{result:?WP_Agent_Workflow_Run_Result,stats:array<string,int|string|bool>}
78+
*/
79+
protected function drain_suspended_run( string $run_id, WP_Agent_Workflow_Run_Recorder $recorder, array $options ): array {
80+
return $this->drain->drain_suspended_run( $run_id, $recorder, $options );
81+
}
82+
83+
private function is_terminal( WP_Agent_Workflow_Run_Result $result ): bool {
84+
return in_array(
85+
$result->get_status(),
86+
array(
87+
WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED,
88+
WP_Agent_Workflow_Run_Result::STATUS_FAILED,
89+
WP_Agent_Workflow_Run_Result::STATUS_SKIPPED,
90+
WP_Agent_Workflow_Run_Result::STATUS_CANCELLED,
91+
),
92+
true
93+
);
94+
}
95+
96+
/**
97+
* Build the scoped-drain stats shape without touching the queue.
98+
*
99+
* @param array<string,mixed> $options Await options.
100+
* @return array<string,int|string|bool>
101+
*/
102+
private function zero_work_stats( string $status, array $options ): array {
103+
$hooks = isset( $options['hooks'] ) && is_array( $options['hooks'] )
104+
? array_values( array_filter( $options['hooks'], 'is_string' ) )
105+
: WP_Agent_Workflow_Scoped_Drain::default_hooks();
106+
$hooks = empty( $hooks ) ? WP_Agent_Workflow_Scoped_Drain::default_hooks() : $hooks;
107+
108+
return array(
109+
'batches' => 0,
110+
'actions_processed' => 0,
111+
'completions' => 0,
112+
'failures' => 0,
113+
'remaining_pending' => 0,
114+
'total_pending' => 0,
115+
'warnings' => 0,
116+
'stop_reason' => 'terminal_status',
117+
'terminal_state' => $status,
118+
'hooks' => implode( ',', $hooks ),
119+
'group' => isset( $options['group'] ) && is_scalar( $options['group'] ) && '' !== (string) $options['group'] ? (string) $options['group'] : WP_Agent_Workflow_Scoped_Drain::default_group(),
120+
'available' => WP_Agent_Workflow_Scoped_Drain::is_available(),
121+
);
122+
}
123+
}

‎tests/bootstrap-smoke.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,7 @@
129129
agents_api_smoke_assert_equals( true, interface_exists( 'AgentsAPI\AI\Channels\WP_Agent_Bridge_Store' ), 'WP_Agent_Bridge_Store contract is available', $failures, $passes );
130130
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 );
131131
agents_api_smoke_assert_equals( true, class_exists( 'AgentsAPI\AI\Channels\WP_Agent_Bridge' ), 'WP_Agent_Bridge facade is available', $failures, $passes );
132+
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 );
132133
foreach ( $namespace_map as $source_class => $target_class ) {
133134
agents_api_smoke_assert_equals( true, class_exists( $target_class ) || interface_exists( $target_class ), $target_class . ' contract is available', $failures, $passes );
134135
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 );

‎tests/workflow-run-awaiter-smoke.php‎

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,157 @@
1+
<?php
2+
/**
3+
* Pure-PHP behavioral smoke test for the workflow run await service.
4+
*
5+
* Run with: php tests/workflow-run-awaiter-smoke.php
6+
*
7+
* @package AgentsAPI\Tests
8+
*/
9+
10+
defined( 'ABSPATH' ) || define( 'ABSPATH', __DIR__ . '/' );
11+
12+
if ( ! class_exists( 'WP_Error' ) ) {
13+
class WP_Error {
14+
public function __construct( private string $code, private string $message ) {}
15+
public function get_error_code(): string {
16+
return $this->code;
17+
}
18+
public function get_error_message(): string {
19+
return $this->message;
20+
}
21+
}
22+
}
23+
24+
require_once __DIR__ . '/../src/Runtime/class-wp-agent-run-result-envelope.php';
25+
require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-result.php';
26+
require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-recorder.php';
27+
require_once __DIR__ . '/../src/Workflows/interface-wp-agent-workflow-branch-executor.php';
28+
require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-action-scheduler-bridge.php';
29+
require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-action-scheduler-branch-executor.php';
30+
require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-scoped-drain.php';
31+
require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-awaiter.php';
32+
33+
use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Awaiter;
34+
use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Recorder;
35+
use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Run_Result;
36+
37+
final class Await_Smoke_Recorder implements WP_Agent_Workflow_Run_Recorder {
38+
/** @var array<string,WP_Agent_Workflow_Run_Result> */
39+
public array $runs = array();
40+
public int $finds = 0;
41+
42+
public function start( WP_Agent_Workflow_Run_Result $result ) {
43+
$this->runs[ $result->get_run_id() ] = $result;
44+
return $result->get_run_id();
45+
}
46+
public function update( WP_Agent_Workflow_Run_Result $result ) {
47+
$this->runs[ $result->get_run_id() ] = $result;
48+
return true;
49+
}
50+
public function find( string $run_id ): ?WP_Agent_Workflow_Run_Result {
51+
++$this->finds;
52+
return $this->runs[ $run_id ] ?? null;
53+
}
54+
public function recent( array $args = array() ): array {
55+
unset( $args );
56+
return array_values( $this->runs );
57+
}
58+
}
59+
60+
final class Await_Smoke_Awaiter extends WP_Agent_Workflow_Run_Awaiter {
61+
public int $drains = 0;
62+
public string $next_status = WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED;
63+
public string $stop_reason = 'terminal_status';
64+
65+
public function __construct() {}
66+
67+
protected function drain_suspended_run( string $run_id, WP_Agent_Workflow_Run_Recorder $recorder, array $options ): array {
68+
++$this->drains;
69+
$current = $recorder->find( $run_id );
70+
if ( null !== $current && WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED !== $this->next_status ) {
71+
$recorder->update( $current->with( array( 'status' => $this->next_status ) ) );
72+
}
73+
$result = $recorder->find( $run_id );
74+
return array(
75+
'result' => $result,
76+
'stats' => array(
77+
'batches' => isset( $options['limit'] ) ? 1 : 2, 'actions_processed' => isset( $options['limit'] ) ? 1 : 2,
78+
'completions' => 2, 'failures' => 0, 'remaining_pending' => WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED === $this->next_status ? 1 : 0,
79+
'total_pending' => 1, 'warnings' => 0, 'stop_reason' => $this->stop_reason,
80+
'terminal_state' => WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED === $this->next_status ? '' : $this->next_status,
81+
'hooks' => 'branch,resume', 'group' => 'workflow', 'available' => true,
82+
),
83+
);
84+
}
85+
}
86+
87+
function await_result( string $id, string $status, array $error = array() ): WP_Agent_Workflow_Run_Result {
88+
return new WP_Agent_Workflow_Run_Result( $id, 'workflow', $status, array(), array( 'answer' => 42 ), array(), $error, 1, 2, array() );
89+
}
90+
91+
$failures = array();
92+
$passes = 0;
93+
function await_assert( $expected, $actual, string $name ): void {
94+
global $failures, $passes;
95+
if ( $expected === $actual ) {
96+
++$passes;
97+
echo " PASS {$name}\n";
98+
return;
99+
}
100+
$failures[] = $name;
101+
echo " FAIL {$name}\n expected: " . var_export( $expected, true ) . "\n actual: " . var_export( $actual, true ) . "\n";
102+
}
103+
104+
echo "workflow-run-awaiter-smoke\n";
105+
$recorder = new Await_Smoke_Recorder();
106+
$awaiter = new Await_Smoke_Awaiter();
107+
108+
$missing = $awaiter->await( 'missing', $recorder );
109+
await_assert( true, $missing instanceof WP_Error, 'missing run returns WP_Error' );
110+
await_assert( 'agents_workflow_run_not_found', $missing->get_error_code(), 'missing run uses generic workflow error code' );
111+
await_assert( 0, $awaiter->drains, 'missing run performs no drain' );
112+
113+
$recorder->start( await_result( 'succeeded', WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED ) );
114+
$succeeded = $awaiter->await( 'succeeded', $recorder );
115+
await_assert( WP_Agent_Workflow_Run_Awaiter::SCHEMA, $succeeded['schema'], 'response has versioned await schema' );
116+
await_assert( true, $succeeded['terminal'], 'already succeeded is terminal' );
117+
await_assert( false, $succeeded['reconnectable'], 'already succeeded is not reconnectable' );
118+
await_assert( 0, $succeeded['drain']['actions_processed'], 'already succeeded performs zero drain work' );
119+
await_assert( 'agents-api/run-result/v1', $succeeded['result']['schema'], 'terminal result uses canonical generic envelope' );
120+
121+
$recorder->start( await_result( 'suspended-success', WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED ) );
122+
$before_finds = $recorder->finds;
123+
$completed = $awaiter->await( 'suspended-success', $recorder );
124+
await_assert( WP_Agent_Workflow_Run_Result::STATUS_SUCCEEDED, $completed['status'], 'suspended run can complete through drain seam' );
125+
await_assert( true, $completed['terminal'], 'completed suspended run is terminal' );
126+
await_assert( true, $recorder->finds - $before_finds >= 3, 'suspended await re-reads recorder live and afterward' );
127+
128+
$recorder->start( await_result( 'limited', WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED ) );
129+
$awaiter->next_status = WP_Agent_Workflow_Run_Result::STATUS_SUSPENDED;
130+
$awaiter->stop_reason = 'limit';
131+
$limited = $awaiter->await( 'limited', $recorder, array( 'limit' => 1 ) );
132+
await_assert( false, $limited['terminal'], 'limit-exhausted suspended run is nonterminal' );
133+
await_assert( true, $limited['reconnectable'], 'limit-exhausted suspended run is reconnectable' );
134+
await_assert( 'limit', $limited['drain']['stop_reason'], 'limit stop is preserved' );
135+
await_assert( null, $limited['result'], 'nonterminal state has no terminal result' );
136+
137+
$awaiter->stop_reason = 'time_limit';
138+
$budgeted = $awaiter->await( 'limited', $recorder, array( 'time_limit_ms' => 1 ) );
139+
await_assert( true, $budgeted['reconnectable'], 'budget-exhausted suspended run is reconnectable' );
140+
await_assert( 'time_limit', $budgeted['drain']['stop_reason'], 'budget stop is preserved' );
141+
142+
$recorder->start( await_result( 'failed', WP_Agent_Workflow_Run_Result::STATUS_FAILED, array( 'code' => 'step_failed', 'message' => 'Nope' ) ) );
143+
$failed = $awaiter->await( 'failed', $recorder );
144+
await_assert( WP_Agent_Workflow_Run_Result::STATUS_FAILED, $failed['status'], 'failed terminal remains failed' );
145+
await_assert( 'step_failed', $failed['result']['error']['code'], 'failed terminal preserves canonical error' );
146+
147+
$awaiter->stop_reason = 'refused_reentrant';
148+
$refused = $awaiter->await( 'limited', $recorder );
149+
await_assert( 'refused_reentrant', $refused['drain']['stop_reason'], 'reentrant refusal is preserved without fallback' );
150+
await_assert( true, $refused['reconnectable'], 'refused suspended run remains reconnectable' );
151+
152+
$source = (string) file_get_contents( __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-awaiter.php' );
153+
await_assert( false, str_contains( $source, 'wp_agent_workflow_run_recorder' ), 'awaiter has no ambient recorder resolver' );
154+
await_assert( false, str_contains( $source, 'implements WP_Agent_Workflow_Run_Recorder' ), 'awaiter adds no storage implementation' );
155+
156+
echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n";
157+
exit( count( $failures ) > 0 ? 1 : 0 );

0 commit comments

Comments
 (0)