test(fixtures): add self-contained openai-default test system YAML

tests/engine/test_engine_flow.py

@@ -14,15 +14,13 @@
 
 from __future__ import annotations
 
-from collections.abc import AsyncIterator, Callable
+from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, cast
+from typing import Any
 
 import pytest
 from langchain_core.language_models import BaseChatModel
-from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
-from langchain_core.messages.tool import ToolCall
 
 from agent_engine.core.spec import (
     AgentSpec,
@@ -40,60 +38,16 @@
 from agent_engine.engine.types import RunResult
 from agent_engine.runtime.hooks import AuthContext, RunContext
 from agent_engine.runtime.state import GraphState
+from tests.fixtures.utils import fake_model_factory
 
 # ---------------------------------------------------------------------------
 # Fake chat model
 # ---------------------------------------------------------------------------
 
 
-class FakeChatModel:
-    """A scriptless stand-in: route through one tool, then answer."""
-
-    def __init__(self, answer: str = "ok", tool_names: list[str] | None = None) -> None:
-        self._answer = answer
-        self._tool_names = tool_names or []
-
-    def bind_tools(self, tools: list[Any]) -> FakeChatModel:
-        return FakeChatModel(self._answer, [t.name for t in tools])
-
-    async def ainvoke(self, messages: list[Any]) -> AIMessage:
-        return self._respond(messages)
-
-    async def astream(self, messages: list[Any]) -> AsyncIterator[AIMessage]:
-        yield self._respond(messages)
-
-    def _respond(self, messages: list[Any]) -> AIMessage:
-        already_called = any(isinstance(m, ToolMessage) for m in messages)
-        if self._tool_names and not already_called:
-            call = ToolCall(
-                name=self._select(messages),
-                args={"message": self._user_text(messages)},
-                id="call_1",
-            )
-            return AIMessage(content="", tool_calls=[call])
-        return AIMessage(content=self._answer)
-
-    def _select(self, messages: list[Any]) -> str:
-        text = self._user_text(messages).lower()
-        for name in self._tool_names:
-            if name.lower() in text:
-                return name
-        return self._tool_names[0]
-
-    @staticmethod
-    def _user_text(messages: list[Any]) -> str:
-        for m in reversed(messages):
-            if isinstance(m, HumanMessage):
-                return str(m.content)
-        return ""
-
-
 @pytest.fixture
 def model_factory() -> Callable[[str, str, float | None], BaseChatModel]:
-    def factory(provider: str, name: str, temperature: float | None) -> BaseChatModel:
-        return cast(BaseChatModel, FakeChatModel())
-
-    return factory
+    return fake_model_factory
 
 
 # ---------------------------------------------------------------------------

tests/fixtures/plugins/plugins.toml

@@ -0,0 +1,12 @@
+
+[tools.echo_tool]
+module = "tools.echo_tool"
+callable = "echo_tool"
+
+[resolvers.shared]
+module = "resolvers.shared"
+class = "SharedResolver"
+
+[resolvers.greeting_agent]
+module = "resolvers.greeting_agent"
+class = "Resolver"

tests/fixtures/plugins/resolvers/greeting_agent.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from shared import SharedResolver
+
+
+class Resolver(SharedResolver):
+    def __init__(self) -> None:
+        super().__init__()

tests/fixtures/plugins/resolvers/shared.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+
+class SharedResolver:
+    def __init__(self) -> None:
+        pass
+
+    def current_date(self, ctx: dict) -> str:
+        """Returns the value for {{current_date}}"""
+        return "2026-07-19"
+
+    def user_name(self, ctx: dict) -> str:
+        """Returns the value for {{user_name}}"""
+        return "Tester"

tests/fixtures/plugins/tools/echo_tool.py

@@ -0,0 +1,5 @@
+def echo_tool(input: dict) -> str:
+    """Echo back the input text. Used to verify tool wiring end to end.
+"""
+    text = input.get("text", "")
+    return f"echo: {text}"

tests/fixtures/prompts/echo_agent/system.md

@@ -0,0 +1,3 @@
+# Echo Agent — system
+
+You echo the user's input back to them using the `echo_tool` tool.

tests/fixtures/prompts/greeting_agent/system.md

@@ -0,0 +1,4 @@
+# Greeting Agent — system
+
+You handle greetings and small talk. Be friendly, brief, and use the
+`current_date` and `user_name` context when natural.

tests/fixtures/prompts/root_router/orchestrator.md

@@ -0,0 +1,7 @@
+# Root Router — orchestrator
+
+You are the entry point of the test system. Read the user's request and route it
+to the most appropriate child specialist:
+
+- `greeting_agent` for greetings and small talk.
+- `echo_agent` for requests that ask to repeat or echo text.

tests/fixtures/prompts/root_router/system.md

@@ -0,0 +1,4 @@
+# Root Router — system
+
+You route each incoming request to a single child specialist. Do not answer
+directly; delegate to `greeting_agent` or `echo_agent`.

tests/fixtures/test_infra.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import sys
+
+import pytest
+from tests.fixtures.utils import FakeEngine, load_test_system
+
+
+@pytest.fixture(autouse=True)
+def _cleanup_shared() -> None:
+    yield
+    sys.modules.pop("shared", None)
+
+
+def test_spec_validates_offline() -> None:
+    load_test_system()
+
+
+@pytest.mark.asyncio
+async def test_engine_builds_from_fixture() -> None:
+    async with FakeEngine():
+        pass
+
+
+@pytest.mark.asyncio
+async def test_orchestrator_routes_to_greeting() -> None:
+    async with FakeEngine() as engine:
+        result = await engine.run("greeting_agent hello")
+        assert "root_router/greeting_agent" in result.visited
+        assert result.answer
+
+
+@pytest.mark.asyncio
+async def test_orchestrator_routes_to_echo_and_calls_tool() -> None:
+    async with FakeEngine() as engine:
+        result = await engine.run("echo_agent echo test")
+        assert "root_router/echo_agent" in result.visited
+        assert any(record.name == "echo_tool" for record in result.used_tools)
+        assert result.answer
+
+
+@pytest.mark.asyncio
+async def test_resolvers_resolve_at_runtime() -> None:
+    async with FakeEngine() as engine:
+        result = await engine.run("hello")
+        assert result.status == "completed"

tests/fixtures/test_system.yaml

@@ -0,0 +1,64 @@
+system:
+  name: "Test Agent System"
+
+
+defaults:
+  model:
+    provider: openai
+    name: gpt-4o-mini
+    temperature: 0.0
+
+execution:
+  max_iterations: 10
+  max_tool_calls: 5
+  max_tool_calls_per_agent: 3
+  max_child_agent_calls: 4
+  allow_duplicate_tool_calls: false
+
+tools:
+  echo_tool:
+    description: >
+      Echo back the input text. Used to verify tool wiring end to end.
+
+resolvers:
+  current_date:
+  user_name:
+
+orchestrators:
+  root_router:
+    name: "Root Router"
+    description: >
+      Entry point. Routes the user request to the appropriate specialist.
+    model:
+      provider: openai
+      name: gpt-4o
+      temperature: 0.0
+    prompts:
+      orchestrator: prompts/root_router/orchestrator.md
+      system: prompts/root_router/system.md
+
+agents:
+  greeting_agent:
+    name: "greeting"
+    description: >
+      Handles simple greetings and small talk.
+    prompts:
+      system: prompts/greeting_agent/system.md
+    resolvers:
+      - current_date
+      - user_name
+
+  echo_agent:
+    name: "echo"
+    description: >
+      Demonstrates tool usage by echoing user input.
+    prompts:
+      system: prompts/echo_agent/system.md
+    tools:
+      - echo_tool
+    auto: true
+
+graph:
+  root_router:
+    greeting_agent:
+    echo_agent:

tests/fixtures/utils.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from langchain_core.messages import AIMessage
+from langchain_core.messages.tool import ToolCall
+
+from agent_engine.core.spec import SystemSpec
+from agent_engine.engine.langgraph.engine import LangGraphEngine
+from agent_engine.engine.types import RunResult
+
+
+class FakeChatModel:
+    def __init__(
+        self,
+        answer: str = "ok",
+        tool_names: list[str] | None = None,
+    ) -> None:
+        self._answer = answer
+        self._tool_names = tool_names or []
+
+    def bind_tools(self, tools: list[Any]) -> FakeChatModel:
+        return FakeChatModel(
+            self._answer,
+            [t.name for t in tools],
+        )
+
+    async def ainvoke(self, messages: list[Any]) -> AIMessage:
+        return self._respond(messages)
+
+    async def astream(self, messages: list[Any]) -> Any:
+        yield self._respond(messages)
+
+    def _respond(self, messages: list[Any]) -> AIMessage:
+        already_called = any(
+            m.__class__.__name__ == "ToolMessage" for m in messages
+        )
+        if self._tool_names and not already_called:
+            return AIMessage(
+                content="",
+                tool_calls=[
+                    ToolCall(
+                        name=self._select(messages),
+                        args={"message": self._user_text(messages)},
+                        id="call_1",
+                    )
+                ],
+            )
+        return AIMessage(content=self._answer)
+
+    def _select(self, messages: list[Any]) -> str:
+        text = self._user_text(messages).lower()
+        for name in self._tool_names:
+            if name.lower() in text:
+                return name
+        return self._tool_names[0]
+
+    @staticmethod
+    def _user_text(messages: list[Any]) -> str:
+        for m in reversed(messages):
+            if m.__class__.__name__ == "HumanMessage":
+                return str(m.content)
+        return ""
+
+
+def fake_model_factory(
+    provider: str = "fake",
+    name: str = "fake",
+    temperature: float | None = None,
+    *,
+    answer: str = "ok",
+) -> Any:
+    return FakeChatModel(answer=answer)
+
+
+FIXTURE_DIR = Path(__file__).resolve().parent
+FIXTURE_SPEC = FIXTURE_DIR / "test_system.yaml"
+
+
+def load_test_system() -> tuple[SystemSpec, Path]:
+    from agent_engine.core.validator import SystemSpecValidator
+    from agent_engine.parsers.yaml.parser import YAMLParser
+
+    spec = YAMLParser().parse(str(FIXTURE_SPEC))
+    errors = SystemSpecValidator().validate(spec, FIXTURE_DIR)
+    if errors:
+        raise AssertionError(
+            "Fixture validation failed:\n" + "\n".join(str(e) for e in errors)
+        )
+    return spec, FIXTURE_DIR
+
+
+class FakeEngine:
+    def __init__(self, answer: str = "ok") -> None:
+        self._answer = answer
+        self._engine: LangGraphEngine | None = None
+
+    async def __aenter__(self) -> FakeEngine:
+        spec, base_dir = load_test_system()
+        self._engine = LangGraphEngine(base_dir, model_factory=fake_model_factory)
+        await self._engine.build(spec)
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        if self._engine is not None:
+            await self._engine.__aexit__(*args)
+
+    async def run(self, message: str) -> RunResult:
+        assert self._engine is not None
+        return await self._engine.run(message)