#!/usr/bin/env python3
"""Independent offline journal verification and adversarial mutations; no DB/HTTP/writes."""
import copy, hashlib, json, struct
from pathlib import Path
R = Path(__file__).resolve().parent
P = json.loads((R / "probe-plan.json").read_bytes())
assert hashlib.sha256((R / "probe-plan.json").read_bytes()).hexdigest() == "68124d2e3fdafa943e70c8c4a4f0b4e0c4dfb67098bb67df8615d7d1da41fc94"
CHECKS = 0
def check(ok, msg):
    global CHECKS
    CHECKS += 1
    if not ok:
        raise AssertionError(msg)
def load(name, sha):
    data = (R / "private-evidence" / name).read_bytes()
    check(hashlib.sha256(data).hexdigest() == sha, name + " immutable bytes SHA")
    es = [json.loads(line) for line in data.splitlines()]
    check([e["seq"] for e in es] == list(range(1, len(es) + 1)), name + " complete sequence")
    return es
B = load("recovered-execute-B.jsonl", "4b12bf80307f4a74fbd015080f69206c35c7f29055c329103263faf6c78cd216")
C = load("final-resumed-C.jsonl", "0cf489264234b72a8230524a27482d620c1ffdb3de4984776feaed8f227da0e0")
C0 = load("recovered-execute-C.jsonl", "97b988ac3f753f8cbda366936d9b163c14957109b26eec29d1c06f8e1cd36724")
C1 = load("catalog-amended-execute-C.jsonl", "580914984d3d85418c90fb2588f116210e5d90b856556d3eee5a42918772b8b6")
for fn, sha in [
 ("initial-execute-B.jsonl", "60d92424f7935aefeaa69cb9b2e23187d4f9a6c37d0c3d8e8e7545c20d3d5c9a"),
 ("initial-execute-C.jsonl", "e1b6dea76ee6656a51650b699b994062c91d0b7fec02b0fcba765f4790400740")]:
    es = load(fn, sha)
    check(not any(e["event"].startswith(("read-", "DML-", "DDL-")) for e in es), "Config errors before DB operations")
def ev(es, name):
    return [e["data"] for e in es if e["event"] == name]
def metadata_cycles(es, lane, existing=False):
    sqls = [q["sql"] for q in P["lanes"][lane]["five_float_group"]["schema_fingerprint_queries"]]
    if lane == "C":
        a = json.loads((R / "catalog-amendment-C.json").read_bytes())
        replacements = {x["original_sql"]: x["approved_replacement_sql"] for x in a["replacements"]}
        sqls = [replacements.get(s, s) for s in sqls]
    active, current, cycles = existing, [], []
    for e in es:
        if e["event"] == "DDL-success":
            active = e["data"]["sql"].startswith("CREATE ")
        if active and e["event"] == "read-success":
            d = e["data"]
            if d["sql"] == sqls[0]:
                current = [d["rows"]]
            elif current and len(current) < len(sqls) and d["sql"] == sqls[len(current)]:
                current.append(d["rows"])
            if len(current) == len(sqls):
                cycles.append(current); current = []
    return cycles
def verify_natural_rows(before_events, after_events, lane):
    sqls = [q["sql"] for q in P["lanes"][lane]["natural_protection"]["queries"]]
    if lane == "C":
        sqls = [s.replace("SELECT TRIGGER_NAME,TABLE_NAME,STATUS,TRIGGER_TYPE,TRIGGERING_EVENT,TRIGGER_BODY FROM USER_TRIGGERS", "SELECT * FROM USER_TRIGGERS") for s in sqls]
    before = [x["rows"] for x in ev(before_events, "read-success") if x["sql"] in sqls][:10]
    after = [x["rows"] for x in ev(after_events, "read-success") if x["sql"] in sqls][-10:]
    check(len(before) == len(after) == 10 and before == after, lane + " complete actual natural metadata/aggregate arrays equal")
def verify_lane(es, lane):
    spec = P["lanes"][lane]["five_float_group"]
    steps = spec["steps"]
    check([x["case_id"] for x in ev(es, "case-start")] == [x["case_id"] for x in steps], lane + " all48 ordered cases")
    starts = ev(es, "DML-start")
    expected = [(st["case_id"], op) for st in steps for op in st["DML"]]
    check(len(starts) == 72, lane + " exactly72 DML attempts")
    for actual, (case, planned) in zip(starts, expected):
        check(actual["case_id"] == case and actual["role"] == planned["role"], "DML case/role/order")
        check(actual["sql"] == planned["sql"], "Exact approved DML SQL")
        check([x["original"] for x in actual["parameters"]] == planned["bindings"], "Every original bind/value/type preserved")
        check(actual.get("column_original_values") == planned.get("column_original_values"), "Original five input lexemes")
        for actual_bind in actual["parameters"]:
            orig = actual_bind["original"]
            rt = actual_bind["runtime"]
            check(actual_bind["pdo_type_code"] == orig["pdo_type_code"], "PDO type exact")
            if orig["php_type"] == "float":
                check(rt["php_type"] == "double" and rt["native_double_hex"] == struct.pack("d", float(orig["original_value"])).hex(), "Original float conversion; no pre-round")
            else:
                check(rt["php_type"] == orig["php_type"] and rt["value"] == orig["original_value"], "String/NULL/int type/value exact")
    check(len(ev(es, "DML-success")) == 72 and not ev(es, "DML-failure"), lane + " actual DML outcomes")
    check(len(ev(es, "rollback-success")) == 48, lane + " all48 rollback")
    check([x["case_id"] for x in ev(es, "case-complete")] == [x["case_id"] for x in steps], "No omitted completion")
    native = [x for x in ev(es, "read-success") if x["sql"] == spec["statements"]["native"]]
    text = [x for x in ev(es, "read-success") if x["sql"] == spec["statements"]["text"]]
    decimal = [x for x in ev(es, "read-success") if x["sql"] == spec["statements"]["decimal30"]]
    check(len(native) == len(text) == len(decimal) == 48, "All native/text/decimal30 readbacks")
    for row in native:
        check(len(row["rows"]) == 1 and row["rows"][0]["id"] == 1 and row["rows"][0]["marker"] == "CDXNF5D9A4", "Own exact one-row fixture")
        check(set(row["rows"][0]) == {"id", "marker", "power", "carbohydrate", "protein", "fat", "sodium"}, "Complete native row")
    zero = [x for x in ev(es, "read-success") if x["sql"] == spec["statements"]["row_count"]]
    check(len(zero) >= 50 and all(x["rows"][0]["n"] == 0 for x in zero), "Every rollback/independent cleanup count0")
    check(ev(es, "cleanup-complete") == [{"fresh_object_count": 0}], "Fresh catalog object0")
    cleanup = ev(es, "cleanup-ownership-gate")
    check(len(cleanup) == 1 and cleanup[0]["created"] is True and cleanup[0]["independent_zero"] is True, "Create ledger+independent0+definition gate")
    after = ev(es, "natural-after")
    check(len(after) == 1 and after[0]["equals_before"] is True, "Full natural baseline unchanged")
    complete = ev(es, "complete")
    check(len(complete) == 1 and complete[0]["completed_cases"] == 48 and complete[0]["DML_attempts"] == 72 and complete[0]["DDL"] == 2 and complete[0]["rollbacks"] == 48, "Cumulative exact DML/DDL/rollback budgets")
    check(complete[0]["probe_queries"] <= 299 and complete[0]["natural_queries"] <= 40 and complete[0]["objects_removed"], "Caps and cleanup")
    check(all(x["sql"].startswith(("SELECT ", "SHOW WARNINGS", "SHOW CREATE TABLE")) for x in ev(es, "read-start")), "All diagnostic reads nonmutating")
    if lane == "B":
        warns = ev(es, "warnings")
        warn_reads = [x for x in ev(es, "read-success") if x["sql"] == "SHOW WARNINGS"]
        check(len(warns) == len(warn_reads) == 72, "Warnings immediately collected for all72")
        check([x["rows"] for x in warns] == [x["rows"] for x in warn_reads], "No warning hidden")
        check(sum(bool(x["rows"]) for x in warns) == 16 and sum(len(x["rows"]) for x in warns) == 24, "Original16 statements/24 warnings retained")
    else:
        check(not ev(es, "warnings"), "C warning API remains unavailable/unknown")
verify_lane(B, "B")
verify_lane(C, "C")
verify_natural_rows(B, B, "B")
verify_natural_rows(C1, C, "C")
bmeta = metadata_cycles(B, "B")
cmeta = metadata_cycles(C1, "C") + metadata_cycles(C, "C", existing=True)
check(len(bmeta) == 3 and all(x == bmeta[0] for x in bmeta), "B actual full CREATE/cleanup/independent metadata equal")
check(len(cmeta) == 5 and all(x == cmeta[0] for x in cmeta), "C original CREATE/resume/independent/cleanup full metadata equal")
digest = hashlib.sha256(json.dumps(cmeta[0], separators=(",", ":"), ensure_ascii=True).encode()).hexdigest()
check(digest == "ad17ef18c54795295c49f34bcb8e16140bcaa434b9c26be9836a26035a07616e", "Independent original raw seven metadata SHA")
check(len(ev(C0, "read-failure")) == 1 and ev(C0, "read-failure")[0]["error"]["errorInfo"][1] == -2111, "Original invalid catalog projection retained")
check(ev(C1, "stopped")[-1]["error"]["message"] == "C exact defaultNULL/no implicit default", "Original default representation guard red retained")
check(len(ev(C1, "DDL-success")) == 1 and not ev(C1, "DML-start") and len(ev(C, "DDL-success")) == 1, "C CREATE only once and final remaining DROP once")
check(ev(C1, "DDL-success")[0]["sql"] == P["lanes"]["C"]["five_float_group"]["statements"]["create_ordinary"], "Original exact C CREATE")
check(ev(C, "DDL-success")[0]["sql"] == P["lanes"]["C"]["five_float_group"]["statements"]["drop_ordinary"], "Final exact C DROP")
for lane, es in [("B", B), ("C", C0+C1+C)]:
    probe = sum(x["data"]["bucket"] == "probe" for x in es if x["event"] == "read-start")
    natural = sum(x["data"]["bucket"] == "natural" for x in es if x["event"] == "read-start")
    final = ev(B if lane == "B" else C, "complete")[-1]
    check((probe, natural) == (final["probe_queries"], final["natural_queries"]), "Independent cumulative query counts " + lane)
base_checks = CHECKS
mutations = {
 "rounded original input": lambda x: x[0]["data"]["parameters"][2]["original"].update(original_value="12.35"),
}
def mutate_event(es, name, fn):
    fn(next(e["data"] for e in es if e["event"] == name))
tests = [
 ("rounded input", lambda es: mutate_event(es, "DML-start", lambda x: x["parameters"][2]["original"].update(original_value="12.35"))),
 ("PDO type changed", lambda es: mutate_event(es, "DML-start", lambda x: x["parameters"][2].update(pdo_type_code=1))),
 ("runtime host type changed", lambda es: mutate_event(es, "DML-start", lambda x: x["parameters"][2]["runtime"].update(php_type="integer"))),
 ("business DML target", lambda es: mutate_event(es, "DML-start", lambda x: x.update(sql=x["sql"].replace("cdx_nf5_261011_d9a4", "ydy_nutrition")))),
 ("case removed", lambda es: es.remove(next(e for e in es if e["event"] == "case-start"))),
 ("rollback removed", lambda es: es.remove(next(e for e in es if e["event"] == "rollback-success"))),
 ("cleanup missing", lambda es: es.remove(next(e for e in es if e["event"] == "cleanup-complete"))),
 ("natural drift", lambda es: mutate_event(es, "natural-after", lambda x: x.update(equals_before=False))),
 ("warning record hidden", lambda es: es.remove(next(e for e in es if e["event"] == "warnings"))),
 ("budget weakened", lambda es: mutate_event(es, "complete", lambda x: x.update(DML_attempts=71))),
]
rejected = []
for label, mutation in tests:
    es = copy.deepcopy(B);mutation(es)
    try:
        verify_lane(es, "B")
    except (AssertionError, KeyError):
        rejected.append(label)
    else:
        raise AssertionError("Mutation survived " + label)
print(json.dumps({"status": "EXECUTION_EVIDENCE_VERIFIED_NOT_COMPATIBILITY_PASS",
                  "baseline_assertions": base_checks, "negative_mutations_rejected": rejected,
                  "B": ev(B, "complete")[-1], "C_cumulative": ev(C, "complete")[-1],
                  "catalog_and_default_tool_reds_retained": True,
                  "C_warning_status": "unknown_not_zero",
                  "new_DB_HTTP_operations": 0}, ensure_ascii=False, indent=2))