wehub-resource-sync
793 lines
30 KiB
Python
793 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yuxi.agents.middlewares.subagent_task as subagent_task_middleware
|
|
import yuxi.services.agent_run_service as agent_run_service
|
|
import yuxi.services.run_queue_service as run_queue_service
|
|
import yuxi.services.subagent_run_service as subagent_run_service
|
|
from langchain.agents._subagent_transformer import SubagentTransformer as LangChainSubagentTransformer
|
|
from langgraph.prebuilt.tool_node import ToolRuntime
|
|
from langgraph.stream._mux import StreamMux
|
|
from langgraph.types import Command
|
|
from yuxi.agents.buildin.chatbot.state import merge_subagent_runs
|
|
from yuxi.agents.middlewares.subagent_task import YUXI_SUBAGENTS_STREAM_KEY, YuxiSubAgentMiddleware
|
|
from yuxi.repositories.agent_repository import SUB_AGENT_BACKEND_ID
|
|
from yuxi.services.input_message_service import AgentRunInputMessage
|
|
from yuxi.utils.hash_utils import subagent_child_thread_id
|
|
|
|
|
|
def make_child_thread_id(parent_thread_id: str, agent_slug: str, tool_call_id: str) -> str:
|
|
return subagent_child_thread_id(parent_thread_id, agent_slug, tool_call_id)
|
|
|
|
|
|
class _ChildContext:
|
|
def __init__(self):
|
|
self.model = None
|
|
|
|
def update_from_dict(self, values: dict):
|
|
for key, value in values.items():
|
|
if hasattr(self, key):
|
|
setattr(self, key, value)
|
|
|
|
|
|
class _SessionContext:
|
|
async def __aenter__(self):
|
|
return object()
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return None
|
|
|
|
|
|
def _patch_session(monkeypatch):
|
|
monkeypatch.setattr(
|
|
subagent_task_middleware,
|
|
"pg_manager",
|
|
SimpleNamespace(get_async_session_context=lambda: _SessionContext()),
|
|
)
|
|
|
|
|
|
def _patch_subagent_run_service(monkeypatch, service_class) -> None:
|
|
monkeypatch.setattr(
|
|
subagent_task_middleware,
|
|
"_subagent_run_service_module",
|
|
lambda: SimpleNamespace(
|
|
SubagentRunService=service_class,
|
|
SubagentRunBusy=subagent_run_service.SubagentRunBusy,
|
|
serialize_subagent_run_state=subagent_run_service.serialize_subagent_run_state,
|
|
subagent_run_urls=subagent_run_service.subagent_run_urls,
|
|
),
|
|
)
|
|
|
|
|
|
def _async_tool_middleware(*, model: str | None = None) -> YuxiSubAgentMiddleware:
|
|
parent_context = SimpleNamespace(thread_id="parent-thread", uid="user-1", run_id="parent-run")
|
|
if model:
|
|
parent_context.model = model
|
|
return YuxiSubAgentMiddleware(
|
|
parent_context=parent_context,
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def _subagent_run(
|
|
*,
|
|
status: str = "running",
|
|
thread_id: str = "child-thread",
|
|
tool_call_id: str = "tool-async",
|
|
subagent_slug: str = "worker",
|
|
subagent_name: str = "Worker",
|
|
description: str = "run in background",
|
|
error_message: str | None = None,
|
|
):
|
|
return SimpleNamespace(
|
|
id="child-run",
|
|
conversation_thread_id=thread_id,
|
|
agent_slug=subagent_slug,
|
|
status=status,
|
|
created_by_run_id="parent-run",
|
|
subagent_thread_relation_id=77,
|
|
input_payload={
|
|
"runtime": {
|
|
"tool_call_id": tool_call_id,
|
|
"subagent_name": subagent_name,
|
|
"description": description,
|
|
},
|
|
},
|
|
created_at=None,
|
|
finished_at=None,
|
|
error_message=error_message,
|
|
)
|
|
|
|
|
|
def _patch_task_start_and_await(
|
|
monkeypatch,
|
|
captured: dict,
|
|
*,
|
|
status: str = "completed",
|
|
output: str = "child done",
|
|
thread_id: str = "child-thread",
|
|
error_message: str | None = None,
|
|
wait_timeout: bool = False,
|
|
):
|
|
class _SubagentRunService:
|
|
def __init__(self, db):
|
|
captured["db"] = db
|
|
|
|
async def start(self, **kwargs):
|
|
captured["start"] = kwargs
|
|
return SimpleNamespace(
|
|
run=_subagent_run(
|
|
status="pending",
|
|
thread_id=thread_id,
|
|
tool_call_id=kwargs["tool_call_id"],
|
|
subagent_slug=kwargs["agent_item"].slug,
|
|
subagent_name=kwargs["agent_item"].name,
|
|
description=kwargs["input_message"].content,
|
|
),
|
|
created=True,
|
|
continuing=bool(kwargs.get("requested_thread_id")),
|
|
relation=SimpleNamespace(id=77, child_thread_id=thread_id),
|
|
)
|
|
|
|
async def get_run_for_creator(self, **kwargs):
|
|
captured["get_run_for_creator"] = kwargs
|
|
started = captured["start"]
|
|
return _subagent_run(
|
|
status=status,
|
|
thread_id=thread_id,
|
|
tool_call_id=started["tool_call_id"],
|
|
subagent_slug=started["agent_item"].slug,
|
|
subagent_name=started["agent_item"].name,
|
|
description=started["input_message"].content,
|
|
error_message=error_message,
|
|
)
|
|
|
|
async def fake_await_agent_run_result(*, run_id: str, current_uid: str):
|
|
captured["await"] = {"run_id": run_id, "current_uid": current_uid}
|
|
result = {
|
|
"status": status,
|
|
"output": output,
|
|
"agent_run_id": run_id,
|
|
"thread_id": thread_id,
|
|
}
|
|
if error_message:
|
|
result["error"] = {"type": "RuntimeError", "message": error_message}
|
|
if wait_timeout:
|
|
raise agent_run_service.AgentRunWaitTimeout(result)
|
|
return result
|
|
|
|
_patch_session(monkeypatch)
|
|
_patch_subagent_run_service(monkeypatch, _SubagentRunService)
|
|
monkeypatch.setattr(agent_run_service, "await_agent_run_result", fake_await_agent_run_result)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_middleware_loads_all_visible_subagents_when_empty(monkeypatch) -> None:
|
|
class _UserRepository:
|
|
async def get_by_uid_with_db(self, _db, uid):
|
|
assert uid == "user-1"
|
|
return SimpleNamespace(uid="user-1", role="user")
|
|
|
|
class _AgentRepository:
|
|
def __init__(self, _db):
|
|
pass
|
|
|
|
async def list_visible_subagents(self, *, user):
|
|
assert user.uid == "user-1"
|
|
return [
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
),
|
|
]
|
|
|
|
async def get_visible_by_slug(self, *, slug, user, kind="main"):
|
|
del slug
|
|
del user
|
|
del kind
|
|
raise AssertionError("empty subagents should load all visible subagents")
|
|
|
|
_patch_session(monkeypatch)
|
|
monkeypatch.setattr(subagent_task_middleware, "UserRepository", _UserRepository)
|
|
monkeypatch.setattr(subagent_task_middleware, "AgentRepository", _AgentRepository)
|
|
|
|
middleware = await subagent_task_middleware.create_subagent_task_middleware(
|
|
SimpleNamespace(thread_id="parent-thread", uid="user-1", subagents=[]),
|
|
)
|
|
|
|
assert isinstance(middleware, YuxiSubAgentMiddleware)
|
|
assert middleware.subagent_names == frozenset({"worker"})
|
|
assert {tool.name for tool in middleware.tools} >= {"task", "subagent_start"}
|
|
|
|
|
|
def test_yuxi_subagent_transformer_does_not_conflict_with_langchain_default() -> None:
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(thread_id="parent-thread", uid="user-1"),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
)
|
|
],
|
|
)
|
|
|
|
mux = StreamMux(
|
|
factories=[LangChainSubagentTransformer, *middleware.transformers],
|
|
is_async=True,
|
|
)
|
|
|
|
assert "subagents" in mux.extensions
|
|
assert YUXI_SUBAGENTS_STREAM_KEY in mux.extensions
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_tool_rejects_unconfigured_subagent() -> None:
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(thread_id="parent-thread", uid="user-1", model=""),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
)
|
|
],
|
|
)
|
|
runtime = ToolRuntime(
|
|
state={},
|
|
context=None,
|
|
tool_call_id="tool-1",
|
|
store=None,
|
|
stream_writer=lambda _: None,
|
|
config={},
|
|
)
|
|
|
|
result = await middleware.tools[0].ainvoke(
|
|
{"description": "do work", "subagent_slug": "missing", "runtime": runtime}
|
|
)
|
|
|
|
assert result == "无法调用子智能体 missing，可用子智能体只有：`worker`"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_tool_invokes_subagent_with_child_scope(monkeypatch) -> None:
|
|
captured = {}
|
|
_patch_task_start_and_await(monkeypatch, captured, thread_id="child-thread")
|
|
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(
|
|
thread_id="child-runtime-thread",
|
|
parent_thread_id="parent-thread",
|
|
file_thread_id="parent-file-thread",
|
|
uid="user-1",
|
|
run_id="parent-run",
|
|
),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker.agent",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={"context": {"model": "provider:model", "subagents": ["nested"]}},
|
|
)
|
|
],
|
|
)
|
|
runtime = SimpleNamespace(tool_call_id="tool-1", state={}, config={})
|
|
|
|
result = await middleware.tools[0].coroutine(
|
|
description="write a report",
|
|
subagent_slug="worker.agent",
|
|
runtime=runtime,
|
|
)
|
|
|
|
assert isinstance(result, Command)
|
|
assert result.update["messages"][0].content == "> 子智能体线程 ID: child-thread\n\n---\n\nchild done"
|
|
assert result.update["messages"][0].tool_call_id == "tool-1"
|
|
assert captured["start"]["uid"] == "user-1"
|
|
assert captured["start"]["created_by_run_id"] == "parent-run"
|
|
assert captured["start"]["requested_thread_id"] is None
|
|
assert captured["start"]["file_thread_id"] == "parent-file-thread"
|
|
assert captured["start"]["model_spec"] is None
|
|
assert captured["await"] == {"run_id": "child-run", "current_uid": "user-1"}
|
|
assert result.update["subagent_runs"][0]["run_id"] == "child-run"
|
|
assert result.update["subagent_runs"][0]["child_thread_id"] == "child-thread"
|
|
assert result.update["subagent_runs"][0]["status"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_tool_inherits_parent_model_when_subagent_model_empty(monkeypatch) -> None:
|
|
captured = {}
|
|
_patch_task_start_and_await(monkeypatch, captured)
|
|
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(
|
|
thread_id="parent-thread",
|
|
uid="user-1",
|
|
run_id="parent-run",
|
|
model="parent:model",
|
|
),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={"context": {"model": ""}},
|
|
)
|
|
],
|
|
)
|
|
|
|
await middleware.tools[0].coroutine(
|
|
description="write a report",
|
|
subagent_slug="worker",
|
|
runtime=SimpleNamespace(tool_call_id="tool-1", state={}, config={}),
|
|
)
|
|
|
|
assert captured["start"]["model_spec"] == "parent:model"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_tool_records_failed_subagent_run(monkeypatch) -> None:
|
|
captured = {}
|
|
_patch_task_start_and_await(monkeypatch, captured, status="failed", output="", error_message="child boom")
|
|
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(thread_id="parent-thread", uid="user-1", run_id="parent-run", model=""),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
)
|
|
],
|
|
)
|
|
|
|
result = await middleware.tools[0].coroutine(
|
|
description="write a report",
|
|
subagent_slug="worker",
|
|
runtime=SimpleNamespace(tool_call_id="tool-1", state={}, config={}),
|
|
)
|
|
|
|
assert isinstance(result, Command)
|
|
assert result.update["messages"][0].content == "> 子智能体线程 ID: child-thread\n\n---\n\nchild boom"
|
|
assert result.update["subagent_runs"][0]["status"] == "failed"
|
|
assert result.update["subagent_runs"][0]["error"] == "child boom"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_tool_reports_running_subagent_after_wait_timeout(monkeypatch) -> None:
|
|
captured = {}
|
|
_patch_task_start_and_await(monkeypatch, captured, status="running", output="", wait_timeout=True)
|
|
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(thread_id="parent-thread", uid="user-1", run_id="parent-run", model=""),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
)
|
|
],
|
|
)
|
|
|
|
result = await middleware.tools[0].coroutine(
|
|
description="write a long report",
|
|
subagent_slug="worker",
|
|
runtime=SimpleNamespace(tool_call_id="tool-1", state={}, config={}),
|
|
)
|
|
|
|
assert isinstance(result, Command)
|
|
content = result.update["messages"][0].content
|
|
assert "子智能体仍在运行" in content
|
|
assert "run_id: child-run" in content
|
|
assert "不要把当前结果视为任务已完成" in content
|
|
assert "子智能体已完成任务" not in content
|
|
assert result.update["subagent_runs"][0]["status"] == "running"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_tool_continues_existing_subagent_thread(monkeypatch) -> None:
|
|
captured = {}
|
|
_patch_task_start_and_await(monkeypatch, captured, output="continued done", thread_id="child-thread")
|
|
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(thread_id="parent-thread", uid="user-1", run_id="parent-run", model=""),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker.agent",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
)
|
|
],
|
|
)
|
|
|
|
result = await middleware.tools[0].coroutine(
|
|
description="continue the report",
|
|
subagent_slug="worker.agent",
|
|
runtime=SimpleNamespace(tool_call_id="tool-2", state={}, config={}),
|
|
thread_id="child-thread",
|
|
)
|
|
|
|
assert isinstance(result, Command)
|
|
assert result.update["messages"][0].content == "> 子智能体线程 ID: child-thread\n\n---\n\ncontinued done"
|
|
assert captured["start"]["requested_thread_id"] == "child-thread"
|
|
assert result.update["subagent_runs"][0]["child_thread_id"] == "child-thread"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_tool_rejects_invalid_continuation_thread(monkeypatch) -> None:
|
|
class _SubagentRunService:
|
|
def __init__(self, db):
|
|
del db
|
|
|
|
async def start(self, **kwargs):
|
|
raise ValueError(
|
|
f"无法继续子智能体线程 {kwargs['requested_thread_id']}：当前对话中没有找到对应的运行记录"
|
|
)
|
|
|
|
_patch_session(monkeypatch)
|
|
_patch_subagent_run_service(monkeypatch, _SubagentRunService)
|
|
middleware = YuxiSubAgentMiddleware(
|
|
parent_context=SimpleNamespace(thread_id="parent-thread", uid="user-1", run_id="parent-run"),
|
|
subagents=[
|
|
SimpleNamespace(
|
|
slug="worker",
|
|
name="Worker",
|
|
description="work on scoped tasks",
|
|
backend_id=SUB_AGENT_BACKEND_ID,
|
|
config_json={},
|
|
)
|
|
],
|
|
)
|
|
|
|
unknown_thread_id = "opaque-child-thread"
|
|
runtime = SimpleNamespace(tool_call_id="tool-2", state={}, config={})
|
|
result = await middleware.tools[0].coroutine(
|
|
description="continue",
|
|
subagent_slug="worker",
|
|
runtime=runtime,
|
|
thread_id=unknown_thread_id,
|
|
)
|
|
|
|
assert result == f"无法继续子智能体线程 {unknown_thread_id}：当前对话中没有找到对应的运行记录"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_start_creates_child_run_and_enqueues(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class _SubagentRunService:
|
|
def __init__(self, db):
|
|
captured["db"] = db
|
|
|
|
async def start(self, **kwargs):
|
|
captured["start"] = kwargs
|
|
child_thread_id = make_child_thread_id("parent-thread", "worker", "tool-async")
|
|
return SimpleNamespace(
|
|
run=_subagent_run(status="pending", thread_id=child_thread_id),
|
|
created=True,
|
|
continuing=False,
|
|
relation=SimpleNamespace(id=77, child_thread_id=child_thread_id),
|
|
)
|
|
|
|
_patch_session(monkeypatch)
|
|
_patch_subagent_run_service(monkeypatch, _SubagentRunService)
|
|
|
|
middleware = _async_tool_middleware(model="provider:parent-model")
|
|
runtime = SimpleNamespace(tool_call_id="tool-async", state={}, config={})
|
|
tool = next(item for item in middleware.tools if item.name == "subagent_start")
|
|
|
|
result = await tool.coroutine(
|
|
description="run in background",
|
|
subagent_slug="worker",
|
|
runtime=runtime,
|
|
)
|
|
|
|
child_thread_id = make_child_thread_id("parent-thread", "worker", "tool-async")
|
|
assert isinstance(result, Command)
|
|
payload = json.loads(result.update["messages"][0].content)
|
|
assert payload["status"] == "started"
|
|
assert payload["run_id"] == "child-run"
|
|
assert payload["thread_id"] == child_thread_id
|
|
assert payload["events_url"] == "/api/agent/runs/child-run/events"
|
|
assert payload["subagent_thread_relation_id"] == 77
|
|
assert captured["start"]["uid"] == "user-1"
|
|
assert captured["start"]["created_by_run_id"] == "parent-run"
|
|
assert isinstance(captured["start"]["input_message"], AgentRunInputMessage)
|
|
assert captured["start"]["input_message"].content == "run in background"
|
|
assert captured["start"]["input_message"].raw_message()["type"] == "human"
|
|
assert captured["start"]["input_message"].raw_message()["content"] == "run in background"
|
|
assert "description" not in captured["start"]
|
|
assert captured["start"]["requested_thread_id"] is None
|
|
assert captured["start"]["model_spec"] == "provider:parent-model"
|
|
assert result.update["subagent_runs"][0]["child_thread_id"] == child_thread_id
|
|
assert result.update["subagent_runs"][0]["run_id"] == "child-run"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_status_returns_terminal_result(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class _SubagentRunService:
|
|
def __init__(self, db):
|
|
captured["db"] = db
|
|
|
|
async def get_run_for_creator(self, **kwargs):
|
|
captured["get_run_for_creator"] = kwargs
|
|
return _subagent_run(status="completed")
|
|
|
|
async def fake_get_agent_run_result(*, run_id: str, current_uid: str, db):
|
|
captured["get_agent_run_result"] = {"run_id": run_id, "current_uid": current_uid, "db": db}
|
|
return {"status": "completed", "output": "final result"}
|
|
|
|
async def fake_get_agent_run_progress(run_id: str):
|
|
captured["get_agent_run_progress"] = run_id
|
|
return {
|
|
"last_seq": "3-0",
|
|
"messages": [{"seq": "3-0", "kind": "assistant_message", "message_id": "msg-1", "content": "working"}],
|
|
}
|
|
|
|
_patch_session(monkeypatch)
|
|
_patch_subagent_run_service(monkeypatch, _SubagentRunService)
|
|
monkeypatch.setattr(agent_run_service, "get_agent_run_result", fake_get_agent_run_result)
|
|
monkeypatch.setattr(agent_run_service, "get_agent_run_progress", fake_get_agent_run_progress)
|
|
|
|
tool = next(item for item in _async_tool_middleware().tools if item.name == "subagent_status")
|
|
result = await tool.coroutine(run_id="child-run", runtime=SimpleNamespace(tool_call_id="status-call"))
|
|
|
|
assert isinstance(result, Command)
|
|
payload = json.loads(result.update["messages"][0].content)
|
|
assert payload["status"] == "completed"
|
|
assert payload["progress"]["messages"][0]["content"] == "working"
|
|
assert payload["result"]["output"] == "final result"
|
|
assert captured["get_agent_run_progress"] == "child-run"
|
|
assert captured["get_run_for_creator"] == {
|
|
"uid": "user-1",
|
|
"created_by_run_id": "parent-run",
|
|
"run_id": "child-run",
|
|
}
|
|
assert captured["get_agent_run_result"]["current_uid"] == "user-1"
|
|
assert result.update["subagent_runs"] == [
|
|
{
|
|
"id": "tool-async",
|
|
"run_id": "child-run",
|
|
"subagent_slug": "worker",
|
|
"subagent_name": "Worker",
|
|
"child_thread_id": "child-thread",
|
|
"status": "completed",
|
|
"events_url": "/api/agent/runs/child-run/events",
|
|
"result_url": "/api/agent/runs/child-run/result",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_status_returns_progress_for_running_run(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class _SubagentRunService:
|
|
def __init__(self, db):
|
|
captured["db"] = db
|
|
|
|
async def get_run_for_creator(self, **kwargs):
|
|
captured["get_run_for_creator"] = kwargs
|
|
return _subagent_run(status="running")
|
|
|
|
async def fake_get_agent_run_result(**kwargs):
|
|
raise AssertionError(f"running status should not load terminal result: {kwargs}")
|
|
|
|
async def fake_get_agent_run_progress(run_id: str):
|
|
captured["get_agent_run_progress"] = run_id
|
|
return {
|
|
"last_seq": "9-0",
|
|
"messages": [
|
|
{"seq": "8-0", "kind": "tool_call", "tool_call_id": "call-1", "content": "调用工具 read_file"},
|
|
{"seq": "9-0", "kind": "assistant_message", "message_id": "msg-2", "content": "正在整理结果"},
|
|
],
|
|
}
|
|
|
|
_patch_session(monkeypatch)
|
|
_patch_subagent_run_service(monkeypatch, _SubagentRunService)
|
|
monkeypatch.setattr(agent_run_service, "get_agent_run_result", fake_get_agent_run_result)
|
|
monkeypatch.setattr(agent_run_service, "get_agent_run_progress", fake_get_agent_run_progress)
|
|
|
|
tool = next(item for item in _async_tool_middleware().tools if item.name == "subagent_status")
|
|
result = await tool.coroutine(run_id="child-run", runtime=SimpleNamespace(tool_call_id="status-call"))
|
|
|
|
payload = json.loads(result.update["messages"][0].content)
|
|
assert payload["status"] == "running"
|
|
assert "result" not in payload
|
|
assert payload["progress"]["last_seq"] == "9-0"
|
|
assert [item["content"] for item in payload["progress"]["messages"]] == ["调用工具 read_file", "正在整理结果"]
|
|
assert captured["get_run_for_creator"] == {
|
|
"uid": "user-1",
|
|
"created_by_run_id": "parent-run",
|
|
"run_id": "child-run",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_events_cancel_and_await_use_parent_run_scope(monkeypatch) -> None:
|
|
captured: dict[str, object] = {"loads": []}
|
|
|
|
async def fake_get_verified_subagent_run(self, *, run_id: str, uid: str, created_by_run_id: str):
|
|
del self
|
|
captured["loads"].append(
|
|
{
|
|
"run_id": run_id,
|
|
"uid": uid,
|
|
"created_by_run_id": created_by_run_id,
|
|
}
|
|
)
|
|
return _subagent_run(status="completed")
|
|
|
|
async def fake_list_run_stream_events(run_id: str, *, after_seq: str, limit: int):
|
|
captured["events"] = {"run_id": run_id, "after_seq": after_seq, "limit": limit}
|
|
return [{"seq": "2-0", "event_type": "messages", "payload": {"chunk": {"status": "loading"}}}]
|
|
|
|
async def fake_request_cancel_agent_run(*, run_id: str, current_uid: str, db):
|
|
captured["cancel"] = {"run_id": run_id, "current_uid": current_uid, "db": db}
|
|
return _subagent_run(status="cancelling")
|
|
|
|
async def fake_await_agent_run_result(*, run_id: str, current_uid: str):
|
|
captured["await"] = {"run_id": run_id, "current_uid": current_uid}
|
|
return {"status": "completed", "output": "awaited result"}
|
|
|
|
_patch_session(monkeypatch)
|
|
monkeypatch.setattr(YuxiSubAgentMiddleware, "_get_verified_subagent_run", fake_get_verified_subagent_run)
|
|
monkeypatch.setattr(run_queue_service, "list_run_stream_events", fake_list_run_stream_events)
|
|
monkeypatch.setattr(agent_run_service, "request_cancel_agent_run", fake_request_cancel_agent_run)
|
|
monkeypatch.setattr(agent_run_service, "await_agent_run_result", fake_await_agent_run_result)
|
|
|
|
tools = {item.name: item for item in _async_tool_middleware().tools}
|
|
|
|
events_result = await tools["subagent_events"].coroutine(
|
|
run_id="child-run",
|
|
after_seq="1-0",
|
|
limit=99,
|
|
runtime=SimpleNamespace(tool_call_id="events-call"),
|
|
)
|
|
events_payload = json.loads(events_result.update["messages"][0].content)
|
|
assert events_payload["last_seq"] == "2-0"
|
|
assert captured["events"] == {"run_id": "child-run", "after_seq": "1-0", "limit": 50}
|
|
|
|
cancel_result = await tools["subagent_cancel"].coroutine(
|
|
run_id="child-run",
|
|
runtime=SimpleNamespace(tool_call_id="cancel-call"),
|
|
)
|
|
cancel_payload = json.loads(cancel_result.update["messages"][0].content)
|
|
assert cancel_payload["status"] == "cancelling"
|
|
assert captured["cancel"]["current_uid"] == "user-1"
|
|
|
|
await_result = await tools["subagent_await"].coroutine(
|
|
run_id="child-run",
|
|
runtime=SimpleNamespace(tool_call_id="await-call"),
|
|
)
|
|
await_payload = json.loads(await_result.update["messages"][0].content)
|
|
assert await_payload["result"]["output"] == "awaited result"
|
|
assert captured["await"] == {"run_id": "child-run", "current_uid": "user-1"}
|
|
assert captured["loads"]
|
|
assert all(
|
|
load == {"run_id": "child-run", "uid": "user-1", "created_by_run_id": "parent-run"}
|
|
for load in captured["loads"]
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_await_reports_timeout_when_run_is_still_active(monkeypatch) -> None:
|
|
captured: dict[str, object] = {"loads": []}
|
|
|
|
async def fake_get_verified_subagent_run(self, *, run_id: str, uid: str, created_by_run_id: str):
|
|
del self
|
|
captured["loads"].append(
|
|
{
|
|
"run_id": run_id,
|
|
"uid": uid,
|
|
"created_by_run_id": created_by_run_id,
|
|
}
|
|
)
|
|
return _subagent_run(status="running")
|
|
|
|
async def fake_await_agent_run_result(*, run_id: str, current_uid: str):
|
|
captured["await"] = {"run_id": run_id, "current_uid": current_uid}
|
|
raise agent_run_service.AgentRunWaitTimeout(
|
|
{"status": "running", "agent_run_id": run_id, "thread_id": "child-thread", "output": ""}
|
|
)
|
|
|
|
_patch_session(monkeypatch)
|
|
monkeypatch.setattr(YuxiSubAgentMiddleware, "_get_verified_subagent_run", fake_get_verified_subagent_run)
|
|
monkeypatch.setattr(agent_run_service, "await_agent_run_result", fake_await_agent_run_result)
|
|
|
|
result = await {item.name: item for item in _async_tool_middleware().tools}["subagent_await"].coroutine(
|
|
run_id="child-run",
|
|
runtime=SimpleNamespace(tool_call_id="await-call"),
|
|
)
|
|
|
|
payload = json.loads(result.update["messages"][0].content)
|
|
assert payload["status"] == "running"
|
|
assert payload["wait_timed_out"] is True
|
|
assert payload["message"] == "子智能体仍在运行，等待最终结果超时；请稍后继续查询。"
|
|
assert payload["result"]["status"] == "running"
|
|
assert captured["await"] == {"run_id": "child-run", "current_uid": "user-1"}
|
|
assert len(captured["loads"]) == 2
|
|
|
|
|
|
def test_merge_subagent_runs_keeps_new_run_on_same_child_thread() -> None:
|
|
child_thread_id = make_child_thread_id("parent-thread", "worker", "tool-old")
|
|
|
|
merged = merge_subagent_runs(
|
|
[
|
|
{
|
|
"id": "tool-old",
|
|
"run_id": "run-old",
|
|
"subagent_slug": "worker",
|
|
"subagent_name": "Worker",
|
|
"child_thread_id": child_thread_id,
|
|
"description": "first task",
|
|
"status": "completed",
|
|
"created_at": "2026-05-31T01:00:00Z",
|
|
"completed_at": "2026-05-31T01:01:00Z",
|
|
}
|
|
],
|
|
[
|
|
{
|
|
"id": "tool-new",
|
|
"run_id": "run-new",
|
|
"subagent_slug": "worker",
|
|
"subagent_name": "Worker",
|
|
"child_thread_id": child_thread_id,
|
|
"description": "continue task",
|
|
"status": "pending",
|
|
"created_at": "2026-05-31T02:00:00Z",
|
|
}
|
|
],
|
|
)
|
|
|
|
assert merged == [
|
|
{
|
|
"id": "tool-old",
|
|
"run_id": "run-old",
|
|
"subagent_slug": "worker",
|
|
"subagent_name": "Worker",
|
|
"child_thread_id": child_thread_id,
|
|
"description": "first task",
|
|
"status": "completed",
|
|
"created_at": "2026-05-31T01:00:00Z",
|
|
"completed_at": "2026-05-31T01:01:00Z",
|
|
},
|
|
{
|
|
"id": "tool-new",
|
|
"run_id": "run-new",
|
|
"subagent_slug": "worker",
|
|
"subagent_name": "Worker",
|
|
"child_thread_id": child_thread_id,
|
|
"description": "continue task",
|
|
"status": "pending",
|
|
"created_at": "2026-05-31T02:00:00Z",
|
|
},
|
|
]
|