"""Architecture Lane Workforce: admit workforce.worker over live data.entity. Freeze deliberately lifted for the Native Capability Baseline (explicit user decision, not harvest ranking -- property/lab stay guarded). No new kernel primitive, no payroll admission, no reuse of membership.member for staff. Shift assignment is a separate Cloud-native primitive (workforce.shift_assign/workforce.shift_release against cloud_shift_assignments in P2HSCloud), not a field on worker_record and not wrapped in this App Block. """ 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/workforce_freeze.receipt.txt" AUDIT = ROOT / "docs/audits/ARCHITECTURE_LANE_WORKFORCE.md" 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", } def workforce_composition(): return _captured_composition( "alembic/versions/157_workforce_directory_pack.py", "workforce-directory" ) def _ids(intent: ProductIntent) -> set[str]: return {block_id for _cap, block_id, _reason in select_capabilities(intent)} def test_workforce_runtime_is_canonical_without_new_kernel(): registry = load_default_registry() block = registry.require("workforce.worker") canonical = registry.canonical.require("workforce.worker") assert block.status == "implemented" assert block.lifecycle == "stable" assert "workforce.worker" in CANONICAL_BLOCK_IDS assert "workforce.worker" not in FEATURE_PACK_ALIASES assert "workforce.worker" not in feature_pack_only_aliases() assert canonical.installation.feature_packs == ("workforce-directory",) primitives = {dep.id for dep in canonical.requires if dep.kind == "primitive"} assert primitives == {"data.entity", "workforce.shift_assign", "workforce.shift_release"} assert "workforce-directory" in default_known_packs() composition = workforce_composition() entity = composition["entities"][0] assert entity["entity_key"] == "worker_record" keys = {field["key"] for field in entity["schema"]["fields"]} assert keys == { "name", "email", "phone", "role", "skills", "credential_name", "credential_expiry_date", "status", } status = next(field for field in entity["schema"]["fields"] if field["key"] == "status") assert status["options"] == ["active", "inactive"] assert "shift_starts_at" not in keys assert "shift_ends_at" not in keys assert "payroll_rate" not in keys def test_workforce_install_plan_does_not_bind_membership(monkeypatch): from src.domain.app_blocks import compiler registry = load_default_registry() monkeypatch.setattr( compiler, "select_capabilities", lambda _intent: [("workforce_worker", "workforce.worker", "Selected.")], ) plan = compiler.compile_install_plan(None, registry=registry) assert_install_plan_installable(plan) assert "workforce-directory" in plan.required_feature_packs assert "saas-membership" not in plan.required_feature_packs assert "payment-checkout" not in plan.required_feature_packs assert "data.entity" in plan.required_primitives assert {"workforce.shift_assign", "workforce.shift_release"} <= set(plan.required_primitives) def test_workforce_does_not_imply_membership_or_payroll(): ids = _ids( ProductIntent( purpose="Staff need a workforce directory. No membership, no payroll.", source_prompt="Staff need a workforce directory. No membership, no payroll.", staff_review=True, ) ) assert "workforce.worker" in ids assert "membership.member" not in ids assert not (ids & set(GATED)) def test_workforce_and_task_management_compose_without_merging_stores(): prompt = "Staff manage a worker directory and internal tasks for workers." ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "workforce.worker" in ids assert "task.management" in ids def test_false_positives_do_not_select_workforce(): for purpose in ( "customers pay us later", "pricing page", "Staff manage a product catalog", "employee handbook onboarding paperwork", ): ids = _ids(ProductIntent(purpose=purpose, source_prompt=purpose, staff_review=True)) assert "workforce.worker" not in ids, purpose def test_negative_intent_skips_workforce(): for prompt in ( "Client records for the firm. Do not add a worker directory.", "Staff CRM. No workforce directory.", ): ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "workforce.worker" not in ids, prompt def test_positive_intent_selects_workforce(): for prompt in ( "Staff need a worker directory with skills and credential expiry.", "Build a workforce directory to track employee credentials.", "Track worker credentials and skills for our field technicians.", ): ids = _ids(ProductIntent(purpose=prompt, source_prompt=prompt, staff_review=True)) assert "workforce.worker" in ids, prompt assert not (ids & set(GATED)), prompt def test_13a_do_not_create_orders_still_misses_workforce(): 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 "workforce.worker" not in ids def test_workforce_reuses_record_starter_contracts(): registry = load_default_registry() block = registry.require("workforce.worker") contracts = { slot.id: slot.component_contract for slot in block.slots if slot.component_contract } assert STARTER_SLOTS.items() <= contracts.items() assert "worker.list" not in known_contract_ids() def test_workforce_runtime_crud_is_staff_managed_records(): kernel = PlatformKernel(namespace="workforce-freeze") kernel.as_anonymous() with pytest.raises(KernelDeniedError): kernel.create_record("worker_record", {"name": "Ada"}) kernel.as_staff() created = kernel.create_record( "worker_record", { "name": "Ada Lovelace", "email": "ada@example.com", "role": "Technician", "skills": "electrical, HVAC", "credential_name": "HVAC Certification", "credential_expiry_date": "2027-01-01", "status": "active", }, ) listed = kernel.list_records("worker_record") assert listed[0]["id"] == created["id"] kernel.patch_record("worker_record", created["id"], {"status": "inactive"}) read = kernel.get_record("worker_record", created["id"]) assert read["status"] == "inactive" assert read["name"] == "Ada Lovelace" other = PlatformKernel(namespace="workforce-other") other.as_staff() assert other.list_records("worker_record") == [] def test_materialized_workforce_hosts_use_records_not_shift_scheduling(): 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"] == "workforce.worker"} assert {"list", "detail", "create", "edit", "empty"} <= set(rows) for slot in ("list", "detail", "create", "edit"): source = _host_fragment(root, rows[slot]["implementation"]) assert "worker_record" in source assert "shift_assign" not in source assert "cloud_shift_assignments" not in source assert "/api/workforce/shifts" not in source assert "/api/jobs/workforce." not in source for slot in ("assign", "release", "shifts", "shift_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="worker_record"' not in source assert 'entityKey="worker_record"' not in source assert 'entity_key: "worker_record"' not in source assert "/api/jobs/workforce.shift_assign" in _host_fragment(root, rows["assign"]["implementation"]) release = _host_fragment(root, rows["release"]["implementation"]) assert "/api/jobs/workforce.shift_release" in release assert "assignment_id" in release assert "/api/workforce/shifts" in _host_fragment(root, rows["shifts"]["implementation"]) detail = _host_fragment(root, rows["shift_detail"]["implementation"]) assert "/api/workforce/shifts/" in detail assert "route.params.id" in detail or "recordId" in detail or "props.id" in detail or "props.recordId" in detail def test_workforce_operational_slots_project_static_query_params(): from src.domain.app_blocks.hosts import host_map_payload from src.integrations.platform_ports import CATALOG payload = host_map_payload() rows = {h["slot"]: h for h in payload["hosts"] if h["block"] == "workforce.worker"} assert rows["shifts"]["query_source"] == "port:workforce.shifts.list" assert rows["shifts"]["query_params"] == {"status": "assigned"} assert rows["shift_detail"]["query_source"] == "port:workforce.shifts.get" assert "{{" not in str(rows["shifts"]["query_params"]) port_ids = {p["port_id"] for p in CATALOG["ports"]} assert {"workforce.shifts.list", "workforce.shifts.get"} <= port_ids for port_id in ("workforce.shifts.list", "workforce.shifts.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_shift_assign_primitive_is_cloud_native_not_a_worker_field(): from src.integrations.platform_ports import QUEUE_REQUEST_TYPES assert "workforce.shift_assign" in QUEUE_REQUEST_TYPES assert "workforce.shift_release" in QUEUE_REQUEST_TYPES from src.domain.cloud.service import _REQUEST_TYPE_POLICY assert "workforce.shift_assign" in _REQUEST_TYPE_POLICY assert "workforce.shift_release" in _REQUEST_TYPE_POLICY registry = load_default_registry() canonical = registry.canonical.require("workforce.worker") primitives = {dep.id for dep in canonical.requires if dep.kind == "primitive"} assert "workforce.shift_assign" in primitives assert "workforce.shift_release" in primitives def test_workforce_freeze_receipts_and_coverage_next_family(): receipt = RECEIPT.read_text(encoding="utf-8") audit = AUDIT.read_text(encoding="utf-8") for required in ( "workforce_admitted=yes", "canonical_app_block=workforce.worker", "canonical_feature_pack=workforce-directory", "workforce_without_payroll=yes", "family_frozen=yes", ): assert required in receipt, required assert "workforce_freeze=yes" in audit registry = yaml.safe_load(REGISTRY.read_text(encoding="utf-8")) by_id = {row["id"]: row for row in registry["capabilities"]} assert by_id["workforce.worker"]["status"] == "frozen" assert by_id["workforce"]["status"] == "frozen" assert by_id["workforce-directory"]["status"] == "frozen" def test_workforce_production_smoke_script(): from scripts.architecture_lane_workforce_smoke import run_smoke payload = run_smoke() assert payload["pass"] is True, payload["failures"] assert payload["on_disk_workforce_recipes"] == []