87 lines
5.6 KiB
Python
87 lines
5.6 KiB
Python
"""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)
|