File size: 1,644 Bytes
31c8075 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | 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",
)
|