"""Dev fixture for hub memo forwarding. No live intake, token, or submitter."""
from __future__ import annotations

import json
from pathlib import Path

import pytest

from api.services.informed_decision_contract import (
    PROVENANCE_NOTE,
    ContractError,
    build_envelope,
    correlate_resolution,
    resolution_applies,
    stable_idempotency_key,
    validate_envelope,
)
from api.services.memo_forwarding import DevIntakeFixture
from api.services.write_idempotency import route_rule_for

DECISION = "62aa95e4-3ccc-4867-8806-5e3db005b483"
MEMO = "infd-20261003-p01"
OTHER = "infd-20261003-p02"
APPROVAL = "ba5ce2d8-8b6d-40be-af89-2e8c147029a3"
BINDING = "sha256:" + "ab" * 32
TOKEN = "dev-fixture-not-a-credential"


class Clock:
    def __init__(self) -> None:
        self.now = 1_000.0

    def __call__(self) -> float:
        return self.now


class Transport:
    def __init__(self, responses):
        self.responses = list(responses)
        self.calls = []

    def __call__(self, method, url, body, headers):
        self.calls.append((method, url, body, dict(headers)))
        item = self.responses.pop(0)
        if isinstance(item, Exception):
            raise item
        return item


def envelope(**overrides):
    memo = {"id": MEMO, "question": "Accept the forwarded draft?"}
    packet = [{"item_id": "brief", "media_type": "text/plain", "content": "Bounded brief."}]
    intent = {
        "action": "accept",
        "principal": "pending-human-session",
        "purpose": "Review the draft.",
        "target": {"id": MEMO, "type": "decision-memo", "system": "informed-decision"},
    }
    fields = dict(
        decision_id=DECISION,
        memo=memo,
        packet=packet,
        memo_class="work-record-accept",
        review_group="net-kingdom-admins",
        approval_intent=intent,
        drafted_by="claude@the-custodian",
        resolves="on-answer",
    )
    fields.update(overrides)
    return build_envelope(**fields)


def receipt(memo_id=MEMO):
    return {
        "memo_id": memo_id,
        "approval_id": APPROVAL,
        "approval_binding_digest": BINDING,
        "echo": "should-not-be-kept",
    }


def fixture(tmp_path, transport, clock=None, token=lambda: TOKEN):
    return DevIntakeFixture(
        tmp_path / "forward.sqlite",
        transport=transport,
        intake_base_url="http://intake.test",
        hub_base_url="http://hub.test",
        token_reader=token,
        clock=clock or Clock(),
    )


def stored_bytes(path) -> bytes:
    data = Path(path).read_bytes()
    for suffix in ("-wal", "-shm"):
        side = Path(str(path) + suffix)
        if side.exists():
            data += side.read_bytes()
    return data


def test_idempotency_key_is_stable_and_drafted_by_is_not_identity():
    first = envelope()
    second = envelope()
    assert first["idempotency_key"] == second["idempotency_key"]
    assert first["idempotency_key"] == stable_idempotency_key(DECISION, MEMO)
    assert "drafted_by" not in first["idempotency_key"]
    assert "submitter" not in first
    changed = dict(first)
    changed["idempotency_key"] = "unstable-key-value"
    with pytest.raises(ContractError) as error:
        validate_envelope(changed)
    assert error.value.code == "idempotency_key_not_stable"


def test_envelope_is_bounded_and_does_not_echo_secrets():
    huge = envelope()
    huge["packet"][0]["content"] = "x" * (256 * 1024 + 1)
    with pytest.raises(ContractError) as too_big:
        validate_envelope(huge)
    assert too_big.value.code == "packet_too_large"
    leaked = envelope()
    leaked["packet"][0]["content"] = "-----BEGIN PRIVATE KEY-----"
    with pytest.raises(ContractError) as secret:
        validate_envelope(leaked)
    assert secret.value.code == "secret_marker"
    assert "PRIVATE KEY" not in str(secret.value)
    extra = envelope()
    extra["highlights"] = ["not-a-notice"]
    with pytest.raises(ContractError) as unknown:
        validate_envelope(extra)
    assert unknown.value.code == "unknown_field"


def test_governed_boolean_leaves_round_trip():
    checked = envelope(memo={
        "id": MEMO,
        "question": "Accept the forwarded draft?",
        "sealed": False,
        "binding": {"target": {"requires_new_bind": True}},
        "highlights": [{"required_ack": True}],
    })
    memo = checked["memo"]
    assert memo["sealed"] is False
    assert memo["binding"]["target"]["requires_new_bind"] is True
    assert memo["highlights"][0]["required_ack"] is True
    rejected = envelope()
    rejected["memo"]["marker"] = object()
    with pytest.raises(ContractError) as error:
        validate_envelope(rejected)
    assert error.value.code == "invalid_type"


def test_pose_replay_conflict_and_asserted_provenance(tmp_path):
    clock = Clock()
    transport = Transport([(201, receipt())])
    fix = fixture(tmp_path, transport, clock)
    posed = fix.submit(envelope())
    assert posed.status == "delivered"
    assert posed.drafted_by == "claude@the-custodian"
    assert posed.drafted_by_provenance == "asserted"
    assert posed.provenance_note == PROVENANCE_NOTE
    assert "asserted provenance" in posed.provenance_note
    assert set(posed.receipt) == {"memo_id", "approval_id", "approval_binding_digest"}
    sent = transport.calls[0]
    assert sent[0:2] == ("POST", "http://intake.test/v1/memos")
    assert sent[2]["drafted_by"] == "claude@the-custodian"
    assert "drafted_by_provenance" not in sent[2]
    assert "submitter" not in sent[2]
    assert sent[3]["Authorization"] == f"Bearer {TOKEN}"
    assert TOKEN.encode() not in stored_bytes(fix.path)

    restarted = DevIntakeFixture(
        fix.path,
        transport=Transport([(201, receipt())]),
        intake_base_url="http://intake.test",
        hub_base_url="http://hub.test",
        token_reader=lambda: TOKEN,
        clock=clock,
    )
    replay = restarted.submit(envelope())
    assert replay.status == "replay"
    assert replay.receipt == posed.receipt
    assert restarted.transport.calls == []

    conflict = envelope()
    conflict["packet"][0]["content"] = "Different brief."
    blocked = restarted.submit(conflict)
    assert blocked.status == "blocked"
    assert blocked.detail == "idempotency_conflict"
    kept = restarted.pose(posed.idempotency_key)
    assert kept["state"] == "delivered"
    assert "Different brief." not in kept["envelope_json"]
    assert restarted.transport.calls == []


def test_delivery_failure_keeps_local_evidence_and_retries(tmp_path):
    clock = Clock()
    transport = Transport([
        ConnectionRefusedError("down"),
        (503, {"error": "intake_unavailable", "body": "LEAKED-BODY"}),
        (201, receipt()),
    ])
    fix = fixture(tmp_path, transport, clock)
    failed = fix.submit(envelope())
    assert failed.status == "pending"
    assert failed.detail == "ConnectionRefusedError"
    row = fix.pose(failed.idempotency_key)
    assert row["state"] == "pending"
    assert "Bounded brief." in row["envelope_json"]
    assert "LEAKED-BODY" not in row["envelope_json"]
    assert fix.run_due() is None

    clock.now += 2
    again = fix.run_due()
    assert again.status == "pending"
    assert again.detail == "http_503"
    assert b"LEAKED-BODY" not in stored_bytes(fix.path)
    assert fix.run_due() is None

    clock.now += 4
    delivered = fix.run_due()
    assert delivered.status == "delivered"
    assert delivered.receipt["memo_id"] == MEMO
    assert fix.run_due() is None
    assert len(transport.calls) == 3


def test_intake_conflict_blocks_without_retry(tmp_path):
    transport = Transport([(409, {"error": "idempotency_conflict", "packet": "LEAK"})])
    fix = fixture(tmp_path, transport)
    blocked = fix.submit(envelope())
    assert blocked.status == "blocked"
    assert blocked.detail == "idempotency_conflict"
    assert b"LEAK" not in stored_bytes(fix.path)
    assert fix.run_due() is None
    assert len(transport.calls) == 1


def test_absent_token_is_not_invented(tmp_path):
    transport = Transport([(201, receipt())])
    fix = fixture(tmp_path, transport, token=None)
    pending = fix.submit(envelope())
    assert pending.status == "pending"
    assert pending.detail == "projected_token_absent"
    assert transport.calls == []
    assert fix.pose(pending.idempotency_key)["state"] == "pending"

    def boom():
        raise RuntimeError("sk-or-this-must-not-be-stored-0123456789")

    fix.token_reader = boom
    fix.clock = Clock()
    fix.clock.now = 1_000 + 2
    retried = fix.run_due()
    assert retried.status == "pending"
    assert retried.detail == "RuntimeError"
    assert transport.calls == []
    assert b"sk-or-this-must-not-be-stored" not in stored_bytes(fix.path)


def test_live_hosts_and_runtime_binding_are_refused(tmp_path):
    with pytest.raises(ValueError):
        DevIntakeFixture(
            tmp_path / "no.sqlite",
            transport=Transport([]),
            intake_base_url="http://127.0.0.1:8000",
            hub_base_url="http://hub.test",
        )
    root = Path(__file__).resolve().parents[1] / "api"
    main = (root / "main.py").read_text()
    assert "memo_notifications" in main
    assert "memo_forwarding" not in main
    assert "DevIntakeFixture" not in main
    owned = [
        root / "main.py",
        root / "services" / "memo_forwarding.py",
        root / "services" / "informed_decision_contract.py",
        root / "services" / "memo_notification.py",
        root / "routers" / "memo_notifications.py",
        root / "models" / "memo_notification.py",
    ]
    for path in owned:
        text = path.read_text()
        assert "INFD_HUB_WRITEBACK" not in text
        assert "IntakeDeliveryWorker" not in text
        assert "/var/run/secrets" not in text
        assert "token_file" not in text
    assert route_rule_for("POST", "/informed-decision/notifications") is None


@pytest.mark.parametrize("verb,serves", [
    ("return", {"hub_decision_id": DECISION, "resolves": "on-answer"}),
    ("discuss", {"hub_decision_id": DECISION, "resolves": "on-answer"}),
    ("accept", {"hub_decision_id": DECISION, "resolves": "never"}),
    ("decline", None),
])
def test_non_terminal_answers_do_not_resolve(tmp_path, verb, serves):
    assert resolution_applies(verb, serves) is False
    transport = Transport([(200, {})])
    fix = fixture(tmp_path, transport)
    status = fix.observe_resolution(
        decision_id=DECISION, memo_id=MEMO, verb=verb, serves=serves, payload={"rationale": "no"},
    )
    assert status == "not_applicable"
    assert transport.calls == []


def test_resolution_failure_retries_and_other_memo_blocks(tmp_path):
    clock = Clock()
    payload = {
        "decided_by": "Bernd Worsch via informed-decision",
        "rationale": f"accept in informed-decision memo {MEMO} v1; ",
        "write_log": False,
    }
    transport = Transport([
        ConnectionRefusedError("down"),
        (200, {"status": "resolved"}),
        (409, {"error": "already resolved"}),
        (200, {"rationale": "resolved by hand"}),
    ])
    fix = fixture(tmp_path, transport, clock)
    failed = fix.observe_resolution(
        decision_id=DECISION, memo_id=MEMO, verb="accept",
        serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
    )
    assert failed == "failed"
    assert fix.resolutions()[0]["payload_json"]
    assert fix.run_due_resolution() is None
    clock.now += 2
    assert fix.run_due_resolution() == "resolved"
    assert [row["status"] for row in fix.resolutions()] == ["failed", "resolved"]
    assert fix.run_due_resolution() is None

    conflict = fix.observe_resolution(
        decision_id=DECISION, memo_id=OTHER, verb="decline",
        serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
    )
    assert conflict == "conflict"
    calls = len(transport.calls)
    assert transport.calls[-2][0] == "POST"
    assert transport.calls[-1][0] == "GET"
    assert fix.observe_resolution(
        decision_id=DECISION, memo_id=OTHER, verb="decline",
        serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
    ) == "conflict"
    assert len(transport.calls) == calls


def test_unreadable_409_stays_retryable(tmp_path):
    clock = Clock()
    transport = Transport([
        (409, {}),
        (503, {}),
        (409, {}),
        (200, {"rationale": f"decline in informed-decision memo {MEMO} v1; "}),
    ])
    fix = fixture(tmp_path, transport, clock)
    payload = {"rationale": "kept locally", "decided_by": "reviewer", "write_log": False}
    assert fix.observe_resolution(
        decision_id=DECISION, memo_id=MEMO, verb="decline",
        serves={"hub_decision_id": DECISION, "resolves": "on-answer"}, payload=payload,
    ) == "failed"
    assert fix.resolutions()[0]["error_type"] == "resolution_unreadable"
    clock.now += 2
    assert fix.run_due_resolution() == "already_resolved"


async def test_hub_409_still_returns_the_existing_resolution(client):
    domain = await client.post("/domains/", json={"slug": "testdomain", "name": "Test Domain"})
    assert domain.status_code == 201, domain.text
    topic = await client.post("/topics/", json={
        "slug": "testtopic", "title": "Test Topic", "domain": "testdomain",
    })
    assert topic.status_code == 201, topic.text
    created = await client.post("/decisions/", json={
        "title": "Should the draft stand?", "topic_id": topic.json()["id"],
    })
    assert created.status_code == 201, created.text
    decision_id = created.json()["id"]
    rationale = f"accept in informed-decision memo {MEMO} v1; approval {APPROVAL}"
    resolved = await client.post(f"/decisions/{decision_id}/resolve", json={
        "rationale": rationale,
        "decided_by": "Bernd Worsch via informed-decision",
        "write_log": False,
    })
    assert resolved.status_code == 200, resolved.text
    conflict = await client.post(f"/decisions/{decision_id}/resolve", json={
        "rationale": "resolved by hand",
        "decided_by": "someone else",
        "write_log": False,
    })
    assert conflict.status_code == 409
    current = await client.get(f"/decisions/{decision_id}")
    assert current.status_code == 200
    body = current.json()
    assert body["rationale"] == rationale
    assert correlate_resolution(MEMO, post_status=409, existing=body) == "already_resolved"
    assert correlate_resolution(OTHER, post_status=409, existing=body) == "conflict"