"""6N CRM Codegen exposure. Runtime catalog truth stays 6M.""" from __future__ import annotations from src.domain.codegen.crm_codegen import ( CRM_CANONICAL_TYPES, CRM_CODEGEN_POLICY, CRM_CONTACT_EXAMPLE, CRM_DEAL_EXAMPLE, CRM_PROMPT_FIXTURES, canonical_crm_invoke_capabilities, example_canonical_invoke, generic_snippet_leaks, select_crm_generation, supported_association_pairs, ) from src.domain.codegen.platform_backend import ( _compact_connectors, build_platform_backend_capabilities, ) from src.domain.codegen.prompt_wrapper import ( _APP_BLOCK_CONTRACT, _P2HS_CLOUD_CONTRACT, _P2HS_CORE_API_GUIDANCE, _app_recipes, ) from src.domain.connectors.catalog import slugs_implementing from src.domain.connectors.crm_events import FROZEN_CRM_EVENT_TYPES from src.domain.connectors.invoke_capabilities import ( CAPABILITY_CONVERSION_PATTERN, CRM_CODEGEN_PHASE_6N, CRM_FAMILY_PLAN, CRM_PUBLIC_INVOKE_VERBS, CRM_PUBLIC_RUNTIME_PHASE_6M, INVOKE_CAPABILITIES, PROVIDER_ACTION_WHEN, capability_surface, catalog_consistency_issues, ) from src.domain.connectors.public_gateway import PHASE_6_STATUS from src.domain.manifest.service import ( _connectors_manifest_envelope, _provider_action_escape_hatches, ) def test_6n_closes_codegen_exposure_without_runtime_truth_changes() -> None: assert PHASE_6_STATUS == "6p_complete" assert CAPABILITY_CONVERSION_PATTERN["phase_6_status"] == "6p_complete" assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == ( "9_app_block_modernization" ) assert CRM_FAMILY_PLAN["status"] == "6p_complete" assert CRM_FAMILY_PLAN["next_slice"] == "forms_first_class_product_domain" assert CRM_CODEGEN_PHASE_6N["status"] == "complete" assert CRM_CODEGEN_PHASE_6N["codegen_default"] is True assert CRM_CODEGEN_PHASE_6N["runtime_unchanged"] is True assert CRM_CODEGEN_PHASE_6N["no_parallel_crm_codegen_registry"] is True assert CRM_CODEGEN_PHASE_6N["inbound_events_unchanged"] is True assert CRM_CODEGEN_PHASE_6N["app_blocks_unchanged"] is True assert CRM_PUBLIC_RUNTIME_PHASE_6M["codegen_default"] is False assert catalog_consistency_issues() == () assert canonical_crm_invoke_capabilities() == CRM_PUBLIC_INVOKE_VERBS assert len(canonical_crm_invoke_capabilities()) == 14 def test_runtime_catalog_fields_remain_6m_truth() -> None: for verb in CRM_PUBLIC_INVOKE_VERBS: spec = INVOKE_CAPABILITIES[verb] slugs = list(slugs_implementing(verb)) surface = capability_surface(verb, slugs) assert spec["status"] == "implemented" assert spec["invocable"] is True assert spec["method"] == f"POST /api/projects/{{project_id}}/capabilities/{verb}" assert spec["runtime"] == "integration.execute" assert surface["role"] == "default" assert surface["implemented_by"] == slugs assert "not a Codegen default" not in spec["codegen"] for event_type in FROZEN_CRM_EVENT_TYPES: assert event_type not in INVOKE_CAPABILITIES assert "crm.external_sync" not in INVOKE_CAPABILITIES assert "crm.search" not in INVOKE_CAPABILITIES def test_codegen_prompts_prefer_canonical_crm() -> None: recipes = _app_recipes() assert "crm.contact.create" in recipes assert "crm.contact.upsert" in recipes assert "crm.company.create" in recipes assert "crm.deal.create" in recipes assert "crm.association.create" in recipes assert "capabilities/crm.contact.create" in recipes assert "logical aliases" in recipes assert "provider_hint" in recipes assert "Do not generate `crm.external_sync`" in recipes assert CRM_CODEGEN_POLICY.strip() in _P2HS_CLOUD_CONTRACT assert "crm.contact.*" in _P2HS_CORE_API_GUIDANCE assert "Do not choose HubSpot or Salesforce actions from the installed connector" in ( _P2HS_CORE_API_GUIDANCE ) assert "Do not rewrite that App Block onto `crm.contact.create`" in _APP_BLOCK_CONTRACT assert "forms.contact` / `contact-capture` plus optional `crm.contact.upsert`" in ( _APP_BLOCK_CONTRACT ) def test_platform_backend_rules_include_crm_default() -> None: class _Manifest: def model_dump(self, mode="json"): return {"credentials": {}, "project": {}, "backend": {}} caps = build_platform_backend_capabilities(_Manifest()) assert any("generic CRM work" in rule for rule in caps["rules"]) compact = _compact_connectors(_connectors_manifest_envelope(items={})) for verb in CRM_PUBLIC_INVOKE_VERBS: assert compact["capabilities"][verb]["role"] == "default" assert compact["capabilities"][verb]["method"].endswith(verb) def test_prompt_fixtures_map_generic_intent_to_crm_star() -> None: expected = { "Create a contact in our CRM": "crm.contact.create", "Sync form submissions to CRM by email": "crm.contact.upsert", "When someone submits the contact form, create/update them in the CRM.": ( "crm.contact.upsert" ), "Search CRM contact by email": "crm.contact.search", "Create a company": "crm.company.create", "Find company called Acme": "crm.company.search", "Create a deal": "crm.deal.create", "Move deal to qualified": "crm.deal.update", "Find qualified deals named Acme": "crm.deal.search", "Associate this contact with the company": "crm.association.create", "Link this customer with Acme Corp.": "crm.association.create", "Attach this deal to the company": "crm.association.create", "Create a CRM contact when this form is submitted.": "crm.contact.create", } for prompt, capability in expected.items(): for provider in (None, "hubspot", "salesforce"): decision = select_crm_generation(prompt, connected_provider=provider) assert decision.surface == "canonical", prompt assert decision.capability == capability assert decision.provider_action is None assert decision.invoke_path.endswith(capability) snippet = example_canonical_invoke(capability, decision.payload or {}) assert generic_snippet_leaks(snippet) == () def test_provider_specific_prompts_use_escape_hatches() -> None: soql = select_crm_generation("Run this Salesforce SOQL query SELECT Id FROM Contact") assert soql.surface == "provider_action" assert soql.provider_action == "salesforce.query_soql" label = select_crm_generation("Use HubSpot association label X") assert label.surface == "provider_action" assert label.provider_action == "hubspot.associate_objects" lifecycle = select_crm_generation("Set the HubSpot lifecycle stage to customer") assert lifecycle.surface == "provider_action" assert lifecycle.provider_action == "hubspot.update_contact" unsupported = select_crm_generation("Associate this contact with the deal") assert unsupported.surface == "gap" def test_cross_provider_generation_is_identical_at_capability_level() -> None: prompt = "Create a CRM contact when this form is submitted." hubspot = select_crm_generation(prompt, connected_provider="hubspot") salesforce = select_crm_generation(prompt, connected_provider="salesforce") assert hubspot.capability == salesforce.capability == "crm.contact.create" assert hubspot.payload == salesforce.payload envelope_hs = _connectors_manifest_envelope(items={"hubspot": {"actions": []}}) envelope_sf = _connectors_manifest_envelope(items={"salesforce": {"actions": []}}) compact_hs = _compact_connectors(envelope_hs) compact_sf = _compact_connectors(envelope_sf) hs_cap = compact_hs["capabilities"]["crm.contact.create"] sf_cap = compact_sf["capabilities"]["crm.contact.create"] assert hs_cap == sf_cap assert compact_hs["connected"] != compact_sf["connected"] def test_negative_generic_generation_must_not_leak_providers() -> None: dirty = """ if (provider === "hubspot") { await hubspot.create_contact({ dealstage: "appointmentscheduled", AccountId: "001" }) } """ leaks = generic_snippet_leaks(dirty) assert "hubspot.create_contact" in leaks assert "AccountId" in leaks assert 'provider === "hubspot"' in leaks for fixture in CRM_PROMPT_FIXTURES: decision = fixture["decision"] if decision.surface != "canonical": continue snippet = example_canonical_invoke(decision.capability, decision.payload or {}) assert generic_snippet_leaks(snippet) == () assert "hubspot.properties" not in snippet assert "salesforce.records" not in snippet def test_examples_are_provider_neutral() -> None: assert CRM_CONTACT_EXAMPLE["email"] == "customer@example.com" assert CRM_CONTACT_EXAMPLE["first_name"] == "Jane" assert "hs_object_id" not in CRM_CONTACT_EXAMPLE assert CRM_DEAL_EXAMPLE["stage"] == "qualified" assert "dealstage" not in CRM_DEAL_EXAMPLE assert "StageName" not in CRM_DEAL_EXAMPLE codegen = INVOKE_CAPABILITIES["crm.contact.create"]["codegen"] assert "customer@example.com" in codegen assert "Jane" in codegen assert "Never emit hubspot.create_contact" in codegen deal_codegen = INVOKE_CAPABILITIES["crm.deal.create"]["codegen"] assert "Enterprise Renewal" in deal_codegen assert "qualified" in deal_codegen assert "never dealstage" in deal_codegen assert CRM_CANONICAL_TYPES == ( "CRMContact", "CRMCompany", "CRMDeal", "CRMObjectRef", "CRMAssociation", ) assert supported_association_pairs() == ("contact->company", "deal->company") def test_provider_action_when_distinguishes_escape_hatches() -> None: hatches = _provider_action_escape_hatches() assert hatches["hubspot.create_contact"]["role"] == "escape_hatch" assert "not generic CRM contact create" in hatches["hubspot.create_contact"]["when"] assert hatches["salesforce.query_soql"]["implements"] == [] assert "SOQL" in hatches["salesforce.query_soql"]["when"] assert "filterGroups" in PROVIDER_ACTION_WHEN["hubspot.search_contacts"] assert "OpportunityContactRole" in PROVIDER_ACTION_WHEN["salesforce.associate_record"]