#!/usr/bin/env python3
"""Offline retirement gate. Fictional, normalized evidence; no HubSpot client.

Only immutable local values are returned. There is no persistence, authentication,
HTTP, ERP operation or concurrent execution guarantee. A generation is local to
the connector, not a native HubSpot version or an event timestamp.
"""
from dataclasses import dataclass, replace, asdict
import json


@dataclass(frozen=True)
class Gate:
    portal: str
    contact: str
    status: str = "active"
    generation: int = 0
    revision: int = 0
    reason: str = "initial"


@dataclass(frozen=True)
class Job:
    portal: str
    contact: str
    generation: int


@dataclass(frozen=True)
class Evidence:
    portal: str
    requested_contact: str
    requested_archived: bool
    http_status: int
    returned_contact: str | None
    archived: bool | None
    generation: int
    transport_qualified: bool


def same(gate, portal, contact):
    return (gate.portal, gate.contact) == (portal, contact)


def pause(gate, reason):
    if gate.status == "privacy_hold":
        return gate
    if gate.status == "paused" and gate.reason == reason:
        return gate
    return replace(gate, status="paused", reason=reason,
                   generation=gate.generation + 1, revision=gate.revision + 1)


def notification(gate, portal, contact, kind):
    """Accept already qualified identity/type, not a raw webhook request."""
    if not same(gate, portal, contact):
        return gate
    if kind == "contact.privacyDeletion":
        if gate.status == "privacy_hold":
            return gate
        return replace(gate, status="privacy_hold", reason="privacy_review",
                       generation=gate.generation + 1, revision=gate.revision + 1)
    if kind == "contact.deletion":
        return pause(gate, "deletion_signal")
    # Property/creation messages cannot lift a retirement gate.
    return gate


def search_absence(gate, portal, contact):
    if not same(gate, portal, contact):
        return gate
    # Local precaution: absence is unresolved, never proof of an archive.
    return pause(gate, "unresolved_absence")


def valid_read(gate, evidence, archived):
    return (same(gate, evidence.portal, evidence.requested_contact)
            and evidence.returned_contact == gate.contact
            and evidence.generation == gate.generation
            and evidence.transport_qualified is True
            and type(evidence.http_status) is int and evidence.http_status == 200
            and evidence.requested_archived is archived
            and evidence.archived is archived)


def observe_archive(gate, evidence):
    if valid_read(gate, evidence, True):
        return pause(gate, "archive_confirmed")
    return gate


def resume(gate, evidence, expected_revision, operator_approved):
    if (gate.status != "paused" or type(expected_revision) is not int
            or expected_revision != gate.revision or operator_approved is not True
            or not valid_read(gate, evidence, False)):
        return gate
    return replace(gate, status="active", reason="reviewed_resumption",
                   generation=gate.generation + 1, revision=gate.revision + 1)


def may_prepare(gate, job):
    return (gate.status == "active" and same(gate, job.portal, job.contact)
            and type(job.generation) is int
            and job.generation == gate.generation)


def run_checks():
    names = []

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

    g = Gate("PORTAL-A", "901")
    old_job = Job(g.portal, g.contact, 0)
    retired = notification(g, g.portal, g.contact, "contact.deletion")
    read = Evidence(g.portal, g.contact, False, 200, g.contact, False,
                    retired.generation, True)
    reopened = resume(retired, read, retired.revision, True)
    check("initial_job_can_be_prepared", may_prepare(g, old_job))
    check("deletion_signal_pauses", retired.status == "paused")
    check("notification_is_not_archive_proof", retired.reason == "deletion_signal")
    check("pending_job_is_blocked", not may_prepare(retired, old_job))
    check("generation_changes_on_pause", retired.generation == 1)
    check("duplicate_signal_keeps_gate", notification(retired, g.portal, g.contact, "contact.deletion") == retired)
    check("late_property_cannot_resume", notification(retired, g.portal, g.contact, "contact.propertyChange") == retired)
    check("creation_message_cannot_resume", notification(retired, g.portal, g.contact, "contact.creation") == retired)
    check("other_portal_signal_isolated", notification(g, "PORTAL-B", g.contact, "contact.deletion") == g)
    check("other_contact_signal_isolated", notification(g, g.portal, "902", "contact.deletion") == g)
    absent = search_absence(g, g.portal, g.contact)
    check("absence_pauses_by_local_policy", absent.status == "paused")
    check("absence_remains_unresolved", absent.reason == "unresolved_absence")
    check("other_portal_absence_isolated", search_absence(g, "PORTAL-B", g.contact) == g)
    archived = replace(read, requested_archived=True, archived=True)
    confirmed = observe_archive(retired, archived)
    check("exact_archive_read_is_confirmed", confirmed.reason == "archive_confirmed")
    check("confirmed_archive_stays_paused", confirmed.status == "paused")
    check("archive_read_from_old_generation_rejected", observe_archive(retired, replace(archived, generation=0)) == retired)
    check("wrong_returned_id_rejected", observe_archive(retired, replace(archived, returned_contact="902")) == retired)
    check("wrong_requested_id_rejected", observe_archive(retired, replace(archived, requested_contact="902")) == retired)
    check("wrong_portal_read_rejected", observe_archive(retired, replace(archived, portal="PORTAL-B")) == retired)
    check("404_is_not_archive_proof", observe_archive(retired, replace(archived, http_status=404)) == retired)
    check("403_is_not_archive_proof", observe_archive(retired, replace(archived, http_status=403)) == retired)
    check("missing_archive_flag_rejected", observe_archive(retired, replace(archived, archived=None)) == retired)
    check("string_archive_flag_rejected", observe_archive(retired, replace(archived, archived="true")) == retired)
    check("wrong_query_mode_rejected", observe_archive(retired, replace(archived, requested_archived=False)) == retired)
    check("unqualified_transport_rejected", observe_archive(retired, replace(archived, transport_qualified=False)) == retired)
    check("active_read_alone_does_not_resume", observe_archive(retired, read) == retired)
    check("operator_approval_required", resume(retired, read, retired.revision, False) == retired)
    check("truthy_string_is_not_approval", resume(retired, read, retired.revision, "yes") == retired)
    check("stale_revision_cannot_resume", resume(retired, read, 0, True) == retired)
    check("stale_read_cannot_resume", resume(retired, replace(read, generation=0), retired.revision, True) == retired)
    check("archived_read_cannot_resume", resume(retired, archived, retired.revision, True) == retired)
    check("404_cannot_resume", resume(retired, replace(read, http_status=404), retired.revision, True) == retired)
    check("wrong_contact_cannot_resume", resume(retired, replace(read, returned_contact="902"), retired.revision, True) == retired)
    check("qualified_review_resumes", reopened.status == "active")
    check("resumption_starts_new_generation", reopened.generation == 2)
    check("pre_retirement_job_stays_blocked_after_resume", not may_prepare(reopened, old_job))
    check("pause_generation_job_stays_blocked_after_resume", not may_prepare(reopened, Job(g.portal, g.contact, 1)))
    check("fresh_generation_job_can_be_prepared", may_prepare(reopened, Job(g.portal, g.contact, 2)))
    check("other_portal_job_blocked", not may_prepare(reopened, Job("PORTAL-B", g.contact, 2)))
    check("other_contact_job_blocked", not may_prepare(reopened, Job(g.portal, "902", 2)))
    privacy = notification(retired, g.portal, g.contact, "contact.privacyDeletion")
    privacy_first = notification(g, g.portal, g.contact, "contact.privacyDeletion")
    check("privacy_escalates_existing_pause", privacy.status == "privacy_hold")
    check("privacy_then_deletion_keeps_hold", notification(privacy_first, g.portal, g.contact, "contact.deletion") == privacy_first)
    check("privacy_duplicates_keep_hold", notification(privacy, g.portal, g.contact, "contact.privacyDeletion") == privacy)
    check("active_read_cannot_clear_privacy_hold", resume(privacy, replace(read, generation=privacy.generation), privacy.revision, True) == privacy)
    check("archive_read_cannot_clear_privacy_hold", observe_archive(privacy, replace(archived, generation=privacy.generation)) == privacy)
    check("late_deletion_after_resume_pauses_again", notification(reopened, g.portal, g.contact, "contact.deletion").status == "paused")
    check("already_active_cannot_reuse_resume", resume(reopened, replace(read, generation=2), reopened.revision, True) == reopened)
    check("inputs_not_mutated", g == Gate("PORTAL-A", "901") and old_job.generation == 0)
    return {
        "checks_passed": len(names), "check_names": names,
        "scope": {"http_calls": 0, "erp_operations": 0, "storage_writes": 0,
                  "native_authentication_verified": False,
                  "concurrent_dispatch_guaranteed": False},
        "examples": {"initial": asdict(g), "retired": asdict(retired),
                     "archive_confirmed": asdict(confirmed),
                     "resumed": asdict(reopened), "privacy_hold": asdict(privacy),
                     "old_job_allowed_after_resume": may_prepare(reopened, old_job)}
    }


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