"""Architecture Lane Inventory: admit inventory_record stock state. P1-RUNTIME-INV-001 (authorized delta, additive): Cloud now has a real inventory.reserve/inventory.release execution primitive (docs/audits/ARCHITECTURE_LANE_INVENTORY.md), operating on a dedicated cloud_inventory_reservations ledger — not new fields on inventory_record. Still no checkout lock, warehouse workflow, purchasing, shipping, or automatic stock movement inferred from orders/billing/checkout. """ import re from pathlib import Path import pytest import yaml from src.domain.app_blocks.canonical import CANONICAL_BLOCK_IDS, feature_pack_only_aliases from src.domain.app_blocks.catalogs import FEATURE_PACK_ALIASES, default_known_packs from src.domain.app_blocks.execution import assert_install_plan_installable from src.domain.app_blocks.generated_product_conformance_12h import ( KernelDeniedError, PlatformKernel, ) from src.domain.app_blocks.planner import ProductIntent, select_capabilities from src.domain.app_blocks.registry import load_default_registry from src.domain.app_blocks.vertical_recipes import match_vertical_recipe from src.domain.starter_library.catalog import known_contract_ids from tests.test_phase11_pack_migration_materialization import _captured_composition ROOT = Path(__file__).resolve().parents[1] RECEIPT = ROOT / "src/domain/app_blocks/recipes/inventory_freeze.receipt.txt" AUDIT = ROOT / "src/domain/app_blocks/recipes/inventory_architecture_audit.receipt.txt" REGISTRY = ROOT / "src/domain/app_blocks/platform_coverage_registry.v1.yaml" GATED = ( "payments.checkout", "membership.portal", ) def _host_fragment(root: Path, implementation: str) -> str: rel, _, symbol = implementation.partition("#") text = (root / rel).read_text(encoding="utf-8") if not symbol: return text match = re.search( rf"(?:export\s+)?(?:async\s+)?function\s+{re.escape(symbol)}\b", text, ) if not match: return text rest = text[match.end() :] nxt = re.search(r"\nexport\s+(?:async\s+)?function\s+", rest) end = match.end() + nxt.start() if nxt else len(text) return text[match.start() : end] STARTER_SLOTS = { "list": "record.table", "detail": "record.detail", "create": "record.create", "edit": "record.edit", "empty": "state.empty", } FIELDS = { "sku", "item_name", "location", "quantity", "reorder_level", "unit", "notes", } def inventory_composition(): return _captured_composition( "alembic/versions/148_inventory_management_pack.py", "inventory-management" ) def _ids(intent: ProductIntent) -> set[str]: return {block_id for _cap, block_id, _reason in select_capabilities(intent)} def test_inventory_runtime_is_canonical_without_new_kernel(): registry = load_default_registry() block = registry.require("inventory.stock") canonical = registry.canonical.require("inventory.stock") assert block.status == "implemented" assert "inventory.stock" in CANONICAL_BLOCK_IDS assert "inventory.stock" not in FEATURE_PACK_ALIASES assert "inventory.stock" not in feature_pack_only_aliases() assert canonical.installation.feature_packs == ("inventory-management",) primitives = {dep.id for dep in canonical.requires if dep.kind == "primitive"} assert primitives == {"data.entity", "inventory.reserve", "inventory.release", "inventory.commit"} assert "payment-checkout" not in canonical.installation.feature_packs assert "product-catalog" not in canonical.installation.feature_packs assert "order-management" not in canonical.installation.feature_packs assert "inventory-management" in default_known_packs() composition = inventory_composition() entity = composition["entities"][0] assert entity["entity_key"] == "inventory_record" keys = {field["key"] for field in entity["schema"]["fields"]} assert keys == FIELDS types = {field["key"]: field["type"] for field in entity["schema"]["fields"]} assert types["quantity"] == "number" assert types["reorder_level"] == "number" assert "status" not in keys assert "reserved_quantity" not in keys assert "available_to_promise" not in keys assert "stripe_product_id" not in keys def test_inventory_install_plan_does_not_bind_checkout(monkeypatch): from src.domain.app_blocks import compiler registry = load_default_registry() monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [("inventory_stock", "inventory.stock", "Selected.")], ) plan = compiler.compile_install_plan(None, registry=registry) assert_install_plan_installable(plan) assert "inventory-management" in plan.required_feature_packs assert "payment-checkout" not in plan.required_feature_packs assert "data.entity" in plan.required_primitives assert {"inventory.reserve", "inventory.release", "inventory.commit"} <= set(plan.required_primitives) def test_positive_intent_selects_inventory(): for prompt in ( "I need to track inventory quantities for our products.", "Staff need to update stock levels.", "Build an inventory management screen for our repair parts.", "Track SKU, location, quantity and reorder level.", "We need staff to manage stock at each location.", ): ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "inventory.stock" in ids, prompt assert not (ids & set(GATED)), prompt def test_negative_intent_skips_inventory(): for prompt in ( "Client records. Do not add inventory.", "Staff CRM. No stock management.", "We don't track inventory.", "Case tracking. No inventory system yet.", "Inventory is handled externally.", "Internal tool. Do not build warehouse features.", ): ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "inventory.stock" not in ids, prompt def test_false_positives_do_not_select_inventory(): for purpose in ( "customer orders 3 items", "show product availability text", "Staff manage a product catalog", "Staff manage customer orders", "pricing page", "warehouse address on the contact page", "shipping labels later", "suppliers and quotes", "SKU printed on the brochure", "materials for the workshop agenda", ): ids = _ids(ProductIntent(purpose=purpose, source_prompt=purpose, staff_review=True)) assert "inventory.stock" not in ids, purpose def test_catalog_and_inventory_compose_without_commerce(): prompt = "Staff product catalog and inventory management. No checkout." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert {"catalog.product", "inventory.stock"} <= ids assert "orders.order" not in ids assert "billing.invoice" not in ids assert not (ids & set(GATED)) def test_orders_and_inventory_compose_without_automation_language(): prompt = "Staff manage customer orders and inventory quantities. No checkout." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert {"orders.order", "inventory.stock"} <= ids assert not (ids & set(GATED)) def test_cross_archetype_inventory_is_the_same_block(): prompts = ( "Retail staff inventory management for SKU stock quantities.", "Repair shop inventory management for parts at each location.", "Restaurant inventory quantities for ingredients with unit kg.", "Clinic inventory records for supply items in storage.", "Event crew inventory management for consumable stock.", ) for prompt in prompts: ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "inventory.stock" in ids, prompt def test_13a_do_not_create_orders_still_misses_inventory(): prompt = ( "Our sales office maintains catalog items in a shared list. " "Employees should sign in, add items and correct descriptions. Do not create orders." ) assert match_vertical_recipe(prompt) is None ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "catalog.product" in ids assert "inventory.stock" not in ids def test_inventory_reuses_record_starter_contracts(): registry = load_default_registry() block = registry.require("inventory.stock") contracts = { slot.id: slot.component_contract for slot in block.slots if slot.component_contract } assert STARTER_SLOTS.items() <= contracts.items() assert "inventory.list" not in known_contract_ids() def test_inventory_runtime_crud_and_decimal_quantity(): kernel = PlatformKernel(namespace="inventory-freeze") kernel.as_anonymous() with pytest.raises(KernelDeniedError): kernel.create_record("inventory_record", {"sku": "WIDGET-BLUE-M"}) kernel.as_staff() created = kernel.create_record( "inventory_record", { "sku": "WIDGET-BLUE-M", "item_name": "Blue widget", "location": "Main Warehouse", "quantity": 42, "reorder_level": 10, "unit": "each", }, ) listed = kernel.list_records("inventory_record") assert listed[0]["id"] == created["id"] kernel.patch_record("inventory_record", created["id"], {"quantity": 1.5, "location": "Back Room", "reorder_level": 4}) read = kernel.get_record("inventory_record", created["id"]) assert read["quantity"] == 1.5 assert read["location"] == "Back Room" assert read["reorder_level"] == 4 assert "status" not in read other = PlatformKernel(namespace="inventory-other") other.as_staff() assert other.list_records("inventory_record") == [] def test_orders_do_not_mutate_inventory(): kernel = PlatformKernel(namespace="order-stock") kernel.as_staff() stock = kernel.create_record( "inventory_record", {"sku": "PART-1", "quantity": 10, "reorder_level": 2}, ) order = kernel.create_record( "order_record", {"reference": "WO-1", "customer_name": "Ada", "status": "open"}, ) kernel.patch_record("order_record", order["id"], {"status": "fulfilled"}) kernel.patch_record("order_record", order["id"], {"status": "cancelled"}) assert kernel.get_record("inventory_record", stock["id"])["quantity"] == 10 def test_billing_does_not_mutate_inventory(): kernel = PlatformKernel(namespace="bill-stock") kernel.as_staff() stock = kernel.create_record("inventory_record", {"sku": "PART-1", "quantity": 7}) invoice = kernel.create_record( "invoice_record", {"reference": "INV-1", "customer_name": "Ada", "status": "issued"}, ) kernel.patch_record("invoice_record", invoice["id"], {"status": "paid"}) assert kernel.get_record("inventory_record", stock["id"])["quantity"] == 7 def test_catalog_and_inventory_records_coexist(): kernel = PlatformKernel(namespace="catalog-stock") kernel.as_staff() item = kernel.create_record( "catalog_item", {"name": "Widget", "sku": "W1", "status": "active"}, ) stock = kernel.create_record( "inventory_record", {"sku": "W1", "item_name": "Widget", "quantity": 3}, ) assert item["id"] != stock["id"] assert kernel.get_record("catalog_item", item["id"])["sku"] == "W1" assert kernel.get_record("inventory_record", stock["id"])["quantity"] == 3 def test_materialized_inventory_hosts_use_records_not_checkout(): import json from src.domain.app_blocks.hosts import merge_host_map for framework in ("html", "nuxt_app", "react_app"): root = ROOT / "boilerplates" / framework hosts = merge_host_map(json.loads((root / "p2hs/app-block-hosts.json").read_text())) rows = {h["slot"]: h for h in hosts["hosts"] if h["block"] == "inventory.stock"} assert {"list", "detail", "create", "edit", "empty"} <= set(rows) for slot in ("list", "detail", "create", "edit"): source = _host_fragment(root, rows[slot]["implementation"]) assert "inventory_record" in source assert "createCheckout" not in source assert "/api/checkout" not in source assert "reserved_quantity" not in source assert "/api/inventory/reservations" not in source assert "/api/jobs/inventory." not in source for slot in ("reserve", "release", "reservations", "reservation_detail"): source = _host_fragment(root, rows[slot]["implementation"]) assert "/v1/projects/" not in source assert "p2hs-cloud" not in source assert "cloudgate" not in source assert "INTERNAL_CONTROL_TOKEN" not in source assert 'entity-key="inventory_record"' not in source assert 'entityKey="inventory_record"' not in source assert 'entity_key: "inventory_record"' not in source assert "/api/jobs/inventory.reserve" in _host_fragment(root, rows["reserve"]["implementation"]) release = _host_fragment(root, rows["release"]["implementation"]) assert "/api/jobs/inventory.release" in release assert "reservation_id" in release assert "/api/inventory/reservations" in _host_fragment(root, rows["reservations"]["implementation"]) detail = _host_fragment(root, rows["reservation_detail"]["implementation"]) assert "/api/inventory/reservations/" in detail assert "route.params.id" in detail or "recordId" in detail or "props.id" in detail or "props.recordId" in detail def test_inventory_operational_slots_project_static_query_params(): from pydantic import ValidationError from src.domain.app_blocks.hosts import host_map_payload from src.domain.app_blocks.schemas import AppBlockSlot from src.integrations.platform_ports import CATALOG payload = host_map_payload() rows = {h["slot"]: h for h in payload["hosts"] if h["block"] == "inventory.stock"} assert rows["reservations"]["query_source"] == "port:inventory.reservations.list" assert rows["reservations"]["query_params"] == {"status": "active"} assert rows["reservation_detail"]["query_source"] == "port:inventory.reservations.get" assert "{{" not in str(rows["reservations"]["query_params"]) with pytest.raises(ValidationError): AppBlockSlot( id="reservations", type="ui_slot", description="x", query_params={"inventory_record_id": "{{ selected.id }}"}, ) port_ids = {p["port_id"] for p in CATALOG["ports"]} assert {"inventory.reservations.list", "inventory.reservations.get"} <= port_ids assert tuple(int(part) for part in CATALOG["version"].split(".")) >= (3, 3, 0) for port_id in ("inventory.reservations.list", "inventory.reservations.get"): port = next(item for item in CATALOG["ports"] if item["port_id"] == port_id) assert port["cloud"]["method"] == "GET" assert port["cloud"]["path"].startswith("/v1/projects/:project_id/") assert port["bff"]["path"].startswith("/api/") assert "/v1/projects/" not in port["bff"]["path"] def test_inventory_freeze_receipts_and_coverage_next_family(): receipt = RECEIPT.read_text(encoding="utf-8") audit = AUDIT.read_text(encoding="utf-8") for required in ( "inventory_admitted=yes", "canonical_app_block=inventory.stock", "canonical_feature_pack=inventory-management", "automatic_order_stock_mutation=no", "family_frozen=yes", "next_architecture_family=membership", "catalog_composition_proven=pass", ): assert required in receipt or required in audit, required registry = yaml.safe_load(REGISTRY.read_text(encoding="utf-8")) by_id = {row["id"]: row for row in registry["capabilities"]} assert by_id["inventory.stock"]["status"] == "frozen" assert by_id["inventory"]["status"] == "frozen" assert registry["next_architecture_family"] == "not_yet_selected" assert registry["family_just_frozen"] == "visualization_calculation" def test_inventory_production_smoke_script(): from scripts.architecture_lane_inventory_smoke import run_smoke payload = run_smoke() assert payload["pass"] is True, payload["failures"] assert payload["on_disk_inventory_recipes"] == []