"""Deterministic refresh-rotation event model; no network or real credentials."""
from dataclasses import dataclass, field
import json
from pathlib import Path


@dataclass
class StrictFamily:
    current: str = "family-A:0"
    used: set = field(default_factory=set)
    revoked: bool = False
    calls: int = 0
    replays: int = 0

    def refresh(self, supplied):
        self.calls += 1
        if self.revoked:
            return None
        if supplied in self.used:
            self.replays += 1
            self.revoked = True
            return None
        if supplied != self.current:
            return None
        self.used.add(supplied)
        generation = int(supplied.rsplit(":", 1)[1]) + 1
        self.current = f"family-A:{generation}"
        return self.current

    def usable(self, token):
        return not self.revoked and token == self.current


@dataclass
class Store:
    token: str = "family-A:0"
    version: int = 0
    state: str = "ready"

    def snapshot(self):
        return self.version, self.token

    def install(self, expected_version, token):
        if self.version != expected_version:
            return False
        self.token = token
        self.version += 1
        self.state = "ready"
        return True


def concurrent(coordinated):
    server, store = StrictFamily(), Store()
    first = store.snapshot()
    second = store.snapshot()
    trace = ["Two callers read version 0 and family-A:0"]
    answer = server.refresh(first[1])
    assert answer == "family-A:1"
    assert store.install(first[0], answer)
    trace.append("Owner exchanges family-A:0 and installs version 1")
    if coordinated:
        # This is an idealized single owner, not an implementation of a lock.
        trace.append("Waiter reads the owner's committed result; no second exchange")
    else:
        assert server.refresh(second[1]) is None
        trace.append("Second caller exchanges consumed family-A:0; family revoked")
    return {
        "case": "coordinated-callers" if coordinated else "duplicate-callers",
        "provider_calls": server.calls,
        "replay_events": server.replays,
        "family_revoked": server.revoked,
        "stored_token_usable": server.usable(store.token),
        "trace": trace,
    }


def late_result(guarded):
    server, store = StrictFamily(), Store()
    version, token = store.snapshot()
    delayed_answer = server.refresh(token)
    assert delayed_answer == "family-A:1"
    # A new authorization grant replaces this connection before the old reply.
    assert store.install(version, "family-B:0")
    expected = store.token
    if guarded:
        installed = store.install(version, delayed_answer)
        assert installed is False
    else:
        store.token = delayed_answer
        store.version += 1
        installed = True
    return {
        "case": "late-result-version-guard" if guarded else "late-result-unconditional",
        "provider_calls": server.calls,
        "replay_events": server.replays,
        "old_result_installed": installed,
        "reconnected_grant_preserved": store.token == expected,
        "trace": ["Old refresh accepted, response delayed", "Reconnect stores family-B:0 at version 1",
                  "Old response rejected by version guard" if guarded else "Old response overwrites new grant"],
    }


def lost_response(retry):
    server, store = StrictFamily(), Store()
    version, token = store.snapshot()
    lost_answer = server.refresh(token)
    assert lost_answer == "family-A:1"
    # The client never receives the accepted response. A timeout is ambiguous.
    store.state = "uncertain"
    trace = ["Exchange accepted: server advances to family-A:1", "Response lost: client still stores family-A:0"]
    if retry:
        assert server.refresh(token) is None
        store.state = "reauthorization-required"
        trace.append("Blind retry of family-A:0 triggers replay revocation")
    else:
        trace.append("No blind retry; connection remains uncertain")
    assert store.version == version
    return {
        "case": "lost-response-blind-retry" if retry else "lost-response-stop",
        "provider_calls": server.calls,
        "replay_events": server.replays,
        "family_revoked": server.revoked,
        "stored_token_usable": server.usable(store.token),
        "client_state": store.state,
        "trace": trace,
    }


def main():
    rows = [concurrent(False), concurrent(True), late_result(False),
            late_result(True), lost_response(True), lost_response(False)]
    assert rows[0]["provider_calls"] == 2 and rows[0]["family_revoked"]
    assert rows[1]["provider_calls"] == 1 and rows[1]["stored_token_usable"]
    assert not rows[2]["reconnected_grant_preserved"]
    assert rows[3]["reconnected_grant_preserved"]
    assert rows[4]["family_revoked"] and not rows[4]["stored_token_usable"]
    assert not rows[5]["family_revoked"] and not rows[5]["stored_token_usable"]
    result = {
        "model": "Strict one-use refresh rotation with family revocation on replay",
        "scenario_count": len(rows),
        "scenarios": rows,
        "limits": ["Constructed event schedules, not a production incident",
                   "No HTTP, actual threads, distributed lock, expiry or cryptography",
                   "Provider grace periods and recovery APIs are not simulated",
                   "No assertion about access-token revocation at resource servers"],
    }
    Path(__file__).with_name("results.json").write_text(json.dumps(result, indent=2) + "\n")
    print("PASS: six event schedules; duplicate and blind-retry paths revoke the modeled family.")


if __name__ == "__main__":
    main()