from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from conftest import make_runtime

from scopewrit.errors import ReplayError
from scopewrit.ledger import verify_ledger
from scopewrit.models import ActionProposal, Decision, Effect, ScopeRule


def _proposal(runtime, index: int) -> ActionProposal:
    return ActionProposal(
        request_id=f"concurrent-{index:03d}",
        run_id=runtime.manifest.run_id,
        action="mcp.call",
        resource="mcp://records/read",
    )


def test_concurrent_rule_limit_is_atomic_and_ledger_stays_contiguous(
    tmp_path: Path,
) -> None:
    runtime = make_runtime(
        tmp_path / "runtime",
        rules=(
            ScopeRule(
                rule_id="five-concurrent-reads",
                effect=Effect.ALLOW,
                action="mcp.call",
                resource="mcp://records/read",
                max_uses=5,
            ),
        ),
    )

    def authorize(index: int) -> Decision:
        return runtime.engine.authorize(
            attestation=runtime.attestation,
            proposal=_proposal(runtime, index),
        ).decision

    with ThreadPoolExecutor(max_workers=10) as pool:
        decisions = list(pool.map(authorize, range(20)))

    assert decisions.count(Decision.ALLOW) == 5
    assert decisions.count(Decision.DENY) == 15
    ledger_path = (
        runtime.engine.data_dir / "runs" / runtime.manifest.run_id / "flight-recorder.jsonl"
    )
    status = verify_ledger(ledger_path)
    assert status["event_count"] == 21


def test_concurrent_ticket_replay_has_one_winner(tmp_path: Path) -> None:
    runtime = make_runtime(
        tmp_path / "runtime",
        rules=(
            ScopeRule(
                rule_id="one-read",
                effect=Effect.ALLOW,
                action="mcp.call",
                resource="mcp://records/read",
                max_uses=1,
            ),
        ),
    )
    proposal = _proposal(runtime, 1)
    decision = runtime.engine.authorize(
        attestation=runtime.attestation,
        proposal=proposal,
    )

    def consume() -> str:
        try:
            runtime.engine.consume_ticket(
                decision.ticket,
                proposal=proposal,
                adapter="mcp",
            )
            return "consumed"
        except ReplayError:
            return "replayed"

    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes = list(pool.map(lambda _: consume(), range(2)))
    assert sorted(outcomes) == ["consumed", "replayed"]
