#!/usr/bin/env python3 """Explicit real-model proof. Isolated Gateway, Codex server and new Pi session. Uses existing model credentials without printing/copying them. All mutations are confined to fresh /tmp/device-lab-*-approval-smoke* paths. No production restart. """ import asyncio import json import os from pathlib import Path import socket import sys import tempfile from uuid import uuid4 REPO = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO / 'server')) ROOT = Path(tempfile.mkdtemp(prefix='device-lab-approval-smoke-')) PI_PROJECT = ROOT / 'pi-project' CODEX_PROJECT = ROOT / 'codex-project' PI_PROJECT.mkdir(); CODEX_PROJECT.mkdir() PI_FILE = Path('/tmp') / ('device-lab-pi-approval-smoke-' + uuid4().hex + '.txt') with socket.socket() as s: s.bind(('127.0.0.1', 0)); PORT = s.getsockname()[1] CREDENTIALS = ROOT / 'device.json' CREDENTIALS.write_text(json.dumps({'device_id':'approval-smoke', 'device_token':'local-smoke-only', 'pi_token':'local-pi-smoke-only'})) CREDENTIALS.chmod(0o600) os.environ.update(GATEWAY_PORT=str(PORT), GATEWAY_CODEX_ENABLED='1', GATEWAY_CODEX_SOCKET=str(ROOT/'codex.sock'), DEVICE_LAB_CONTROL_DIR=str(ROOT/'runtime'), GATEWAY_PROJECTS_FILE=str(ROOT/'projects.json'), GATEWAY_STATE_FILE=str(ROOT/'targets.json'), DEVICE_CREDENTIALS_FILE=str(CREDENTIALS), DEVICE_LAB_CREDENTIALS_FILE=str(CREDENTIALS), DEVICE_LAB_GATEWAY_URL=f'http://127.0.0.1:{PORT}') import httpx import uvicorn from websockets.asyncio.client import connect import device_gateway.app as gateway from device_gateway.control.models import Project async def until(fn, *, timeout=180): async with asyncio.timeout(timeout): while True: result = await fn() if result: return result await asyncio.sleep(.1) async def main(): print('Proof directory:', ROOT, 'Gateway port:', PORT, flush=True) result = {'proof_directory':str(ROOT), 'pi_file':str(PI_FILE)} gateway.project_registry.add(Project(id='pi-smoke',name='Pi approval smoke',path=str(PI_PROJECT))) gateway.project_registry.add(Project(id='codex-smoke',name='Codex approval smoke',path=str(CODEX_PROJECT))) log = (ROOT/'codex-server.log').open('w') codex_process = await asyncio.create_subprocess_exec('codex','app-server','--listen','unix://'+str(ROOT/'codex.sock'), stdout=log, stderr=log) server = uvicorn.Server(uvicorn.Config(gateway.app,host='127.0.0.1',port=PORT,log_level='warning')) server_task = asyncio.create_task(server.serve()) pi = gateway.adapters['pi']; codex = gateway.adapters['codex'] events=[] original_event=codex.event async def capture(message): events.append(message) await original_event(message) codex.rpc.on_event=capture async with httpx.AsyncClient(base_url=f'http://127.0.0.1:{PORT}',timeout=30) as client: async def ready(): try: return (await client.get('/health')).status_code==200 except httpx.ConnectError:return False async def pending(agent,sid): data=(await client.get('/api/approvals')).json() return next((a for a in data if a['agent']==agent and a['session_id']==sid),None) async def decide(a,decision): response=await client.post(f"/api/approvals/{a['approval_id']}/decision",json={'decision':decision}) response.raise_for_status() async def pi_done(sid): session=gateway.pi_sessions.get(sid) return session and session.state in ('DONE','FAILED') try: await until(ready,timeout=15) async def socket_ready():return (ROOT/'codex.sock').exists() await until(socket_ready,timeout=15) # One real Pi SDK runtime/session for read, reject and approve. sid=(await pi.rpc.call('create_smoke',{'cwd':str(PI_PROJECT)}))['session_id'] result['pi_session']=sid print('Pi session:',sid,flush=True) seed=PI_PROJECT/'read-only.txt';seed.write_text('READ_ONLY_PROOF\n') await pi.rpc.call('input',{'cwd':str(PI_PROJECT),'session_id':sid,'text':f'Use the read tool to read {seed}. Do not use other tools. Reply READ_ONLY_PROOF.'}) await until(lambda:pi_done(sid)) assert not gateway.approvals.entries, 'read-only Pi tool unexpectedly needed approval' result['pi_read_no_approval']=True; print('Pi read: no approval',flush=True) for label, decision, channel in [('reject','reject','admin_api'), ('device_reject','reject','simulated_device_B'), ('approve','approve','simulated_device_A')]: text=(f'Use the write tool exactly once to create {PI_FILE} with exact content PI_APPROVED. ' 'This is an approval integration test. If the tool is rejected, stop immediately without retry or another tool. ' 'Do not ask for confirmation in text; call the write tool now.') response=await client.post(f'/api/agents/pi/sessions/{sid}/input',json={'text':text});response.raise_for_status() a=await until(lambda:pending('pi',sid)) print('Pi pending',decision,a['approval_id'],flush=True) await asyncio.sleep(2) assert not PI_FILE.exists(), 'Pi crossed pending gate' assert (await client.get(f'/api/agents/pi/sessions/{sid}')).json()['state']=='WAITING_APPROVAL' assert gateway.pi_sessions.get(sid).state == 'RUNNING' if channel=='admin_api': await decide(a,decision) else: async with connect(f'ws://127.0.0.1:{PORT}/device/ws') as ws: await ws.send(json.dumps({'type':'hello','device_id':'approval-smoke','device_token':'local-smoke-only', 'board':'simulated-stopwatch','firmware':'0.3.0'})) while True: wire=json.loads(await asyncio.wait_for(ws.recv(),5)) if wire['type']=='approval_request':break assert wire['approval_id']==a['approval_id'] await ws.send(json.dumps({'type':'approval_response','approval_id':wire['approval_id'],'decision':decision})) while True: wire=json.loads(await asyncio.wait_for(ws.recv(),5)) if wire['type']=='approval_cancelled':break assert wire['state']==('APPROVED' if decision=='approve' else 'REJECTED') await until(lambda:pi_done(sid)) assert PI_FILE.exists()==(decision=='approve') if decision=='approve':assert PI_FILE.read_text().strip()=='PI_APPROVED' result['pi_'+label]={'approval_id':a['approval_id'],'gate_wait_seconds':2,'file_exists':PI_FILE.exists(), 'channel':channel} print('Pi',label,'proof passed',flush=True) # New Codex thread, on-request + read-only forces an explicit sandbox escape. created=await codex.call('thread/start',{'cwd':str(CODEX_PROJECT),'approvalPolicy':'on-request', 'approvalsReviewer':'user','sandbox':'read-only', 'baseInstructions':'You are testing a human approval gate. Only perform the exact requested single command. If declined, stop; never retry or use a different tool. Do not inspect any other files.'}) sid=created['thread']['id'];result['codex_thread']=sid for decision in ('reject','approve'): target=CODEX_PROJECT/(decision+'.txt') mark=len(events) command=f"printf CODEX_APPROVED > {target}" prompt=(f'Call exec_command exactly once with command: {command}\n' 'Set sandbox_permissions="require_escalated" and justification="Approval smoke test: create only the named temporary proof file". ' 'The read-only sandbox needs user approval for this write. Request the tool approval now. ' 'Do not use apply_patch, another command, or retry if rejected. Then finish.') turn=await codex.call('turn/start',{'threadId':sid,'input':[{'type':'text','text':prompt}]}) a=await until(lambda:pending('codex',sid)) print('Codex pending',decision,a['approval_id'],flush=True) await asyncio.sleep(2);assert not target.exists(), 'Codex crossed pending gate' assert any(e.get('method')=='item/commandExecution/requestApproval' for e in events[mark:]) await decide(a,decision) async def codex_done():return next((e for e in events[mark:] if e.get('method')=='turn/completed'),None) done=await until(codex_done) assert done['params']['turn']['status']=='completed', done assert target.exists()==(decision=='approve') if decision=='approve':assert target.read_text()=='CODEX_APPROVED' result['codex_'+decision]={'approval_id':a['approval_id'],'turn_id':turn['turn']['id'], 'file':str(target),'file_exists':target.exists(),'gate_wait_seconds':2,'channel':'admin_api'} print('Codex',decision,'proof passed',flush=True) result['pi_ready']=result['codex_ready']=True except Exception as exc: result['error']=repr(exc) # A compact protocol trace contains only this dedicated smoke thread. (ROOT/'codex-events.json').write_text(json.dumps(events,indent=2)) raise finally: (ROOT/'codex-events.json').write_text(json.dumps(events,indent=2)) (ROOT/'result.json').write_text(json.dumps(result,indent=2)) print(json.dumps(result,indent=2),flush=True) server.should_exit=True await asyncio.wait_for(server_task,20) codex_process.terminate() try:await asyncio.wait_for(codex_process.wait(),10) except TimeoutError:codex_process.kill();await codex_process.wait() log.close() if __name__=='__main__':asyncio.run(main())