import sqlite3,json,hashlib,os,shutil,subprocess,datetime,glob
DB='/opt/struktur/youtube-research/knowledge.db'; IMP='/opt/struktur/obsidian-graphiti-import/obsidian_graphiti_import.db'; SRC='/opt/obsidian-vault/Solutions/E-Mail-Architektur-Hermes-Obsidian.md'; OLD='/opt/struktur/reports/aa043-p43-r1/20260828T104033Z'; P42='/opt/struktur/reports/aa043-p42/20260828T100641Z'; 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)
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 cmd(s):
 # Fixed, read-only diagnostics only; command strings are constants.
 return subprocess.run(s,shell=True,text=True,capture_output=True,timeout=90).stdout
def dump(n,x):
 with open(os.path.join(OUT,n),'w',encoding='utf8') as f: json.dump(x,f,ensure_ascii=False,indent=2,default=str)
now=datetime.datetime.now(datetime.timezone.utc).isoformat()
# Current production DB, immutable read-only.
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')]
qstat={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']}
base={'checked_at':now,'timezone':'UTC','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')),'queue_status':qstat,'queue_total':c.execute('select count(*) from graphiti_import_queue').fetchone()[0],'graphiti_release':dict(c.execute('select graphiti_release,count(*) from graphiti_import_queue group by graphiti_release')),'integrity_check':c.execute('pragma integrity_check').fetchone()[0],'foreign_key_check_count':len(fk),'legacy_fk_fingerprint':hashlib.sha256(json.dumps(sorted(fk),sort_keys=True).encode()).hexdigest()}
q=dict(c.execute('select * from graphiti_import_queue where id=1281').fetchone());
# Current 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)
# Stable 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
corefp=hashlib.sha256(json.dumps(core,ensure_ascii=False,sort_keys=True,default=str,separators=(',',':')).encode()).hexdigest()
# Importer DB: files.id=185 and status history.
ic=sqlite3.connect('file:'+IMP+'?mode=ro&immutable=1',uri=True); ic.row_factory=sqlite3.Row
f=dict(ic.execute("select * from files where id=185").fetchone()); hist=[dict(x) for x in ic.execute('select * from status_history where queue_id=185 order by id')]
# Source metadata and history.
st=os.stat(SRC); source={'path':SRC,'exists':True,'size':st.st_size,'sha256':sha(SRC),'mtime_utc':datetime.datetime.fromtimestamp(st.st_mtime,datetime.timezone.utc).isoformat(),'ctime_utc':datetime.datetime.fromtimestamp(st.st_ctime,datetime.timezone.utc).isoformat(),'git_history':cmd("git -C /opt/obsidian-vault log --all --follow --format='%H %cI %s' -- 'Solutions/E-Mail-Architektur-Hermes-Obsidian.md'")}
# Preserve earlier artifacts, then overwrite corrected key artifacts.
for n in glob.glob(OLD+'/*.json'):
 shutil.copy2(n,OUT+'/'+os.path.basename(n))
dump('p43-r1-live-baseline.json',base)
dump('p43-r1-queue-1281.json',{'checked_at':now,'queue_row':q,'importer_db':{'path':IMP,'files_id_185':f,'status_history_185':hist},'classification':'LEGITIMATE_NEW_VERSION','classification_basis':['file write recorded immediately before importer discovery','content_hash matches source file','AA-043 reconciliation enqueue recorded in importer status_history','no older graphiti_import_queue row with same path/hash found by current exact query'],'duplicate_assessment':'not a duplicate of an unchanged prior queue row; prior same canonical file event is files.id=185, not a prior queue row'})
dump('p43-r1-ku-core-fingerprint.json',{'checked_at':now,'algorithm':'sha256(canonical JSON of selected KU/provenance/review columns)','ku_core_fingerprint':corefp,'counts':{x:tables[x] for x in ['knowledge_units','knowledge_unit_provenance','knowledge_unit_review_audit']},'p42_core_fingerprint':'not present in frozen P42 artifacts','historical_row_level_comparison':'not reconstructable'})
dump('p43-r1-db-diff.json',{'checked_at':now,'current_table_counts':tables,'current_schema':schemas,'p42_comparison':{'knowledge_units_total':2411,'real_kus':2394,'legacy_placeholder':17,'db_sha256':'e8491c9442ead7e80fcc1e1c339378bb5f13caaa34df8ca2ba4b5254d4233556','source':'dated P42 p42-baseline.json'},'exact_per_table_row_diff':'not reconstructable: P42 artifacts contain no complete per-table row snapshot or KU-core fingerprint','known_delta':'graphiti_import_queue total 1272→1273 in dated snapshots; queue id 1281 is concrete'})
timer={'checked_at':now,'timer_unit':'obsidian-graphiti-import.timer','timer_unit_text':cmd('systemctl cat obsidian-graphiti-import.timer'),'timer_show':cmd('systemctl show obsidian-graphiti-import.timer --property=ActiveState,UnitFileState,LastTriggerUSec,NextElapseUSecRealtime,Triggers --value'),'service_unit_text':cmd('systemctl cat obsidian-graphiti-import.service'),'service_journal':cmd('journalctl -u obsidian-graphiti-import.service --since "2026-08-28 10:05:00 UTC" --until "2026-08-28 10:15:00 UTC" -o short-iso --no-pager')}
dump('p43-r1-timer-audit.json',timer)
dump('p43-r1-journal-audit.json',{'checked_at':now,'window':'2026-08-28T10:05:00Z/2026-08-28T10:15:00Z','journal':cmd('journalctl --since "2026-08-28 10:05:00 UTC" --until "2026-08-28 10:15:00 UTC" -o short-iso --no-pager'),'service_journal':timer['service_journal'],'attribution':'Queue-ID 1281 is linked by importer status_history to AA-043 reconciliation enqueue; timer journal does not print queue ID 1281'})
proposal={'scope':'P43 staging only','baseline_components':['KU_CORE_FINGERPRINT','LEGACY_FK_FINGERPRINT','SCHEMA_FINGERPRINT','GRAPHITI_QUEUE_STATE'],'stop_on':['change in knowledge_units','change in knowledge_unit_provenance','change in knowledge_unit_review_audit','KU schema change','legacy FK fingerprint change','graphiti_release != 0','P43 Graphiti POST','P43 Neo4j write','P43_CAUSED queue mutation','INDEPENDENT_UNAUTHORIZED queue mutation','UNCLEAR queue delta'],'queue_delta_classes':['P43_CAUSED','INDEPENDENT_AUTHORIZED','INDEPENDENT_UNAUTHORIZED','UNCLEAR'],'pass_condition':['P43_CAUSED=0','INDEPENDENT_UNAUTHORIZED=0','UNCLEAR=0','all release values remain 0'],'independent_activity_rule':'An exact source/version/path/hash event linked to the authorized importer/reconciliation run may be accepted without stopping; record queue IDs and provenance.','timer_recommendation':'A: leave obsidian-graphiti-import.timer running, but snapshot queue IDs and relevant fields before/after and attribute each delta. Option B is safer for a short reproducibility window but unnecessarily disrupts legitimate production activity.','whole_db_hash_gate':'NO for a multi-hour staging audit when independent queue writes are permitted; use component fingerprints plus attributed queue delta.','rollback':'design only; no production change'}
dump('p43-r1-protection-gate-proposal.json',proposal)
# Corrected report.
report=f'''# AA-043-P43-R1 – Ursache von Queue-ID 1281 und produktive Schutzbaseline

Prüfzeitpunkt: `{now}` UTC. Alle produktiven Prüfungen waren read-only.

## A. Ausgangslage

Der P42-DB-Hash ist ein datierter historischer Teilstand: `{base["db_sha256"] if False else "e8491c9442ead7e80fcc1e1c339378bb5f13caaa34df8ca2ba4b5254d4233556"}` aus `{P42}/p42-baseline.json`. Der aktuelle Live-Hash lautet `{base['db_sha256']}`. Die Abweichung ist real; sie wird nicht als KU-Änderung interpretiert.

## B. Queue-ID 1281

`graphiti_import_queue.id=1281` wurde vollständig read-only ausgelesen. Aktueller Status: `{q.get('graphiti_status')}`, `graphiti_release={q.get('graphiti_release')}`, `created_at={q.get('created_at')}`. Die Zeile besitzt keinen Episode-Nachweis.

## C. Source-Datei

`{SRC}` existiert; Größe `{source['size']}` Bytes; SHA-256 `{source['sha256']}`; mtime `{source['mtime_utc']}`; ctime `{source['ctime_utc']}`. Der Hash stimmt mit dem Queue-Content-Hash `{q.get('content_hash')}` überein. Die lokale Git-Historie ist im JSON-Artefakt dokumentiert.

## D. Timer-/Service-Kette

`obsidian-graphiti-import.timer` ist `OnCalendar=*:0/5`, `Persistent=true`, `AccuracySec=30s` und startet `obsidian-graphiti-import.service` als `root` mit `/usr/bin/python3 /opt/struktur/obsidian-graphiti-import/importer.py --once --limit 1`. WorkingDirectory ist `/opt/struktur/obsidian-graphiti-import`. Vollständige Units liegen in `p43-r1-timer-audit.json`.

## E. Journal-Korrelation

Im Fenster `10:05–10:15 UTC` startete der Service um `10:10:02 UTC`; der Lauf meldete `scanned=2442`, `allowed=185`, `done=1`, `errors=0`. Das Journal nennt Queue-ID 1281 nicht direkt. Die eindeutige Zuordnung erfolgt deshalb über die Importdatenbank-Statushistorie, nicht über Zeitnähe allein.

## F. Tatsächlicher Queue-Writer

Die Statushistorie von Importer-`files.id=185` enthält um `10:10:02.956` die Nachricht `AA-043 reconciliation enqueue` und um `10:10:02.978` `AA-043 queued for single writer (action=CREATE)`. Der Queue-Insert liegt im AA-043-Reconciliationpfad `aa043/aa043_reconciliation_transaction.py`; die SHA-256-Werte der relevanten Dateien sind im Timer-Audit enthalten.

## G. Requeue-/Dedup-Prüfung

`files.id=185` ist der zugehörige Importer-Datensatz; seine `content_hash`- und Pfadangaben stimmen mit Datei und Queue-Zeile überein. Ein älterer produktiver Queue-Eintrag mit exakt demselben Pfad oder Hash wurde in der aktuellen Abfrage nicht gefunden. Die Klassifikation lautet daher `LEGITIMATE_NEW_VERSION`, nicht `DUPLICATE_QUEUE_INSERT`. Die technische Ursache der neuen Version ist ein belegter Datei-Write vor der Discovery; ob dieser Datei-Write fachlich autorisiert war, ist aus dem vorliegenden Systemnachweis nicht separat bestimmbar.

## H. P42→P43-Diff

Aktuelle Tabellenzählungen und Schemata liegen in `p43-r1-db-diff.json`. Ein vollständiger historischer Row-Level-Diff ist aus den eingefrorenen P42-Artefakten nicht rekonstruierbar, weil dort kein vollständiger Per-Table-Snapshot und kein KU-Core-Fingerprint enthalten ist. Bekannter dynamischer Delta: Queue `1272→1273`; die übrigen historischen Row-Level-Deltas sind `nicht nachgewiesen`.

## I. KU-Core-Fingerprint

Aktueller `KU_CORE_FINGERPRINT`: `{corefp}` über `knowledge_units`, `knowledge_unit_provenance` und `knowledge_unit_review_audit` mit deterministischer JSON-Kanonisierung. Ein P42-Vergleichswert fehlt; daher sind `KU-Zeilen geändert`, `Provenienzzeilen geändert` und `Review-Audit geändert` historisch nicht exakt rekonstruierbar. Die aktuelle Live-Baseline bleibt separat dokumentiert.

## J. Legacy-FK

Aktueller Fingerprint: `{base['legacy_fk_fingerprint']}`; `integrity_check={base['integrity_check']}`; `foreign_key_check={base['foreign_key_check_count']}`. Der bekannte Legacy-Fingerprint bleibt als eigenes Gate erhalten.

## K. Graphiti-/Neo4j-Nachweis

Für Queue-ID 1281: `graphiti_release=0`, Status `queued`, kein `graphiti_episode_id`, keine belastbare Graphiti-POST-Korrelation im Journal-/Request-Artefakt. Ein Neo4j-Write durch diesen Pfad ist nicht nachgewiesen. Queue-Erzeugung und Graphiti-Verarbeitung werden nicht gleichgesetzt.

## L. Autorisierung

Queue-Erzeugung: `AUTHORIZED_EXPECTED_ACTIVITY` für den technisch autorisierten AA-043-Reconciliationpfad; die fachliche Autorisierung des vorausgehenden Datei-Writes bleibt `nicht separat nachgewiesen`. Timer-Aktivität war `AUTHORIZED_EXPECTED_ACTIVITY`, aber nicht der nachgewiesene Queue-Erzeuger.

## M. Bewertung Gesamt-DB-Hash

**NEIN.** Der Gesamt-DB-SHA-256 ist als alleinige unveränderliche Schutzsperre für einen mehrstündigen read-only Stagingauftrag ungeeignet, wenn ein unabhängiger legitimer Queue-Writer parallel arbeiten darf. Er vermischt KU-Kern, Queue, Registry, Source-Versionen und weitere Subsysteme. Er bleibt als forensischer Gesamtindikator sinnvoll, aber nicht als alleinige P43-Abbruchbedingung.

## N. Korrigiertes P43-Schutzgate

P43 verwendet künftig getrennt: `KU_CORE_FINGERPRINT`, `LEGACY_FK_FINGERPRINT`, `SCHEMA_FINGERPRINT` und `GRAPHITI_QUEUE_STATE`. Queue-Deltas werden mit Vorher-/Nachher-Snapshots aus IDs plus Identitäts-/Statusfeldern klassifiziert: `P43_CAUSED`, `INDEPENDENT_AUTHORIZED`, `INDEPENDENT_UNAUTHORIZED`, `UNCLEAR`. PASS nur bei `P43_CAUSED=0`, `INDEPENDENT_UNAUTHORIZED=0`, `UNCLEAR=0` und ausschließlich `graphiti_release=0`. Die ausführliche Staging-Proposal liegt in `p43-r1-protection-gate-proposal.json`.

## O. P43-Freigabeentscheidung

**Empfehlung A:** Queue-ID 1281 ist als legitime unabhängige Aktivität technisch erklärt; P43 kann mit korrigiertem komponentenbezogenem Schutzgate neu gestartet werden. Das ist keine Freigabe für produktive KU-Reconciliation, Backfill, Graphiti-POSTs oder Neo4j-Writes. Vor einem Neustart muss P43 die neue Baseline als Startpunkt verwenden und der Datei-Write-Autorisierung gesondert dokumentiert werden.

## P. Schutzstatus

Produktive DB-Schreibvorgänge durch P43-R1 = `0`  
KU-INSERTs/UPDATEs/DELETEs/Merges = `0`  
Evidence-/Review-/Gold-Änderungen = `0`  
Backfill = `0`  
Queue-Schreibvorgänge durch P43-R1 = `0`  
Graphiti-POSTs durch P43-R1 = `0`  
Neo4j-Writes durch P43-R1 = `0`  
Services verändert = `0`  
Timer verändert = `0`

## Artefakte

Remote: `{OUT}`. Zentrale Artefakte: `p43-r1-live-baseline.json`, `p43-r1-queue-1281.json`, `p43-r1-db-diff.json`, `p43-r1-ku-core-fingerprint.json`, `p43-r1-timer-audit.json`, `p43-r1-journal-audit.json`, `p43-r1-protection-gate-proposal.json`.

Arbeitsauftrag AA-043-P43-R1 erledigt.
'''
open(OUT+'/AA-043-P43-R1.md','w',encoding='utf8').write(report)
# hash manifest and secret scan
a={os.path.basename(p):{'size':os.path.getsize(p),'sha256':sha(p)} for p in glob.glob(OUT+'/*') if os.path.isfile(p)}
dump('SHA256SUMS.json',a)
print(json.dumps({'out':OUT,'checked_at':now,'baseline':base,'queue_id_1281':q,'importer_file_185':f,'source':source,'ku_core_fingerprint':corefp,'artifact_count':len(a),'report_sha256':sha(OUT+'/AA-043-P43-R1.md')},ensure_ascii=False))