import sqlite3,json,urllib.request,subprocess,os,hashlib,re
from pathlib import Path
from datetime import datetime,timezone
DB='/opt/struktur/youtube-research/knowledge.db'; LOG=Path('/opt/struktur/reports/aa049-worker-events.jsonl')
now=datetime.now(timezone.utc); now_s=now.isoformat(); out=[]
# Live inventory exact-match gate.
eps=json.load(urllib.request.urlopen('http://127.0.0.1:8644/episodes/inventory',timeout=90)).get('episodes',[])
# No active queue worker process and no held worker lock.
ps=subprocess.run(['ps','-eo','pid,cmd'],capture_output=True,text=True).stdout
active=[x for x in ps.splitlines() if any(t in x for t in ['aa049_queue_worker.py','aa043_single_writer.py'])]
lock=Path('/run/graphiti-control/aa049-queue-worker.lock')
lock_held=False
if lock.exists():
 import fcntl
 with lock.open('r+') as f:
  try: fcntl.flock(f,fcntl.LOCK_EX|fcntl.LOCK_NB); fcntl.flock(f,fcntl.LOCK_UN)
  except BlockingIOError: lock_held=True
if active or lock_held: raise SystemExit({'blocked':'active_writer_or_lock','processes':active,'lock_held':lock_held})
c=sqlite3.connect(DB,timeout=30,isolation_level=None); c.row_factory=sqlite3.Row; c.execute('pragma busy_timeout=30000')
try:
 c.execute('BEGIN IMMEDIATE')
 for qid in (1092,1268):
  r=c.execute('select * from graphiti_import_queue where id=?',(qid,)).fetchone()
  if not r: raise RuntimeError(f'missing_queue:{qid}')
  d=dict(r); p=Path(d['obsidian_path']); raw=p.read_bytes(); file_hash=hashlib.sha256(raw).hexdigest()
  exact=[e for e in eps if d.get('import_identity') and (d['import_identity'] in str(e.get('source_description','')) or e.get('name')=='aa043_'+d['import_identity'])]
  fm=raw.decode('utf8','replace').split('---',2)[1] if raw.startswith(b'---') else ''
  y=re.search(r'(?im)^youtube_id:\s*(.+)$',fm)
  checks={'status':d['graphiti_status']=='processing','lease_expired':bool(d.get('lease_expires_at') and d['lease_expires_at']<now_s),'owner':d.get('lock_owner')=='aa043-single-writer','remote_unknown':d.get('remote_outcome')=='UNKNOWN_REMOTE_OUTCOME','episode_null':d.get('graphiti_episode_id') is None,'exact_episode_count':len(exact),'hash_match':file_hash==d.get('content_hash'),'youtube_id_match':bool(y and y.group(1).strip()==d.get('video_id'))}
  if not all([checks['status'],checks['lease_expired'],checks['owner'],checks['remote_unknown'],checks['episode_null'],checks['exact_episode_count']==0,checks['hash_match'],checks['youtube_id_match']]): raise RuntimeError(json.dumps({'qid':qid,'checks':checks},ensure_ascii=False))
  old={k:d.get(k) for k in ['lock_owner','lease_expires_at','heartbeat_at','lock_token','remote_outcome','last_error_class','last_http_status','attempt_count','graphiti_attempts','import_identity','content_hash']}
  cur=c.execute("update graphiti_import_queue set graphiti_status='lease_expired',lock_owner=NULL,lock_token=NULL,lease_expires_at=NULL,heartbeat_at=NULL,remote_outcome='UNKNOWN_REMOTE_OUTCOME',last_error_class='lease_expired_recovery',last_error_message=?,updated_at=? where id=? and graphiti_status='processing' and lease_expires_at < ? and lock_owner='aa043-single-writer' and graphiti_episode_id is null",(f"controlled recovery {now_s}; prior_error={d.get('last_error_message','')[:700]}",now_s,qid,now_s))
  if cur.rowcount != 1: raise RuntimeError(f'guard_update_failed:{qid}')
  out.append({'event':'LEASE_RECOVERY','queue_id':qid,'at_utc':now_s,'old':old,'checks':checks,'new_status':'lease_expired','action':'regular_aa049_writer_only'})
 c.execute('COMMIT')
except:
 c.rollback(); c.close(); raise
c.close()
with LOG.open('a',encoding='utf8') as f:
 for e in out: f.write(json.dumps(e,ensure_ascii=False)+'\n')
print(json.dumps({'recovered':out,'active_processes':active,'lock_held':lock_held},ensure_ascii=False,indent=2))