#!/usr/bin/env python3
"""
AA-043-P2S: Remove legacy UNIQUE(obsidian_path,content_hash) from the
productive graphiti_import_queue. Full pre/post verification, transactional
rebuild, abort+restore on any error.
"""
import hashlib
import json
import sqlite3
import subprocess
import sys
DB = "/opt/struktur/youtube-research/knowledge.db"
BK = "/opt/struktur/backups/aa043-p2s-20260822_201448"
report = {}
def sha256(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()
def row_checksum(c):
h = hashlib.sha256()
n = 0
for r in c.execute("SELECT * FROM graphiti_import_queue ORDER BY id"):
h.update(repr(tuple(r)).encode())
n += 1
return h.hexdigest(), n
c = sqlite3.connect(DB, timeout=60)
c.row_factory = sqlite3.Row
c.execute("PRAGMA busy_timeout=60000")
# ---------- PRE ----------
cols_before = [(r["name"], r["type"], r["notnull"], r["dflt_value"], r["pk"])
for r in c.execute("PRAGMA table_info(graphiti_import_queue)")]
idx_before = [dict(r) for r in c.execute("PRAGMA index_list(graphiti_import_queue)")]
fks = [dict(r) for r in c.execute("PRAGMA foreign_key_list(graphiti_import_queue)")]
trig_views = [dict(r) for r in c.execute(
"SELECT type,name FROM sqlite_master WHERE tbl_name='graphiti_import_queue' AND type IN ('trigger','view')")]
chk_before, n_before = row_checksum(c)
print("PRE: columns =", len(cols_before))
print("PRE: rows =", n_before)
print("PRE: checksum =", chk_before[:24])
print("PRE: indexes =", sorted(r["name"] for r in idx_before))
print("PRE: fks =", len(fks), "| triggers/views =", len(trig_views))
report.update({"cols_before": len(cols_before), "rows_before": n_before,
"chk_before": chk_before})
# ensure no writer holds a lock
try:
c.execute("BEGIN IMMEDIATE")
except sqlite3.OperationalError as e:
print("LOCKED - STOP:", e); sys.exit(75)
try:
# exact DDL from the real productive schema minus the UNIQUE constraint
c.executescript("""
CREATE TABLE graphiti_import_queue_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_type TEXT NOT NULL,
video_id TEXT,
obsidian_path TEXT NOT NULL,
content_hash TEXT NOT NULL,
file_modified_at TEXT,
graphiti_status TEXT NOT NULL DEFAULT 'pending',
graphiti_attempts INTEGER NOT NULL DEFAULT 0,
graphiti_last_attempt_at TEXT,
graphiti_processed_at TEXT,
graphiti_episode_id TEXT,
graphiti_error TEXT,
next_attempt_at TEXT,
locked_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
attempt_count INTEGER NOT NULL DEFAULT 0,
last_attempt_at TEXT,
last_error_class TEXT,
last_error_message TEXT,
last_http_status INTEGER,
completed_at TEXT,
reconciliation_record_id TEXT,
source_version_id INTEGER,
import_identity TEXT,
knowledge_unit_identity TEXT NOT NULL DEFAULT '__source__',
action TEXT NOT NULL DEFAULT 'CREATE',
payload_hash TEXT,
payload_json TEXT,
lock_owner TEXT,
lock_token TEXT,
lease_expires_at TEXT,
heartbeat_at TEXT,
remote_outcome TEXT,
unknown_since TEXT,
last_lookup_at TEXT,
lookup_attempt_count INTEGER NOT NULL DEFAULT 0,
post_request_id TEXT,
file_state TEXT
);
INSERT INTO graphiti_import_queue_new SELECT * FROM graphiti_import_queue;
DROP TABLE graphiti_import_queue;
ALTER TABLE graphiti_import_queue_new RENAME TO graphiti_import_queue;
CREATE UNIQUE INDEX uq_queue_decision ON graphiti_import_queue(reconciliation_record_id, action);
CREATE UNIQUE INDEX uq_queue_identity_action_version_unit ON graphiti_import_queue(import_identity, action, source_version_id, knowledge_unit_identity);
CREATE UNIQUE INDEX uq_queue_version_action_unit ON graphiti_import_queue(source_version_id, action, knowledge_unit_identity);
CREATE INDEX idx_graphiti_queue_due ON graphiti_import_queue(graphiti_status, next_attempt_at);
""")
c.commit()
except Exception as e:
print("MIGRATION FAILED - ROLLBACK:", e)
c.rollback()
c.close()
# restore from backup
subprocess.run(["cp", "-a", BK + "/knowledge.db", DB], check=True)
print("RESTORED from backup; md5 now:")
subprocess.run(["md5sum", DB])
sys.exit(1)
# ---------- POST ----------
cols_after = [(r["name"], r["type"], r["notnull"], r["dflt_value"], r["pk"])
for r in c.execute("PRAGMA table_info(graphiti_import_queue)")]
chk_after, n_after = row_checksum(c)
schema = c.execute("SELECT sql FROM sqlite_master WHERE name='graphiti_import_queue'").fetchone()[0]
legacy_gone = "UNIQUE(obsidian_path,content_hash)" not in schema
idx_after = sorted(r[0] for r in c.execute(
"SELECT name FROM sqlite_master WHERE tbl_name='graphiti_import_queue' AND type='index' AND name LIKE 'uq_%'"))
due_idx = c.execute("SELECT 1 FROM sqlite_master WHERE name='idx_graphiti_queue_due'").fetchone() is not None
row34 = dict(c.execute("SELECT id,reconciliation_record_id,action,graphiti_status FROM graphiti_import_queue WHERE id=34").fetchone())
integrity = c.execute("PRAGMA integrity_check").fetchone()[0]
fk_errors = list(c.execute("PRAGMA foreign_key_check"))
print("POST: columns =", len(cols_after))
print("POST: rows =", n_after)
print("POST: checksum =", chk_after[:24])
print("POST: legacy UNIQUE removed =", legacy_gone)
print("POST: AA-043 indexes =", idx_after)
print("POST: due index present =", due_idx)
print("POST: row34 =", row34)
print("POST: integrity_check =", integrity)
print("POST: foreign_key_check errors =", len(fk_errors))
assert cols_before == cols_after, "column definitions changed!"
assert n_before == n_after and chk_before == chk_after, "DATA CHANGED!"
assert legacy_gone
assert idx_after == ["uq_queue_decision", "uq_queue_identity_action_version_unit", "uq_queue_version_action_unit"]
assert due_idx
assert integrity == "ok"
assert not fk_errors
# behavioral check inside a safe savepoint (rolled back immediately)
c.execute("SAVEPOINT behavior_test")
try:
c.execute("""INSERT INTO graphiti_import_queue(source_type,obsidian_path,content_hash,
graphiti_status,created_at,updated_at,reconciliation_record_id,
source_version_id,import_identity,knowledge_unit_identity,action)
VALUES('obsidian','/opt/obsidian-vault/Solutions/SOL-005-content-briefing-modul.md',
'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa','queued',
'2026-01-01','2026-01-01','aa043:obsidian-general:test',NULL,'t','__source__','CREATE')""")
print("BEHAVIOR: insert with legacy path/hash OK")
finally:
c.execute("ROLLBACK TO behavior_test"); c.release("behavior_test")
n_final = c.execute("SELECT COUNT(*) FROM graphiti_import_queue").fetchone()[0]
assert n_final == n_before
c.close()
json.dump(report, open(BK + "/p2s-report.json", "w"), indent=2)
print("MIGRATION SUCCESS")