"""Cockpit tests: the design law is testable — every panel renders
recomputed evidence with provenance, failures render loud, and the HTTP
surface cannot mutate wallet state (asserted by hashing the wallet
directory before and after a full request sweep, including a POST)."""

import hashlib
import json
import stat
import threading
import urllib.request
import urllib.parse
from pathlib import Path

import pytest

from pacta.quorum import binary_path
from pacta.signing import generate_ed25519_keypair
from pacta.wallet import Wallet
from pacta.walletui import (collect, collect_airgap, collect_incidents,
                            collect_refusals, inspect_receipt, render_incidents,
                            render_inspect, render_posture, render_queue, serve)


def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _fake_member(path: Path, verdict: str) -> None:
    code = {"accept": 0, "reject": 1}[verdict]
    out = {"accept": "OK", "reject": "INVALID"}[verdict]
    path.write_text(f"#!/bin/sh\necho {out}\nexit {code}\n")
    path.chmod(path.stat().st_mode | stat.S_IEXEC)


def _seal_wallet(tmp_path: Path) -> Wallet:
    state_dir = tmp_path / "state"
    state_dir.mkdir(parents=True, exist_ok=True)
    members = []
    for name in ("dalek", "anza"):
        binary = binary_path(name, state_dir)
        _fake_member(binary, "accept")
        members.append({
            "backend": name,
            "component": f"{name}-ed25519-verified",
            "semantics": "test", "entry_point": "test",
            "source_commit": "deadbeef" * 5, "repo_commit": "cafe" * 10,
            "binary_sha256": _sha256(binary.read_bytes()),
            "backend_cfg": "test", "risk_tier": "R4",
            "evidence": {"leaf_hash": "00", "leaf_index": 0, "tree_size": 1,
                         "inclusion_proof": [],
                         "sth": {"timestamp": "2099-01-01T00:00:00Z"}},
        })
    wallet = Wallet(tmp_path / "w")
    for sub in (wallet.keys_dir, wallet.incidents_dir, wallet.receipts_dir,
                wallet.quarantine_dir, wallet.airgap_dir / "outbox",
                wallet.airgap_dir / "inbox"):
        sub.mkdir(parents=True, exist_ok=True)
    capsule = {
        "type": "pacta.wallet.custody_capsule.v1",
        "created_at": "2026-07-06T00:00:00Z", "members": members,
        "policy": {"require_unanimity": True, "min_members": 2,
                   "require_tier": "R4", "freshness_max_age_days": 0},
        "signing": {"backend": "test"}, "problems_at_init": [],
    }
    capsule_bytes = json.dumps(capsule, sort_keys=True, separators=(",", ":")).encode()
    wallet.capsule_path.write_text(json.dumps(capsule, indent=2, sort_keys=True) + "\n")
    wallet._append_ledger("genesis", {"type": "pacta.wallet.ledger_genesis.v1",
                                      "capsule_sha256": _sha256(capsule_bytes)})
    generate_ed25519_keypair(wallet.keys_dir / "warden.key.pem",
                             wallet.keys_dir / "warden.pub.pem")
    return wallet


def _dir_fingerprint(root: Path) -> dict[str, str]:
    out = {}
    for path in sorted(root.rglob("*")):
        if path.is_file():
            out[str(path.relative_to(root))] = _sha256(path.read_bytes())
    return out


def test_posture_renders_recomputed_evidence(tmp_path):
    wallet = _seal_wallet(tmp_path)
    html = render_posture(collect("Wallet.posture()", wallet.posture))
    assert "CUSTODY HEALTHY" in html  # the verdict, in words, before any evidence
    assert "chain verified" in html
    assert "unlatched" in html
    assert "dalek-ed25519-verified" in html and "anza-ed25519-verified" in html
    assert "recomputed" in html and "Wallet.verify_ledger()" in html
    # the panel states its own honesty boundary
    assert "does NOT" in html and "reproducible builds" in html.lower()


def test_broken_ledger_renders_red(tmp_path):
    wallet = _seal_wallet(tmp_path)
    ledger = wallet.dir / "ledger.jsonl"
    ledger.write_text(ledger.read_text().replace("genesis", "gene-sis"), encoding="utf-8")
    html = render_posture(collect("Wallet.posture()", wallet.posture))
    assert "CHAIN BROKEN" in html
    assert "CUSTODY EVIDENCE BROKEN" in html  # verdict banner states it in words


def test_collector_failure_is_loud_not_gray(tmp_path):
    def boom():
        raise RuntimeError("evidence unavailable")
    html = render_posture(collect("Wallet.posture()", boom))
    assert "FAILED TO VERIFY" in html and "evidence unavailable" in html
    assert "pill ok" not in html  # no green anywhere on a failed panel


def test_latched_state_renders_frozen():
    posture = {"ok": True, "via": "x", "data": {
        "capsule_sha256": "ab" * 32,
        "members": [], "policy": {}, "spending_policy": {},
        "latch": {"latched": True, "reason": "quorum divergence",
                  "incident": "incident-1.json", "at": "2026-07-19T00:00:00Z"},
        "ledger": {"entries": 1, "head": "cd" * 32, "chain_ok": True, "problems": []},
        "incidents": 1, "refusal_receipts": 0, "generated_at": "now"}}
    html = render_posture(posture)
    assert "CUSTODY FROZEN" in html  # verdict banner states it in words
    assert "LATCHED" in html and "outbound custody frozen" in html
    assert "quorum divergence" in html and "runbook-latch" in html


def test_queue_lists_airgap_requests(tmp_path):
    wallet = _seal_wallet(tmp_path)
    outbox = wallet.airgap_dir / "outbox"
    (outbox / "req-1.request.json").write_text(json.dumps(
        {"created_at": "2026-07-19T00:00:00Z", "payload_sha256": "aa" * 32}))
    html = render_queue(collect_airgap(wallet))
    assert "req-1" in html and "awaiting device" in html
    (wallet.airgap_dir / "inbox" / "req-1.response.json").write_text("{}")
    html = render_queue(collect_airgap(wallet))
    assert "answered" in html
    # observed, never operated
    assert "OBSERVED" in html


def test_incidents_and_refusals_render_verbatim(tmp_path):
    wallet = _seal_wallet(tmp_path)
    (wallet.incidents_dir / "incident-1.json").write_text(json.dumps(
        {"severity": "tamper", "detail": "member disagreement on payload"}))
    (wallet.receipts_dir / "refusal-1.json").write_text(json.dumps(
        {"code": "CUSTODY_LATCHED", "remediation": "see runbook"}))
    html = render_incidents(collect_incidents(wallet), collect_refusals(wallet))
    assert "incident-1.json" in html and "member disagreement" in html
    assert "refusal-1.json" in html and "CUSTODY_LATCHED" in html


def test_inspect_garbage_fails_closed():
    result = inspect_receipt("not json at all", "{}", "")
    assert result["ok"] is False
    html = render_inspect(result)
    assert "FAILED TO VERIFY" in html
    assert "ACCEPTED" not in html


def test_inspect_real_evidence_roundtrip():
    evidence = Path("examples") / "wallet-evidence"
    key = evidence / "log.pub"
    attestations = sorted(evidence.glob("*attestation*.json")) if evidence.exists() else []
    receipts = sorted(evidence.glob("*receipt*.json")) if evidence.exists() else []
    if not (key.exists() and attestations and receipts):
        pytest.skip("example wallet evidence not present")
    result = inspect_receipt(attestations[0].read_text(), receipts[0].read_text(),
                             key.read_text())
    assert result["ok"] is True
    html = render_inspect(result)
    assert ("ACCEPTED" in html) or ("REJECTED" in html)
    assert "verify_receipt" in html  # provenance names the deployed verifier


def test_server_routes_and_read_only_guarantee(tmp_path):
    wallet = _seal_wallet(tmp_path)
    before = _dir_fingerprint(wallet.dir)
    server = serve(wallet.dir, host="127.0.0.1", port=0)
    port = server.server_address[1]
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        routes = ("/", "/posture", "/queue", "/incidents", "/inspect", "/guide",
                  "/manual", "/deck", "/station/proposer", "/station/quorum",
                  "/station/operator", "/station/operator?probe=1",
                  "/station/operator?pane=1", "/station/cryptographer",
                  "/station/architect", "/station/newcomer")
        for route in routes:
            with urllib.request.urlopen(f"http://127.0.0.1:{port}{route}") as resp:
                body = resp.read().decode()
                assert resp.status == 200
                assert "READ-ONLY" in body
        data = urllib.parse.urlencode(
            {"attestation": "junk", "receipt": "junk", "pubkey": "junk"}).encode()
        with urllib.request.urlopen(f"http://127.0.0.1:{port}/inspect", data=data) as resp:
            assert resp.status == 200
            assert "FAILED TO VERIFY" in resp.read().decode()
    finally:
        server.shutdown()
        thread.join(timeout=5)
    # the whole sweep, POST included, changed not one byte of wallet state
    assert _dir_fingerprint(wallet.dir) == before


def test_serve_refuses_non_wallet(tmp_path):
    with pytest.raises(Exception):
        serve(tmp_path / "empty")


def test_estate_view_served_with_runtime_dimension(tmp_path):
    wallet = _seal_wallet(tmp_path)
    server = serve(wallet.dir, host="127.0.0.1", port=0)
    port = server.server_address[1]
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{port}/estate") as resp:
            body = resp.read().decode()
        assert resp.status == 200
        assert "ALWAYS ON" in body and "ON-DEMAND" in body and "NOT RUNNING" in body
        assert "Runtime:" in body  # dossier panel wiring
        assert "lean-transparency-log" in body and "ltl-accumulator-verified" in body
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_estate_view_and_estate_md_do_not_drift():
    """Name-level sync guard between the cockpit estate page and ESTATE.md
    (the published_assets lesson: two renderings of one model need a
    tripwire)."""
    from pacta.estateview import ESTATE_HTML
    estate_md = Path("ESTATE.md").read_text(encoding="utf-8")
    for name in ("lean-transparency-log", "ltl-accumulator-verified",
                 "proof-aware-crypto-tooling-agent", "verifying-crypto-with-lean",
                 "dalek-ed25519-verified", "pasta-pallas-verified",
                 "ltl.zkdefi.org", "Forgejo"):
        assert name in ESTATE_HTML, f"estate view lost: {name}"
        assert name in estate_md, f"ESTATE.md lost: {name}"
    # the runtime dimension must exist in BOTH renderings
    assert "What is running" in estate_md
    assert "ALWAYS ON" in ESTATE_HTML


def test_demo_wallet_seals_and_serves(tmp_path):
    from pacta.walletui import seal_demo_wallet
    wallet_dir = seal_demo_wallet(tmp_path)
    server = serve(wallet_dir, host="127.0.0.1", port=0)
    port = server.server_address[1]
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{port}/") as resp:
            body = resp.read().decode()
        assert resp.status == 200
        assert "dalek-ed25519-verified" in body and "4 pinned" in body
        assert "chain verified" in body
    finally:
        server.shutdown()
        thread.join(timeout=5)


# ---------------------------------------------------------------------------
# UX contract: the cockpit never leaves a human in the dark. Every view must
# explain itself in plain language, every verdict must be stated in words,
# every page must carry navigation and interpretation help.
# ---------------------------------------------------------------------------

def _serve(wallet_dir):
    server = serve(wallet_dir, host="127.0.0.1", port=0)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, thread, server.server_address[1]


def _get(port, route):
    with urllib.request.urlopen(f"http://127.0.0.1:{port}{route}") as resp:
        return resp.status, resp.read().decode()


def test_guide_view_explains_every_term(tmp_path):
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/guide")
        assert status == 200
        assert "renders evidence" in body  # the design law, in the user's face
        for term in ("custody capsule", "quorum member", "hash chain",
                     "custody latch", "refusal receipt", "air-gap",
                     "evidence grade", "DEMO wallet",
                     "What this cockpit cannot tell you",
                     "A five-minute tour"):
            assert term in body, f"guide is missing: {term}"
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_every_view_carries_lead_nav_and_explainers(tmp_path):
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        for route in ("/", "/posture", "/queue", "/incidents", "/inspect"):
            status, body = _get(port, route)
            assert status == 200
            assert 'href="/guide"' in body, f"{route}: no path to the guide"
            assert 'class="lead"' in body, f"{route}: no plain-language lead"
            assert "<details" in body, f"{route}: no interpretation expander"
            assert 'class="navsub"' in body, f"{route}: tabs do not say what they answer"
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_empty_states_are_explained(tmp_path):
    """A blank table is a UX failure; every empty list must say what empty
    means and whether it is good news."""
    wallet = _seal_wallet(tmp_path)
    q = render_queue(collect_airgap(wallet))
    assert "No signing requests are waiting" in q
    inc = render_incidents(collect_incidents(wallet), collect_refusals(wallet))
    assert "Empty is the good state" in inc
    assert "nothing has been refused" in inc


def test_estate_page_links_back_to_cockpit(tmp_path):
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/estate")
        assert status == 200
        assert "Back to cockpit" in body
    finally:
        server.shutdown()
        thread.join(timeout=5)


# ---------------------------------------------------------------------------
# Crew law: the bridge presents six DISTINCT role stations that cooperate
# through explicit handoffs — everything a human crew would need with no AI
# around, as runnable commands. Roles never melt into each other.
# ---------------------------------------------------------------------------

STATION_IDS = ("proposer", "quorum", "operator", "cryptographer",
               "architect", "newcomer")


def test_bridge_shows_crew_and_dispatch(tmp_path):
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/")
        assert status == 200
        assert "CUSTODY HEALTHY" in body  # whole-system verdict, in words, on top
        for name in ("Proposer", "Quorum bench", "Operator", "Cryptographer",
                     "Architect", "Newcomer"):
            assert name in body, f"bridge is missing the {name} card"
        assert "Take this station" in body
        assert "If this happens, who acts" in body  # the dispatch (andon) board
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_every_station_defines_role_contract(tmp_path):
    """Each station page must state: mission, duties as runnable commands,
    the never-list (separation of duties), and explicit handoffs."""
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        for sid in STATION_IDS:
            status, body = _get(port, f"/station/{sid}")
            assert status == 200, f"{sid}: not served"
            assert "Mission" in body, f"{sid}: no mission"
            assert "Duties" in body, f"{sid}: no duties"
            assert 'class="cmd"' in body, f"{sid}: no runnable command (the no-AI drill)"
            assert "This station never" in body, f"{sid}: no separation-of-duties list"
            assert "Handoffs" in body, f"{sid}: no handoffs"
            assert "RECEIVES" in body and "DELIVERS" in body, f"{sid}: handoffs empty"
            assert 'class="lead"' in body, f"{sid}: no lead"
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_stations_are_distinct_roles(tmp_path):
    """No melting: each role's signature phrase appears on its own station
    and on no other station."""
    markers = {
        "proposer": "front door",
        "quorum": "unanimity or latch",
        "operator": "Probe now",
        "cryptographer": "Replay the whole log offline",
        "architect": "Drift tripwire",
        "newcomer": "Your first hour",
    }
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        bodies = {sid: _get(port, f"/station/{sid}")[1] for sid in STATION_IDS}
        for sid, marker in markers.items():
            assert marker in bodies[sid], f"{sid}: lost its signature duty ({marker})"
            for other, body in bodies.items():
                if other != sid:
                    assert marker not in body, (
                        f"{other} bleeds into {sid}'s role ({marker})")
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_deck_serves_all_panes_and_wizard(tmp_path):
    """The deck: every role live in its own pane (tmux-style grid), plus the
    color-camouflaged wizard rail that walks a newcomer through each role."""
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/deck")
        assert status == 200
        for sid in STATION_IDS:
            assert f'src="/station/{sid}?pane=1"' in body, f"deck missing pane: {sid}"
        assert "The wizard" in body and "your first watch" in body
        assert body.count('class="wizstep"') >= 8, "wizard lost its steps"
        for sid in STATION_IDS:
            assert f'data-role="{sid}"' in body, f"wizard never visits {sid}"
        assert "Success looks like" in body  # every step tells the learner when they're done
        assert "YOU ARE THE" in body         # the role-camouflage chip
        assert "READ-ONLY" in body
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_pane_mode_is_chromeless_but_labeled(tmp_path):
    """Panes are independent viewports: full station content, no page chrome,
    still labeled read-only; links inside stay in pane mode via the shim."""
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/station/operator?pane=1")
        assert status == 200
        assert 'class="navrow"' not in body      # no nav chrome inside a pane
        assert "warden custody cockpit" not in body  # no page h1 either
        assert "Probe now" in body               # the actual station content
        assert "READ-ONLY pane" in body          # but the guarantee stays visible
        assert "searchParams.set('pane','1')" in body  # the stay-in-pane shim
        status, body = _get(port, "/inspect?pane=1")
        assert status == 200 and 'class="navrow"' not in body
        assert "Verify (read-only)" in body
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_inspect_sample_prefill(tmp_path):
    evidence = Path("examples") / "wallet-evidence"
    if not (evidence / "log.pub").exists():
        pytest.skip("example wallet evidence not present")
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/inspect?sample=1")
        assert status == 200
        assert "sample loaded" in body
        assert "BEGIN PUBLIC KEY" in body  # log.pub really pre-filled
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_manual_is_a_complete_course(tmp_path):
    """The lab manual: a full study-club course — every role gets a session,
    every session has a lab with checkpoints and a self-test, the capstone
    crosses all chairs, and the answers exist."""
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/manual")
        assert status == 200
        assert "Syllabus" in body  # the TOC
        for session in ("the world you are entering", "the Newcomer",
                        "the Proposer", "the Quorum bench", "the Operator",
                        "the Cryptographer", "the Architect",
                        "Capstone: the incident day", "Graduation"):
            assert session in body, f"manual lost session: {session}"
        assert body.count("Checkpoint") >= 8, "labs lost their checkpoints"
        assert body.count("Self-test") >= 6, "sessions lost their self-tests"
        assert "Appendix B" in body and "self-test answers" in body.lower()
        assert "tamper drill" in body.lower()  # the hands-on heart of Session 4
        assert "Recap card" in body
    finally:
        server.shutdown()
        thread.join(timeout=5)


def test_manual_markdown_renders_cleanly(tmp_path):
    """mdlite over the real manual: no raw Markdown artifacts may leak."""
    from pacta.mdlite import render as render_markdown
    md = (Path("docs") / "warden-lab-manual.md").read_text(encoding="utf-8")
    html_out, toc = render_markdown(md)
    for artifact in ("```", "](", "\n## ", "> ✔"):
        assert artifact not in html_out, f"raw markdown leaked: {artifact!r}"
    assert html_out.count("**") == 0, "unrendered bold markers leaked"
    assert len(toc) >= 10  # sessions + appendices reach the syllabus
    assert "<pre class='cmd'>" in html_out  # command blocks rendered
    assert "<blockquote>" in html_out  # checkpoints rendered


def test_operator_probe_is_explicit_and_live(tmp_path):
    """The liveness board never phones home on an ordinary page load; probes
    run only on the operator's explicit demand, then show per-target rows."""
    wallet = _seal_wallet(tmp_path)
    server, thread, port = _serve(wallet.dir)
    try:
        status, body = _get(port, "/station/operator")
        assert status == 200
        assert "has not probed" in body and "Probe now" in body
        assert "Public services" not in body  # no results without demand
        status, body = _get(port, "/station/operator?probe=1")
        assert status == 200
        assert "Public services" in body and "Local working copies" in body
        for target in ("log head", "paper", "blog", "lean-transparency-log"):
            assert target in body, f"probe results missing target: {target}"
    finally:
        server.shutdown()
        thread.join(timeout=5)