FIX7 P0 Final Pre-Real-Data — surrogate_rehearsal.py
#!/usr/bin/env python3
-- coding: ascii --
"""FIX7 P0 - rehearsal on the GENERATED production-shaped surrogate.
Proves, on an isolated generated surrogate (NOT real production data): before -> gated birth apply -> after_apply (hash differs) -> rollback (snapshot restore) -> after_rollback (hash equals before)
Fail-closed guards (each refuses with a REJECT_* token and exit 1):
- target path must live under an allowed staging root and must not look production-like;
- the DB must carry the surrogate provenance marker (GENERATED_SURROGATE_NOT_REAL_PRODUCTION_DUMP) with real_production_data=NO / contains_secrets=NO / contains_pii=NO;
- the birth mutation only runs if the modeled Tier-0 gate holds (db_target=directus AND approvals>=1 for the named proposal).
No production contact, no network, no secrets, no REAL_RUN/QT001/cutover. Emits rollback-evidence.json (hardened-validator schema) and surrogate-rehearsal-execution-evidence.json. """ import hashlib import json import os import shutil import sqlite3 import sys
MARKER = "GENERATED_SURROGATE_NOT_REAL_PRODUCTION_DUMP" PROPOSAL = "SURR-PROPOSAL-APPROVED" REHEARSAL_OBJECT_ID = "FIX7-P0-SURR-BIRTH-REHEARSAL-0001" ALLOWED_STAGING_ROOTS = ("/private/tmp/", "/tmp/", "/var/folders/") PRODUCTION_LIKE_TOKENS = ("prod", "live", "directus-vps", "incomex-vps")
def sha256_file(path): with open(path, "rb") as fh: return hashlib.sha256(fh.read()).hexdigest()
def logical_dump_sha(conn): dump = "\n".join(conn.iterdump()) return hashlib.sha256(dump.encode("utf-8")).hexdigest()
def row_counts(conn): counts = {} for t in ("birth_registry", "os_proposal_approvals", "directus_objects", "system_issues", "rollback_anchors"): counts[t] = conn.execute("SELECT COUNT(*) FROM " + t).fetchone()[0] return counts
def check_target_path(db_path): """Refuse production-looking or non-staging targets. Fail-closed.""" rp = os.path.realpath(db_path) if not rp.startswith(ALLOWED_STAGING_ROOTS): return "REJECT_TARGET_NOT_IN_STAGING_ROOT: " + rp low = rp.lower() for tok in PRODUCTION_LIKE_TOKENS: if tok in low: return "REJECT_TARGET_LOOKS_PRODUCTION_LIKE: token '%s' in path" % tok return None
def check_surrogate_marker(conn): """The DB must prove it is a generated surrogate. Fail-closed.""" try: row = conn.execute( "SELECT marker, real_production_data, contains_secrets, contains_pii " "FROM surrogate_provenance").fetchone() except sqlite3.Error: return "REJECT_NOT_A_SURROGATE: surrogate_provenance table missing" if row is None: return "REJECT_NOT_A_SURROGATE: surrogate_provenance empty" marker, real_prod, secrets, pii = row if marker != MARKER: return "REJECT_SURROGATE_PROVENANCE_INVALID: marker mismatch" if (real_prod, secrets, pii) != ("NO", "NO", "NO"): return ("REJECT_SURROGATE_PROVENANCE_INVALID: " "real_production_data/contains_secrets/contains_pii must all be NO") return None
def tier0_gate_ok(conn, proposal_id): """Model of the governed Tier-0 executor gate: db=directus AND os_proposal_approvals>=1. Fail-closed.""" n = conn.execute( "SELECT COUNT(*) FROM os_proposal_approvals " "WHERE proposal_id=? AND approved=1 AND db_target='directus'", (proposal_id,)).fetchone()[0] return n >= 1
def apply_birth(conn, proposal_id): """The rehearsal mutation: a gated object birth (INSERTs only).""" if not tier0_gate_ok(conn, proposal_id): raise SystemExit("REJECT_TIER0_GATE_NOT_SATISFIED: proposal " + proposal_id) payload_sha = hashlib.sha256( ("FIX7-P0-SURROGATE-BIRTH::" + REHEARSAL_OBJECT_ID).encode("ascii")).hexdigest() c = conn.cursor() c.execute("INSERT INTO birth_registry " "(object_id, object_kind, status, content_sha256, content_bytes, " " created_at, updated_at, born_by, approval_id) " "VALUES (?,?,?,?,?,?,?,?,?)", (REHEARSAL_OBJECT_ID, "rehearsal-birth", "BORN", payload_sha, 4242, "2026-06-12T00:01:00Z", "2026-06-12T00:01:00Z", "SURROGATE_REHEARSAL", 1)) c.execute("INSERT INTO directus_objects (collection, item_pk, payload_sha256, created_at) " "VALUES (?,?,?,?)", ("objects", REHEARSAL_OBJECT_ID, payload_sha, "2026-06-12T00:01:00Z")) c.execute("INSERT INTO system_issues (issue_code, severity, object_id, opened_at, status) " "VALUES (?,?,?,?,?)", ("SURR-ISSUE-BIRTH-NOTE", "info", REHEARSAL_OBJECT_ID, "2026-06-12T00:01:00Z", "OPEN")) conn.commit() return payload_sha
def main(): if len(sys.argv) != 3: print("usage: surrogate_rehearsal.py <surrogate_db_path> <out_dir>") sys.exit(2) db_path, out_dir = sys.argv[1], sys.argv[2]
err = check_target_path(db_path)
if err:
print(err)
sys.exit(1)
if not os.path.exists(db_path):
print("REJECT_SURROGATE_DB_MISSING: " + db_path)
sys.exit(1)
conn = sqlite3.connect(db_path)
err = check_surrogate_marker(conn)
if err:
conn.close()
print(err)
sys.exit(1)
# ---- BEFORE ----
counts_before = row_counts(conn)
dump_before = logical_dump_sha(conn)
conn.close()
before_hash = sha256_file(db_path)
# ---- SNAPSHOT (rollback anchor) ----
snap_path = db_path + ".snapshot"
shutil.copy2(db_path, snap_path)
snap_hash = sha256_file(snap_path)
if snap_hash != before_hash:
print("REJECT_SNAPSHOT_HASH_MISMATCH")
sys.exit(1)
# ---- APPLY (gated birth) ----
conn = sqlite3.connect(db_path)
payload_sha = apply_birth(conn, PROPOSAL)
counts_after_apply = row_counts(conn)
dump_after_apply = logical_dump_sha(conn)
born = conn.execute("SELECT COUNT(*) FROM birth_registry WHERE object_id=?",
(REHEARSAL_OBJECT_ID,)).fetchone()[0]
conn.close()
after_apply_hash = sha256_file(db_path)
if after_apply_hash == before_hash:
print("REJECT_APPLY_DID_NOT_MUTATE")
sys.exit(1)
if born != 1:
print("REJECT_BIRTH_ROW_NOT_PRESENT_AFTER_APPLY")
sys.exit(1)
# ---- ROLLBACK (restore snapshot) ----
shutil.copy2(snap_path, db_path)
after_rollback_hash = sha256_file(db_path)
conn = sqlite3.connect(db_path)
counts_after_rollback = row_counts(conn)
dump_after_rollback = logical_dump_sha(conn)
born_after_rb = conn.execute("SELECT COUNT(*) FROM birth_registry WHERE object_id=?",
(REHEARSAL_OBJECT_ID,)).fetchone()[0]
conn.close()
restored = (after_rollback_hash == before_hash
and dump_after_rollback == dump_before
and counts_after_rollback == counts_before
and born_after_rb == 0)
if not restored:
print("REJECT_ROLLBACK_DID_NOT_RESTORE")
sys.exit(1)
rollback_evidence = {
"doc": "fix7-p0-final-pre-real-data-surrogate-rollback-evidence",
"date": "2026-06-12",
"target": "GENERATED production-shaped surrogate (sqlite), "
"GENERATED_SURROGATE_NOT_REAL_PRODUCTION_DUMP",
"rollback_proof_status": "PROVEN_IN_STAGING",
"staging_mutation_occurred": True,
"production_rollback_status": "NOT_APPLICABLE",
"production_rollback_reason": "no production surface was touched; the "
"rehearsal ran on a generated surrogate in isolated staging; a "
"production rollback proof can only exist after a separately "
"authorized production phase",
"entries": [
{
"id": "RB-SURR-1",
"surface": "surrogate sqlite db FILE (production-shaped surrogate)",
"method": "file snapshot copy -> Tier-0-gated birth INSERTs "
"(birth_registry + directus_objects + system_issues) "
"-> restore snapshot over db",
"before_hash": before_hash,
"after_apply_hash": after_apply_hash,
"after_rollback_hash": after_rollback_hash,
"expected_restored_hash": before_hash,
"restored_match": True,
"verify_after_rollback": "file sha256 == before AND rehearsal "
"birth row absent AND all row counts restored",
},
{
"id": "RB-SURR-2",
"surface": "surrogate LOGICAL content (sqlite iterdump)",
"method": "logical-dump sha256 comparison across "
"before/apply/rollback",
"before_hash": dump_before,
"after_apply_hash": dump_after_apply,
"after_rollback_hash": dump_after_rollback,
"expected_restored_hash": dump_before,
"restored_match": True,
"verify_after_rollback": "logical dump sha256 == before",
},
],
}
execution_evidence = {
"doc": "fix7-p0-final-pre-real-data-surrogate-rehearsal-execution-evidence",
"date": "2026-06-12",
"macro": "FIX7_P0_FINAL_PRE_REAL_DATA_READINESS_LANE_MACRO_2026_06_12",
"surrogate_marker": MARKER,
"surrogate_is_real_production_data": False,
"isolation": {
"db_realpath_root_allowed": True,
"production_like_token_in_path": False,
"network_used": False,
"live_system_contact": False,
"secrets_used": False,
"pii_used": False,
},
"tier0_gate_model": {
"rule": "db_target=directus AND os_proposal_approvals>=1",
"approved_proposal_used": PROPOSAL,
"gate_satisfied_before_apply": True,
"unapproved_and_wrong_db_paths_probed_separately":
"see bad-input-probes.json (must fail closed)",
},
"birth_payload_sha256": payload_sha,
"rehearsal_object_id": REHEARSAL_OBJECT_ID,
"row_counts": {
"before": counts_before,
"after_apply": counts_after_apply,
"after_rollback": counts_after_rollback,
},
"hashes": {
"file": {"before": before_hash, "after_apply": after_apply_hash,
"after_rollback": after_rollback_hash},
"logical_dump": {"before": dump_before, "after_apply": dump_after_apply,
"after_rollback": dump_after_rollback},
"snapshot": snap_hash,
},
"predicates": {
"after_apply_hash_differs_from_before": True,
"after_rollback_hash_equals_before": True,
"rehearsal_row_absent_after_rollback": True,
"row_counts_restored": True,
},
"production_mutation": False,
"real_run": False,
"qt001": False,
"cutover": False,
"ci_triggered": False,
}
for name, payload in (("rollback-evidence.json", rollback_evidence),
("surrogate-rehearsal-execution-evidence.json",
execution_evidence)):
with open(os.path.join(out_dir, name), "w", encoding="ascii") as fh:
json.dump(payload, fh, indent=2, sort_keys=True, ensure_ascii=True)
fh.write("\n")
print("SURROGATE_REHEARSAL: before=%s after_apply=%s after_rollback=%s"
% (before_hash[:12], after_apply_hash[:12], after_rollback_hash[:12]))
print("SURROGATE_REHEARSAL_RESULT: PROVEN_IN_STAGING (apply mutated, "
"rollback restored byte-exact)")
sys.exit(0)
if name == "main": main()