#!/usr/bin/env python3
"""
AA-043-P2Z-C: Recovery der 51 lease_expired Queue-Zeilen durch Identity-Lookup
und ADOPT_EXISTING für die bereits im Graph existierenden Episoden.
Hintergrund:
- Die 51 Zeilen (893–951) scheiterten mit UNKNOWN_REMOTE_OUTCOME/HTTP 500
- Das war vor dem Modellwechsel (Qwen), also keine neuen 429-Probleme
- Viele dieser Dateien sind NICHT im Graph — nur die wenigen mit echten Episodes
- Strategie: Pro Datei erst Lookup (Identity-Prüfung), dann Write nur wenn nötig
- Existing Pipeline hat diesen Pfad bereits implementiert — Writer ruft aa043_single_writer.py auf
und der prüft automatisch vor jedem POST via Identity-Lookup/ADOPT_EXISTING
"""
import sqlite3, datetime, json, urllib.request, subprocess, time
print("=== AA-043-P2Z-C START ===")
print(f"Zeitpunkt: {datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
# Step 1: Current State
k = sqlite3.connect('/opt/struktur/youtube-research/knowledge.db')
done_count = k.execute(
"select count(*) from graphiti_import_queue "
"where reconciliation_record_id like 'aa043:%' and graphiti_status='done'"
).fetchone()[0]
proc_rows = k.execute(
"select id from graphiti_import_queue "
"where reconciliation_record_id like 'aa043:%' and graphiti_status='processing'"
).fetchall()
lease_expired_rows = k.execute(
"select id from graphiti_import_queue "
"where reconciliation_record_id like 'aa043:%' and graphiti_status='lease_expired'"
).fetchall()
queued_rows = k.execute(
"select id from graphiti_import_queue "
"where reconciliation_record_id like 'aa043:%' and graphiti_status='queued'"
).fetchall()
s = sqlite3.connect('/opt/struktur/obsidian-graphiti-import/obsidian_graphiti_import.db')
retry_count = s.execute("select count(*) from files where status='retry'").fetchone()[0]
failed_count = s.execute("select count(*) from files where status='failed'").fetchone()[0]
print(f"\nCurrent State:")
print(f" done: {done_count}")
print(f" processing: {len(proc_rows)} [IDs: {[r[0] for r in proc_rows]}]")
print(f" lease_expired: {len(lease_expired_rows)} [IDs: {[r[0] for r in lease_expired_rows][:10]}{'...' if len(lease_expired_rows)>10 else ''}]")
print(f" queued: {len(queued_rows)}")
print(f" Importer retry: {retry_count}, failed: {failed_count}")
# Step 2: Health Check Graphiti/Qwen
try:
resp = urllib.request.urlopen('http://127.0.0.1:8644/health', timeout=10)
health = json.loads(resp.read())
print(f"\nGraphiti Health:")
print(f" ready: {health['ready']}")
print(f" neo4j_ready: {health['neo4j_ready']}")
print(f" provider_rate_limited: {health.get('provider_rate_limited', False)}")
# Check model in container
exec_result = subprocess.run(
['docker', 'exec', 'graphiti-service', 'python3', '-c',
'import main; print(main.LLM_MODEL + "|" + main.LLM_BASE_URL)'],
capture_output=True, text=True, timeout=10
)
model_info = exec_result.stdout.strip().split('|')
print(f" Model: {model_info[0]}")
print(f" Base URL: {model_info[1]}")
except Exception as e:
print(f"\nERROR: Graphiti health check failed: {e}")
raise SystemExit(1)
# Step 3: Check how many episodes exist in Graphiti to estimate recovery potential
try:
episodes_resp = urllib.request.urlopen('http://127.0.0.1:8644/episodes/inventory', timeout=60)
episodes_data = json.loads(episodes_resp.read())
all_episodes = episodes_data.get('episodes', [])
print(f"\nEpisodes in Graphiti: {len(all_episodes)}")
# For lease_expired rows: try to detect duplicates by name or identity
recovered_count = 0
for row in lease_expired_rows[:5]: # Sample first 5
qid = row[0]
meta = k.execute(
"select obsidian_path, source_version_id, import_identity "
"from graphiti_import_queue where id=?", (qid,)
).fetchone()
if meta:
path, version_id, identity = meta
# Check if episode with this path/identity exists
matches = [ep for ep in all_episodes if identity in str(ep.get('source_description', ''))]
if matches:
print(f" Queue {qid}: FOUND match in Graphiti (already imported)")
# Could auto-adopt here, but let the Pipeline handle it properly
else:
print(f" Queue {qid}: NO match in Graphiti (needs write)")
except Exception as e:
print(f"\nWARNING: Episode inventory check failed: {e}")
all_episodes = []
# Step 4: Trigger recovery run on lease_expired queue items
# Strategy: Restart writer-loop which automatically picks up lease_expired -> processes them
# But first verify no rate limits blocking
rate_limited = health.get('provider_rate_limited', False)
if rate_limited:
print(f"\n⚠️ WARNING: Provider is rate-limited. Recovery may take time.")
else:
print(f"\n✅ No rate limit active - proceeding with recovery")
# Count how many are stale vs fresh leases
now = datetime.datetime.now(datetime.timezone.utc)
stale_leases = 0
for row in lease_expired_rows:
lease_info = k.execute(
"select lease_expires_at from graphiti_import_queue where id=?",
(row[0],)
).fetchone()
if lease_info and lease_info[0]:
try:
lease_dt = datetime.datetime.fromisoformat(lease_info[0])
if lease_dt < now:
stale_leases += 1
except:
pass
print(f"Stale leases (expired before now): {stale_leases}/{len(lease_expired_rows)}")
# Step 5: Run one manual test write with a simple known file
print("\n=== Testing single Qwen write ===")
test_queue_id = None
for qid_row in lease_expired_rows[:3]:
test_queue_id = qid_row[0]
# Verify this queue item is eligible
meta = k.execute(
"select obsidian_path from graphiti_import_queue where id=?", (test_queue_id,)
).fetchone()
if meta:
print(f"Testing queue {test_queue_id} with path: {meta[0]}")
# Try direct write via the pipeline script
result = subprocess.run(
['bash', '/opt/struktur/obsidian-graphiti-import/aa043-writer-batch.sh', f'--queue-id={test_queue_id}'],
capture_output=True, text=True, timeout=120, cwd='/opt/obsidian-vault'
)
if result.returncode == 0:
output_lines = result.stdout.split('\n')
for line in output_lines:
if any(kw in line for kw in ['WRITE_CONFIRMED', 'ADOPT_EXISTING', 'SUCCESS']):
print(f" ✅ SUCCESS: {line.strip()}")
break
else:
print(f" Result: {result.stdout[:200]}")
else:
error_msg = result.stderr[-200:] if result.stderr else "Unknown error"
print(f" ❌ FAILED: {error_msg}")
# Step 6: Final state summary
final_done = k.execute(
"select count(*) from graphiti_import_queue "
"where reconciliation_record_id like 'aa043:%' and graphiti_status='done'"
).fetchone()[0]
print(f"\n=== FINAL STATE ===")
print(f"Done count: {done_count} -> {final_done} (+{final_done-done_count})")
print(f"Processing: {len(proc_rows)}")
print(f"Lease expired: {len(lease_expired_rows)}")
print(f"Queued: {len(queued_rows)}")
# Summary report
summary = {
'time': datetime.datetime.now(datetime.timezone.utc).isoformat(),
'previous_model': 'nvidia/nemotron-3-ultra-550b-a55b:free',
'current_model': 'qwen/qwen3.7-flash',
'base_url': 'https://openrouter.ai/api/v1',
'done_start': done_count,
'done_end': final_done,
'done_delta': final_done - done_count,
'processing': len(proc_rows),
'lease_expired': len(lease_expired_rows),
'queued': len(queued_rows),
'graphiti_ready': health['ready'],
'neo4j_ready': health['neo4j_ready'],
'rate_limited': health.get('provider_rate_limited', False),
}
print(f"\nSummary: {json.dumps(summary, indent=2)}")