import asyncio import json import pytest from fastapi.testclient import TestClient import device_gateway.app as gateway from device_gateway.auth import DeviceCredentialStore from device_gateway.control.codex import CodexAdapter from device_gateway.control.models import Project from device_gateway.control.pi import PiAdapter from device_gateway.control.projects import ProjectRegistry from device_gateway.integrations.pi import PiSessionRegistry from device_gateway.speech import SpeechPreferences from device_gateway.state import TargetStore from fake_device import FakeDevice @pytest.fixture def downlink(tmp_path, monkeypatch): credentials = tmp_path/'credentials.json' credentials.write_text(json.dumps(dict(device_id='downlink-test', device_token='test-device-only', pi_token='test-pi-only'))) registry = PiSessionRegistry() projects = ProjectRegistry(tmp_path/'projects.json') projects.add(Project(id='p', name='Test', path=str(tmp_path))) class RPC: inputs = [] async def call(self, method, params): if method == 'list': return [dict(session_id='s', name='Test', cwd=str(tmp_path), state='IDLE', updated_at='2026-10-03T00:00:00Z', live=True)] assert method == 'input' self.inputs.append(params['text']) return dict(accepted=True, session_id='s', resumed=False) async def close(self): pass adapter = PiAdapter(projects, registry, rpc=RPC()) for key, value in dict(credential_store=DeviceCredentialStore(credentials), devices={}, target_store=TargetStore(tmp_path/'targets.json'), pi_sessions=registry, project_registry=projects, adapters={'pi': adapter}).items(): monkeypatch.setattr(gateway, key, value) monkeypatch.setattr(gateway.speeches, 'provider', None) monkeypatch.setattr(gateway.speeches, 'preferences', SpeechPreferences(tmp_path/'speech.json')) with TestClient(gateway.app) as client, client.websocket_connect('/device/ws') as ws: device = FakeDevice(ws) response = client.post('/devices/downlink-test/target', json={'target':'pi:s'}) assert response.status_code == 200 and response.json()['target'] == 'agent:pi:s' device.wait('agent_state', session_id='s') def post(kind, **extra): response = client.post('/integrations/pi/events', headers={'x-device-lab-pi-token':'test-pi-only'}, json=dict(event=kind, session_id='s', name='Test', cwd=str(tmp_path), **{'turn_id':'t', **extra})) assert response.status_code == 200 return response.json() yield client, device, post, adapter def test_direct_answer_without_message_updates(downlink): client, device, post, adapter = downlink response = client.post('/api/agents/pi/sessions/s/input', json={'text':'哈喽哈喽'}) assert response.status_code == 202 and adapter.rpc.inputs == ['哈喽哈喽'] post('agent_start') assert device.wait('agent_state', state='RUNNING')['final_text'] == '' post('agent_end', final_text='你好！有什么可以帮你？') end = device.wait('agent_state', state='DONE') assert end['final_text'] == '你好！有什么可以帮你？' assert end['recent_text'] == '' and end['notification'] == 'done' # Reconnect/target refresh must carry the same authoritative result. assert client.post('/devices/downlink-test/target', json={'target':'agent:pi:s'}).status_code == 200 assert device.wait('agent_state', state='DONE')['final_text'] == end['final_text'] @pytest.mark.parametrize('failed', [False, True]) def test_tool_run_and_terminal_result_are_separate(downlink, failed): _, device, post, _ = downlink post('agent_start'); device.wait('agent_state', state='RUNNING') post('activity', activity_type='tool', activity='Running\nsleep 3', tool_name='bash') assert device.wait('agent_activity')['activity'] == 'Running\nsleep 3' post('activity', activity_type='text', activity='正在检查。') assert device.wait('agent_activity')['activity'] == '正在检查。' final = '依赖不可用，尚未完成。' if failed else '检查完成，结果正常。' post('agent_end', failed=failed, final_text=final) end = device.wait('agent_state', state='FAILED' if failed else 'DONE') assert end['final_text'] == final assert end['notification'] == ('failed' if failed else 'done') assert 'sleep 3' in end['recent_text'] and '正在检查。' in end['recent_text'] # Fire-and-forget activity POSTs can arrive after terminal; they must not replace it. post('activity', activity_type='tool', activity='Completed\nbash') device.wait('agent_activity') post('session_info_changed') assert device.wait('agent_state', state=end['state'])['final_text'] == final def test_final_bound_redaction_and_new_run_reset(downlink): _, device, post, _ = downlink post('agent_start'); device.wait('agent_state', state='RUNNING') final = '结果🙂 Authorization: Bearer private-credential\napi_key=private-key\n' + '中文🙂'*3000 post('agent_end', final_text=final) end = device.wait('agent_state', state='DONE') assert 0 < len(end['final_text'].encode()) <= 2048 assert 'private-' not in end['final_text'] and '\ufffd' not in end['final_text'] assert end['final_text'].startswith('结果🙂') post('agent_start', turn_id='next') assert device.wait('agent_state', state='RUNNING')['final_text'] == '' post('agent_end', turn_id='next', failed=True) end = device.wait('agent_state', state='FAILED') assert end['final_text'] == '' and end['state'] == 'FAILED' def test_terminal_state_carries_notification_and_generation(downlink): _, device, post, _ = downlink post('agent_start'); device.wait('agent_state', state='RUNNING') post('agent_end', final_text='完成。') end = device.wait('agent_state', state='DONE') # The immediate terminal push must survive the firmware target-select guard: # notification for board_notify_pi() plus the current target_generation. assert end['notification'] == 'done' assert end['target_generation'] == gateway.target_store.generation('downlink-test') def test_failed_terminal_state_carries_notification_and_generation(downlink): _, device, post, _ = downlink post('agent_start'); device.wait('agent_state', state='RUNNING') post('agent_end', failed=True) end = device.wait('agent_state', state='FAILED') assert end['notification'] == 'failed' assert end['target_generation'] == gateway.target_store.generation('downlink-test') def test_terminal_notification_never_reaches_switched_session(tmp_path, monkeypatch): credentials = tmp_path/'credentials.json' credentials.write_text(json.dumps(dict(device_id='downlink-test', device_token='test-device-only', pi_token='test-pi-only'))) registry = PiSessionRegistry() projects = ProjectRegistry(tmp_path/'projects.json') projects.add(Project(id='p', name='Test', path=str(tmp_path))) class RPC: async def call(self, method, params): if method == 'list': return [dict(session_id=s, name=s, cwd=str(tmp_path), state='IDLE', updated_at='2026-10-03T00:00:00Z', live=True) for s in ('s', 's2')] assert method == 'input' return dict(accepted=True, session_id=params.get('session_id', 's'), resumed=False) async def close(self): pass adapter = PiAdapter(projects, registry, rpc=RPC()) for key, value in dict(credential_store=DeviceCredentialStore(credentials), devices={}, target_store=TargetStore(tmp_path/'targets.json'), pi_sessions=registry, project_registry=projects, adapters={'pi': adapter}).items(): monkeypatch.setattr(gateway, key, value) monkeypatch.setattr(gateway.speeches, 'provider', None) monkeypatch.setattr(gateway.speeches, 'preferences', SpeechPreferences(tmp_path/'speech.json')) with TestClient(gateway.app) as client, client.websocket_connect('/device/ws') as ws: device = FakeDevice(ws) def post(session, kind, **extra): response = client.post('/integrations/pi/events', headers={'x-device-lab-pi-token':'test-pi-only'}, json=dict(event=kind, session_id=session, name=session, cwd=str(tmp_path), **{'turn_id':'t', **extra})) assert response.status_code == 200 return response.json() post('s', 'session_start') assert client.post('/devices/downlink-test/target', json={'target':'pi:s'}).status_code == 200 device.wait('agent_state', session_id='s') post('s', 'agent_start'); device.wait('agent_state', state='RUNNING') # User switches to session B while session A is still running. post('s2', 'session_start') assert client.post('/devices/downlink-test/target', json={'target':'pi:s2'}).status_code == 200 device.wait('agent_state', session_id='s2') # A's stale terminal notification must not reach the device now on B. post('s', 'agent_end', final_text='A 完成') before = len(device.recorder.records) client.post('/devices/downlink-test/message', json={'title':'barrier', 'body':'test'}) device.wait('display_message') stale = [r for r in device.recorder.records[before:] if r['type']=='agent_state'] assert not stale def test_unselected_session_never_pushes_activity(downlink): client, device, post, _ = downlink client.post('/devices/downlink-test/target', json={'target':'server'}) device.wait('target_state') post('agent_start') post('activity', activity_type='text', activity='must stay on s') client.post('/devices/downlink-test/message', json={'title':'barrier', 'body':'test'}) before = len(device.recorder.records) device.wait('display_message') assert not any(r['type'] in ('agent_state', 'agent_activity') for r in device.recorder.records[before:]) def test_codex_authoritative_final_is_preserved(tmp_path, monkeypatch): async def check(): projects = ProjectRegistry(tmp_path/'projects.json') projects.add(Project(id='p', name='Test', path=str(tmp_path))) adapter = CodexAdapter(projects) async def call(method, params): assert method == 'thread/read' return {'thread':dict(id='s', cwd=str(tmp_path), updatedAt=0, status={'type':'idle'})} async def observe(sid): pass adapter.call = call adapter.observe_session = observe monkeypatch.setattr(gateway, 'adapters', {'codex':adapter}) await adapter.event(dict(method='turn/started', params={'threadId':'s', 'turn':{'id':'t'}})) await adapter.event(dict(method='item/completed', params={'threadId':'s', 'turnId':'t', 'item':dict(id='final', type='agentMessage', phase='final_answer', text='Codex 结果。token=private-key')})) await adapter.event(dict(method='turn/completed', params={'threadId':'s', 'turn':{'id':'t', 'status':'completed'}})) payload = await gateway.agent_target_payload('agent:codex:s') assert payload['state'] == 'DONE' and payload['agent'] == 'codex' assert payload['final_text'].startswith('Codex 结果。') and 'private-key' not in payload['final_text'] asyncio.run(check())