"""P1-4 local proof: CRM + Tasks + Notifications on one generated composition. Core pack lookup/install-state and schema-snapshot persistence are mocked. Cloud gateway/D1/JWT/records HTTP are real and local. No contact pack is installed. Notifications are created by the admitted internal caller, not by inventing a CRM→notification coupling. Run from Core: python scripts/probe_cross_family_12b.py [--generated] """ from __future__ import annotations import asyncio import base64 import json from pathlib import Path import sys import time import tomllib from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 import httpx from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding, rsa ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.domain.app_blocks.compiler import compile_install_plan from src.domain.feature_packs.crm_authorization import staff_policy from src.domain.feature_packs.service import PackInstallService from src.integrations.cloud_control_client import ( CloudControlClient, _rsa_private_jwk, _rsa_public_jwk, ) from src.domain.projects.discovery_schemas import ( Actor, AuthenticationSpec, DiscoverySpec, Entity, ProductDefinition, TenancySpec, Workflow, ) from tests.test_crm_authorization_11e2 import lead_composition from tests.test_phase11_pack_migration_materialization import _captured_composition BASE = "http://127.0.0.1:8801" PURPOSE = ( "Staff manage customer records, a sales pipeline board, " "an internal task board, and an in-app notification inbox." ) def b64(value: bytes) -> str: return base64.urlsafe_b64encode(value).rstrip(b"=").decode() def jwt(key, project, workspace, role, sub=None): now = int(time.time()) header = b64(json.dumps({"alg": "RS256", "typ": "JWT", "kid": "12b"}).encode()) body = b64(json.dumps({ "sub": sub or f"probe-{role}", "project_id": str(project), "workspace_id": str(workspace), "role": role, "email": f"{role}@example.invalid", "iat": now, "exp": now + 600, "app_metadata": {}, "user_metadata": {}, }).encode()) message = f"{header}.{body}" return f"{message}.{b64(key.sign(message.encode(), padding.PKCS1v15(), hashes.SHA256()))}" def spec() -> DiscoverySpec: return DiscoverySpec( product=ProductDefinition( name="Staff operations", name_status="working", purpose=PURPOSE, first_build_goal="Let staff work customers, pipeline stages, tasks, and their notification inbox.", ), archetypes=["internal_tool"], actors=[Actor(id="staff", name="Staff"), Actor(id="visitor", name="Website visitor")], workflows=[ Workflow(id="review", name="Review customers", description="Staff update records"), Workflow(id="tasks", name="Track staff tasks", description="Staff move tasks"), ], entities=[ Entity(id="customer", name="Customer", description="A customer record"), Entity(id="task", name="Staff task", description="An internal task"), ], tenancy=TenancySpec(model="single_workspace"), authentication=AuthenticationSpec(required=True), ) async def install(cloud, project, composition, slug, version): pack = SimpleNamespace( id=uuid4(), slug=slug, status="active", version=version, composition=composition, input_schema=None, ) service = PackInstallService(AsyncMock(), MagicMock(), cloud) service._get_pack_by_slug = AsyncMock(return_value=pack) service._installed_config = AsyncMock(return_value={}) service._upsert_install_state = AsyncMock() with patch( "src.domain.projects.cloud_schema_catalog.ProjectCloudSchemaSnapshotRepository.upsert", new_callable=AsyncMock, ): return await service.install_pack(project_id=project, pack_slug=pack.slug, config={}) async def main(): config = tomllib.loads((ROOT.parent / "P2HSCloud/apps/pc-gateway/wrangler.toml").read_text()) token = config["vars"]["INTERNAL_CONTROL_TOKEN"] cloud = CloudControlClient(api_url=BASE, token=token) workspace, project, other = uuid4(), uuid4(), uuid4() key = rsa.generate_private_key(public_exponent=65537, key_size=2048) private_pem = key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption() ).decode() public_pem = key.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo ).decode() observations = [] def check(name, actual, expected): observations.append({ "check": name, "actual": actual, "expected": expected, "pass": actual == expected, }) plan = compile_install_plan(spec()) ids = {block.id for block in plan.blocks} check("plan satisfiable", plan.unsatisfiable, False) check("crm.customer selected", "crm.customer" in ids, True) check("crm.pipeline selected", "crm.pipeline" in ids, True) check("task.management selected", "task.management" in ids, True) check("notifications.in_app selected", "notifications.in_app" in ids, True) check("accidental contact absent", "forms.contact" in ids, False) check("contact pack absent", "contact-capture" in plan.required_feature_packs, False) check( "required packs", sorted(plan.required_feature_packs), sorted(["customer-records", "lead-lifecycle", "task-management", "notifications-in-app"]), ) packs = [ ("customer-records", "1.0.0", _captured_composition(str(ROOT / "alembic/versions/144_customer_records_pack.py"), "customer-records")), ("lead-lifecycle", "1.1.0", lead_composition()), ("task-management", "1.0.0", _captured_composition(str(ROOT / "alembic/versions/137_task_management_pack.py"), "task-management")), ("notifications-in-app", "1.0.0", _captured_composition(str(ROOT / "alembic/versions/140_notifications_in_app_pack.py"), "notifications-in-app")), ] async with httpx.AsyncClient(base_url=BASE, timeout=15) as http: async def req(method, path, body=None, bearer=token): kwargs = {"headers": {"Authorization": f"Bearer {bearer}"}} if body is not None: kwargs["json"] = body return await http.request(method, path, **kwargs) async def control(path, body): (await req("PUT", "/v1/cloud/control/" + path, body)).raise_for_status() await control(f"workspaces/{workspace}", { "workspace_id": str(workspace), "name": "12B disposable cross-family probe", "status": "active", "cloud_enabled": True, "suspended": False, }) for target in (project, other): await control(f"projects/{target}", { "project_id": str(target), "workspace_id": str(workspace), "name": "12B probe", "status": "active", "cloud_enabled": True, "suspended": False, "allowed_request_types": [], "origins": [], }) await control(f"projects/{target}/auth-keys", { "project_id": str(target), "key_id": "12b", "public_key_jwk": _rsa_public_jwk(public_pem, key_id="12b"), "private_key_jwk": _rsa_private_jwk(private_pem, key_id="12b"), }) entities = [] for slug, version, composition in packs: result = await install(cloud, project, composition, slug, version) check(f"{slug} installs", result["status"], "installed") for entity in composition.get("entities") or []: entities.append(entity["entity_key"]) check("distinct entity keys", sorted(entities), ["customer", "lead", "platform_notification", "platform_task"]) customer_live = await cloud.get_project_entity(str(project), "customer") check("customer staff policy", json.loads(customer_live["entity"]["schema_json"])["access_policy"], staff_policy()) lead_live = await cloud.get_project_entity(str(project), "lead") check("lead staff policy", json.loads(lead_live["entity"]["schema_json"])["access_policy"], staff_policy()) missing = await req("GET", f"/v1/projects/{project}/entities/contact_submission") check("contact entity absent", missing.status_code, 404) admin = jwt(key, project, workspace, "admin") member = jwt(key, project, workspace, "member") customer_path = f"/v1/projects/{project}/entities/customer/records" task_path = f"/v1/projects/{project}/entities/platform_task/records" notify_path = f"/v1/projects/{project}/entities/platform_notification/records" lead_path = f"/v1/projects/{project}/entities/lead/records" response = await req("POST", customer_path, {"data": {"name": "Coexist customer", "status": "active"}}, admin) check("admin customer create", response.status_code, 201) customer_id = response.json()["record"]["id"] response = await req("POST", lead_path, {"data": {"name": "Coexist lead", "email": "lead@example.invalid", "status": "new"}}, admin) check("admin lead create", response.status_code, 201) lead_id = response.json()["record"]["id"] response = await req("POST", task_path, {"data": {"title": "Follow up", "status": "todo"}}, admin) check("admin task create", response.status_code, 201) task_id = response.json()["record"]["id"] response = await req("PATCH", f"{task_path}/{task_id}", {"data": {"status": "in_progress"}}, admin) check("admin task transition", response.status_code, 200) created = await req("POST", notify_path, { "data": { "recipient_user_id": "probe-admin", "source_event_id": "evt_12b_fixture", "canonical_event_id": "task.created", "subject_entity_key": "platform_task", "subject_record_id": task_id, "title": "Inbox fixture", "category": "created", }, "owner_user_id": "probe-admin", }, token) if created.status_code != 201: print("notification create body", created.status_code, created.text) check("internal notification create", created.status_code, 201) notify_id = created.json()["record"]["id"] if created.status_code == 201 else None listed = await req("GET", notify_path, bearer=admin) check("admin notification list", listed.status_code, 200) denied = await req("POST", notify_path, {"data": {"title": "Denied", "recipient_user_id": "x"}}, member) check("member notification create denied", denied.status_code, 403) for method, route, body in ( ("POST", customer_path, {"data": {"name": "Denied", "status": "active"}}), ("GET", customer_path, None), ("PATCH", f"{customer_path}/{customer_id}", {"data": {"status": "inactive"}}), ): response = await req(method, route, body, member) check(f"customer member {method} denied", response.status_code, 403) response = await req("GET", customer_path, bearer=jwt(key, other, workspace, "admin")) check("cross-project JWT denied", response.status_code, 401) if "--generated" in sys.argv: await control(f"projects/{project}/auth-config", { "project_id": str(project), "workspace_id": str(workspace), "providers": ["email"], "jwt_expiry_seconds": 600, "refresh_token_expiry_seconds": 3600, "allow_signup": False, "require_email_verify": False, "google_oauth_mode": "platform", "roles": [{"name": role, "is_default": role == "member", "permissions": []} for role in ("admin", "super_admin", "member")], }) users = {} for role in ("admin", "member"): response = await req("POST", f"/v1/projects/{project}/auth/users", { "email": f"{role}@example.invalid", "role": role, "email_verified": True, }) response.raise_for_status() users[role] = response.json()["id"] if notify_id: await req("PATCH", f"{notify_path}/{notify_id}", { "data": {"recipient_user_id": users["admin"]}, "owner_user_id": users["admin"], }, token) child = await asyncio.create_subprocess_exec( "node", str(ROOT / "scripts/prove-cross-family-generated-12b.mjs"), stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) output, error = await child.communicate(json.dumps({ "gateway": BASE, "project": str(project), "admin": jwt(key, project, workspace, "admin", users["admin"]), "member": jwt(key, project, workspace, "member", users["member"]), "other_project": jwt(key, other, workspace, "admin"), "lead_id": lead_id, "task_id": task_id, "notify_id": notify_id, "customer_id": customer_id, }).encode()) print(output.decode()) if child.returncode: print(error.decode()) check("generated host to BFF to gateway proof", child.returncode, 0) failed = sum(not item["pass"] for item in observations) print(json.dumps({ "phase": "12B-cross-family", "gateway": BASE, "fixture_projects": [str(project), str(other)], "core_database": "mocked lookup and install/snapshot persistence", "cloud_runtime": "real local gateway, D1, JWT, install and records HTTP", "mocked": ["Core pack catalog lookup", "install-state persistence", "schema snapshot persistence"], "real": ["local pc-gateway", "D1 entities/records", "JWT records HTTP", "HTML Starter hosts", "generated records BFF"], "accidental_contact_capture_present": False, "event_coupling": "none invented; notifications created via internal caller, not a new CRM/task route", "observations": observations, "failed": failed, "note": "Synthetic local fixtures only; no new primitive or App Block.", }, indent=2)) return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))