from __future__ import annotations

import hashlib
import importlib
import importlib.util
import json
import os
import stat
import subprocess
import sys
from pathlib import Path

import pytest

HERE = Path(__file__).resolve().parent
CONTROLLER = HERE / "task27_supplemental_cleanup_controller.py"
SESSION = "cb_mYUoMIsk_CRzSDKYpPE9dg"
USER = "8527916639"
ROUTE = (USER, "0")
WHEEL = (
    HERE.parent
    / "task26/task26-combined-v38-delivered-st_01a019d7/artifacts"
    / "hermes_agent-0.17.0-py3-none-any.whl"
)


def load():
    spec = importlib.util.spec_from_file_location("task27_supplement", CONTROLLER)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def canonical(value):
    return json.dumps(
        value, ensure_ascii=True, sort_keys=True, separators=(",", ":")
    ).encode()


def private_json(path: Path, value) -> None:
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    path.write_bytes(canonical(value) + b"\n")
    path.chmod(0o600)


def profile(tmp_path: Path) -> Path:
    root = tmp_path / "profile"
    root.mkdir(parents=True, mode=0o700)
    marker = root / ".task27-supplemental-disposable-copy"
    marker.write_text("TASK27_SUPPLEMENTAL_DISPOSABLE_COPY\n")
    marker.chmod(0o600)
    authority_lock = root / "data/.profile-authority.lock"
    authority_lock.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    authority_lock.touch(mode=0o600)
    authority_lock.chmod(0o600)
    sys.path.insert(0, str(WHEEL))
    outbox_module = importlib.import_module(
        "gateway.platforms.telegram_nutrition_onboarding_publication_outbox"
    )
    membership_module = importlib.import_module(
        "gateway.platforms.telegram_staff_membership_gate"
    )
    outbox_type = getattr(outbox_module, "GatewayOnboardingPublicationOutbox")
    journal_type = getattr(membership_module, "MembershipJournal")

    payload = {"body_digest": "a" * 64, "state": "collecting"}
    render = "b" * 64
    outbox = outbox_type(root)
    outbox.claim(
        session_id=SESSION,
        generation=0,
        payload=payload,
        route=ROUTE,
        role="customer",
        render_identity=render,
    )
    outbox.record_receipt(
        session_id=SESSION,
        generation=0,
        chat_id=USER,
        topic_id="0",
        message_id=304,
    )
    outbox.mark_committed(
        session_id=SESSION,
        generation=0,
        payload=payload,
        route=ROUTE,
        role="customer",
        render_identity=render,
        message_id=304,
    )
    journal_type(
        root / "data/onboarding/telegram-staff-membership-v1/events.jsonl"
    ).append(
        {
            "event": "subscription_armed",
            "subscription_epoch_id": "epoch-1",
            "observed_at_utc": "2026-08-18T14:55:11+00:00",
            "staff_chat_inventory_sha256": "c" * 64,
            "customer_user_ids": [USER],
        }
    )
    return root


def service_state(tmp_path: Path, active: bool = False) -> Path:
    path = tmp_path / ("active.json" if active else "inactive.json")
    private_json(
        path,
        {
            "active_state": "active" if active else "inactive",
            "sub_state": "running" if active else "dead",
            "main_pid": 42 if active else 0,
            "matching_processes": 1 if active else 0,
        },
    )
    return path


def permission(module, root: Path, tmp_path: Path):
    path = tmp_path / "permission.json"
    result = module.dry_run(
        root, path, service_override=service_state(tmp_path)
    )
    return path, result["permission_seal"], result


def mutation_digest(module, root: Path) -> str:
    return module.sha_bytes(
        canonical(
            {
                key: module.compact(module.tree_inventory(value))
                for key, value in module.mutation_paths(root).items()
            }
        )
    )


def test_dry_run_binds_contract(tmp_path: Path) -> None:
    module = load()
    root = profile(tmp_path)
    _, _, result = permission(module, root, tmp_path)
    payload = result["permission_payload"]
    assert set(payload["exact_state"]) == {
        "outbox",
        "membership",
        "supplemental_archive",
    }
    assert payload["cleanup_receipt"]["sha256"] == module.CLEANUP_RECEIPT_SHA256
    assert payload["proposed_mutations"] == module.proposed_mutations()
    assert any(
        mutation["path"]
        == (
            ".omo/evidence/task27/"
            ".task27-supplemental-live-cleanup-receipt.json.*"
        )
        for mutation in payload["proposed_mutations"]
    )
    assert payload["tools"] == module.tool_binding()


def test_live_execution_requires_exact_authorization(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    module = load()
    root = profile(tmp_path)
    authorization = tmp_path / "authorization.json"
    live_output = tmp_path / "live-receipt.json"
    monkeypatch.setattr(module, "LIVE", root)
    monkeypatch.setattr(module, "LIVE_AUTHORIZATION", authorization, raising=False)
    monkeypatch.setattr(module, "LIVE_OUTPUT", live_output, raising=False)
    preview = tmp_path / "preview.json"
    monkeypatch.setattr(module, "LIVE_PREVIEW", preview, raising=False)
    preview_result = module.dry_run(
        root,
        preview,
        service_override=service_state(tmp_path),
    )
    assert preview_result["status"] == "AWAITING_AUTHORIZATION"
    approval: dict[str, object] = {
        "approved_at": "2026-08-20T03:56:29Z",
        "approved_controller_sha256": module.tool_binding()[
            "controller_sha256"
        ],
        "approved_dry_run_path": str(preview),
        "approved_dry_run_sha256": module.sha_file(preview),
        "approved_permission_seal": preview_result["permission_seal"],
        "thread_session_id": "test-session",
        "user_message": "ㅇㅇ",
    }
    authorization_document = {
        "schema": "task27-supplemental-live-cleanup-authorization-v1",
        "status": "APPROVED_ONCE",
        "target": str(root),
        "candidate_digest": module.CANDIDATE,
        "cleanup_receipt_sha256": module.CLEANUP_RECEIPT_SHA256,
        "identity": {
            "customer_key": module.CUSTOMER,
            "session_id": module.SESSION,
            "user_id": module.USER_ID,
            "route": list(module.ROUTE),
        },
        "proposed_mutations": module.proposed_mutations(),
        "constraints": {
            "activation": False,
            "commit": "none",
            "customer_delivery": False,
            "external_network_actions": 0,
            "gateway_must_remain_stopped": True,
            "provider_actions": 0,
            "push": "none",
            "release": False,
            "telegram_actions": 0,
        },
        "approval": approval,
    }
    private_json(authorization, authorization_document)
    authorization.chmod(0o400)
    with pytest.raises(module.Refusal, match="authorization"):
        module.execute(
            root,
            preview,
            preview_result["permission_seal"],
            live_output,
            service_override=service_state(tmp_path),
        )
    permission_path = tmp_path / "live-permission.json"
    permission_result = module.dry_run(
        root,
        permission_path,
        live_authorization_file=authorization,
        service_override=service_state(tmp_path),
    )
    authorized_payload = permission_result["permission_payload"]
    approved_projection = dict(authorized_payload)
    approved_projection["live_authorization"] = None
    assert approved_projection == preview_result["permission_payload"]
    forged_permission = json.loads(json.dumps(permission_result))
    forged_payload = forged_permission["permission_payload"]
    forged_payload["observed"]["membership_rows"] = 99
    forged_permission["permission_seal"] = module.sha_bytes(
        canonical(forged_payload)
    )
    forged_path = tmp_path / "forged-permission.json"
    private_json(forged_path, forged_permission)
    with pytest.raises(module.Refusal, match="approved preview"):
        module.execute(
            root,
            forged_path,
            forged_permission["permission_seal"],
            live_output,
            live_authorization_file=authorization,
            service_override=service_state(tmp_path),
        )
    result = module.execute(
        root,
        permission_path,
        permission_result["permission_seal"],
        live_output,
        live_authorization_file=authorization,
        service_override=service_state(tmp_path),
    )
    assert result["status"] == "COMMITTED"
    assert result["execution_target"] == "live"
    assert result["live_authorization"]["sha256"] == module.sha_file(authorization)
    consumption = (
        Path(result["archive"]["root"]) / "authorization-consumed.json"
    )
    assert consumption.is_file()
    assert json.loads(consumption.read_text())["permission_seal"] == (
        permission_result["permission_seal"]
    )


def test_live_execution_refuses_unbound_output(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    module = load()
    root = profile(tmp_path)
    expected_output = tmp_path / "expected.json"
    monkeypatch.setattr(module, "LIVE", root)
    monkeypatch.setattr(module, "LIVE_OUTPUT", expected_output, raising=False)
    with pytest.raises(module.Refusal, match="output"):
        module.execute(
            root,
            tmp_path / "unused-permission.json",
            "0" * 64,
            tmp_path / "arbitrary.json",
            live_authorization_file=tmp_path / "unused-authorization.json",
            service_override=service_state(tmp_path),
        )


@pytest.mark.parametrize(
    "kind", ["foreign_session", "foreign_route", "mixed_membership"]
)
def test_rejects_foreign_or_mixed_state(tmp_path: Path, kind: str) -> None:
    module = load()
    root = profile(tmp_path)
    if kind.startswith("foreign"):
        path = root / "data/onboarding/telegram-publication-outbox-v1/ledger.json"
        document = json.loads(path.read_text())
        if kind == "foreign_session":
            document["records"][0]["session_id"] = "foreign"
        else:
            document["records"][0]["route"] = ["999", "0"]
        private_json(path, document)
    else:
        path = root / "data/onboarding/telegram-staff-membership-v1/events.jsonl"
        row = json.loads(path.read_text())
        row["customer_user_ids"] = [USER, "999"]
        row.pop("row_sha256")
        row["row_sha256"] = hashlib.sha256(canonical(row)).hexdigest()
        path.write_bytes(canonical(row) + b"\n")
    with pytest.raises(module.Refusal):
        module.dry_run(
            root,
            tmp_path / "bad-permission.json",
            service_override=service_state(tmp_path),
        )


@pytest.mark.parametrize(
    "attack", ["hmac", "unknown", "symlink", "hardlink", "mode"]
)
def test_rejects_unsafe_or_unknown_outbox(tmp_path: Path, attack: str) -> None:
    module = load()
    root = profile(tmp_path)
    outbox = root / "data/onboarding/telegram-publication-outbox-v1"
    ledger = outbox / "ledger.json"
    if attack == "hmac":
        document = json.loads(ledger.read_text())
        document["records"][0]["receipt_integrity"] = "0" * 64
        private_json(ledger, document)
    elif attack == "unknown":
        path = outbox / "unexpected.json"
        path.write_text("{}\n")
        path.chmod(0o600)
    elif attack == "symlink":
        ledger.unlink()
        ledger.symlink_to("emergency.json")
    elif attack == "hardlink":
        os.link(ledger, outbox / "ledger-copy.json")
    else:
        ledger.chmod(0o644)
    with pytest.raises(module.Refusal):
        module.dry_run(
            root,
            tmp_path / "bad-permission.json",
            service_override=service_state(tmp_path),
        )


@pytest.mark.parametrize(
    "attack", ["unknown", "symlink", "hardlink", "mode", "malformed"]
)
def test_rejects_unsafe_membership(tmp_path: Path, attack: str) -> None:
    module = load()
    root = profile(tmp_path)
    membership = root / "data/onboarding/telegram-staff-membership-v1"
    events = membership / "events.jsonl"
    if attack == "unknown":
        path = membership / "unexpected"
        path.write_text("x")
        path.chmod(0o600)
    elif attack == "symlink":
        events.unlink()
        events.symlink_to("events.jsonl.lock")
    elif attack == "hardlink":
        os.link(events, membership / "copy")
    elif attack == "mode":
        events.chmod(0o644)
    else:
        events.write_text("{")
    with pytest.raises(module.Refusal):
        module.dry_run(
            root,
            tmp_path / "bad-permission.json",
            service_override=service_state(tmp_path),
        )


def test_rejects_active_service_and_state_drift(tmp_path: Path) -> None:
    module = load()
    root = profile(tmp_path)
    with pytest.raises(module.Refusal):
        module.dry_run(
            root,
            tmp_path / "active-permission.json",
            service_override=service_state(tmp_path, True),
        )
    permission_path, seal, _ = permission(module, root, tmp_path)
    (root / "data/onboarding/telegram-publication-outbox-v1/.lock").write_text(
        "drift"
    )
    with pytest.raises(module.Refusal, match="drift"):
        module.execute(
            root,
            permission_path,
            seal,
            tmp_path / "receipt.json",
            service_override=service_state(tmp_path),
        )


@pytest.mark.parametrize("fault", ["after_archive_copy", "after_prune"])
def test_fault_rollback_is_exact(tmp_path: Path, fault: str) -> None:
    module = load()
    root = profile(tmp_path)
    permission_path, seal, _ = permission(module, root, tmp_path)
    before = mutation_digest(module, root)
    with pytest.raises(RuntimeError, match=fault):
        module.execute(
            root,
            permission_path,
            seal,
            tmp_path / "receipt.json",
            fault=fault,
            service_override=service_state(tmp_path),
        )
    assert mutation_digest(module, root) == before
    assert not (root / "data/task27-supplemental-cleanup").exists()


def test_success_is_frozen_terminal_and_one_use(tmp_path: Path) -> None:
    module = load()
    root = profile(tmp_path)
    permission_path, seal, _ = permission(module, root, tmp_path)
    output = tmp_path / "receipt.json"
    result = module.execute(
        root,
        permission_path,
        seal,
        output,
        service_override=service_state(tmp_path),
    )
    assert result["status"] == "COMMITTED"
    assert result["terminal"]["active_target_matches"] == 0
    assert not (
        root / "data/onboarding/telegram-publication-outbox-v1"
    ).exists()
    assert not (
        root / "data/onboarding/telegram-staff-membership-v1"
    ).exists()
    archive = Path(result["archive"]["root"])
    assert stat.S_IMODE(archive.stat().st_mode) == 0o500
    assert all(
        not (path.stat().st_mode & 0o222)
        for path in [archive, *archive.rglob("*")]
    )
    with pytest.raises(module.Refusal, match="output already exists|reuse"):
        module.execute(
            root,
            permission_path,
            seal,
            output,
            service_override=service_state(tmp_path),
        )


def test_execute_is_disposable_only(tmp_path: Path) -> None:
    module = load()
    root = profile(tmp_path)
    permission_path, seal, _ = permission(module, root, tmp_path)
    (root / ".task27-supplemental-disposable-copy").unlink()
    with pytest.raises(module.Refusal, match="disposable"):
        module.execute(
            root,
            permission_path,
            seal,
            tmp_path / "receipt.json",
            service_override=service_state(tmp_path),
        )


def test_cli_bad_seal_and_success(tmp_path: Path) -> None:
    root = profile(tmp_path)
    state = service_state(tmp_path)
    permission_path = tmp_path / "permission.json"
    dry = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER),
            "dry-run",
            "--profile",
            str(root),
            "--output",
            str(permission_path),
            "--test-service-state",
            str(state),
        ],
        text=True,
        capture_output=True,
    )
    assert dry.returncode == 0, dry.stderr
    seal = json.loads(permission_path.read_text())["permission_seal"]
    bad = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER),
            "execute",
            "--profile",
            str(root),
            "--permission-file",
            str(permission_path),
            "--permission-seal",
            "0" * 64,
            "--output",
            str(tmp_path / "bad.json"),
            "--test-service-state",
            str(state),
        ],
        text=True,
        capture_output=True,
    )
    assert bad.returncode == 2
    assert "permission seal mismatch" in bad.stderr
    good = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER),
            "execute",
            "--profile",
            str(root),
            "--permission-file",
            str(permission_path),
            "--permission-seal",
            seal,
            "--output",
            str(tmp_path / "good.json"),
            "--test-service-state",
            str(state),
        ],
        text=True,
        capture_output=True,
    )
    assert good.returncode == 0, good.stderr