Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions core/agent_runtime/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
UserSteer,
compose_injection_callbacks,
)
from core.agent_runtime.parallel import run_parallel_llm
from core.agent_runtime.runner import AgentRunner, AgentRunResult, AgentRunSpec
from core.agent_runtime.tools.base import Tool
from core.agent_runtime.tools.mcp import (
Expand Down Expand Up @@ -50,4 +51,5 @@
"UserSteer",
"compose_injection_callbacks",
"connect_mcp_servers",
"run_parallel_llm",
]
74 changes: 74 additions & 0 deletions core/agent_runtime/parallel.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
"""Public ``run_parallel_llm`` entry point (see ``core.agent_runtime.__init__``).

The fan-out / fan-in machinery lives in :class:`core.compat.parallel.ParallelLLM`
(a drop-in for the legacy ``mcp_agent`` helper). This module only exposes the
documented, ergonomic entry point so the name ``run_parallel_llm`` actually
resolves: build the :class:`~core.compat.agent.Agent` participants, hand them to
:class:`ParallelLLM`, and reduce the branch outputs through the fan-in agent.

Typical use — one prompt fanned out to N independent workers, then aggregated::

from core.agent_runtime.parallel import run_parallel_llm
from core.compat.agent import Agent

workers = [Agent(name=f"worker-{i}", instruction="...") for i in range(3)]
fan_in = Agent(name="aggregator", instruction="Synthesize the workers.")
result = await run_parallel_llm(
"Summarize the tradeoffs of X", fan_in_agent=fan_in, fan_out_agents=workers
)

Each worker branch runs concurrently (``asyncio.gather``); a branch that raises
is logged and folded in as ``[<name> error: ...]`` rather than aborting the run.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Callable, Iterable, Sequence, Type

if TYPE_CHECKING:
from core.compat.agent import Agent, AugmentedLLM
from core.compat.request_params import RequestParams

__all__ = ["run_parallel_llm"]


async def run_parallel_llm(
message: str,
*,
fan_in_agent: Agent,
fan_out_agents: Sequence[Agent] | Iterable[Agent],
llm_factory: Callable[[Agent], Any] | Type[AugmentedLLM] | None = None,
request_params: RequestParams | None = None,
) -> str:
"""Fan ``message`` out to ``fan_out_agents`` concurrently and aggregate via ``fan_in_agent``.

Thin wrapper over :meth:`core.compat.parallel.ParallelLLM.generate_str`. The
named keyword arguments mirror :class:`ParallelLLM`'s constructor so the
call site reads as "run this prompt in parallel"; the return value is the
fan-in agent's synthesized answer.

Args:
message: The prompt sent to every fan-out worker.
fan_in_agent: Agent that reduces the worker outputs into one answer.
fan_out_agents: The workers that each answer ``message`` concurrently.
llm_factory: Optional factory/marker used to attach an LLM to each
agent (see :meth:`ParallelLLM._attach`). ``None`` uses each agent's
default provider.
request_params: Optional request parameters forwarded to every branch
and to the fan-in call.

Returns:
The fan-in agent's aggregated text.
"""
# Imported lazily: core.compat.parallel transitively pulls core.compat.agent
# -> core.compat.runtime -> core.config, while core.config imports this
# package's core.agent_runtime.tools.mcp at module load. A module-level
# import here would close that cycle; deferring to call time keeps it open.
from core.compat.parallel import ParallelLLM

parallel = ParallelLLM(
fan_in_agent=fan_in_agent,
fan_out_agents=fan_out_agents,
llm_factory=llm_factory,
)
return await parallel.generate_str(message, request_params)
7 changes: 4 additions & 3 deletions core/compat/parallel.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,13 @@
from __future__ import annotations

import asyncio
from typing import Any, Callable, Iterable, Type
from typing import TYPE_CHECKING, Any, Callable, Iterable, Type

from loguru import logger

from core.compat.agent import Agent, AugmentedLLM
from core.compat.request_params import RequestParams
if TYPE_CHECKING:
from core.compat.agent import Agent, AugmentedLLM
from core.compat.request_params import RequestParams


class ParallelLLM:
Expand Down
104 changes: 104 additions & 0 deletions tests/test_import_layering.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
"""Regression guard: the core stack must import without order-dependent cycles.

Topology
--------
The core layering has three nodes that historically formed a 3-way import cycle::

core.config ──imports──▶ core.agent_runtime.tools.mcp
▲ │
│ ▼
core.compat.runtime ◀──core.compat.agent── core.compat (``__init__``)

``core/compat/parallel.py`` defers its ``core.compat.agent`` import under
``TYPE_CHECKING`` precisely because a module-level import closes the cycle
above and turns ``import core.config`` into
``ImportError: cannot import name 'DeepCodeConfig'`` — a failure that used to
surface only as a confusing *collection* error in ``test_config_layering.py``.

``core/agent_runtime/parallel.py`` likewise defers its ``core.compat.parallel``
import into ``run_parallel_llm`` (call time) so that re-exporting the entry
point from ``core/agent_runtime/__init__.py`` does not re-close the cycle.

Two tests pin the invariant, both cheap and fully local:

1. ``test_entrypoints_import_in_fresh_interpreter`` — symptom guard: each
entrypoint must import in a brand-new interpreter (circular imports are
order-sensitive, so an already-warmed session proves nothing).
2. ``test_agent_runtime_has_no_module_level_compat_import`` — cause guard:
nothing under ``core/agent_runtime/`` may import ``core.compat`` at module
level, even when the cycle is not currently closed (catches the landmine
before it detonates).
"""

from __future__ import annotations

import ast
import subprocess
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
RUNTIME_PACKAGE = ROOT / "core" / "agent_runtime"

ENTRYPOINTS = ("core.config", "core.compat", "core.agent_runtime")


@pytest.mark.parametrize("module", ENTRYPOINTS)
def test_entrypoints_import_in_fresh_interpreter(module: str) -> None:
result = subprocess.run(
[sys.executable, "-c", f"import {module}"],
cwd=ROOT,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=120,
)
assert result.returncode == 0, (
f"import {module} failed in a fresh interpreter (rc={result.returncode})."
" This is usually the core.config <-> core.agent_runtime <-> core.compat"
" circular import regressing: grep for a new module-level `core.compat`"
f" import under core/agent_runtime/.\nstderr: {result.stderr.strip()}"
)


def _is_deferred(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> bool:
"""True when the import runs at call time or never (function / TYPE_CHECKING)."""
cur = parents.get(node)
while cur is not None:
if isinstance(cur, (ast.FunctionDef, ast.AsyncFunctionDef)):
return True
if isinstance(cur, ast.If) and "TYPE_CHECKING" in ast.dump(cur.test):
return True
cur = parents.get(cur)
return False


def test_agent_runtime_has_no_module_level_compat_import() -> None:
offenders: list[str] = []
for path in sorted(RUNTIME_PACKAGE.rglob("*.py")):
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
parents = {
child: node
for node in ast.walk(tree)
for child in ast.iter_child_nodes(node)
}
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
refs = (node.module or "",)
elif isinstance(node, ast.Import):
refs = tuple(alias.name for alias in node.names)
else:
continue
mentions_compat = any(ref.startswith("core.compat") for ref in refs)
if mentions_compat and not _is_deferred(node, parents):
offenders.append(f"{path.relative_to(ROOT)}:{node.lineno}")
assert not offenders, (
"core/agent_runtime must not import core.compat at module level (the"
" new core stack must not depend on the legacy shim at import time;"
" that edge closes the core.config cycle). Move the import under"
" `if TYPE_CHECKING:` (annotations only) or into the calling function."
f" Offenders: {offenders}"
)
Loading