Files
placeholder/features/steps/plan_executor_subplan_spawning_steps.py
freemo 8608584e99 fix(plan-executor): wire SubplanService and SubplanExecutionService into Execute phase
Implemented optional SubplanService and SubplanExecutionService wiring in PlanExecutor.__init__() (None = no-op).
- Added _spawn_subplans() helper:
  - Queries spawn decisions via SubplanService.get_spawn_decisions() and calls SubplanService.spawn() for each decision.
  - No-ops when there are no spawn decisions.
- Added _execute_subplans() helper:
  - Delegates to SubplanExecutionService.execute_all() to run spawned subplans, handling both sequential and parallel groups as dictated by decisions.
- Added _apply_subplan_results_to_plan() helper:
  - Updates parent plan status tracking when child subplans fail.
  - Annotates error_details with failed_subplan_ids when appropriate.
- Integrated spawning and execution into existing flow:
  - Called _spawn_subplans() and _execute_subplans() from both _run_execute_with_runtime() and _run_execute_with_stub() after actor completion.
- Introduced PlanExecutor properties:
  - subplan_service and subplan_execution_service for external wiring and testability.
- Added tests and scenarios:
  - Behave feature file with 6 scenarios covering subplan_spawn, subplan_parallel_spawn, no-op, and failure tracking.
  - Robot Framework integration test suite with 6 end-to-end subplan spawning test cases.

Key design decisions
- Optional services (None = no-op) to maintain backward compatibility with existing deployments.
- Subplan spawning is a no-op when no spawn decisions exist, avoiding unnecessary work.
- Parent plan error_details is annotated with failed_subplan_ids when a child subplan fails to aid debugging and traceability.
- Both runtime and stub execute paths share the same spawning logic to ensure consistent behavior across execution modes.

ISSUES CLOSED: #3561
2026-04-05 20:46:24 +00:00

559 lines
19 KiB
Python

"""Step definitions for plan_executor_subplan_spawning.feature.
Tests that PlanExecutor correctly wires SubplanService and
SubplanExecutionService into the Execute phase so that subplan_spawn
and subplan_parallel_spawn decisions are realised as actual child plan
executions.
"""
from __future__ import annotations
from unittest.mock import MagicMock
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.plan_executor import (
PlanExecutor,
)
from cleveragents.application.services.subplan_execution_service import (
SubplanExecutionResult,
SubplanExecutionService,
)
from cleveragents.application.services.subplan_service import (
SpawnEntry,
SpawnResult,
SubplanService,
)
from cleveragents.domain.models.core.decision import (
ContextSnapshot,
Decision,
DecisionType,
)
from cleveragents.domain.models.core.plan import (
ExecutionMode,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
ProcessingState,
SubplanConfig,
SubplanStatus,
)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_PLAN_ID = "01KNFMGJSG67S6RG9TVXV205TQ"
_ROOT_ID = "01KNFMGJSH67S6RG9TVXV205TR"
_DEC_ID = "01KNFMGJSH67S6RG9TVXV205TS"
_SUBPLAN_ID = "01KNFMGJSH67S6RG9TVXV205TT"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_plan(
*,
phase: PlanPhase = PlanPhase.EXECUTE,
state: ProcessingState = ProcessingState.QUEUED,
subplan_config: SubplanConfig | None = None,
) -> Plan:
"""Build a minimal Plan domain object for testing."""
return Plan(
identity=PlanIdentity(
plan_id=_PLAN_ID,
root_plan_id=_ROOT_ID,
),
namespaced_name=NamespacedName(namespace="local", name="test-plan"),
description="Test plan for subplan spawning",
action_name="local/test-action",
phase=phase,
processing_state=state,
subplan_config=subplan_config,
)
def _make_spawn_decision(
decision_type: DecisionType = DecisionType.SUBPLAN_SPAWN,
) -> Decision:
"""Build a spawn-type Decision."""
return Decision(
decision_id=_DEC_ID,
plan_id=_PLAN_ID,
decision_type=decision_type,
sequence_number=0,
question="Spawn a child plan?",
chosen_option="local/sub-action",
context_snapshot=ContextSnapshot(relevant_resources=[]),
)
def _make_subplan_status(
status: ProcessingState = ProcessingState.COMPLETE,
error: str | None = None,
) -> SubplanStatus:
"""Build a SubplanStatus for testing."""
return SubplanStatus(
subplan_id=_SUBPLAN_ID,
action_name="local/sub-action",
status=status,
error=error,
)
def _make_spawn_result(
execution_mode: str = ExecutionMode.SEQUENTIAL,
status: ProcessingState = ProcessingState.QUEUED,
) -> SpawnResult:
"""Build a SpawnResult with one subplan status."""
sub_status = SubplanStatus(
subplan_id=_SUBPLAN_ID,
action_name="local/sub-action",
status=status,
)
return SpawnResult(
spawned_statuses=[sub_status],
metadata={},
total_spawned=1,
execution_mode=execution_mode,
child_plans=[],
)
def _make_exec_result(
all_succeeded: bool = True,
failed_ids: list[str] | None = None,
) -> SubplanExecutionResult:
"""Build a SubplanExecutionResult."""
failed = failed_ids or []
status = ProcessingState.COMPLETE if all_succeeded else ProcessingState.ERRORED
sub_status = SubplanStatus(
subplan_id=_SUBPLAN_ID,
action_name="local/sub-action",
status=status,
error=None if all_succeeded else "ChildExecError: timeout",
)
return SubplanExecutionResult(
all_succeeded=all_succeeded,
statuses=[sub_status],
merge_result=None,
total_duration_ms=10,
failed_subplan_ids=failed,
)
def _make_lifecycle(plan: Plan) -> MagicMock:
"""Build a mock lifecycle service that returns the given plan."""
lcs = MagicMock()
lcs.get_plan.return_value = plan
lcs.start_execute = MagicMock()
lcs.complete_execute = MagicMock()
lcs.fail_execute = MagicMock()
lcs._commit_plan = MagicMock()
return lcs
def _make_subplan_service(
decisions: list[Decision] | None = None,
spawn_result: SpawnResult | None = None,
) -> MagicMock:
"""Build a mock SubplanService."""
svc = MagicMock(spec=SubplanService)
svc.get_spawn_decisions.return_value = decisions or []
svc.build_spawn_entries.return_value = [
SpawnEntry(
decision=d,
action_name="local/sub-action",
)
for d in (decisions or [])
]
svc.spawn.return_value = spawn_result or _make_spawn_result()
return svc
def _make_subplan_execution_service(
exec_result: SubplanExecutionResult | None = None,
) -> MagicMock:
"""Build a mock SubplanExecutionService."""
svc = MagicMock(spec=SubplanExecutionService)
svc.execute_all.return_value = exec_result or _make_exec_result()
return svc
def _make_executor(
plan: Plan,
subplan_service: MagicMock | None = None,
subplan_execution_service: MagicMock | None = None,
*,
runtime_mode: bool = False,
) -> tuple[PlanExecutor, MagicMock]:
"""Build a PlanExecutor with mocked dependencies."""
lcs = _make_lifecycle(plan)
if runtime_mode:
# Minimal execution context mock
exec_ctx = MagicMock()
exec_ctx.changeset_store = MagicMock()
exec_ctx.decision_root_id = None
exec_ctx.sandbox_root = None
exec_ctx.sandbox_manager = None
tool_runner = MagicMock()
tool_runner.discover.return_value = []
# Mock RuntimeExecuteActor to avoid real LLM calls
runtime_actor_result = MagicMock()
runtime_actor_result.changeset_id = "01JSPAWN0000000000000CS0001"
runtime_actor_result.sandbox_refs = []
runtime_actor_result.tool_call_count = 0
runtime_actor_result.decision_ids_processed = []
runtime_actor_result.execution_duration_ms = 5
import unittest.mock as _mock
with _mock.patch(
"cleveragents.application.services.plan_executor.RuntimeExecuteActor"
) as _MockActor:
_MockActor.return_value.execute.return_value = runtime_actor_result
executor = PlanExecutor(
lifecycle_service=lcs,
tool_runner=tool_runner,
execution_context=exec_ctx,
subplan_service=subplan_service,
subplan_execution_service=subplan_execution_service,
)
# Store the mock actor class for later assertions
executor._mock_runtime_actor_cls = _MockActor
else:
# Stub mode: patch ExecuteStubActor.execute to avoid real work
execute_actor = MagicMock()
execute_result = MagicMock()
execute_result.changeset_id = "01JSPAWN0000000000000CS0001"
execute_result.sandbox_refs = []
execute_result.tool_calls_count = 0
execute_actor.execute.return_value = execute_result
executor = PlanExecutor(
lifecycle_service=lcs,
execute_actor=execute_actor,
subplan_service=subplan_service,
subplan_execution_service=subplan_execution_service,
)
return executor, lcs
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("a PlanExecutor configured with SubplanService and SubplanExecutionService")
def step_given_executor_with_subplan_services(context: Context) -> None:
"""Set up a PlanExecutor with both subplan services mocked."""
decision = _make_spawn_decision(DecisionType.SUBPLAN_SPAWN)
spawn_result = _make_spawn_result()
exec_result = _make_exec_result(all_succeeded=True)
context.spawn_decision = decision
context.spawn_result = spawn_result
context.exec_result = exec_result
context.mock_subplan_svc = _make_subplan_service(
decisions=[decision],
spawn_result=spawn_result,
)
context.mock_exec_svc = _make_subplan_execution_service(exec_result=exec_result)
plan = _make_plan()
plan.decision_root_id = _ROOT_ID
context.plan = plan
executor, lcs = _make_executor(
plan,
subplan_service=context.mock_subplan_svc,
subplan_execution_service=context.mock_exec_svc,
)
context.executor = executor
context.lcs = lcs
@given(
"a PlanExecutor configured with SubplanService and SubplanExecutionService "
"in runtime mode"
)
def step_given_executor_with_subplan_services_runtime(context: Context) -> None:
"""Set up a PlanExecutor in runtime mode with both subplan services mocked."""
decision = _make_spawn_decision(DecisionType.SUBPLAN_SPAWN)
spawn_result = _make_spawn_result()
exec_result = _make_exec_result(all_succeeded=True)
context.spawn_decision = decision
context.spawn_result = spawn_result
context.exec_result = exec_result
context.mock_subplan_svc = _make_subplan_service(
decisions=[decision],
spawn_result=spawn_result,
)
context.mock_exec_svc = _make_subplan_execution_service(exec_result=exec_result)
plan = _make_plan()
plan.decision_root_id = _ROOT_ID
context.plan = plan
context.runtime_mode = True
@given("a parent plan in Execute phase with a subplan_spawn decision")
def step_given_plan_with_spawn_decision(context: Context) -> None:
"""Ensure the plan has a subplan_spawn decision (already set in Given above)."""
# Plan is already configured in the previous Given step
assert context.plan is not None
assert context.plan.phase == PlanPhase.EXECUTE
@given("a parent plan in Execute phase with a subplan_parallel_spawn decision")
def step_given_plan_with_parallel_spawn_decision(context: Context) -> None:
"""Set up a plan with a subplan_parallel_spawn decision."""
decision = _make_spawn_decision(DecisionType.SUBPLAN_PARALLEL_SPAWN)
spawn_result = _make_spawn_result(execution_mode=ExecutionMode.PARALLEL)
exec_result = _make_exec_result(all_succeeded=True)
context.spawn_decision = decision
context.spawn_result = spawn_result
context.exec_result = exec_result
context.mock_subplan_svc = _make_subplan_service(
decisions=[decision],
spawn_result=spawn_result,
)
context.mock_exec_svc = _make_subplan_execution_service(exec_result=exec_result)
plan = _make_plan()
plan.decision_root_id = _ROOT_ID
context.plan = plan
executor, lcs = _make_executor(
plan,
subplan_service=context.mock_subplan_svc,
subplan_execution_service=context.mock_exec_svc,
)
context.executor = executor
context.lcs = lcs
@given("a parent plan in Execute phase with no spawn decisions")
def step_given_plan_with_no_spawn_decisions(context: Context) -> None:
"""Set up a plan with no spawn decisions."""
context.mock_subplan_svc = _make_subplan_service(decisions=[])
context.mock_exec_svc = _make_subplan_execution_service()
plan = _make_plan()
plan.decision_root_id = _ROOT_ID
context.plan = plan
executor, lcs = _make_executor(
plan,
subplan_service=context.mock_subplan_svc,
subplan_execution_service=context.mock_exec_svc,
)
context.executor = executor
context.lcs = lcs
@given("a PlanExecutor with no SubplanService configured")
def step_given_executor_without_subplan_service(context: Context) -> None:
"""Set up a PlanExecutor without SubplanService."""
plan = _make_plan()
plan.decision_root_id = _ROOT_ID
context.plan = plan
executor, lcs = _make_executor(
plan,
subplan_service=None,
subplan_execution_service=None,
)
context.executor = executor
context.lcs = lcs
context.mock_subplan_svc = None
context.mock_exec_svc = None
@given("the SubplanExecutionService will report a failed subplan")
def step_given_exec_svc_reports_failure(context: Context) -> None:
"""Reconfigure the execution service to report a failure."""
failed_exec_result = _make_exec_result(
all_succeeded=False,
failed_ids=[_SUBPLAN_ID],
)
context.exec_result = failed_exec_result
context.mock_exec_svc.execute_all.return_value = failed_exec_result
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when("I call run_execute on the parent plan")
def step_when_run_execute(context: Context) -> None:
"""Call run_execute on the executor."""
try:
context.execute_result = context.executor.run_execute(
plan_id=_PLAN_ID,
)
context.execute_error = None
except Exception as exc:
context.execute_result = None
context.execute_error = exc
@when("I call run_execute on the parent plan in runtime mode")
def step_when_run_execute_runtime(context: Context) -> None:
"""Call run_execute in runtime mode."""
import unittest.mock as _mock
plan = context.plan
lcs = _make_lifecycle(plan)
exec_ctx = MagicMock()
exec_ctx.changeset_store = MagicMock()
exec_ctx.decision_root_id = None
exec_ctx.sandbox_root = None
exec_ctx.sandbox_manager = None
tool_runner = MagicMock()
tool_runner.discover.return_value = []
runtime_actor_result = MagicMock()
runtime_actor_result.changeset_id = "01JSPAWN0000000000000CS0001"
runtime_actor_result.sandbox_refs = []
runtime_actor_result.tool_call_count = 0
runtime_actor_result.decision_ids_processed = []
runtime_actor_result.execution_duration_ms = 5
with _mock.patch(
"cleveragents.application.services.plan_executor.RuntimeExecuteActor"
) as MockActor:
MockActor.return_value.execute.return_value = runtime_actor_result
executor = PlanExecutor(
lifecycle_service=lcs,
tool_runner=tool_runner,
execution_context=exec_ctx,
subplan_service=context.mock_subplan_svc,
subplan_execution_service=context.mock_exec_svc,
)
try:
context.execute_result = executor.run_execute(plan_id=_PLAN_ID)
context.execute_error = None
except Exception as exc:
context.execute_result = None
context.execute_error = exc
context.executor = executor
context.lcs = lcs
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("SubplanService.get_spawn_decisions should have been called")
def step_then_get_spawn_decisions_called(context: Context) -> None:
"""Verify get_spawn_decisions was called with the plan ID."""
assert context.execute_error is None, f"run_execute raised: {context.execute_error}"
context.mock_subplan_svc.get_spawn_decisions.assert_called_once_with(_PLAN_ID)
@then("SubplanService.spawn should have been called with the spawn entries")
def step_then_spawn_called(context: Context) -> None:
"""Verify spawn was called."""
context.mock_subplan_svc.spawn.assert_called_once()
@then("SubplanExecutionService.execute_all should have been called")
def step_then_execute_all_called(context: Context) -> None:
"""Verify execute_all was called."""
context.mock_exec_svc.execute_all.assert_called_once()
@then("the parent plan subplan_statuses should be updated")
def step_then_subplan_statuses_updated(context: Context) -> None:
"""Verify the plan's subplan_statuses were updated via _commit_plan."""
context.lcs._commit_plan.assert_called()
@then("SubplanService.spawn should NOT have been called")
def step_then_spawn_not_called(context: Context) -> None:
"""Verify spawn was NOT called (no-op path)."""
assert context.execute_error is None, f"run_execute raised: {context.execute_error}"
context.mock_subplan_svc.spawn.assert_not_called()
@then("SubplanExecutionService.execute_all should NOT have been called")
def step_then_execute_all_not_called(context: Context) -> None:
"""Verify execute_all was NOT called (no-op path)."""
context.mock_exec_svc.execute_all.assert_not_called()
@then("the execute phase completes without error")
def step_then_execute_completes_without_error(context: Context) -> None:
"""Verify run_execute completed without raising."""
assert context.execute_error is None, (
f"Expected no error but got: {context.execute_error}"
)
@then("no subplan spawning occurs")
def step_then_no_subplan_spawning(context: Context) -> None:
"""Verify no subplan spawning occurred (no SubplanService configured)."""
# No mock_subplan_svc to assert on — just verify no error
assert context.execute_error is None, (
f"Expected no error but got: {context.execute_error}"
)
@then("the parent plan error_details should contain failed_subplan_ids")
def step_then_error_details_has_failed_ids(context: Context) -> None:
"""Verify error_details contains failed_subplan_ids."""
assert context.execute_error is None, f"run_execute raised: {context.execute_error}"
# The _commit_plan call should have been made with a plan that has
# error_details containing failed_subplan_ids
commit_calls = context.lcs._commit_plan.call_args_list
assert commit_calls, "Expected _commit_plan to be called"
# Find the call where error_details was set with failed_subplan_ids
found = False
for c in commit_calls:
committed_plan = c[0][0]
details = getattr(committed_plan, "error_details", None) or {}
if "failed_subplan_ids" in details:
found = True
break
assert found, (
"Expected error_details to contain 'failed_subplan_ids' in a "
f"_commit_plan call. Calls: {commit_calls}"
)
@then("the parent plan error_details should contain subplan_execution_failed true")
def step_then_error_details_has_exec_failed(context: Context) -> None:
"""Verify error_details contains subplan_execution_failed=true."""
commit_calls = context.lcs._commit_plan.call_args_list
found = False
for c in commit_calls:
committed_plan = c[0][0]
details = getattr(committed_plan, "error_details", None) or {}
if details.get("subplan_execution_failed") == "true":
found = True
break
assert found, (
"Expected error_details to contain 'subplan_execution_failed'='true' "
f"in a _commit_plan call. Calls: {commit_calls}"
)