from __future__ import annotations

import hashlib
import importlib.util
import json
import os
import shutil
import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest

HERE = Path(__file__).resolve().parent
HARNESS = HERE / "invite_harness_v2.py"
PERMISSION = HERE / "permission-seal-v2.json"
VENV_PYTHON = Path("/home/cube/projects/richard/hermes-agent/.venv/bin/python")
RESET = Path("/home/cube/.hermes/profiles/dualcoachtest/data/profile-reset-archives/task26-live-reset-2e0894ea")


def load_harness():
    spec = importlib.util.spec_from_file_location("invite_harness_v2", HARNESS)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def sha_tree(root: Path) -> str:
    rows = []
    for path in sorted(item for item in root.rglob("*") if item.is_file()):
        rows.append(path.relative_to(root).as_posix().encode() + b"\0" + hashlib.sha256(path.read_bytes()).digest())
    return hashlib.sha256(b"\n".join(rows)).hexdigest()


def post_reset_profile(tmp_path: Path) -> Path:
    profile = tmp_path / "profile"
    data = profile / "data"
    data.mkdir(parents=True, mode=0o700)
    archive = data / "profile-reset-archives/task26-live-reset-2e0894ea"
    archive.mkdir(parents=True, mode=0o700)
    for name in ("manifest.json", "receipt.json"):
        shutil.copyfile(RESET / name, archive / name)
        (archive / name).chmod(0o600)
    profile.chmod(0o700)
    data.chmod(0o700)
    return profile


def draft(path: Path, key: str = "task26_same_actor_v2") -> Path:
    value = {
        "customer_key": key,
        "display_name": "Task26 Synthetic Customer",
        "starts_on": "2026-08-16",
        "daily_time": "08:00",
        "weekly_weekday": 0,
        "monthly_day": 1,
        "calories_kcal": 2200,
        "protein_g": 140,
        "meals": ["breakfast", "lunch", "dinner"],
        "primary_goal": "synthetic rehearsal",
    }
    path.write_text(json.dumps(value))
    path.chmod(0o600)
    return path


def run(*args: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [str(VENV_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_missing_ledger_is_canonical_read_only_baseline(tmp_path: Path, mode: str) -> None:
    profile = post_reset_profile(tmp_path)
    before = sha_tree(profile)
    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["baseline"] == "RESET_BOUND_LEDGER_ABSENT"
    assert value["ledger_present"] is False
    assert value["reset_manifest_sha256"].startswith("7ef0eb8f")
    assert value["reset_evidence_digest"].startswith("d2879b12")
    assert value["mutations"] == 0
    assert sha_tree(profile) == before
    assert not (profile / "data/onboarding").exists()


@pytest.mark.parametrize(
    "attack",
    ["unknown-parent", "unknown-file", "symlink", "wrong-mode", "wrong-schema", "unexpected-sibling"],
)
def test_absent_baseline_rejects_unclean_or_unsafe_authority(tmp_path: Path, attack: str) -> None:
    profile = post_reset_profile(tmp_path)
    onboarding = profile / "data/onboarding"
    bootstrap = onboarding / "telegram-customer-bootstrap-v1"
    if attack == "unknown-parent":
        onboarding.mkdir(mode=0o700)
    elif attack == "unknown-file":
        bootstrap.mkdir(parents=True, mode=0o700)
        p = bootstrap / "unknown"
        p.write_bytes(b"x")
        p.chmod(0o600)
    elif attack == "symlink":
        onboarding.symlink_to(profile / "data/profile-reset-archives", target_is_directory=True)
    elif attack == "wrong-mode":
        onboarding.mkdir(mode=0o755)
    elif attack == "wrong-schema":
        bootstrap.mkdir(parents=True, mode=0o700)
        p = bootstrap / "ledger.json"
        p.write_text("{}\n")
        p.chmod(0o600)
    else:
        p = profile / "customers"
        p.mkdir(mode=0o700)
    result = run("dry-run", "--profile", str(profile), "--permission", str(PERMISSION),
                 "--receipt", str(tmp_path / f"{attack}.json"))
    assert result.returncode == 2
    assert json.loads((tmp_path / f"{attack}.json").read_text())["status"] == "FAIL"


def test_prepare_candidate_initializes_absent_tree_after_subscription_once(tmp_path: Path) -> None:
    profile = post_reset_profile(tmp_path)
    handoff = tmp_path / "handoff.json"
    receipt = tmp_path / "prepared.json"
    result = run(
        "prepare", "--profile", str(profile), "--permission", str(PERMISSION),
        "--draft", str(draft(tmp_path / "draft.json")), "--handoff", str(handoff),
        "--receipt", str(receipt), "--evidence-root", str(tmp_path),
    )
    assert result.returncode == 0, result.stderr
    value = json.loads(receipt.read_text())
    ledger_path = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    ledger = json.loads(ledger_path.read_text())
    assert value["baseline_before"] == "RESET_BOUND_LEDGER_ABSENT"
    assert value["watch_root"] == "data"
    assert value["event_subscription_before_initializer"] is True
    assert value["onboarding_create_event_count"] == 1
    assert value["candidate_initializer_invocations"] == 1
    assert value["prepare_invocations"] == 1
    assert value["final_ledger_publications_expected_from_candidate"] == 2
    assert len(ledger["sessions"]) == 1
    assert ledger["sessions"][0]["state"] == "PREPARED"
    state = ledger_path.parent
    assert (profile / "data/onboarding").stat().st_mode & 0o777 == 0o775
    assert state.stat().st_mode & 0o777 == 0o700
    assert ledger_path.stat().st_mode & 0o777 == 0o600
    assert (state / "ledger.lock").stat().st_mode & 0o777 == 0o600
    assert set(item.name for item in state.iterdir()) == {"ledger.json", "ledger.lock"}
    assert handoff.stat().st_mode & 0o777 == 0o600
    assert "rc1_" not in receipt.read_text()


def test_prepare_rejects_race_creation_after_subscription(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    module = load_harness()
    profile = post_reset_profile(tmp_path)
    real = module.Inotify

    class RacingWatch(real):
        def __init__(self, directory: Path) -> None:
            # Simulate creation after preflight but immediately before subscription.
            (profile / "data/onboarding").mkdir(mode=0o700)
            super().__init__(directory)

    monkeypatch.setattr(module, "Inotify", RacingWatch)
    args = SimpleNamespace(
        profile=profile, draft=draft(tmp_path / "draft.json"), handoff=tmp_path / "handoff.json",
        receipt=tmp_path / "receipt.json", evidence_root=None,
    )
    with pytest.raises(module.HarnessError, match="race"):
        module.prepare(args)
    assert not args.handoff.exists()
    assert not (profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json").exists()


def test_prepare_rejects_duplicate_creation_events(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    module = load_harness()
    profile = post_reset_profile(tmp_path)

    class DuplicateWatch:
        def __init__(self, _directory: Path) -> None: pass
        def close(self) -> None: pass
        def await_name_count(self, _name: bytes, _timeout: float) -> int: return 2

    monkeypatch.setattr(module, "Inotify", DuplicateWatch)
    args = SimpleNamespace(
        profile=profile, draft=draft(tmp_path / "draft.json"), handoff=tmp_path / "handoff.json",
        receipt=tmp_path / "receipt.json", evidence_root=None,
    )
    with pytest.raises(module.HarnessError, match="exactly one"):
        module.prepare(args)
    assert not args.handoff.exists()


def test_candidate_source_and_upstream_test_prove_supported_initializer() -> None:
    module = load_harness()
    proof = module.loaded_byte_proof()
    assert proof["candidate_absent_ledger_initializer"] == "PASS"
    assert proof["initializer_source_sha256"] == module.MODULE_SHA256
    source = module.wheel_module_bytes().decode()
    assert "self.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)" in source
    assert "else:\n                self._write_unlocked(())" in source
    upstream = Path("/home/cube/projects/richard/hermes-agent/tests/gateway/test_telegram_customer_bootstrap.py").read_text()
    assert "state_dir = tmp_path / \"bootstrap\"" in upstream
    assert "RoomBootstrapStore(\n        state_dir" in upstream


def test_reset_binding_drift_fails_closed(tmp_path: Path) -> None:
    profile = post_reset_profile(tmp_path)
    manifest = profile / "data/profile-reset-archives/task26-live-reset-2e0894ea/manifest.json"
    manifest.write_bytes(manifest.read_bytes() + b" ")
    result = run("verify", "--profile", str(profile), "--permission", str(PERMISSION),
                 "--receipt", str(tmp_path / "verify.json"))
    assert result.returncode == 2
    assert "reset manifest" in json.loads((tmp_path / "verify.json").read_text())["blocker"]


def test_independent_verifier_passes_disposable_read_only_modes(tmp_path: Path) -> None:
    output = tmp_path / "independent.json"
    result = subprocess.run(
        [str(VENV_PYTHON), "-B", str(HERE / "independent_verify_v2.py"), str(output)],
        text=True, capture_output=True, check=False,
    )
    assert result.returncode == 0, result.stderr
    value = json.loads(output.read_text())
    assert value["status"] == "PASS"
    assert value["profile_mutations"] == 0