From 799ef24842433de1c0225e7bf112ebcf5eb4ca40 Mon Sep 17 00:00:00 2001 From: Ousama Ben Younes Date: Mon, 27 Jul 2026 19:45:51 +0000 Subject: [PATCH] fix(runtime): make child agent limit reservation atomic --- strix/core/agents.py | 28 +++++++++++++++++++++++ strix/core/execution.py | 24 +++++++++----------- tests/test_execution.py | 49 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 14 deletions(-) diff --git a/strix/core/agents.py b/strix/core/agents.py index c96204df..d72236e4 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -168,6 +168,34 @@ class AgentCoordinator: logger.info("agent.register %s (%s) parent=%s", agent_id, name, parent_id or "-") await self._maybe_snapshot() + async def register_child_if_capacity( + self, + agent_id: str, + name: str, + parent_id: str, + *, + max_child_agents: int, + task: str | None = None, + skills: list[str] | None = None, + ) -> tuple[bool, int]: + async with self._lock: + child_count = sum(parent is not None for parent in self.parent_of.values()) + if max_child_agents > 0 and child_count >= max_child_agents: + return False, child_count + self.statuses[agent_id] = "running" + self.parent_of[agent_id] = parent_id + self.names[agent_id] = name + self.pending_counts.setdefault(agent_id, 0) + self.metadata[agent_id] = { + "task": task or "", + "skills": list(skills or []), + } + self.runtimes.setdefault(agent_id, AgentRuntime()) + child_count += 1 + logger.info("agent.register %s (%s) parent=%s", agent_id, name, parent_id) + await self._maybe_snapshot() + return True, child_count + async def attach_runtime( self, agent_id: str, diff --git a/strix/core/execution.py b/strix/core/execution.py index 6b7c3462..b8ce1cfa 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -310,8 +310,16 @@ async def spawn_child_agent( if not isinstance(parent_id, str): raise TypeError("Parent agent_id missing from context") - child_count = _child_agent_count(coordinator) - if max_child_agents > _UNLIMITED_CHILD_AGENTS and child_count >= max_child_agents: + child_id = uuid.uuid4().hex[:8] + registered, child_count = await coordinator.register_child_if_capacity( + child_id, + name, + parent_id, + max_child_agents=max_child_agents, + task=task, + skills=skills, + ) + if not registered: logger.info( "refusing to spawn child agent %r: limit %d already reached", name, @@ -328,15 +336,7 @@ async def spawn_child_agent( "current_child_agents": child_count, } - child_id = uuid.uuid4().hex[:8] child_agent = factory(name=name, skills=skills) - await coordinator.register( - child_id, - name, - parent_id, - task=task, - skills=skills, - ) await _start_child_runner( parent_ctx=parent_ctx, @@ -371,10 +371,6 @@ async def spawn_child_agent( } -def _child_agent_count(coordinator: AgentCoordinator) -> int: - return sum(parent_id is not None for parent_id in coordinator.parent_of.values()) - - async def respawn_subagents( *, coordinator: AgentCoordinator, diff --git a/tests/test_execution.py b/tests/test_execution.py index 4d3b5d25..94dccc1f 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -168,6 +168,55 @@ async def test_spawn_child_agent_respects_child_limit( assert set(coordinator.parent_of) == {"root", "child-1"} +@pytest.mark.asyncio +async def test_concurrent_spawn_child_agent_reserves_limit_atomically( + monkeypatch: pytest.MonkeyPatch, tmp_path: Any +) -> None: + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + + async def _noop_start(**_kwargs: Any) -> None: + return None + + monkeypatch.setattr("strix.core.execution._start_child_runner", _noop_start) + + def _factory(**_kwargs: Any) -> object: + return object() + + async def _spawn(name: str) -> dict[str, Any]: + return await spawn_child_agent( + coordinator=coordinator, + factory=_factory, + agents_db_path=tmp_path / "agents.db", + sessions_to_close=[], + run_config=cast("Any", object()), + max_turns=1, + interactive=False, + parent_ctx={"agent_id": "root"}, + name=name, + task="do more recon", + skills=[], + parent_history=[], + max_child_agents=1, + ) + + async with coordinator._lock: + tasks = [ + asyncio.create_task(_spawn("extra-a")), + asyncio.create_task(_spawn("extra-b")), + ] + await asyncio.sleep(0) + + results = await asyncio.gather(*tasks) + + assert [result["success"] for result in results].count(True) == 1 + assert _child_count(coordinator) == 1 + + +def _child_count(coordinator: AgentCoordinator) -> int: + return sum(parent_id is not None for parent_id in coordinator.parent_of.values()) + + @pytest.mark.asyncio async def test_concurrent_reserve_claims_yield_single_root() -> None: coordinator = AgentCoordinator()