"""Phase 11D4: files capability freeze — static + composition + isolation proofs."""

from __future__ import annotations

from pathlib import Path

from src.domain.app_blocks import compiler
from src.domain.app_blocks.hosts import host_map_payload
from src.domain.app_blocks.planner import ProductIntent, default_slots, select_capabilities
from src.domain.app_blocks.public import _BLOCK_COPY
from src.domain.app_blocks.registry import load_default_registry
from src.domain.connectors.catalog import CONNECTORS
from src.domain.connectors.invoke_capabilities import (
    EXECUTION_DOMAINS,
    STORAGE_FAMILY_PLAN,
)
from src.domain.feature_packs.normalize import load_pack_contract, stamp_composition

REPO = Path(__file__).resolve().parents[1]
MANIFEST = REPO / "src/domain/app_blocks/manifests/files.attachments.v1.yaml"
CATALOG_V2 = REPO / "src/domain/feature_packs/catalog_v2.py"
PLANNER = REPO / "src/domain/app_blocks/planner.py"
MIGRATION = REPO / "alembic/versions/141_file_attachments_pack.py"
ACTIONS_PIPELINE = REPO / "src/domain/codegen/actions_pipeline.py"
STORAGE_TS = (
    Path(r"E:\workroom\2026\P2HSCloud\apps\pc-gateway\src\db\storage.ts")
)
RECORDS_TS = (
    Path(r"E:\workroom\2026\P2HSCloud\apps\pc-gateway\src\routes\records.ts")
)
FILE_ATTACHMENT_SVC = (
    Path(r"E:\workroom\2026\P2HSCloud\apps\pc-gateway\src\services\file-attachment.ts")
)

_PROVIDER_LEAKS = (
    "r2",
    "cloudflare",
    "amazon ses",
    "ses sdk",
    "s3-compatible",
    "storage_key",
    "object_key",
    "bucket_name",
    "access_key",
    "secret_access_key",
)

# 11D3 list filter-after-limit: unauthorized rows are omitted, not returned.
# next_cursor is taken from the last *visible* row. Failure mode is an
# incomplete page, not unauthorized disclosure.
LIST_FILTER_AFTER_LIMIT_DEBT = {
    "classification": ("correctness_only", "ux"),
    "security": False,
    "data_leak": False,
    "availability": False,
    "fix_in_11d4": False,
}


def _load_migration_composition() -> dict:
    import importlib.util
    from unittest.mock import patch

    spec = importlib.util.spec_from_file_location("m141", MIGRATION)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    captured: list[dict] = []

    def fake_bulk_insert(_table, rows):
        captured.extend(rows)

    with patch("alembic.op.bulk_insert", side_effect=fake_bulk_insert):
        module.upgrade()
    matches = [row for row in captured if row.get("slug") == "file-attachments"]
    assert len(matches) == 1
    return matches[0]["composition"]


def test_pagination_debt_is_correctness_and_ux_not_a_leak():
    assert LIST_FILTER_AFTER_LIMIT_DEBT["security"] is False
    assert LIST_FILTER_AFTER_LIMIT_DEBT["data_leak"] is False
    assert LIST_FILTER_AFTER_LIMIT_DEBT["fix_in_11d4"] is False
    assert "correctness_only" in LIST_FILTER_AFTER_LIMIT_DEBT["classification"]
    src = RECORDS_TS.read_text(encoding="utf-8")
    assert "callerCanSeeAttachment" in src
    assert "visible.length === limit" in src
    assert "encodeCursor(lastRecord)" in src


def test_provider_neutrality_product_and_codegen_file_surfaces():
    registry = load_default_registry()
    manifest = registry.require("files.attachments")
    contract = load_pack_contract("file-attachments", {"requires": [], "entities": []})
    hosts = [
        item for item in host_map_payload()["hosts"] if item["block"] == "files.attachments"
    ]
    catalog = CATALOG_V2.read_text(encoding="utf-8")
    overlay = catalog.split('"file-attachments"')[1].split('"approval-workflow"')[0]
    actions = ACTIONS_PIPELINE.read_text(encoding="utf-8")
    media_notes = actions.split("prompt_media_manifest = {", 1)[1].split("}", 1)[0]
    surfaces = [
        MANIFEST.read_text(encoding="utf-8"),
        manifest.description,
        " ".join(manifest.tags),
        _BLOCK_COPY["files.attachments"],
        " ".join(contract.provide_ids()),
        str(default_slots(manifest, ProductIntent())),
        str(hosts),
        overlay,
        media_notes,
    ]
    blob = "\n".join(surfaces).lower()
    for leak in _PROVIDER_LEAKS:
        assert leak not in blob, f"files product/codegen surface leaked {leak!r}"
    assert "platform file_id values are durable" in actions.lower()
    assert "cloud/r2 file_id" not in actions.lower()


def test_planner_still_does_not_treat_storage_attach_as_authority():
    planner = PLANNER.read_text(encoding="utf-8")
    assert "storage.attach" not in planner
    hit = select_capabilities(ProductIntent(purpose="Attach files to cases and leads"))
    assert any(block_id == "files.attachments" for _cap, block_id, _reason in hit)


def test_legacy_storage_attach_is_not_product_or_delete_authority():
    storage = STORAGE_TS.read_text(encoding="utf-8")
    svc = FILE_ATTACHMENT_SVC.read_text(encoding="utf-8")
    records = RECORDS_TS.read_text(encoding="utf-8")
    planner = PLANNER.read_text(encoding="utf-8")
    assert "entity_key = 'file_attachment'" in storage
    assert "countActiveAttachmentsForFile" in storage
    assert "FROM cloud_files" not in svc
    assert "prepareFileAttachmentCreate" in records
    assert "storage.attach" not in records
    assert "storage.attach" not in planner
    for path in (REPO / "src/domain/codegen").rglob("*.py"):
        assert "storage.attach" not in path.read_text(encoding="utf-8"), path.name


def test_p_storage_cannot_satisfy_c_and_c_cannot_replace_p():
    domains = EXECUTION_DOMAINS
    assert domains["managed_r2_does_not_implement_storage_file"] is True
    assert domains["customer_r2_does_not_replace_managed_r2"] is True
    assert "p2hs_managed_r2" not in CONNECTORS
    assert STORAGE_FAMILY_PLAN["managed_r2_in_scope"] is False
    assert STORAGE_FAMILY_PLAN["customer_absent_does_not_use_managed_r2"] is True
    assert "customer_r2_absent_then_managed_r2_for_storage_file" in domains["never"]
    assert domains["p2hs_owned_execution"]["resolver_may_inspect_customer_connections"] is False
    assert domains["customer_bound_execution"]["resolver_may_inspect_p2hs_owned_resources"] is False


def test_composition_source_migration_db_loaded_contract_install_plan():
    persisted = _load_migration_composition()
    entity = persisted["entities"][0]
    assert persisted["schema_version"] == 2
    assert entity["entity_key"] == "file_attachment"
    assert entity["schema"]["schema_mode"] == "protected"
    keys = [field["key"] for field in entity["schema"]["fields"]]
    assert keys == ["file_id", "subject_entity_key", "subject_record_id"]
    assert "workflow" not in entity["schema"]
    assert "domain_events" not in entity["schema"]

    stamped = stamp_composition(
        "file-attachments",
        {
            "requires": [],
            "entities": [
                {
                    "entity_key": "file_attachment",
                    "schema": {
                        "fields": [
                            {"key": "file_id", "type": "text", "required": True, "indexed": True},
                            {"key": "subject_entity_key", "type": "text", "required": True, "indexed": True},
                            {"key": "subject_record_id", "type": "text", "required": True, "indexed": True},
                        ]
                    },
                }
            ],
        },
    )
    assert stamped["entities"][0]["schema"]["schema_mode"] == "protected"
    contract = load_pack_contract("file-attachments", persisted)
    assert "files.attachments" in contract.provide_ids()


def test_install_plan_still_materializes_file_attachments(monkeypatch):
    registry = load_default_registry()
    monkeypatch.setattr(
        compiler,
        "select_capabilities",
        lambda _intent: [("files_attachments", "files.attachments", "Selected.")],
    )
    plan = compiler.compile_install_plan(None, registry=registry)
    assert not plan.unsatisfiable
    assert "file-attachments" in plan.required_feature_packs
    assert "data.entity" in plan.required_primitives