#!/usr/bin/env python3
"""Fictional Intercom part references, latest-window audit and real in-memory SQL.

Tested separately from any REST client. No message bodies, HTTP, Intercom/CRM/ERP
writes, durable archive or full-history recovery. SQLite writes below are real,
but confined to explicitly created in-memory databases with memory temp storage.
"""
from collections import Counter
from dataclasses import dataclass, replace
import json
import sqlite3
import sys


CAP = 500


@dataclass(frozen=True)
class Scope:
    workspace: str
    conversation: str
    version: str
    visibility_profile: str


@dataclass(frozen=True)
class Window:
    scope: Scope
    part_ids: tuple[str, ...]
    source_id: str | None
    access_qualified: bool
    transport_qualified: bool
    reported_total_count: object = None


def text(value):
    return isinstance(value, str) and bool(value.strip())


def key(scope):
    return (scope.workspace, scope.conversation, scope.version, scope.visibility_profile)


def validate(requested, window):
    if (not all(text(v) for v in key(requested)) or requested.version != "2.10"):
        return "requested_scope_outside_local_profile"
    if window.scope != requested:
        return "scope_changed"
    if window.access_qualified is not True or window.transport_qualified is not True:
        return "observation_unqualified"
    if (not isinstance(window.part_ids, tuple) or len(window.part_ids) > CAP
            or not all(text(v) for v in window.part_ids)
            or len(set(window.part_ids)) != len(window.part_ids)
            or (window.source_id is not None and not text(window.source_id))):
        return "reference_window_outside_local_profile"
    return None


def make_database(trace, databases):
    db = sqlite3.connect(":memory:", isolation_level=None)
    db.set_trace_callback(trace.append)
    db.execute("PRAGMA temp_store=MEMORY")
    db.execute("""CREATE TABLE observed_refs (
        workspace TEXT NOT NULL, conversation TEXT NOT NULL,
        version TEXT NOT NULL, visibility_profile TEXT NOT NULL,
        kind TEXT NOT NULL CHECK(kind IN ('source','part')),
        identifier TEXT NOT NULL,
        PRIMARY KEY(workspace,conversation,version,visibility_profile,kind,identifier)
    ) WITHOUT ROWID""")
    databases.append(db)
    return db


def identifiers(db, scope, kind="part"):
    return frozenset(r[0] for r in db.execute("""SELECT identifier FROM observed_refs
        WHERE workspace=? AND conversation=? AND version=?
        AND visibility_profile=? AND kind=?""", key(scope)+(kind,)))


def insert_refs(db, scope, ids, kind="part"):
    db.executemany("""INSERT INTO observed_refs VALUES(?,?,?,?,?,?)
        ON CONFLICT(workspace,conversation,version,visibility_profile,kind,identifier)
        DO NOTHING""", [key(scope)+(kind, value) for value in ids])


def seed(db, scope, parts, source):
    db.execute("BEGIN")
    try:
        insert_refs(db, scope, parts)
        insert_refs(db, scope, (source,), "source")
        db.execute("COMMIT")
    except Exception:
        db.execute("ROLLBACK")
        raise


def audit_and_merge(db, requested, window):
    reason = validate(requested, window)
    if reason:
        return {"decision": "held", "reason": reason}
    known = identifiers(db, requested)
    observed = frozenset(window.part_ids)
    db.execute("BEGIN")
    try:
        insert_refs(db, requested, window.part_ids)
        if window.source_id is not None:
            insert_refs(db, requested, (window.source_id,), "source")
        db.execute("COMMIT")
    except Exception:
        db.execute("ROLLBACK")
        raise
    return {"decision": "reference_inventory_updated",
            "window_state": "at_cap" if len(observed) == CAP else "below_cap",
            "observed_parts": len(observed), "known_parts_before": len(known),
            "known_parts_after": len(identifiers(db, requested)),
            "newly_observed_ids": sorted(observed-known),
            "known_ids_not_in_window": sorted(known-observed),
            "history_complete": False,
            "native_deletions_inferred": 0,
            "reported_total_count_used_as_history_proof": False}


def destructive_counterexample(db, scope, window):
    """Intentionally wrong comparison, executed only on the bad fixture DB."""
    assert validate(scope, window) is None
    db.execute("CREATE TEMP TABLE current_window(identifier TEXT PRIMARY KEY)")
    db.executemany("INSERT INTO current_window VALUES(?)", [(v,) for v in window.part_ids])
    db.execute("BEGIN")
    try:
        db.execute("""DELETE FROM observed_refs WHERE workspace=? AND conversation=?
            AND version=? AND visibility_profile=? AND kind='part'
            AND identifier NOT IN (SELECT identifier FROM current_window)""", key(scope))
        insert_refs(db, scope, window.part_ids)
        db.execute("COMMIT")
    except Exception:
        db.execute("ROLLBACK")
        raise


def run_checks():
    names, trace, databases = [], [], []

    def check(name, condition):
        if name in names or not condition:
            raise AssertionError(name)
        names.append(name)

    def parts(start, stop):
        # Fixture ordinals construct our sample; native IDs are not a clock.
        return tuple("p"+str(n).zfill(3) for n in range(start, stop))

    try:
        scope = Scope("WORKSPACE-A", "CONVERSATION-A", "2.10", "fixture-visible")
        older = parts(1, 501)
        recent = parts(2, 502)
        window = Window(scope, recent, "initial-1", True, True, 500)
        good, bad = make_database(trace, databases), make_database(trace, databases)
        for db in (good, bad):
            seed(db, scope, older, "initial-1")
        check("seed_has_500_known_parts", len(identifiers(good, scope)) == 500)
        check("initial_source_is_separate", len(identifiers(good, scope, "source")) == 1)
        first = audit_and_merge(good, scope, window)
        destructive_counterexample(bad, scope, window)
        check("actual_sql_preserves_older_reference", "p001" in identifiers(good, scope))
        check("actual_bad_sql_erases_older_reference", "p001" not in identifiers(bad, scope))
        check("good_inventory_has_501_known_parts", len(identifiers(good, scope)) == 501)
        check("bad_inventory_stays_plausibly_at_500", len(identifiers(bad, scope)) == 500)
        check("new_reference_is_received", first["newly_observed_ids"] == ["p501"])
        check("omitted_known_reference_is_named", first["known_ids_not_in_window"] == ["p001"])
        check("cap_is_flagged", first["window_state"] == "at_cap")
        check("cap_is_not_completeness", first["history_complete"] is False)
        check("no_deletion_inferred_from_absence", first["native_deletions_inferred"] == 0)
        check("native_total_count_not_used_as_proof", first["reported_total_count_used_as_history_proof"] is False)
        before_repeat = good.total_changes
        second = audit_and_merge(good, scope, window)
        check("repeat_adds_no_sql_rows", good.total_changes == before_repeat)
        check("repeat_adds_no_new_reference", second["newly_observed_ids"] == [])
        check("repeat_keeps_older_reference", "p001" in identifiers(good, scope))
        check("order_change_does_not_replace_membership", audit_and_merge(good, scope, replace(window, part_ids=tuple(reversed(recent)))) == second)
        check("larger_reported_count_does_not_create_proof", audit_and_merge(good, scope, replace(window, reported_total_count=900)) == second)
        check("missing_reported_count_does_not_create_proof", audit_and_merge(good, scope, replace(window, reported_total_count=None)) == second)
        below = audit_and_merge(good, scope, replace(window, part_ids=recent[:-1]))
        check("below_cap_is_reported_separately", below["window_state"] == "below_cap")
        check("below_cap_not_full_history_certificate", below["history_complete"] is False)
        check("below_cap_keeps_all_known_references", below["known_parts_after"] == 501)
        empty = audit_and_merge(good, scope, replace(window, part_ids=(), source_id=None))
        check("empty_window_does_not_delete_parts", empty["known_parts_after"] == 501)
        check("absent_initial_source_does_not_delete_it", identifiers(good, scope, "source") == {"initial-1"})
        other_workspace = replace(scope, workspace="WORKSPACE-B")
        seed(good, other_workspace, ("p001",), "initial-1")
        check("same_native_id_in_other_workspace_is_distinct", len(identifiers(good, other_workspace)) == 1 and len(identifiers(good, scope)) == 501)
        bad_windows = [
            ("workspace", replace(window, scope=other_workspace), "scope_changed"),
            ("conversation", replace(window, scope=replace(scope, conversation="OTHER")), "scope_changed"),
            ("version", replace(window, scope=replace(scope, version="2.17")), "scope_changed"),
            ("visibility", replace(window, scope=replace(scope, visibility_profile="OTHER")), "scope_changed"),
            ("access", replace(window, access_qualified=False), "observation_unqualified"),
            ("transport", replace(window, transport_qualified=False), "observation_unqualified"),
            ("truthy_access", replace(window, access_qualified="yes"), "observation_unqualified"),
            ("duplicate", replace(window, part_ids=("p001", "p001")), "reference_window_outside_local_profile"),
            ("blank_id", replace(window, part_ids=(" ",)), "reference_window_outside_local_profile"),
            ("boolean_id", replace(window, part_ids=(True,)), "reference_window_outside_local_profile"),
            ("list_input", replace(window, part_ids=["p001"]), "reference_window_outside_local_profile"),
            ("over_cap", replace(window, part_ids=parts(1, 502)), "reference_window_outside_local_profile"),
            ("blank_source", replace(window, source_id=" "), "reference_window_outside_local_profile"),
        ]
        changes_before_rejections = good.total_changes
        for label, supplied, reason in bad_windows:
            check(label+"_is_held_before_sql", audit_and_merge(good, scope, supplied) == {"decision": "held", "reason": reason})
        check("held_observations_write_no_rows", good.total_changes == changes_before_rejections)
        check("other_workspace_keeps_its_reference", identifiers(good, other_workspace) == {"p001"})
        unsupported = replace(scope, version="Preview")
        check("preview_outside_local_profile", audit_and_merge(good, unsupported, replace(window, scope=unsupported))["reason"] == "requested_scope_outside_local_profile")
        blank_scope = replace(scope, workspace=" ")
        check("blank_requested_workspace_held", audit_and_merge(good, blank_scope, replace(window, scope=blank_scope))["reason"] == "requested_scope_outside_local_profile")
        exact = make_database(trace, databases)
        history_exact = recent
        history_larger = ("p001",)+recent
        # Synthetic tail selection demonstrates ambiguity, not a native API call.
        from_exact = replace(window, part_ids=history_exact[-CAP:])
        from_larger = replace(window, part_ids=history_larger[-CAP:])
        exact_observation = audit_and_merge(exact, scope, from_exact)
        check("exact_500_and_larger_history_share_same_observation", len(history_exact) == 500 and len(history_larger) == 501 and from_exact == from_larger)
        check("without_prior_inventory_no_omission_can_be_named", exact_observation["known_ids_not_in_window"] == [])
        check("500_only_inventory_still_not_full_history", exact_observation["history_complete"] is False)
        namespaced = make_database(trace, databases)
        one = replace(window, part_ids=("same-id",), source_id="same-id")
        audit_and_merge(namespaced, scope, one)
        check("source_and_part_use_separate_namespaces", identifiers(namespaced, scope) == {"same-id"} and identifiers(namespaced, scope, "source") == {"same-id"})
        check("source_not_added_to_part_cap", len(identifiers(namespaced, scope)) == 1)
        check("all_databases_are_in_memory", all(all(row[2] == "" for row in db.execute("PRAGMA database_list")) for db in databases))
        check("actual_sql_delete_counterexample_was_executed", any(q.lstrip().startswith("DELETE FROM observed_refs") for q in trace))
        examples = {"first_merge": first,
                    "good_known_parts": len(identifiers(good, scope)),
                    "bad_known_parts": len(identifiers(bad, scope)),
                    "known_older_reference_lost_by_bad_sql": "p001" not in identifiers(bad, scope),
                    "initial_source_counted_separately": len(identifiers(good, scope, "source")),
                    "unproven_500_only": exact_observation}
        counts = dict(Counter(q.split()[0].upper() for q in trace if q.strip()))
        return {"checks_passed": len(names), "check_names": names,
                "python_version": sys.version, "sqlite_version": sqlite3.sqlite_version,
                "scope": {"http_calls": 0, "intercom_operations": 0,
                          "crm_erp_operations": 0, "persistent_sqlite_files": 0,
                          "message_bodies": 0, "in_memory_sql_executed": True,
                          "in_memory_sql_statements_traced": len(trace),
                          "sql_statement_counts": counts,
                          "native_history_completeness_certified": False},
                "examples": examples}
    finally:
        for db in databases:
            db.close()


if __name__ == "__main__":
    print(json.dumps(run_checks(), ensure_ascii=False, indent=2))