"""Phase 12D: restore status-only record.edit on workflow product surfaces.

Bounded composition. No new Starter contract, App Block, Feature Pack,
primitive, Cloud store, planner rule, or authorization model.
"""
from __future__ import annotations

import json
import shutil
from pathlib import Path

import pytest
from src.domain.app_blocks import compiler
from src.domain.app_blocks.canonical import CANONICAL_BLOCK_IDS
from src.domain.app_blocks.hosts import host_map_payload, merge_host_map
from src.domain.app_blocks.planner import ProductIntent, select_capabilities
from src.domain.app_blocks.registry import load_default_registry
from src.domain.feature_packs.crm_authorization import staff_policy
from src.domain.starter_library.catalog import known_contract_ids
from src.domain.starter_library.record_form import merge_submit_values, plan_record_form_fields

from tests.test_crm_authorization_11e2 import lead_composition
from tests.test_phase11_pack_migration_materialization import _captured_composition

ROOT = Path(__file__).resolve().parents[1]
STARTERS = {
    "nuxt": ROOT / "boilerplates" / "nuxt_app",
    "html": ROOT / "boilerplates" / "html",
    "react": ROOT / "boilerplates" / "react_app",
}

WORKFLOW_BLOCKS = (
    {
        "block": "task.management",
        "detail": "TaskDetail",
        "status": "TaskStatusEdit",
        "entity_key": "platform_task",
        "pack": "task-management",
        "migration": "alembic/versions/137_task_management_pack.py",
        "valid": ("todo", "in_progress"),
        "invalid": ("todo", "bogus"),
    },
    {
        "block": "case.management",
        "detail": "CaseDetail",
        "status": "CaseStatusEdit",
        "entity_key": "case_record",
        "pack": "case-lifecycle",
        "migration": "alembic/versions/133_case_lifecycle_pack.py",
        "valid": ("open", "in_review"),
        "invalid": ("open", "resolved"),
    },
    {
        "block": "workflow.assignment",
        "detail": "AssignmentDetail",
        "status": "AssignmentStatusEdit",
        "entity_key": "assignment",
        "pack": "assignment-tracking",
        "migration": "alembic/versions/136_assignment_tracking_pack.py",
        "valid": ("active", "completed"),
        "invalid": ("completed", "active"),
    },
)

STARTER_IDS_BEFORE_12D = frozenset(known_contract_ids())
BYPASS = (
    "ALLOWED_TRANSITIONS",
    "/api/workflow",
    "transitionTask",
    "localStorage.setItem('status'",
    "pipeline.board",
    "task.board",
    "calendar.schedule",
    "notification.inbox",
)


def _schema(spec: dict) -> dict:
    return _captured_composition(spec["migration"], spec["pack"])["entities"][0]["schema"]


def validate_workflow_transition(schema: dict, existing: dict, incoming: dict) -> dict | None:
    """Mirror Cloud packages/shared/src/records.ts validateWorkflowTransition."""
    workflow = schema.get("workflow") or {}
    field = workflow.get("field")
    if not field:
        return None
    frm = existing.get(field)
    to = incoming.get(field)
    if not isinstance(to, str) or to == frm:
        return None
    if not isinstance(frm, str):
        return None
    allowed = (workflow.get("transitions") or {}).get(frm) or []
    if to in allowed:
        return None
    return {"field": field, "from": frm, "to": to, "allowed": allowed}


def check_access_policy(schema: dict, role: str | None, action: str) -> bool:
    """Mirror Cloud packages/shared/src/records.ts checkAccessPolicy."""
    policy = schema.get("access_policy")
    if not policy:
        return True
    if not role:
        return False
    entry = ((policy.get("roles") or {}).get(role) or {}).get(action) or {}
    return entry.get("effect") == "allow"


def _host_source(root: Path, implementation: str) -> tuple[Path, str]:
    path, _, export = implementation.partition("#")
    source_path = root / path
    return source_path, source_path.read_text(encoding="utf-8")


def _merged_hosts(root: Path) -> dict:
    payload = json.loads((root / "p2hs/app-block-hosts.json").read_text(encoding="utf-8"))
    return merge_host_map(payload)


def test_12d_does_not_mint_contracts_blocks_or_planner_rules():
    assert known_contract_ids() == STARTER_IDS_BEFORE_12D
    assert "record.edit" in known_contract_ids()
    assert "pipeline.board" not in known_contract_ids()
    assert "task.board" not in known_contract_ids()
    # 12D itself minted nothing; GR-05 (Phase 13, explicitly authorized) later
    # admitted support.ticket and GR-06 admitted workflow.approval -- see
    # docs/audits/GR05_EXIT_REVIEW.md and docs/audits/GR06_EXIT_REVIEW.md.
    # Capability Model Phase 4 admitted contract.contract (31 -> 32).
    # Capability Model Phase 5 admitted workforce.worker; later capability
    # waves admitted the current catalog. Keep this assertion tied to the
    # imported canonical set: this test is about 12D not minting contracts,
    # not about freezing an obsolete block count.
    assert len(CANONICAL_BLOCK_IDS) == 61
    assert load_default_registry().canonical.ids == CANONICAL_BLOCK_IDS
    hit = select_capabilities(ProductIntent(purpose="We need an internal task board for staff"))
    assert any(block_id == "task.management" for _cap, block_id, _reason in hit)
    miss = select_capabilities(ProductIntent(purpose="Our task is to help students enroll"))
    assert not any(block_id == "task.management" for _cap, block_id, _reason in miss)


def test_auth_profile_remains_deferred_with_no_edit_slot():
    profile = load_default_registry().require("auth.profile")
    assert not any(slot.type == "ui_slot" for slot in profile.slots)
    payload = host_map_payload()
    assert not any(item["block"] == "auth.profile" for item in payload["hosts"])


def test_yaml_declares_status_only_record_edit():
    registry = load_default_registry()
    for block_id in (
        "task.management",
        "case.management",
        "workflow.assignment",
        "crm.pipeline",
    ):
        manifest = registry.require(block_id)
        slots = {slot.id: slot for slot in manifest.slots}
        assert slots["detail"].component_contract == "record.detail"
        assert slots["status"].component_contract == "record.edit"
        assert 'visible_fields=["status"]' in slots["status"].description
        assert "workflow.transition" in slots["status"].description


@pytest.mark.parametrize("framework", ["html", "nuxt_app", "react_app"])
def test_generated_hosts_compose_status_only_record_edit(framework):
    root = ROOT / "boilerplates" / framework
    hosts = _merged_hosts(root)
    by_key = {(item["block"], item["slot"]): item for item in hosts["hosts"]}
    for spec in WORKFLOW_BLOCKS:
        detail = by_key[(spec["block"], "detail")]
        status = by_key[(spec["block"], "status")]
        assert detail["component_contract"] == "record.detail"
        assert status["component_contract"] == "record.edit"
        _, detail_src = _host_source(root, detail["implementation"])
        _, status_src = _host_source(root, status["implementation"])
        assert spec["detail"] in detail["implementation"] or spec["detail"] in detail_src
        assert "RecordEdit" in detail_src or "mountRecordEdit" in detail_src
        assert "visible-fields" in detail_src or "visibleFields" in detail_src
        assert "['status']" in detail_src or '["status"]' in detail_src
        assert "RecordEdit" in status_src or "mountRecordEdit" in status_src
        assert "['status']" in status_src or '["status"]' in status_src
        for leaked in BYPASS:
            assert leaked not in detail_src
            assert leaked not in status_src
        if "visible-fields=" in detail_src:
            assert 'visible-fields="[\'title\']"' not in detail_src
            assert 'visible-fields="[\'title\', \'status\']"' not in detail_src


@pytest.mark.parametrize("framework", ["html", "nuxt_app", "react_app"])
def test_crm_pipeline_declaration_parity_does_not_change_detail_host(framework):
    root = ROOT / "boilerplates" / framework
    hosts = _merged_hosts(root)
    pipeline = {item["slot"]: item for item in hosts["hosts"] if item["block"] == "crm.pipeline"}
    assert pipeline["detail"]["component_contract"] == "record.detail"
    assert pipeline["status"]["component_contract"] == "record.edit"
    _, detail_src = _host_source(root, pipeline["detail"]["implementation"])
    _, status_src = _host_source(root, pipeline["status"]["implementation"])
    assert "LeadDetail" in detail_src or "mountLeadDetail" in detail_src
    assert "visibleFields" in detail_src or "visible-fields" in detail_src
    assert "Change pipeline stage" in detail_src
    assert "Change stage" in detail_src
    assert "RecordEdit" in status_src or "mountRecordEdit" in status_src
    assert "['status']" in status_src or '["status"]' in status_src
    assert "ALLOWED_TRANSITIONS" not in detail_src
    files = {item["slot"]: item for item in hosts["hosts"] if item["block"] == "files.attachments"}
    assert {"pipeline", "detail", "empty"} <= set(files)
    assert files["upload"]["component_contract"] == "file.upload"


def test_status_editor_cannot_submit_arbitrary_non_status_fields():
    for spec in WORKFLOW_BLOCKS:
        schema = {"fields": _schema(spec)["fields"]}
        planned = plan_record_form_fields(schema, ["status"])
        assert [field["key"] for field in planned["visible"]] == ["status"]
        payload = merge_submit_values(
            schema,
            {"status": spec["valid"][1], "title": "hijack", "notes": "nope", "assigned_to": "x"},
            ["status"],
        )
        assert payload == {"status": spec["valid"][1]}
        unrestricted = merge_submit_values(
            schema,
            {"status": spec["valid"][1], "title": "hijack"},
        )
        assert "title" in unrestricted or spec["entity_key"] == "assignment"


def test_workflow_transition_remains_authoritative():
    for spec in WORKFLOW_BLOCKS:
        schema = _schema(spec)
        assert schema["workflow"]["field"] == "status"
        frm, to = spec["valid"]
        assert validate_workflow_transition(schema, {"status": frm}, {"status": to}) is None
        bad_from, bad_to = spec["invalid"]
        violation = validate_workflow_transition(schema, {"status": bad_from}, {"status": bad_to})
        assert violation is not None
        assert violation["field"] == "status"


def test_unauthorized_actor_is_rejected_on_the_existing_governance_path():
    lead_schema = lead_composition()["entities"][0]["schema"]
    assert lead_schema["access_policy"] == staff_policy()
    assert check_access_policy(lead_schema, "admin", "update") is True
    assert check_access_policy(lead_schema, None, "update") is False
    assert check_access_policy(lead_schema, "visitor", "update") is False
    assert check_access_policy(lead_schema, "member", "update") is False

    adapter = (ROOT / "boilerplates/nuxt_app/lib/p2hs-app.js").read_text(encoding="utf-8")
    after_patch = adapter.split("export async function patchRecord", 1)[1]
    patch = after_patch.split("export async function", 1)[0]
    assert "authHeaders" in patch
    assert "method: \"PATCH\"" in patch or "method: 'PATCH'" in patch
    for spec in WORKFLOW_BLOCKS:
        for framework, root in STARTERS.items():
            hosts = json.loads((root / "p2hs/app-block-hosts.json").read_text(encoding="utf-8"))
            status = next(
                item
                for item in hosts["hosts"]
                if item["block"] == spec["block"] and item["slot"] == "status"
            )
            _, source = _host_source(root, status["implementation"])
            assert "patchRecord" in source or "RecordEdit" in source or "mountRecordEdit" in source


def _disposable_app(name: str) -> Path:
    root = ROOT / ".test-tmp" / "12d" / name
    if root.exists():
        shutil.rmtree(root, ignore_errors=True)
    root.mkdir(parents=True)
    return root


@pytest.mark.parametrize("spec", WORKFLOW_BLOCKS, ids=lambda item: item["block"])
def test_disposable_generated_app_contains_status_only_editor(monkeypatch, spec):
    registry = load_default_registry()
    monkeypatch.setattr(
        compiler,
        "select_capabilities",
        lambda _intent: [(spec["block"], spec["block"], "Selected.")],
    )
    plan = compiler.compile_install_plan(None, registry=registry)
    assert not plan.unsatisfiable
    assert "record.edit" in plan.starter_contracts
    assert "workflow.transition" in plan.required_primitives
    assert spec["pack"] in plan.required_feature_packs
    assert spec["block"] in {block.id for block in plan.blocks}
    assert "pipeline.board" not in plan.starter_contracts
    assert "file.upload" not in plan.starter_contracts or spec["block"] != "files.attachments"

    app = _disposable_app(spec["block"])
    p2hs = app / "p2hs"
    p2hs.mkdir(parents=True)
    (p2hs / "app-blocks-plan.json").write_text(plan.model_dump_json(indent=2), encoding="utf-8")
    source_hosts = json.loads(
        (STARTERS["nuxt"] / "p2hs/app-block-hosts.json").read_text(encoding="utf-8")
    )
    filtered = [item for item in source_hosts["hosts"] if item["block"] == spec["block"]]
    (p2hs / "app-block-hosts.json").write_text(
        json.dumps({"version": 1, "hosts": filtered}, indent=2),
        encoding="utf-8",
    )
    by_slot = {item["slot"]: item for item in filtered}
    for slot in ("detail", "status"):
        rel = by_slot[slot]["implementation"]
        src = STARTERS["nuxt"] / rel
        dest = app / rel
        dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src, dest)
        text = dest.read_text(encoding="utf-8")
        assert "RecordEdit" in text
        assert ":visible-fields=\"['status']\"" in text
        assert spec["detail"] in by_slot["detail"]["implementation"]
    detail_text = (app / by_slot["detail"]["implementation"]).read_text(encoding="utf-8")
    assert "AdminRecordDetail" in detail_text
    assert "RecordEdit" in detail_text


def test_disposable_pipeline_app_keeps_existing_detail_and_gains_declaration(monkeypatch):
    registry = load_default_registry()
    monkeypatch.setattr(
        compiler,
        "select_capabilities",
        lambda _intent: [("crm.pipeline", "crm.pipeline", "Selected.")],
    )
    plan = compiler.compile_install_plan(None, registry=registry)
    assert "record.edit" in plan.starter_contracts
    assert "record.detail" in plan.starter_contracts
    app = _disposable_app("crm.pipeline")
    p2hs = app / "p2hs"
    p2hs.mkdir(parents=True)
    (p2hs / "app-blocks-plan.json").write_text(plan.model_dump_json(indent=2), encoding="utf-8")
    source_hosts = json.loads(
        (STARTERS["nuxt"] / "p2hs/app-block-hosts.json").read_text(encoding="utf-8")
    )
    filtered = [item for item in source_hosts["hosts"] if item["block"] == "crm.pipeline"]
    (p2hs / "app-block-hosts.json").write_text(
        json.dumps({"version": 1, "hosts": filtered}, indent=2),
        encoding="utf-8",
    )
    by_slot = {item["slot"]: item for item in filtered}
    for slot in ("detail", "status"):
        rel = by_slot[slot]["implementation"]
        dest = app / rel
        dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(STARTERS["nuxt"] / rel, dest)
    detail = (app / by_slot["detail"]["implementation"]).read_text(encoding="utf-8")
    status = (app / by_slot["status"]["implementation"]).read_text(encoding="utf-8")
    assert "LeadDetail" in detail
    assert ":visible-fields=\"['status']\"" in detail
    assert "Change pipeline stage" in detail
    assert ":visible-fields=\"['status']\"" in status
    assert "LeadDetail" not in status


def test_files_attachments_mounts_existing_upload_without_a_new_block():
    files = load_default_registry().require("files.attachments")
    ui = [slot.id for slot in files.slots if slot.type == "ui_slot"]
    assert ui[:3] == ["pipeline", "detail", "empty"]
    assert "upload" in ui
    assert any(slot.component_contract == "file.upload" for slot in files.slots)
    assert "files.attachments" in CANONICAL_BLOCK_IDS
    # See GR-05/GR-06 comment above -- 19 -> 20 -> 21. Membership admission is 24 -> 25.
    # Capability Model Phase 4 admitted contract.contract (31 -> 32).
    # Capability Model Phase 5 admitted workforce.worker (32 -> 33).
    assert len(CANONICAL_BLOCK_IDS) == 61