diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index e53b921a..e37b9376 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -52,6 +52,24 @@ VALID_RETRY_POLICIES = frozenset(_RETRY_POLICY_MAP.keys()) +def _iter_child_agents(agent: Agent): + """Yield an agent's direct sub-agents. + + Covers both the positional ``agents=[...]`` list and the PLAN_EXECUTE + named slots ``planner=`` / ``fallback=``. The named slots are stored as + separate attributes (``agent.planner`` / ``agent.fallback``) and are not + present in ``agent.agents``, so any tree walk that only iterates + ``agent.agents`` silently skips them — leaving the planner's tools with no + workers started (tasks sit SCHEDULED forever). + """ + for sub in agent.agents: + yield sub + for slot in ("planner", "fallback"): + sub = getattr(agent, slot, None) + if isinstance(sub, Agent): + yield sub + + def _resolve_retry_logic(policy: str) -> str: """Convert a user-friendly retry policy name to the Conductor retry logic constant.""" key = policy.lower().strip() @@ -1246,7 +1264,7 @@ def _collect_worker_names(self, agent: Agent, *, required_workers: Optional[set] continue # Recurse into sub-agents for their tool names - for sub in agent.agents: + for sub in _iter_child_agents(agent): if getattr(sub, "is_claude_code", False): tool_names.add(sub.name) elif not getattr(sub, "external", False): @@ -1508,7 +1526,7 @@ def _server_needs(task_name: str) -> bool: self._register_manual_selection_worker(agent, domain=domain) # Recurse into sub-agents - for sub in agent.agents: + for sub in _iter_child_agents(agent): if getattr(sub, "is_claude_code", False): if _server_needs(sub.name): # Register passthrough worker for claude-code sub-agent @@ -2235,7 +2253,7 @@ def _collect(a: Agent) -> None: return if a.model and a.model not in seen: seen.add(a.model) - for sub in a.agents: + for sub in _iter_child_agents(a): _collect(sub) _collect(agent) @@ -2295,7 +2313,7 @@ def _has_worker_tools(self, agent: Agent) -> bool: if not getattr(nested_agent, "external", False): if self._has_worker_tools(nested_agent): return True - return any(self._has_worker_tools(sub) for sub in agent.agents) + return any(self._has_worker_tools(sub) for sub in _iter_child_agents(agent)) # ── Plan (compile without executing) ──────────────────────────── diff --git a/tests/unit/ai/test_plan_execute_worker_traversal.py b/tests/unit/ai/test_plan_execute_worker_traversal.py new file mode 100644 index 00000000..09b29f04 --- /dev/null +++ b/tests/unit/ai/test_plan_execute_worker_traversal.py @@ -0,0 +1,246 @@ +"""Regression tests: PLAN_EXECUTE ``planner=`` / ``fallback=`` tools get workers. + +Issue #478: tools declared on an agent passed via the ``planner=`` (or +``fallback=``) named slot were never traversed when the runtime collected / +registered worker names. Because PLAN_EXECUTE stores those agents as separate +attributes (``agent.planner`` / ``agent.fallback``) rather than in +``agent.agents``, a tree walk that only iterated ``agent.agents`` skipped them, +so the planner's tasks stayed SCHEDULED with nothing polling them — the +``86_coding_agent.py`` example hung forever. + +These tests are fully deterministic: pure dataclasses and mocks, no LLM, no +live server, no Docker. They assert only against pre-existing public/private +methods of ``AgentRuntime`` (``_collect_worker_names``, ``_register_workers``, +``_has_worker_tools``, ``_ensure_models_for_agent``), never against helpers +introduced by the fix, so they fail on the original code. +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from conductor.ai.agents import Agent, Strategy +from conductor.ai.agents.runtime.runtime import AgentRuntime +from conductor.ai.agents.tool import get_tool_def, tool + + +@tool +def planner_only_tool(path: str) -> str: + """A tool that exists ONLY on the planner slot.""" + return path + + +@tool +def fallback_only_tool(path: str) -> str: + """A tool that exists ONLY on the fallback slot.""" + return path + + +@tool +def harness_tool(path: str) -> str: + """A tool declared on the parent harness.""" + return path + + +def _collect_names(agent: Agent) -> set: + """Call _collect_worker_names without a real server connection.""" + rt = AgentRuntime.__new__(AgentRuntime) + return rt._collect_worker_names(agent) + + +class TestPlannerSlotWorkerCollection: + """_collect_worker_names must walk the planner= / fallback= named slots.""" + + def test_planner_tool_is_collected(self): + planner = Agent(name="planner", model="openai/gpt-4o", tools=[planner_only_tool]) + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[harness_tool], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + ) + + names = _collect_names(parent) + + # The parent's own tool is collected... + assert "harness_tool" in names + # ...and so is the planner's tool (THE BUG: this was missing). + assert "planner_only_tool" in names + + def test_fallback_tool_is_collected(self): + planner = Agent(name="planner", model="openai/gpt-4o", tools=[planner_only_tool]) + fallback = Agent(name="fallback", model="openai/gpt-4o", tools=[fallback_only_tool]) + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[harness_tool], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + fallback=fallback, + ) + + names = _collect_names(parent) + + assert "planner_only_tool" in names + assert "fallback_only_tool" in names + + def test_planner_system_workers_are_collected(self): + """System workers on the planner (guardrails) must also be walked.""" + planner = Agent( + name="planner", + model="openai/gpt-4o", + termination=lambda result, iteration: False, + ) + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[harness_tool], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + ) + + names = _collect_names(parent) + + assert "planner_termination" in names + + +class TestPlannerSlotHasWorkerTools: + """_has_worker_tools must recurse into planner= / fallback= slots.""" + + def test_planner_only_worker_tool_is_detected(self): + planner = Agent(name="planner", model="openai/gpt-4o", tools=[planner_only_tool]) + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[harness_tool], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + ) + + rt = AgentRuntime.__new__(AgentRuntime) + + # Sanity: the parent itself has a worker tool, so it's true regardless. + assert rt._has_worker_tools(parent) is True + # The planner slot must independently report True. + assert rt._has_worker_tools(planner) is True + + def test_planner_slot_detected_when_parent_has_only_server_tools(self): + """A planner with a worker tool is detected even if the parent's own + tools are server-side (http) and need no local worker.""" + from conductor.ai.agents.tool import http_tool + + parent_http = http_tool(name="fetch", description="d", url="http://example.test") + planner = Agent(name="planner", model="openai/gpt-4o", tools=[planner_only_tool]) + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[parent_http], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + ) + + rt = AgentRuntime.__new__(AgentRuntime) + + # Parent alone has no worker tools; True only because of the planner slot. + assert rt._has_worker_tools(parent) is True + + +class TestPlannerSlotRegistration: + """_register_workers must register the planner's tool workers.""" + + def test_planner_tool_worker_registered(self): + planner = Agent(name="planner", model="openai/gpt-4o", tools=[planner_only_tool]) + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[harness_tool], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + ) + + rt = AgentRuntime.__new__(AgentRuntime) + registered: list = [] + + def _fake_register_tool_workers(tools, agent_name, **kwargs): + for t in tools: + registered.append((agent_name, get_tool_def(t).name)) + + with patch( + "conductor.ai.agents.runtime.tool_registry.ToolRegistry.register_tool_workers", + side_effect=_fake_register_tool_workers, + ): + rt._register_workers(parent, required_workers=None, domain=None) + + registered_names = {name for _, name in registered} + assert "harness_tool" in registered_names + # THE BUG: the planner's tool worker was never registered. + assert "planner_only_tool" in registered_names + + def test_fallback_tool_worker_registered(self): + planner = Agent(name="planner", model="openai/gpt-4o", tools=[planner_only_tool]) + fallback = Agent(name="fallback", model="openai/gpt-4o", tools=[fallback_only_tool]) + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[harness_tool], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + fallback=fallback, + ) + + rt = AgentRuntime.__new__(AgentRuntime) + registered: list = [] + + def _fake_register_tool_workers(tools, agent_name, **kwargs): + for t in tools: + registered.append((agent_name, get_tool_def(t).name)) + + with patch( + "conductor.ai.agents.runtime.tool_registry.ToolRegistry.register_tool_workers", + side_effect=_fake_register_tool_workers, + ): + rt._register_workers(parent, required_workers=None, domain=None) + + registered_names = {name for _, name in registered} + assert "planner_only_tool" in registered_names + assert "fallback_only_tool" in registered_names + + +class TestPlannerSlotModelEnsuring: + """_ensure_models_for_agent must walk planner= / fallback= slots.""" + + def _make_runtime(self): + with ( + patch("conductor.client.orkes_clients.OrkesClients") as MockClients, + patch("conductor.ai.agents.runtime.worker_manager.TaskHandler", create=True), + ): + mock_clients = MagicMock() + MockClients.return_value = mock_clients + mock_clients.get_integration_client.return_value = MagicMock() + + from conductor.ai.agents.runtime.config import AgentConfig + + runtime = AgentRuntime( + settings=AgentConfig(auto_register_integrations=True) + ) + return runtime + + def test_planner_model_is_ensured(self): + planner = Agent(name="planner", model="anthropic/claude-sonnet-4-20250514") + parent = Agent( + name="harness", + model="openai/gpt-4o", + tools=[harness_tool], + strategy=Strategy.PLAN_EXECUTE, + planner=planner, + ) + + runtime = self._make_runtime() + with patch.object(runtime, "_ensure_model") as mock_ensure: + runtime._ensure_models_for_agent(parent) + + ensured = {c.args[0] for c in mock_ensure.call_args_list} + assert "openai/gpt-4o" in ensured + # THE BUG: the planner's model was never ensured. + assert "anthropic/claude-sonnet-4-20250514" in ensured