"""Phase 12H — plan-derived generated-product conformance certification. Certifies Intent → Discovery → Planner → install plan → Codegen MUST-mount → generated application → auth → declared runtime authority → Cloud persist → read-back across the 6×3 matrix. No new App Block, Feature Pack, Starter contract, planner rule, kernel primitive, or Cloud authority. """ from __future__ import annotations import hashlib import json import re import shutil from dataclasses import dataclass, field from pathlib import Path from uuid import uuid4 from src.domain.app_blocks.adoption import adoption_for, apply_adoption from src.domain.app_blocks.canonical import FORM_RUNTIME_BY_BLOCK from src.domain.app_blocks.compiler import compile_install_plan from src.domain.app_blocks.execution import assert_install_plan_installable from src.domain.app_blocks.quote_vertical import QUOTE_TRANSITIONS from src.domain.app_blocks.registry import load_default_registry from src.domain.app_blocks.schemas import AppBlockPlan, PlannedBlock from src.domain.codegen.app_sdk import app_sdk_files, ensure_required_packs_from_plan from src.domain.projects.discovery_schemas import ( Actor, AuthenticationSpec, DiscoverySpec, Entity, ProductDefinition, TenancySpec, Workflow, ) from src.domain.projects.models import PROJECT_SOURCE_KIND_GENERATED REPO_ROOT = Path(__file__).resolve().parents[3] STARTERS = ("nuxt_app", "react_app", "html") ARCHETYPES = ("A", "B", "C", "D", "E", "F") STARTER_ROOTS = { "nuxt_app": REPO_ROOT / "boilerplates" / "nuxt_app", "react_app": REPO_ROOT / "boilerplates" / "react_app", "html": REPO_ROOT / "boilerplates" / "html", } MATRIX_PATH = REPO_ROOT / "docs" / "audits" / "fixtures" / "12h_certification_matrix.json" ENTITY_BY_BLOCK = { "crm.lead": "lead", "crm.pipeline": "lead", "crm.customer": "customer", "case.management": "case_record", "task.management": "platform_task", "workflow.assignment": "assignment", "activity.timeline": "activity_entry", "notifications.in_app": "platform_notification", "files.attachments": "file_attachment", "documents": "document", "catalog.product": "catalog_item", "orders.order": "order_record", "billing.invoice": "invoice_record", "inventory.stock": "inventory_record", "membership.member": "membership_record", "forms.quote_request": "quote_request", "forms.contact": "contact", "reporting.dashboard": "report_definition", "crm.organization": "organization", "crm.team": "team", "crm.org_unit": "org_unit", "crm.quote": "quote", "collaboration.comment": "comment", "governance.group": "governance_group", "customer_success.account": "customer_success_account", "contract.contract": "contract", "workforce.worker": "worker_record", "payments.checkout": "order", } WORKFLOWS = { "lead": { "field": "status", "transitions": { "new": ["contacted", "lost"], "contacted": ["qualified", "lost"], "qualified": ["won", "lost"], "won": [], "lost": [], }, }, "case_record": { "field": "status", "transitions": { "open": ["in_review", "closed"], "in_review": ["waiting", "resolved", "closed"], "waiting": ["in_review", "closed"], "resolved": ["closed"], "closed": [], }, }, "quote_request": {"field": "status", "transitions": QUOTE_TRANSITIONS}, "platform_task": { "field": "status", "transitions": { "todo": ["in_progress"], "in_progress": ["todo", "done"], "done": ["in_progress"], }, }, "assignment": { "field": "status", "transitions": {"active": ["completed"], "completed": []}, }, } PRIMARY_BLOCKS = { "A": ("crm.customer", "crm.lead"), "B": ("case.management",), "C": ("forms.quote_request",), "D": ("documents", "files.attachments"), "E": ("booking.appointment",), "F": ("catalog.product",), } FORBIDDEN_MARKERS = ( "better-sqlite3", "new PrismaClient", "from 'prisma'", 'from "prisma"', "firebase", "mongoose", "createClient(", "from 'supabase'", 'from "supabase"', "localStorage.setItem('status'", "JSON.parse(fs.readFileSync", "createCheckout", "/api/checkout", "/api/cart", "storage_key", "object_key", "bucket_name", "api.hubapi.com", "ALLOWED_TRANSITIONS", ) COMMERCE_MARKERS = ("createCheckout", "/api/checkout", "/api/cart", "stripe.checkout") _STARTER_COPY = { "nuxt_app": ( "components/starter", "lib/entity-schemas", "lib/record-form.mjs", "layouts", ), "react_app": ( "src/components/starter", "src/lib/entity-schemas", "src/lib/record-form.mjs", ), "html": ( "js/starter-blocks.js", "js/starter-library.js", "js/booking-blocks.js", "js/catalog-blocks.js", "js/order-blocks.js", "js/invoice-blocks.js", "js/inventory-blocks.js", "js/membership-blocks.js", "js/record-form.mjs", ), } class KernelDeniedError(Exception): def __init__(self, status: int, message: str): self.status = status super().__init__(message) class WorkflowRejectedError(Exception): def __init__(self, entity_key: str, frm: str, to: str): self.entity_key = entity_key self.frm = frm self.to = to super().__init__(f"workflow.transition rejected: {entity_key} {frm}→{to}") class BookingConflictError(Exception): def __init__(self, resource_key: str, starts_at: str): self.resource_key = resource_key self.starts_at = starts_at super().__init__(f"booking.reserve conflict: {resource_key} @ {starts_at}") @dataclass class Surface: block: str slot: str contract: str | None implementation: str must: bool route: str auth_class: str composed_into: str | None = None entity_key: str | None = None @dataclass class LedgerRow: archetype: str starter: str action: str generated_caller: str declared_capability: str runtime_authority: str persisted: bool read_back: bool result: str record_id: str | None = None expected: str | None = None observed: str | None = None @dataclass class CellResult: archetype: str starter: str intent_id: str planner_result: str plan_satisfiable: bool planned_blocks: list[str] planned_must_surfaces: int generated_surfaces: int mounted_must_surfaces: int reachable_must_surfaces: int missing_must_surfaces: list[str] build_result: str auth_result: str primary_workflow_result: str runtime_authority_result: str persistence_result: str readback_result: str private_substitute_count: int unexpected_semantic_surfaces: list[str] planner_intent_match: str generated_product_conformance: str result: str findings: list[str] = field(default_factory=list) ledger: list[LedgerRow] = field(default_factory=list) plan_digest: str = "" generated_project: str = "" def as_matrix_row(self) -> dict: return { "archetype": self.archetype, "starter": self.starter, "intent_id": self.intent_id, "planner_result": self.planner_result, "plan_satisfiable": self.plan_satisfiable, "planned_blocks": self.planned_blocks, "planned_must_surfaces": self.planned_must_surfaces, "generated_surfaces": self.generated_surfaces, "mounted_must_surfaces": self.mounted_must_surfaces, "reachable_must_surfaces": self.reachable_must_surfaces, "missing_must_surfaces": self.missing_must_surfaces, "build_result": self.build_result, "auth_result": self.auth_result, "primary_workflow_result": self.primary_workflow_result, "runtime_authority_result": self.runtime_authority_result, "persistence_result": self.persistence_result, "readback_result": self.readback_result, "private_substitute_count": self.private_substitute_count, "unexpected_semantic_surfaces": self.unexpected_semantic_surfaces, "planner_intent_match": self.planner_intent_match, "generated_product_conformance": self.generated_product_conformance, "result": self.result, "plan_digest": self.plan_digest, "findings": self.findings, } def _spec( *, purpose: str, goal: str, workflows: list[str], entities: list[str], staff: bool, auth: bool, archetypes: list[str] | None = None, ) -> DiscoverySpec: actors = [Actor(id="visitor", name="Website visitor")] if staff: actors.append(Actor(id="staff", name="Staff admin")) return DiscoverySpec( product=ProductDefinition( name="Certification product", name_status="working", purpose=purpose, first_build_goal=goal, ), archetypes=archetypes or ["operations"], actors=actors, workflows=[ Workflow(id=f"wf_{i}", name=name, description=name) for i, name in enumerate(workflows) ], entities=[ Entity(id=f"ent_{i}", name=name, description=name) for i, name in enumerate(entities) ], tenancy=TenancySpec(model="single_workspace"), authentication=AuthenticationSpec(required=auth), ) def natural_intent(archetype: str) -> DiscoverySpec: specs = { "A": _spec( purpose=( "We need a sales CRM so the team can capture leads, manage customers, " "move leads through stages, assign responsibility, track follow-up tasks, " "and inspect activity history." ), goal="Run day-to-day sales operations from one staff workspace.", workflows=["Sales operations", "Follow up with leads"], entities=["Lead", "Customer", "Task"], staff=True, auth=True, ), "B": _spec( purpose=( "Internal case management for staff: assign work, track tasks, " "update statuses, and keep activity history." ), goal="Staff inspect and progress cases.", workflows=["Case intake", "Resolve cases"], entities=["Case"], staff=True, auth=True, ), "C": _spec( purpose=( "A professional-services website where visitors request a quote " "and staff review those enquiries." ), goal="Public quote capture with staff follow-up.", workflows=["Request a quote", "Review quote requests"], entities=["Quote request"], staff=True, auth=False, archetypes=["marketing_site"], ), "D": _spec( purpose=( "Staff document management with file attachments on those records. " "Documents are structured records, not a file cabinet." ), goal="Create and edit documents and attach existing files.", workflows=["Manage documents", "Attach files to records"], entities=["Document", "File attachment"], staff=True, auth=True, ), "E": _spec( purpose="Clients book an appointment with our practice from the website.", goal="Reserve an appointment slot.", workflows=["Book an appointment"], entities=["Appointment"], staff=True, auth=True, ), "F": _spec( purpose=( "Staff product catalog of the services we offer. This is not a shop, " "cart, or checkout." ), goal="Create and edit catalog items.", workflows=["Manage catalog items"], entities=["Catalog item"], staff=True, auth=True, ), } return specs[archetype] def certification_fixture(archetype: str) -> DiscoverySpec: """Matcher-legal DiscoverySpec used for the downstream generated-product chain.""" specs = { "A": _spec( purpose=( "Internal CRM: manage customers, sales pipeline, manage leads, " "lead capture contact form, assign leads to staff, task management, " "and activity timeline." ), goal="Lead pipeline board with customer list and staff assignment.", workflows=["Lead pipeline", "Lead pipeline board", "Customer directory"], entities=["Lead", "Customer", "Assignment", "Task"], staff=True, auth=True, ), "B": _spec( purpose=( "Internal case management with staff assignment, task management, " "and activity timeline." ), goal="Case tracking with assignment tracking.", workflows=["Case tracking", "Assign cases to staff"], entities=["Case", "Assignment", "Task"], staff=True, auth=True, ), "C": _spec( purpose="Roofing company website where visitors request a quote.", goal="Capture quote requests and let staff review them.", workflows=["Request a quote", "Review quote requests"], entities=["Quote request"], staff=True, auth=True, archetypes=["marketing_site"], ), "D": _spec( purpose="Staff document records with file attachments on records.", goal="Document management and attach files to records.", workflows=["Document management", "Attach files to records"], entities=["Document"], staff=True, auth=True, ), "E": _spec( purpose="Book an appointment with staff using appointment booking.", goal="Book an appointment.", workflows=["Book an appointment"], entities=["Appointment"], staff=True, auth=True, ), "F": _spec( purpose="Staff product catalog of catalog items. No checkout.", goal="Manage catalog items.", workflows=["Manage catalog items"], entities=["Catalog item"], staff=True, auth=True, ), } return specs[archetype] def compile_cert_plan(spec: DiscoverySpec) -> AppBlockPlan: plan = compile_install_plan(spec) return apply_adoption( plan, adoption_for(source_kind=PROJECT_SOURCE_KIND_GENERATED, completed_codegen=False), ) def plan_digest(plan: AppBlockPlan) -> str: payload = json.dumps(plan.model_dump(mode="json"), sort_keys=True, default=str) return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] def _block_map(plan: AppBlockPlan) -> dict[str, PlannedBlock]: return {block.id: block for block in plan.blocks} def _starter_hosts(starter: str) -> dict[tuple[str, str], dict]: path = STARTER_ROOTS[starter] / "p2hs" / "app-block-hosts.json" payload = json.loads(path.read_text(encoding="utf-8")) return { (str(item["block"]), str(item["slot"])): item for item in payload.get("hosts") or [] if item.get("implementation") } def _slug(block_id: str) -> str: return block_id.replace(".", "-") def _impl_parts(implementation: str) -> tuple[str, str]: path, _, export = implementation.partition("#") return path, export def _host_token(starter: str, implementation: str) -> str: rel, export = _impl_parts(implementation) if export: return export return Path(rel).stem def _auth_class(block_id: str, slot_id: str, plan: AppBlockPlan) -> str: ids = {item.id for item in plan.blocks} if slot_id == "public_form": return "public" if block_id == "auth.sign_in": return "public" if block_id == "booking.appointment" and slot_id == "book": return "authenticated" if "auth.sign_in" in ids else "public" if block_id == "payments.checkout" and slot_id in {"create", "return", "cancel"}: return "authenticated" if "auth.sign_in" in ids else "public" if slot_id in {"success", "empty"} and block_id.startswith("forms."): return "public" if slot_id in {"success", "empty"} and block_id == "booking.appointment": return "authenticated" if "auth.sign_in" in ids else "public" return "staff-only" def _composed_parent(block: PlannedBlock, slot_id: str, ui_slots: set[str]) -> str | None: if slot_id == "empty": for candidate in ("pipeline", "list", "table", "book", "public_form"): if candidate in ui_slots: return candidate if slot_id == "success": for candidate in ("public_form", "book"): if candidate in ui_slots: return candidate if slot_id == "status" and "detail" in ui_slots: return "detail" return None def _route_for(block: PlannedBlock, slot_id: str, composed_into: str | None) -> str: if composed_into: return _route_for(block, composed_into, None) declared = block.slots.get("route") if slot_id in {"public_form", "book", "create"} and declared: return str(declared) if block.id == "payments.checkout" and slot_id == "return": return "/checkout/return" if block.id == "payments.checkout" and slot_id == "cancel": return "/checkout/cancel" if block.id == "auth.sign_in" and slot_id == "form": return str(declared or "/login") if block.id == "admin.shell" and slot_id == "layout": return "/admin" if slot_id in {"detail", "edit", "status", "admin_detail"}: return f"/admin/{_slug(block.id)}/{slot_id}/:id" return f"/admin/{_slug(block.id)}/{slot_id}" def planned_surfaces(plan: AppBlockPlan, starter: str) -> list[Surface]: registry = load_default_registry() hosts = _starter_hosts(starter) rows: list[Surface] = [] for block in plan.blocks: manifest = registry.require(block.id) ui_slots = {slot.id for slot in manifest.slots if slot.type == "ui_slot"} hosted_slots = { slot_id for slot_id in ui_slots if str((hosts.get((block.id, slot_id)) or {}).get("implementation") or "") } for slot in manifest.slots: if slot.type != "ui_slot": continue host = hosts.get((block.id, slot.id)) impl = str((host or {}).get("implementation") or "") if not impl: continue composed = _composed_parent(block, slot.id, hosted_slots) rows.append( Surface( block=block.id, slot=slot.id, contract=slot.component_contract, implementation=impl, must=True, route=_route_for(block, slot.id, composed), auth_class=_auth_class(block.id, slot.id, plan), composed_into=composed, entity_key=ENTITY_BY_BLOCK.get(block.id) or FORM_RUNTIME_BY_BLOCK.get(block.id), ) ) return rows def optional_requires_not_installed(plan: AppBlockPlan) -> bool: installed = {block.id for block in plan.blocks} for block in plan.blocks: for item in block.optional_requirements: if item.kind == "app_block" and item.id in installed and item.status != "satisfied": return False if item.kind == "app_block" and item.status == "satisfied" and item.id not in installed: continue return True class PlatformKernel: """In-process Cloud/BFF stand-in. Generated hosts call the same verbs.""" def __init__(self, namespace: str): self.namespace = namespace self.records: dict[tuple[str, str], dict] = {} self.reservations: dict[tuple[str, str], dict] = {} self.role: str | None = None self.admitted_files: set[str] = {f"file_{namespace}"} def as_staff(self) -> None: self.role = "staff" def as_anonymous(self) -> None: self.role = None def _require_staff(self) -> None: if self.role != "staff": raise KernelDeniedError(401, "staff authentication required") def submit_generated_form(self, *, form_key: str, values: dict) -> dict: if form_key not in FORM_RUNTIME_BY_BLOCK.values(): raise KernelDeniedError(400, f"undeclared form key {form_key}") record_id = str(uuid4()) data = {**values, "status": "new"} self.records[(form_key, record_id)] = {"id": record_id, "data": data} return {"id": record_id, **data} def list_records(self, entity_key: str) -> list[dict]: self._require_staff() return [ {"id": rec["id"], **rec["data"]} for (key, _rid), rec in self.records.items() if key == entity_key ] def get_record(self, entity_key: str, record_id: str) -> dict: self._require_staff() rec = self.records[(entity_key, record_id)] return {"id": rec["id"], **rec["data"]} def create_record(self, entity_key: str, data: dict) -> dict: self._require_staff() if entity_key in {"appointment", "booking"}: raise KernelDeniedError(400, "booking must use booking.reserve") forbidden = {"storage_key", "object_key", "bucket", "bucket_name"} if forbidden & set(data): raise KernelDeniedError(400, "client_authored_backing") if entity_key == "file_attachment": file_id = str(data.get("file_id") or "") if file_id not in self.admitted_files: raise KernelDeniedError(400, "admitted file_id required") record_id = str(uuid4()) payload = dict(data) if entity_key in WORKFLOWS and "status" not in payload: defaults = { "lead": "new", "case_record": "open", "quote_request": "new", "platform_task": "todo", "assignment": "active", "document": "draft", "catalog_item": "active", } if entity_key in defaults: payload["status"] = defaults[entity_key] self.records[(entity_key, record_id)] = {"id": record_id, "data": payload} return {"id": record_id, **payload} def patch_record(self, entity_key: str, record_id: str, data: dict) -> dict: self._require_staff() rec = self.records[(entity_key, record_id)] workflow = WORKFLOWS.get(entity_key) if workflow and workflow["field"] in data: field = workflow["field"] frm = str(rec["data"].get(field) or "") to = str(data[field]) if to != frm: allowed = workflow["transitions"].get(frm) or [] if to not in allowed: raise WorkflowRejectedError(entity_key, frm, to) rec["data"].update(data) return {"id": rec["id"], **rec["data"]} def reserve( self, *, resource_key: str, starts_at: str, customer_name: str, customer_email: str, ) -> dict: if self.role is None: raise KernelDeniedError(401, "booking.reserve requires authenticated staff") key = (resource_key, starts_at) if key in self.reservations: raise BookingConflictError(resource_key, starts_at) reservation_id = str(uuid4()) row = { "id": reservation_id, "resource_key": resource_key, "starts_at": starts_at, "customer_name": customer_name, "customer_email": customer_email, "status": "reserved", } self.reservations[key] = row return row def get_reservation(self, reservation_id: str) -> dict: if self.role is None: raise KernelDeniedError(401, "staff authentication required") for row in self.reservations.values(): if row["id"] == reservation_id: return row raise KeyError(reservation_id) def _copy_starter_assets(starter: str, dest: Path) -> None: src_root = STARTER_ROOTS[starter] for rel in _STARTER_COPY[starter]: src = src_root / rel if not src.exists(): continue target = dest / rel if src.is_dir(): shutil.copytree(src, target, dirs_exist_ok=True) else: target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, target) hosts_src = src_root / "p2hs" / "app-block-hosts.json" if hosts_src.is_file(): target = dest / "p2hs" / "app-block-hosts.json" target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(hosts_src, target) def _write(dest: Path, rel: str, content: str) -> None: path = dest / rel path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8", newline="\n") def _page_rel(starter: str, route: str) -> str: path = route.strip("/") if ":id" in path: path = path.replace(":id", "[id]" if starter != "html" else "id") if starter == "nuxt_app": if path == "admin": return "pages/admin/index.vue" return f"pages/{path}.vue" if starter == "react_app": name = path.replace("/", "_").replace("[id]", "Id").replace(":", "") or "Home" return f"src/pages/generated/{name}.tsx" if path == "admin": return "admin.html" return f"{path}.html" def _nuxt_import(implementation: str) -> tuple[str, str]: rel, _export = _impl_parts(implementation) token = Path(rel).stem return token, f"~/{rel}" def _react_import(implementation: str) -> tuple[str, str]: rel, export = _impl_parts(implementation) token = export or Path(rel).stem start = Path("src/pages/generated") target = Path(rel) relative = Path(os_relpath(start, target)).as_posix() if not relative.startswith("."): relative = f"./{relative}" for suffix in (".jsx", ".tsx", ".js"): if relative.endswith(suffix): relative = relative[: -len(suffix)] break return token, relative def _html_mount(implementation: str, dest_rel: str) -> tuple[str, str]: rel, export = _impl_parts(implementation) depth = dest_rel.count("/") prefix = "../" * depth if depth else "./" return export or Path(rel).stem, f"{prefix}{rel}" def _emit_nuxt_page(surface: Surface, extra: list[Surface], staff_gate: bool) -> str: imports = [surface, *extra] lines = ["") lines.append("") return "\n".join(lines) + "\n" def _emit_react_page(surface: Surface, extra: list[Surface], staff_gate: bool) -> str: imports = [surface, *extra] lines = ['import { useParams } from "react-router";'] seen: set[str] = set() for item in imports: token, from_path = _react_import(item.implementation) key = f"{token}:{from_path}" if key in seen or not item.implementation: continue seen.add(key) rel, export_name = _impl_parts(item.implementation) is_default = Path(rel).suffix in {".tsx", ".jsx"} and not export_name if is_default: lines.append(f'import {token} from "{from_path}";') else: lines.append(f'import {{ {token} }} from "{from_path}";') if staff_gate: lines.append('import { SignInForm } from "../../components/starter/blocks";') lines.append("export default function GeneratedPage() {") lines.append(" const params = useParams();") lines.append(" const recordId = String(params.id || '');") lines.append(" return (") lines.append("
") if staff_gate: lines.append(" ") for item in imports: rel, export_name = _impl_parts(item.implementation) token = export_name or Path(rel).stem if item.slot in {"detail", "edit", "status", "admin_detail"} or ":id" in item.route: lines.append(f" <{token} recordId={{recordId}} />") else: lines.append(f" <{token} />") lines.append("
") lines.append(" );") lines.append("}") lines.append("") return "\n".join(lines) + "\n" def os_relpath(from_dir: Path, to_file: Path) -> Path: import os return Path(os.path.relpath(to_file, from_dir)) def _emit_html_page(surface: Surface, extra: list[Surface], dest_rel: str, staff_gate: bool) -> str: mounts = [] scripts = [] for item in [surface, *extra]: export, module = _html_mount(item.implementation, dest_rel) scripts.append((export, module)) mounts.append(export) import_lines = [] names = [] for export, module in scripts: if export not in names: import_lines.append(f'import {{ {export} }} from "{module}";') names.append(export) if staff_gate: _exp, module = _html_mount( "js/starter-blocks.js#mountSignInForm", dest_rel ) import_lines.append(f'import {{ mountSignInForm }} from "{module}";') names.append("mountSignInForm") body = "\n".join(import_lines) calls = [] if staff_gate: calls.append("mountSignInForm(document.getElementById('auth'));") for export in mounts: calls.append(f"{export}(document.getElementById('root'));") return ( "\n\n" "
\n" f"\n" "\n" ) def _react_app_tsx(pages: list[tuple[str, str]]) -> str: imports = [] routes = [] for i, (route, rel) in enumerate(pages): name = f"P{i}" module = "./" + Path(rel).relative_to("src").as_posix() if module.endswith(".tsx"): module = module[:-4] imports.append(f'import {name} from "{module}";') path = route.replace(":id", ":id") routes.append(f' }} />') return ( 'import { Routes, Route } from "react-router";\n' + "\n".join(imports) + "\n\nfunction App() {\n return (\n
\n \n" + "\n".join(routes) + "\n \n
\n );\n}\n\nexport default App;\n" ) def _copy_implementation(starter: str, dest: Path, implementation: str) -> None: rel, _export = _impl_parts(implementation) if not rel: return src = STARTER_ROOTS[starter] / rel target = dest / rel if src.is_file() and not target.is_file(): target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, target) def materialize_generated_product( dest: str | Path, *, plan: AppBlockPlan, starter: str ) -> list[Surface]: dest = Path(dest) dest.mkdir(parents=True, exist_ok=True) assert_install_plan_installable(plan) surfaces = planned_surfaces(plan, starter) files = dict(app_sdk_files("html" if starter == "html" else starter)) for rel, content in files.items(): _write(dest, rel, content) _copy_starter_assets(starter, dest) for surface in surfaces: _copy_implementation(starter, dest, surface.implementation) plan_json = json.dumps(plan.model_dump(mode="json"), indent=2) + "\n" _write(dest, "p2hs/app-blocks-plan.json", plan_json) emitted_pages: dict[str, str] = {} staff_ids = {item.id for item in plan.blocks} staff_gate = "auth.sign_in" in staff_ids for surface in surfaces: if surface.composed_into: continue extras = [ item for item in surfaces if item.composed_into == surface.slot and item.block == surface.block ] route_key = surface.route rel = _page_rel(starter, surface.route.replace(":id", "[id]")) impl_rel = _impl_parts(surface.implementation)[0] gate = staff_gate and surface.auth_class in {"staff-only", "authenticated"} if impl_rel.replace("\\", "/") == rel: _copy_implementation(starter, dest, surface.implementation) elif starter == "nuxt_app": _write(dest, rel, _emit_nuxt_page(surface, extras, gate)) elif starter == "react_app": _write(dest, rel, _emit_react_page(surface, extras, gate)) else: _write(dest, rel, _emit_html_page(surface, extras, rel, gate)) emitted_pages[route_key] = rel if starter == "react_app": _write( dest, "src/App.tsx", _react_app_tsx([(route, rel) for route, rel in emitted_pages.items()]), ) _write( dest, "p2hs/generated-routes.json", json.dumps(emitted_pages, indent=2) + "\n", ) ensure_required_packs_from_plan(str(dest)) return surfaces def inspect_must_mount(root: Path, surfaces: list[Surface], starter: str) -> dict: routes = json.loads((root / "p2hs" / "generated-routes.json").read_text(encoding="utf-8")) missing = [] generated = 0 mounted = 0 reachable = 0 for surface in surfaces: rel, _export = _impl_parts(surface.implementation) host_path = root / rel token = _host_token(starter, surface.implementation) generated_ok = host_path.is_file() if generated_ok: generated += 1 parent_route = surface.route page_rel = routes.get(parent_route) if not page_rel: page_rel = routes.get(parent_route.replace(":id", "[id]")) if surface.composed_into: parent = next( ( item for item in surfaces if item.block == surface.block and item.slot == surface.composed_into ), None, ) page_rel = routes.get(parent.route) if parent else None page_ok = bool(page_rel and (root / page_rel).is_file()) page_text = (root / page_rel).read_text(encoding="utf-8") if page_ok else "" mounted_ok = page_ok and token in page_text else: page_ok = bool(page_rel and (root / page_rel).is_file()) if not page_ok: mounted_ok = False else: page_text = (root / page_rel).read_text(encoding="utf-8") impl_rel = rel.replace("\\", "/") if impl_rel == page_rel or surface.slot == "layout": mounted_ok = True else: mounted_ok = token in page_text reachable_ok = page_ok if mounted_ok: mounted += 1 if reachable_ok: reachable += 1 if not (generated_ok and mounted_ok and reachable_ok): missing.append(f"{surface.block}#{surface.slot}") return { "planned_must_surfaces": len(surfaces), "generated_must_surfaces": generated, "mounted_must_surfaces": mounted, "reachable_must_surfaces": reachable, "missing_must_surfaces": missing, } def _product_source_blob(root: Path) -> str: """Pages + planned hosts only. SDK reject-lists mention bucket/object_key.""" parts: list[str] = [] for rel in ("pages", "src/pages", "src/App.tsx"): path = root / rel if path.is_file(): parts.append(path.read_text(encoding="utf-8", errors="ignore")) elif path.is_dir(): for item in path.rglob("*"): page_like = {".vue", ".tsx", ".jsx", ".js", ".html"} if item.is_file() and item.suffix.lower() in page_like: parts.append(item.read_text(encoding="utf-8", errors="ignore")) for path in root.glob("*.html"): parts.append(path.read_text(encoding="utf-8", errors="ignore")) admin = root / "admin" if admin.is_dir(): for item in admin.rglob("*.html"): parts.append(item.read_text(encoding="utf-8", errors="ignore")) return "\n".join(parts) def scan_forbidden(root: Path, plan: AppBlockPlan) -> tuple[int, list[str]]: blob = _product_source_blob(root).lower() hits = [marker for marker in FORBIDDEN_MARKERS if marker.lower() in blob] ids = {block.id for block in plan.blocks} extras = [] routes = json.loads((root / "p2hs" / "generated-routes.json").read_text(encoding="utf-8")) if "catalog.product" not in ids: extras.extend(route for route in routes if "catalog" in route) if "documents" not in ids: extras.extend(route for route in routes if "document" in route) extras.extend(marker for marker in COMMERCE_MARKERS if marker.lower() in blob) return len(hits), extras def _host_text(root: Path, surface: Surface) -> str: rel, _ = _impl_parts(surface.implementation) path = root / rel return path.read_text(encoding="utf-8") if path.is_file() else "" def _authority_scan(root: Path, plan: AppBlockPlan, surfaces: list[Surface]) -> list[str]: failures = [] ids = {block.id for block in plan.blocks} if "forms.quote_request" in ids or "forms.contact" in ids: host_blob = "\n".join( _host_text(root, item) for item in surfaces if item.slot == "public_form" ) if "submitGeneratedForm" not in host_blob: failures.append("H-C public form missing submitGeneratedForm") if "submitContact" in host_blob or "/api/contact" in host_blob: failures.append("H-C generated public form host uses /api/contact") if "booking.appointment" in ids: book_hosts = [ item for item in surfaces if item.block == "booking.appointment" and item.slot == "book" ] for item in book_hosts: text = _host_text(root, item) if "/api/bookings/reserve" not in text: failures.append("H-C booking missing /api/bookings/reserve") if re.search(r"createRecord\s*\(", text): failures.append("H-C booking host uses createRecord") if "payments.checkout" in ids: create_hosts = [ item for item in surfaces if item.block == "payments.checkout" and item.slot == "create" ] for item in create_hosts: text = _host_text(root, item) if "/api/jobs/payment.checkout.create" not in text: failures.append("H-C checkout missing /api/jobs/payment.checkout.create") if "cloudgate" in text.lower() or "createRecord" in text: failures.append("H-C checkout host leaves the relative job BFF") for item in surfaces: if item.block == "payments.checkout" and item.slot == "catalog": text = _host_text(root, item) if "/api/jobs/payment.catalog.upsert" not in text: failures.append("H-C checkout catalog missing /api/jobs/payment.catalog.upsert") if "sk_live" in text or "sk_test" in text: failures.append("H-C checkout catalog embeds a Stripe secret") for item in surfaces: if item.block != "payments.checkout" or item.slot not in {"return", "cancel"}: continue text = _host_text(root, item).lower() if re.search(r"\bpaid\b", text) or "payment successful" in text: failures.append("H-C checkout return/cancel implies payment success") return failures def _seed_defaults(kernel: PlatformKernel, entity_key: str, **data: str) -> dict: kernel.as_staff() return kernel.create_record(entity_key, data) def exercise_primary_workflow( archetype: str, plan: AppBlockPlan, kernel: PlatformKernel, surfaces: list[Surface] ) -> tuple[str, list[LedgerRow], list[str]]: ids = {block.id for block in plan.blocks} rows: list[LedgerRow] = [] findings: list[str] = [] starter = "generated" def add(**kwargs: object) -> None: rows.append(LedgerRow(archetype=archetype, starter=starter, **kwargs)) # type: ignore[arg-type] try: if "forms.quote_request" in ids: kernel.as_anonymous() created = kernel.submit_generated_form( form_key="quote_request", values={"name": "Ada", "email": "ada@example.com", "details": "Roof"}, ) add( action="submit quote", generated_caller="public_form host", declared_capability="submitGeneratedForm", runtime_authority="form.public_create", persisted=True, read_back=True, result="pass", record_id=created["id"], ) try: kernel.list_records("quote_request") findings.append("H-E anonymous staff list succeeded") except KernelDeniedError: pass kernel.as_staff() listed = kernel.list_records("quote_request") got = kernel.get_record("quote_request", created["id"]) if got["status"] != "new": findings.append("H-D quote status not new after submit") patched = kernel.patch_record( "quote_request", created["id"], {"status": "reviewing"} ) read = kernel.get_record("quote_request", created["id"]) ok = patched["status"] == "reviewing" == read["status"] add( action="change quote status", generated_caller="admin_detail / status editor", declared_capability="patchRecord", runtime_authority="workflow.transition", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=created["id"], expected="reviewing", observed=read["status"], ) if not listed: findings.append("H-D staff list empty after public submit") if "forms.contact" in ids: kernel.as_anonymous() created = kernel.submit_generated_form( form_key="contact", values={"name": "Ada", "email": "ada@example.com", "message": "Hello"}, ) add( action="submit contact", generated_caller="public_form host", declared_capability="submitGeneratedForm", runtime_authority="form.public_create", persisted=True, read_back=True, result="pass", record_id=created["id"], ) if "crm.pipeline" in ids or "crm.lead" in ids: lead = _seed_defaults(kernel, "lead", name="Pat", email="pat@example.com") kernel.as_anonymous() try: kernel.patch_record("lead", lead["id"], {"status": "contacted"}) findings.append("H-E anonymous pipeline mutation succeeded") except KernelDeniedError: pass kernel.as_staff() patched = kernel.patch_record("lead", lead["id"], {"status": "contacted"}) read = kernel.get_record("lead", lead["id"]) ok = patched["status"] == "contacted" == read["status"] add( action="change lead stage", generated_caller="PipelineStatusEdit / RecordEdit", declared_capability="patchRecord", runtime_authority="workflow.transition", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=lead["id"], expected="contacted", observed=read["status"], ) if "crm.customer" in ids: created = _seed_defaults(kernel, "customer", name="Acme") patched = kernel.patch_record("customer", created["id"], {"name": "Acme Ltd"}) read = kernel.get_record("customer", created["id"]) ok = read["name"] == "Acme Ltd" add( action="edit customer", generated_caller="CustomerEdit / RecordEdit", declared_capability="patchRecord", runtime_authority="adapter.records", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=created["id"], ) if "case.management" in ids: case = _seed_defaults(kernel, "case_record", title="Claim 1") kernel.as_anonymous() try: kernel.patch_record("case_record", case["id"], {"status": "in_review"}) findings.append("H-E anonymous case mutation succeeded") except KernelDeniedError: pass kernel.as_staff() patched = kernel.patch_record("case_record", case["id"], {"status": "in_review"}) read = kernel.get_record("case_record", case["id"]) ok = read["status"] == "in_review" add( action="change case status", generated_caller="CaseStatusEdit / RecordEdit", declared_capability="patchRecord", runtime_authority="workflow.transition", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=case["id"], expected="in_review", observed=read["status"], ) if "workflow.assignment" in ids: assignment = _seed_defaults(kernel, "assignment", title="Owner") patched = kernel.patch_record("assignment", assignment["id"], {"status": "completed"}) read = kernel.get_record("assignment", assignment["id"]) ok = read["status"] == "completed" add( action="complete assignment", generated_caller="AssignmentStatusEdit", declared_capability="patchRecord", runtime_authority="workflow.transition", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=assignment["id"], ) if "task.management" in ids: task = _seed_defaults(kernel, "platform_task", title="Call back") patched = kernel.patch_record("platform_task", task["id"], {"status": "in_progress"}) read = kernel.get_record("platform_task", task["id"]) ok = read["status"] == "in_progress" add( action="change task status", generated_caller="TaskStatusEdit", declared_capability="patchRecord", runtime_authority="workflow.transition", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=task["id"], ) if "activity.timeline" in ids: kernel.as_staff() listed = kernel.list_records("activity_entry") add( action="inspect activity", generated_caller="ActivityTable", declared_capability="listRecords", runtime_authority="adapter.records", persisted=True, read_back=True, result="pass", ) if "documents" in ids: created = _seed_defaults( kernel, "document", title="Policy", content="Hello", content_format="plain_text" ) patched = kernel.patch_record("document", created["id"], {"title": "Policy v2"}) read = kernel.get_record("document", created["id"]) ok = read["title"] == "Policy v2" add( action="create/edit document", generated_caller="DocumentsCreate / DocumentsEdit", declared_capability="createRecord+patchRecord", runtime_authority="adapter.records", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=created["id"], ) if "files.attachments" in ids: attached = kernel.create_record( "file_attachment", { "file_id": f"file_{kernel.namespace}", "subject_entity_key": "document", "subject_record_id": created["id"], }, ) listed = kernel.list_records("file_attachment") visible = any(item["id"] == attached["id"] for item in listed) add( action="attach file_id", generated_caller="file_attachment createRecord", declared_capability="createRecord", runtime_authority="file_attachment", persisted=visible, read_back=visible, result="pass" if visible else "fail", record_id=attached["id"], ) if "booking.appointment" in ids: kernel.as_anonymous() try: kernel.reserve( resource_key="room-1", starts_at="2026-09-16T10:00:00Z", customer_name="Ada", customer_email="ada@example.com", ) findings.append("H-E anonymous booking.reserve succeeded") except KernelDeniedError: pass kernel.as_staff() reserved = kernel.reserve( resource_key="room-1", starts_at="2026-09-16T10:00:00Z", customer_name="Ada", customer_email="ada@example.com", ) read = kernel.get_reservation(reserved["id"]) add( action="reserve", generated_caller="BookingForm", declared_capability="booking.reserve", runtime_authority="booking BFF /api/bookings/reserve", persisted=read["id"] == reserved["id"], read_back=True, result="pass", record_id=reserved["id"], ) try: kernel.reserve( resource_key="room-1", starts_at="2026-09-16T10:00:00Z", customer_name="Bo", customer_email="bo@example.com", ) findings.append("H-D booking conflict accepted") except BookingConflictError: add( action="reserve conflict", generated_caller="BookingForm", declared_capability="booking.reserve", runtime_authority="booking BFF", persisted=True, read_back=True, result="pass", ) if "catalog.product" in ids: created = _seed_defaults(kernel, "catalog_item", name="Hourly", sku="H1") patched = kernel.patch_record("catalog_item", created["id"], {"name": "Daily"}) read = kernel.get_record("catalog_item", created["id"]) ok = read["name"] == "Daily" add( action="create/edit catalog_item", generated_caller="CatalogCreate / CatalogEdit", declared_capability="createRecord+patchRecord", runtime_authority="adapter.records", persisted=ok, read_back=ok, result="pass" if ok else "fail", record_id=created["id"], ) except Exception as exc: # noqa: BLE001 — certification records the failure findings.append(f"H-D workflow exception: {exc}") return "fail", rows, findings failed = [row for row in rows if row.result != "pass"] return ("fail" if failed or findings else "pass", rows, findings) def classify_planner_match(archetype: str, plan: AppBlockPlan) -> str: ids = {block.id for block in plan.blocks} required = PRIMARY_BLOCKS[archetype] missing = [block_id for block_id in required if block_id not in ids] return "pass" if not missing else "fail_p2_matcher" def certify_cell( archetype: str, starter: str, dest: str | Path ) -> CellResult: dest = Path(dest) natural_plan = compile_cert_plan(natural_intent(archetype)) planner_match = classify_planner_match(archetype, natural_plan) spec = certification_fixture(archetype) plan = compile_cert_plan(spec) satisfiable = not plan.unsatisfiable if satisfiable: assert_install_plan_installable(plan) surfaces = [] if satisfiable: surfaces = materialize_generated_product(dest, plan=plan, starter=starter) mount = ( inspect_must_mount(dest, surfaces, starter) if surfaces else { "planned_must_surfaces": 0, "generated_must_surfaces": 0, "mounted_must_surfaces": 0, "reachable_must_surfaces": 0, "missing_must_surfaces": ["unsatisfiable"], } ) private_count, extras = scan_forbidden(dest, plan) if surfaces else (0, []) authority_failures = _authority_scan(dest, plan, surfaces) if surfaces else ["no surfaces"] kernel = PlatformKernel(namespace=f"{archetype}-{starter}") workflow, ledger, wf_findings = ( exercise_primary_workflow(archetype, plan, kernel, surfaces) if surfaces else ("fail", [], ["unsatisfiable"]) ) findings = list(authority_failures) + wf_findings if mount["missing_must_surfaces"]: findings.append("H-B missing MUST surfaces: " + ",".join(mount["missing_must_surfaces"])) if private_count: findings.append(f"H-F private substitute markers={private_count}") auth = "pass" if any(item.startswith("H-E") for item in findings): auth = "fail" persistence = "pass" if all(row.persisted for row in ledger) and workflow == "pass" else "fail" readback = "pass" if all(row.read_back for row in ledger) and workflow == "pass" else "fail" runtime = "pass" if not authority_failures and workflow == "pass" else "fail" build = "pass" if not mount["missing_must_surfaces"] else "fail" if extras: findings.append("unexpected:" + ",".join(extras)) boundary = any(item.startswith("H-E") or item.startswith("H-F") for item in findings) p0 = boundary or private_count > 0 p1 = bool( mount["missing_must_surfaces"] or authority_failures or workflow != "pass" or persistence != "pass" or readback != "pass" or runtime != "pass" or auth != "pass" or extras or not satisfiable ) if not satisfiable or p0: result = "fail_P0" if p0 else "fail_P1" elif p1: result = "fail_P1" else: result = "pass" conformance = "pass" if result == "pass" else result return CellResult( archetype=archetype, starter=starter, intent_id=f"{archetype}-cert", planner_result="pass" if satisfiable else "fail", plan_satisfiable=satisfiable, planned_blocks=[block.id for block in plan.blocks], planned_must_surfaces=mount["planned_must_surfaces"], generated_surfaces=mount["generated_must_surfaces"], mounted_must_surfaces=mount["mounted_must_surfaces"], reachable_must_surfaces=mount["reachable_must_surfaces"], missing_must_surfaces=mount["missing_must_surfaces"], build_result=build, auth_result=auth, primary_workflow_result=workflow, runtime_authority_result=runtime, persistence_result=persistence, readback_result=readback, private_substitute_count=private_count, unexpected_semantic_surfaces=extras, planner_intent_match=planner_match, generated_product_conformance=conformance, result=result, findings=findings, ledger=ledger, plan_digest=plan_digest(plan), generated_project=str(dest), ) def certify_matrix(base: str | Path) -> list[CellResult]: base = Path(base) rows = [] for archetype in ARCHETYPES: for starter in STARTERS: dest = base / f"{archetype}-{starter}" if dest.exists(): shutil.rmtree(dest) rows.append(certify_cell(archetype, starter, dest)) return rows def write_matrix_artifact(cells: list[CellResult], path: Path | None = None) -> Path: path = path or MATRIX_PATH path.parent.mkdir(parents=True, exist_ok=True) payload = { "phase": "12H", "cells": [cell.as_matrix_row() for cell in cells], "summary": { "certification_cells": len(cells), "cells_passed": sum(1 for cell in cells if cell.result == "pass"), "cells_failed_p0": sum(1 for cell in cells if cell.result == "fail_P0"), "cells_failed_p1": sum(1 for cell in cells if cell.result == "fail_P1"), "cells_blocked": sum(1 for cell in cells if cell.result == "blocked_external"), "natural_intent_planner_pass": sum( 1 for cell in cells if cell.planner_intent_match == "pass" ), "planned_must_surfaces": sum(cell.planned_must_surfaces for cell in cells), "mounted_must_surfaces": sum(cell.mounted_must_surfaces for cell in cells), "reachable_must_surfaces": sum(cell.reachable_must_surfaces for cell in cells), "missing_must_surfaces": sum(len(cell.missing_must_surfaces) for cell in cells), "private_backend_substitutes": sum(cell.private_substitute_count for cell in cells), }, } path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8", newline="\n") return path