mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-10-07 16:19:09 +00:00
9d3600ad8f
Adapted from upstream langchain-ai/open-swe #1684 to the fork's direct-import middleware registry and middleware stack ordering. Adds SubdirAgentsReadMiddleware, which appends applicable ancestor AGENTS.md instructions to read_file results once per run, so scoped rules are visible before edits. Wired into get_agent immediately after ToolErrorMiddleware, matching upstream's relative position. Note: this changes file-read behavior for every main-agent run. The reviewer graph uses its own leaner middleware stack and is unaffected. (cherry picked from commit 7f7af71547be2199cea699284676f8ceefba7691)
113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
"""Assembly contract for the main agent's context-management + middleware wiring.
|
|
|
|
Locks in that `get_agent` hands a sandbox `backend` to `create_deep_agent` (which
|
|
is what makes deepagents auto-wire `FilesystemMiddleware` tool-result eviction and
|
|
`SummarizationMiddleware` history offloading), and that the redundant custom
|
|
`RepairOrphanedToolCallsMiddleware` is no longer added explicitly — the built-in
|
|
`PatchToolCallsMiddleware` that `create_deep_agent` adds covers it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from langgraph.graph.state import RunnableConfig
|
|
|
|
from agent.server import get_agent
|
|
|
|
|
|
class _DummyAgent:
|
|
def with_config(self, config: RunnableConfig) -> _DummyAgent:
|
|
self.config = config
|
|
return self
|
|
|
|
|
|
def _base_config() -> RunnableConfig:
|
|
return {
|
|
"configurable": {
|
|
"__is_for_execution__": True,
|
|
"thread_id": "thread-ctx",
|
|
"github_login": "octocat",
|
|
},
|
|
"metadata": {},
|
|
}
|
|
|
|
|
|
async def _capture_create_deep_agent_kwargs() -> dict[str, object]:
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_create_deep_agent(**kwargs: object) -> _DummyAgent:
|
|
captured.update(kwargs)
|
|
return _DummyAgent()
|
|
|
|
with (
|
|
patch(
|
|
"agent.server.resolve_github_token",
|
|
new_callable=AsyncMock,
|
|
return_value=("ghp", None),
|
|
),
|
|
patch("agent.server.resolve_triggering_user_identity", return_value=None),
|
|
patch(
|
|
"agent.server.ensure_sandbox_for_thread",
|
|
new_callable=AsyncMock,
|
|
return_value=MagicMock(),
|
|
),
|
|
patch(
|
|
"agent.server.aresolve_sandbox_work_dir",
|
|
new_callable=AsyncMock,
|
|
return_value="/workspace",
|
|
),
|
|
patch(
|
|
"agent.server.get_team_default_model_pair",
|
|
new_callable=AsyncMock,
|
|
return_value=(("openai:gpt-5.5", "medium"), ("openai:gpt-5.5", "low")),
|
|
),
|
|
patch("agent.server.load_profile", new_callable=AsyncMock, return_value=None),
|
|
patch("agent.server.fallback_model_id_for", return_value=None),
|
|
patch("agent.server.make_model", side_effect=[MagicMock(), MagicMock()]),
|
|
patch("agent.server.construct_system_prompt", return_value="prompt"),
|
|
patch("agent.server.create_deep_agent", side_effect=fake_create_deep_agent),
|
|
):
|
|
await get_agent(_base_config())
|
|
|
|
return captured
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_is_built_with_a_backend_for_eviction_and_summarization() -> None:
|
|
captured = await _capture_create_deep_agent_kwargs()
|
|
# The backend is what enables deepagents' auto-wired FilesystemMiddleware
|
|
# eviction + SummarizationMiddleware offloading.
|
|
assert callable(captured["backend"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_does_not_add_custom_repair_middleware() -> None:
|
|
captured = await _capture_create_deep_agent_kwargs()
|
|
middleware = captured["middleware"]
|
|
assert isinstance(middleware, list)
|
|
names = {type(m).__name__ for m in middleware}
|
|
# Built-in PatchToolCallsMiddleware (added by create_deep_agent) replaces it.
|
|
assert "RepairOrphanedToolCallsMiddleware" not in names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_wires_subdir_agents_middleware_after_tool_error() -> None:
|
|
captured = await _capture_create_deep_agent_kwargs()
|
|
middleware = captured["middleware"]
|
|
assert isinstance(middleware, list)
|
|
names = [type(m).__name__ for m in middleware]
|
|
assert "SubdirAgentsReadMiddleware" in names
|
|
assert names.index("ToolErrorMiddleware") < names.index("SubdirAgentsReadMiddleware")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_keeps_message_queue_and_step_limit_middleware() -> None:
|
|
captured = await _capture_create_deep_agent_kwargs()
|
|
middleware = captured["middleware"]
|
|
# The dashboard depends on check_message_queue_before_model; the step-limit
|
|
# notifier must still fire when the lowered run budget is hit.
|
|
present = {type(m).__name__ for m in middleware}
|
|
assert "check_message_queue_before_model" in present
|
|
assert "notify_step_limit_reached" in present
|