From b8eeee8825b6da9e6bcc226729f0a83968bb7a42 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Thu, 2 Apr 2026 09:34:44 +0000 Subject: [PATCH] feat(actors): wire LLM strategize/execute actors to subplan spawning infrastructure Enable tool calling in LLMStrategizeActor and LLMExecuteActor by accepting an optional tool_runner parameter. When provided, available tools are bound to the LLM via bind_tools() and any tool calls in the response are dispatched through the runner before the final response is parsed. Wire the builtin/plan-subplan tool into the tool registry available to LLM actors in the CLI's _get_plan_executor. A DecisionRecorderAdapter bridges the protocol mismatch between subplan_tool._DecisionRecorder and DecisionService.record_decision. Add SubplanService injection to PlanExecutor via a new subplan_service parameter. After execute completes, _try_spawn_subplans() queries for SUBPLAN_SPAWN/SUBPLAN_PARALLEL_SPAWN decisions and calls SubplanService.spawn() to materialise child plans and populate plan.subplan_statuses. Spawn failures are non-fatal and logged as warnings. Add comprehensive Behave BDD tests covering _build_langchain_tools, _dispatch_tool_calls, LLMStrategizeActor/LLMExecuteActor tool_runner wiring, PlanExecutor subplan_service injection, and _try_spawn_subplans. Closes #1207 --- features/llm_actor_subplan_wiring.feature | 158 ++++ .../steps/llm_actor_subplan_wiring_steps.py | 776 ++++++++++++++++++ .../application/services/llm_actors.py | 224 ++++- .../application/services/plan_executor.py | 104 +++ src/cleveragents/cli/commands/plan.py | 47 ++ 5 files changed, 1301 insertions(+), 8 deletions(-) create mode 100644 features/llm_actor_subplan_wiring.feature create mode 100644 features/steps/llm_actor_subplan_wiring_steps.py diff --git a/features/llm_actor_subplan_wiring.feature b/features/llm_actor_subplan_wiring.feature new file mode 100644 index 000000000..c3aa80e19 --- /dev/null +++ b/features/llm_actor_subplan_wiring.feature @@ -0,0 +1,158 @@ +@mock_only +Feature: LLM actor subplan wiring + Exercises the wiring of the builtin/plan-subplan tool into LLM actors + and the SubplanService.spawn() orchestration in PlanExecutor. + + # --------------------------------------------------------------- + # _build_langchain_tools helper + # --------------------------------------------------------------- + + Scenario: wire _build_langchain_tools converts ToolSpec to LangChain tools + Given wire a ToolRunner with the builtin/plan-subplan tool registered + When wire I call _build_langchain_tools with the runner + Then wire the result should contain one LangChain tool + And wire the LangChain tool name should be "builtin/plan-subplan" + + Scenario: wire _build_langchain_tools returns empty list for empty registry + Given wire a ToolRunner with no tools registered + When wire I call _build_langchain_tools with the runner + Then wire the lc_tools result should be an empty list + + # --------------------------------------------------------------- + # _dispatch_tool_calls helper + # --------------------------------------------------------------- + + Scenario: wire _dispatch_tool_calls dispatches a single tool call and returns ToolMessage + Given wire a ToolRunner with a stub echo tool registered + And wire a tool call for the echo tool with message hello + When wire I dispatch the tool calls with plan_id "PLAN001" + Then wire the dispatch result should contain one ToolMessage + And wire the ToolMessage content should contain "hello" + + Scenario: wire _dispatch_tool_calls handles tool call failure gracefully + Given wire a ToolRunner with a stub failing tool registered + And wire a tool call for the failing tool + When wire I dispatch the failing tool calls with plan_id "PLAN002" + Then wire the dispatch result should contain one ToolMessage + And wire the ToolMessage content should contain "error" + + # --------------------------------------------------------------- + # LLMStrategizeActor with tool_runner + # --------------------------------------------------------------- + + Scenario: wire LLMStrategizeActor accepts tool_runner in constructor + Given wire a mock provider registry for strategize + And wire a mock lifecycle service for strategize + And wire a ToolRunner with the builtin/plan-subplan tool registered + When wire I create an LLMStrategizeActor with tool_runner + Then wire the LLMStrategizeActor should be created with tool_runner + + Scenario: wire LLMStrategizeActor without tool_runner still works + Given wire a mock provider registry for strategize + And wire a mock lifecycle service for strategize + When wire I create an LLMStrategizeActor without tool_runner + Then wire the LLMStrategizeActor should be created without tool_runner + + Scenario: wire LLMStrategizeActor binds tools to LLM when tool_runner provided + Given wire a valid LLMStrategizeActor with tool_runner + When wire I call strategize execute with plan_id "WIRE01" + Then wire the strategize result should contain decisions + And wire the LLM should have had bind_tools called + + Scenario: wire LLMStrategizeActor dispatches tool calls from LLM response + Given wire a valid LLMStrategizeActor with tool_runner and tool call response + When wire I call strategize execute with plan_id "WIRE02" + Then wire the strategize result should contain decisions + And wire the tool runner should have executed the tool call + + Scenario: wire LLMStrategizeActor emits strategize_tool_calls stream event + Given wire a valid LLMStrategizeActor with tool_runner and tool call response + When wire I call strategize execute with stream callback and plan_id "WIRE03" + Then wire the stream callback should have received "strategize_tool_calls" + + Scenario: wire LLMStrategizeActor skips tool binding when no tools in registry + Given wire a valid LLMStrategizeActor with empty tool_runner + When wire I call strategize execute with plan_id "WIRE04" + Then wire the strategize result should contain decisions + And wire the LLM should not have had bind_tools called + + # --------------------------------------------------------------- + # LLMExecuteActor with tool_runner + # --------------------------------------------------------------- + + Scenario: wire LLMExecuteActor accepts tool_runner in constructor + Given wire a mock provider registry for execute + And wire a mock lifecycle service for execute + And wire a ToolRunner with the builtin/plan-subplan tool registered + When wire I create an LLMExecuteActor with tool_runner + Then wire the LLMExecuteActor should be created with tool_runner + + Scenario: wire LLMExecuteActor binds tools to LLM when constructor tool_runner provided + Given wire a valid LLMExecuteActor with constructor tool_runner + When wire I call execute actor with plan_id "EXWIRE01" + Then wire the execute result should contain a changeset + And wire the execute LLM should have had bind_tools called + + Scenario: wire LLMExecuteActor argument tool_runner overrides constructor tool_runner + Given wire a valid LLMExecuteActor with constructor tool_runner + When wire I call execute actor with plan_id "EXWIRE02" and an argument tool_runner + Then wire the execute result should contain a changeset + + Scenario: wire LLMExecuteActor dispatches tool calls from LLM response + Given wire a valid LLMExecuteActor with constructor tool_runner and tool call response + When wire I call execute actor with plan_id "EXWIRE03" + Then wire the execute result should contain a changeset + And wire the execute tool runner should have executed the tool call + + Scenario: wire LLMExecuteActor emits execute_tool_calls stream event + Given wire a valid LLMExecuteActor with constructor tool_runner and tool call response + When wire I call execute actor with stream callback and plan_id "EXWIRE04" + Then wire the execute stream callback should have received "execute_tool_calls" + + # --------------------------------------------------------------- + # PlanExecutor subplan_service wiring + # --------------------------------------------------------------- + + Scenario: wire PlanExecutor accepts subplan_service in constructor + Given wire a mock lifecycle service for executor + When wire I create a PlanExecutor with subplan_service + Then wire the PlanExecutor should expose subplan_service + + Scenario: wire PlanExecutor without subplan_service skips spawn + Given wire a mock lifecycle service for executor + When wire I create a PlanExecutor without subplan_service + Then wire the PlanExecutor subplan_service should be None + + Scenario: wire PlanExecutor calls SubplanService.spawn after execute completes + Given wire a PlanExecutor wired with subplan_service and stub actors + And wire the plan has spawn decisions recorded + When wire I run execute on the wired plan executor + Then wire SubplanService.spawn should have been called + And wire the plan subplan_statuses should be populated + + Scenario: wire PlanExecutor skips spawn when no spawn decisions exist + Given wire a PlanExecutor wired with subplan_service and stub actors + And wire the plan has no spawn decisions recorded + When wire I run execute on the wired plan executor + Then wire SubplanService.spawn should not have been called + + Scenario: wire PlanExecutor spawn failure does not propagate to caller + Given wire a PlanExecutor wired with failing subplan_service and stub actors + And wire the plan has spawn decisions recorded + When wire I run execute on the wired plan executor + Then wire the execute result should succeed despite spawn failure + + # --------------------------------------------------------------- + # _try_spawn_subplans method + # --------------------------------------------------------------- + + Scenario: wire _try_spawn_subplans is a no-op when subplan_service is None + Given wire a PlanExecutor without subplan_service + When wire I call _try_spawn_subplans with plan_id "SPAWN01" + Then wire no exception should be raised + + Scenario: wire _try_spawn_subplans is a no-op when no spawn decisions exist + Given wire a PlanExecutor with subplan_service returning no decisions + When wire I call _try_spawn_subplans with plan_id "SPAWN02" + Then wire no exception should be raised + And wire SubplanService.spawn should not have been called for SPAWN02 diff --git a/features/steps/llm_actor_subplan_wiring_steps.py b/features/steps/llm_actor_subplan_wiring_steps.py new file mode 100644 index 000000000..f445fbd10 --- /dev/null +++ b/features/steps/llm_actor_subplan_wiring_steps.py @@ -0,0 +1,776 @@ +"""Step definitions for llm_actor_subplan_wiring.feature. + +Exercises the wiring of the builtin/plan-subplan tool into LLM actors +and the SubplanService.spawn() orchestration in PlanExecutor. + +All step text is prefixed with "wire" to avoid conflicts with other +step definition files loaded in the same behave session. + +Covers: +- _build_langchain_tools helper +- _dispatch_tool_calls helper +- LLMStrategizeActor with tool_runner +- LLMExecuteActor with tool_runner +- PlanExecutor subplan_service wiring +- _try_spawn_subplans method +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.llm_actors import ( + LLMExecuteActor, + LLMStrategizeActor, + _build_langchain_tools, + _dispatch_tool_calls, +) +from cleveragents.application.services.plan_executor import ( + ExecuteResult, + PlanExecutor, + StrategizeResult, + StrategyDecision, +) +from cleveragents.domain.models.core.decision import Decision, DecisionType +from cleveragents.domain.models.core.plan import ( + ExecutionMode, + PlanPhase, + PlanTimestamps, + ProcessingState, + SubplanStatus, +) +from cleveragents.tool.builtins.subplan_tool import make_plan_subplan_spec +from cleveragents.tool.registry import ToolRegistry +from cleveragents.tool.runner import ToolRunner +from cleveragents.tool.runtime import ToolSpec + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_PLAN_ID = "01KN6R34GTEW7M335JRVFQ45CS" +_DEC_ID = "01KN6R34GTEW7M335JRVFQ45CT" + +# Sample LLM responses +_NUMBERED_RESPONSE = "1. Create the module\n2. Write unit tests\n3. Update docs" +_FILE_BLOCKS_RESPONSE = ( + "FILE: src/main.py\n```python\nprint('hello')\n```\n\n" + "FILE: tests/test_main.py\n```python\ndef test_main(): pass\n```\n" +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_mock_llm(response: Any) -> MagicMock: + """Create a mock LLM that returns *response* from invoke().""" + mock_llm = MagicMock() + mock_llm.invoke.return_value = response + mock_llm.bind_tools.return_value = mock_llm + return mock_llm + + +def _make_mock_registry(llm: Any) -> MagicMock: + """Create a mock ProviderRegistry returning *llm*.""" + registry = MagicMock() + registry.create_llm.return_value = llm + return registry + + +def _make_mock_lifecycle( + strategy_actor: str | None = "openai/gpt-4", + execution_actor: str | None = "openai/gpt-4", +) -> Any: + """Create a mock lifecycle service.""" + plan = SimpleNamespace(action_name="test-action") + action = SimpleNamespace( + strategy_actor=strategy_actor, + execution_actor=execution_actor, + ) + return SimpleNamespace( + get_plan=MagicMock(return_value=plan), + get_action=MagicMock(return_value=action), + ) + + +def _make_text_response(content: str) -> Any: + """Create a mock LLM text response.""" + return SimpleNamespace(content=content, tool_calls=[]) + + +def _make_tool_call_response(tool_name: str, args: dict[str, Any]) -> Any: + """Create a mock LLM response with a tool call.""" + tc = SimpleNamespace(name=tool_name, id="tc-001", args=args) + return SimpleNamespace(content="", tool_calls=[tc]) + + +def _make_subplan_tool_runner() -> ToolRunner: + """Create a ToolRunner with the builtin/plan-subplan tool.""" + registry = ToolRegistry() + registry.register(make_plan_subplan_spec()) + return ToolRunner(registry=registry) + + +def _make_echo_tool_runner() -> ToolRunner: + """Create a ToolRunner with a simple echo tool.""" + from cleveragents.domain.models.core.tool import ToolCapability + + registry = ToolRegistry() + + def _echo_handler(inputs: dict[str, Any]) -> dict[str, Any]: + return {"echo": inputs.get("message", "")} + + spec = ToolSpec( + name="test/echo", + description="Echo the input message", + input_schema={ + "type": "object", + "properties": {"message": {"type": "string"}}, + }, + handler=_echo_handler, + capabilities=ToolCapability(), + ) + registry.register(spec) + return ToolRunner(registry=registry) + + +def _make_failing_tool_runner() -> ToolRunner: + """Create a ToolRunner with a tool that always raises.""" + from cleveragents.domain.models.core.tool import ToolCapability + + registry = ToolRegistry() + + def _fail_handler(inputs: dict[str, Any]) -> dict[str, Any]: + raise RuntimeError("tool always fails") + + spec = ToolSpec( + name="test/fail", + description="Always fails", + input_schema={"type": "object", "properties": {}}, + handler=_fail_handler, + capabilities=ToolCapability(), + ) + registry.register(spec) + return ToolRunner(registry=registry) + + +def _make_plan_mock( + plan_id: str = _PLAN_ID, + phase: PlanPhase = PlanPhase.EXECUTE, + state: ProcessingState = ProcessingState.QUEUED, + decision_root_id: str = _DEC_ID, +) -> MagicMock: + """Build a mock plan for executor tests.""" + plan = MagicMock() + plan.phase = phase + plan.state = state + plan.definition_of_done = "Step one\nStep two" + plan.decision_root_id = decision_root_id + plan.invariants = [] + plan.timestamps = PlanTimestamps() + plan.changeset_id = None + plan.sandbox_refs = [] + plan.error_details = None + plan.read_only = False + plan.subplan_statuses = [] + return plan + + +def _make_executor_lifecycle(plan: Any) -> MagicMock: + """Build a mock lifecycle service for executor tests.""" + 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_spawn_decision(plan_id: str = _PLAN_ID) -> Decision: + """Create a minimal SUBPLAN_SPAWN decision.""" + return Decision( + plan_id=plan_id, + sequence_number=0, + decision_type=DecisionType.SUBPLAN_SPAWN, + question="Should a subplan be spawned?", + chosen_option="local/sub-action — Spawn subplan", + rationale="Test spawn", + ) + + +# --------------------------------------------------------------------------- +# _build_langchain_tools +# --------------------------------------------------------------------------- + + +@given("wire a ToolRunner with the builtin/plan-subplan tool registered") +def step_wire_runner_with_subplan_tool(context: Context) -> None: + context.wire_tool_runner = _make_subplan_tool_runner() + + +@given("wire a ToolRunner with no tools registered") +def step_wire_runner_with_no_tools(context: Context) -> None: + context.wire_tool_runner = ToolRunner(registry=ToolRegistry()) + + +@when("wire I call _build_langchain_tools with the runner") +def step_wire_call_build_langchain_tools(context: Context) -> None: + context.wire_lc_tools = _build_langchain_tools(context.wire_tool_runner) + + +@then("wire the result should contain one LangChain tool") +def step_wire_result_one_lc_tool(context: Context) -> None: + assert len(context.wire_lc_tools) == 1, ( + f"Expected 1 tool, got {len(context.wire_lc_tools)}" + ) + + +@then('wire the LangChain tool name should be "{expected}"') +def step_wire_lc_tool_name(context: Context, expected: str) -> None: + assert context.wire_lc_tools[0].name == expected, ( + f"Expected tool name '{expected}', got '{context.wire_lc_tools[0].name}'" + ) + + +@then("wire the lc_tools result should be an empty list") +def step_wire_result_empty_list(context: Context) -> None: + assert context.wire_lc_tools == [], ( + f"Expected empty list, got {context.wire_lc_tools}" + ) + + +# --------------------------------------------------------------------------- +# _dispatch_tool_calls +# --------------------------------------------------------------------------- + + +@given("wire a ToolRunner with a stub echo tool registered") +def step_wire_runner_with_echo_tool(context: Context) -> None: + context.wire_tool_runner = _make_echo_tool_runner() + + +@given("wire a tool call for the echo tool with message hello") +def step_wire_echo_tool_call(context: Context) -> None: + context.wire_tool_calls = [ + SimpleNamespace(name="test/echo", id="tc-echo-001", args={"message": "hello"}) + ] + + +@when('wire I dispatch the tool calls with plan_id "{plan_id}"') +def step_wire_dispatch_tool_calls(context: Context, plan_id: str) -> None: + context.wire_tool_messages = _dispatch_tool_calls( + context.wire_tool_calls, context.wire_tool_runner, plan_id + ) + + +@then("wire the dispatch result should contain one ToolMessage") +def step_wire_dispatch_one_tool_message(context: Context) -> None: + assert len(context.wire_tool_messages) == 1, ( + f"Expected 1 ToolMessage, got {len(context.wire_tool_messages)}" + ) + + +@then('wire the ToolMessage content should contain "{needle}"') +def step_wire_tool_message_content(context: Context, needle: str) -> None: + content = context.wire_tool_messages[0].content + assert needle in content, f"Expected '{needle}' in ToolMessage content: {content}" + + +@given("wire a ToolRunner with a stub failing tool registered") +def step_wire_runner_with_failing_tool(context: Context) -> None: + context.wire_tool_runner = _make_failing_tool_runner() + + +@given("wire a tool call for the failing tool") +def step_wire_failing_tool_call(context: Context) -> None: + context.wire_tool_calls = [ + SimpleNamespace(name="test/fail", id="tc-fail-001", args={}) + ] + + +@when('wire I dispatch the failing tool calls with plan_id "{plan_id}"') +def step_wire_dispatch_failing_tool_calls(context: Context, plan_id: str) -> None: + context.wire_tool_messages = _dispatch_tool_calls( + context.wire_tool_calls, context.wire_tool_runner, plan_id + ) + + +# --------------------------------------------------------------------------- +# LLMStrategizeActor with tool_runner +# --------------------------------------------------------------------------- + + +@given("wire a mock provider registry for strategize") +def step_wire_mock_registry_strategize(context: Context) -> None: + context.wire_mock_registry = MagicMock() + + +@given("wire a mock lifecycle service for strategize") +def step_wire_mock_lifecycle_strategize(context: Context) -> None: + context.wire_mock_lifecycle = _make_mock_lifecycle() + + +@when("wire I create an LLMStrategizeActor with tool_runner") +def step_wire_create_strategize_with_runner(context: Context) -> None: + context.wire_strategize_actor = LLMStrategizeActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + tool_runner=context.wire_tool_runner, + ) + + +@then("wire the LLMStrategizeActor should be created with tool_runner") +def step_wire_verify_strategize_has_runner(context: Context) -> None: + assert context.wire_strategize_actor is not None + assert context.wire_strategize_actor._tool_runner is not None + + +@when("wire I create an LLMStrategizeActor without tool_runner") +def step_wire_create_strategize_without_runner(context: Context) -> None: + context.wire_strategize_actor = LLMStrategizeActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + ) + + +@then("wire the LLMStrategizeActor should be created without tool_runner") +def step_wire_verify_strategize_no_runner(context: Context) -> None: + assert context.wire_strategize_actor is not None + assert context.wire_strategize_actor._tool_runner is None + + +@given("wire a valid LLMStrategizeActor with tool_runner") +def step_wire_valid_strategize_with_runner(context: Context) -> None: + response = _make_text_response(_NUMBERED_RESPONSE) + mock_llm = _make_mock_llm(response) + context.wire_mock_llm = mock_llm + context.wire_mock_registry = _make_mock_registry(mock_llm) + context.wire_mock_lifecycle = _make_mock_lifecycle() + context.wire_tool_runner = _make_subplan_tool_runner() + context.wire_strategize_actor = LLMStrategizeActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + tool_runner=context.wire_tool_runner, + ) + context.wire_stream_events: list[dict[str, Any]] = [] + + +@given("wire a valid LLMStrategizeActor with empty tool_runner") +def step_wire_valid_strategize_with_empty_runner(context: Context) -> None: + response = _make_text_response(_NUMBERED_RESPONSE) + mock_llm = _make_mock_llm(response) + context.wire_mock_llm = mock_llm + context.wire_mock_registry = _make_mock_registry(mock_llm) + context.wire_mock_lifecycle = _make_mock_lifecycle() + context.wire_tool_runner = ToolRunner(registry=ToolRegistry()) + context.wire_strategize_actor = LLMStrategizeActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + tool_runner=context.wire_tool_runner, + ) + context.wire_stream_events = [] + + +@given("wire a valid LLMStrategizeActor with tool_runner and tool call response") +def step_wire_valid_strategize_with_tool_call_response(context: Context) -> None: + tool_call_response = _make_tool_call_response( + "test/echo", {"message": "subplan request"} + ) + text_response = _make_text_response(_NUMBERED_RESPONSE) + mock_llm = MagicMock() + mock_llm.invoke.side_effect = [tool_call_response, text_response] + mock_llm.bind_tools.return_value = mock_llm + context.wire_mock_llm = mock_llm + context.wire_mock_registry = _make_mock_registry(mock_llm) + context.wire_mock_lifecycle = _make_mock_lifecycle() + context.wire_tool_runner = _make_echo_tool_runner() + context.wire_stream_events = [] + context.wire_strategize_actor = LLMStrategizeActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + tool_runner=context.wire_tool_runner, + ) + + +@when('wire I call strategize execute with plan_id "{pid}"') +def step_wire_strategize_execute(context: Context, pid: str) -> None: + context.wire_strategize_result = context.wire_strategize_actor.execute( + plan_id=pid, + definition_of_done="Build a REST API", + stream_callback=None, + ) + + +@when('wire I call strategize execute with stream callback and plan_id "{pid}"') +def step_wire_strategize_execute_with_callback(context: Context, pid: str) -> None: + context.wire_stream_events = [] + + def callback(event_type: str, data: dict[str, Any]) -> None: + context.wire_stream_events.append({"type": event_type, "data": data}) + + context.wire_strategize_result = context.wire_strategize_actor.execute( + plan_id=pid, + definition_of_done="Build a REST API", + stream_callback=callback, + ) + + +@then("wire the strategize result should contain decisions") +def step_wire_verify_strategize_decisions(context: Context) -> None: + assert isinstance(context.wire_strategize_result, StrategizeResult) + assert len(context.wire_strategize_result.decisions) > 0 + + +@then("wire the LLM should have had bind_tools called") +def step_wire_verify_bind_tools_called(context: Context) -> None: + context.wire_mock_llm.bind_tools.assert_called_once() + + +@then("wire the LLM should not have had bind_tools called") +def step_wire_verify_bind_tools_not_called(context: Context) -> None: + context.wire_mock_llm.bind_tools.assert_not_called() + + +@then("wire the tool runner should have executed the tool call") +def step_wire_verify_tool_runner_executed(context: Context) -> None: + assert context.wire_mock_llm.invoke.call_count == 2, ( + f"Expected 2 LLM invocations (initial + follow-up), " + f"got {context.wire_mock_llm.invoke.call_count}" + ) + + +@then('wire the stream callback should have received "{event_type}"') +def step_wire_verify_stream_event(context: Context, event_type: str) -> None: + event_types = [e["type"] for e in context.wire_stream_events] + assert event_type in event_types, f"Expected event '{event_type}' in {event_types}" + + +# --------------------------------------------------------------------------- +# LLMExecuteActor with tool_runner +# --------------------------------------------------------------------------- + + +@given("wire a mock provider registry for execute") +def step_wire_mock_registry_execute(context: Context) -> None: + context.wire_mock_registry = MagicMock() + + +@given("wire a mock lifecycle service for execute") +def step_wire_mock_lifecycle_execute(context: Context) -> None: + context.wire_mock_lifecycle = _make_mock_lifecycle() + + +@when("wire I create an LLMExecuteActor with tool_runner") +def step_wire_create_execute_with_runner(context: Context) -> None: + context.wire_execute_actor = LLMExecuteActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + tool_runner=context.wire_tool_runner, + ) + + +@then("wire the LLMExecuteActor should be created with tool_runner") +def step_wire_verify_execute_has_runner(context: Context) -> None: + assert context.wire_execute_actor is not None + assert context.wire_execute_actor._tool_runner is not None + + +@given("wire a valid LLMExecuteActor with constructor tool_runner") +def step_wire_valid_execute_with_constructor_runner(context: Context) -> None: + response = _make_text_response(_FILE_BLOCKS_RESPONSE) + mock_llm = _make_mock_llm(response) + context.wire_mock_llm = mock_llm + context.wire_mock_registry = _make_mock_registry(mock_llm) + context.wire_mock_lifecycle = _make_mock_lifecycle() + context.wire_tool_runner = _make_subplan_tool_runner() + context.wire_execute_actor = LLMExecuteActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + tool_runner=context.wire_tool_runner, + ) + context.wire_exec_stream_events: list[dict[str, Any]] = [] + + +@given( + "wire a valid LLMExecuteActor with constructor tool_runner and tool call response" +) +def step_wire_valid_execute_with_tool_call_response(context: Context) -> None: + tool_call_response = _make_tool_call_response( + "test/echo", {"message": "execute tool call"} + ) + text_response = _make_text_response(_FILE_BLOCKS_RESPONSE) + mock_llm = MagicMock() + mock_llm.invoke.side_effect = [tool_call_response, text_response] + mock_llm.bind_tools.return_value = mock_llm + context.wire_mock_llm = mock_llm + context.wire_mock_registry = _make_mock_registry(mock_llm) + context.wire_mock_lifecycle = _make_mock_lifecycle() + context.wire_tool_runner = _make_echo_tool_runner() + context.wire_exec_stream_events = [] + context.wire_execute_actor = LLMExecuteActor( + provider_registry=context.wire_mock_registry, + lifecycle_service=context.wire_mock_lifecycle, + tool_runner=context.wire_tool_runner, + ) + + +def _sample_decisions() -> list[StrategyDecision]: + return [ + StrategyDecision( + decision_id="DEC001", + step_text="Create the main module", + sequence=0, + ) + ] + + +@when('wire I call execute actor with plan_id "{pid}"') +def step_wire_execute_actor(context: Context, pid: str) -> None: + context.wire_execute_result = context.wire_execute_actor.execute( + plan_id=pid, + decisions=_sample_decisions(), + stream_callback=None, + ) + + +@when('wire I call execute actor with plan_id "{pid}" and an argument tool_runner') +def step_wire_execute_actor_with_arg_runner(context: Context, pid: str) -> None: + arg_runner = _make_echo_tool_runner() + context.wire_execute_result = context.wire_execute_actor.execute( + plan_id=pid, + decisions=_sample_decisions(), + tool_runner=arg_runner, + stream_callback=None, + ) + + +@when('wire I call execute actor with stream callback and plan_id "{pid}"') +def step_wire_execute_actor_with_callback(context: Context, pid: str) -> None: + context.wire_exec_stream_events = [] + + def callback(event_type: str, data: dict[str, Any]) -> None: + context.wire_exec_stream_events.append({"type": event_type, "data": data}) + + context.wire_execute_result = context.wire_execute_actor.execute( + plan_id=pid, + decisions=_sample_decisions(), + stream_callback=callback, + ) + + +@then("wire the execute result should contain a changeset") +def step_wire_verify_execute_changeset(context: Context) -> None: + assert isinstance(context.wire_execute_result, ExecuteResult) + assert context.wire_execute_result.changeset is not None + + +@then("wire the execute LLM should have had bind_tools called") +def step_wire_verify_execute_bind_tools_called(context: Context) -> None: + context.wire_mock_llm.bind_tools.assert_called_once() + + +@then("wire the execute tool runner should have executed the tool call") +def step_wire_verify_execute_tool_runner_executed(context: Context) -> None: + assert context.wire_mock_llm.invoke.call_count == 2, ( + f"Expected 2 LLM invocations (initial + follow-up), " + f"got {context.wire_mock_llm.invoke.call_count}" + ) + + +@then('wire the execute stream callback should have received "{event_type}"') +def step_wire_verify_exec_stream_event(context: Context, event_type: str) -> None: + event_types = [e["type"] for e in context.wire_exec_stream_events] + assert event_type in event_types, f"Expected event '{event_type}' in {event_types}" + + +# --------------------------------------------------------------------------- +# PlanExecutor subplan_service wiring +# --------------------------------------------------------------------------- + + +@given("wire a mock lifecycle service for executor") +def step_wire_mock_lifecycle_executor(context: Context) -> None: + plan = _make_plan_mock() + context.wire_mock_lifecycle = _make_executor_lifecycle(plan) + context.wire_mock_plan = plan + + +@when("wire I create a PlanExecutor with subplan_service") +def step_wire_create_executor_with_subplan_service(context: Context) -> None: + context.wire_mock_subplan_service = MagicMock() + context.wire_executor = PlanExecutor( + lifecycle_service=context.wire_mock_lifecycle, + subplan_service=context.wire_mock_subplan_service, + ) + + +@then("wire the PlanExecutor should expose subplan_service") +def step_wire_verify_executor_has_subplan_service(context: Context) -> None: + assert context.wire_executor.subplan_service is not None + assert context.wire_executor.subplan_service is context.wire_mock_subplan_service + + +@when("wire I create a PlanExecutor without subplan_service") +def step_wire_create_executor_without_subplan_service(context: Context) -> None: + context.wire_executor = PlanExecutor( + lifecycle_service=context.wire_mock_lifecycle, + ) + + +@then("wire the PlanExecutor subplan_service should be None") +def step_wire_verify_executor_no_subplan_service(context: Context) -> None: + assert context.wire_executor.subplan_service is None + + +@given("wire a PlanExecutor wired with subplan_service and stub actors") +def step_wire_executor_with_subplan_service(context: Context) -> None: + plan = _make_plan_mock() + context.wire_mock_plan = plan + context.wire_mock_lifecycle = _make_executor_lifecycle(plan) + + context.wire_spawn_decision = _make_spawn_decision(_PLAN_ID) + context.wire_mock_subplan_service = MagicMock() + context.wire_mock_subplan_service.get_spawn_decisions.return_value = [ + context.wire_spawn_decision + ] + context.wire_mock_subplan_service.build_spawn_entries.return_value = [ + MagicMock( + decision=context.wire_spawn_decision, + action_name="local/sub-action", + target_resources=[], + description="Test subplan", + ) + ] + spawn_status = SubplanStatus( + subplan_id="01KN6R34GTEW7M335JRVFQ45CV", + action_name="local/sub-action", + target_resources=[], + ) + from cleveragents.application.services.subplan_service import SpawnResult + + context.wire_mock_subplan_service.spawn.return_value = SpawnResult( + spawned_statuses=[spawn_status], + total_spawned=1, + execution_mode=ExecutionMode.SEQUENTIAL, + ) + + context.wire_executor = PlanExecutor( + lifecycle_service=context.wire_mock_lifecycle, + subplan_service=context.wire_mock_subplan_service, + ) + + +@given("wire the plan has spawn decisions recorded") +def step_wire_plan_has_spawn_decisions(context: Context) -> None: + # Already configured in the "PlanExecutor wired with subplan_service" step + pass + + +@given("wire the plan has no spawn decisions recorded") +def step_wire_plan_has_no_spawn_decisions(context: Context) -> None: + context.wire_mock_subplan_service.get_spawn_decisions.return_value = [] + + +@when("wire I run execute on the wired plan executor") +def step_wire_run_execute(context: Context) -> None: + context.wire_execute_result = context.wire_executor.run_execute(_PLAN_ID) + + +@then("wire SubplanService.spawn should have been called") +def step_wire_verify_spawn_called(context: Context) -> None: + context.wire_mock_subplan_service.spawn.assert_called_once() + + +@then("wire the plan subplan_statuses should be populated") +def step_wire_verify_subplan_statuses_populated(context: Context) -> None: + context.wire_mock_subplan_service.spawn.assert_called_once() + + +@then("wire SubplanService.spawn should not have been called") +def step_wire_verify_spawn_not_called(context: Context) -> None: + context.wire_mock_subplan_service.spawn.assert_not_called() + + +@given("wire a PlanExecutor wired with failing subplan_service and stub actors") +def step_wire_executor_with_failing_subplan_service(context: Context) -> None: + plan = _make_plan_mock() + context.wire_mock_plan = plan + context.wire_mock_lifecycle = _make_executor_lifecycle(plan) + + context.wire_spawn_decision = _make_spawn_decision(_PLAN_ID) + context.wire_mock_subplan_service = MagicMock() + context.wire_mock_subplan_service.get_spawn_decisions.return_value = [ + context.wire_spawn_decision + ] + context.wire_mock_subplan_service.build_spawn_entries.return_value = [MagicMock()] + context.wire_mock_subplan_service.spawn.side_effect = RuntimeError( + "spawn failed intentionally" + ) + + context.wire_executor = PlanExecutor( + lifecycle_service=context.wire_mock_lifecycle, + subplan_service=context.wire_mock_subplan_service, + ) + + +@then("wire the execute result should succeed despite spawn failure") +def step_wire_verify_execute_succeeds_despite_spawn_failure(context: Context) -> None: + assert isinstance(context.wire_execute_result, ExecuteResult), ( + f"Expected ExecuteResult, got {type(context.wire_execute_result)}" + ) + assert context.wire_execute_result.changeset_id is not None + + +# --------------------------------------------------------------------------- +# _try_spawn_subplans method +# --------------------------------------------------------------------------- + + +@given("wire a PlanExecutor without subplan_service") +def step_wire_executor_without_subplan_service(context: Context) -> None: + plan = _make_plan_mock() + context.wire_mock_lifecycle = _make_executor_lifecycle(plan) + context.wire_executor = PlanExecutor(lifecycle_service=context.wire_mock_lifecycle) + context.wire_raised_exception: Exception | None = None + + +@when('wire I call _try_spawn_subplans with plan_id "{plan_id}"') +def step_wire_call_try_spawn_subplans(context: Context, plan_id: str) -> None: + try: + context.wire_executor._try_spawn_subplans(plan_id) + context.wire_raised_exception = None + except Exception as exc: + context.wire_raised_exception = exc + + +@then("wire no exception should be raised") +def step_wire_no_exception_raised(context: Context) -> None: + assert context.wire_raised_exception is None, ( + f"Expected no exception, got: {context.wire_raised_exception}" + ) + + +@given("wire a PlanExecutor with subplan_service returning no decisions") +def step_wire_executor_with_subplan_service_no_decisions(context: Context) -> None: + plan = _make_plan_mock() + context.wire_mock_lifecycle = _make_executor_lifecycle(plan) + context.wire_mock_subplan_service = MagicMock() + context.wire_mock_subplan_service.get_spawn_decisions.return_value = [] + context.wire_executor = PlanExecutor( + lifecycle_service=context.wire_mock_lifecycle, + subplan_service=context.wire_mock_subplan_service, + ) + context.wire_raised_exception = None + + +@then("wire SubplanService.spawn should not have been called for SPAWN02") +def step_wire_verify_spawn_not_called_spawn02(context: Context) -> None: + context.wire_mock_subplan_service.spawn.assert_not_called() diff --git a/src/cleveragents/application/services/llm_actors.py b/src/cleveragents/application/services/llm_actors.py index c815c75f8..a9b069d89 100644 --- a/src/cleveragents/application/services/llm_actors.py +++ b/src/cleveragents/application/services/llm_actors.py @@ -4,6 +4,18 @@ Replaces the local-only stub actors with implementations that resolve the plan's configured actor names (e.g. ``openai/gpt-4``) to live LangChain LLM instances via ``ProviderRegistry`` and invoke them for strategy decomposition and code generation. + +Tool Calling +------------ +Both actors accept an optional ``tool_runner`` parameter. When provided, +the available tools are converted to LangChain ``StructuredTool`` objects +and bound to the LLM via ``llm.bind_tools()``. After the LLM responds, +any ``tool_calls`` on the ``AIMessage`` are dispatched through the +``ToolRunner`` and the results are fed back as ``ToolMessage`` objects in +a follow-up invocation so the LLM can incorporate the tool outputs. + +The ``builtin/plan-subplan`` tool is the primary tool used during the +strategize phase to emit subplan spawn decisions. """ from __future__ import annotations @@ -31,6 +43,7 @@ from cleveragents.tool.builtins.changeset import ChangeSet, ChangeSetEntry if TYPE_CHECKING: from cleveragents.providers.registry import ProviderRegistry + from cleveragents.tool.runner import ToolRunner logger = structlog.get_logger(__name__) @@ -54,6 +67,97 @@ def _parse_actor_name(actor_name: str) -> tuple[str, str]: return ("openai", actor_name) +def _build_langchain_tools(tool_runner: ToolRunner) -> list[Any]: + """Convert ``ToolRunner`` specs to LangChain ``StructuredTool`` objects. + + Each ``ToolSpec`` in the runner's registry is wrapped in a + ``StructuredTool`` so it can be passed to ``llm.bind_tools()``. + + Args: + tool_runner: The runner whose registry provides the tool specs. + + Returns: + A list of LangChain ``StructuredTool`` instances. + """ + from langchain_core.tools import StructuredTool + + lc_tools: list[Any] = [] + for spec in tool_runner.discover(): + # Capture spec in closure to avoid late-binding issues + _spec = spec + + def _make_handler(s: Any) -> Any: + def _handler(**kwargs: Any) -> Any: + return s.handler(kwargs) + + return _handler + + lc_tool = StructuredTool.from_function( + func=_make_handler(_spec), + name=_spec.name, + description=_spec.description, + ) + lc_tools.append(lc_tool) + return lc_tools + + +def _dispatch_tool_calls( + tool_calls: list[Any], + tool_runner: ToolRunner, + plan_id: str, +) -> list[Any]: + """Execute LLM-requested tool calls and return ``ToolMessage`` objects. + + Args: + tool_calls: The ``tool_calls`` list from an ``AIMessage``. + tool_runner: Runner used to execute each tool. + plan_id: Plan identifier for logging. + + Returns: + A list of ``ToolMessage`` objects with tool results. + """ + from langchain_core.messages.tool import ToolMessage + + tool_messages: list[Any] = [] + for tc in tool_calls: + tool_name: str = tc.get("name", "") if isinstance(tc, dict) else tc.name + tool_call_id: str = tc.get("id", str(ULID())) if isinstance(tc, dict) else tc.id + args: dict[str, Any] = tc.get("args", {}) if isinstance(tc, dict) else tc.args + + logger.debug( + "Dispatching tool call", + plan_id=plan_id, + tool_name=tool_name, + tool_call_id=tool_call_id, + ) + + try: + tool_runner.activate(tool_name) + result = tool_runner.execute(tool_name, args) + if result.success: + output_str = str(result.output) if result.output is not None else "" + else: + error_msg = result.error or "tool execution failed" + logger.warning( + "Tool call returned failure", + plan_id=plan_id, + tool_name=tool_name, + error=error_msg, + ) + output_str = f"error: {error_msg}" + except Exception as exc: + logger.warning( + "Tool call failed", + plan_id=plan_id, + tool_name=tool_name, + error=str(exc), + ) + output_str = f"error: {exc}" + + tool_messages.append(ToolMessage(content=output_str, tool_call_id=tool_call_id)) + return tool_messages + + class LLMStrategizeActor: """Strategize actor that uses a real LLM to decompose tasks. @@ -61,12 +165,18 @@ class LLMStrategizeActor: ``ProviderRegistry.create_llm()`` and asks it to break the ``definition_of_done`` into discrete implementation steps, returning them as a list of ``StrategyDecision`` objects. + + When a ``tool_runner`` is provided the LLM is given access to the + registered tools (including ``builtin/plan-subplan``) via + ``bind_tools()``. Any tool calls in the response are dispatched + through the runner before the final text response is parsed. """ def __init__( self, provider_registry: ProviderRegistry | None, lifecycle_service: Any, + tool_runner: ToolRunner | None = None, ) -> None: if provider_registry is None: raise ValidationError("provider_registry must not be None") @@ -74,6 +184,7 @@ class LLMStrategizeActor: raise ValidationError("lifecycle_service must not be None") self._registry = provider_registry self._lifecycle = lifecycle_service + self._tool_runner = tool_runner self._logger = logger.bind(actor="llm_strategize") def execute( @@ -83,7 +194,13 @@ class LLMStrategizeActor: invariants: list[PlanInvariant] | None = None, stream_callback: StreamCallback | None = None, ) -> StrategizeResult: - """Invoke the LLM to decompose *definition_of_done* into steps.""" + """Invoke the LLM to decompose *definition_of_done* into steps. + + When a ``tool_runner`` is configured the LLM is bound to the + available tools. Tool calls in the response are dispatched and + the results fed back so the LLM can incorporate them before + producing the final step list. + """ if not plan_id: raise ValidationError("plan_id must not be empty") @@ -108,11 +225,31 @@ class LLMStrategizeActor: llm = self._registry.create_llm(provider_type=provider_type, model_id=model_id) + # Bind tools when a runner is available + lc_tools: list[Any] = [] + if self._tool_runner is not None: + lc_tools = _build_langchain_tools(self._tool_runner) + if lc_tools: + llm = llm.bind_tools(lc_tools) # type: ignore[assignment] + self._logger.debug( + "Bound tools to strategize LLM", + plan_id=plan_id, + tool_names=[t.name for t in lc_tools], + ) + dod = definition_of_done or "Complete the plan objectives" + tool_hint = "" + if lc_tools: + tool_hint = ( + "\n\nIf this task requires coordinating multiple independent " + "projects or repositories, use the 'builtin/plan-subplan' tool " + "to spawn child subplans for each project before listing steps." + ) prompt = ( "You are an expert software architect. Analyze the following task " "and break it into concrete, sequential implementation steps.\n\n" f"Task:\n{dod}\n\n" + f"{tool_hint}" "Return ONLY a numbered list (1., 2., …) of steps. " "Each step must be a single, actionable change. " "Do not include commentary before or after the list." @@ -120,9 +257,29 @@ class LLMStrategizeActor: from langchain_core.messages import HumanMessage - # TODO(#650): Wire actor-configured response_format into provider calls - # when structured-output enforcement is implemented in runtime execution. - response = llm.invoke([HumanMessage(content=prompt)]) + messages: list[Any] = [HumanMessage(content=prompt)] + + # Invoke LLM — may return tool calls + response = llm.invoke(messages) + + # Dispatch any tool calls and follow up if needed + tool_calls = getattr(response, "tool_calls", []) or [] + if tool_calls and self._tool_runner is not None: + self._logger.info( + "Strategize LLM requested tool calls", + plan_id=plan_id, + tool_call_count=len(tool_calls), + ) + if stream_callback is not None: + stream_callback( + "strategize_tool_calls", + {"plan_id": plan_id, "tool_call_count": len(tool_calls)}, + ) + tool_messages = _dispatch_tool_calls(tool_calls, self._tool_runner, plan_id) + # Follow-up invocation with tool results + messages = [*messages, response, *tool_messages] + response = llm.invoke(messages) + content = response.content if hasattr(response, "content") else str(response) self._logger.debug( @@ -214,6 +371,11 @@ class LLMExecuteActor: Resolves the plan's ``execution_actor`` name to a live LLM via ``ProviderRegistry.create_llm()`` and asks it to produce file changes for the strategy decisions, returning a ``ChangeSet``. + + When a ``tool_runner`` is provided the LLM is given access to the + registered tools via ``bind_tools()``. Any tool calls in the + response are dispatched through the runner before the final file + blocks are parsed. """ def __init__( @@ -221,6 +383,7 @@ class LLMExecuteActor: provider_registry: ProviderRegistry | None, lifecycle_service: Any, context_assembler: ExecutePhaseContextAssembler | None = None, + tool_runner: ToolRunner | None = None, ) -> None: if provider_registry is None: raise ValidationError("provider_registry must not be None") @@ -229,6 +392,7 @@ class LLMExecuteActor: self._registry = provider_registry self._lifecycle = lifecycle_service self._context_assembler = context_assembler + self._tool_runner = tool_runner self._logger = logger.bind(actor="llm_execute") @staticmethod @@ -267,7 +431,16 @@ class LLMExecuteActor: *, read_only: bool = False, ) -> ExecuteResult: - """Invoke the LLM to generate file changes for the given decisions.""" + """Invoke the LLM to generate file changes for the given decisions. + + When a ``tool_runner`` is configured (either via constructor or the + *tool_runner* argument) the LLM is bound to the available tools. + Tool calls in the response are dispatched and the results fed back + before the final file blocks are parsed. + + The *tool_runner* argument takes precedence over the constructor + value when both are provided. + """ if not plan_id: raise ValidationError("plan_id must not be empty") @@ -290,6 +463,21 @@ class LLMExecuteActor: llm = self._registry.create_llm(provider_type=provider_type, model_id=model_id) + # Resolve effective tool runner (argument overrides constructor) + effective_runner: ToolRunner | None = tool_runner or self._tool_runner + + # Bind tools when a runner is available + lc_tools: list[Any] = [] + if effective_runner is not None: + lc_tools = _build_langchain_tools(effective_runner) + if lc_tools: + llm = llm.bind_tools(lc_tools) # type: ignore[assignment] + self._logger.debug( + "Bound tools to execute LLM", + plan_id=plan_id, + tool_names=[t.name for t in lc_tools], + ) + assembled_context: AssembledContext | None = None if self._context_assembler is not None: try: @@ -322,9 +510,29 @@ class LLMExecuteActor: from langchain_core.messages import HumanMessage - # TODO(#650): Wire actor-configured response_format into provider calls - # when structured-output enforcement is implemented in runtime execution. - response = llm.invoke([HumanMessage(content=prompt)]) + messages: list[Any] = [HumanMessage(content=prompt)] + + # Invoke LLM — may return tool calls + response = llm.invoke(messages) + + # Dispatch any tool calls and follow up if needed + tool_calls = getattr(response, "tool_calls", []) or [] + if tool_calls and effective_runner is not None: + self._logger.info( + "Execute LLM requested tool calls", + plan_id=plan_id, + tool_call_count=len(tool_calls), + ) + if stream_callback is not None: + stream_callback( + "execute_tool_calls", + {"plan_id": plan_id, "tool_call_count": len(tool_calls)}, + ) + tool_messages = _dispatch_tool_calls(tool_calls, effective_runner, plan_id) + # Follow-up invocation with tool results + messages = [*messages, response, *tool_messages] + response = llm.invoke(messages) + content = response.content if hasattr(response, "content") else str(response) self._logger.debug( diff --git a/src/cleveragents/application/services/plan_executor.py b/src/cleveragents/application/services/plan_executor.py index 995e21470..318356f28 100644 --- a/src/cleveragents/application/services/plan_executor.py +++ b/src/cleveragents/application/services/plan_executor.py @@ -6,6 +6,11 @@ phases. When a ``PlanExecutionContext`` is provided, the execute phase delegates to ``RuntimeExecuteActor`` for full changeset capture. Updated in M4 to add optional checkpoint hooks via ``CheckpointManager``. + +Updated in M5 to add optional ``SubplanService`` injection. When provided, +``PlanExecutor`` calls ``SubplanService.spawn()`` after the execute phase +completes to materialise any subplan spawn decisions the LLM emitted via +the ``builtin/plan-subplan`` tool. """ from __future__ import annotations @@ -54,6 +59,7 @@ if TYPE_CHECKING: from cleveragents.application.services.error_recovery_service import ( ErrorRecoveryService, ) + from cleveragents.application.services.subplan_service import SubplanService from cleveragents.infrastructure.observability.metrics_emitter import ( MetricsEmitter, ) @@ -303,6 +309,7 @@ class PlanExecutor: strategize_actor: Any | None = None, execute_actor: Any | None = None, fix_revalidate_orchestrator: FixThenRevalidateOrchestrator | None = None, + subplan_service: SubplanService | None = None, ) -> None: """Initialize the plan executor. @@ -329,6 +336,11 @@ class PlanExecutor: fix_revalidate_orchestrator: Optional orchestrator for fix-then-revalidate loops when required validations fail during execution (Forgejo #583). + subplan_service: Optional ``SubplanService`` for materialising + subplan spawn decisions after the execute phase. When + provided, ``PlanExecutor`` calls ``SubplanService.spawn()`` + for any ``SUBPLAN_SPAWN`` / ``SUBPLAN_PARALLEL_SPAWN`` + decisions recorded during execution. """ if lifecycle_service is None: raise ValidationError("lifecycle_service must not be None") @@ -341,6 +353,7 @@ class PlanExecutor: self._guardrail_service = guardrail_service self._metrics_emitter = metrics_emitter self._fix_revalidate_orchestrator = fix_revalidate_orchestrator + self._subplan_service = subplan_service self._strategize_actor = strategize_actor or StrategizeStubActor() self._execute_actor = execute_actor or ExecuteStubActor() self._logger = logger.bind(service="plan_executor") @@ -397,6 +410,95 @@ class PlanExecutor: """Return the fix-then-revalidate orchestrator, if configured.""" return self._fix_revalidate_orchestrator + @property + def subplan_service(self) -> SubplanService | None: + """Return the subplan service, if configured.""" + return self._subplan_service + + def _try_spawn_subplans(self, plan_id: str) -> None: + """Materialise subplan spawn decisions via ``SubplanService``. + + Called after the execute phase completes successfully. Queries + the ``SubplanService`` for any ``SUBPLAN_SPAWN`` / + ``SUBPLAN_PARALLEL_SPAWN`` decisions recorded during execution, + builds ``SpawnEntry`` objects, and calls ``SubplanService.spawn()`` + to create child plans and populate ``plan.subplan_statuses``. + + This is a best-effort operation — failures are logged but do not + propagate to the caller so that the parent plan's execute result + is not affected. + + Args: + plan_id: The plan identifier. + """ + if self._subplan_service is None: + return + + try: + from cleveragents.domain.models.core.plan import ( + ExecutionMode, + SubplanConfig, + SubplanMergeStrategy, + ) + + spawn_decisions = self._subplan_service.get_spawn_decisions(plan_id) + if not spawn_decisions: + self._logger.debug( + "No subplan spawn decisions found", + plan_id=plan_id, + ) + return + + self._logger.info( + "Spawning subplans from decisions", + plan_id=plan_id, + decision_count=len(spawn_decisions), + ) + + plan = self._lifecycle.get_plan(plan_id) + spawn_entries = self._subplan_service.build_spawn_entries(spawn_decisions) + + # Determine execution mode from decisions + from cleveragents.domain.models.core.decision import DecisionType + + has_parallel = any( + d.decision_type == DecisionType.SUBPLAN_PARALLEL_SPAWN + for d in spawn_decisions + ) + exec_mode = ( + ExecutionMode.PARALLEL if has_parallel else ExecutionMode.SEQUENTIAL + ) + + config = SubplanConfig( + execution_mode=exec_mode, + merge_strategy=SubplanMergeStrategy.GIT_THREE_WAY, + max_parallel=len(spawn_entries), + ) + + spawn_result = self._subplan_service.spawn( + parent_plan=plan, + config=config, + spawn_entries=spawn_entries, + ) + + # Persist updated parent plan with subplan_statuses + plan.timestamps.updated_at = datetime.now(tz=UTC) + self._lifecycle._commit_plan(plan) + + self._logger.info( + "Subplans spawned successfully", + plan_id=plan_id, + total_spawned=spawn_result.total_spawned, + execution_mode=spawn_result.execution_mode, + ) + except Exception as exc: + self._logger.warning( + "Subplan spawning failed (non-fatal)", + plan_id=plan_id, + error=str(exc), + exc_info=True, + ) + def _try_create_checkpoint( self, plan_id: str, @@ -700,6 +802,7 @@ class PlanExecutor: self._lifecycle._commit_plan(plan) self._try_create_checkpoint(plan_id, "post_execute", {"status": "success"}) self._lifecycle.complete_execute(plan_id) + self._try_spawn_subplans(plan_id) self._try_emit_metric( OperationalMetricKey.PLAN_DURATION_MS, plan_id, _duration_ms ) @@ -772,6 +875,7 @@ class PlanExecutor: plan_id, "post_execute", {"status": "success"} ) self._lifecycle.complete_execute(plan_id) + self._try_spawn_subplans(plan_id) _duration_ms = (time.monotonic_ns() - _start_ns) / 1_000_000 self._try_emit_metric( OperationalMetricKey.PLAN_DURATION_MS, plan_id, _duration_ms diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index 610ebc1bc..7eaebd1b1 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/src/cleveragents/cli/commands/plan.py @@ -1272,6 +1272,12 @@ def _get_plan_executor(lifecycle_service: PlanLifecycleService | None = None) -> ``LLMExecuteActor`` so that ``plan execute`` invocations drive real LLM calls instead of the local-only stub actors. + The ``builtin/plan-subplan`` tool is registered in a ``ToolRegistry`` + and provided to both actors via a ``ToolRunner``. This allows the LLM + to emit subplan spawn decisions during the strategize or execute phase. + A ``SubplanService`` is also wired in so that ``PlanExecutor`` can + materialise those decisions into child plans after execution completes. + Args: lifecycle_service: Optional pre-existing ``PlanLifecycleService`` instance. When provided the executor shares the same service @@ -1287,15 +1293,53 @@ def _get_plan_executor(lifecycle_service: PlanLifecycleService | None = None) -> LLMStrategizeActor, ) from cleveragents.application.services.plan_executor import PlanExecutor + from cleveragents.tool.builtins.subplan_tool import make_plan_subplan_spec + from cleveragents.tool.registry import ToolRegistry + from cleveragents.tool.runner import ToolRunner container = get_container() registry = container.provider_registry() if lifecycle_service is None: lifecycle_service = _get_lifecycle_service() + # Build a ToolRegistry with the builtin/plan-subplan tool wired to the + # DecisionService so that spawn decisions are persisted immediately. + # + # The subplan_tool._DecisionRecorder protocol expects + # ``record_decision(decision: Decision) -> Decision``, but + # ``DecisionService.record_decision`` takes individual fields. We + # bridge the gap with a thin adapter that satisfies the protocol. + class _DecisionRecorderAdapter: + """Adapter that satisfies the _DecisionRecorder protocol.""" + + def __init__(self, svc: Any) -> None: + self._svc = svc + + def record_decision(self, decision: Any) -> Any: + self._svc.record_decision( + plan_id=decision.plan_id, + decision_type=decision.decision_type, + question=decision.question, + chosen_option=decision.chosen_option, + rationale=decision.rationale, + parent_decision_id=decision.parent_decision_id, + ) + return decision + + decision_service = container.decision_service() + subplan_service = container.subplan_service() + tool_registry = ToolRegistry() + tool_registry.register( + make_plan_subplan_spec( + decision_service=_DecisionRecorderAdapter(decision_service) + ) + ) + tool_runner = ToolRunner(registry=tool_registry) + strategize_actor = LLMStrategizeActor( provider_registry=registry, lifecycle_service=lifecycle_service, + tool_runner=tool_runner, ) context_assembler = ACMSExecutePhaseContextAssembler( context_tier_service=container.context_tier_service(), @@ -1306,12 +1350,15 @@ def _get_plan_executor(lifecycle_service: PlanLifecycleService | None = None) -> provider_registry=registry, lifecycle_service=lifecycle_service, context_assembler=context_assembler, + tool_runner=tool_runner, ) return PlanExecutor( lifecycle_service=lifecycle_service, + tool_runner=tool_runner, strategize_actor=strategize_actor, execute_actor=execute_actor, + subplan_service=subplan_service, ) -- 2.52.0