Merge a60ab98a9c into 851743adcf

docs/advanced/configuration.mdx

@@ -198,6 +198,10 @@ When remote vars are set, Strix dual-writes telemetry to both local JSONL and th
   Runtime backend for the sandbox environment.
 </ParamField>
 
+<ParamField path="STRIX_MAX_CHILD_AGENTS" default="0" type="integer">
+  Maximum number of child agents a scan may spawn. Set to `0` for no limit.
+</ParamField>
+
 ## Sandbox Configuration
 
 <ParamField path="STRIX_SANDBOX_EXECUTION_TIMEOUT" default="120" type="integer">

strix/config/settings.py

@@ -170,6 +170,8 @@ class RuntimeSettings(BaseSettings):
     backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
     # Max screenshot/image tool outputs kept live per agent context (0 = none).
     max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
+    # Max spawned child agents per scan (0 = unlimited).
+    max_child_agents: int = Field(default=0, ge=0, alias="STRIX_MAX_CHILD_AGENTS")
 
 
 class TelemetrySettings(BaseSettings):

strix/core/agents.py

@@ -254,6 +254,26 @@ class AgentCoordinator:
                 runtime.wake.set()
             return next((aid for aid, parent in self.parent_of.items() if parent is None), None)
 
+    def _record_registration(
+        self,
+        agent_id: str,
+        name: str,
+        parent_id: str | None,
+        *,
+        task: str | None,
+        skills: list[str] | None,
+    ) -> None:
+        """Record a new agent in the graph; callers must hold ``self._lock``."""
+        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())
+
     async def register(
         self,
         agent_id: str,
@@ -264,18 +284,50 @@ class AgentCoordinator:
         skills: list[str] | None = None,
     ) -> None:
         async with self._lock:
-            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())
+            self._record_registration(agent_id, name, parent_id, task=task, skills=skills)
         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._record_registration(agent_id, name, parent_id, task=task, skills=skills)
+            child_count += 1
+        logger.info("agent.register %s (%s) parent=%s", agent_id, name, parent_id)
+        try:
+            await self._maybe_snapshot()
+        except BaseException:
+            # The slot is already recorded under the lock; if persistence is
+            # cancelled (CancelledError is a BaseException, so it escapes the
+            # best-effort guard in ``_maybe_snapshot``) undo the reservation so
+            # the atomic capacity count never leaks.
+            await self.discard_child_reservation(agent_id)
+            raise
+        return True, child_count
+
+    async def discard_child_reservation(self, agent_id: str) -> None:
+        """Forget a reserved child whose runner never started, freeing its slot."""
+        async with self._lock:
+            self.statuses.pop(agent_id, None)
+            self.parent_of.pop(agent_id, None)
+            self.names.pop(agent_id, None)
+            self.pending_counts.pop(agent_id, None)
+            self.metadata.pop(agent_id, None)
+            self.runtimes.pop(agent_id, None)
+        logger.info("agent.discard %s", agent_id)
+        await self._maybe_snapshot()
+
     async def attach_runtime(
         self,
         agent_id: str,

strix/core/execution.py

@@ -74,6 +74,8 @@ _IMAGE_REJECTION = re.compile(
     re.IGNORECASE,
 )
 _MAX_COMPACTIONS_PER_CYCLE = 2
+_UNLIMITED_CHILD_AGENTS = 0
+_CHILD_AGENT_LIMIT_ERROR = "child agent limit reached"
 
 
 @cache
@@ -402,45 +404,70 @@ async def spawn_child_agent(
     parent_history: list[Any],
     event_sink: StreamEventSink | None = None,
     hooks: RunHooks[dict[str, Any]] | None = None,
+    max_child_agents: int = _UNLIMITED_CHILD_AGENTS,
 ) -> dict[str, Any]:
     parent_id = parent_ctx.get("agent_id")
     if not isinstance(parent_id, str):
         raise TypeError("Parent agent_id missing from context")
 
     child_id = uuid.uuid4().hex[:8]
-    child_agent = factory(name=name, skills=skills)
-    await coordinator.register(
+    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,
+            max_child_agents,
+        )
+        return {
+            "success": False,
+            "error": _CHILD_AGENT_LIMIT_ERROR,
+            "message": (
+                f"Cannot spawn '{name}': configured child agent limit "
+                f"({max_child_agents}) is already reached."
+            ),
+            "limit": max_child_agents,
+            "current_child_agents": child_count,
+        }
 
-    await _start_child_runner(
-        parent_ctx=parent_ctx,
-        coordinator=coordinator,
-        agents_db_path=agents_db_path,
-        sessions_to_close=sessions_to_close,
-        run_config=run_config,
-        max_turns=max_turns,
-        interactive=interactive,
-        child_agent=child_agent,
-        child_id=child_id,
-        name=name,
-        parent_id=parent_id,
-        task=task,
-        skills=skills,
-        initial_input=child_initial_input(
-            name=name,
+    try:
+        child_agent = factory(name=name, skills=skills)
+        await _start_child_runner(
+            parent_ctx=parent_ctx,
+            coordinator=coordinator,
+            agents_db_path=agents_db_path,
+            sessions_to_close=sessions_to_close,
+            run_config=run_config,
+            max_turns=max_turns,
+            interactive=interactive,
+            child_agent=child_agent,
             child_id=child_id,
+            name=name,
             parent_id=parent_id,
             task=task,
-            parent_history=parent_history,
-        ),
-        event_sink=event_sink,
-        hooks=hooks,
-    )
+            skills=skills,
+            initial_input=child_initial_input(
+                name=name,
+                child_id=child_id,
+                parent_id=parent_id,
+                task=task,
+                parent_history=parent_history,
+            ),
+            event_sink=event_sink,
+            hooks=hooks,
+        )
+    except BaseException:
+        # Free the reserved slot on any failure, including cancellation
+        # (CancelledError is a BaseException), so a failed spawn never leaks
+        # child-agent capacity. The error/cancellation is re-raised unchanged.
+        await coordinator.discard_child_reservation(child_id)
+        raise
 
     return {
         "success": True,

strix/core/runner.py

@@ -583,6 +583,7 @@ async def run_strix_scan(
                 interactive=interactive,
                 event_sink=event_sink,
                 hooks=hooks,
+                max_child_agents=settings.runtime.max_child_agents,
                 **kwargs,
             )
 

tests/test_config_loader.py

@@ -10,7 +10,7 @@ from pydantic import AliasChoices, Field, ValidationError
 from pydantic.fields import FieldInfo
 
 from strix.config import loader
-from strix.config.settings import ContextSettings
+from strix.config.settings import ContextSettings, RuntimeSettings
 
 
 if TYPE_CHECKING:
@@ -35,6 +35,7 @@ _LLM_ENV_KEYS = [
     # RuntimeSettings
     "STRIX_IMAGE",
     "STRIX_RUNTIME_BACKEND",
+    "STRIX_MAX_CHILD_AGENTS",
     # TelemetrySettings
     "STRIX_TELEMETRY",
     # JSON-object settings
@@ -148,6 +149,10 @@ def test_tool_output_max_bytes_accepts_floor() -> None:
     assert ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=1024).tool_output_max_bytes == 1024
 
 
+def test_max_child_agents_env_alias() -> None:
+    assert RuntimeSettings(STRIX_MAX_CHILD_AGENTS=7).max_child_agents == 7
+
+
 # --------------------------------------------------------------------------- #
 # _aliases_for
 # --------------------------------------------------------------------------- #

tests/test_execution.py

@@ -22,6 +22,7 @@ from strix.core.agents import AgentCoordinator, WaitKind
 from strix.core.execution import (
     _notify_root_on_budget_reserve,
     notify_parent_on_terminal,
+    spawn_child_agent,
 )
 from strix.core.sessions import seed_initial_input
 from strix.tools.agents_graph.tools import agent_finish, stop_agent
@@ -153,6 +154,256 @@ async def test_reserve_notice_names_the_root_finish_tool(monkeypatch: pytest.Mon
     assert "finish_scan" not in str(sent[0]["content"])
 
 
+@pytest.mark.asyncio
+async def test_spawn_child_agent_respects_child_limit(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child-1", "recon", parent_id="root")
+
+    def _unexpected_factory(**_kwargs: Any) -> object:
+        raise AssertionError("child factory should not be called at the child limit")
+
+    async def _unexpected_start(**_kwargs: Any) -> None:
+        raise AssertionError("child runner should not start at the child limit")
+
+    monkeypatch.setattr("strix.core.execution._start_child_runner", _unexpected_start)
+
+    result = await spawn_child_agent(
+        coordinator=coordinator,
+        factory=_unexpected_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="extra",
+        task="do more recon",
+        skills=[],
+        parent_history=[],
+        max_child_agents=1,
+    )
+
+    assert result["success"] is False
+    assert "child agent limit" in result["error"]
+    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
+
+
+@pytest.mark.asyncio
+async def test_spawn_child_agent_releases_reservation_when_factory_fails(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    started: list[str] = []
+
+    async def _record_start(**kwargs: Any) -> None:
+        started.append(kwargs["name"])
+
+    monkeypatch.setattr("strix.core.execution._start_child_runner", _record_start)
+
+    def _failing_factory(**_kwargs: Any) -> object:
+        raise RuntimeError("agent build failed")
+
+    def _factory(**_kwargs: Any) -> object:
+        return object()
+
+    async def _spawn(name: str, factory: Any) -> 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,
+        )
+
+    with pytest.raises(RuntimeError, match="agent build failed"):
+        await _spawn("broken", _failing_factory)
+
+    assert set(coordinator.parent_of) == {"root"}
+    assert set(coordinator.statuses) == {"root"}
+
+    result = await _spawn("retry", _factory)
+
+    assert result["success"] is True
+    assert started == ["retry"]
+
+
+@pytest.mark.asyncio
+async def test_spawn_child_agent_releases_reservation_when_runner_start_fails(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    started: list[str] = []
+
+    def _factory(**_kwargs: Any) -> object:
+        return object()
+
+    async def _boom_start(**_kwargs: Any) -> None:
+        raise RuntimeError("runner start failed")
+
+    async def _ok_start(**kwargs: Any) -> None:
+        started.append(kwargs["name"])
+
+    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,
+        )
+
+    monkeypatch.setattr("strix.core.execution._start_child_runner", _boom_start)
+    with pytest.raises(RuntimeError, match="runner start failed"):
+        await _spawn("broken")
+
+    # The reserved slot must be freed when the runner never started, otherwise the
+    # child-agent limit leaks capacity and blocks every later spawn.
+    assert set(coordinator.parent_of) == {"root"}
+    assert set(coordinator.statuses) == {"root"}
+    assert _child_count(coordinator) == 0
+
+    monkeypatch.setattr("strix.core.execution._start_child_runner", _ok_start)
+    result = await _spawn("retry")
+
+    assert result["success"] is True
+    assert started == ["retry"]
+
+
+@pytest.mark.asyncio
+async def test_spawn_child_agent_releases_reservation_on_cancellation(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+
+    def _factory(**_kwargs: Any) -> object:
+        return object()
+
+    async def _cancelled_start(**_kwargs: Any) -> None:
+        raise asyncio.CancelledError
+
+    monkeypatch.setattr("strix.core.execution._start_child_runner", _cancelled_start)
+
+    with pytest.raises(asyncio.CancelledError):
+        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="cancelled",
+            task="do more recon",
+            skills=[],
+            parent_history=[],
+            max_child_agents=1,
+        )
+
+    # A cancellation while the runner is starting must still free the slot;
+    # CancelledError is a BaseException, so an ``except Exception`` would leak it.
+    assert set(coordinator.parent_of) == {"root"}
+    assert _child_count(coordinator) == 0
+
+
+@pytest.mark.asyncio
+async def test_register_child_if_capacity_undoes_reservation_on_snapshot_cancel(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+
+    # Only the reservation's own persistence is cancelled; the subsequent
+    # discard is allowed to complete so the slot is actually released.
+    snapshot_calls = {"n": 0}
+
+    async def _cancel_first_snapshot() -> None:
+        snapshot_calls["n"] += 1
+        if snapshot_calls["n"] == 1:
+            raise asyncio.CancelledError
+
+    monkeypatch.setattr(coordinator, "_maybe_snapshot", _cancel_first_snapshot)
+
+    with pytest.raises(asyncio.CancelledError):
+        await coordinator.register_child_if_capacity("leaked", "recon", "root", max_child_agents=5)
+
+    # Cancellation during post-lock persistence must not leave a phantom child
+    # consuming a capacity slot.
+    assert "leaked" not in coordinator.parent_of
+    assert _child_count(coordinator) == 0
+
+
+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()

tests/test_runner_root_prompt.py

@@ -67,7 +67,7 @@ def _patch_engine_scaffold(
             extra_headers=None,
             extra_body=None,
         ),
-        runtime=types.SimpleNamespace(max_context_images=3),
+        runtime=types.SimpleNamespace(max_context_images=3, max_child_agents=0),
         telemetry=types.SimpleNamespace(run_record_timings=False),
     )
     monkeypatch.setattr(runner, "load_settings", lambda: settings)