import asyncio from types import SimpleNamespace import pytest import device_gateway.app as gateway from device_gateway.control.approvals import ApprovalRegistry from device_gateway.control.catalog import Catalog, LIVE_PROJECT_ID from device_gateway.control.models import Project from device_gateway.control.navigator import Navigator from device_gateway.control.operations import Operations from device_gateway.control.pi import PiAdapter from device_gateway.control.projects import ProjectRegistry from device_gateway.integrations.pi import PiSessionEvent, PiSessionRegistry from device_gateway.state import TargetStore @pytest.mark.parametrize('heartbeat', ['session_info_changed', 'session_start']) def test_running_native_pi_survives_restart_heartbeat_catalog_and_selection(tmp_path, monkeypatch, heartbeat): async def run(): projects = ProjectRegistry(tmp_path / 'projects.json') project = projects.add(Project(id='p', name='Project', path=str(tmp_path))) live = PiSessionRegistry() # Gateway restarted during an existing turn, so agent_start was missed. live.apply(PiSessionEvent(event=heartbeat, session_id='long-task', name='Long task', cwd=project.path)) assert live.get('long-task').state == 'IDLE' class RPC: state = 'RUNNING' async def call(self, method, params): assert method == 'list' and params['cwd'] == project.path return [dict(session_id='long-task', name='Long task', cwd=project.path, updated_at='2026-10-07T00:00:00Z', live=True, state=self.state)] rpc = RPC() adapter = PiAdapter(projects, live, rpc) monkeypatch.setattr(gateway, 'pi_sessions', live) monkeypatch.setattr(gateway, 'adapters', {'pi': adapter}) monkeypatch.setattr(gateway, 'approvals', ApprovalRegistry()) catalog = Catalog(projects, {'pi': adapter}) page = await catalog.page('c', 'sessions', LIVE_PROJECT_ID) assert page['status'] == 'Ready' row, = page['items'] assert (row['state'], row['project_id'], row['session_id']) == ('RUNNING', 'p', 'long-task') messages = [] async def send(device, payload): messages.append(payload) 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, {'pi': adapter}, targets, Operations(tmp_path / 'ops.sqlite'), gateway.user_inputs, send, lambda d: True, gateway.agent_target_payload, lambda d: None) await nav.select(device, dict(request_id='select', expected_target_generation=0, project_id=row['project_id'], agent=row['agent'], session_id=row['session_id'])) assert [m['type'] for m in messages] == ['target_selected', 'agent_state'] assert messages[-1]['state'] == 'RUNNING' and messages[-1]['target'] == 'agent:pi:long-task' assert targets.selection('d')['project_id'] == 'p' # Loaded but idle is still excluded; a cached snapshot stays immutable. rpc.state = 'IDLE' assert (await catalog.page('c', 'sessions', LIVE_PROJECT_ID))['status'] == 'Empty' assert (await catalog.page('c', 'sessions', LIVE_PROJECT_ID, snapshot=page['snapshot']))['items'][0]['state'] == 'RUNNING' assert (await gateway.agent_target_payload('agent:pi:long-task'))['state'] == 'IDLE' asyncio.run(run()) @pytest.mark.parametrize('terminal,expected', [(None, 'RUNNING'), (False, 'DONE'), (True, 'FAILED')]) def test_known_pi_lifecycle_still_overlays_native_idle(tmp_path, monkeypatch, terminal, expected): async def run(): projects = ProjectRegistry(tmp_path / 'projects.json') project = projects.add(Project(id='p', name='Project', path=str(tmp_path))) live = PiSessionRegistry() values = dict(session_id='task', name='Task', cwd=project.path, turn_id='turn') live.apply(PiSessionEvent(event='agent_start', **values)) if terminal is not None: live.apply(PiSessionEvent(event='agent_end', failed=terminal, final_text='Final answer', **values)) class RPC: async def call(self, method, params): assert method == 'list' return [dict(session_id='task', name='Task', cwd=project.path, updated_at='2026-10-07T00:00:00Z', live=True, state='IDLE')] adapter = PiAdapter(projects, live, RPC()) monkeypatch.setattr(gateway, 'pi_sessions', live) monkeypatch.setattr(gateway, 'adapters', {'pi': adapter}) monkeypatch.setattr(gateway, 'approvals', ApprovalRegistry()) assert (await adapter.get_session('task')).state == expected result = await gateway.agent_target_payload('agent:pi:task') assert result['state'] == expected if terminal is not None: assert result['final_text'] == 'Final answer' asyncio.run(run()) def test_restarted_pi_elapsed_keeps_advancing_across_refreshes(tmp_path, monkeypatch): async def run(): projects = ProjectRegistry(tmp_path / 'projects.json') project = projects.add(Project(id='p', name='Project', path=str(tmp_path))) live = PiSessionRegistry() values = dict(session_id='long-task', name='Long task', cwd=project.path) live.apply(PiSessionEvent(event='session_info_changed', **values)) class RPC: state = 'RUNNING' async def call(self, method, params): assert method == 'list' return [dict(**values, updated_at='2026-10-07T00:00:00Z', live=True, state=self.state)] rpc = RPC();adapter = PiAdapter(projects, live, rpc) now = [100.0];adapter.clock = lambda: now[0] monkeypatch.setattr(gateway, 'pi_sessions', live) monkeypatch.setattr(gateway, 'adapters', {'pi': adapter}) monkeypatch.setattr(gateway, 'approvals', ApprovalRegistry()) for elapsed in (0, 2, 5, 37): now[0] = 100.0 + elapsed live.apply(PiSessionEvent(event='session_info_changed', **values)) payload = await gateway.agent_target_payload('agent:pi:long-task') assert payload['state'] == 'RUNNING' and payload['elapsed_seconds'] == elapsed assert (await adapter.get_session('long-task')).elapsed_seconds == elapsed # An idle observation ends that inferred run; a later run starts fresh. rpc.state = 'IDLE';now[0] = 140 assert (await gateway.agent_target_payload('agent:pi:long-task'))['elapsed_seconds'] == 0 rpc.state = 'RUNNING';now[0] = 150 assert (await gateway.agent_target_payload('agent:pi:long-task'))['elapsed_seconds'] == 0 now[0] = 156 assert (await gateway.agent_target_payload('agent:pi:long-task'))['elapsed_seconds'] == 6 asyncio.run(run()) def test_original_pi_turn_clock_and_terminal_elapsed_remain_authoritative(tmp_path, monkeypatch): from datetime import datetime, timedelta, timezone async def run(): projects = ProjectRegistry(tmp_path / 'projects.json') project = projects.add(Project(id='p', name='Project', path=str(tmp_path))) live = PiSessionRegistry() values = dict(session_id='task', name='Task', cwd=project.path, turn_id='turn') started = datetime.now(timezone.utc) - timedelta(seconds=120) live.apply(PiSessionEvent(event='agent_start', timestamp=started, **values)) class RPC: async def call(self, method, params): assert method == 'list' return [dict(session_id='task', name='Task', cwd=project.path, updated_at='2026-10-07T00:00:00Z', live=True, state='RUNNING')] adapter = PiAdapter(projects, live, RPC()) monkeypatch.setattr(gateway, 'pi_sessions', live) monkeypatch.setattr(gateway, 'adapters', {'pi': adapter}) monkeypatch.setattr(gateway, 'approvals', ApprovalRegistry()) assert 120 <= (await adapter.get_session('task')).elapsed_seconds <= 122 assert 120 <= (await gateway.agent_target_payload('agent:pi:task'))['elapsed_seconds'] <= 122 live.apply(PiSessionEvent(event='agent_end', timestamp=started+timedelta(seconds=123), **values)) assert (await adapter.get_session('task')).elapsed_seconds == 123 payload = await gateway.agent_target_payload('agent:pi:task') assert payload['state'] == 'DONE' and payload['elapsed_seconds'] == 123 asyncio.run(run())