import asyncio
import base64
import json
from types import SimpleNamespace
from pathlib import Path
import pytest
from device_gateway.control.catalog import Catalog
from device_gateway.control.models import AgentSession,ControlError,Project,Settings
from device_gateway.control.projects import ProjectRegistry
from device_gateway.control.operations import Operations
from device_gateway.control.navigator import Navigator
from device_gateway.control.user_input import UserInputs
from device_gateway.control.approvals import ApprovalRegistry
from device_gateway.state import TargetStore
from device_gateway.voice import VoiceManager

class Agent:
    def __init__(self,agent,path,count=0,fail=False):self.agent=agent;self.path=str(path);self.count=count;self.fail=fail;self.creations=0;self.inputs=[];self.pause=None
    async def get_session(self,sid):
        if self.pause:await self.pause.wait()
        if self.fail or not sid.startswith('s'):raise ControlError('Session disappeared',404)
        return AgentSession(agent=self.agent,session_id=sid,project_id='p',name='同名很长'+('中文'*80),cwd=self.path,updated_at='2026-10-05T00:00:00Z')
    async def list_sessions(self,p):
        if self.fail:raise ControlError('Adapter unavailable')
        return [await self.get_session('s'+str(i)) for i in range(self.count)]
    async def create_session(self,p,text,settings,created,**kw):
        assert p.path==self.path and text.strip();self.creations+=1;sid='s-new'+str(self.creations);created(sid)
        self.inputs.append((sid,text));return dict(agent=self.agent,session_id=sid,project_id='p')
    async def send_input(self,sid,text):self.inputs.append((sid,text))

def registry(tmp):
    p=ProjectRegistry(tmp/'projects.json');p.add(Project(id='p',name='Demo 中文',path=str(tmp)));return p

@pytest.mark.parametrize('agents,count,partial',[(('pi',),0,False),(('codex',),9,False),(('pi','codex'),253,False),(('pi','codex'),9,True)])
def test_catalog_bounded_merged_identity_partial_ttl(tmp_path,agents,count,partial):
    async def run():
        now=[0];adapters={n:Agent(n,tmp_path,count,partial and n=='pi') for n in agents}
        c=Catalog(registry(tmp_path),adapters,clock=lambda:now[0]);page=await c.page('conn','sessions','p');snapshot=page['snapshot'];seen=[]
        assert page['status']==('Partial' if partial else 'Empty' if not count else 'Ready')
        while True:
            assert len(page['items'])<=4 and len(json.dumps(page,ensure_ascii=False).encode())<=4096
            for item in page['items']:assert 'cwd' not in item and 'path' not in item
            seen.extend((i['agent'],i['session_id']) for i in page['items'])
            if page['next_cursor'] is None:break
            page=await c.page('conn','sessions','p',snapshot,page['next_cursor'])
        assert len(set(seen))==len(seen)==count*(len(agents)-int(partial))
        with pytest.raises(ControlError,match='Stale'):await c.page('new-conn','sessions','p',snapshot)
        with pytest.raises(ControlError,match='Stale'):await c.page('conn','sessions','p',snapshot,1)
        now[0]=61
        with pytest.raises(ControlError,match='Expired'):await c.page('conn','sessions','p',snapshot)
    asyncio.run(run())

def test_target_generation_migration_aba_and_restart(tmp_path):
    path=tmp_path/'targets.json';path.write_text(json.dumps({'d':'pi:s0'}));t=TargetStore(path)
    assert t.get('d')=='agent:pi:s0' and t.generation('d')==0
    assert t.set('d','agent:pi:s1','p',expected=0)['generation']==1
    assert t.set('d','agent:pi:s0','p',expected=1)['generation']==2
    assert t.set('d','agent:pi:s2','p',expected=0) is None
    assert TargetStore(path).selection('d')==dict(target='agent:pi:s0',generation=2,project_id='p')

def test_target_admin_race_and_ack_lost_after_persistence(tmp_path):
    async def run():
        project=registry(tmp_path);a=Agent('pi',tmp_path,2);targets=TargetStore(tmp_path/'t.json');targets.set('d','agent:pi:s0','p');messages=[]
        class WS:
            fail=False
            async def send_json(self,v):
                if self.fail:raise ConnectionError('ACK lost')
                messages.append(v)
        d=SimpleNamespace(device_id='d',connection_id='c',send_lock=asyncio.Lock(),websocket=WS())
        async def send(d,v):await d.websocket.send_json(v)
        async def payload(t):return dict(type='agent_state',target=t,state='IDLE',model='m',effort='high',permissions={'current':'user'},latest_activity='now',recent_text='recent',final_text='final',elapsed_seconds=0)
        n=Navigator(project,{'pi':a},targets,Operations(tmp_path/'o.db'),UserInputs(),send,lambda d:True,payload,lambda d:None)
        a.pause=asyncio.Event();task=asyncio.create_task(n.select(d,dict(expected_target_generation=1,project_id='p',agent='pi',session_id='s1',request_id='r')));await asyncio.sleep(0)
        targets.set('d','agent:pi:s1');targets.set('d','agent:pi:s0');a.pause.set()
        with pytest.raises(ControlError,match='Stale'):await task
        assert messages==[]
        d.websocket.fail=True
        with pytest.raises(ConnectionError):await n.select(d,dict(expected_target_generation=3,project_id='p',agent='pi',session_id='s1',request_id='r2'))
        assert TargetStore(targets.path).generation('d')==4
        d.websocket.fail=False;await n.deliver_target(d,targets.selection('d'),'r2')
        assert [m['type'] for m in messages]==['target_selected','agent_state']
        assert all(m['target_generation']==4 for m in messages)
    asyncio.run(run())

@pytest.mark.parametrize('agent',['pi','codex'])
def test_create_first_input_journal_idempotent_ack_loss_restart(tmp_path,agent,caplog):
    async def run():
        a=Agent(agent,tmp_path);targets=TargetStore(tmp_path/'t.json');ops=Operations(tmp_path/'ops.db');d=SimpleNamespace(device_id='d',connection_id='c',send_lock=asyncio.Lock())
        async def send(d,v):raise ConnectionError()
        async def payload(t):return dict(type='agent_state',target=t)
        d.websocket=SimpleNamespace(send_json=send)
        n=Navigator(registry(tmp_path),{agent:a},targets,ops,UserInputs(),send,lambda d:True,payload,lambda d:None)
        event=dict(operation_id='op',project_id='p',agent=agent,expected_target_generation=0,settings={'model':'native-m'})
        first=await n.create(d,event,'first prompt');second=await n.create(d,event,'first prompt')
        assert first['state']==second['state']=='DONE' and a.creations==1 and a.inputs[0][1]=='first prompt'
        assert Operations(ops.path).get('d','op')['session_id']==first['session_id']
        with pytest.raises(ControlError,match='reused'):await n.create(d,event,'different prompt')
        assert a.creations==1
        # An UNKNOWN journal row survives a Gateway crash before native identity.
        ops.begin('d','unknown',{'first':'text'})
        assert not Operations(ops.path).begin('d','unknown',{'first':'text'})
    with caplog.at_level('INFO',logger='uvicorn.error.device_gateway.navigator'):
        asyncio.run(run())
    assert f'[SESSION_CREATE] stage=target_delivery agent={agent} state=DONE' in caplog.text
    assert 'first prompt' not in caplog.text

def test_native_user_input_semantics_lifecycle_secret_and_duplicates():
    async def run():
        u=UserInputs();questions=[dict(id='two',header='Q',question='Choose',options=[dict(label='one'),dict(label='two')]),dict(id='many',header='Q',question='Other',options=[dict(label=str(i)) for i in range(3)],isOther=True),dict(id='free',header='Q',question='Explain')]
        e=u.create('codex','s','p','native',questions,'o');rid=e['request_id'];u.presentations['conn']=(rid,'two')
        assert not u.resolve('old',rid,'two','one') and not u.resolve('conn',rid,'two','invented')
        assert u.resolve('conn',rid,'two','one') and not u.resolve('conn',rid,'two','one')
        u.presentations['conn']=(rid,'many');assert u.resolve('conn',rid,'many','different native answer')
        u.presentations['conn']=(rid,'free');assert u.resolve('conn',rid,'free','Voice text')
        assert (await u.wait(e))['answers']['free']=={'answers':['Voice text']}
        secret=u.create('pi','s','p','secret',[dict(id='secret',header='Q',question='Enter token',isSecret=True)],'o')
        u.presentations['conn']=(secret['request_id'],'secret');wire=u.wire(secret)
        assert wire['secret'] and not wire['free_text'] and wire['question']=='Answer on desktop'
        assert not u.resolve('conn',secret['request_id'],'secret','never-cloud')
        u.cancel_session('pi','s');assert secret['state']=='CANCELLED'
        again=u.create('codex','s','p','again',questions,'o');u.cancel_owner('o');assert again['state']=='CANCELLED'
    asyncio.run(run())

def test_native_approval_exact_choices_no_synthetic_hook():
    async def run():
        r=ApprovalRegistry();a=r.create(agent='codex',session_id='s',project_id='p',kind='command',title='Native',detail='Command',owner='o',request_identity='native',choices=['accept','acceptForSession','decline','cancel'])
        assert r.resolve(a.approval_id,'acceptForSession') and a.decision=='acceptForSession'
        assert not r.resolve(a.approval_id,'accept')
        b=r.create(agent='codex',session_id='s',project_id='p',kind='command',title='Native',detail='Command',owner='o',request_identity='other',choices=['accept','cancel'])
        assert not r.resolve(b.approval_id,'reject') and r.resolve(b.approval_id,'cancel')
        assert 'pi.on("tool_call"' not in (Path(__file__).parents[2]/'integrations/pi/device-lab-pager.ts').read_text()
    asyncio.run(run())

@pytest.mark.parametrize('duration',[999,1000,1001])
def test_short_voice_pcm_discard_no_stt_or_agent(tmp_path,duration):
    async def run():
        a=Agent('pi',tmp_path);events=[]
        class STT:
            calls=0
            async def transcribe(self,wav):self.calls+=1;return 'Valid transcript'
        stt=STT();d=SimpleNamespace(connection_id='c',device_id='d');generation=[1]
        async def send(d,v):events.append(v)
        v=VoiceManager(stt,lambda d:'agent:pi:s0',lambda agent:a,send,lambda d:True,lambda:False,generation_for=lambda d:generation[0])
        await v.handle(d,dict(type='voice_start',voice_id='v',sample_rate=44100,bits=16,channels=1))
        await v.handle(d,dict(type='voice_chunk',voice_id='v',sequence=0,audio=base64.b64encode(b'\0\0'*1764).decode()))
        req=v.requests['c']
        if duration<=1000:await v.handle(d,dict(type='voice_cancel',voice_id='v'))
        else:await v.handle(d,dict(type='voice_end',voice_id='v',recorded_duration_ms=duration))
        await asyncio.sleep(.02)
        assert stt.calls==int(duration>1000) and a.inputs==[]
        assert sum(e['type']=='voice_transcript' for e in events)==int(duration>1000)
        if duration<=1000:assert not req.pcm and not req.transcript and 'c' not in v.requests
        else:
            await v.handle(d,dict(type='voice_decision',voice_id='v',decision='send'))
            await v.handle(d,dict(type='voice_decision',voice_id='v',decision='send'))
            assert len(a.inputs)==1
        await v.close()
    asyncio.run(run())

@pytest.mark.parametrize('direct',[False,True])
def test_voice_native_input_confirm_direct_and_stale_secret(tmp_path,direct):
    async def run():
        u=UserInputs();p=registry(tmp_path);t=TargetStore(tmp_path/'t.json');t.set('d','agent:codex:s0','p');ops=Operations(tmp_path/'o.db');a=Agent('codex',tmp_path);messages=[]
        d=SimpleNamespace(device_id='d',connection_id='c',capabilities={'native-interaction'},send_lock=asyncio.Lock())
        async def send(d,v):messages.append(v)
        async def payload(t):return {}
        n=Navigator(p,{'codex':a},t,ops,u,send,lambda d:True,payload,lambda d:None)
        class STT:
            calls=0
            async def transcribe(self,wav):self.calls+=1;return 'native voice answer'
        provider=STT()
        v=VoiceManager(provider,t.get,lambda name:a,send,lambda d:True,lambda:False,generation_for=t.generation,context_prepare=n.voice_prepare,context_valid=n.voice_valid,context_submit=n.voice_submit)
        e=u.create('codex','s0','p','native',[dict(id='q',header='Q',question='Explain')],'o');rid=e['request_id'];u.presentations['c']=(rid,'q')
        await v.handle(d,dict(type='voice_start',voice_id='native',sample_rate=44100,bits=16,channels=1,context=dict(kind='user_input',request_id=rid,question_id='q')))
        await v.handle(d,dict(type='voice_chunk',voice_id='native',sequence=0,audio=base64.b64encode(b'\0\0'*1764).decode()))
        await v.handle(d,dict(type='voice_end',voice_id='native',recorded_duration_ms=1001));await asyncio.sleep(.02)
        assert e['state']=='PENDING' and a.inputs==[]
        # Confirm waits for button; Direct sends the identical existing decision.
        assert v.requests['c'].state=='AWAITING_CONFIRMATION'
        await v.handle(d,dict(type='voice_decision',voice_id='native',decision='send'))
        await v.handle(d,dict(type='voice_decision',voice_id='native',decision='send'))
        assert e['state']=='ANSWERED' and (await u.wait(e))['answers']['q']['answers']==['native voice answer']
        assert a.inputs==[] and provider.calls==1
        secret=u.create('codex','s0','p','secret',[dict(id='q',header='Q',question='Password',isSecret=True)],'o');u.presentations['c']=(secret['request_id'],'q')
        await v.handle(d,dict(type='voice_start',voice_id='secret',sample_rate=44100,bits=16,channels=1,context=dict(kind='user_input',request_id=secret['request_id'],question_id='q')))
        assert 'c' not in v.requests and provider.calls==1
        u.cancel(secret);await v.close()
    asyncio.run(run())

@pytest.mark.parametrize('agent',['pi','codex'])
@pytest.mark.parametrize('direct',[False,True])
def test_first_voice_prompt_creates_once_confirm_direct(tmp_path,agent,direct):
    async def run():
        a=Agent(agent,tmp_path);t=TargetStore(tmp_path/'t.json');messages=[]
        d=SimpleNamespace(device_id='d',connection_id='c',capabilities={'native-interaction','session-create'},send_lock=asyncio.Lock())
        async def send(device,event):messages.append(event)
        d.websocket=SimpleNamespace(send_json=lambda event:send(d,event))
        async def payload(target):return {'type':'agent_state','target':target}
        n=Navigator(registry(tmp_path),{agent:a},t,Operations(tmp_path/'o.db'),UserInputs(),send,lambda d:True,payload,lambda d:None)
        class STT:
            async def transcribe(self,wav):return 'First Voice prompt'
        v=VoiceManager(STT(),t.get,lambda name:a,send,lambda d:True,lambda:False,generation_for=t.generation,context_prepare=n.voice_prepare,context_valid=n.voice_valid,context_submit=n.voice_submit)
        context=dict(kind='new_session',operation_id='one',project_id='p',agent=agent,settings={'model':'native-m','effort':'high',**({'permissions':'user'} if agent=='codex' else {})},expected_target_generation=0)
        await v.handle(d,dict(type='voice_start',voice_id='first',sample_rate=44100,bits=16,channels=1,context=context))
        await v.handle(d,dict(type='voice_chunk',voice_id='first',sequence=0,audio=base64.b64encode(b'\0\0'*1764).decode()))
        await v.handle(d,dict(type='voice_end',voice_id='first',recorded_duration_ms=1001));await asyncio.sleep(.02)
        assert a.creations==0 and v.requests['c'].state=='AWAITING_CONFIRMATION'
        # Firmware Confirm waits; Direct emits this same once-only decision.
        await v.handle(d,dict(type='voice_decision',voice_id='first',decision='send'))
        await v.handle(d,dict(type='voice_decision',voice_id='first',decision='send'))
        assert a.creations==1 and a.inputs==[('s-new1','First Voice prompt')]
        status=n.operations.get('d','one');assert status['state']=='DONE'
        assert t.get('d')==f'agent:{agent}:s-new1'
        # A reconnect queries journal status, never submits another first prompt.
        d.connection_id='reconnected';assert n.operations.get('d','one')['session_id']=='s-new1'
        assert (await n.create(d,context,'First Voice prompt'))['state']=='DONE' and a.creations==1
        await v.close()
    asyncio.run(run())

@pytest.mark.parametrize('cause',['aba','disconnected','approval','native-resolved'])
def test_voice_direct_stale_preemption_never_submits(tmp_path,cause):
    async def run():
        a=Agent('codex',tmp_path);t=TargetStore(tmp_path/'t.json');t.set('d','agent:codex:s0','p');current=[True];approval=[False];u=UserInputs()
        d=SimpleNamespace(device_id='d',connection_id='c',capabilities={'native-interaction','target-select'})
        async def send(d,event):pass
        async def payload(target):return {}
        n=Navigator(registry(tmp_path),{'codex':a},t,Operations(tmp_path/'o.db'),u,send,lambda d:current[0],payload,lambda d:None)
        class STT:
            calls=0
            async def transcribe(self,wav):self.calls+=1;return 'Valid transcript'
        provider=STT();v=VoiceManager(provider,t.get,lambda name:a,send,lambda d:current[0],lambda:approval[0],generation_for=t.generation,context_prepare=n.voice_prepare,context_valid=n.voice_valid,context_submit=n.voice_submit)
        context=dict(kind='message',target=t.get('d'),expected_target_generation=1)
        if cause=='native-resolved':
            e=u.create('codex','s0','p','native',[dict(id='q',header='Q',question='Explain')],'owner');u.presentations['c']=(e['request_id'],'q');context=dict(kind='user_input',request_id=e['request_id'],question_id='q')
        await v.handle(d,dict(type='voice_start',voice_id='stale',sample_rate=44100,bits=16,channels=1,context=context))
        await v.handle(d,dict(type='voice_chunk',voice_id='stale',sequence=0,audio=base64.b64encode(b'\0\0'*1764).decode()))
        await v.handle(d,dict(type='voice_end',voice_id='stale',recorded_duration_ms=1001));await asyncio.sleep(.02)
        assert provider.calls==1
        if cause=='aba':t.set('d','agent:codex:s1');t.set('d','agent:codex:s0')
        elif cause=='disconnected':current[0]=False
        elif cause=='approval':approval[0]=True
        else:u.cancel(e)
        await v.handle(d,dict(type='voice_decision',voice_id='stale',decision='send'))
        assert a.inputs==[];await v.close()
    asyncio.run(run())

def test_native_long_unicode_options_fit_frame_and_keep_exact_answer():
    async def run():
        u=UserInputs();labels=['选项'+str(i)+'中文'*150 for i in range(32)]
        e=u.create('codex','s','p','native',[dict(id='q',header='Q',question='中文'*1000,options=[{'label':label} for label in labels])],'owner')
        wire=u.wire(e);assert len(json.dumps(wire,ensure_ascii=False).encode())<=4096
        u.presentations['c']=(e['request_id'],'q');assert u.resolve('c',e['request_id'],'q',None,31)
        assert (await u.wait(e))['answers']['q']['answers']==[labels[31]]
        again=u.create('pi','s','p','next',[dict(id='q',header='Q',question='Explain')],'owner');u.presentations['old']=(again['request_id'],'q')
        assert not u.resolve('new',again['request_id'],'q','stale')
        u.cancel_session('pi','s');assert not u.resolve('old',again['request_id'],'q','late')
    asyncio.run(run())

def test_corrupt_target_store_cannot_reset_generation(tmp_path):
    path=tmp_path/'t.json';path.write_text('{broken')
    with pytest.raises(ControlError,match='Invalid target'):TargetStore(path).set('d','agent:pi:s0',expected=0)
    assert path.read_text()=='{broken'

def test_codex_native_permissions_read_update_readback_busy_unsupported(tmp_path):
    from device_gateway.control.codex import CodexAdapter
    async def run():
        class RPC:
            generation=1
            def __init__(self):self.reviewer='user';self.calls=[];self.busy=False
            async def start(self):pass
            async def write(self,message):pass
            async def call(self,method,params):
                self.calls.append((method,params))
                thread=dict(id='s',cwd=str(tmp_path),updatedAt=1,preview='Native',model='m',reasoningEffort='high',status={'type':'active' if self.busy else 'idle'})
                if method=='thread/read':return {'thread':thread}
                if method=='thread/resume':return dict(thread=thread,approvalsReviewer=self.reviewer,approvalPolicy='on-request',sandbox={'type':'workspaceWrite','networkAccess':False},activePermissionProfile=None)
                if method=='thread/settings/update':self.reviewer=params.get('approvalsReviewer',self.reviewer);return {}
                if method=='model/list':return dict(data=[dict(id='m',model='m',displayName='M',isDefault=True,supportedReasoningEfforts=[{'reasoningEffort':'high','description':'High'}],defaultReasoningEffort='high')],nextCursor=None)
                return {}
        rpc=RPC();a=CodexAdapter(registry(tmp_path),rpc)
        read=await a.permissions('s');assert read['current']=='user' and [m['id'] for m in read['available_modes']]==['user','auto_review']
        updated=await a.update_settings('s',Settings(permissions='auto_review'));assert updated['permissions']['current']=='auto_review'
        update=next(p for m,p in rpc.calls if m=='thread/settings/update');assert update=={'threadId':'s','approvalsReviewer':'auto_review'}
        assert updated['permissions']['approvalPolicy']=='on-request' and updated['permissions']['sandboxPolicy']['networkAccess'] is False
        count=sum(m=='thread/settings/update' for m,p in rpc.calls)
        with pytest.raises(ControlError,match='Unsupported'):await a.update_settings('s',Settings(permissions='invented'))
        rpc.busy=True
        with pytest.raises(ControlError,match='busy'):await a.update_settings('s',Settings(permissions='user'))
        assert sum(m=='thread/settings/update' for m,p in rpc.calls)==count
    asyncio.run(run())

def test_creation_model_snapshot_paging_connection_and_removed_session(tmp_path):
    async def run():
        class Models(Agent):
            calls=0
            async def creation_models(self,project):
                self.calls+=1;return [dict(id=f'm{i}',name='中文'*100,available_efforts=['low','high']) for i in range(9)]
        a=Models('pi',tmp_path);d=SimpleNamespace(device_id='d',connection_id='c',send_lock=asyncio.Lock())
        async def send(d,event):pass
        async def payload(target):return {}
        t=TargetStore(tmp_path/'t.json');n=Navigator(registry(tmp_path),{'pi':a},t,Operations(tmp_path/'o.db'),UserInputs(),send,lambda d:True,payload,lambda d:None)
        event=dict(project_id='p',agent='pi');page=await n.creation_options(d,event)
        assert len(page['items'])==4 and not page['permissions'] and len(json.dumps(page,ensure_ascii=False).encode())<4096
        second=await n.creation_options(d,dict(event,snapshot=page['snapshot'],cursor=4))
        assert second['items'][0]['id']=='m4' and a.calls==1
        d.connection_id='new'
        with pytest.raises(ControlError,match='Stale'):await n.creation_options(d,dict(event,snapshot=page['snapshot'],cursor=4))
        with pytest.raises(ControlError,match='disappeared'):await n.select(d,dict(expected_target_generation=0,project_id='p',agent='pi',session_id='removed'))
        assert t.generation('d')==0
    asyncio.run(run())

def test_pi_native_history_survives_empty_live_heartbeat_overlay(tmp_path):
    from device_gateway.control.pi import PiAdapter
    from device_gateway.integrations.pi import PiSessionEvent,PiSessionRegistry
    async def run():
        class RPC:
            async def call(self,method,params):
                return [dict(session_id='saved',name='Saved',cwd=str(tmp_path),updated_at='2026-10-05T00:00:00Z',live=True,state='IDLE',model='provider/model',recent_text='Actual saved response',final_text='Actual saved response',latest_activity='Actual saved response')]
        live=PiSessionRegistry();live.apply(PiSessionEvent(event='session_start',session_id='saved',name='Saved',cwd=str(tmp_path)))
        adapter=PiAdapter(registry(tmp_path),live,RPC())
        session=await adapter.get_session('saved');assert session.recent_text==session.final_text=='Actual saved response'
        live.apply(PiSessionEvent(event='agent_start',session_id='saved',name='Saved',cwd=str(tmp_path)))
        session=await adapter.get_session('saved');assert session.state=='RUNNING' and session.final_text==''
    asyncio.run(run())

def test_device_snapshot_retains_native_pi_text_under_heartbeat_overlay(tmp_path,monkeypatch):
    import device_gateway.app as gateway
    from device_gateway.integrations.pi import PiSessionEvent,PiSessionRegistry
    async def run():
        live=PiSessionRegistry();live.apply(PiSessionEvent(event='session_start',session_id='saved',name='Saved',cwd=str(tmp_path)))
        class Adapter:
            async def get_session(self,sid):return AgentSession(agent='pi',session_id=sid,project_id='p',name='Saved',cwd=str(tmp_path),updated_at='2026-10-05T00:00:00Z',live=True,state='IDLE',recent_text='Actual saved answer',final_text='Actual saved answer',latest_activity='Actual saved answer',model='native')
        monkeypatch.setattr(gateway,'pi_sessions',live);monkeypatch.setattr(gateway,'adapters',{'pi':Adapter()})
        payload=await gateway.agent_target_payload('agent:pi:saved')
        assert payload['recent_text']==payload['final_text']==payload['latest_activity']=='Actual saved answer'
        assert payload['model']=='native' and payload['target']=='agent:pi:saved'
    asyncio.run(run())