from __future__ import annotations

import json
import threading
from concurrent.futures import ThreadPoolExecutor

import pytest

from scopewrit.errors import LedgerError, StateError
from scopewrit.gateway import GuardedExecutor
from scopewrit.ledger import verify_audit_packet, verify_ledger
from scopewrit.models import ActionProposal


def proposal(runtime, name="closure-action"):
    return ActionProposal(
        request_id=name,
        run_id=runtime.manifest.run_id,
        action="network.http",
        resource="https://example.com/authorized/value",
        method="GET",
    )


def test_close_refuses_in_flight_execution_then_returns_stable_packet(runtime):
    started = threading.Event()
    finish = threading.Event()
    action = proposal(runtime)

    def operation():
        started.set()
        assert finish.wait(5)
        return "completed"

    with ThreadPoolExecutor(max_workers=1) as pool:
        pending = pool.submit(
            GuardedExecutor(runtime.engine, adapter="http").execute,
            attestation=runtime.attestation,
            proposal=action,
            operation=operation,
        )
        assert started.wait(5)
        try:
            with pytest.raises(LedgerError, match="await execution outcomes"):
                runtime.engine.close_run(runtime.manifest.run_id)
            assert runtime.engine.state.run_status(runtime.manifest.run_id) == "active"
        finally:
            finish.set()
        assert pending.result(timeout=5).value == "completed"
    packet = runtime.engine.close_run(runtime.manifest.run_id)
    ledger = runtime.engine._recorder(runtime.manifest.run_id)
    before = ledger.path.read_bytes()
    assert runtime.engine.close_run(runtime.manifest.run_id) == packet
    assert runtime.engine.read_audit_packet(runtime.manifest.run_id) == packet
    with pytest.raises(LedgerError, match="closed"):
        ledger.append("late", {})
    assert ledger.path.read_bytes() == before
    assert (
        verify_audit_packet(packet, runtime.engine.issuer_trust, ledger.path).event_count
        == verify_ledger(ledger.path)["event_count"]
    )


def test_closed_state_and_log_recover_after_packet_write_failure(runtime, monkeypatch):
    original = runtime.engine._write_json_atomic

    def fail(*args, **kwargs):
        raise OSError("injected disk interruption")

    monkeypatch.setattr(runtime.engine, "_write_json_atomic", fail)
    with pytest.raises(OSError, match="injected"):
        runtime.engine.close_run(runtime.manifest.run_id)
    assert runtime.engine.state.run_status(runtime.manifest.run_id) == "closed"
    monkeypatch.setattr(runtime.engine, "_write_json_atomic", original)
    packet = runtime.engine.close_run(runtime.manifest.run_id)
    rows = [
        json.loads(line)
        for line in runtime.engine._recorder(runtime.manifest.run_id).path.read_text().splitlines()
    ]
    assert sum(row["event_type"] == "run_closed" for row in rows) == 1
    assert runtime.engine.read_audit_packet(runtime.manifest.run_id) == packet


def test_concurrent_closers_receive_identical_audit_packet(runtime):
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(
            pool.map(lambda _: runtime.engine.close_run(runtime.manifest.run_id), range(8))
        )
    assert all(item == results[0] for item in results)


def test_revocation_stops_new_authority_and_can_be_finalized(runtime):
    runtime.engine.revoke_run(runtime.manifest.run_id, "test withdrawal")
    with pytest.raises(StateError):
        runtime.engine.authorize(attestation=runtime.attestation, proposal=proposal(runtime))
    packet = runtime.engine.close_run(runtime.manifest.run_id)
    assert runtime.engine.read_audit_packet(runtime.manifest.run_id) == packet
    assert runtime.engine.state.run_status(runtime.manifest.run_id) == "revoked"


def test_execution_outcome_requires_matching_consumed_proposal(runtime):
    action = proposal(runtime)
    with pytest.raises(LedgerError, match="consumed ticket"):
        runtime.engine.record_execution(proposal=action, outcome={"status": "completed"})
