#!/usr/bin/env python3
"""Read-only, allowlisted runtime proof; never emits messages, credentials or DB blobs."""
import argparse, hashlib, json, pathlib, platform, sqlite3
p=argparse.ArgumentParser(); p.add_argument('--db',required=True); p.add_argument('--before-db',required=True); p.add_argument('--run-ordinal',type=int,default=3); a=p.parse_args()
def connect(path):
 c=sqlite3.connect(pathlib.Path(path).resolve().as_uri()+'?mode=ro',uri=True); c.row_factory=sqlite3.Row; c.execute('PRAGMA query_only=ON'); return c
c=connect(a.db); b=connect(a.before_db)
def one(c,sql,args=()):return c.execute(sql,args).fetchone()
def emit(label,value):print(label+': '+json.dumps(value,sort_keys=True))
item='triage-16441-malformed-background'
old=one(b,'select * from orchestration_v2_projection_turn_items where turn_item_id=?',(item,))
now=one(c,'select * from orchestration_v2_projection_turn_items where turn_item_id=?',(item,))
assert old and now
thread=now['thread_id']
run=one(c,'select * from orchestration_v2_projection_runs where thread_id=? and ordinal=?',(thread,a.run_ordinal)); assert run
r=json.loads(run['payload_json']); rid=run['run_id']
attempt=one(c,'select * from orchestration_v2_projection_run_attempts where attempt_id=?',(r['activeAttemptId'],)); ap=json.loads(attempt['payload_json'])
node=one(c,'select * from orchestration_v2_projection_nodes where node_id=?',(r['rootNodeId'],)); np=json.loads(node['payload_json'])
errors=[json.loads(x[0]) for x in c.execute("select payload_json from orchestration_v2_projection_turn_items where run_id=? and type='error'",(rid,))]
outbox=[dict(x) for x in c.execute("select status,attempt_count,last_error,completed_at from orchestration_v2_effect_outbox where effect_type='provider-turn.start' and json_extract(payload_json,'$.runId')=?",(rid,))]
turns=c.execute('select count(*) from orchestration_v2_projection_provider_turns where run_attempt_id=?',(r['activeAttemptId'],)).fetchone()[0]
native=c.execute('select count(*) from orchestration_v2_projection_turn_items where run_id=? and provider_turn_id is not null',(rid,)).fetchone()[0]
events=[dict(x) for x in c.execute("select sequence,event_type from orchestration_events where json_extract(payload_json,'$.runId')=? or (event_type='run.updated' and json_extract(payload_json,'$.id')=?) order by sequence",(rid,rid))]
def digest(row):return hashlib.sha256(json.dumps(dict(row),sort_keys=True,separators=(',',':')).encode()).hexdigest()
print('Issue 16441 AFTER runtime proof — persisted backend state only')
print('Environment: Linux x86_64; Node 24.21.0; primary-owned dev server port 8492; isolated worktree .t3 database')
print('Fix tested: 016889ea432eeb7c7adf98d7e613b8b0b6617dea')
print('Method: read-only SQLite URI + PRAGMA query_only; allowlisted fields; no database uploaded')
print('Primary submitted browser follow-up at 2026-10-22 00:26:39 UTC; run ordinal 3')
print('Prior base run 2 stuck starting/outbox failed after 5 attempts; startup recovery cancelled it on fixed-server restart after malformed-item cleanup. The scoped fault trigger was then reinstalled before the fixed browser follow-up.')
for label,obj in [('run',r),('active_attempt',ap),('root_node',np)]:emit(label,{k:obj.get(k) for k in ('status','startedAt','completedAt')})
emit('failure_items',[{k:x.get(k) for k in ('type','status','title')} for x in errors]); emit('provider-turn.start',outbox)
emit('execution_evidence',{'provider_turns_for_active_attempt':turns,'native_turn_items_for_run':native,'attempt_provider_turn_id':ap.get('providerTurnId'),'root_provider_turn_id':np.get('providerTurnId')})
emit('run_events',events)
emit('malformed_prior_run_row',{'present':True,'unchanged_all_columns':dict(old)==dict(now),'before_sha256':digest(old),'after_sha256':digest(now),'belongs_to_prior_run':now['run_id']!=rid,'type':now['type'],'status':now['status'],'payload_missing_threadId':'threadId' not in json.loads(now['payload_json'])})
assert all(x['status']=='failed' and x.get('startedAt') is None for x in (r,ap,np))
assert len(errors)==1 and errors[0]['title']=='Thread history failed to load'
assert len(outbox)==1 and outbox[0]['status']=='succeeded' and outbox[0]['attempt_count']==5 and outbox[0]['last_error'] is None
assert turns==native==0 and ap.get('providerTurnId') is None and np.get('providerTurnId') is None
assert dict(old)==dict(now) and now['run_id']!=rid
print('ASSERTIONS: PASS')
print('Scope limit: backend terminal persistence verified. Corrupt history still prevents browser bounded-history loading; no UI render success claimed. Zero visual rows/captures/recordings.')