wehub-resource-sync
1485 lines
52 KiB
Python
1485 lines
52 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import yuxi.services.agent_run_service as agent_run_service
|
|
from yuxi.services.input_message_service import (
|
|
build_chat_input_message,
|
|
build_chat_input_message_from_openai_content,
|
|
restore_chat_input_message,
|
|
)
|
|
|
|
|
|
def _chat_input(content: str, image_content: str | None = None):
|
|
return build_chat_input_message(content, image_content)
|
|
|
|
|
|
def _sse_data(chunk: str) -> dict:
|
|
for line in chunk.splitlines():
|
|
if line.startswith("data: "):
|
|
return json.loads(line.removeprefix("data: "))
|
|
raise AssertionError(f"SSE chunk has no data line: {chunk}")
|
|
|
|
|
|
def test_openai_content_parts_build_and_restore_multimodal_message():
|
|
input_message = build_chat_input_message_from_openai_content(
|
|
[
|
|
{"type": "text", "text": "看图"},
|
|
{"type": "image_url", "image_url": {"url": "https://example.test/image.png"}},
|
|
]
|
|
)
|
|
|
|
assert input_message.content == "看图"
|
|
assert input_message.message_type == "multimodal_image"
|
|
assert input_message.image_content is None
|
|
raw_message = input_message.raw_message()
|
|
assert raw_message["content"][1]["image_url"]["url"] == "https://example.test/image.png"
|
|
|
|
restored = restore_chat_input_message(
|
|
content=input_message.content, image_content=None, metadata={"raw_message": raw_message}
|
|
)
|
|
assert restored.message_type == "multimodal_image"
|
|
assert restored.require_langchain_message().content == raw_message["content"]
|
|
|
|
|
|
def test_prepare_run_input_message_keeps_invocation_meta_namespaced():
|
|
input_message = agent_run_service._prepare_run_input_message(
|
|
run_type="chat",
|
|
input_message=build_chat_input_message("hello"),
|
|
resume=None,
|
|
request_id="req-1",
|
|
model_spec="provider:model",
|
|
meta={
|
|
"source": "agent_call",
|
|
"agent_invocation_meta": {"trace_id": "trace-1"},
|
|
"evaluation": {"dataset_name": "legacy-top-level"},
|
|
"custom_variables": {"system_prompt": "legacy"},
|
|
},
|
|
)
|
|
|
|
assert input_message.extra_metadata["source"] == "agent_call"
|
|
assert input_message.extra_metadata["agent_invocation_meta"] == {"trace_id": "trace-1"}
|
|
assert "evaluation" not in input_message.extra_metadata
|
|
assert "custom_variables" not in input_message.extra_metadata
|
|
|
|
|
|
def _progress_event(seq: str, chunks: list[dict]) -> dict:
|
|
return {
|
|
"seq": seq,
|
|
"event_type": "messages",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "messages",
|
|
"payload": {"items": chunks},
|
|
"created_at": "2026-06-30T00:00:00+00:00",
|
|
},
|
|
"ts": 1700000000000,
|
|
}
|
|
|
|
|
|
def _progress_chunk(stream_event: dict) -> dict:
|
|
return {"status": "loading", "stream_event": stream_event}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_agent_run_progress_returns_empty_progress_for_empty_events(monkeypatch: pytest.MonkeyPatch):
|
|
async def fake_list_recent_run_stream_events(run_id: str, *, limit: int):
|
|
assert run_id == "run-1"
|
|
assert limit == agent_run_service.RUN_PROGRESS_RECENT_EVENT_SCAN_LIMIT
|
|
return []
|
|
|
|
monkeypatch.setattr(agent_run_service, "list_recent_run_stream_events", fake_list_recent_run_stream_events)
|
|
|
|
assert await agent_run_service.get_agent_run_progress("run-1") == {"last_seq": "0-0", "messages": []}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_agent_run_progress_extracts_recent_message_delta_events(monkeypatch: pytest.MonkeyPatch):
|
|
async def fake_list_recent_run_stream_events(run_id: str, *, limit: int):
|
|
del run_id, limit
|
|
return [
|
|
_progress_event(
|
|
"2-0",
|
|
[
|
|
_progress_chunk({"type": "message_delta", "message_id": "msg-1", "content": " world"}),
|
|
_progress_chunk({"type": "message_delta", "message_id": "msg-2", "content": "new"}),
|
|
],
|
|
),
|
|
_progress_event(
|
|
"1-0",
|
|
[_progress_chunk({"type": "message_delta", "message_id": "msg-1", "content": "hello"})],
|
|
),
|
|
]
|
|
|
|
monkeypatch.setattr(agent_run_service, "list_recent_run_stream_events", fake_list_recent_run_stream_events)
|
|
|
|
progress = await agent_run_service.get_agent_run_progress("run-1")
|
|
|
|
assert progress["last_seq"] == "2-0"
|
|
assert progress["messages"] == [
|
|
{
|
|
"seq": "1-0",
|
|
"kind": "assistant_message",
|
|
"message_id": "msg-1",
|
|
"content": "hello",
|
|
},
|
|
{
|
|
"seq": "2-0",
|
|
"kind": "assistant_message",
|
|
"message_id": "msg-1",
|
|
"content": "world",
|
|
},
|
|
{
|
|
"seq": "2-0",
|
|
"kind": "assistant_message",
|
|
"message_id": "msg-2",
|
|
"content": "new",
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_agent_run_progress_keeps_latest_three_readable_items(monkeypatch: pytest.MonkeyPatch):
|
|
events = [
|
|
_progress_event(
|
|
f"{seq}-0",
|
|
[_progress_chunk({"type": "message_delta", "message_id": f"msg-{seq}", "content": f"message-{seq}"})],
|
|
)
|
|
for seq in range(4, 0, -1)
|
|
]
|
|
|
|
async def fake_list_recent_run_stream_events(run_id: str, *, limit: int):
|
|
del run_id, limit
|
|
return events
|
|
|
|
monkeypatch.setattr(agent_run_service, "list_recent_run_stream_events", fake_list_recent_run_stream_events)
|
|
|
|
progress = await agent_run_service.get_agent_run_progress("run-1")
|
|
|
|
assert progress["last_seq"] == "4-0"
|
|
assert [item["message_id"] for item in progress["messages"]] == ["msg-2", "msg-3", "msg-4"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_agent_run_progress_extracts_tool_call_events(monkeypatch: pytest.MonkeyPatch):
|
|
async def fake_list_recent_run_stream_events(run_id: str, *, limit: int):
|
|
del run_id, limit
|
|
return [
|
|
_progress_event(
|
|
"3-0",
|
|
[
|
|
_progress_chunk(
|
|
{
|
|
"type": "tool_call",
|
|
"message_id": "msg-3",
|
|
"tool_call_id": "call-3",
|
|
"name": "write_file",
|
|
"args": {"path": "/home/gem/user-data/outputs/report.md"},
|
|
}
|
|
)
|
|
],
|
|
),
|
|
_progress_event(
|
|
"2-0",
|
|
[
|
|
_progress_chunk(
|
|
{
|
|
"type": "tool_call_delta",
|
|
"message_id": "msg-2",
|
|
"tool_call_id": "call-2",
|
|
"name": "read_file",
|
|
"args_delta": '{"path":',
|
|
}
|
|
)
|
|
],
|
|
),
|
|
]
|
|
|
|
monkeypatch.setattr(agent_run_service, "list_recent_run_stream_events", fake_list_recent_run_stream_events)
|
|
|
|
progress = await agent_run_service.get_agent_run_progress("run-1")
|
|
|
|
assert progress["messages"][0]["kind"] == "tool_call_delta"
|
|
assert progress["messages"][0]["tool_call_id"] == "call-2"
|
|
assert progress["messages"][0]["content"] == "正在准备工具 read_file"
|
|
assert progress["messages"][1]["kind"] == "tool_call"
|
|
assert progress["messages"][1]["tool_call_id"] == "call-3"
|
|
assert progress["messages"][1]["content"] == "调用工具 write_file"
|
|
|
|
|
|
class _FakeContext:
|
|
def __init__(self):
|
|
self.model = "agent-default-model"
|
|
|
|
def update_from_dict(self, data: dict):
|
|
for key, value in data.items():
|
|
if hasattr(self, key):
|
|
setattr(self, key, value)
|
|
|
|
|
|
class _FakeBackend:
|
|
context_schema = _FakeContext
|
|
|
|
|
|
class _UserResult:
|
|
def scalar_one_or_none(self):
|
|
return SimpleNamespace(uid="user-1", role="user")
|
|
|
|
|
|
class _CreateRunDb:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
message_id: int = 10,
|
|
active_run: SimpleNamespace | None = None,
|
|
active_run_after_rollback: SimpleNamespace | None = None,
|
|
existing_run: SimpleNamespace | None = None,
|
|
existing_run_after_rollback: SimpleNamespace | None = None,
|
|
runs_by_id: dict[str, SimpleNamespace] | None = None,
|
|
raise_create_integrity_error: bool = False,
|
|
):
|
|
self.added = []
|
|
self.deleted = []
|
|
self.committed = False
|
|
self.created_run = None
|
|
self.created_run_kwargs = None
|
|
self.enqueued: list[tuple[str, str, str]] = []
|
|
self.order: list[str] = []
|
|
self.request_id_lookups: list[str] = []
|
|
self.active_run_lookup = None
|
|
self.active_run = active_run
|
|
self.active_run_after_rollback = active_run_after_rollback
|
|
self.existing_run = existing_run
|
|
self.existing_run_after_rollback = existing_run_after_rollback
|
|
self.runs_by_id = runs_by_id or {}
|
|
self.raise_create_integrity_error = raise_create_integrity_error
|
|
self._message_id = message_id
|
|
|
|
async def execute(self, stmt):
|
|
del stmt
|
|
return _UserResult()
|
|
|
|
def add(self, item):
|
|
self.added.append(item)
|
|
|
|
async def flush(self):
|
|
self.order.append("flush")
|
|
for item in self.added:
|
|
if getattr(item, "id", None) is None:
|
|
item.id = self._message_id
|
|
|
|
async def commit(self):
|
|
self.order.append("commit")
|
|
self.committed = True
|
|
|
|
async def rollback(self):
|
|
self.order.append("rollback")
|
|
|
|
async def delete(self, item):
|
|
self.deleted.append(item)
|
|
self.order.append("delete")
|
|
|
|
def begin_nested(self):
|
|
db = self
|
|
|
|
class NestedTransaction:
|
|
async def __aenter__(self):
|
|
db.order.append("begin_nested")
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
if exc_type is agent_run_service.IntegrityError:
|
|
db.order.append("rollback_savepoint")
|
|
else:
|
|
db.order.append("release_savepoint")
|
|
return False
|
|
|
|
return NestedTransaction()
|
|
|
|
|
|
class _CreateRunRepo:
|
|
def __init__(self, db_session):
|
|
self.db = db_session
|
|
|
|
async def get_run_by_request_id(self, request_id: str):
|
|
self.db.request_id_lookups.append(request_id)
|
|
if "rollback_savepoint" in self.db.order and self.db.existing_run_after_rollback:
|
|
return self.db.existing_run_after_rollback
|
|
return self.db.existing_run
|
|
|
|
async def get_active_run_by_thread_for_user(self, *, agent_slug: str, conversation_thread_id: str, uid: str):
|
|
self.db.active_run_lookup = {
|
|
"agent_slug": agent_slug,
|
|
"conversation_thread_id": conversation_thread_id,
|
|
"uid": uid,
|
|
}
|
|
if "rollback_savepoint" in self.db.order and self.db.active_run_after_rollback:
|
|
return self.db.active_run_after_rollback
|
|
return self.db.active_run
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
assert uid == "user-1"
|
|
return self.db.runs_by_id.get(run_id)
|
|
|
|
async def create_run(self, **kwargs):
|
|
self.db.created_run_kwargs = kwargs
|
|
if self.db.raise_create_integrity_error:
|
|
raise agent_run_service.IntegrityError("insert agent_run", kwargs, Exception("duplicate request_id"))
|
|
self.db.created_run = SimpleNamespace(
|
|
id=kwargs["run_id"],
|
|
conversation_thread_id=kwargs["conversation_thread_id"],
|
|
agent_slug=kwargs["agent_slug"],
|
|
status="pending",
|
|
request_id=kwargs["request_id"],
|
|
uid=kwargs["uid"],
|
|
run_type=kwargs["run_type"],
|
|
created_by_run_id=kwargs.get("created_by_run_id"),
|
|
subagent_thread_relation_id=kwargs.get("subagent_thread_relation_id"),
|
|
)
|
|
return self.db.created_run
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_agent_run_events_emits_error_on_db_error(monkeypatch: pytest.MonkeyPatch):
|
|
@asynccontextmanager
|
|
async def fake_session_ctx():
|
|
yield object()
|
|
|
|
class BrokenRepo:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
del run_id, uid
|
|
raise RuntimeError("db down")
|
|
|
|
monkeypatch.setattr(agent_run_service.pg_manager, "get_async_session_context", fake_session_ctx)
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", BrokenRepo)
|
|
|
|
chunks = []
|
|
async for chunk in agent_run_service.stream_agent_run_events(
|
|
run_id="run-1",
|
|
after_seq="0",
|
|
current_uid="user-1",
|
|
):
|
|
chunks.append(chunk)
|
|
|
|
assert len(chunks) == 1
|
|
assert chunks[0].startswith("event: error")
|
|
assert '"reason": "db_error"' in chunks[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_agent_run_events_reads_redis_and_ends_on_end_event(monkeypatch: pytest.MonkeyPatch):
|
|
@asynccontextmanager
|
|
async def fake_session_ctx():
|
|
yield object()
|
|
|
|
class Repo:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
del run_id, uid
|
|
return SimpleNamespace(status="completed", conversation_thread_id="thread-1")
|
|
|
|
calls = {"count": 0}
|
|
|
|
async def fake_list_events(run_id: str, *, after_seq: str, limit: int):
|
|
del run_id, after_seq, limit
|
|
calls["count"] += 1
|
|
if calls["count"] == 1:
|
|
return [
|
|
{
|
|
"seq": "1700000000000-0",
|
|
"event_type": "messages",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "messages",
|
|
"payload": {"items": [{"status": "loading", "response": "你"}]},
|
|
"created_at": "2026-05-27T00:00:00+00:00",
|
|
},
|
|
"ts": 1700000000000,
|
|
},
|
|
{
|
|
"seq": "1700000000001-0",
|
|
"event_type": "end",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "end",
|
|
"payload": {"status": "completed"},
|
|
"created_at": "2026-05-27T00:00:01+00:00",
|
|
},
|
|
"ts": 1700000000001,
|
|
},
|
|
]
|
|
return []
|
|
|
|
monkeypatch.setattr(agent_run_service.pg_manager, "get_async_session_context", fake_session_ctx)
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo)
|
|
monkeypatch.setattr(agent_run_service, "list_run_stream_events", fake_list_events)
|
|
monkeypatch.setattr(agent_run_service, "SSE_POLL_INTERVAL_SECONDS", 0)
|
|
|
|
chunks = []
|
|
async for chunk in agent_run_service.stream_agent_run_events(
|
|
run_id="run-1",
|
|
after_seq="0",
|
|
current_uid="user-1",
|
|
):
|
|
chunks.append(chunk)
|
|
|
|
assert chunks[0].startswith("event: messages")
|
|
assert "id: 1700000000000-0" in chunks[0]
|
|
assert chunks[-1].startswith("event: end")
|
|
assert "id: 1700000000001-0" in chunks[-1]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_agent_run_events_compacts_verbose_false(monkeypatch: pytest.MonkeyPatch):
|
|
@asynccontextmanager
|
|
async def fake_session_ctx():
|
|
yield object()
|
|
|
|
class Repo:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
del run_id, uid
|
|
return SimpleNamespace(status="completed", conversation_thread_id="thread-1")
|
|
|
|
async def fake_list_events(run_id: str, *, after_seq: str, limit: int):
|
|
del run_id, after_seq, limit
|
|
return [
|
|
{
|
|
"seq": "1700000000000-0",
|
|
"event_type": "metadata",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "metadata",
|
|
"payload": {
|
|
"request_id": "req-1",
|
|
"agent_slug": "deep-research",
|
|
"backend_id": "ChatbotAgent",
|
|
"uid": "user-1",
|
|
},
|
|
"created_at": "2026-05-27T00:00:00+00:00",
|
|
},
|
|
"ts": 1700000000000,
|
|
},
|
|
{
|
|
"seq": "1700000000001-0",
|
|
"event_type": "custom",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "custom",
|
|
"payload": {
|
|
"name": "yuxi.init",
|
|
"chunk": {
|
|
"request_id": "req-1",
|
|
"response": None,
|
|
"thread_id": "thread-1",
|
|
"status": "init",
|
|
"meta": {"query": "写一个冒泡排序", "uid": "user-1"},
|
|
"msg": {
|
|
"role": "user",
|
|
"content": "写一个冒泡排序",
|
|
"type": "human",
|
|
"image_content": "base64-image-data",
|
|
"extra_metadata": {
|
|
"request_id": "req-1",
|
|
"attachments": [],
|
|
"debug": "drop-me",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
"created_at": "2026-05-27T00:00:00+00:00",
|
|
},
|
|
"ts": 1700000000001,
|
|
},
|
|
{
|
|
"seq": "1700000000002-0",
|
|
"event_type": "custom",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "custom",
|
|
"payload": {
|
|
"name": "yuxi.agent_state",
|
|
"chunk": {
|
|
"request_id": "req-1",
|
|
"response": None,
|
|
"thread_id": "thread-1",
|
|
"status": "agent_state",
|
|
"agent_state": {
|
|
"todos": [],
|
|
"files": {},
|
|
"artifacts": [],
|
|
"subagent_runs": [],
|
|
},
|
|
"meta": {"uid": "user-1"},
|
|
},
|
|
"agent_state": {
|
|
"todos": [],
|
|
"files": {},
|
|
"artifacts": [],
|
|
"subagent_runs": [],
|
|
},
|
|
},
|
|
"created_at": "2026-05-27T00:00:00+00:00",
|
|
},
|
|
"ts": 1700000000002,
|
|
},
|
|
{
|
|
"seq": "1700000000003-0",
|
|
"event_type": "messages",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "messages",
|
|
"payload": {
|
|
"items": [
|
|
{
|
|
"request_id": "req-1",
|
|
"response": "你",
|
|
"thread_id": "thread-1",
|
|
"status": "loading",
|
|
"stream_event": {
|
|
"type": "tool_call",
|
|
"message_id": "msg-1",
|
|
"tool_call_id": "call-1",
|
|
"name": "ls",
|
|
"args": {"path": "/home/gem/user-data/outputs"},
|
|
"thread_id": "thread-1",
|
|
"namespace": [],
|
|
},
|
|
"metadata": {
|
|
"langfuse_user_id": "user-1",
|
|
"langgraph_checkpoint_ns": "model:checkpoint",
|
|
},
|
|
}
|
|
]
|
|
},
|
|
"created_at": "2026-05-27T00:00:01+00:00",
|
|
},
|
|
"ts": 1700000000003,
|
|
},
|
|
{
|
|
"seq": "1700000000004-0",
|
|
"event_type": "end",
|
|
"payload": {
|
|
"schema_version": 1,
|
|
"run_id": "run-1",
|
|
"thread_id": "thread-1",
|
|
"event": "end",
|
|
"payload": {
|
|
"status": "completed",
|
|
"chunk": {"status": "finished", "request_id": "req-1", "meta": {"uid": "user-1"}},
|
|
},
|
|
"created_at": "2026-05-27T00:00:02+00:00",
|
|
},
|
|
"ts": 1700000000004,
|
|
},
|
|
]
|
|
|
|
monkeypatch.setattr(agent_run_service.pg_manager, "get_async_session_context", fake_session_ctx)
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo)
|
|
monkeypatch.setattr(agent_run_service, "list_run_stream_events", fake_list_events)
|
|
|
|
chunks = []
|
|
async for chunk in agent_run_service.stream_agent_run_events(
|
|
run_id="run-1",
|
|
after_seq="0",
|
|
current_uid="user-1",
|
|
verbose=False,
|
|
):
|
|
chunks.append(chunk)
|
|
|
|
assert len(chunks) == 3
|
|
|
|
init_data = _sse_data(chunks[0])
|
|
init_chunk = init_data["payload"]["chunk"]
|
|
assert init_data["request_id"] == "req-1"
|
|
assert init_data["payload"]["name"] == "yuxi.init"
|
|
assert "meta" not in init_chunk
|
|
assert "request_id" not in init_chunk
|
|
assert "response" not in init_chunk
|
|
assert "thread_id" not in init_chunk
|
|
assert "image_content" not in init_chunk["msg"]
|
|
assert "extra_metadata" not in init_chunk["msg"]
|
|
|
|
message_data = _sse_data(chunks[1])
|
|
message_chunk = message_data["payload"]["items"][0]
|
|
assert message_data["request_id"] == "req-1"
|
|
assert "request_id" not in message_chunk
|
|
assert "metadata" not in message_chunk
|
|
assert "response" not in message_chunk
|
|
assert "thread_id" not in message_chunk
|
|
assert message_chunk["stream_event"]["tool_call_id"] == "call-1"
|
|
assert "thread_id" not in message_chunk["stream_event"]
|
|
assert "namespace" not in message_chunk["stream_event"]
|
|
|
|
end_data = _sse_data(chunks[2])
|
|
assert end_data["request_id"] == "req-1"
|
|
assert end_data["payload"]["status"] == "completed"
|
|
assert "request_id" not in end_data["payload"]["chunk"]
|
|
assert "meta" not in end_data["payload"]["chunk"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_agent_run_events_compact_fallback_end_keeps_request_id(monkeypatch: pytest.MonkeyPatch):
|
|
@asynccontextmanager
|
|
async def fake_session_ctx():
|
|
yield object()
|
|
|
|
class Repo:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
del run_id, uid
|
|
return SimpleNamespace(status="completed", conversation_thread_id="thread-1", request_id="req-1")
|
|
|
|
async def fake_list_events(run_id: str, *, after_seq: str, limit: int):
|
|
del run_id, after_seq, limit
|
|
return []
|
|
|
|
async def fake_get_last_run_stream_seq(run_id: str):
|
|
del run_id
|
|
return "1700000000004-0"
|
|
|
|
monkeypatch.setattr(agent_run_service.pg_manager, "get_async_session_context", fake_session_ctx)
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo)
|
|
monkeypatch.setattr(agent_run_service, "list_run_stream_events", fake_list_events)
|
|
monkeypatch.setattr(agent_run_service, "get_last_run_stream_seq", fake_get_last_run_stream_seq)
|
|
|
|
chunks = []
|
|
async for chunk in agent_run_service.stream_agent_run_events(
|
|
run_id="run-1",
|
|
after_seq="0",
|
|
current_uid="user-1",
|
|
verbose=False,
|
|
):
|
|
chunks.append(chunk)
|
|
|
|
assert len(chunks) == 1
|
|
assert chunks[0].startswith("event: end")
|
|
assert "id: 1700000000004-0" in chunks[0]
|
|
data = _sse_data(chunks[0])
|
|
assert data["request_id"] == "req-1"
|
|
assert data["payload"] == {"status": "completed"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_run_persists_input_before_enqueue(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(monkeypatch)
|
|
|
|
result = await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert db.order[-2:] == ["commit", "enqueue"]
|
|
assert result["run_id"] == db.created_run.id
|
|
assert result["request_id"] == "req-1"
|
|
assert db.request_id_lookups == ["req-1"]
|
|
assert db.created_run_kwargs["request_id"] == "req-1"
|
|
assert db.created_run_kwargs["conversation_id"] == 1
|
|
assert db.created_run_kwargs["input_message_id"] == 10
|
|
assert db.added[0].run_id == db.created_run.id
|
|
assert db.added[0].request_id == "req-1"
|
|
assert db.enqueued == [("process_agent_run", db.created_run.id, f"run:{db.created_run.id}")]
|
|
assert db.created_run_kwargs["input_payload"] == {"model_spec": "agent-default-model"}
|
|
assert "model_spec" not in db.added[0].extra_metadata
|
|
assert db.added[0].extra_metadata["raw_message"]["type"] == "human"
|
|
assert db.added[0].extra_metadata["raw_message"]["content"] == "hello"
|
|
assert "run_id" not in db.added[0].extra_metadata
|
|
assert "run_type" not in db.added[0].extra_metadata
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_run_reuses_existing_only_with_same_request_scope(monkeypatch: pytest.MonkeyPatch):
|
|
existing_run = SimpleNamespace(
|
|
id="existing-run",
|
|
conversation_thread_id="thread-1",
|
|
agent_slug="default",
|
|
status="pending",
|
|
request_id="req-1",
|
|
uid="user-1",
|
|
run_type="chat",
|
|
created_by_run_id=None,
|
|
subagent_thread_relation_id=None,
|
|
)
|
|
db = _patch_agent_run_creation(monkeypatch, existing_run=existing_run)
|
|
|
|
result = await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert result["run_id"] == "existing-run"
|
|
assert db.active_run_lookup is None
|
|
assert db.created_run_kwargs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_run_rejects_request_id_scope_mismatch(monkeypatch: pytest.MonkeyPatch):
|
|
existing_run = SimpleNamespace(
|
|
id="existing-run",
|
|
conversation_thread_id="other-thread",
|
|
agent_slug="default",
|
|
status="pending",
|
|
request_id="req-1",
|
|
uid="user-1",
|
|
run_type="chat",
|
|
created_by_run_id=None,
|
|
subagent_thread_relation_id=None,
|
|
)
|
|
db = _patch_agent_run_creation(monkeypatch, existing_run=existing_run)
|
|
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert exc.value.status_code == 409
|
|
assert exc.value.detail == "request_id 冲突"
|
|
assert db.created_run_kwargs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_run_integrity_error_reuses_same_request_scope(monkeypatch: pytest.MonkeyPatch):
|
|
existing_run = SimpleNamespace(
|
|
id="existing-run",
|
|
conversation_thread_id="thread-1",
|
|
agent_slug="default",
|
|
status="pending",
|
|
request_id="req-1",
|
|
uid="user-1",
|
|
run_type="chat",
|
|
created_by_run_id=None,
|
|
subagent_thread_relation_id=None,
|
|
)
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
existing_run_after_rollback=existing_run,
|
|
raise_create_integrity_error=True,
|
|
)
|
|
|
|
result = await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert result["run_id"] == "existing-run"
|
|
assert "rollback_savepoint" in db.order
|
|
assert "rollback" not in db.order
|
|
assert db.deleted == [db.added[0]]
|
|
assert "commit" not in db.order
|
|
assert "enqueue" not in db.order
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_run_integrity_error_rejects_scope_mismatch(monkeypatch: pytest.MonkeyPatch):
|
|
existing_run = SimpleNamespace(
|
|
id="existing-run",
|
|
conversation_thread_id="other-thread",
|
|
agent_slug="default",
|
|
status="pending",
|
|
request_id="req-1",
|
|
uid="user-1",
|
|
run_type="chat",
|
|
created_by_run_id=None,
|
|
subagent_thread_relation_id=None,
|
|
)
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
existing_run_after_rollback=existing_run,
|
|
raise_create_integrity_error=True,
|
|
)
|
|
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert exc.value.status_code == 409
|
|
assert exc.value.detail == "request_id 冲突"
|
|
assert "rollback_savepoint" in db.order
|
|
assert "rollback" not in db.order
|
|
assert "commit" not in db.order
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_run_integrity_error_returns_run_busy_for_active_thread(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
active_run_after_rollback=SimpleNamespace(id="active-run", status="pending"),
|
|
raise_create_integrity_error=True,
|
|
)
|
|
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-2"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert exc.value.status_code == 409
|
|
assert exc.value.detail["code"] == "run_busy"
|
|
assert exc.value.detail["active_run_id"] == "active-run"
|
|
assert db.request_id_lookups == ["req-2", "req-2"]
|
|
assert "rollback_savepoint" in db.order
|
|
assert "rollback" not in db.order
|
|
assert "commit" not in db.order
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_resume_run_marks_input_message_source(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
message_id=11,
|
|
parent_run=SimpleNamespace(
|
|
id="parent-run",
|
|
conversation_thread_id="thread-1",
|
|
status="interrupted",
|
|
input_payload={"model_spec": "parent-model"},
|
|
),
|
|
)
|
|
|
|
result = await agent_run_service.create_agent_run_view(
|
|
input_message=None,
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "resume-req"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
resume={"language": "python"},
|
|
created_by_run_id="parent-run",
|
|
)
|
|
|
|
assert result["run_id"] == db.created_run.id
|
|
assert db.created_run_kwargs["run_type"] == "resume"
|
|
assert db.created_run_kwargs["created_by_run_id"] == "parent-run"
|
|
assert db.created_run_kwargs["input_message_id"] == 11
|
|
assert db.added[0].message_type == "resume"
|
|
assert db.added[0].extra_metadata["source"] == "ask_user_question_resume"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_resume_run_without_request_id_reuses_stable_key(monkeypatch: pytest.MonkeyPatch):
|
|
parent_run = SimpleNamespace(
|
|
id="parent-run",
|
|
conversation_thread_id="thread-1",
|
|
status="interrupted",
|
|
input_payload={"model_spec": "parent-model"},
|
|
)
|
|
first_db = _patch_agent_run_creation(monkeypatch, parent_run=parent_run)
|
|
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=None,
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={},
|
|
current_uid="user-1",
|
|
db=first_db,
|
|
resume={"language": "python", "answer": "ok"},
|
|
created_by_run_id="parent-run",
|
|
)
|
|
|
|
request_id = first_db.created_run_kwargs["request_id"]
|
|
existing_run = SimpleNamespace(
|
|
id="existing-resume-run",
|
|
conversation_thread_id="thread-1",
|
|
agent_slug="default",
|
|
status="pending",
|
|
request_id=request_id,
|
|
uid="user-1",
|
|
run_type="resume",
|
|
created_by_run_id="parent-run",
|
|
subagent_thread_relation_id=None,
|
|
)
|
|
retry_db = _patch_agent_run_creation(monkeypatch, existing_run=existing_run, parent_run=parent_run)
|
|
|
|
result = await agent_run_service.create_agent_run_view(
|
|
input_message=None,
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={},
|
|
current_uid="user-1",
|
|
db=retry_db,
|
|
resume={"answer": "ok", "language": "python"},
|
|
created_by_run_id="parent-run",
|
|
)
|
|
|
|
assert result["run_id"] == "existing-resume-run"
|
|
assert request_id.startswith("resume:")
|
|
assert len(request_id) <= 64
|
|
assert retry_db.request_id_lookups == [request_id]
|
|
assert retry_db.created_run_kwargs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_resume_run_requires_parent_run_id(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(monkeypatch)
|
|
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=None,
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "resume-req"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
resume={"language": "python"},
|
|
)
|
|
|
|
assert exc.value.status_code == 422
|
|
assert exc.value.detail == "created_by_run_id 不能为空"
|
|
assert db.created_run_kwargs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_resume_run_rejects_non_interrupted_parent(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
parent_run=SimpleNamespace(
|
|
id="parent-run",
|
|
conversation_thread_id="thread-1",
|
|
status="running",
|
|
input_payload={"model_spec": "parent-model"},
|
|
),
|
|
)
|
|
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=None,
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "resume-req"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
resume={"language": "python"},
|
|
created_by_run_id="parent-run",
|
|
)
|
|
|
|
assert exc.value.status_code == 409
|
|
assert exc.value.detail == "只有 interrupted run 可以恢复"
|
|
assert db.created_run_kwargs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_resume_run_rejects_parent_without_model_snapshot(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
parent_run=SimpleNamespace(
|
|
id="parent-run",
|
|
conversation_thread_id="thread-1",
|
|
status="interrupted",
|
|
input_payload={},
|
|
),
|
|
)
|
|
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=None,
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "resume-req"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
resume={"language": "python"},
|
|
created_by_run_id="parent-run",
|
|
)
|
|
|
|
assert exc.value.status_code == 409
|
|
assert exc.value.detail == "被恢复的运行任务缺少模型快照"
|
|
assert db.created_run_kwargs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_agent_run_rejects_active_checkpoint_run(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(monkeypatch, active_run=SimpleNamespace(id="active-run", status="running"))
|
|
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert exc.value.status_code == 409
|
|
assert exc.value.detail["code"] == "run_busy"
|
|
assert exc.value.detail["active_run_id"] == "active-run"
|
|
assert db.active_run_lookup == {
|
|
"agent_slug": "default",
|
|
"conversation_thread_id": "thread-1",
|
|
"uid": "user-1",
|
|
}
|
|
assert db.created_run_kwargs is None
|
|
|
|
|
|
# ==================== run 结果基础能力 ====================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_agent_run_result_uses_output_message_id(monkeypatch: pytest.MonkeyPatch):
|
|
run = SimpleNamespace(
|
|
id="run-1",
|
|
status="completed",
|
|
agent_slug="default-chatbot",
|
|
conversation_thread_id="thread-1",
|
|
conversation_id=10,
|
|
request_id="req-1",
|
|
output_message_id=2,
|
|
error_type=None,
|
|
error_message=None,
|
|
)
|
|
messages = [
|
|
SimpleNamespace(id=1, role="user", content="question", extra_metadata={}),
|
|
SimpleNamespace(id=2, role="assistant", content="older", extra_metadata={"langfuse_trace_id": "trace-old"}),
|
|
SimpleNamespace(id=3, role="assistant", content="final", extra_metadata={"langfuse_trace_id": "trace-final"}),
|
|
]
|
|
|
|
class FakeScalars:
|
|
def unique(self):
|
|
return self
|
|
|
|
def all(self):
|
|
return messages
|
|
|
|
class FakeResult:
|
|
def scalars(self):
|
|
return FakeScalars()
|
|
|
|
class FakeDB:
|
|
async def execute(self, _stmt):
|
|
return FakeResult()
|
|
|
|
class RunRepo:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
assert run_id == "run-1"
|
|
assert uid == "user-1"
|
|
return run
|
|
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", RunRepo)
|
|
|
|
payload = await agent_run_service.get_agent_run_result(run_id="run-1", current_uid="user-1", db=FakeDB())
|
|
|
|
assert payload["status"] == "completed"
|
|
assert payload["output"] == "older"
|
|
assert payload["final_message_id"] == 2
|
|
assert payload["langfuse_trace_id"] == "trace-old"
|
|
assert "debug" not in payload
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_agent_run_result_missing_run_returns_failed(monkeypatch: pytest.MonkeyPatch):
|
|
class RunRepo:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
del run_id, uid
|
|
return None
|
|
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", RunRepo)
|
|
|
|
payload = await agent_run_service.get_agent_run_result(run_id="run-x", current_uid="user-1", db=object())
|
|
|
|
assert payload["status"] == "failed"
|
|
assert payload["error"]["type"] == "run_not_found"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_await_agent_run_result_drains_stream_then_loads_result(monkeypatch: pytest.MonkeyPatch):
|
|
drained: list[str] = []
|
|
|
|
async def fake_stream(*, run_id: str, after_seq: str, current_uid: str, verbose: bool):
|
|
assert run_id == "run-1"
|
|
assert after_seq == "0-0"
|
|
assert current_uid == "user-1"
|
|
assert verbose is False
|
|
for chunk in ("event: messages\n\n", "event: end\n\n"):
|
|
drained.append(chunk)
|
|
yield chunk
|
|
|
|
async def fake_load(*, run_id: str, current_uid: str):
|
|
assert run_id == "run-1"
|
|
assert current_uid == "user-1"
|
|
return {"status": "completed", "output": "final"}
|
|
|
|
monkeypatch.setattr(agent_run_service, "stream_agent_run_events", fake_stream)
|
|
monkeypatch.setattr(agent_run_service, "load_agent_run_result", fake_load)
|
|
|
|
payload = await agent_run_service.await_agent_run_result(run_id="run-1", current_uid="user-1")
|
|
|
|
assert len(drained) == 2
|
|
assert payload == {"status": "completed", "output": "final"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_await_agent_run_result_raises_when_stream_ends_before_terminal(monkeypatch: pytest.MonkeyPatch):
|
|
async def fake_stream(*, run_id: str, after_seq: str, current_uid: str, verbose: bool):
|
|
assert run_id == "run-1"
|
|
assert after_seq == "0-0"
|
|
assert current_uid == "user-1"
|
|
assert verbose is False
|
|
yield ": heartbeat\n\n"
|
|
|
|
async def fake_load(*, run_id: str, current_uid: str):
|
|
assert run_id == "run-1"
|
|
assert current_uid == "user-1"
|
|
return {"status": "running", "agent_run_id": run_id, "output": ""}
|
|
|
|
monkeypatch.setattr(agent_run_service, "stream_agent_run_events", fake_stream)
|
|
monkeypatch.setattr(agent_run_service, "load_agent_run_result", fake_load)
|
|
|
|
with pytest.raises(agent_run_service.AgentRunWaitTimeout) as exc_info:
|
|
await agent_run_service.await_agent_run_result(run_id="run-1", current_uid="user-1")
|
|
|
|
assert exc_info.value.result == {"status": "running", "agent_run_id": "run-1", "output": ""}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_agent_run_view_cascades_children(monkeypatch: pytest.MonkeyPatch):
|
|
parent_run = SimpleNamespace(id="parent-run", uid="user-1", to_dict=lambda: {"id": "parent-run"})
|
|
child_runs = [SimpleNamespace(id="child-1"), SimpleNamespace(id="child-2")]
|
|
requested: list[str] = []
|
|
signals: list[tuple[str, bool]] = []
|
|
|
|
class Db:
|
|
committed = False
|
|
|
|
async def commit(self):
|
|
self.committed = True
|
|
|
|
class RunRepo:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
|
|
async def get_run_for_user(self, run_id: str, uid: str):
|
|
assert run_id == "parent-run"
|
|
assert uid == "user-1"
|
|
return parent_run
|
|
|
|
async def list_active_child_runs_for_user(self, created_by_run_id: str, uid: str):
|
|
assert created_by_run_id == "parent-run"
|
|
assert uid == "user-1"
|
|
return child_runs
|
|
|
|
async def request_cancel(self, run_id: str):
|
|
requested.append(run_id)
|
|
return parent_run if run_id == "parent-run" else SimpleNamespace(id=run_id)
|
|
|
|
async def fake_publish_cancel_signal(run_id: str):
|
|
signals.append((run_id, db.committed))
|
|
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", RunRepo)
|
|
monkeypatch.setattr(agent_run_service, "publish_cancel_signal", fake_publish_cancel_signal)
|
|
db = Db()
|
|
|
|
result = await agent_run_service.cancel_agent_run_view(
|
|
run_id="parent-run",
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert result["run"]["id"] == "parent-run"
|
|
assert requested == ["child-1", "child-2", "parent-run"]
|
|
assert signals == [("child-1", True), ("child-2", True), ("parent-run", True)]
|
|
|
|
|
|
def test_resolve_agent_run_model_spec_rejects_unknown_explicit_model(monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr(agent_run_service.model_cache, "get_model_info", lambda spec: None)
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
agent_run_service.resolve_agent_run_model_spec("nope", SimpleNamespace(config_json={}), _FakeBackend())
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
def test_resolve_agent_run_model_spec_rejects_non_chat_explicit_model(monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr(
|
|
agent_run_service.model_cache,
|
|
"get_model_info",
|
|
lambda spec: SimpleNamespace(model_type="embedding"),
|
|
)
|
|
with pytest.raises(agent_run_service.HTTPException) as exc:
|
|
agent_run_service.resolve_agent_run_model_spec("embed-1", SimpleNamespace(config_json={}), _FakeBackend())
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
def test_resolve_agent_run_model_spec_strips_explicit_chat_model(monkeypatch: pytest.MonkeyPatch):
|
|
seen = []
|
|
|
|
def fake_get_model_info(spec):
|
|
seen.append(spec)
|
|
return SimpleNamespace(model_type="chat")
|
|
|
|
monkeypatch.setattr(agent_run_service.model_cache, "get_model_info", fake_get_model_info)
|
|
|
|
assert (
|
|
agent_run_service.resolve_agent_run_model_spec(
|
|
" gpt-x ",
|
|
SimpleNamespace(config_json={}),
|
|
_FakeBackend(),
|
|
)
|
|
== "gpt-x"
|
|
)
|
|
assert seen == ["gpt-x"]
|
|
|
|
|
|
def _patch_agent_run_creation(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
*,
|
|
agent_config_json: dict | None = None,
|
|
message_id: int = 10,
|
|
active_run: SimpleNamespace | None = None,
|
|
active_run_after_rollback: SimpleNamespace | None = None,
|
|
existing_run: SimpleNamespace | None = None,
|
|
existing_run_after_rollback: SimpleNamespace | None = None,
|
|
parent_run: SimpleNamespace | None = None,
|
|
raise_create_integrity_error: bool = False,
|
|
):
|
|
runs_by_id = {
|
|
"parent-agent-run": SimpleNamespace(
|
|
id="parent-agent-run",
|
|
conversation_id=99,
|
|
conversation_thread_id="parent-thread",
|
|
)
|
|
}
|
|
if parent_run:
|
|
runs_by_id["parent-run"] = parent_run
|
|
db = _CreateRunDb(
|
|
message_id=message_id,
|
|
active_run=active_run,
|
|
active_run_after_rollback=active_run_after_rollback,
|
|
existing_run=existing_run,
|
|
existing_run_after_rollback=existing_run_after_rollback,
|
|
runs_by_id=runs_by_id,
|
|
raise_create_integrity_error=raise_create_integrity_error,
|
|
)
|
|
|
|
class ConvRepo:
|
|
def __init__(self, db_session):
|
|
del db_session
|
|
|
|
async def get_conversation_by_thread_id(self, thread_id: str):
|
|
del thread_id
|
|
return SimpleNamespace(id=1, uid="user-1", status="active", agent_id="default")
|
|
|
|
class AgentRepo:
|
|
def __init__(self, db_session):
|
|
del db_session
|
|
|
|
async def get_visible_by_slug(self, *, slug: str, user, kind="main"):
|
|
del user
|
|
is_subagent = kind == "subagent"
|
|
return SimpleNamespace(
|
|
slug=slug,
|
|
name="Default",
|
|
backend_id="ChatbotAgent",
|
|
config_json=agent_config_json or {"context": {}},
|
|
is_subagent=is_subagent,
|
|
)
|
|
|
|
class Queue:
|
|
async def enqueue_job(self, job_name: str, run_id: str, _job_id: str):
|
|
assert db.committed is True
|
|
db.order.append("enqueue")
|
|
db.enqueued.append((job_name, run_id, _job_id))
|
|
|
|
async def fake_get_arq_pool():
|
|
return Queue()
|
|
|
|
monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: _FakeBackend())
|
|
monkeypatch.setattr(agent_run_service, "AgentRepository", AgentRepo)
|
|
monkeypatch.setattr(agent_run_service, "ConversationRepository", ConvRepo)
|
|
monkeypatch.setattr(agent_run_service, "AgentRunRepository", _CreateRunRepo)
|
|
monkeypatch.setattr(agent_run_service, "get_arq_pool", fake_get_arq_pool)
|
|
return db
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_chat_run_persists_validated_model_spec(monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr(
|
|
agent_run_service.model_cache,
|
|
"get_model_info",
|
|
lambda spec: SimpleNamespace(model_type="chat"),
|
|
)
|
|
db = _patch_agent_run_creation(monkeypatch)
|
|
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
model_spec="claude-x",
|
|
)
|
|
|
|
assert db.created_run_kwargs["input_payload"]["model_spec"] == "claude-x"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_chat_run_with_image_persists_multimodal_message_type(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(monkeypatch)
|
|
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("看图", "base64-image"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
)
|
|
|
|
assert db.created_run_kwargs["input_payload"] == {"model_spec": "agent-default-model"}
|
|
assert db.added[0].message_type == "multimodal_image"
|
|
assert db.added[0].image_content == "base64-image"
|
|
raw_message = db.added[0].extra_metadata["raw_message"]
|
|
assert raw_message["type"] == "human"
|
|
assert raw_message["content"][0] == {"type": "text", "text": "看图"}
|
|
assert raw_message["content"][1]["image_url"]["url"] == "data:image/jpeg;base64,base64-image"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_chat_run_snapshots_agent_configured_model_spec(monkeypatch: pytest.MonkeyPatch):
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
agent_config_json={"context": {"model": "agent-config-model"}},
|
|
)
|
|
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
model_spec=None,
|
|
)
|
|
|
|
assert db.created_run_kwargs["input_payload"]["model_spec"] == "agent-config-model"
|
|
assert "model_spec" not in db.added[0].extra_metadata
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_chat_run_snapshots_system_default_when_agent_model_empty(monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr(
|
|
agent_run_service,
|
|
"resolve_chat_model_spec",
|
|
lambda model_spec: str(model_spec).strip() if str(model_spec or "").strip() else "system-default-model",
|
|
)
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
agent_config_json={"context": {"model": ""}},
|
|
)
|
|
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=_chat_input("hello"),
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "req-1"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
model_spec=None,
|
|
)
|
|
|
|
assert db.created_run_kwargs["input_payload"]["model_spec"] == "system-default-model"
|
|
assert "model_spec" not in db.added[0].extra_metadata
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_resume_run_inherits_parent_model_spec(monkeypatch: pytest.MonkeyPatch):
|
|
# 即使 resume 入参传了别的模型，也必须沿用父运行的模型
|
|
db = _patch_agent_run_creation(
|
|
monkeypatch,
|
|
parent_run=SimpleNamespace(
|
|
id="parent-run",
|
|
conversation_thread_id="thread-1",
|
|
status="interrupted",
|
|
input_payload={"model_spec": "parent-model"},
|
|
),
|
|
)
|
|
|
|
await agent_run_service.create_agent_run_view(
|
|
input_message=None,
|
|
agent_slug="default",
|
|
thread_id="thread-1",
|
|
meta={"request_id": "resume-req"},
|
|
current_uid="user-1",
|
|
db=db,
|
|
model_spec="ignored-model",
|
|
resume={"language": "python"},
|
|
created_by_run_id="parent-run",
|
|
)
|
|
|
|
assert db.created_run_kwargs["input_payload"]["model_spec"] == "parent-model"
|
|
|
|
|
|
def test_compact_stream_chunk_retains_compression_field():
|
|
chunk = {
|
|
"request_id": "req-1",
|
|
"response": None,
|
|
"thread_id": "thread-1",
|
|
"status": "context_compression",
|
|
"compression": {"type": "yuxi.context_compression", "status": "started"},
|
|
"meta": {"uid": "user-1"},
|
|
}
|
|
|
|
compact = agent_run_service._compact_stream_chunk(chunk)
|
|
|
|
assert compact["status"] == "context_compression"
|
|
assert compact["compression"] == {"type": "yuxi.context_compression", "status": "started"}
|