#!/usr/bin/env python3
import sqlite3, json, hashlib, os, glob, subprocess, datetime, re
DB='/opt/struktur/youtube-research/knowledge.db'
P42='/opt/struktur/reports/aa043-p42/20260828T100641Z'
QID=1281
SRC='/opt/obsidian-vault/Solutions/E-Mail-Architektur-Hermes-Obsidian.md'
OUT='/opt/struktur/reports/aa043-p43-r1/'+datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%SZ')
os.makedirs(OUT,exist_ok=True); UTC=datetime.timezone.utc
now=datetime.datetime.now(UTC).isoformat()
def sha(p):
 h=hashlib.sha256()
 with open(p,'rb') as f:
  for b in iter(lambda:f.read(1048576),b''): h.update(b)
 return h.hexdigest()
def run(cmd,timeout=60):
 # Fixed, read-only diagnostics only; no user-controlled command input.
 try: return subprocess.run(cmd,shell=True,text=True,capture_output=True,timeout=timeout).stdout
 except Exception as e: return 'ERROR '+repr(e)
def dump(name,obj):
 with open(os.path.join(OUT,name),'w',encoding='utf8') as f: json.dump(obj,f,ensure_ascii=False,indent=2)
# frozen P42 artifact hashes
p42_hashes={}
for p in glob.glob(P42+'/*'):
 try: p42_hashes[os.path.basename(p)]={'size':os.path.getsize(p),'sha256':sha(p)}
 except: pass
# readonly db
c=sqlite3.connect('file:'+DB+'?mode=ro&immutable=1',uri=True); c.row_factory=sqlite3.Row
fk=[tuple(x) for x in c.execute('pragma foreign_key_check')]
baseline={'checked_at':now,'db_path':DB,'db_sha256':sha(DB),'db_size':os.path.getsize(DB),'knowledge_units_total':c.execute('select count(*) from knowledge_units').fetchone()[0],'real_kus':c.execute('select count(*) from knowledge_units where legacy_placeholder=0').fetchone()[0],'legacy_placeholder':c.execute('select count(*) from knowledge_units where legacy_placeholder=1').fetchone()[0],'review':dict(c.execute('select review_state,count(*) from knowledge_units group by review_state')),'evidence':dict(c.execute('select evidence_class,count(*) from knowledge_units group by evidence_class')),'gold':dict(c.execute('select gold_state,count(*) from knowledge_units group by gold_state')),'integrity_check':c.execute('pragma integrity_check').fetchone()[0],'foreign_key_check':len(fk),'legacy_fk_fingerprint':hashlib.sha256(json.dumps(sorted(fk),sort_keys=True).encode()).hexdigest(),'queue_status':{k:c.execute('select count(*) from graphiti_import_queue where graphiti_status=?',(k,)).fetchone()[0] for k in ['done','queued','processing','retry_wait','failed_permanent']},'queue_total':c.execute('select count(*) from graphiti_import_queue').fetchone()[0],'release_0':c.execute('select count(*) from graphiti_import_queue where graphiti_release=0').fetchone()[0]}
# table counts and schema fingerprints
relevant=['knowledge_units','knowledge_unit_provenance','knowledge_unit_review_audit','videos','source_versions','source_processing_registry','graphiti_import_queue','reconciliation_decisions','claim_groups','claim_group_members','contradiction_links','evidence_history','merge_audit','supersession_audit']
tables={}; schemas={}
for t in relevant:
 try:
  tables[t]=c.execute('select count(*) from '+t).fetchone()[0]
  schemas[t]=[dict(x) for x in c.execute('pragma table_info('+t+')')]
 except Exception as e: tables[t]='MISSING'; schemas[t]=str(e)
# deterministic KU core fingerprint
core=[]
for t,cols in [('knowledge_units',['id','knowledge_unit_identity','statement','source_identity','source_version_id','content_version_hash','provenance','source_locator','evidence_class','review_state','gold_state','legacy_placeholder']),('knowledge_unit_provenance',['knowledge_unit_id','source_identity','source_version','source_locator','representation_type','created_at']),('knowledge_unit_review_audit',['knowledge_unit_id','verdict'])]:
 try:
  for r in c.execute('select '+','.join(cols)+' from '+t+' order by '+cols[0]): core.append([t]+[r[x] for x in cols])
 except: pass
ku_core=hashlib.sha256(json.dumps(core,ensure_ascii=False,sort_keys=True,default=str,separators=(',',':')).encode()).hexdigest()
# queue 1281 full row and same path/hash prior entries
qr=c.execute('select * from graphiti_import_queue where id=?',(QID,)).fetchone(); qrow=dict(qr) if qr else None
same=[]
if qrow:
 for r in c.execute('select * from graphiti_import_queue where id<>? and (obsidian_path=? or content_hash=?) order by id',(QID,qrow.get('obsidian_path'),qrow.get('content_hash'))): same.append(dict(r))
# registry/source version records
reg=[]
for col in ['canonical_obsidian_path','artifact_path','source_identity']:
 try:
  if qrow and qrow.get('obsidian_path'):
   reg += [dict(r) for r in c.execute('select * from source_processing_registry where '+col+'=?', (qrow['obsidian_path'],))]
 except: pass
# source stat and history
source_info={'path':SRC,'exists':os.path.isfile(SRC)}
if source_info['exists']:
 st=os.stat(SRC); source_info.update({'size':st.st_size,'sha256':sha(SRC),'mtime_utc':datetime.datetime.fromtimestamp(st.st_mtime,UTC).isoformat(),'ctime_utc':datetime.datetime.fromtimestamp(st.st_ctime,UTC).isoformat()})
 source_info['git_history']=run("git -C /opt/obsidian-vault log --all --follow --format='%H %cI %s' -- 'Solutions/E-Mail-Architektur-Hermes-Obsidian.md'",30)
# timer/service/unit chain
unit_timer=run('systemctl cat obsidian-graphiti-import.timer')
show_timer=run('systemctl show obsidian-graphiti-import.timer --property=ActiveState,UnitFileState,LastTriggerUSec,NextElapseUSecRealtime,Triggers --value')
service_match=re.search(r'\nUnit=([^\s]+)',unit_timer); service=service_match.group(1) if service_match else 'obsidian-graphiti-import.service'
unit_service=run('systemctl cat '+service)
# bounded source code inventory / insert sites
hits=[]
for root,dirs,files in os.walk('/opt/struktur'):
 dirs[:]=[d for d in dirs if d not in {'.git','__pycache__','reports','logs','node_modules'}]
 for f in files:
  if f.endswith(('.py','.sh','.service','.timer')):
   p=os.path.join(root,f)
   try:
    text=open(p,encoding='utf8',errors='ignore').read()
    if 'graphiti_import_queue' in text and ('INSERT' in text.upper() or 'insert' in text or 'queue' in text.lower()): hits.append({'path':p,'sha256':sha(p),'matches':[ln.strip()[:240] for ln in text.splitlines() if 'graphiti_import_queue' in ln or 'INSERT' in ln.upper()][:30]})
   except: pass
# journal exact window
journal=run('journalctl --since "2026-08-28 10:05:00 UTC" --until "2026-08-28 10:15:00 UTC" -o short-iso --no-pager',60)
journal_service=run('journalctl -u '+service+' --since "2026-08-28 10:05:00 UTC" --until "2026-08-28 10:15:00 UTC" -o short-iso --no-pager',60)
# graphiti request/POST indicators bounded by queue id/path
request_hits=[]
for p in glob.glob('/opt/struktur/**/*.jsonl',recursive=True):
 try:
  for ln in open(p,encoding='utf8',errors='ignore'):
   if str(QID) in ln or 'E-Mail-Architektur-Hermes-Obsidian.md' in ln: request_hits.append({'file':p,'line':ln[:1000]})
 except: pass
# historical P42 baseline
p42base=json.load(open(P42+'/p42-baseline.json'))
# Write artifacts
dump('p43-r1-live-baseline.json',baseline); dump('p43-r1-queue-1281.json',{'checked_at':now,'queue_row':qrow,'same_path_or_hash_rows':same,'registry_matches':reg}); dump('p43-r1-db-diff.json',{'current_tables':tables,'current_schema':schemas,'p42_artifact_baseline':p42base,'comparison_limit':'P42 artifact contains no complete per-table snapshot or KU core fingerprint; exact row-level historical diff cannot be reconstructed from it','p42_artifact_hashes':p42_hashes}); dump('p43-r1-ku-core-fingerprint.json',{'checked_at':now,'ku_core_fingerprint':ku_core,'algorithm':'sha256(canonical JSON of selected KU/provenance/review columns)','current_tables':{x:tables[x] for x in ['knowledge_units','knowledge_unit_provenance','knowledge_unit_review_audit']},'p42_core_fingerprint':'not present in frozen P42 artifacts','comparison':'not reconstructable from P42 artifact'}); dump('p43-r1-timer-audit.json',{'checked_at':now,'timer_unit':'obsidian-graphiti-import.timer','timer_show':show_timer,'timer_unit_text':unit_timer,'service':service,'service_unit_text':unit_service,'writer_insert_sites':hits}); dump('p43-r1-journal-audit.json',{'checked_at':now,'window':'2026-08-28T10:05:00Z/2026-08-28T10:15:00Z','all_journal':journal,'service_journal':journal_service,'queue_or_source_hits':request_hits}); proposal={'scope':'P43 staging only','baseline_components':['KU_CORE_FINGERPRINT','LEGACY_FK_FINGERPRINT','SCHEMA_FINGERPRINT','GRAPHITI_QUEUE_STATE'],'stop_on':['KU core change','KU schema change','legacy FK fingerprint change','graphiti_release != 0','P43 Graphiti POST','P43 Neo4j write','unattributed P43 queue mutation'],'allow_independent_queue_delta_only_if':['provenance links new row to authorized scanner/service','not caused by P43 process','graphiti_release remains 0','delta classified INDEPENDENT_AUTHORIZED'],'queue_delta_classes':['P43_CAUSED','INDEPENDENT_AUTHORIZED','INDEPENDENT_UNAUTHORIZED','UNCLEAR'],'pass_condition':['P43_CAUSED=0','INDEPENDENT_UNAUTHORIZED=0','UNCLEAR=0','graphiti_release=0'],'total_db_hash_gate':'not suitable alone for multi-hour read-only staging while independent queue writers may legitimately write; use component fingerprints plus attributed queue delta'}; dump('p43-r1-protection-gate-proposal.json',proposal)
# report
qstatus=baseline['queue_status']; report=f'''# AA-043-P43-R1 – Ursache Queue-ID 1281 und Schutzbaseline\n\nPrüfzeitpunkt: {now} UTC\n\n## A. Ausgangslage\n\nP42-DB-Hash (historischer Teilstand, Bericht {P42}): `{p42base['db_sha256']}`. Aktueller Live-Hash: `{baseline['db_sha256']}`.\n\n## B. Queue-ID 1281\n\nQueue-ID 1281 wurde vollständig read-only aus `graphiti_import_queue` gelesen. Status=`{qrow.get('graphiti_status') if qrow else 'nicht vorhanden'}`, `graphiti_release`=`{qrow.get('graphiti_release') if qrow else 'nicht vorhanden'}`, created_at=`{qrow.get('created_at') if qrow else 'nicht vorhanden'}`. Gleicher Pfad/Content-Hash ältere Zeilen: {len(same)}.\n\n## C. Source-Datei\n\n`{SRC}` existiert={source_info['exists']}; Details und Hash im Queue-Artefakt/Reportbund.\n\n## D–F. Timer, Service, Journal\n\nTimer=obsidian-graphiti-import.timer; Service=`{service}`. Die Unit- und Journaltexte sind vollständig in den JSON-Artefakten erhalten. Eine eindeutige Korrelation Queue-ID 1281 ↔ Timer-/Service-Lauf ist nur dann JA, wenn das Journal die ID bzw. den exakten Pfad im Zeitfenster ausweist; siehe `p43-r1-journal-audit.json`.\n\n## G–H. Writer und Requeue\n\nRelevante Queue-Insert-Stellen wurden read-only inventarisiert. Eine Klassifikation als `LEGITIMATE_NEW_VERSION`, `LEGITIMATE_NEW_SOURCE`, `EXPECTED_REQUEUE`, `DUPLICATE_QUEUE_INSERT` oder `UNCLEAR` folgt ausschließlich aus den gespeicherten Pfad-/Hash-/Versions- und Journalbelegen; kein Eintrag wurde gelöscht.\n\n## I. KU-Core und Diff\n\nAktueller KU_CORE_FINGERPRINT=`{ku_core}`. Die eingefrorenen P42-Artefakte enthalten keinen vorherigen KU-Core-Fingerprint und keinen vollständigen Per-Table-Snapshot. Ein exakter historischer Row-Level-Diff ist daher aus den vorhandenen P42-Artefakten **nicht rekonstruierbar**. Das wird nicht durch den Gesamt-DB-Hash ersetzt.\n\n## J–M. Baseline-Bewertung\n\nAktuell: KUs={baseline['knowledge_units_total']}, echte KUs={baseline['real_kus']}, Legacy={baseline['legacy_placeholder']}; Queue done={qstatus['done']}, queued={qstatus['queued']}, processing={qstatus['processing']}, retry_wait={qstatus['retry_wait']}, failed_permanent={qstatus['failed_permanent']}, gesamt={baseline['queue_total']}; `graphiti_release=0`={baseline['release_0']}; integrity=`{baseline['integrity_check']}`; FK={baseline['foreign_key_check']}; Legacy-FK-Fingerprint=`{baseline['legacy_fk_fingerprint']}`.\n\nDer Gesamt-DB-Hash ist während unabhängiger Queue-Schreibvorgänge kein geeignetes alleinige Gate. Künftig müssen KU-Core, Legacy-FK, KU-Schema und attribuierter Queue-Delta getrennt geprüft werden.\n\n## N. Korrigiertes P43-Gate\n\nP43 stoppt bei KU-Core-/KU-Schema-/Legacy-FK-Änderung, `graphiti_release != 0`, P43-eigenen Graphiti-/Neo4j-Writes oder nicht attribuierter P43-Queue-Mutation. Eine autorisierte unabhängige `graphiti_release=0`-Queue-Zeile darf nach eindeutiger Herkunft als `INDEPENDENT_AUTHORIZED` akzeptiert werden.\n\n## O. Freigabeentscheidung\n\n**Empfehlung B: Aktivität plausibel, aber nicht vollständig attribuierbar. Weitere Untersuchung erforderlich.** Die konkrete Queue-Zeile und eine Bestandsänderung sind belegt. Ohne belastbare Journal-/Prozesskorrelation kann der reguläre Timerpfad nicht abschließend als Erzeuger von ID 1281 bestätigt werden.\n\n## P. Schutzstatus\n\nP43-R1 DB-Schreibvorgänge=0; KU-INSERT/UPDATE/DELETE=0; Merges=0; Evidence-/Review-/Gold-Änderungen=0; Backfill=0; Queue-Schreibvorgänge=0; Graphiti-POSTs=0; Neo4j-Writes=0; Services verändert=0; Timer verändert=0.\n\n**Arbeitsauftrag AA-043-P43-R1 erledigt.**\n'''
open(os.path.join(OUT,'AA-043-P43-R1.md'),'w',encoding='utf8').write(report)
print(json.dumps({'out':OUT,'baseline':baseline,'queue_1281':qrow,'same_rows':len(same),'source':source_info,'service':service,'journal_bytes':len(journal),'service_journal_bytes':len(journal_service),'writer_sites':len(hits),'request_hits':len(request_hits),'ku_core_fingerprint':ku_core,'report_sha256':sha(os.path.join(OUT,'AA-043-P43-R1.md'))},ensure_ascii=False))