story/firmware/echoear-platform-agent/tools/com6_reconnect_acceptance.py

"""One bounded COM6 reboot; observe real MQTT lifecycle and durable command ACK.

Requires the existing, COM6-restricted operator helper in the API container.
Records operational metadata only. Does not open serial or change ownership.
"""
import datetime
import json
from pathlib import Path
import subprocess

REMOTE = r'''
import datetime,json,subprocess,time,uuid
p=json.loads(subprocess.check_output(['docker','inspect','data-terminal-cloud-postgres-1']))[0]
env=dict(x.split('=',1) for x in p['Config']['Env'] if '=' in x)
psql=['docker','exec',p['Id'],'psql','-U',env['POSTGRES_USER'],'-d',env['POSTGRES_DB'],'-At','-v','ON_ERROR_STOP=1','-c']
query="""SELECT json_build_object(
'utc',now(),'connected',p.connected,'online',COALESCE(p.connected AND p.last_seen_at>now()-interval '90 seconds',false),
'session',p.connection_id,'heartbeat',p.last_seen_at,'firmware',t.firmware_version,
'desired_revision',t.desired_revision,'reported_revision',t.reported_revision,
'speaker',t.reported_config->'volume'->'speaker','power',t.reported_config->'power'->>'mode',
'bound',(SELECT count(*) FROM device_binding WHERE device_id=d.id AND unbound_at IS NULL),
'pending',(SELECT count(*) FROM device_command WHERE device_id=d.id AND status IN ('QUEUED','DELIVERED','ACKNOWLEDGED') AND expires_at>now()),
'busy',EXISTS(SELECT 1 FROM call_session WHERE device_id=d.id AND call_state IN ('PREPARING','STARTING','TALKING','STOPPING')) OR
EXISTS(SELECT 1 FROM conversation WHERE device_id=d.id AND status NOT IN ('COMPLETED','FAILED','EXPIRED')) OR
EXISTS(SELECT 1 FROM fleet_ota WHERE device_id=d.id AND status IN ('pending','accepted')))
FROM device d JOIN fleet_presence p ON p.device_id=d.id JOIN device_twin t ON t.device_id=d.id
WHERE d.public_id='623a4e54-bdc4-4c98-b03e-27267d3ff360' AND d.factory_device_id='KP01-020026091601' AND d.status='registered'"""
def state():
 return json.loads(subprocess.check_output(psql+[query],text=True,timeout=10))
def emit(key,value):
 print(json.dumps({key:value}),flush=True)
initial=state()
assert initial['bound']==1 and not initial['busy'] and initial['pending']==0 and initial['online'], 'COM6 not idle/online or binding changed'
assert initial['desired_revision']==initial['reported_revision'] and initial['power']=='active', 'COM6 config unsettled or standby'
session=initial.pop('session');emit('initial',initial)
start=time.monotonic()
command=subprocess.Popen(['docker','exec','data-terminal-cloud-api-1','/tmp/com6-power-verify','reboot'],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
first=command.stdout.readline().strip()
assert first, 'reboot helper did not return a command ID'
receipt=json.loads(first);uuid.UUID(receipt['command_id']);emit('command',receipt)
observations=[];offline=False;reconnected=False
while time.monotonic()-start<100:
 current=state();changed=current.pop('session')!=session
 current['session_changed']=changed;current['elapsed_sec']=round(time.monotonic()-start,3)
 observations.append(current)
 offline=offline or not current['online']
 reconnected=reconnected or (changed and current['online'])
 if len(observations)==1 or (len(observations)>1 and (current['online'],changed)!=(observations[-2]['online'],observations[-2]['session_changed'])):
  emit('transition',current)
 if command.poll() is not None and reconnected and current['pending']==0 and current['desired_revision']==current['reported_revision']:
  break
 time.sleep(.25)
output,error=command.communicate(timeout=15)
receipts=[json.loads(line) for line in output.splitlines() if line.strip()]
terminal=receipts[-1] if receipts else {}
emit('observations',observations)
emit('result',{'exit':command.returncode,'command_id':receipt['command_id'],'status':terminal.get('status'),
 'delivery_attempts':terminal.get('delivery_attempts'),'offline_observed':offline,'reconnected':reconnected,
 'settings_preserved':current['speaker']==initial['speaker'] and current['power']==initial['power'] and current['bound']==initial['bound'] and current['desired_revision']==initial['desired_revision'] and current['reported_revision']==current['desired_revision'],
 'error':error.strip()})
assert command.returncode==0 and terminal.get('status')=='SUCCEEDED' and reconnected, 'reboot/reconnect failed'
assert current['speaker']==initial['speaker'] and current['power']==initial['power'] and current['bound']==1 and current['desired_revision']==initial['desired_revision'] and current['reported_revision']==current['desired_revision']
# A very short reboot may reconnect before TCP detects the old peer is gone.
# Report the observation faithfully; do not call it an offline pass.
'''

if __name__ == '__main__':
    stamp = datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%SZ')
    evidence = Path('E:/Codex/story/.local-evidence/com6-controls-20260922') / ('reconnect-' + stamp + '.jsonl')
    ssh = ['ssh', '-i', 'E:/Codex/story/.local-evidence/account-reset-18642041700-20260916/migration-key.pem',
           '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', 'root@82.157.201.114', 'python3 -u -']
    process = subprocess.Popen(ssh, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding='utf-8')
    process.stdin.write(REMOTE)
    process.stdin.close()
    with evidence.open('w', encoding='utf-8') as out:
        for line in process.stdout:
            out.write(line)
            out.flush()
            if '"observations"' not in line:
                print(line.strip(), flush=True)
    error = process.stderr.read()
    result = process.wait()
    if error:
        print(error, flush=True)
    print(json.dumps({'evidence': str(evidence), 'exit': result}), flush=True)
    raise SystemExit(result)