feat(actors): wire LLM strategize/execute actors to subplan spawning infrastructure #1304
@@ -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
|
||||
@@ -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()
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user