fix(acp): follow-up — named-const page size, alias kwarg, tests

- Replace kwargs.get('limit', 50) with module-level _LIST_SESSIONS_PAGE_SIZE
  constant. ListSessionsRequest schema has no 'limit' field, so the kwarg
  path was dead. Constant is the single source of truth for the page cap.
- Use next_cursor= (field name) instead of nextCursor= (alias). Both work
  under the schema's populate_by_name config, but using the declared
  Python field name is the consistent style in this file.
- Add docstring explaining cwd pass-through and cursor semantics.
- Add 4 tests: first-page with next_cursor, single-page no next_cursor,
  cursor resumes after match, unknown cursor returns empty page.
This commit is contained in:
Teknium
2026-04-21 05:59:19 -07:00
committed by Teknium
parent c1fb7b6d27
commit 4cc5065f63
2 changed files with 69 additions and 8 deletions

View File

@@ -270,6 +270,57 @@ class TestListAndFork:
mock_list.assert_called_once_with(cwd="/mnt/e/Projects/AI/browser-link-3")
@pytest.mark.asyncio
async def test_list_sessions_pagination_first_page(self, agent):
from acp_adapter import server as acp_server
infos = [
{"session_id": f"s{i}", "cwd": "/tmp", "title": None, "updated_at": 0.0}
for i in range(acp_server._LIST_SESSIONS_PAGE_SIZE + 5)
]
with patch.object(agent.session_manager, "list_sessions", return_value=infos):
resp = await agent.list_sessions()
assert len(resp.sessions) == acp_server._LIST_SESSIONS_PAGE_SIZE
assert resp.next_cursor == resp.sessions[-1].session_id
@pytest.mark.asyncio
async def test_list_sessions_pagination_no_more(self, agent):
infos = [
{"session_id": f"s{i}", "cwd": "/tmp", "title": None, "updated_at": 0.0}
for i in range(3)
]
with patch.object(agent.session_manager, "list_sessions", return_value=infos):
resp = await agent.list_sessions()
assert len(resp.sessions) == 3
assert resp.next_cursor is None
@pytest.mark.asyncio
async def test_list_sessions_cursor_resumes_after_match(self, agent):
infos = [
{"session_id": "s1", "cwd": "/tmp", "title": None, "updated_at": 0.0},
{"session_id": "s2", "cwd": "/tmp", "title": None, "updated_at": 0.0},
{"session_id": "s3", "cwd": "/tmp", "title": None, "updated_at": 0.0},
]
with patch.object(agent.session_manager, "list_sessions", return_value=infos):
resp = await agent.list_sessions(cursor="s1")
assert [s.session_id for s in resp.sessions] == ["s2", "s3"]
assert resp.next_cursor is None
@pytest.mark.asyncio
async def test_list_sessions_unknown_cursor_returns_empty(self, agent):
infos = [
{"session_id": "s1", "cwd": "/tmp", "title": None, "updated_at": 0.0},
{"session_id": "s2", "cwd": "/tmp", "title": None, "updated_at": 0.0},
]
with patch.object(agent.session_manager, "list_sessions", return_value=infos):
resp = await agent.list_sessions(cursor="does-not-exist")
assert resp.sessions == []
assert resp.next_cursor is None
# ---------------------------------------------------------------------------
# session configuration / model routing
# ---------------------------------------------------------------------------