"""9D: the existing version-1 plan is a deterministic diagnostic contract.""" import json import pytest from pydantic import ValidationError from src.domain.app_blocks import compiler from src.domain.app_blocks.canonical import CANONICAL_BLOCK_IDS from src.domain.app_blocks.catalogs import default_known_packs, default_primitives from src.domain.app_blocks.errors import InvalidAppBlockManifestError from src.domain.app_blocks.forms_binding import sources_from_plan_and_packs from src.domain.app_blocks.registry import AppBlockRegistry, load_default_registry from src.domain.app_blocks.schemas import ( AppBlockManifest, AppBlockPlan, CompositionFallback, DependencyRef, ProviderBinding, ) from src.domain.forms.composition import InstalledPackView @pytest.fixture def registry(): return load_default_registry() def select(monkeypatch, *ids): # The product selector intentionally prefers quote over contact. Exercise # multi-root compilation at its selection seam, without changing that rule. monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [(block_id, block_id, f"Selected {block_id}.") for block_id in ids], ) def changed(registry, **updates): return AppBlockRegistry( [block.model_copy(update=updates.get(block.id, {}), deep=True) for block in registry.all()] ) def dep(kind, id): return DependencyRef(kind=kind, id=id) def test_determinism_across_catalog_and_dependency_order(monkeypatch, registry): select(monkeypatch, "forms.contact", "crm.lead") originals = registry.all() expected = compiler.compile_install_plan(None, registry=registry).model_dump_json() for _ in range(3): assert compiler.compile_install_plan(None, registry=registry).model_dump_json() == expected shuffled = AppBlockRegistry( [ manifest.model_copy( update={ "requires": list(reversed(manifest.requires)), "bindings": list(reversed(manifest.bindings)), "provider_bindings": list(reversed(manifest.provider_bindings)), } ) for manifest in reversed(originals) ] ) assert compiler.compile_install_plan(None, registry=shuffled).model_dump_json() == expected def test_shared_pack_and_selection_presentation_priority(monkeypatch, registry): select(monkeypatch, "forms.contact", "forms.quote_request") first = compiler.compile_install_plan(None, registry=registry) assert [b.id for b in first.blocks] == ["forms.contact", "forms.quote_request"] assert first.required_feature_packs == ["contact-capture"] select(monkeypatch, "forms.quote_request", "forms.contact") reverse = compiler.compile_install_plan(None, registry=registry) # Root order is explicitly presentation priority; install sets are order-free. assert [b.id for b in reverse.blocks] == ["forms.quote_request", "forms.contact"] assert first.required_feature_packs == reverse.required_feature_packs assert first.required_primitives == reverse.required_primitives assert first.configuration_requirements == reverse.configuration_requirements assert first.runtime_dependencies == reverse.runtime_dependencies assert first.gaps == reverse.gaps def test_diamond_closure_is_unique_and_nested_order_is_stable(monkeypatch, registry): select(monkeypatch, "forms.contact") contact = registry.require("forms.contact") nested = [dep("app_block", "auth.profile"), dep("app_block", "auth.sign_in")] overrides = { "forms.contact": {"requires": [*contact.requires, *contact.requires, *nested]}, "auth.profile": {"requires": [dep("app_block", "forms.quote_request")]}, "auth.sign_in": {"requires": [dep("app_block", "forms.quote_request")]}, } first = compiler.compile_install_plan(None, registry=changed(registry, **overrides)) overrides["forms.contact"]["requires"].reverse() second = compiler.compile_install_plan(None, registry=changed(registry, **overrides)) assert first.model_dump_json() == second.model_dump_json() assert not first.unsatisfiable assert first.required_feature_packs == ["contact-capture"] assert len(first.blocks) == 4 assert first.required_primitives == sorted(set(first.required_primitives)) assert first.starter_contracts == sorted(set(first.starter_contracts)) assert len(first.blocks[0].dependencies) == len(contact.requires) + len(nested) @pytest.mark.parametrize( "kind,id,expected", [ ("primitive", "data.entity", "primitive_missing"), ("feature_pack", "contact-capture", "feature_pack_missing"), ], ) def test_missing_required_availability_is_blocking(monkeypatch, registry, kind, id, expected): select(monkeypatch, "forms.contact") kwargs = ( {"primitives": default_primitives() - {id}} if kind == "primitive" else {"known_packs": default_known_packs() - {id}} ) plan = compiler.compile_install_plan(None, registry=registry, **kwargs) assert plan.unsatisfiable assert any( g.type == expected and g.id == id and g.severity == "build_blocking" for g in plan.gaps ) @pytest.mark.parametrize("id", ["missing.block", "crm-pipeline", "contact-capture", "contact"]) def test_unknown_selection_never_falls_back_across_namespaces(monkeypatch, registry, id): select(monkeypatch, id) plan = compiler.compile_install_plan(None, registry=registry) assert plan.unsatisfiable assert not plan.blocks and not plan.required_feature_packs assert plan.gaps[0].type == "block_missing" @pytest.mark.parametrize("implemented", [True, False]) def test_invalid_nested_required_block_is_blocking(monkeypatch, registry, implemented): select(monkeypatch, "forms.contact") nested_id = "missing.block" if implemented else "auth.profile" custom = changed( registry, **{ "forms.contact": {"requires": [dep("app_block", nested_id)]}, "auth.profile": {"status": "planned"}, }, ) plan = compiler.compile_install_plan(None, registry=custom) assert plan.unsatisfiable assert any(g.type == "block_missing" and g.id == nested_id for g in plan.gaps) @pytest.mark.parametrize("fallback", [False, True]) @pytest.mark.parametrize("length", [1, 3]) def test_cycles_are_deterministic_blocking_diagnostics(monkeypatch, registry, fallback, length): ids = ["forms.contact", "auth.profile", "auth.sign_in"][:length] select(monkeypatch, ids[0]) overrides = {} for index, id in enumerate(ids): target = ids[(index + 1) % len(ids)] overrides[id] = { "requires": [] if fallback else [dep("app_block", target)], "optional_requires": [], "status": "planned" if fallback else "implemented", "composition_fallback": CompositionFallback(supported=True, blocks=[target]) if fallback else None, } custom = changed(registry, **overrides) plan = compiler.compile_install_plan(None, registry=custom) assert plan.unsatisfiable cycles = [g for g in plan.gaps if g.type == "unsupported"] assert len(cycles) == 1 assert set(cycles[0].affects) == set(ids) assert " -> " in cycles[0].message and "dependency cycle" in cycles[0].message reordered = AppBlockRegistry(list(reversed(custom.all()))) assert compiler.compile_install_plan(None, registry=reordered).model_dump() == plan.model_dump() def test_unselected_and_optional_cycles_do_not_enter_hard_graph(monkeypatch, registry): select(monkeypatch, "forms.contact") custom = changed( registry, **{ "auth.profile": {"requires": [dep("app_block", "auth.profile")]}, "forms.contact": {"optional_requires": [dep("app_block", "forms.contact")]}, }, ) plan = compiler.compile_install_plan(None, registry=custom) assert not plan.unsatisfiable assert [b.id for b in plan.blocks] == ["forms.contact"] def test_fallback_pack_availability_is_validated(monkeypatch, registry): select(monkeypatch, "forms.contact") custom = changed( registry, **{ "forms.contact": { "status": "planned", "requires": [], "composition_fallback": CompositionFallback( supported=True, feature_packs=["forms.public_submission"] ), } }, ) missing = compiler.compile_install_plan(None, registry=custom, known_packs=frozenset()) assert missing.unsatisfiable assert missing.required_feature_packs == ["contact-capture"] assert any(g.type == "feature_pack_missing" for g in missing.gaps) assert not compiler.compile_install_plan(None, registry=custom).unsatisfiable def test_optional_availability_cannot_change_any_serialized_field(monkeypatch, registry): select(monkeypatch, "forms.contact") custom = changed( registry, **{ "forms.contact": { "optional_requires": [ *registry.require("forms.contact").optional_requires, dep("primitive", "email.transactional"), dep("feature_pack", "crm.lead"), ] } }, ) unavailable = AppBlockRegistry([b for b in custom.all() if b.family != "admin"]) first = compiler.compile_install_plan( None, registry=unavailable, primitives=default_primitives() - {"email.send"}, known_packs=default_known_packs() - {"lead-lifecycle"}, ) second = compiler.compile_install_plan( None, registry=custom, primitives=default_primitives() | {"crm.contact.upsert"}, ) assert not first.unsatisfiable and not second.unsatisfiable assert first.model_dump_json() == second.model_dump_json() assert first.required_feature_packs == ["contact-capture"] assert [b.id for b in first.blocks] == ["forms.contact"] observations = first.blocks[0].optional_requirements assert all(o.status == "unavailable" for o in observations) assert all(o.status == "satisfied" for o in second.blocks[0].optional_requirements) assert len([o for o in observations if o.id == "email.send"]) == 1 assert "optional_requirements" not in first.model_dump_json() def test_canonical_aliases_in_all_public_projections(monkeypatch, registry): select(monkeypatch, "forms.contact") custom = changed( registry, **{ "forms.contact": { "requires": [ dep("primitive", "email.transactional"), dep("primitive", "email.send"), ], "provider_bindings": [ ProviderBinding(capability="email.transactional"), ProviderBinding(capability="email.send"), ], } }, ) plan = compiler.compile_install_plan(None, registry=custom) assert "email.transactional" not in plan.model_dump_json() assert plan.required_primitives == ["email.send"] assert len(plan.blocks[0].dependencies) == 1 assert len(plan.configuration_requirements) == 1 assert len(plan.gaps) == 1 assert plan.runtime_dependencies == ["email.send"] # Raw YAML and canonical ontology remain frozen; no manifest rewrite. assert custom.require("forms.contact").requires[0].id == "email.transactional" def test_registry_rejects_invalid_in_memory_starter_reference(registry): contact = registry.require("forms.contact").model_copy(deep=True) contact.slots[0].component_contract = "unknown.starter" with pytest.raises(InvalidAppBlockManifestError, match="unknown Starter contract"): AppBlockRegistry([contact]) def test_invalid_canonical_input_and_artifact_version_are_rejected(registry): with pytest.raises(ValidationError): AppBlockManifest.model_validate({"id": "malformed"}) with pytest.raises(InvalidAppBlockManifestError, match="unknown capability"): changed( registry, **{"forms.contact": {"requires": [dep("primitive", "unknown.capability")]}} ) with pytest.raises(ValidationError): AppBlockPlan.model_validate({"schema_version": "2"}) def test_artifact_shape_and_actual_writer_optional_isolation(monkeypatch, tmp_path, registry): from src.workers.discovery_apply import write_brief_files select(monkeypatch, "forms.contact") result = {"specification": "Contact form", "spec": {"product": {"purpose": "Contact"}}} write_brief_files(tmp_path, result) path = tmp_path / "p2hs" / "app-blocks-plan.json" first = path.read_text(encoding="utf-8") payload = json.loads(first) assert set(payload) == { "schema_version", "requested_capabilities", "blocks", "required_feature_packs", "required_primitives", "configuration_requirements", "gaps", "runtime_dependencies", "starter_contracts", "adoption", "matched_recipe_id", "starter_id", "feature_pack_ids", "forbidden_capabilities", "product_archetype", "unsupported_assumptions", "composition_recipe_version", "collection_recipe_ids", } assert set(payload["blocks"][0]) == { "id", "version", "reason", "slots", "dependencies", "protected_bindings", } assert set(payload["blocks"][0]["dependencies"][0]) == {"kind", "id", "min_version"} assert set(payload["configuration_requirements"][0]) == {"capability", "providers", "behavior"} assert set(payload["gaps"][0]) == {"type", "id", "severity", "affects", "message"} assert payload["schema_version"] == "1" and payload["adoption"] == "authoritative" assert AppBlockPlan.model_validate(payload).model_dump() == payload monkeypatch.setattr( compiler, "default_primitives", lambda: frozenset(default_primitives() - {"email.send"}) ) write_brief_files(tmp_path, result) assert path.read_text(encoding="utf-8") == first assert "optional_requirements" not in first packs = json.loads((tmp_path / "p2hs" / "feature-packs.json").read_text()) assert [p["pack_slug"] for p in packs["install"]] == ["contact-capture"] def test_catalog_and_forms_runtime_identity_stay_distinct(monkeypatch, registry): select(monkeypatch, "forms.contact", "forms.quote_request") plan = compiler.compile_install_plan(None, registry=registry) assert registry.canonical.ids == CANONICAL_BLOCK_IDS assert len(registry.canonical.blocks) == len(CANONICAL_BLOCK_IDS) sources = sources_from_plan_and_packs( plan, [ InstalledPackView(slug="contact-capture", status="installed", config={}), ], ) assert {s.form_key for s in sources} == {"contact", "quote_request"} assert all(s.form_key not in {b.id for b in plan.blocks} for s in sources)