From 6a4144149599ec9e636df66f32076e3eb5e02f96 Mon Sep 17 00:00:00 2001 From: Miguel Lezama Date: Mon, 3 Aug 2026 11:29:07 -0300 Subject: [PATCH] Refactor: share parallel-roles/map spec validation across sync and async paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SYNC in-process loops (`run_parallel_roles()` / `run_parallel_map()`) and the ASYNC dispatch-plan builders (`build_roles_dispatch_plan()` / `build_map_dispatch_plan()`) independently re-implemented the SAME parallel spec validation. Because the two copies were maintained by hand, they could silently DRIFT — a rule tightened on one path but not the other would make the substrate accept a spec synchronously that it rejects asynchronously (or vice versa), a hard-to-diagnose correctness gap. Extract two shared private helpers — `validate_parallel_roles_spec()` and `validate_parallel_map_spec()` — that enforce the rules once (roles: entries-are-arrays / non-empty branches / at-most-one aggregator; map: items resolve to an array / non-empty nested steps) and return the parsed split (sibling branches + optional aggregator, or items + steps), or a WP_Error with the shared error code on an invalid spec. Route all four entry points through them. Behavior is IDENTICAL: same error codes, same messages, same accepted and rejected specs — only the duplication is removed. Adds tests/workflow-parallel-spec-validation-parity-smoke.php, which asserts the sync and async entry points reject the same invalid specs with the same error codes AND accept the same valid roles (with aggregator) and map specs, so the two paths can never drift again. All existing workflow smoke tests still pass; PHPStan (max) is clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- composer.json | 1 + .../class-wp-agent-workflow-runner.php | 167 +++++----- ...-parallel-spec-validation-parity-smoke.php | 304 ++++++++++++++++++ 3 files changed, 395 insertions(+), 77 deletions(-) create mode 100644 tests/workflow-parallel-spec-validation-parity-smoke.php diff --git a/composer.json b/composer.json index 8b864bf..ff36136 100644 --- a/composer.json +++ b/composer.json @@ -130,6 +130,7 @@ "php tests/workflow-runner-smoke.php", "php tests/workflow-parallel-smoke.php", "php tests/workflow-parallel-async-smoke.php", + "php tests/workflow-parallel-spec-validation-parity-smoke.php", "php tests/workflow-as-branch-smoke.php", "php tests/workflow-branch-concurrency-gate-smoke.php", "php tests/workflow-scoped-drain-smoke.php", diff --git a/src/Workflows/class-wp-agent-workflow-runner.php b/src/Workflows/class-wp-agent-workflow-runner.php index 8de2306..5fa2b4f 100644 --- a/src/Workflows/class-wp-agent-workflow-runner.php +++ b/src/Workflows/class-wp-agent-workflow-runner.php @@ -970,17 +970,23 @@ private static function dispatch_parallel_async( WP_Agent_Workflow_Branch_Execut } /** - * Build the dispatch plan for the roles shape: one branch descriptor per - * sibling role, the collect plan (with its OPTIONAL aggregator branch), and - * the shared immutable context. Mirrors the sync `run_parallel_roles()` - * validation so the async path rejects the same malformed specs. + * Validate the roles-shape parallel spec ONCE for both the sync loop and the + * async dispatch-plan builder so the two paths can never drift on which specs + * they accept. Enforces the three shared rules: every branch entry is an + * array (`workflow_parallel_branch_invalid`), at least one branch + * (`workflow_parallel_branches_empty`), and at most one aggregator + * (`workflow_parallel_aggregator_invalid`). Splits the branches into the + * sibling set and the OPTIONAL aggregator (zero aggregators is valid — + * scatter-collect-return; one runs after the siblings over their collected + * outputs). Returns the split, or a WP_Error with the shared code on an + * invalid spec. * * @since 0.5.0 * * @param array $step Resolved parallel step. - * @return array{branches:array>,shared_context:array,aggregate:array}|\WP_Error + * @return array{sibling_branches:list>,aggregator:array|null}|\WP_Error */ - private static function build_roles_dispatch_plan( array $step ) { + private static function validate_parallel_roles_spec( array $step ) { $branch_specs = array(); foreach ( (array) $step['branches'] as $branch_spec ) { if ( ! is_array( $branch_spec ) ) { @@ -999,9 +1005,6 @@ private static function build_roles_dispatch_plan( array $step ) { ); } - // At most one branch may be the aggregator. Zero aggregators is valid - // (scatter-collect-return); one aggregator runs after the siblings over - // their collected outputs. $aggregator = null; $aggregator_roles = array(); $sibling_branches = array(); @@ -1023,6 +1026,68 @@ private static function build_roles_dispatch_plan( array $step ) { ); } + return array( + 'sibling_branches' => $sibling_branches, + 'aggregator' => $aggregator, + ); + } + + /** + * Validate the map-shape parallel spec ONCE for both the sync loop and the + * async dispatch-plan builder so the two paths can never drift on which specs + * they accept. Enforces the two shared rules: `items` resolves to an array + * (`workflow_parallel_items_invalid`) and a non-empty nested `steps` list + * (`workflow_parallel_steps_invalid`). Returns the resolved items + steps, or + * a WP_Error with the shared code on an invalid spec. + * + * @since 0.5.0 + * + * @param array $step Resolved parallel step. + * @return array{items:array,steps:array}|\WP_Error + */ + private static function validate_parallel_map_spec( array $step ) { + $items = $step['items'] ?? array(); + if ( ! is_array( $items ) ) { + return new \WP_Error( + 'workflow_parallel_items_invalid', + 'parallel-map step `items` must resolve to an array.' + ); + } + + $steps = $step['steps'] ?? array(); + if ( empty( $steps ) || ! is_array( $steps ) ) { + return new \WP_Error( + 'workflow_parallel_steps_invalid', + 'parallel-map step must include a non-empty nested `steps` list.' + ); + } + + return array( + 'items' => $items, + 'steps' => $steps, + ); + } + + /** + * Build the dispatch plan for the roles shape: one branch descriptor per + * sibling role, the collect plan (with its OPTIONAL aggregator branch), and + * the shared immutable context. Routes through the shared + * `validate_parallel_roles_spec()` so the async path rejects EXACTLY the same + * malformed specs as the sync loop. + * + * @since 0.5.0 + * + * @param array $step Resolved parallel step. + * @return array{branches:array>,shared_context:array,aggregate:array}|\WP_Error + */ + private static function build_roles_dispatch_plan( array $step ) { + $spec = self::validate_parallel_roles_spec( $step ); + if ( is_wp_error( $spec ) ) { + return $spec; + } + $sibling_branches = $spec['sibling_branches']; + $aggregator = $spec['aggregator']; + $shared_context = is_array( $step['context'] ?? null ) ? self::string_keyed_array( $step['context'] ) : array(); /** @var array> $descriptors */ @@ -1061,21 +1126,12 @@ private static function build_roles_dispatch_plan( array $step ) { * @return array{branches:array>,shared_context:array,aggregate:array}|\WP_Error */ private static function build_map_dispatch_plan( array $step ) { - $items = $step['items'] ?? array(); - if ( ! is_array( $items ) ) { - return new \WP_Error( - 'workflow_parallel_items_invalid', - 'parallel-map step `items` must resolve to an array.' - ); - } - - $steps = $step['steps'] ?? array(); - if ( empty( $steps ) || ! is_array( $steps ) ) { - return new \WP_Error( - 'workflow_parallel_steps_invalid', - 'parallel-map step must include a non-empty nested `steps` list.' - ); + $spec = self::validate_parallel_map_spec( $step ); + if ( is_wp_error( $spec ) ) { + return $spec; } + $items = $spec['items']; + $steps = $spec['steps']; $as_value = self::string_value( $step['as'] ?? null ); $index_as_value = self::string_value( $step['index_as'] ?? null ); @@ -1236,21 +1292,12 @@ public static function aggregate_branch_results( array $aggregate, array $branch * @return array|WP_Error */ private static function run_parallel_map( array $step, array $context, array $handlers ) { - $items = $step['items'] ?? array(); - if ( ! is_array( $items ) ) { - return new \WP_Error( - 'workflow_parallel_items_invalid', - 'parallel-map step `items` must resolve to an array.' - ); - } - - $steps = $step['steps'] ?? array(); - if ( empty( $steps ) || ! is_array( $steps ) ) { - return new \WP_Error( - 'workflow_parallel_steps_invalid', - 'parallel-map step must include a non-empty nested `steps` list.' - ); + $spec = self::validate_parallel_map_spec( $step ); + if ( is_wp_error( $spec ) ) { + return $spec; } + $items = $spec['items']; + $steps = $spec['steps']; $as_value = self::string_value( $step['as'] ?? null ); $index_as_value = self::string_value( $step['index_as'] ?? null ); @@ -1308,40 +1355,12 @@ private static function run_parallel_map( array $step, array $context, array $ha * @return array|WP_Error */ private static function run_parallel_roles( array $step, array $context, array $handlers ) { - $branch_specs = array(); - foreach ( (array) $step['branches'] as $branch_spec ) { - if ( ! is_array( $branch_spec ) ) { - return new \WP_Error( - 'workflow_parallel_branch_invalid', - 'parallel branch entries must be arrays.' - ); - } - $branch_specs[] = $branch_spec; - } - - if ( empty( $branch_specs ) ) { - return new \WP_Error( - 'workflow_parallel_branches_empty', - 'parallel-roles step must declare a non-empty `branches` list.' - ); - } - - // At most one branch may be the aggregator (optional). - $aggregator_roles = array(); - foreach ( $branch_specs as $branch_spec ) { - if ( ! empty( $branch_spec['is_aggregator'] ) ) { - $aggregator_roles[] = self::string_value( $branch_spec['role'] ?? '' ); - } - } - if ( count( $aggregator_roles ) > 1 ) { - return new \WP_Error( - 'workflow_parallel_aggregator_invalid', - sprintf( - 'parallel-roles step may declare at most one aggregator branch (`is_aggregator` true); found %d.', - count( $aggregator_roles ) - ) - ); + $spec = self::validate_parallel_roles_spec( $step ); + if ( is_wp_error( $spec ) ) { + return $spec; } + $sibling_branches = $spec['sibling_branches']; + $aggregator = $spec['aggregator']; // Shared immutable context: deep-copied per branch so a branch cannot // mutate it for its siblings or an aggregator. Arrays are copied by @@ -1351,16 +1370,10 @@ private static function run_parallel_roles( array $step, array $context, array $ $branch_outputs = array(); $branch_records = array(); - $aggregator = null; // Scatter: every non-aggregator branch, each against its own snapshot of // the shared context. - foreach ( $branch_specs as $branch_spec ) { - if ( ! empty( $branch_spec['is_aggregator'] ) ) { - $aggregator = $branch_spec; - continue; - } - + foreach ( $sibling_branches as $branch_spec ) { $run = self::run_role_branch( $branch_spec, $shared_context, array(), $context, $executor, $handlers ); if ( is_wp_error( $run ) ) { return $run; diff --git a/tests/workflow-parallel-spec-validation-parity-smoke.php b/tests/workflow-parallel-spec-validation-parity-smoke.php new file mode 100644 index 0000000..7168760 --- /dev/null +++ b/tests/workflow-parallel-spec-validation-parity-smoke.php @@ -0,0 +1,304 @@ +code; } + public function get_error_message(): string { return $this->message; } + public function get_error_data() { return $this->data; } + } +} +if ( ! function_exists( 'is_wp_error' ) ) { + function is_wp_error( $value ): bool { + return $value instanceof WP_Error; + } +} +if ( ! class_exists( 'WP_Ability' ) ) { + class WP_Ability { + public function __construct( private string $name, private array $args ) {} + public function get_name(): string { return $this->name; } + public function get_input_schema(): array { return isset( $this->args['input_schema'] ) && is_array( $this->args['input_schema'] ) ? $this->args['input_schema'] : array(); } + public function get_meta_item( string $key, $default = null ) { return $this->args['meta'][ $key ] ?? $default; } + public function execute( $input = null ) { + $callback = $this->args['execute_callback'] ?? null; + return is_callable( $callback ) ? call_user_func( $callback, is_array( $input ) ? $input : array() ) : null; + } + } +} + +$GLOBALS['__filters'] = array(); +$GLOBALS['__abilities'] = array(); +$GLOBALS['__options'] = array(); + +if ( ! function_exists( 'add_filter' ) ) { + function add_filter( string $hook, callable $cb, int $priority = 10, int $accepted_args = 1 ): void { + unset( $accepted_args ); + $GLOBALS['__filters'][ $hook ][ $priority ][] = $cb; + } +} +if ( ! function_exists( 'apply_filters' ) ) { + function apply_filters( string $hook, $value, ...$args ) { + $cbs = $GLOBALS['__filters'][ $hook ] ?? array(); + ksort( $cbs ); + foreach ( $cbs as $bucket ) { + foreach ( $bucket as $cb ) { + $value = call_user_func_array( $cb, array_merge( array( $value ), $args ) ); + } + } + return $value; + } +} +if ( ! function_exists( 'add_action' ) ) { + function add_action( string $hook, callable $cb, int $priority = 10, int $accepted_args = 1 ): void { + add_filter( $hook, $cb, $priority, $accepted_args ); + } +} +if ( ! function_exists( 'do_action' ) ) { + function do_action( string $hook, ...$args ): void { + $cbs = $GLOBALS['__filters'][ $hook ] ?? array(); + ksort( $cbs ); + foreach ( $cbs as $bucket ) { + foreach ( $bucket as $cb ) { + call_user_func_array( $cb, $args ); + } + } + } +} +if ( ! function_exists( 'wp_get_ability' ) ) { + function wp_get_ability( string $name ) { + return $GLOBALS['__abilities'][ $name ] ?? null; + } +} +if ( ! function_exists( 'get_option' ) ) { + function get_option( string $option, $default = false ) { + return $GLOBALS['__options'][ $option ] ?? $default; + } +} +if ( ! function_exists( 'update_option' ) ) { + function update_option( string $option, $value, $autoload = null ): bool { + unset( $autoload ); + $GLOBALS['__options'][ $option ] = $value; + return true; + } +} + +function parity_register_ability( string $name, \Closure $handler ): void { + $GLOBALS['__abilities'][ $name ] = new WP_Ability( + $name, + array( + 'label' => $name, + 'description' => 'Parallel spec-validation parity smoke stub.', + 'input_schema' => array( 'type' => 'object' ), + 'execute_callback' => $handler, + ) + ); +} + +function smoke_assert( $expected, $actual, string $name, array &$failures, int &$passes ): void { + if ( $expected === $actual ) { + ++$passes; + echo " PASS {$name}\n"; + return; + } + $failures[] = $name; + echo " FAIL {$name}\n"; + echo ' expected: ' . var_export( $expected, true ) . "\n"; + echo ' actual: ' . var_export( $actual, true ) . "\n"; +} + +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-bindings.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-spec-validator.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-spec.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-result.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-store.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-recorder.php'; +require_once __DIR__ . '/../src/Abilities/class-wp-agent-ability-dispatcher.php'; +require_once __DIR__ . '/../src/Runtime/interface-wp-agent-run-control-store.php'; +require_once __DIR__ . '/../src/Runtime/class-wp-agent-option-run-control-store.php'; +require_once __DIR__ . '/../src/Runtime/class-wp-agent-run-control.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-run-context.php'; +require_once __DIR__ . '/../src/Workflows/interface-wp-agent-workflow-branch-executor.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-step-executor.php'; +require_once __DIR__ . '/../src/Workflows/class-wp-agent-workflow-runner.php'; + +use AgentsAPI\AI\Workflows\WP_Agent_Workflow_Runner; + +// A trivial worker so a VALID spec's branches actually execute on the sync path. +parity_register_ability( + 'demo/echo', + static function ( array $input ): array { + return array( 'echoed' => (string) ( $input['value'] ?? '' ) ); + } +); + +// ── Reflection handles to the private static entry points ──────────────────── + +$runner_class = new ReflectionClass( WP_Agent_Workflow_Runner::class ); + +$sync_roles = $runner_class->getMethod( 'run_parallel_roles' ); +$async_roles = $runner_class->getMethod( 'build_roles_dispatch_plan' ); +$sync_map = $runner_class->getMethod( 'run_parallel_map' ); +$async_map = $runner_class->getMethod( 'build_map_dispatch_plan' ); +$handlers_m = $runner_class->getMethod( 'default_step_handlers' ); +foreach ( array( $sync_roles, $async_roles, $sync_map, $async_map, $handlers_m ) as $m ) { + $m->setAccessible( true ); +} + +/** @var array $handlers */ +$handlers = $handlers_m->invoke( null ); + +/** + * Invoke a sync entry point (run_parallel_roles / run_parallel_map) and return + * the error code, or '' when it returned a non-error result. + */ +function parity_sync_code( ReflectionMethod $method, array $step, array $handlers ): string { + $result = $method->invoke( null, $step, array(), $handlers ); + return is_wp_error( $result ) ? $result->get_error_code() : ''; +} + +/** + * Invoke an async entry point (build_*_dispatch_plan) and return the error code, + * or '' when it returned a non-error result. + */ +function parity_async_code( ReflectionMethod $method, array $step ): string { + $result = $method->invoke( null, $step ); + return is_wp_error( $result ) ? $result->get_error_code() : ''; +} + +// ── 1. Roles: invalid specs rejected with the SAME code on both paths ───────── + +$roles_invalid = array( + 'branch-not-array' => array( + 'step' => array( 'branches' => array( 'not-an-array' ) ), + 'code' => 'workflow_parallel_branch_invalid', + ), + 'empty-branches' => array( + 'step' => array( 'branches' => array() ), + 'code' => 'workflow_parallel_branches_empty', + ), + 'two-aggregators' => array( + 'step' => array( + 'branches' => array( + array( 'role' => 'a', 'is_aggregator' => true, 'steps' => array( array( 'id' => 's1', 'type' => 'ability', 'ability' => 'demo/echo' ) ) ), + array( 'role' => 'b', 'is_aggregator' => true, 'steps' => array( array( 'id' => 's2', 'type' => 'ability', 'ability' => 'demo/echo' ) ) ), + ), + ), + 'code' => 'workflow_parallel_aggregator_invalid', + ), +); + +foreach ( $roles_invalid as $label => $case ) { + $sync = parity_sync_code( $sync_roles, $case['step'], $handlers ); + $async = parity_async_code( $async_roles, $case['step'] ); + smoke_assert( $case['code'], $sync, "roles sync rejects {$label} with {$case['code']}", $failures, $passes ); + smoke_assert( $case['code'], $async, "roles async rejects {$label} with {$case['code']}", $failures, $passes ); + smoke_assert( $sync, $async, "roles sync/async PARITY on {$label} (same code)", $failures, $passes ); +} + +// ── 2. Map: invalid specs rejected with the SAME code on both paths ─────────── + +$map_invalid = array( + 'items-not-array' => array( + 'step' => array( 'items' => 'not-an-array', 'steps' => array( array( 'id' => 'd', 'type' => 'ability', 'ability' => 'demo/echo' ) ) ), + 'code' => 'workflow_parallel_items_invalid', + ), + 'empty-steps' => array( + 'step' => array( 'items' => array( 1, 2 ), 'steps' => array() ), + 'code' => 'workflow_parallel_steps_invalid', + ), +); + +foreach ( $map_invalid as $label => $case ) { + $sync = parity_sync_code( $sync_map, $case['step'], $handlers ); + $async = parity_async_code( $async_map, $case['step'] ); + smoke_assert( $case['code'], $sync, "map sync rejects {$label} with {$case['code']}", $failures, $passes ); + smoke_assert( $case['code'], $async, "map async rejects {$label} with {$case['code']}", $failures, $passes ); + smoke_assert( $sync, $async, "map sync/async PARITY on {$label} (same code)", $failures, $passes ); +} + +// ── 3. Roles: the SAME valid spec (siblings + aggregator) is ACCEPTED on both ─ + +$roles_valid = array( + 'context' => array( 'marker' => 'M' ), + 'branches' => array( + array( + 'role' => 'headline', + 'required' => true, + 'is_aggregator' => false, + 'steps' => array( array( 'id' => 'h', 'type' => 'ability', 'ability' => 'demo/echo', 'args' => array( 'value' => 'H' ) ) ), + ), + array( + 'role' => 'body', + 'required' => true, + 'is_aggregator' => false, + 'steps' => array( array( 'id' => 'b', 'type' => 'ability', 'ability' => 'demo/echo', 'args' => array( 'value' => 'B' ) ) ), + ), + array( + 'role' => 'fuse', + 'required' => true, + 'is_aggregator' => true, + 'steps' => array( array( 'id' => 'agg', 'type' => 'ability', 'ability' => 'demo/echo', 'args' => array( 'value' => 'F' ) ) ), + ), + ), +); + +$sync_roles_result = $sync_roles->invoke( null, $roles_valid, array(), $handlers ); +$async_roles_result = $async_roles->invoke( null, $roles_valid ); + +smoke_assert( false, is_wp_error( $sync_roles_result ), 'roles sync ACCEPTS the valid spec (no WP_Error)', $failures, $passes ); +smoke_assert( false, is_wp_error( $async_roles_result ), 'roles async ACCEPTS the valid spec (no WP_Error)', $failures, $passes ); +smoke_assert( 'roles', is_array( $sync_roles_result ) ? ( $sync_roles_result['shape'] ?? '' ) : '', 'roles sync valid path produces the roles shape (no regression)', $failures, $passes ); +smoke_assert( 'fuse', is_array( $sync_roles_result ) ? ( $sync_roles_result['aggregator'] ?? '' ) : '', 'roles sync valid path ran the aggregator', $failures, $passes ); +// Async plan: 2 sibling descriptors dispatched, aggregator deferred into the plan. +smoke_assert( 2, is_array( $async_roles_result ) ? count( $async_roles_result['branches'] ?? array() ) : -1, 'roles async valid path builds one descriptor per sibling', $failures, $passes ); +smoke_assert( 'fuse', is_array( $async_roles_result ) ? ( $async_roles_result['aggregate']['aggregator_role'] ?? '' ) : '', 'roles async valid path carries the aggregator role in the collect plan', $failures, $passes ); + +// ── 4. Map: the SAME valid spec is ACCEPTED on both paths ───────────────────── + +$map_valid = array( + 'items' => array( 10, 20 ), + 'as' => 'num', + 'steps' => array( array( 'id' => 'd', 'type' => 'ability', 'ability' => 'demo/echo', 'args' => array( 'value' => '${vars.num}' ) ) ), +); + +$sync_map_result = $sync_map->invoke( null, $map_valid, array(), $handlers ); +$async_map_result = $async_map->invoke( null, $map_valid ); + +smoke_assert( false, is_wp_error( $sync_map_result ), 'map sync ACCEPTS the valid spec (no WP_Error)', $failures, $passes ); +smoke_assert( false, is_wp_error( $async_map_result ), 'map async ACCEPTS the valid spec (no WP_Error)', $failures, $passes ); +smoke_assert( 'map', is_array( $sync_map_result ) ? ( $sync_map_result['shape'] ?? '' ) : '', 'map sync valid path produces the map shape (no regression)', $failures, $passes ); +smoke_assert( 2, is_array( $sync_map_result ) ? ( $sync_map_result['count'] ?? 0 ) : -1, 'map sync valid path fans out one branch per item', $failures, $passes ); +smoke_assert( 2, is_array( $async_map_result ) ? count( $async_map_result['branches'] ?? array() ) : -1, 'map async valid path builds one descriptor per item', $failures, $passes ); + +echo "Passed: {$passes}, Failed: " . count( $failures ) . "\n"; +exit( count( $failures ) > 0 ? 1 : 0 );