from __future__ import annotations

import hashlib
import json
import os
import shutil
import subprocess
from pathlib import Path

import pytest

HERE = Path(__file__).resolve().parent
HARNESS = HERE / "invite_harness_v3.py"
PERMISSION = HERE / "permission-seal-v3.json"
DRAFT = HERE / "task26-live-2e-r2-customer-draft.private.json"
PYTHON = Path("/home/cube/projects/richard/hermes-agent/.venv/bin/python")
LIVE = Path("/home/cube/.hermes/profiles/dualcoachtest")


def canonical_ledger(value: dict[str, object]) -> None:
    payload = {"schema": value["schema"], "sessions": value["sessions"]}
    raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
    value["digest"] = hashlib.sha256(raw).hexdigest()


def fixture(tmp_path: Path) -> Path:
    profile = tmp_path / "profile"
    paths = (
        "customers/registry.json",
        "data/onboarding/telegram-customer-bootstrap-v1/ledger.json",
        "data/onboarding/telegram-customer-bootstrap-v1/ledger.lock",
        "data/owner-actions/draft-deliveries.json.lock",
        "gateway.lock",
        "gateway_state.json",
        "data/profile-reset-archives/task26-live-reset-2e0894ea/manifest.json",
        "data/profile-reset-archives/task26-live-reset-2e0894ea/receipt.json",
    )
    for relative in paths:
        source = LIVE / relative
        target = profile / relative
        target.parent.mkdir(parents=True, mode=0o700, exist_ok=True)
        shutil.copyfile(source, target)
        target.chmod(0o600)
    for path in (profile, profile / "data", profile / "customers",
                 profile / "data/onboarding/telegram-customer-bootstrap-v1",
                 profile / "data/owner-actions"):
        path.chmod(0o700)
    (profile / "data/onboarding").chmod(0o775)
    return profile


def run(*args: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [str(PYTHON), "-B", str(HARNESS), *args], text=True, capture_output=True,
        check=False, env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
    )


@pytest.mark.parametrize("mode", ["dry-run", "verify"])
def test_exact_expired_lineage_is_read_only_ready(tmp_path: Path, mode: str) -> None:
    profile = fixture(tmp_path)
    ledger = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    before = hashlib.sha256(ledger.read_bytes()).hexdigest()
    receipt = tmp_path / f"{mode}.json"
    result = run(mode, "--profile", str(profile), "--permission", str(PERMISSION),
                 "--receipt", str(receipt))
    assert result.returncode == 0, result.stderr
    value = json.loads(receipt.read_text())
    assert value["status"] == "PASS"
    assert value["ready_for_exactly_one_replacement_invite"] is True
    assert value["prior_session_count"] == 1
    assert value["prior_state"] == "EXPIRED"
    assert value["prepared_session_count"] == 0
    assert value["accepted_claim_count"] == 0
    assert value["registry_customer_count"] == 0
    assert value["mutations"] == 0
    assert hashlib.sha256(ledger.read_bytes()).hexdigest() == before


@pytest.mark.parametrize("attack", ["zero", "two", "prepared", "claim", "wrong-id", "customer"])
def test_rejects_wrong_lineage_or_live_authority(tmp_path: Path, attack: str) -> None:
    profile = fixture(tmp_path)
    ledger_path = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    ledger = json.loads(ledger_path.read_text())
    session = ledger["sessions"][0]
    if attack == "zero":
        ledger["sessions"] = []
    elif attack == "two":
        ledger["sessions"].append(dict(session))
        ledger["sessions"][1]["session_id"] = "cb_0000000000000000000000"
        ledger["sessions"][1]["sid_hash"] = "0" * 64
    elif attack == "prepared":
        session["state"] = "PREPARED"
    elif attack == "claim":
        session["role_claims"] = [{"role": "customer"}]
    elif attack == "wrong-id":
        session["session_id"] = "cb_0000000000000000000000"
    else:
        registry = profile / "customers/registry.json"
        value = json.loads(registry.read_text())
        value["customers"] = [{"customer_key": "x"}]
        registry.write_text(json.dumps(value))
        registry.chmod(0o600)
    if attack != "customer":
        canonical_ledger(ledger)
        ledger_path.write_text(json.dumps(ledger, sort_keys=True, separators=(",", ":")))
        ledger_path.chmod(0o600)
    result = run("dry-run", "--profile", str(profile), "--permission", str(PERMISSION),
                 "--receipt", str(tmp_path / "receipt.json"))
    assert result.returncode == 2
    assert json.loads((tmp_path / "receipt.json").read_text())["status"] == "FAIL"


def test_only_hard_pinned_replacement_draft_is_accepted(tmp_path: Path) -> None:
    profile = fixture(tmp_path)
    altered = tmp_path / "draft.json"
    value = json.loads(DRAFT.read_text())
    value["customer_key"] = "some_other_new_key"
    altered.write_text(json.dumps(value))
    altered.chmod(0o600)
    result = run(
        "prepare", "--profile", str(profile), "--permission", str(PERMISSION),
        "--draft", str(altered), "--handoff", str(tmp_path / "handoff.json"),
        "--receipt", str(tmp_path / "prepared.json"),
    )
    assert result.returncode == 2
    assert "replacement draft" in result.stderr
    ledger = json.loads((profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json").read_text())
    assert len(ledger["sessions"]) == 1


def test_one_replacement_then_third_invite_is_rejected(tmp_path: Path) -> None:
    profile = fixture(tmp_path)
    result = run(
        "prepare", "--profile", str(profile), "--permission", str(PERMISSION),
        "--draft", str(DRAFT), "--handoff", str(tmp_path / "handoff.json"),
        "--receipt", str(tmp_path / "prepared.json"), "--evidence-root", str(tmp_path),
    )
    assert result.returncode == 0, result.stderr
    receipt = json.loads((tmp_path / "prepared.json").read_text())
    ledger = json.loads((profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json").read_text())
    assert receipt["replacement_prepare_invocations"] == 1
    assert receipt["event_subscription_before_prepare"] is True
    assert len(ledger["sessions"]) == 2
    assert [item["state"] for item in ledger["sessions"]] == ["EXPIRED", "PREPARED"]
    assert ledger["sessions"][1]["session_id"] != ledger["sessions"][0]["session_id"]
    third = run(
        "prepare", "--profile", str(profile), "--permission", str(PERMISSION),
        "--draft", str(DRAFT), "--handoff", str(tmp_path / "third-handoff.json"),
        "--receipt", str(tmp_path / "third.json"),
    )
    assert third.returncode == 2
    assert not (tmp_path / "third-handoff.json").exists()


def test_seal_binds_current_plan_approval_expiry_and_v6() -> None:
    seal = json.loads(PERMISSION.read_text())
    assert seal["plan_sha256"].startswith("d816dbbd")
    assert seal["replacement_authorization_event_sha256"].startswith("4a3551a9")
    assert seal["baseline_readiness_sha256"].startswith("16497e3b")
    assert seal["prior_expiry_receipt_sha256"].startswith("d906f1f")
    assert seal["replacement_maximum"] == 1
    assert seal["third_invite_authorized"] is False