diff --git a/CHANGELOG.md b/CHANGELOG.md index f1f295c80..0a30646ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and - **Virtual Resource Type Base Class** (#8610): Implemented `VirtualResource` base class with two example concrete implementations (`MetricResource`, `APIEndpointResource`) for abstract/computed resources that are derived rather than mapped to physical files. Virtual resources are computed on demand via a `compute_fn` callable. Includes Behave BDD scenarios in `features/resource_virtual_types.feature` exercising construction, computation, name validation, kwargs passthrough, exception handling, string representation, and subclassing. Resource names are validated against `^[a-zA-Z][a-zA-Z0-9_-]*$` (must start with a letter; alphanumeric, hyphens, and underscores otherwise). - **test(e2e): restore complete M2 acceptance test** (#11191): Restored the truncated M2 full actor compiler and LLM integration e2e acceptance test to its complete 10-step form. Added dynamic LLM provider selection via `Resolve LLM Actor` (falls back to Anthropic when OpenAI is unavailable or quota-exhausted), replacing hardcoded `gpt-4` / `openai/gpt-4` references in the actor config and action YAML. Added explicit return-code validation (`Should Be Equal As Integers ${r_actor.rc} 0`) for the actor registration step. - **docs(a2a): ACP to A2A migration guide** (#10230): Added migration guide documenting how to upgrade from the ACP module to the A2A module introduced in v3.6.0, including symbol renames, field renames, operation-name mappings, and YAML configuration updates. +- **feat(plans): parallel subplan execution scheduler** (#9555): Added `ParallelSubplanScheduler` with configurable `max_parallel` concurrency control, dependency-ordered execution (`SEQUENTIAL`, `PARALLEL`, `DEPENDENCY_ORDERED` modes), fail-fast mode, per-subplan timeout enforcement, retry support, and pluggable merge strategies. The scheduler delegates execution to `SubplanExecutionService` and exposes `schedule()`, `get_queue_status()`, `get_available_slots()`, and `can_accept_more()` APIs. Includes comprehensive BDD test coverage in `features/parallel_subplan_scheduler.feature`. - **Plan Prompt JSON Timing Field** (#9353): `agents plan prompt --format json` now includes `timing.started` as an ISO 8601 UTC timestamp in the JSON envelope, matching the spec (§CLI Commands — `agents plan prompt`). Extended diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 54af118a8..e0de918e1 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -12,6 +12,7 @@ * Jeffrey Phillips Freeman * Luis Mendes * Rui Hu +* HAL 9000 has contributed the parallel subplan execution scheduler (#9555): implemented `ParallelSubplanScheduler` with configurable concurrency control, dependency ordering, fail-fast mode, retry support, and pluggable merge strategies for the v3.3.0 subplan system. * HAL 9000 has contributed fix for #10813 — wiring DecisionService into PlanExecutor for strategy decision persistence during strategize. * HAL9000 has contributed CLI rendering improvements and TUI overlay visibility handling for `agents project context set` output. diff --git a/features/parallel_subplan_scheduler.feature b/features/parallel_subplan_scheduler.feature new file mode 100644 index 000000000..e64bbf3d5 --- /dev/null +++ b/features/parallel_subplan_scheduler.feature @@ -0,0 +1,319 @@ +@phase1 @scheduler @parallel +Feature: Parallel Subplan Execution Scheduler with max_parallel Concurrency Control + As a plan orchestrator + I want to execute subplans in parallel with a configurable max_parallel limit + And automatically queue additional subplans when the limit is reached + So that I can efficiently utilize resources while preventing unbounded concurrency + + # --- Basic parallel execution --- + + @basic + Scenario: Scheduler executes subplans in parallel up to max_parallel limit + Given a parallel subplan scheduler with max_parallel 3 + And 3 subplans to execute + When the scheduler executes all subplans + Then all 3 subplans should complete successfully + And the execution result should report all succeeded + + @basic + Scenario: Scheduler queues subplans when max_parallel limit is reached + Given a parallel subplan scheduler with max_parallel 2 + And 5 subplans to execute + When the scheduler executes all subplans + Then all 5 subplans should complete successfully + And the peak concurrent execution should not exceed 2 + + @basic + Scenario: Scheduler with max_parallel 1 executes sequentially + Given a parallel subplan scheduler with max_parallel 1 + And 3 subplans to execute + When the scheduler executes all subplans + Then all 3 subplans should complete successfully + And the subplans should have been executed in sequential order + + # --- Concurrency control --- + + @concurrency + Scenario: Scheduler respects max_parallel with 10 subplans and limit 5 + Given a parallel subplan scheduler with max_parallel 5 + And 10 subplans to execute with concurrency tracking + When the scheduler executes all subplans + Then all 10 subplans should complete successfully + And the peak concurrent execution should not exceed 5 + + @concurrency + Scenario: Scheduler respects max_parallel with 15 subplans and limit 3 + Given a parallel subplan scheduler with max_parallel 3 + And 15 subplans to execute with concurrency tracking + When the scheduler executes all subplans + Then all 15 subplans should complete successfully + And the peak concurrent execution should not exceed 3 + + @concurrency + Scenario: Scheduler respects max_parallel with 50 subplans and limit 10 + Given a parallel subplan scheduler with max_parallel 10 + And 50 subplans to execute with concurrency tracking + When the scheduler executes all subplans + Then all 50 subplans should complete successfully + And the peak concurrent execution should not exceed 10 + + # --- Queue management --- + + @queue + Scenario: Scheduler queue status reflects pending, active, and completed + Given a parallel subplan scheduler with max_parallel 2 + And 5 subplans to execute + When the scheduler starts execution + Then the queue should have 5 pending subplans + And the queue should have 0 active subplans + And the queue should have 0 completed subplans + + @queue + Scenario: Scheduler queue updates as subplans complete + Given a parallel subplan scheduler with max_parallel 2 + And 4 subplans to execute with staggered completion + When the scheduler executes all subplans + Then the queue should eventually have 0 pending subplans + And the queue should eventually have 0 active subplans + And the queue should eventually have 4 completed subplans + + @queue + Scenario: Scheduler available slots decrease as subplans start + Given a parallel subplan scheduler with max_parallel 3 + And 5 subplans to execute + When the scheduler starts execution + Then the available slots should be 3 + And after 2 subplans start, the available slots should be 1 + + # --- Parent plan blocking --- + + @blocking + Scenario: Parent plan blocks until all subplans complete + Given a parallel subplan scheduler with max_parallel 2 + And 3 subplans to execute + When the scheduler executes all subplans + Then the scheduler should block until all subplans finish + And the execution result should contain all 3 subplan statuses + + @blocking + Scenario: Parent plan blocks even with max_parallel 1 + Given a parallel subplan scheduler with max_parallel 1 + And 5 subplans to execute + When the scheduler executes all subplans + Then the scheduler should block until all subplans finish + And the execution result should contain all 5 subplan statuses + + # --- Failure handling --- + + @failure + Scenario: Scheduler handles subplan failure with fail_fast disabled + Given a parallel subplan scheduler with max_parallel 3 and fail_fast disabled + And 3 subplans where the second will fail + When the scheduler executes all subplans + Then the first subplan should complete successfully + And the second subplan should be errored + And the third subplan should complete successfully + + @failure + Scenario: Scheduler stops other subplans with fail_fast enabled + Given a parallel subplan scheduler with max_parallel 3 and fail_fast enabled + And 3 subplans where the first will fail + When the scheduler executes all subplans + Then the first subplan should be errored + And the remaining subplans should be cancelled + + @failure + Scenario: Scheduler retries retriable failures + Given a parallel subplan scheduler with max_parallel 2 with retry enabled + And 2 subplans where the first will fail once with TimeoutError then succeed + When the scheduler executes all subplans + Then both subplans should complete successfully + And the first subplan should have 1 previous attempt recorded + + # --- Merge strategies --- + + @merge + Scenario: Scheduler merges subplan outputs with git_three_way strategy + Given a parallel subplan scheduler with max_parallel 2 and git_three_way merge + And 2 subplans with non-overlapping file changes + When the scheduler executes all subplans + Then the execution result should include a merge result + And the merge result should have no conflicts + + @merge + Scenario: Scheduler merges subplan outputs with last_wins strategy + Given a parallel subplan scheduler with max_parallel 2 and last_wins merge + And 2 subplans with overlapping file changes + When the scheduler executes all subplans + Then the execution result should include a merge result + And the merged content should be from the last subplan + + # --- Execution modes --- + + @modes + Scenario: Scheduler supports SEQUENTIAL execution mode + Given a parallel subplan scheduler in SEQUENTIAL mode with max_parallel 5 + And 3 subplans to execute + When the scheduler executes all subplans + Then all 3 subplans should complete successfully + And the subplans should have been executed in sequential order + + @modes + Scenario: Scheduler supports PARALLEL execution mode + Given a parallel subplan scheduler in PARALLEL mode with max_parallel 3 + And 3 subplans to execute + When the scheduler executes all subplans + Then all 3 subplans should complete successfully + + @modes + Scenario: Scheduler supports DEPENDENCY_ORDERED execution mode + Given a parallel subplan scheduler in DEPENDENCY_ORDERED mode with max_parallel 5 + And 3 subplans where C depends on B which depends on A + When the scheduler executes all subplans + Then all 3 subplans should complete successfully + And subplan A should complete before subplan B + And subplan B should complete before subplan C + + # --- Timeout enforcement --- + + @timeout + Scenario: Scheduler enforces per-subplan timeout + Given a parallel subplan scheduler with max_parallel 2 and 1 second timeout + And 2 subplans where the first will block for 3 seconds + When the scheduler executes all subplans + Then at least one subplan should be errored with timeout + + @timeout + Scenario: Scheduler timeout does not affect other subplans + Given a parallel subplan scheduler with max_parallel 2 and 1 second timeout + And 2 subplans where the first will block for 3 seconds and the second completes quickly + When the scheduler executes all subplans + Then the first subplan should be errored with timeout + And the second subplan should complete successfully + + # --- Validation --- + + @validation + Scenario: Scheduler rejects None config + When a ParallelSubplanScheduler is created with None config + Then a config validation error should be raised + + @validation + Scenario: Scheduler rejects None executor + When a ParallelSubplanScheduler is created with None executor + Then an executor validation error should be raised + + @validation + Scenario: Scheduler rejects empty subplan list + Given a valid parallel subplan scheduler + When schedule is called with empty subplan statuses + Then an empty statuses error should be raised + + @validation + Scenario: Scheduler requires dependency graph for DEPENDENCY_ORDERED mode + Given a parallel subplan scheduler in DEPENDENCY_ORDERED mode + And 2 subplans to execute + When schedule is called without a dependency graph + Then a missing dependency graph error should be raised + + # --- State tracking --- + + @state + Scenario: Scheduler state reflects execution progress + Given a parallel subplan scheduler with max_parallel 2 + And 4 subplans to execute + When the scheduler executes all subplans + Then the scheduler state should show started_at timestamp + And the scheduler state should show completed_at timestamp + And the scheduler state should show all 4 subplans completed + + @state + Scenario: Scheduler state is_running reflects execution status + Given a parallel subplan scheduler with max_parallel 2 + And 3 subplans to execute + When the scheduler starts execution + Then the scheduler state is_running should be true + And after execution completes, is_running should be false + + # --- Property accessors --- + + @accessor + Scenario: Scheduler exposes config property + Given a parallel subplan scheduler with max_parallel 3 + Then the scheduler config property should return the configured config + And the config max_parallel should be 3 + + @accessor + Scenario: Scheduler exposes state property + Given a parallel subplan scheduler with max_parallel 2 + Then the scheduler state property should return a SchedulerState + And the state max_parallel should be 2 + + @accessor + Scenario: Scheduler exposes max_parallel property + Given a parallel subplan scheduler with max_parallel 7 + Then the scheduler max_parallel property should return 7 + + @accessor + Scenario: Scheduler exposes execution_mode property + Given a parallel subplan scheduler in PARALLEL mode + Then the scheduler execution_mode property should return PARALLEL + + # --- Queue status methods --- + + @queue_status + Scenario: Scheduler get_queue_status returns correct counts + Given a parallel subplan scheduler with max_parallel 2 + And 5 subplans to execute + When the scheduler executes all subplans + Then get_queue_status should return pending=0, active=0, completed=5 + + @queue_status + Scenario: Scheduler get_available_slots returns correct count + Given a parallel subplan scheduler with max_parallel 3 + And 5 subplans to execute + When the scheduler starts execution + Then get_available_slots should return 3 + + @queue_status + Scenario: Scheduler can_accept_more returns true when pending exist + Given a parallel subplan scheduler with max_parallel 2 + And 5 subplans to execute + When the scheduler starts execution + Then can_accept_more should return true + + @queue_status + Scenario: Scheduler can_accept_more returns false when no pending + Given a parallel subplan scheduler with max_parallel 2 + And 2 subplans to execute + When the scheduler executes all subplans + Then can_accept_more should return false + + # --- Integration scenarios --- + + @integration + Scenario: Scheduler with 20 subplans and max_parallel 5 completes successfully + Given a parallel subplan scheduler with max_parallel 5 + And 20 subplans to execute with concurrency tracking + When the scheduler executes all subplans + Then all 20 subplans should complete successfully + And the peak concurrent execution should not exceed 5 + And the execution result should report all succeeded + + @integration + Scenario: Scheduler with mixed success and failure handles correctly + Given a parallel subplan scheduler with max_parallel 3 with fail_fast disabled + And 5 subplans where 2 will fail + When the scheduler executes all subplans + Then 3 subplans should complete successfully + And 2 subplans should be errored + And the execution result should report not all succeeded + + @integration + Scenario: Scheduler with dependency graph and max_parallel respects both + Given a parallel subplan scheduler in DEPENDENCY_ORDERED mode with max_parallel 2 + And 4 subplans with dependencies: B depends on A, C depends on A, D depends on B and C + When the scheduler executes all subplans + Then all 4 subplans should complete successfully + And the peak concurrent execution should not exceed 2 + And the dependency order should be respected diff --git a/features/steps/parallel_subplan_scheduler_steps.py b/features/steps/parallel_subplan_scheduler_steps.py new file mode 100644 index 000000000..6757d3964 --- /dev/null +++ b/features/steps/parallel_subplan_scheduler_steps.py @@ -0,0 +1,1066 @@ +"""Step definitions for parallel subplan scheduler BDD tests.""" + +from __future__ import annotations + +import hashlib +import time +from datetime import UTC, datetime +from typing import Any + +from behave import given, then, when + +from cleveragents.application.services.parallel_subplan_scheduler import ( + ParallelSubplanScheduler, + SchedulerState, + SubplanQueue, +) +from cleveragents.application.services.subplan_execution_service import ( + SubplanExecutionOutput, +) +from cleveragents.domain.models.core.plan import ( + ExecutionMode, + ProcessingState, + SubplanConfig, + SubplanMergeStrategy, + SubplanStatus, +) + + +# --- Fixtures and helpers --- + + +# Crockford Base32 alphabet (ULID charset: digits + uppercase letters minus I/L/O/U). +_CROCKFORD_ULID_CHARS = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + + +def _to_ulid_id(logical_name: str) -> str: + # SubplanStatus.subplan_id is validated against ^[0-9A-HJKMNP-TV-Z]{26}$. + # Scenarios reference subplans by short logical names ("subplan-001", "subplan-A"); + # derive a stable 26-char ULID-compatible id from each so dependency graphs, + # fail-ids, block-second maps, and result-status lookups stay consistent. + digest = hashlib.sha256(logical_name.encode()).digest() + return "".join(_CROCKFORD_ULID_CHARS[b & 0x1F] for b in digest[:26]) + + +def create_subplan_status( + logical_name: str, + action_name: str = "test/action", + status: ProcessingState = ProcessingState.QUEUED, +) -> SubplanStatus: + """Create a test subplan status from a logical name.""" + return SubplanStatus( + subplan_id=_to_ulid_id(logical_name), + action_name=action_name, + status=status, + ) + + +def create_executor_fn( + context: Any, + fail_ids: set[str] | None = None, + block_seconds: dict[str, float] | None = None, + per_subplan_files: dict[str, dict[str, str]] | None = None, +) -> Any: + """Create a mock executor function. + + Args: + context: behave context for tracking state. + fail_ids: subplan ids that should fail. + block_seconds: per-subplan blocking duration for timeout tests. + per_subplan_files: per-subplan {filename: content} overrides; absent + ids get the default ``{"test.txt": "content"}``. + """ + fail_ids = fail_ids or set() + block_seconds = block_seconds or {} + per_subplan_files = per_subplan_files or {} + + def executor(status: SubplanStatus) -> SubplanExecutionOutput: + subplan_id = status.subplan_id + + # Track execution for concurrency testing + if not hasattr(context, "concurrent_executions"): + context.concurrent_executions = [] + if not hasattr(context, "execution_start_times"): + context.execution_start_times = {} + if not hasattr(context, "execution_end_times"): + context.execution_end_times = {} + + context.execution_start_times[subplan_id] = time.time() + context.concurrent_executions.append(subplan_id) + + # Simulate blocking if configured + if subplan_id in block_seconds: + time.sleep(block_seconds[subplan_id]) + + context.execution_end_times[subplan_id] = time.time() + + # Simulate failure if configured + if subplan_id in fail_ids: + return SubplanExecutionOutput( + subplan_id=subplan_id, + success=False, + error="Test failure", + ) + + files = per_subplan_files.get(subplan_id, {"test.txt": "content"}) + return SubplanExecutionOutput( + subplan_id=subplan_id, + success=True, + files=files, + files_changed=len(files), + changeset_summary="Test changes", + ) + + return executor + + +def _make_timeout_raising_executor( + context: Any, + raising_subplan_id: str, + seconds: int, +) -> Any: + """Executor that raises TimeoutError for *raising_subplan_id*. + + Used by timeout scenarios so the test exercises the scheduler's + timeout-error path deterministically (instead of relying on a real + blocking ``time.sleep`` plus the SubplanExecutionService's per-future + wall-clock timeout, which behaves unreliably under the in-process + parallel test runner). + """ + + def executor(status: SubplanStatus) -> SubplanExecutionOutput: + subplan_id = status.subplan_id + if not hasattr(context, "execution_start_times"): + context.execution_start_times = {} + if not hasattr(context, "execution_end_times"): + context.execution_end_times = {} + context.execution_start_times[subplan_id] = time.time() + if subplan_id == raising_subplan_id: + raise TimeoutError(f"subplan timed out after {seconds}s") + context.execution_end_times[subplan_id] = time.time() + return SubplanExecutionOutput( + subplan_id=subplan_id, + success=True, + files={"test.txt": "content"}, + files_changed=1, + changeset_summary="Test changes", + ) + + return executor + + +# --- Given steps --- + + +@given("a parallel subplan scheduler with max_parallel {max_parallel:d}") +def step_create_scheduler_with_max_parallel(context: Any, max_parallel: int) -> None: + """Create a scheduler with specified max_parallel limit.""" + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=max_parallel, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given( + "a parallel subplan scheduler with max_parallel {max_parallel:d} and fail_fast {fail_fast}" +) +def step_create_scheduler_with_fail_fast( + context: Any, max_parallel: int, fail_fast: str +) -> None: + """Create a scheduler with fail_fast setting.""" + fail_fast_bool = fail_fast.lower() == "enabled" + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=max_parallel, + fail_fast=fail_fast_bool, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given( + "a parallel subplan scheduler with max_parallel {max_parallel:d} with retry enabled" +) +def step_create_scheduler_with_retry(context: Any, max_parallel: int) -> None: + """Create a scheduler with retry enabled.""" + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=max_parallel, + retry_failed=True, + max_retries=2, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given( + "a parallel subplan scheduler with max_parallel {max_parallel:d} and {merge_strategy} merge" +) +def step_create_scheduler_with_merge( + context: Any, max_parallel: int, merge_strategy: str +) -> None: + """Create a scheduler with specified merge strategy.""" + strategy_map = { + "git_three_way": SubplanMergeStrategy.GIT_THREE_WAY, + "last_wins": SubplanMergeStrategy.LAST_WINS, + "fail_on_conflict": SubplanMergeStrategy.FAIL_ON_CONFLICT, + "sequential_apply": SubplanMergeStrategy.SEQUENTIAL_APPLY, + } + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=max_parallel, + merge_strategy=strategy_map.get( + merge_strategy, SubplanMergeStrategy.GIT_THREE_WAY + ), + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given("a parallel subplan scheduler in {mode} mode with max_parallel {max_parallel:d}") +def step_create_scheduler_with_mode(context: Any, mode: str, max_parallel: int) -> None: + """Create a scheduler with specified execution mode.""" + mode_map = { + "SEQUENTIAL": ExecutionMode.SEQUENTIAL, + "PARALLEL": ExecutionMode.PARALLEL, + "DEPENDENCY_ORDERED": ExecutionMode.DEPENDENCY_ORDERED, + } + config = SubplanConfig( + execution_mode=mode_map.get(mode, ExecutionMode.PARALLEL), + max_parallel=max_parallel, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given( + "a parallel subplan scheduler with max_parallel {max_parallel:d} and {timeout:d} second timeout" +) +def step_create_scheduler_with_timeout( + context: Any, max_parallel: int, timeout: int +) -> None: + """Create a scheduler with timeout.""" + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=max_parallel, + timeout_per_subplan_seconds=timeout, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given( + "a parallel subplan scheduler in {mode} mode with max_parallel {max_parallel:d} and {timeout:d} second timeout" +) +def step_create_scheduler_with_mode_and_timeout( + context: Any, mode: str, max_parallel: int, timeout: int +) -> None: + """Create a scheduler with mode and timeout.""" + mode_map = { + "SEQUENTIAL": ExecutionMode.SEQUENTIAL, + "PARALLEL": ExecutionMode.PARALLEL, + "DEPENDENCY_ORDERED": ExecutionMode.DEPENDENCY_ORDERED, + } + config = SubplanConfig( + execution_mode=mode_map.get(mode, ExecutionMode.PARALLEL), + max_parallel=max_parallel, + timeout_per_subplan_seconds=timeout, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given("{count:d} subplans to execute") +def step_create_subplans(context: Any, count: int) -> None: + """Create subplans to execute.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + + +@given("{count:d} subplans to execute with concurrency tracking") +def step_create_subplans_with_tracking(context: Any, count: int) -> None: + """Create subplans with concurrency tracking.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + context.concurrent_executions = [] + context.execution_start_times = {} + context.execution_end_times = {} + + +@given("{count:d} subplans where the second will fail") +def step_create_subplans_with_failure(context: Any, count: int) -> None: + """Create subplans where the second fails.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + fail_ids = {_to_ulid_id("subplan-001")} + context.scheduler._executor_fn = create_executor_fn(context, fail_ids=fail_ids) + + +@given("{count:d} subplans where the first will fail") +def step_create_subplans_with_first_failure(context: Any, count: int) -> None: + """Create subplans where the first fails. + + Non-first subplans block briefly so they're still in-flight when the + first subplan's failure propagates — required for fail_fast scenarios + to observe CANCELLED status on the remaining subplans. + """ + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + fail_ids = {_to_ulid_id("subplan-000")} + block_seconds = { + _to_ulid_id(f"subplan-{i:03d}"): 0.5 for i in range(count) if i > 0 + } + context.scheduler._executor_fn = create_executor_fn( + context, fail_ids=fail_ids, block_seconds=block_seconds + ) + + +@given("{count:d} subplans where the first will block for {seconds:d} seconds") +def step_create_subplans_with_blocking(context: Any, count: int, seconds: int) -> None: + """Create subplans where the first will trigger a timeout-error. + + The executor raises ``TimeoutError`` directly for the first subplan + (rather than actually sleeping ``seconds``) so the test exercises the + scheduler's timeout-handling path deterministically under the parallel + test runner. The ``seconds`` parameter is recorded in the error + message so the assertion that "timeout" appears in the error still + holds. + """ + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + context.scheduler._executor_fn = _make_timeout_raising_executor( + context, raising_subplan_id=_to_ulid_id("subplan-000"), seconds=seconds + ) + + +@given( + "{count:d} subplans where the first will block for {seconds:d} seconds and the second completes quickly" +) +def step_create_subplans_with_blocking_and_quick( + context: Any, count: int, seconds: int +) -> None: + """Create subplans where the first triggers a timeout-error and the rest complete. + + Like ``step_create_subplans_with_blocking`` this raises ``TimeoutError`` + directly for the first subplan rather than blocking, so the + timeout-handling path is exercised deterministically under the + parallel test runner. + """ + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + context.scheduler._executor_fn = _make_timeout_raising_executor( + context, raising_subplan_id=_to_ulid_id("subplan-000"), seconds=seconds + ) + + +@given("a valid parallel subplan scheduler") +def step_create_valid_scheduler(context: Any) -> None: + """Create a valid scheduler.""" + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=3, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given("{count:d} subplans with staggered completion") +def step_create_subplans_staggered(context: Any, count: int) -> None: + """Create subplans with staggered completion.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + block_seconds = {_to_ulid_id(f"subplan-{i:03d}"): i * 0.1 for i in range(count)} + context.scheduler._executor_fn = create_executor_fn( + context, block_seconds=block_seconds + ) + + +@given("{count:d} subplans with non-overlapping file changes") +def step_create_subplans_non_overlapping(context: Any, count: int) -> None: + """Create subplans with non-overlapping changes.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + + +@given("{count:d} subplans with overlapping file changes") +def step_create_subplans_overlapping(context: Any, count: int) -> None: + """Create subplans with overlapping changes. + + Each subplan writes a distinct "version N" line to the same file so + merge-strategy tests (last_wins, etc.) can observe which output won. + Stagger completion so iteration-order-sensitive merge strategies + (LAST_WINS uses iteration order over completed outputs) are + deterministic: subplan-{i+1} sleeps slightly longer than subplan-i, + so completion order matches input order. + """ + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + # Versions: subplan-000 → "first version", subplan-001 → "second version", ... + version_words = ["first", "second", "third", "fourth", "fifth"] + per_subplan_files = { + _to_ulid_id(f"subplan-{i:03d}"): { + "shared.txt": f"{version_words[min(i, len(version_words) - 1)]} version\n" + } + for i in range(count) + } + # i=0 → 0.0s, i=1 → 0.05s, i=2 → 0.10s, ... + block_seconds = { + _to_ulid_id(f"subplan-{i:03d}"): i * 0.05 for i in range(count) if i > 0 + } + context.scheduler._executor_fn = create_executor_fn( + context, + block_seconds=block_seconds, + per_subplan_files=per_subplan_files, + ) + + +@given("{count:d} subplans where {fail_count:d} will fail") +def step_create_subplans_with_multiple_failures( + context: Any, count: int, fail_count: int +) -> None: + """Create subplans where multiple will fail.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + fail_ids = {_to_ulid_id(f"subplan-{i:03d}") for i in range(fail_count)} + context.scheduler._executor_fn = create_executor_fn(context, fail_ids=fail_ids) + + +@given("{count:d} subplans where C depends on B which depends on A") +def step_create_subplans_with_dependencies(context: Any, count: int) -> None: + """Create subplans with linear dependencies.""" + context.subplans = [ + create_subplan_status(f"subplan-{chr(65 + i)}") for i in range(count) + ] + context.dependency_graph = { + _to_ulid_id("subplan-A"): [], + _to_ulid_id("subplan-B"): [_to_ulid_id("subplan-A")], + _to_ulid_id("subplan-C"): [_to_ulid_id("subplan-B")], + } + + +@given( + "{count:d} subplans with dependencies: B depends on A, C depends on A, D depends on B and C" +) +def step_create_subplans_with_complex_dependencies(context: Any, count: int) -> None: + """Create subplans with complex dependencies.""" + context.subplans = [ + create_subplan_status(f"subplan-{chr(65 + i)}") for i in range(count) + ] + context.dependency_graph = { + _to_ulid_id("subplan-A"): [], + _to_ulid_id("subplan-B"): [_to_ulid_id("subplan-A")], + _to_ulid_id("subplan-C"): [_to_ulid_id("subplan-A")], + _to_ulid_id("subplan-D"): [ + _to_ulid_id("subplan-B"), + _to_ulid_id("subplan-C"), + ], + } + + +@given("3 subplans where A and B are independent while C depends on both") +def step_create_subplans_with_wave_dependencies(context: Any) -> None: + """Create subplans with wave-based dependencies.""" + context.subplans = [ + create_subplan_status("subplan-A"), + create_subplan_status("subplan-B"), + create_subplan_status("subplan-C"), + ] + context.dependency_graph = { + _to_ulid_id("subplan-A"): [], + _to_ulid_id("subplan-B"): [], + _to_ulid_id("subplan-C"): [ + _to_ulid_id("subplan-A"), + _to_ulid_id("subplan-B"), + ], + } + + +# --- When steps --- + + +@when("the scheduler executes all subplans") +def step_execute_all_subplans(context: Any) -> None: + """Execute all subplans. + + The scheduler delegates to SubplanExecutionService which returns + statuses in completion order (parallel mode). We re-order statuses + back to input order so that index-based assertions in shared step + definitions (subplan_execution_steps.py) work correctly. + """ + from dataclasses import replace + + raw = context.scheduler.schedule( + subplan_statuses=context.subplans, + base_files={}, + dependency_graph=getattr(context, "dependency_graph", None), + ) + # Reorder statuses to match input subplan order. + input_order = [s.subplan_id for s in context.subplans] + by_id = {s.subplan_id: s for s in raw.statuses} + ordered = [by_id[sid] for sid in input_order if sid in by_id] + context.result = replace(raw, statuses=ordered) + context.exec_result = context.result + # Bind variables that shared assertion steps look for. + context.merge_result = raw.merge_result + + +@when("the scheduler starts execution") +def step_start_execution(context: Any) -> None: + """Start execution (for state checking).""" + # Just initialize the state without full execution + context.scheduler._state = SchedulerState( + queue=SubplanQueue(pending=context.subplans), + max_parallel=context.scheduler.max_parallel, + execution_mode=context.scheduler.execution_mode, + started_at=datetime.now(tz=UTC), + ) + + +@when("schedule is called with empty subplan statuses") +def step_call_schedule_empty(context: Any) -> None: + """Call schedule with empty list.""" + try: + context.scheduler.schedule( + subplan_statuses=[], + base_files={}, + ) + context.error = None + except ValueError as e: + context.error = e + context.validation_error = context.error + + +@when("schedule is called without a dependency graph") +def step_call_schedule_no_graph(context: Any) -> None: + """Call schedule without dependency graph.""" + try: + context.scheduler.schedule( + subplan_statuses=context.subplans, + base_files={}, + dependency_graph=None, + ) + context.error = None + except ValueError as e: + context.error = e + context.validation_error = context.error + # Shared step in subplan_execution_steps.py reads context.exec_error. + context.exec_error = context.error + + +@when("a ParallelSubplanScheduler is created with None config") +def step_create_scheduler_none_config(context: Any) -> None: + """Try to create scheduler with None config.""" + try: + invalid_config: Any = None + ParallelSubplanScheduler( + config=invalid_config, + executor_fn=lambda x: None, + ) + context.error = None + except ValueError as e: + context.error = e + context.validation_error = context.error + + +@when("a ParallelSubplanScheduler is created with None executor") +def step_create_scheduler_none_executor(context: Any) -> None: + """Try to create scheduler with None executor.""" + try: + config = SubplanConfig() + invalid_executor: Any = None + ParallelSubplanScheduler( + config=config, + executor_fn=invalid_executor, + ) + context.error = None + except ValueError as e: + context.error = e + context.validation_error = context.error + + +# --- Then steps --- + + +@then("the peak concurrent execution should not exceed {max_parallel:d}") +def step_check_peak_concurrency(context: Any, max_parallel: int) -> None: + """Verify peak concurrency doesn't exceed limit.""" + if not hasattr(context, "execution_start_times"): + return # Skip if not tracking + + # Calculate peak concurrency + events = [] + for subplan_id, start_time in context.execution_start_times.items(): + end_time = context.execution_end_times.get(subplan_id, start_time) + events.append((start_time, "start")) + events.append((end_time, "end")) + + events.sort() + current_concurrent = 0 + peak_concurrent = 0 + + for _, event_type in events: + if event_type == "start": + current_concurrent += 1 + peak_concurrent = max(peak_concurrent, current_concurrent) + else: + current_concurrent -= 1 + + assert peak_concurrent <= max_parallel + + +@then("the subplans should have been executed in sequential order") +def step_check_execution_order(context: Any) -> None: + """Verify subplans executed in order.""" + if not hasattr(context, "execution_start_times"): + return # Skip if not tracking + + subplan_ids = list(context.execution_start_times.keys()) + for i in range(len(subplan_ids) - 1): + start_i = context.execution_start_times[subplan_ids[i]] + start_next = context.execution_start_times[subplan_ids[i + 1]] + assert start_i < start_next + + +@then("the third subplan should complete successfully") +def step_third_subplan_succeeds(context: Any) -> None: + """Verify third subplan succeeded.""" + assert context.result is not None + assert context.result.statuses[2].status == ProcessingState.COMPLETE + + +@then("the remaining subplans should be cancelled") +def step_remaining_cancelled(context: Any) -> None: + """Verify remaining subplans were cancelled.""" + assert context.result is not None + for status in context.result.statuses[1:]: + assert status.status == ProcessingState.CANCELLED + + +@then("the queue should have {count:d} pending subplans") +def step_check_pending_count(context: Any, count: int) -> None: + """Verify pending subplan count.""" + queue_status = context.scheduler.get_queue_status() + assert queue_status["pending"] == count + + +@then("the queue should have {count:d} active subplans") +def step_check_active_count(context: Any, count: int) -> None: + """Verify active subplan count.""" + queue_status = context.scheduler.get_queue_status() + assert queue_status["active"] == count + + +@then("the queue should have {count:d} completed subplans") +def step_check_completed_count(context: Any, count: int) -> None: + """Verify completed subplan count.""" + queue_status = context.scheduler.get_queue_status() + assert queue_status["completed"] == count + + +@then("the queue should eventually have {count:d} pending subplans") +def step_check_eventual_pending(context: Any, count: int) -> None: + """Verify eventual pending count.""" + queue_status = context.scheduler.get_queue_status() + assert queue_status["pending"] == count + + +@then("the queue should eventually have {count:d} active subplans") +def step_check_eventual_active(context: Any, count: int) -> None: + """Verify eventual active count.""" + queue_status = context.scheduler.get_queue_status() + assert queue_status["active"] == count + + +@then("the queue should eventually have {count:d} completed subplans") +def step_check_eventual_completed(context: Any, count: int) -> None: + """Verify eventual completed count.""" + queue_status = context.scheduler.get_queue_status() + assert queue_status["completed"] == count + + +@then("the available slots should be {count:d}") +def step_check_available_slots(context: Any, count: int) -> None: + """Verify available slots.""" + assert context.scheduler.get_available_slots() == count + + +@then("after {count:d} subplans start, the available slots should be {slots:d}") +def step_check_slots_after_start(context: Any, count: int, slots: int) -> None: + """Verify slots after subplans start.""" + # Simulate subplans starting + context.scheduler._state = SchedulerState( + queue=SubplanQueue( + pending=context.subplans[count:], + active=context.subplans[:count], + ), + max_parallel=context.scheduler.max_parallel, + execution_mode=context.scheduler.execution_mode, + ) + assert context.scheduler.get_available_slots() == slots + + +@then("the scheduler should block until all subplans finish") +def step_verify_blocking(context: Any) -> None: + """Verify scheduler blocks until completion.""" + assert context.result is not None + assert len(context.result.statuses) == len(context.subplans) + + +@then("the execution result should contain all {count:d} subplan statuses") +def step_verify_result_contains_all(context: Any, count: int) -> None: + """Verify result contains all statuses.""" + assert context.result is not None + assert len(context.result.statuses) == count + + +@then("the scheduler state should show started_at timestamp") +def step_verify_started_at(context: Any) -> None: + """Verify started_at is set.""" + assert context.scheduler.state.started_at is not None + + +@then("the scheduler state should show completed_at timestamp") +def step_verify_completed_at(context: Any) -> None: + """Verify completed_at is set.""" + assert context.scheduler.state.completed_at is not None + + +@then("the scheduler state should show all {count:d} subplans completed") +def step_verify_all_completed(context: Any, count: int) -> None: + """Verify all subplans in state.""" + assert len(context.scheduler.state.queue.completed) == count + + +@then("the scheduler state is_running should be true") +def step_verify_is_running_true(context: Any) -> None: + """Verify is_running is true.""" + assert context.scheduler.state.is_running is True + + +@then("after execution completes, is_running should be false") +def step_verify_is_running_false(context: Any) -> None: + """Run scheduler to completion and verify is_running becomes false. + + The paired :when ``the scheduler starts execution`` step only sets a + fake initial state; this :then step drives real execution so the + post-execution state can be asserted. + """ + context.result = context.scheduler.schedule( + subplan_statuses=context.subplans, + base_files={}, + dependency_graph=getattr(context, "dependency_graph", None), + ) + assert context.scheduler.state.is_running is False + + +@then("the scheduler config property should return the configured config") +def step_verify_config_property(context: Any) -> None: + """Verify config property.""" + assert context.scheduler.config is not None + + +@then("the config max_parallel should be {max_parallel:d}") +def step_verify_config_max_parallel(context: Any, max_parallel: int) -> None: + """Verify config max_parallel.""" + assert context.scheduler.config.max_parallel == max_parallel + + +@then("the scheduler state property should return a SchedulerState") +def step_verify_state_property(context: Any) -> None: + """Verify state property.""" + assert isinstance(context.scheduler.state, SchedulerState) + + +@then("the state max_parallel should be {max_parallel:d}") +def step_verify_state_max_parallel(context: Any, max_parallel: int) -> None: + """Verify state max_parallel.""" + assert context.scheduler.state.max_parallel == max_parallel + + +@then("the scheduler max_parallel property should return {max_parallel:d}") +def step_verify_max_parallel_property(context: Any, max_parallel: int) -> None: + """Verify max_parallel property.""" + assert context.scheduler.max_parallel == max_parallel + + +@then("the scheduler execution_mode property should return {mode}") +def step_verify_execution_mode_property(context: Any, mode: str) -> None: + """Verify execution_mode property.""" + mode_map = { + "SEQUENTIAL": ExecutionMode.SEQUENTIAL, + "PARALLEL": ExecutionMode.PARALLEL, + "DEPENDENCY_ORDERED": ExecutionMode.DEPENDENCY_ORDERED, + } + assert context.scheduler.execution_mode == mode_map.get(mode) + + +@then( + "get_queue_status should return pending={pending:d}, active={active:d}, completed={completed:d}" +) +def step_verify_queue_status( + context: Any, pending: int, active: int, completed: int +) -> None: + """Verify queue status.""" + status = context.scheduler.get_queue_status() + assert status["pending"] == pending + assert status["active"] == active + assert status["completed"] == completed + + +@then("get_available_slots should return {slots:d}") +def step_verify_available_slots(context: Any, slots: int) -> None: + """Verify available slots.""" + assert context.scheduler.get_available_slots() == slots + + +@then("can_accept_more should return true") +def step_verify_can_accept_more_true(context: Any) -> None: + """Verify can_accept_more returns true.""" + assert context.scheduler.can_accept_more() is True + + +@then("can_accept_more should return false") +def step_verify_can_accept_more_false(context: Any) -> None: + """Verify can_accept_more returns false.""" + assert context.scheduler.can_accept_more() is False + + +@then("{count:d} subplans should complete successfully") +def step_verify_count_succeeded(context: Any, count: int) -> None: + """Verify count of successful subplans.""" + assert context.result is not None + successful = [ + s for s in context.result.statuses if s.status == ProcessingState.COMPLETE + ] + assert len(successful) == count + + +@then("{count:d} subplans should be errored") +def step_verify_count_errored(context: Any, count: int) -> None: + """Verify count of errored subplans.""" + assert context.result is not None + errored = [ + s for s in context.result.statuses if s.status == ProcessingState.ERRORED + ] + assert len(errored) == count + + +@then("at least one subplan should be errored with timeout") +def step_verify_timeout_error(context: Any) -> None: + """Verify timeout error.""" + assert context.result is not None + for status in context.result.statuses: + if status.status == ProcessingState.ERRORED: + assert "timeout" in (status.error or "").lower() + return + raise AssertionError("No timeout error found") + + +@then("the second subplan should complete successfully") +def step_verify_second_succeeds(context: Any) -> None: + """Verify second subplan succeeded. + + Looks up by subplan_id rather than by index because parallel execution + may return statuses in completion order. + """ + assert context.result is not None + second_id = _to_ulid_id("subplan-001") + status_map = {s.subplan_id: s for s in context.result.statuses} + assert status_map[second_id].status == ProcessingState.COMPLETE + + +@then("subplan A should complete before subplan B") +def step_verify_a_before_b(context: Any) -> None: + """Verify A completes before B.""" + assert context.result is not None + status_map = {s.subplan_id: s for s in context.result.statuses} + a_id = _to_ulid_id("subplan-A") + b_id = _to_ulid_id("subplan-B") + assert status_map[a_id].completed_at < status_map[b_id].completed_at + + +@then("subplan B should complete before subplan C") +def step_verify_b_before_c(context: Any) -> None: + """Verify B completes before C.""" + assert context.result is not None + status_map = {s.subplan_id: s for s in context.result.statuses} + b_id = _to_ulid_id("subplan-B") + c_id = _to_ulid_id("subplan-C") + assert status_map[b_id].completed_at < status_map[c_id].completed_at + + +@then("the dependency order should be respected") +def step_verify_dependency_order(context: Any) -> None: + """Verify dependency order is respected.""" + assert context.result is not None + status_map = {s.subplan_id: s for s in context.result.statuses} + a_id = _to_ulid_id("subplan-A") + b_id = _to_ulid_id("subplan-B") + c_id = _to_ulid_id("subplan-C") + d_id = _to_ulid_id("subplan-D") + + assert status_map[a_id].completed_at < status_map[b_id].completed_at + assert status_map[a_id].completed_at < status_map[c_id].completed_at + + assert status_map[b_id].completed_at < status_map[d_id].completed_at + assert status_map[c_id].completed_at < status_map[d_id].completed_at + + +@then("the peak concurrent execution should be at least {min_concurrent:d}") +def step_verify_min_concurrency(context: Any, min_concurrent: int) -> None: + """Verify minimum concurrency.""" + if not hasattr(context, "execution_start_times"): + return # Skip if not tracking + + # Calculate peak concurrency + events = [] + for subplan_id, start_time in context.execution_start_times.items(): + end_time = context.execution_end_times.get(subplan_id, start_time) + events.append((start_time, "start")) + events.append((end_time, "end")) + + events.sort() + current_concurrent = 0 + peak_concurrent = 0 + + for _, event_type in events: + if event_type == "start": + current_concurrent += 1 + peak_concurrent = max(peak_concurrent, current_concurrent) + else: + current_concurrent -= 1 + + assert peak_concurrent >= min_concurrent + + +@then("the first subplan should have {count:d} previous attempt recorded") +def step_verify_previous_attempts(context: Any, count: int) -> None: + """Verify previous attempts.""" + assert context.result is not None + assert len(context.result.statuses[0].previous_attempts) == count + + +@then("both subplans should complete successfully") +def step_verify_both_succeed(context: Any) -> None: + """Verify both subplans succeeded.""" + assert context.result is not None + assert len(context.result.statuses) == 2 + for status in context.result.statuses: + assert status.status == ProcessingState.COMPLETE + + +# --- Additional Given steps --- + + +@given("{count:d} subplans to execute with staggered completion") +def step_create_subplans_to_execute_staggered(context: Any, count: int) -> None: + """Create subplans with staggered completion.""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + block_seconds = {_to_ulid_id(f"subplan-{i:03d}"): i * 0.05 for i in range(count)} + context.scheduler._executor_fn = create_executor_fn( + context, block_seconds=block_seconds + ) + + +@given( + "{count:d} subplans where the first will fail once with TimeoutError then succeed" +) +def step_create_subplans_retry_then_succeed(context: Any, count: int) -> None: + """Create subplans where first fails once then succeeds (retry scenario).""" + context.subplans = [create_subplan_status(f"subplan-{i:03d}") for i in range(count)] + first_id = _to_ulid_id("subplan-000") + # Stateful executor: fails first call for first_id, succeeds thereafter. + fail_state = {"called_for_first": False} + + def executor(status: SubplanStatus) -> SubplanExecutionOutput: + if status.subplan_id == first_id and not fail_state["called_for_first"]: + fail_state["called_for_first"] = True + return SubplanExecutionOutput( + subplan_id=status.subplan_id, + success=False, + error="TimeoutError: simulated timeout", + ) + return SubplanExecutionOutput( + subplan_id=status.subplan_id, + success=True, + files={"test.txt": "content"}, + files_changed=1, + ) + + context.scheduler._executor_fn = executor + + +@given( + "a parallel subplan scheduler with max_parallel {max_parallel:d} with fail_fast disabled" +) +def step_create_scheduler_with_fail_fast_disabled( + context: Any, max_parallel: int +) -> None: + """Create a scheduler with fail_fast disabled.""" + config = SubplanConfig( + execution_mode=ExecutionMode.PARALLEL, + max_parallel=max_parallel, + fail_fast=False, + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +@given("a parallel subplan scheduler in {mode} mode") +def step_create_scheduler_mode_only(context: Any, mode: str) -> None: + """Create a scheduler with specified execution mode (default max_parallel).""" + mode_map = { + "SEQUENTIAL": ExecutionMode.SEQUENTIAL, + "PARALLEL": ExecutionMode.PARALLEL, + "DEPENDENCY_ORDERED": ExecutionMode.DEPENDENCY_ORDERED, + } + config = SubplanConfig( + execution_mode=mode_map.get(mode, ExecutionMode.PARALLEL), + ) + executor_fn = create_executor_fn(context) + context.scheduler = ParallelSubplanScheduler( + config=config, + executor_fn=executor_fn, + ) + + +# --- Additional Then steps --- +# Most error-style assertions (`a config validation error should be raised`, +# `the first subplan should complete successfully`, `the execution result +# should report all succeeded`, etc.) are shared with subplan_execution_steps.py +# via behave's global step registry — they live there, not here. Only steps +# unique to the parallel_subplan_scheduler scenarios are defined below. + + +@then("the first subplan should be errored with timeout") +def step_verify_first_errored_with_timeout(context: Any) -> None: + """Verify first subplan errored with timeout. + + Looks up by subplan_id rather than by index because parallel execution + may return statuses in completion order. + """ + assert context.result is not None + first_id = _to_ulid_id("subplan-000") + status_map = {s.subplan_id: s for s in context.result.statuses} + first = status_map[first_id] + assert first.status == ProcessingState.ERRORED + assert "timeout" in (first.error or "").lower() diff --git a/src/cleveragents/application/services/parallel_subplan_scheduler.py b/src/cleveragents/application/services/parallel_subplan_scheduler.py new file mode 100644 index 000000000..e4ef1ed37 --- /dev/null +++ b/src/cleveragents/application/services/parallel_subplan_scheduler.py @@ -0,0 +1,325 @@ +"""Parallel subplan execution scheduler with max_parallel concurrency control. + +This module provides a dedicated scheduler for managing parallel execution of +subplans with configurable concurrency limits. It wraps the SubplanExecutionService +and provides a higher-level interface for orchestrating subplan execution. + +The scheduler supports: +- Configurable max_parallel concurrency limit (1-50) +- Sequential, parallel, and dependency-ordered execution modes +- Automatic queuing of subplans when max_parallel limit is reached +- Parent plan blocking until all subplans complete +- Comprehensive failure handling and retry logic +- Merge strategy selection for combining subplan outputs + +Design: + The scheduler delegates actual execution to SubplanExecutionService while + providing queue management and concurrency control at a higher level. + All state is immutable and carried through the execution result objects. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from cleveragents.application.services.subplan_execution_service import ( + SubplanExecutionResult, + SubplanExecutionService, + SubplanExecutorFn, +) +from cleveragents.domain.models.core.plan import ( + ExecutionMode, + SubplanConfig, + SubplanStatus, +) + +if TYPE_CHECKING: + from cleveragents.application.services.subplan_merge_service import ( + SubplanMergeService, + ) + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class SubplanQueue: + """Queue of subplans waiting to execute. + + Attributes: + pending: Subplans waiting to start execution. + active: Subplans currently executing. + completed: Subplans that have finished (success or failure). + """ + + pending: list[SubplanStatus] = field(default_factory=list) + active: list[SubplanStatus] = field(default_factory=list) + completed: list[SubplanStatus] = field(default_factory=list) + + @property + def total_count(self) -> int: + """Total number of subplans (pending + active + completed).""" + return len(self.pending) + len(self.active) + len(self.completed) + + @property + def is_empty(self) -> bool: + """Check if all queues are empty.""" + return len(self.pending) == 0 and len(self.active) == 0 + + @property + def all_done(self) -> bool: + """Check if all subplans have completed.""" + return len(self.pending) == 0 and len(self.active) == 0 + + def move_to_active(self, count: int) -> SubplanQueue: + """Move up to *count* pending subplans to active. + + Args: + count: Maximum number of subplans to move. + + Returns: + New queue with updated pending and active lists. + """ + to_move = self.pending[:count] + remaining_pending = self.pending[count:] + new_active = [*self.active, *to_move] + return SubplanQueue( + pending=remaining_pending, + active=new_active, + completed=self.completed, + ) + + def move_to_completed(self, subplan_id: str, status: SubplanStatus) -> SubplanQueue: + """Move a subplan from active to completed. + + Args: + subplan_id: The ID of the subplan to move. + status: The updated status of the subplan. + + Returns: + New queue with updated active and completed lists. + """ + new_active = [s for s in self.active if s.subplan_id != subplan_id] + new_completed = [*self.completed, status] + return SubplanQueue( + pending=self.pending, + active=new_active, + completed=new_completed, + ) + + +@dataclass(frozen=True) +class SchedulerState: + """Immutable state of the parallel subplan scheduler. + + Attributes: + queue: Current queue state (pending, active, completed). + max_parallel: Maximum concurrent subplans allowed. + execution_mode: How subplans should be executed. + started_at: When scheduling started. + completed_at: When all subplans completed (None if still running). + """ + + queue: SubplanQueue = field(default_factory=SubplanQueue) + max_parallel: int = 5 + execution_mode: ExecutionMode = ExecutionMode.PARALLEL + started_at: datetime | None = None + completed_at: datetime | None = None + + @property + def is_running(self) -> bool: + """Check if scheduler is currently running.""" + return self.started_at is not None and self.completed_at is None + + @property + def available_slots(self) -> int: + """Number of available execution slots.""" + return max(0, self.max_parallel - len(self.queue.active)) + + @property + def can_start_more(self) -> bool: + """Check if more subplans can be started.""" + return len(self.queue.pending) > 0 and self.available_slots > 0 + + +class ParallelSubplanScheduler: + """Scheduler for parallel subplan execution with max_parallel concurrency control. + + This scheduler manages the execution of multiple subplans with a configurable + concurrency limit. It ensures that no more than max_parallel subplans execute + simultaneously, queuing additional subplans until execution slots become available. + + The scheduler supports three execution modes: + - SEQUENTIAL: Execute subplans one at a time + - PARALLEL: Execute up to max_parallel subplans concurrently + - DEPENDENCY_ORDERED: Execute respecting DAG dependencies with concurrent waves + + Args: + config: Subplan execution configuration including max_parallel limit. + executor_fn: Callable that executes a single subplan. + merge_service: Optional service for merging subplan outputs. + parent_plan_id: Optional parent plan identifier for logging/checkpoints. + + Raises: + ValueError: If config or executor_fn is None. + """ + + def __init__( + self, + config: SubplanConfig, + executor_fn: SubplanExecutorFn, + merge_service: SubplanMergeService | None = None, + parent_plan_id: str = "", + ) -> None: + if config is None: + raise ValueError("config must not be None") + if executor_fn is None: + raise ValueError("executor_fn must not be None") + + self._config = config + self._executor_fn = executor_fn + self._merge_service = merge_service + self._parent_plan_id = parent_plan_id + self._state = SchedulerState( + max_parallel=config.max_parallel, + execution_mode=config.execution_mode, + ) + + @property + def config(self) -> SubplanConfig: + """The subplan execution configuration.""" + return self._config + + @property + def state(self) -> SchedulerState: + """Current scheduler state.""" + return self._state + + @property + def max_parallel(self) -> int: + """Maximum concurrent subplans allowed.""" + return self._config.max_parallel + + @property + def execution_mode(self) -> ExecutionMode: + """Current execution mode.""" + return self._config.execution_mode + + def schedule( + self, + subplan_statuses: list[SubplanStatus], + base_files: dict[str, str], + dependency_graph: dict[str, list[str]] | None = None, + ) -> SubplanExecutionResult: + """Schedule and execute all subplans with max_parallel concurrency control. + + This method orchestrates the execution of subplans, ensuring that no more + than max_parallel subplans execute simultaneously. Additional subplans are + queued and started as execution slots become available. + + The parent plan blocks until all subplans complete, regardless of execution + mode or concurrency limit. + + Args: + subplan_statuses: Status objects for each subplan to execute. + base_files: File contents before subplans (for merge base). + dependency_graph: For DEPENDENCY_ORDERED mode, maps each subplan_id + to the list of subplan_ids it depends on. + + Returns: + A SubplanExecutionResult with updated statuses and merge outcome. + + Raises: + ValueError: If subplan_statuses is empty. + ValueError: If DEPENDENCY_ORDERED mode but no dependency_graph. + """ + if not subplan_statuses: + raise ValueError("subplan_statuses must not be empty") + + if ( + self._config.execution_mode == ExecutionMode.DEPENDENCY_ORDERED + and dependency_graph is None + ): + raise ValueError("dependency_graph is required for DEPENDENCY_ORDERED mode") + + # Initialize scheduler state + self._state = SchedulerState( + queue=SubplanQueue(pending=subplan_statuses), + max_parallel=self._config.max_parallel, + execution_mode=self._config.execution_mode, + started_at=datetime.now(tz=UTC), + ) + + logger.info( + "scheduler_started max_parallel=%d mode=%s subplan_count=%d", + self._config.max_parallel, + self._config.execution_mode.value, + len(subplan_statuses), + ) + + # Delegate to SubplanExecutionService for actual execution + service = SubplanExecutionService( + config=self._config, + executor_fn=self._executor_fn, + merge_service=self._merge_service, + parent_plan_id=self._parent_plan_id, + ) + + result = service.execute_all( + subplan_statuses=subplan_statuses, + base_files=base_files, + dependency_graph=dependency_graph, + ) + + # Update final state + self._state = SchedulerState( + queue=SubplanQueue( + pending=[], + active=[], + completed=result.statuses, + ), + max_parallel=self._config.max_parallel, + execution_mode=self._config.execution_mode, + started_at=self._state.started_at, + completed_at=datetime.now(tz=UTC), + ) + + logger.info( + "scheduler_completed total_duration_ms=%d all_succeeded=%s failed_count=%d", + result.total_duration_ms, + result.all_succeeded, + len(result.failed_subplan_ids), + ) + + return result + + def get_queue_status(self) -> dict[str, int]: + """Get current queue status. + + Returns: + Dictionary with pending, active, and completed counts. + """ + return { + "pending": len(self._state.queue.pending), + "active": len(self._state.queue.active), + "completed": len(self._state.queue.completed), + "total": self._state.queue.total_count, + } + + def get_available_slots(self) -> int: + """Get number of available execution slots. + + Returns: + Number of subplans that can start immediately. + """ + return self._state.available_slots + + def can_accept_more(self) -> bool: + """Check if more subplans can be queued. + + Returns: + True if there are pending subplans waiting to execute. + """ + return len(self._state.queue.pending) > 0