from __future__ import annotations

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

HERE = Path(__file__).resolve().parent
HARNESS = HERE / "invite_harness.py"
PERMISSION = HERE / "permission-seal.json"
FIXTURE = HERE.parent / "reset-controller-st_01a0054d/fixtures/current-empty-profile"
VENV_PYTHON = Path("/home/cube/projects/richard/hermes-agent/.venv/bin/python")


def sha(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def profile(tmp_path: Path) -> Path:
    target = tmp_path / "profile"
    shutil.copytree(FIXTURE, target)
    for root, dirs, files in os.walk(target):
        Path(root).chmod(0o700)
        for name in files:
            (Path(root) / name).chmod(0o600)
    # A stopped clean invite baseline has no process lock.
    (target / "gateway.lock").unlink(missing_ok=True)
    return target


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"},
    )


def draft(path: Path, key: str = "task26_same_actor_20260815") -> 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), encoding="utf-8")
    path.chmod(0o600)
    return path


def test_dry_run_is_profile_byte_stable_and_redacted(tmp_path: Path) -> None:
    target = profile(tmp_path)
    ledger = target / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    before = sha(ledger)
    receipt = tmp_path / "dry-run.json"
    result = run("dry-run", "--profile", str(target), "--permission", str(PERMISSION),
                 "--receipt", str(receipt))
    assert result.returncode == 0, result.stderr
    assert json.loads(result.stdout)["status"] == "PASS"
    assert sha(ledger) == before
    assert json.loads(receipt.read_text())["ready_for_one_invite"] is True
    assert "rc1_" not in receipt.read_text()


def test_prepare_uses_api_after_watch_and_emits_only_one_private_token_artifact(tmp_path: Path) -> None:
    target = profile(tmp_path)
    handoff = tmp_path / "handoff.json"
    receipt = tmp_path / "prepared.json"
    result = run(
        "prepare", "--profile", str(target), "--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
    public = json.loads(receipt.read_text())
    private = json.loads(handoff.read_text())
    ledger = json.loads((target / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json").read_text())
    assert len(ledger["sessions"]) == 4
    assert sum(item["state"] == "PREPARED" for item in ledger["sessions"]) == 1
    assert private["start_token"].startswith("rc1_")
    assert hashlib.sha256(private["start_token"].encode("ascii")).hexdigest() == public["start_token_sha256"]
    assert hashlib.sha256(private["start_token"][4:].encode("ascii")).hexdigest() == public["sid_hash"]
    assert private["session_id"] == public["session_id"]
    assert handoff.stat().st_mode & 0o777 == 0o600
    assert receipt.stat().st_mode & 0o777 == 0o600
    assert private["start_token"] not in receipt.read_text()
    assert public["prepare_invocations"] == 1
    assert public["event_subscription_before_prepare"] is True

    duplicate = run(
        "prepare", "--profile", str(target), "--permission", str(PERMISSION),
        "--draft", str(draft(tmp_path / "draft2.json", "another_new_key")),
        "--handoff", str(tmp_path / "second-handoff.json"),
        "--receipt", str(tmp_path / "second-receipt.json"),
    )
    assert duplicate.returncode != 0
    assert not (tmp_path / "second-handoff.json").exists()


def test_prepare_rejects_nonterminal_duplicate_recovery_and_existing_authority(tmp_path: Path) -> None:
    scenarios = ["nonterminal", "duplicate", "recovery", "registry"]
    for scenario in scenarios:
        target = profile(tmp_path / scenario)
        ledger_path = target / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
        ledger = json.loads(ledger_path.read_text())
        if scenario == "nonterminal":
            ledger["sessions"][0]["state"] = "PREPARED"
        elif scenario == "duplicate":
            ledger["sessions"].append(dict(ledger["sessions"][0]))
        elif scenario == "recovery":
            ledger["sessions"][0]["recovery_attempts"] = [{"unsafe": True}]
        else:
            registry = target / "customers/registry.json"
            value = json.loads(registry.read_text())
            value["customers"] = [{"customer_key": "existing"}]
            registry.write_text(json.dumps(value))
            registry.chmod(0o600)
        if scenario != "registry":
            # Deliberately leave the old digest: malformed authority must fail closed.
            ledger_path.write_text(json.dumps(ledger))
            ledger_path.chmod(0o600)
        receipt = tmp_path / scenario / "receipt.json"
        result = run(
            "prepare", "--profile", str(target), "--permission", str(PERMISSION),
            "--draft", str(draft(tmp_path / scenario / "draft.json", f"new_{scenario}")),
            "--handoff", str(tmp_path / scenario / "handoff.json"), "--receipt", str(receipt),
        )
        assert result.returncode != 0, scenario
        assert not (tmp_path / scenario / "handoff.json").exists()


def test_observe_subscribes_before_exact_single_private_actor_claim(tmp_path: Path) -> None:
    target = profile(tmp_path)
    handoff = tmp_path / "handoff.json"
    prepared_receipt = tmp_path / "prepared.json"
    assert run(
        "prepare", "--profile", str(target), "--permission", str(PERMISSION),
        "--draft", str(draft(tmp_path / "draft.json")), "--handoff", str(handoff),
        "--receipt", str(prepared_receipt),
    ).returncode == 0
    prepared = json.loads(prepared_receipt.read_text())
    ready = tmp_path / "ready.json"
    observed = tmp_path / "observed.json"
    read_fd, write_fd = os.pipe()
    process = subprocess.Popen(
        [str(VENV_PYTHON), "-B", str(HARNESS), "observe", "--profile", str(target),
         "--permission", str(PERMISSION), "--session-id", prepared["session_id"],
         "--sid-hash", prepared["sid_hash"], "--ready", str(ready),
         "--receipt", str(observed), "--timeout-seconds", "5", "--ready-fd", str(write_fd)],
        text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, pass_fds=(write_fd,),
    )
    os.close(write_fd)
    assert os.read(read_fd, 1) == b"1"  # exact readiness signal, no timing wait
    os.close(read_fd)

    # Claim through the actual installed candidate API, never JSON mutation.
    module = importlib.import_module("gateway.platforms.telegram_customer_bootstrap")
    token = json.loads(handoff.read_text())["start_token"]
    module.RoomBootstrapStore(module.room_bootstrap_state_dir(target)).claim_rehearsal_customer_invite(
        token, user_id="8527916639", chat_id="8527916639", message_id="101"
    )
    stdout, stderr = process.communicate(timeout=10)
    assert process.returncode == 0, stderr
    value = json.loads(observed.read_text())
    assert value["status"] == "PASS_CLAIM_OBSERVED"
    assert value["accepted_start_claim_count"] == 1
    assert value["actor_id"] == "8527916639"
    assert value["private_dm"] is True
    assert value["event_subscription_before_claim"] is True
    assert "rc1_" not in stdout + stderr + observed.read_text()


def test_expire_is_canonical_and_refuses_before_deadline(tmp_path: Path) -> None:
    target = profile(tmp_path)
    handoff = tmp_path / "handoff.json"
    prepared = tmp_path / "prepared.json"
    assert run(
        "prepare", "--profile", str(target), "--permission", str(PERMISSION),
        "--draft", str(draft(tmp_path / "draft.json")), "--handoff", str(handoff),
        "--receipt", str(prepared),
    ).returncode == 0
    value = json.loads(prepared.read_text())
    result = run(
        "expire", "--profile", str(target), "--permission", str(PERMISSION),
        "--session-id", value["session_id"], "--sid-hash", value["sid_hash"],
        "--receipt", str(tmp_path / "expired.json"),
    )
    assert result.returncode != 0
    assert not (tmp_path / "expired.json").exists()


def test_permission_and_loaded_byte_pins_fail_closed(tmp_path: Path) -> None:
    target = profile(tmp_path)
    altered = tmp_path / "permission.json"
    value = json.loads(PERMISSION.read_text())
    value["actor_id"] = "8693203710"
    altered.write_text(json.dumps(value))
    altered.chmod(0o600)
    result = run("dry-run", "--profile", str(target), "--permission", str(altered),
                 "--receipt", str(tmp_path / "receipt.json"))
    assert result.returncode != 0