steven_li
6e9e74d1ee
添加 RuntimeContext 类用于捕获模型运行时的日期时间信息， 包括UTC时间、本地时间和时区信息，并在系统提示中显示这些信息。 同时增加最大上下文消息数和工具迭代次数的配置选项， 将验证服务从引擎加载器中移除，并更新相关的数据结构和接口。 BREAKING CHANGE: 移除了验证服务，相关字段被替换为证据状态和接受状态。 - 添加 RuntimeContext 类和相关渲染方法 - 增加 max_context_messages 和 max_tool_iterations 配置 - 移除 ValidationService 相关代码 - 更新消息记录中的验证状态字段 - 添加原始工具调用检测和回退处理
292 lines
9.2 KiB
Python
292 lines
9.2 KiB
Python
import asyncio
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from beaver.foundation.events import InboundMessage, MessageBus
|
|
from beaver.interfaces.channels import ChannelManager, MemoryChannelAdapter
|
|
from beaver.interfaces.gateway.main import run_gateway
|
|
from beaver.services.agent_service import AgentService
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class FakeResult:
|
|
session_id: str
|
|
run_id: str = "run-1"
|
|
output_text: str = ""
|
|
finish_reason: str = "stop"
|
|
provider_name: str | None = "fake"
|
|
model: str | None = "fake-model"
|
|
usage: dict[str, Any] = field(default_factory=dict)
|
|
task_id: str | None = "task-1"
|
|
task_status: str | None = "awaiting_acceptance"
|
|
validation_result: dict[str, Any] | None = None
|
|
|
|
|
|
class FakeService:
|
|
is_running = True
|
|
|
|
async def submit_direct(self, message: str, **kwargs: Any) -> FakeResult:
|
|
return FakeResult(
|
|
session_id=kwargs.get("session_id") or "s1",
|
|
output_text=f"echo:{message}",
|
|
)
|
|
|
|
async def handle_inbound_message(self, inbound: InboundMessage):
|
|
result = await self.submit_direct(inbound.content, session_id=inbound.session_id)
|
|
return AgentService.build_outbound_message(inbound, result)
|
|
|
|
|
|
class SlowService:
|
|
is_running = True
|
|
|
|
async def submit_direct(self, message: str, **kwargs: Any) -> FakeResult:
|
|
await asyncio.sleep(10)
|
|
return FakeResult(session_id=kwargs.get("session_id") or "s1")
|
|
|
|
async def handle_inbound_message(self, inbound: InboundMessage):
|
|
result = await self.submit_direct(inbound.content, session_id=inbound.session_id)
|
|
return AgentService.build_outbound_message(inbound, result)
|
|
|
|
|
|
class InvalidService:
|
|
is_running = True
|
|
|
|
|
|
def test_gateway_routes_memory_channel_roundtrip() -> None:
|
|
async def run() -> None:
|
|
bus = MessageBus()
|
|
channel = MemoryChannelAdapter(bus)
|
|
stop_event = asyncio.Event()
|
|
task = asyncio.create_task(
|
|
run_gateway(
|
|
service=FakeService(),
|
|
manage_service_lifecycle=False,
|
|
bus=bus,
|
|
channels=[channel],
|
|
stop_event=stop_event,
|
|
)
|
|
)
|
|
|
|
await channel.publish_text("hello", session_id="s1")
|
|
for _ in range(40):
|
|
if channel.sent_messages:
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert channel.sent_messages
|
|
message = channel.sent_messages[0]
|
|
assert message.content == "echo:hello"
|
|
assert message.session_id == "s1"
|
|
assert message.finish_reason == "stop"
|
|
assert message.metadata["task_id"] == "task-1"
|
|
assert message.metadata["task_status"] == "awaiting_acceptance"
|
|
assert message.metadata["evidence_status"] == "recorded"
|
|
assert message.metadata["validation_result"] is None
|
|
|
|
stop_event.set()
|
|
await asyncio.wait_for(task, timeout=2)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_gateway_delivers_cancelled_outbound_to_channel() -> None:
|
|
async def run() -> None:
|
|
bus = MessageBus()
|
|
channel = MemoryChannelAdapter(bus)
|
|
stop_event = asyncio.Event()
|
|
task = asyncio.create_task(
|
|
run_gateway(
|
|
service=SlowService(),
|
|
manage_service_lifecycle=False,
|
|
bus=bus,
|
|
channels=[channel],
|
|
stop_event=stop_event,
|
|
)
|
|
)
|
|
|
|
await channel.publish_text("slow", session_id="s1")
|
|
await asyncio.sleep(0.05)
|
|
stop_event.set()
|
|
await asyncio.wait_for(task, timeout=3)
|
|
|
|
assert channel.sent_messages
|
|
assert channel.sent_messages[0].finish_reason == "cancelled"
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_gateway_rejects_channel_manager_and_channels_together() -> None:
|
|
async def run() -> None:
|
|
bus = MessageBus()
|
|
try:
|
|
await run_gateway(
|
|
service=FakeService(),
|
|
manage_service_lifecycle=False,
|
|
bus=bus,
|
|
channel_manager=ChannelManager(bus),
|
|
channels=[MemoryChannelAdapter(bus)],
|
|
stop_event=asyncio.Event(),
|
|
)
|
|
except ValueError as exc:
|
|
assert "either channel_manager or channels" in str(exc)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_gateway_fails_fast_for_service_without_handle_inbound_message() -> None:
|
|
async def run() -> None:
|
|
try:
|
|
await run_gateway(
|
|
service=InvalidService(),
|
|
manage_service_lifecycle=False,
|
|
bus=MessageBus(),
|
|
stop_event=asyncio.Event(),
|
|
)
|
|
except TypeError as exc:
|
|
assert "handle_inbound_message" in str(exc)
|
|
else:
|
|
raise AssertionError("expected TypeError")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_agent_service_maps_inbound_error_to_structured_outbound() -> None:
|
|
async def run() -> None:
|
|
service = AgentService()
|
|
|
|
async def failing_submit_direct(message: str, **kwargs: Any) -> FakeResult:
|
|
raise RuntimeError("boom")
|
|
|
|
service.submit_direct = failing_submit_direct # type: ignore[method-assign]
|
|
outbound = await service.handle_inbound_message(
|
|
InboundMessage(channel="memory", content="hello", session_id="s1", metadata={"source": "test"})
|
|
)
|
|
|
|
assert outbound.finish_reason == "error"
|
|
assert outbound.session_id == "s1"
|
|
assert outbound.metadata["error"] == "boom"
|
|
assert outbound.metadata["inbound_metadata"] == {"source": "test"}
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_agent_service_maps_stopped_runtime_to_stopped_outbound() -> None:
|
|
async def run() -> None:
|
|
service = AgentService()
|
|
|
|
async def stopped_submit_direct(message: str, **kwargs: Any) -> FakeResult:
|
|
raise RuntimeError("AgentLoop.submit_direct() is not accepting new tasks after stop()")
|
|
|
|
service.submit_direct = stopped_submit_direct # type: ignore[method-assign]
|
|
outbound = await service.handle_inbound_message(
|
|
InboundMessage(channel="memory", content="hello", session_id="s1")
|
|
)
|
|
|
|
assert outbound.finish_reason == "stopped"
|
|
assert "not accepting new tasks" in outbound.metadata["error"]
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_channel_manager_keeps_unknown_channel_outbound_undeliverable() -> None:
|
|
async def run() -> None:
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
stop_event = asyncio.Event()
|
|
await bus.publish_outbound(
|
|
AgentService.build_outbound_message(
|
|
InboundMessage(channel="missing", content="hello", session_id="missing:1"),
|
|
FakeResult(session_id="missing:1", output_text="ok"),
|
|
)
|
|
)
|
|
stop_event.set()
|
|
|
|
await manager.dispatch_outbound(stop_event)
|
|
|
|
assert len(manager.undeliverable) == 1
|
|
assert manager.undeliverable[0].channel == "missing"
|
|
assert manager.undeliverable[0].session_id == "missing:1"
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_memory_channel_adapts_old_style_payload_to_stable_session_id() -> None:
|
|
async def run() -> None:
|
|
bus = MessageBus()
|
|
channel = MemoryChannelAdapter(bus, name="telegram")
|
|
inbound = await channel.publish_external_text(
|
|
"hello",
|
|
chat_id="chat-1",
|
|
message_id="message-1",
|
|
raw_payload={"platform": "telegram", "text": "hello"},
|
|
)
|
|
|
|
queued = await bus.consume_inbound()
|
|
assert queued is inbound
|
|
assert queued.channel == "telegram"
|
|
assert queued.session_id == "telegram:chat-1"
|
|
assert queued.metadata["chat_id"] == "chat-1"
|
|
assert queued.metadata["message_id"] == "message-1"
|
|
assert queued.metadata["raw_channel_payload"] == {"platform": "telegram", "text": "hello"}
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_channel_manager_start_cancellation_rolls_back_started_channels() -> None:
|
|
class StartedChannel:
|
|
name = "started"
|
|
|
|
def __init__(self, bus: MessageBus) -> None:
|
|
self.bus = bus
|
|
self.stopped = False
|
|
|
|
async def start(self) -> None:
|
|
pass
|
|
|
|
async def stop(self) -> None:
|
|
self.stopped = True
|
|
|
|
async def send(self, message: Any) -> None:
|
|
pass
|
|
|
|
class BlockingChannel:
|
|
name = "blocking"
|
|
|
|
def __init__(self, bus: MessageBus) -> None:
|
|
self.bus = bus
|
|
self.entered = asyncio.Event()
|
|
|
|
async def start(self) -> None:
|
|
self.entered.set()
|
|
await asyncio.sleep(10)
|
|
|
|
async def stop(self) -> None:
|
|
pass
|
|
|
|
async def send(self, message: Any) -> None:
|
|
pass
|
|
|
|
async def run() -> None:
|
|
bus = MessageBus()
|
|
started = StartedChannel(bus)
|
|
blocking = BlockingChannel(bus)
|
|
manager = ChannelManager(bus)
|
|
manager.register(started)
|
|
manager.register(blocking)
|
|
|
|
task = asyncio.create_task(manager.start())
|
|
await blocking.entered.wait()
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
else:
|
|
raise AssertionError("expected cancellation")
|
|
|
|
assert started.stopped
|
|
|
|
asyncio.run(run())
|