import asyncio import json from types import SimpleNamespace import pytest import device_gateway.control.catalog as catalog_module from device_gateway.control.catalog import Catalog, LIVE_PROJECT_ID, LIVE_QUERY_CONCURRENCY from device_gateway.control.models import AgentSession, ControlError, Project from device_gateway.control.navigator import Navigator from device_gateway.control.operations import Operations from device_gateway.control.projects import ProjectRegistry from device_gateway.control.user_input import UserInputs from device_gateway.state import TargetStore def registry(tmp_path, count=3): result = ProjectRegistry(tmp_path / 'projects.json') for i in range(count): path = tmp_path / f'p{i}' path.mkdir() result.add(Project(id=f'p{i}', name=f'Project {i}', path=str(path))) return result def session(projects, project_id, sid, state='RUNNING', agent='pi', updated='2026-10-07T00:00:00Z', name='Work'): return AgentSession(project_id=project_id, agent=agent, session_id=sid, name=name, state=state, updated_at=updated, cwd=projects.get(project_id).path) class Adapter: def __init__(self, rows=None, failures=None): self.rows, self.failures = rows or {}, failures or {} self.calls, self.creations = [], 0 async def list_sessions(self, project): self.calls.append(project.id) if project.id in self.failures: raise self.failures[project.id] return self.rows.get(project.id, []) async def get_session(self, sid): return next(s for rows in self.rows.values() for s in rows if s.session_id == sid) async def permissions(self, sid): return dict(current='Native', available_modes=[]) async def create_session(self, *args, **kwargs): self.creations += 1 raise AssertionError('Virtual project must not create a session') def test_projects_virtual_first_real_preserved_not_persisted(tmp_path): projects = registry(tmp_path, 5) original = projects.path.read_bytes() async def run(): catalog = Catalog(projects, {}) page = await catalog.page('c', 'projects') assert page['items'][0] == {'project_id': LIVE_PROJECT_ID, 'name': 'Live Sessions'} rows = list(page['items']) page = await catalog.page('c', 'projects', snapshot=page['snapshot'], cursor=4) rows += page['items'] assert [row['project_id'] for row in rows[1:]] == [p.id for p in projects.list()] assert all(set(row) == {'project_id', 'name'} for row in rows) assert projects.path.read_bytes() == original with pytest.raises(ControlError, match='project not found'): projects.get(LIVE_PROJECT_ID) asyncio.run(run()) def test_active_filter_ignores_loaded_idle_and_preserves_original_name(tmp_path): projects = registry(tmp_path) states = ['RUNNING', 'WAITING_APPROVAL', 'WAITING_INPUT', 'IDLE', 'DORMANT', 'DONE', 'FAILED', 'OFFLINE'] rows = [session(projects, 'p0', state, state, name='中文原名') for state in states] for row in rows: row.live = row.state not in ('RUNNING', 'WAITING_INPUT', 'WAITING_APPROVAL') original = [row.model_dump() for row in rows] async def run(): page = await Catalog(projects, {'pi': Adapter({'p0': rows})}).page('c', 'sessions', LIVE_PROJECT_ID) assert [row['state'] for row in page['items']] == ['WAITING_APPROVAL', 'WAITING_INPUT', 'RUNNING'] assert page['project_id'] == LIVE_PROJECT_ID assert all(row['project_id'] == 'p0' and row['name'] == 'Project 0 / 中文原名' for row in page['items']) assert [row.model_dump() for row in rows] == original asyncio.run(run()) def test_cross_project_priority_time_and_stable_identity_ties(tmp_path): projects = registry(tmp_path) pi = Adapter({'p0': [session(projects, 'p0', 'run', updated='2027-01-01T00:00:00Z'), session(projects, 'p0', 'tie')], 'p2': [session(projects, 'p2', 'ignored', 'IDLE')]}) codex = Adapter({'p1': [session(projects, 'p1', 'approval', 'WAITING_APPROVAL', 'codex', '2020-01-01T00:00:00Z'), session(projects, 'p1', 'old', 'WAITING_INPUT', 'codex', '2026-10-06T00:00:00Z'), session(projects, 'p1', 'new', 'WAITING_INPUT', 'codex', '2026-10-07T09:00:00+08:00'), session(projects, 'p1', 'aaa', agent='codex'), session(projects, 'p1', 'tie', agent='codex')]}) async def run(): c = Catalog(projects, {'pi': pi, 'codex': codex}) page = await c.page('c', 'sessions', LIVE_PROJECT_ID) rows = list(page['items']) rows += (await c.page('c', 'sessions', LIVE_PROJECT_ID, page['snapshot'], 4))['items'] assert [(r['agent'], r['session_id']) for r in rows] == [ ('codex', 'approval'), ('codex', 'new'), ('codex', 'old'), ('pi', 'run'), ('codex', 'aaa'), ('codex', 'tie'), ('pi', 'tie')] assert set(pi.calls) == set(codex.calls) == {'p0', 'p1', 'p2'} asyncio.run(run()) @pytest.mark.parametrize('left,right,expected', [ (('RUNNING', '2027-01-01T00:00:00Z', 'rich'), ('WAITING_APPROVAL', '2020-01-01T00:00:00Z', ''), 'p1'), (('RUNNING', '2020-01-01T00:00:00Z', 'rich'), ('RUNNING', '2027-01-01T00:00:00Z', ''), 'p1'), (('RUNNING', '2026-10-07T00:00:00Z', ''), ('RUNNING', '2026-10-07T00:00:00Z', 'rich'), 'p1'), ]) def test_duplicate_prefers_priority_then_time_then_completeness(tmp_path, left, right, expected): projects = registry(tmp_path) def row(pid, values): state, date, name = values return session(projects, pid, 'duplicate', state, updated=date, name=name) pi = Adapter({'p0': [row('p0', left)], 'p1': [row('p1', right)]}) async def run(): page = await Catalog(projects, {'pi': pi}).page('c', 'sessions', LIVE_PROJECT_ID) assert len(page['items']) == 1 and page['items'][0]['project_id'] == expected asyncio.run(run()) def test_partial_failure_sanitizes_error_and_keeps_other_project(tmp_path): projects = registry(tmp_path) a = Adapter({'p0': [session(projects, 'p0', 'good')]}, {'p1': RuntimeError('/private/path token=private-value')}) async def run(): page = await Catalog(projects, {'pi': a}).page('c', 'sessions', LIVE_PROJECT_ID) assert page['status'] == 'Partial' and len(page['items']) == 1 assert page['errors'] == {'pi:p1': 'Adapter unavailable'} assert '/private' not in json.dumps(page) and 'private-value' not in json.dumps(page) asyncio.run(run()) def test_timeout_is_per_query_and_other_results_survive(tmp_path, monkeypatch): projects = registry(tmp_path) original_wait = asyncio.wait_for timeouts = [] async def short_wait(coro, timeout): timeouts.append(timeout) return await original_wait(coro, .02) monkeypatch.setattr(catalog_module.asyncio, 'wait_for', short_wait) class Slow(Adapter): async def list_sessions(self, project): if project.id == 'p1': await asyncio.sleep(60) return await super().list_sessions(project) async def run(): page = await Catalog(projects, {'pi': Slow({'p0': [session(projects, 'p0', 'good')]})}).page('c', 'sessions', LIVE_PROJECT_ID) assert page['status'] == 'Partial' and page['errors'] == {'pi:p1': 'Adapter unavailable'} assert page['items'][0]['session_id'] == 'good' and timeouts == [15, 15, 15] asyncio.run(run()) def test_queries_are_concurrent_and_bounded(tmp_path): projects = registry(tmp_path, 12) running, maximum, started = 0, 0, 0 gate = None class Batch(Adapter): async def list_sessions(self, project): nonlocal running, maximum, started running += 1 maximum = max(maximum, running) started += 1 if started == LIVE_QUERY_CONCURRENCY: gate.set() try: await gate.wait() await asyncio.sleep(0) return [] finally: running -= 1 async def run(): nonlocal gate gate = asyncio.Event() catalog = Catalog(projects, {'pi': Batch(), 'codex': Batch()}) pages = await asyncio.wait_for(asyncio.gather( catalog.page('c1', 'sessions', LIVE_PROJECT_ID), catalog.page('c2', 'sessions', LIVE_PROJECT_ID)), 1) assert all(page['status'] == 'Empty' and page['items'] == [] for page in pages) assert maximum == LIVE_QUERY_CONCURRENCY and started == 48 asyncio.run(run()) @pytest.mark.parametrize('count', [0, 3]) def test_empty_never_fills_done_or_creates_session(tmp_path, count): projects = registry(tmp_path, count) a = Adapter({'p0': [session(projects, 'p0', 'done', 'DONE')]} if count else {}) async def run(): page = await Catalog(projects, {'pi': a}).page('c', 'sessions', LIVE_PROJECT_ID) assert page['status'] == 'Empty' and page['items'] == [] and a.creations == 0 asyncio.run(run()) def test_snapshot_stable_under_state_change_connection_cursor_and_ttl(tmp_path): projects = registry(tmp_path) rows = [session(projects, 'p0', str(i), updated=f'2026-10-07T00:00:{i:02}Z') for i in range(9)] now = [0] async def run(): c = Catalog(projects, {'pi': Adapter({'p0': rows})}, clock=lambda: now[0]) first = await c.page('c', 'sessions', LIVE_PROJECT_ID) for row in rows: row.state = 'DONE' second = await c.page('c', 'sessions', LIVE_PROJECT_ID, first['snapshot'], 4) third = await c.page('c', 'sessions', LIVE_PROJECT_ID, first['snapshot'], 8) assert [r['session_id'] for page in [first, second, third] for r in page['items']] == list(map(str, range(8, -1, -1))) assert all(r['state'] == 'RUNNING' for r in second['items']) assert (await c.page('c', 'sessions', LIVE_PROJECT_ID))['status'] == 'Empty' for connection, kind, pid, cursor in [('other', 'sessions', LIVE_PROJECT_ID, 4), ('c', 'projects', LIVE_PROJECT_ID, 4), ('c', 'sessions', 'p0', 4), ('c', 'sessions', LIVE_PROJECT_ID, 1)]: with pytest.raises(ControlError, match='Stale'): await c.page(connection, kind, pid, first['snapshot'], cursor) now[0] = 60 with pytest.raises(ControlError, match='Expired'): await c.page('c', 'sessions', LIVE_PROJECT_ID, first['snapshot'], 4) asyncio.run(run()) def test_utf8_frame_and_security_projection(tmp_path): projects = registry(tmp_path) rows = [session(projects, 'p0', f's{i}', name='中文🙂' * 200) for i in range(4)] for row in rows: row.model = 'provider-secret' row.recent_text = row.final_text = row.latest_activity = 'private-content' async def run(): page = await Catalog(projects, {'pi': Adapter({'p0': rows})}).page('c', 'sessions', LIVE_PROJECT_ID) assert len(json.dumps(page, ensure_ascii=False).encode()) < 3900 for row in page['items']: assert set(row) == {'project_id', 'agent', 'session_id', 'name', 'state', 'updated_at'} assert len(row['name'].encode()) <= 240 and '\ufffd' not in row['name'] assert 'private-content' not in json.dumps(page) and 'provider-secret' not in json.dumps(page) asyncio.run(run()) def make_navigator(projects, adapters, tmp_path): messages = [] async def send(device, payload): messages.append(payload) async def target_payload(target): return dict(type='agent_state', target=target) device = SimpleNamespace(device_id='d', connection_id='c', send_lock=asyncio.Lock(), websocket=SimpleNamespace(send_json=lambda payload: send(None, payload))) targets = TargetStore(tmp_path / 'targets.json') nav = Navigator(projects, adapters, targets, Operations(tmp_path / 'ops.sqlite'), UserInputs(), send, lambda d: True, target_payload, lambda d: None) return nav, device, targets, messages @pytest.mark.parametrize('agent', ['pi', 'codex']) def test_live_row_selects_through_existing_real_project_cwd_checks(tmp_path, agent): projects = registry(tmp_path) a = Adapter({'p1': [session(projects, 'p1', 'real', agent=agent)]}) async def run(): n, d, targets, messages = make_navigator(projects, {agent: a}, tmp_path) page = await n.catalog.page('c', 'sessions', LIVE_PROJECT_ID) row = page['items'][0] result = await n.select(d, dict(request_id='r', expected_target_generation=0, project_id=row['project_id'], agent=row['agent'], session_id=row['session_id'])) assert result['project_id'] == 'p1' and targets.get('d') == f'agent:{agent}:real' assert [m['type'] for m in messages] == ['target_selected', 'agent_state'] assert a.creations == 0 asyncio.run(run()) @pytest.mark.parametrize('wrong', ['project_id', 'cwd']) def test_live_navigation_cannot_bypass_project_identity(tmp_path, wrong): projects = registry(tmp_path) row = session(projects, 'p1', 'real') a = Adapter({'p1': [row]}) async def run(): n, d, targets, messages = make_navigator(projects, {'pi': a}, tmp_path) page = await n.catalog.page('c', 'sessions', LIVE_PROJECT_ID) selected = page['items'][0] setattr(row, wrong, 'p0' if wrong == 'project_id' else projects.get('p0').path) with pytest.raises(ControlError, match='Session disappeared'): await n.select(d, dict(expected_target_generation=0, **{k: selected[k] for k in ['project_id', 'agent', 'session_id']})) assert targets.generation('d') == 0 and messages == [] asyncio.run(run()) def test_virtual_creation_is_controlled_rejected_and_not_journaled(tmp_path): projects = registry(tmp_path) a = Adapter() original = projects.path.read_bytes() async def run(): n, d, targets, messages = make_navigator(projects, {'pi': a}, tmp_path) event = dict(project_id=LIVE_PROJECT_ID, agent='pi', operation_id='op', expected_target_generation=0, settings={}) for action in [lambda: n.creation_options(d, event), lambda: n.create(d, event, 'unused prompt')]: with pytest.raises(ControlError, match='project not found') as error: await action() assert error.value.status == 404 assert a.creations == 0 and targets.generation('d') == 0 and messages == [] assert projects.path.read_bytes() == original asyncio.run(run()) def test_same_instant_with_different_timezone_uses_stable_identity_tie(tmp_path): projects = registry(tmp_path) pi = Adapter({'p0': [session(projects, 'p0', 'same', updated='2026-10-07T09:00:00+08:00')]}) codex = Adapter({'p1': [session(projects, 'p1', 'same', agent='codex', updated='2026-10-07T01:00:00Z')]}) async def run(): page = await Catalog(projects, {'pi': pi, 'codex': codex}).page('c', 'sessions', LIVE_PROJECT_ID) assert [row['agent'] for row in page['items']] == ['codex', 'pi'] asyncio.run(run()) def test_mismatched_project_or_adapter_identity_is_not_advertised(tmp_path): projects = registry(tmp_path) a = Adapter({'p0': [session(projects, 'p1', 'wrong-project'), session(projects, 'p0', 'wrong-agent', agent='codex'), session(projects, 'p0', 'good')]}) async def run(): page = await Catalog(projects, {'pi': a}).page('c', 'sessions', LIVE_PROJECT_ID) assert [row['session_id'] for row in page['items']] == ['good'] asyncio.run(run())