| import tempfile |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from arena.agent import AgentSession |
|
|
|
|
| class RecordingAgent: |
| """Records invocations and returns deterministic responses.""" |
|
|
| def __init__(self, response: str = "Built the app.") -> None: |
| self.invocations: list[dict] = [] |
| self._response = response |
|
|
| def invoke(self, payload: dict, config: dict) -> dict: |
| self.invocations.append({"payload": payload, "config": config}) |
| return {"messages": [{"role": "assistant", "content": self._response}]} |
|
|
|
|
| def make_local_session(thread_id: str | None = None) -> AgentSession: |
| """Create a real LocalShellBackend session for behavior testing.""" |
| from deepagents.backends import LocalShellBackend, CompositeBackend, FilesystemBackend |
|
|
| root = Path(tempfile.mkdtemp(prefix="agent-swarm-test-")) |
| (root / "workspace").mkdir(parents=True, exist_ok=True) |
| backend = LocalShellBackend(root_dir=root, virtual_mode=True, inherit_env=False) |
| backend._workspace_dir = "/workspace" |
| agent = RecordingAgent() |
|
|
| ctx_root = Path(tempfile.mkdtemp(prefix="agent-swarm-ctx-")) |
| (ctx_root / "context").mkdir(parents=True, exist_ok=True) |
| context_backend = FilesystemBackend(root_dir=ctx_root, virtual_mode=True) |
| composite = CompositeBackend( |
| default=backend, |
| routes={"/context/": context_backend}, |
| ) |
| composite._workspace_dir = "/workspace" |
| backend.execute("mkdir -p /workspace") |
| return AgentSession( |
| thread_id=thread_id or "session-1", |
| provider="local", |
| backend=composite, |
| agent=agent, |
| workspace_dir="/workspace", |
| ) |
|
|