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

CONTRIBUTORS.md

@@ -12,6 +12,7 @@
 * Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
 * Luis Mendes <luis.p.mendes@gmail.com>
 * Rui Hu <rui.hu@cleverthis.com>
+* HAL 9000 <hal9000@cleverthis.com> 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 <hal9000@cleverthis.com> has contributed fix for #10813 — wiring DecisionService into PlanExecutor for strategy decision persistence during strategize.
 
 * HAL9000 <HAL9000@cleverthis.com> has contributed CLI rendering improvements and TUI overlay visibility handling for `agents project context set` output.

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

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