from __future__ import annotations
import importlib
import unittest
from unittest.mock import AsyncMock, patch
from agent import sessions
session_module = importlib.import_module("agent.sessions")
SESSION = {
"activeSessionId": "active-root",
"sessionId": "session-root",
"sessionName": "root",
"cwd": "/repo",
"status": "idle",
"isCurrent": True,
"isStreaming": False,
"isCompacting": False,
"attachedClients": 1,
"messageCount": 2,
"queuedCount": 0,
"isSessionActive": False,
"capabilities": ["send", "recent_turns", "stop", "restart"],
}
class AgentSessionsTest(unittest.IsolatedAsyncioTestCase):
async def test_namespace_and_handle_route_typed_operations(self) -> None:
calls: list[tuple[str, dict[str, object]]] = []
async def request(request_type: str, payload: dict[str, object] | None = None) -> dict[str, object]:
calls.append((request_type, payload or {}))
if request_type == "agent.sessions.list":
return {"sessions": [SESSION]}
if request_type == "agent.sessions.get":
return {"session": SESSION}
if request_type == "agent.sessions.recent_turns":
return {
"agent": SESSION,
"turns": [
{
"user": "question",
"assistant": "answer",
"toolCalls": 1,
"startedAt": 10,
"endedAt": 20,
"messageCount": 3,
"truncated": False,
}
],
"count": 3,
"maxChars": 6000,
"truncated": False,
}
if request_type == "agent.sessions.send":
return {
"operationId": "op-send",
"status": "queued",
"delivery": "follow_up",
"target": {"activeSessionId": "active-root", "sessionId": "session-root"},
}
if request_type == "agent.sessions.stop":
return {"operationId": "op-stop", "status": "accepted"}
raise AssertionError(request_type)
with patch.object(session_module, "host_request", new=AsyncMock(side_effect=request)):
listed = await sessions.list()
handle = await sessions.get(active_session_id="active-root")
recent = await handle.recent_turns()
receipt = await handle.send("continue")
stopped = await handle.stop()
self.assertEqual([entry.active_session_id for entry in listed], ["active-root"])
self.assertEqual(handle.name, "root")
self.assertEqual(recent.turns[0].tool_calls, 1)
self.assertEqual(receipt.operation_id, "op-send")
self.assertEqual(stopped.operation_id, "op-stop")
self.assertEqual(handle.status, "stopped")
self.assertEqual(
[request_type for request_type, _ in calls],
[
"agent.sessions.list",
"agent.sessions.get",
"agent.sessions.recent_turns",
"agent.sessions.send",
"agent.sessions.stop",
],
)
self.assertEqual(calls[2][1], {"target": "active-root", "count": 3, "max_chars": 6000})
async def test_missing_capabilities_fail_closed(self) -> None:
async def request(request_type: str, payload: dict[str, object] | None = None) -> dict[str, object]:
self.assertEqual(request_type, "agent.sessions.get")
return {"session": {"activeSessionId": "active-root", "sessionId": "session-root"}}
with patch.object(session_module, "host_request", new=AsyncMock(side_effect=request)):
handle = await sessions.get(active_session_id="active-root")
self.assertEqual(handle.capabilities, ())
for operation in (
handle.send("message"),
handle.recent_turns(),
handle.stop(),
handle.restart(),
):
with self.assertRaises(RuntimeError):
await operation
async def test_get_rejects_multiple_selectors(self) -> None:
with self.assertRaises(TypeError):
await sessions.get("active-root", active_session_id="also-root")
async def test_handle_validates_send_arguments(self) -> None:
handle = session_module.AgentSessionHandle(
"active-root",
"session-root",
capabilities=("send", "recent_turns", "stop", "restart"),
)
with self.assertRaises(TypeError):
await handle.send("")
with self.assertRaises(ValueError):
await handle.send("message", "invalid") # type: ignore[arg-type]
if __name__ == "__main__":
unittest.main()