From 9c6dc613017c08a8e79ec16e99e92ca337f10e5d Mon Sep 17 00:00:00 2001 From: Ken Chau Date: Wed, 28 Oct 2026 21:22:41 +0000 Subject: [PATCH 1/2] Preserve descendant limits and settle owned subagent execution --- .../subagent_tools.py | 84 ++++++- .../loop-agent/tests/test_subagent_tools.py | 212 ++++++++++++++---- 2 files changed, 245 insertions(+), 51 deletions(-) diff --git a/modules/loop-agent/amplifier_module_loop_agent/subagent_tools.py b/modules/loop-agent/amplifier_module_loop_agent/subagent_tools.py index ef297eb..4702876 100644 --- a/modules/loop-agent/amplifier_module_loop_agent/subagent_tools.py +++ b/modules/loop-agent/amplifier_module_loop_agent/subagent_tools.py @@ -14,6 +14,7 @@ from __future__ import annotations +import asyncio import json import logging import uuid @@ -42,6 +43,7 @@ class SubagentState: model: str = "" # M-8: optional model override for subagent pending_messages: list[str] = field(default_factory=list) result: str | None = None + execution: asyncio.Task | None = None # --------------------------------------------------------------------------- @@ -287,6 +289,14 @@ async def execute(self, arguments: dict[str, Any]) -> ToolResult: output=f"Agent not found: {agent_id}", ) + if state.status == "closed": + return ToolResult( + success=False, + output=json.dumps({"agent_id": agent_id, "status": state.status}), + ) + if state.execution is not None: + return await self._join(state) + # If already completed or failed, return cached result if state.status in ("completed", "failed"): return ToolResult( @@ -327,17 +337,48 @@ async def execute(self, arguments: dict[str, Any]) -> ToolResult: "parent_session": parent_session, "agent_configs": agent_configs, } - # Pass max_turns via orchestrator_config if set + # Canonical coding-agent spec §7.3: descendants retain the limit and + # consume one level. Forward both loop and delegate depth channels. + spawn_kwargs["orchestrator_config"] = { + "current_depth": self._manager._current_depth + 1, + "max_subagent_depth": self._manager._max_depth, + } + spawn_kwargs["self_delegation_depth"] = self._manager._current_depth + 1 if state.max_turns: - spawn_kwargs["orchestrator_config"] = { - "max_turns": state.max_turns, - } + spawn_kwargs["orchestrator_config"]["max_turns"] = state.max_turns # Pass model as provider_preferences if set (M-8) if state.model: spawn_kwargs["provider_preferences"] = [{"model": state.model}] # Execute via session.spawn capability state.status = "running" + state.execution = asyncio.create_task(self._run(state, spawn_fn, spawn_kwargs)) + return await self._join(state) + + async def _join(self, state: SubagentState) -> ToolResult: + try: + return await asyncio.shield(state.execution) + except asyncio.CancelledError: + # close_agent cancelled the child, rather than this waiter. Return + # its actual terminal state to the pursuing agent. + if not asyncio.current_task().cancelling(): + return ToolResult( + success=False, + output=json.dumps( + {"agent_id": state.agent_id, "status": state.status} + ), + ) + state.execution.cancel() + try: + await state.execution + except asyncio.CancelledError: + pass + raise + + async def _run( + self, state: SubagentState, spawn_fn: Any, spawn_kwargs: dict + ) -> ToolResult: + agent_id = state.agent_id try: result = await spawn_fn(**spawn_kwargs) @@ -347,6 +388,26 @@ async def execute(self, arguments: dict[str, Any]) -> ToolResult: ) session_id = result.get("session_id") if isinstance(result, dict) else None + execution_status = ( + result.get("status", "success") + if isinstance(result, dict) + else "success" + ) + if execution_status != "success": + state.status = "failed" + state.result = output + return ToolResult( + success=False, + output=json.dumps( + { + "agent_id": agent_id, + "status": "failed", + "execution_status": execution_status, + "output": output, + "session_id": session_id, + } + ), + ) state.status = "completed" state.result = output state.pending_messages.clear() @@ -363,6 +424,10 @@ async def execute(self, arguments: dict[str, Any]) -> ToolResult: } ), ) + except asyncio.CancelledError: + state.status = "closed" if state.status == "closing" else "failed" + state.result = "Subagent execution cancelled" + raise except Exception as e: state.status = "failed" state.result = str(e) @@ -410,7 +475,16 @@ async def execute(self, arguments: dict[str, Any]) -> ToolResult: output=f"Agent not found: {agent_id}", ) - # Idempotent: closing already-closed agent is fine + # Terminate actual owned work before reporting closure (§7.2), and + # never allow a subsequent wait to restart a closed pending handle. + state.status = "closing" + if state.execution is not None and not state.execution.done(): + state.execution.cancel() + try: + await state.execution + except asyncio.CancelledError: + if asyncio.current_task().cancelling(): + raise state.status = "closed" state.pending_messages.clear() diff --git a/modules/loop-agent/tests/test_subagent_tools.py b/modules/loop-agent/tests/test_subagent_tools.py index 039c4d5..07d8f5c 100644 --- a/modules/loop-agent/tests/test_subagent_tools.py +++ b/modules/loop-agent/tests/test_subagent_tools.py @@ -98,9 +98,11 @@ async def test_spawn_returns_agent_id(self): mgr = SubagentManager(coordinator=coordinator) tool = _find_tool(mgr, "spawn_agent") - result = await tool.execute({ - "task": "Plan the feature", - }) + result = await tool.execute( + { + "task": "Plan the feature", + } + ) assert result.success assert "agent_id" in result.output @@ -151,11 +153,13 @@ async def test_spawn_accepts_optional_params(self): mgr = SubagentManager(coordinator=coordinator) tool = _find_tool(mgr, "spawn_agent") - result = await tool.execute({ - "task": "Plan feature", - "working_dir": "/tmp/work", - "max_turns": 25, - }) + result = await tool.execute( + { + "task": "Plan feature", + "working_dir": "/tmp/work", + "max_turns": 25, + } + ) assert result.success agent_id = _extract_agent_id(result.output) state = mgr._agents[agent_id] @@ -169,10 +173,12 @@ async def test_spawn_accepts_model_param(self): mgr = SubagentManager(coordinator=coordinator) tool = _find_tool(mgr, "spawn_agent") - result = await tool.execute({ - "task": "Analyze code", - "model": "claude-sonnet-4-20250514", - }) + result = await tool.execute( + { + "task": "Analyze code", + "model": "claude-sonnet-4-20250514", + } + ) assert result.success agent_id = _extract_agent_id(result.output) state = mgr._agents[agent_id] @@ -205,10 +211,12 @@ async def test_wait_passes_provider_preferences_when_model_set(self): spawn_tool = _find_tool(mgr, "spawn_agent") wait_tool = _find_tool(mgr, "wait") - spawn_result = await spawn_tool.execute({ - "task": "Work", - "model": "gpt-4o", - }) + spawn_result = await spawn_tool.execute( + { + "task": "Work", + "model": "gpt-4o", + } + ) agent_id = _extract_agent_id(spawn_result.output) await wait_tool.execute({"agent_id": agent_id}) @@ -253,10 +261,12 @@ async def test_send_input_stores_message(self): spawn_result = await spawn_tool.execute({"task": "Work"}) agent_id = _extract_agent_id(spawn_result.output) - result = await send_tool.execute({ - "agent_id": agent_id, - "message": "Focus on tests first", - }) + result = await send_tool.execute( + { + "agent_id": agent_id, + "message": "Focus on tests first", + } + ) assert result.success assert mgr._agents[agent_id].pending_messages == ["Focus on tests first"] @@ -267,10 +277,12 @@ async def test_send_input_unknown_agent_fails(self): mgr = SubagentManager(coordinator=coordinator) send_tool = _find_tool(mgr, "send_input") - result = await send_tool.execute({ - "agent_id": "nonexistent", - "message": "hello", - }) + result = await send_tool.execute( + { + "agent_id": "nonexistent", + "message": "hello", + } + ) assert not result.success assert "not found" in result.output.lower() @@ -287,10 +299,12 @@ async def test_send_input_closed_agent_fails(self): agent_id = _extract_agent_id(spawn_result.output) await close_tool.execute({"agent_id": agent_id}) - result = await send_tool.execute({ - "agent_id": agent_id, - "message": "hello", - }) + result = await send_tool.execute( + { + "agent_id": agent_id, + "message": "hello", + } + ) assert not result.success assert "closed" in result.output.lower() @@ -349,10 +363,12 @@ async def test_wait_includes_pending_messages(self): spawn_result = await spawn_tool.execute({"task": "Build feature"}) agent_id = _extract_agent_id(spawn_result.output) - await send_tool.execute({ - "agent_id": agent_id, - "message": "Focus on tests", - }) + await send_tool.execute( + { + "agent_id": agent_id, + "message": "Focus on tests", + } + ) await wait_tool.execute({"agent_id": agent_id}) # The instruction passed to session.spawn should include the task and message @@ -437,7 +453,9 @@ async def test_wait_passes_correct_spawn_kwargs(self): coordinator = _make_coordinator( spawn_result={"output": "done", "session_id": "child-123"}, session="parent-session-abc", - agents={"test-agent": {"session": {"orchestrator": {"module": "loop-agent"}}}}, + agents={ + "test-agent": {"session": {"orchestrator": {"module": "loop-agent"}}} + }, ) mgr = SubagentManager(coordinator=coordinator) spawn_tool = _find_tool(mgr, "spawn_agent") @@ -455,13 +473,16 @@ async def test_wait_passes_correct_spawn_kwargs(self): call_kwargs = coordinator._mock_spawn.call_args[1] assert call_kwargs["instruction"] == "Write hello.py" assert call_kwargs["parent_session"] == "parent-session-abc" - assert call_kwargs["agent_configs"] == {"test-agent": {"session": {"orchestrator": {"module": "loop-agent"}}}} + assert call_kwargs["agent_configs"] == { + "test-agent": {"session": {"orchestrator": {"module": "loop-agent"}}} + } assert "agent_name" in call_kwargs @pytest.mark.asyncio async def test_wait_returns_session_id_from_spawn(self): """wait returns session_id from the spawn result dict.""" import json + coordinator = _make_coordinator( spawn_result={"output": "Task done", "session_id": "child-999"}, ) @@ -552,15 +573,20 @@ async def test_subagent_tools_registered_when_depth_allows(self): coordinator.register_capability = MagicMock() provider = AsyncMock() - provider.complete = AsyncMock(return_value=ChatResponse( - content=[{"type": "text", "text": "done"}], - tool_calls=None, - usage=Usage(input_tokens=10, output_tokens=5, total_tokens=15), - )) + provider.complete = AsyncMock( + return_value=ChatResponse( + content=[{"type": "text", "text": "done"}], + tool_calls=None, + usage=Usage(input_tokens=10, output_tokens=5, total_tokens=15), + ) + ) hooks = MagicMock() hooks.emit = AsyncMock(return_value=MagicMock(action="continue")) - orch = AgentOrchestrator(coordinator, {"system_prompt": "You are a test coding agent.", "max_subagent_depth": 1}) + orch = AgentOrchestrator( + coordinator, + {"system_prompt": "You are a test coding agent.", "max_subagent_depth": 1}, + ) await orch.execute("hi", MagicMock(), {"test": provider}, {}, hooks) # Session should have the 4 subagent tools @@ -580,17 +606,24 @@ async def test_subagent_tools_not_registered_when_depth_exhausted(self): coordinator.register_capability = MagicMock() provider = AsyncMock() - provider.complete = AsyncMock(return_value=ChatResponse( - content=[{"type": "text", "text": "done"}], - tool_calls=None, - usage=Usage(input_tokens=10, output_tokens=5, total_tokens=15), - )) + provider.complete = AsyncMock( + return_value=ChatResponse( + content=[{"type": "text", "text": "done"}], + tool_calls=None, + usage=Usage(input_tokens=10, output_tokens=5, total_tokens=15), + ) + ) hooks = MagicMock() hooks.emit = AsyncMock(return_value=MagicMock(action="continue")) # current_depth=1, max_subagent_depth=1 => no subagent tools orch = AgentOrchestrator( - coordinator, {"system_prompt": "You are a test coding agent.", "max_subagent_depth": 1, "current_depth": 1} + coordinator, + { + "system_prompt": "You are a test coding agent.", + "max_subagent_depth": 1, + "current_depth": 1, + }, ) await orch.execute("hi", MagicMock(), {"test": provider}, {}, hooks) @@ -619,6 +652,7 @@ def _extract_agent_id(output: str) -> str: Expects output to contain 'agent_id: ' or similar. """ import json + try: data = json.loads(output) return data["agent_id"] @@ -631,3 +665,89 @@ def _extract_agent_id(output: str) -> str: if len(parts) >= 2: return parts[-1].strip().strip('"') raise ValueError(f"Could not extract agent_id from: {output}") + + +@pytest.mark.asyncio +async def test_descendant_depth_and_limit_reach_host_spawn(): + coordinator = _make_coordinator() + manager = SubagentManager(coordinator, max_depth=3, current_depth=1) + handle = _extract_agent_id( + (await _find_tool(manager, "spawn_agent").execute({"task": "bounded"})).output + ) + await _find_tool(manager, "wait").execute({"agent_id": handle}) + sent = coordinator._mock_spawn.call_args.kwargs + assert sent["orchestrator_config"]["current_depth"] == 2 + assert sent["orchestrator_config"]["max_subagent_depth"] == 3 + assert sent["self_delegation_depth"] == 2 + + +@pytest.mark.asyncio +async def test_closed_pending_handle_cannot_be_started_by_wait(): + coordinator = _make_coordinator() + manager = SubagentManager(coordinator) + handle = _extract_agent_id( + (await _find_tool(manager, "spawn_agent").execute({"task": "pending"})).output + ) + await _find_tool(manager, "close_agent").execute({"agent_id": handle}) + result = await _find_tool(manager, "wait").execute({"agent_id": handle}) + assert not result.success and "closed" in result.output + coordinator._mock_spawn.assert_not_called() + + +@pytest.mark.asyncio +async def test_active_close_stops_owned_execution_and_concurrent_waits_do_not_duplicate(): + import asyncio + import json + + coordinator = _make_coordinator() + entered, stopped = asyncio.Event(), asyncio.Event() + calls = [] + + async def execute(**kwargs): + calls.append(kwargs) + entered.set() + try: + await asyncio.Event().wait() + finally: + stopped.set() + + coordinator.get_capability.return_value = execute + manager = SubagentManager(coordinator) + handle = _extract_agent_id( + (await _find_tool(manager, "spawn_agent").execute({"task": "active"})).output + ) + first = asyncio.create_task( + _find_tool(manager, "wait").execute({"agent_id": handle}) + ) + await asyncio.wait_for(entered.wait(), 2) + second = asyncio.create_task( + _find_tool(manager, "wait").execute({"agent_id": handle}) + ) + closed = await _find_tool(manager, "close_agent").execute({"agent_id": handle}) + results = await asyncio.gather(first, second) + assert stopped.is_set() + assert len(calls) == 1 + assert json.loads(closed.output)["status"] == "closed" + assert all(not result.success and "closed" in result.output for result in results) + assert not (await _find_tool(manager, "wait").execute({"agent_id": handle})).success + assert len(calls) == 1 + + +@pytest.mark.parametrize("status", ["incomplete", "cancelled", "failed"]) +@pytest.mark.asyncio +async def test_unfinished_helper_is_not_reported_as_completed(status): + coordinator = _make_coordinator( + spawn_result={ + "output": "partial result", + "session_id": "child", + "status": status, + } + ) + manager = SubagentManager(coordinator) + handle = _extract_agent_id( + (await _find_tool(manager, "spawn_agent").execute({"task": "work"})).output + ) + result = await _find_tool(manager, "wait").execute({"agent_id": handle}) + assert not result.success + assert status in result.output and "partial result" in result.output + assert manager._agents[handle].status == "failed" From 00df371957f1f0d85d3d6138dd774358ce92f903 Mon Sep 17 00:00:00 2001 From: Ken Chau Date: Wed, 28 Oct 2026 21:38:42 +0000 Subject: [PATCH 2/2] Keep root CI candidate overrides within declared runtime dependencies --- .github/workflows/ci.yml | 29 +++++++++++++++++++++++------ 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 25aa12e..18b426b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -145,16 +145,33 @@ jobs: - name: Install root package without uv source mappings run: | - # Only redirect same-repo packages to this candidate. Never override - # binding/engine/Foundation sources: their conflicts must remain visible. + # Redirect the root's same-repo runtime dependency closure only. + # A module elsewhere in this repository can share a distribution name + # with an engine dependency without implementing that dependency's API. + # Never override binding/engine/Foundation-owned sources. python - <<'PY' from pathlib import Path import tomllib + modules = { + tomllib.loads(path.read_text())["project"]["name"]: path + for path in Path("modules").glob("*/pyproject.toml") + } + pending = [Path("pyproject.toml")] + visited = set() lines = [] - for path in Path("modules").glob("*/pyproject.toml"): - name = tomllib.loads(path.read_text())["project"]["name"] - lines.append(f"{name} @ {path.parent.resolve().as_uri()}") - Path("candidate-overrides.txt").write_text("\n".join(lines) + "\n") + while pending: + project = tomllib.loads(pending.pop().read_text())["project"] + for dependency in project.get("dependencies", []): + if " @ git+https://github.com/microsoft/amplifier-bundle-dot-runner@" not in dependency: + continue + name = dependency.split(" @ ", 1)[0].split("[", 1)[0] + if name in visited: + continue + visited.add(name) + path = modules[name] + lines.append(f"{name} @ {path.parent.resolve().as_uri()}") + pending.append(path) + Path("candidate-overrides.txt").write_text("\n".join(sorted(lines)) + "\n") PY uv --no-config pip install --python .venv-root-install -e . \ --no-sources --overrides candidate-overrides.txt