diff --git a/api/routes/organization.py b/api/routes/organization.py index a35eb292..01278140 100644 --- a/api/routes/organization.py +++ b/api/routes/organization.py @@ -566,34 +566,6 @@ def _set_nested_field(value: dict, dotted_path: str, field_value) -> None: current[parts[-1]] = field_value -async def _enforce_external_pbx_feature( - organization_id: int, - provider: str, - credentials: dict, - *, - existing_credentials: Optional[dict] = None, -) -> None: - if provider != "ari": - return - if await external_pbx_integrations_enabled(organization_id): - return - requested_external_pbx = credentials.get("external_pbx") - existing_external_pbx = (existing_credentials or {}).get("external_pbx") - if not requested_external_pbx and not existing_external_pbx: - return - # Disabling the feature hides and disables the integration, but does not - # destroy saved credentials. Allow only an unchanged masked round-trip. - if requested_external_pbx == existing_external_pbx: - return - raise HTTPException( - status_code=403, - detail=( - "External PBX integrations are disabled for this organization. " - "Enable them in Platform Settings before changing this configuration." - ), - ) - - def _credentials_from_payload(config: TelephonyConfigRequest) -> dict: """Provider credentials only — strip provider/from_numbers from the payload.""" payload = config.model_dump() @@ -690,9 +662,6 @@ async def create_telephony_configuration( raise HTTPException(status_code=400, detail="No organization selected") credentials = _credentials_from_payload(request.config) - await _enforce_external_pbx_feature( - user.selected_organization_id, request.config.provider, credentials - ) credentials = await _run_preprocess_hook(request.config.provider, credentials) try: @@ -775,12 +744,6 @@ async def update_telephony_configuration( preserve_masked_fields( existing.provider, credentials, existing.credentials or {} ) - await _enforce_external_pbx_feature( - user.selected_organization_id, - existing.provider, - credentials, - existing_credentials=existing.credentials or {}, - ) credentials = await _run_preprocess_hook(existing.provider, credentials) row = await db_client.update_telephony_configuration( diff --git a/api/routes/workflow.py b/api/routes/workflow.py index c030790d..800bf6aa 100644 --- a/api/routes/workflow.py +++ b/api/routes/workflow.py @@ -41,10 +41,14 @@ from api.services.configuration.resolve import ( resolve_effective_config, ) from api.services.mps_service_key_client import mps_service_key_client -from api.services.organization_preferences import external_pbx_integrations_enabled from api.services.posthog_client import capture_event from api.services.reports import generate_workflow_report_csv from api.services.storage import storage_fs +from api.services.workflow.configuration_policy import ( + ExternalPBXConfigurationDisabledError, + WorkflowConfigurationNotFoundError, + apply_external_pbx_mapping_policy, +) from api.services.workflow.dto import ReactFlowDTO, sanitize_workflow_definition from api.services.workflow.duplicate import duplicate_workflow from api.services.workflow.errors import ItemKind, WorkflowError @@ -1051,41 +1055,16 @@ async def update_workflow( if request.workflow_configurations is not None else None ) - if workflow_configurations is not None and not ( - await external_pbx_integrations_enabled(user.selected_organization_id) - ): - existing_workflow = await db_client.get_workflow( - workflow_id, organization_id=user.selected_organization_id + try: + workflow_configurations = await apply_external_pbx_mapping_policy( + workflow_configurations, + workflow_id=workflow_id, + organization_id=user.selected_organization_id, ) - if existing_workflow is None: - raise HTTPException( - status_code=404, detail=f"Workflow with id {workflow_id} not found" - ) - existing_draft = await db_client.get_draft_version(workflow_id) - existing_configs = ( - existing_draft.workflow_configurations - if existing_draft - else existing_workflow.released_definition.workflow_configurations - ) - existing_mappings = (existing_configs or {}).get( - "external_pbx_field_mappings", [] - ) - incoming_mappings = workflow_configurations.get( - "external_pbx_field_mappings", existing_mappings - ) - if incoming_mappings != existing_mappings: - raise HTTPException( - status_code=403, - detail=( - "External PBX integrations are disabled for this organization. " - "Enable them in Platform Settings before changing field " - "mappings." - ), - ) - if existing_mappings: - workflow_configurations["external_pbx_field_mappings"] = ( - existing_mappings - ) + except WorkflowConfigurationNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except ExternalPBXConfigurationDisabledError as exc: + raise HTTPException(status_code=403, detail=str(exc)) from exc if workflow_configurations and workflow_configurations.get( WORKFLOW_MODEL_CONFIGURATION_V2_OVERRIDE_KEY ): diff --git a/api/schemas/tool.py b/api/schemas/tool.py index 00e950d4..742b6591 100644 --- a/api/schemas/tool.py +++ b/api/schemas/tool.py @@ -258,7 +258,10 @@ class ContextDestinationMappingConfig(BaseModel): @field_validator("context_path") @classmethod def strip_context_path(cls, value: str) -> str: - return value.strip() + stripped = value.strip() + if not stripped: + raise ValueError("context path cannot be blank") + return stripped @field_validator("fallback_destination") @classmethod diff --git a/api/schemas/workflow_configurations.py b/api/schemas/workflow_configurations.py index 1125447d..b74d3cf0 100644 --- a/api/schemas/workflow_configurations.py +++ b/api/schemas/workflow_configurations.py @@ -22,15 +22,15 @@ class ExternalPBXFieldMapping(BaseModel): context_path: str = Field(min_length=1, max_length=255) destination_field: str = Field(pattern=r"^[A-Za-z][A-Za-z0-9_]{0,63}$") - @field_validator("context_path") + @field_validator("context_path", mode="before") @classmethod - def strip_context_path(cls, value: str) -> str: - return value.strip() + def strip_context_path(cls, value: object) -> object: + return value.strip() if isinstance(value, str) else value - @field_validator("destination_field") + @field_validator("destination_field", mode="before") @classmethod - def strip_destination_field(cls, value: str) -> str: - return value.strip() + def strip_destination_field(cls, value: object) -> object: + return value.strip() if isinstance(value, str) else value class AmbientNoiseConfigurationDefaults(BaseModel): diff --git a/api/services/workflow/configuration_policy.py b/api/services/workflow/configuration_policy.py new file mode 100644 index 00000000..a77b33f5 --- /dev/null +++ b/api/services/workflow/configuration_policy.py @@ -0,0 +1,70 @@ +"""Workflow-configuration policies shared by workflow update surfaces.""" + +from __future__ import annotations + +from copy import deepcopy +from typing import Any + +from api.db import db_client +from api.services.organization_preferences import external_pbx_integrations_enabled + + +class WorkflowConfigurationNotFoundError(LookupError): + """Raised when configuration policy cannot resolve the target workflow.""" + + +class ExternalPBXConfigurationDisabledError(PermissionError): + """Raised when a disabled organization changes external-PBX mappings.""" + + +async def apply_external_pbx_mapping_policy( + workflow_configurations: dict[str, Any] | None, + *, + workflow_id: int, + organization_id: int, +) -> dict[str, Any] | None: + """Preserve hidden mappings and reject edits while external PBX is disabled. + + Workflow configuration updates replace the stored configuration document. When + the External PBX UI is hidden, its field mappings are absent from the request, + so they must be copied from the active draft or published definition before the + update is persisted. + """ + if workflow_configurations is None or await external_pbx_integrations_enabled( + organization_id + ): + return workflow_configurations + + workflow = await db_client.get_workflow( + workflow_id, organization_id=organization_id + ) + if workflow is None: + raise WorkflowConfigurationNotFoundError( + f"Workflow with id {workflow_id} not found" + ) + + draft = await db_client.get_draft_version(workflow_id) + stored_configurations = ( + draft.workflow_configurations + if draft + else workflow.released_definition.workflow_configurations + ) + stored_mappings = (stored_configurations or {}).get( + "external_pbx_field_mappings", [] + ) + incoming_mappings = workflow_configurations.get( + "external_pbx_field_mappings", stored_mappings + ) + + if incoming_mappings != stored_mappings: + raise ExternalPBXConfigurationDisabledError( + "External PBX integrations are disabled for this organization. " + "Enable them in Platform Settings before changing field mappings." + ) + + if not stored_mappings: + return workflow_configurations + + prepared_configurations = dict(workflow_configurations) + prepared_configurations["external_pbx_field_mappings"] = deepcopy(stored_mappings) + return prepared_configurations diff --git a/api/services/workflow/pipecat_engine_custom_tools.py b/api/services/workflow/pipecat_engine_custom_tools.py index c41f8664..3bc71167 100644 --- a/api/services/workflow/pipecat_engine_custom_tools.py +++ b/api/services/workflow/pipecat_engine_custom_tools.py @@ -626,18 +626,6 @@ class CustomToolManager: is_dynamic_transfer = config.get( "destination_source", "static" ) == "dynamic" and isinstance(resolver, dict) - resolver_phase_muted = False - - def clear_transfer_setup_mute_state() -> None: - nonlocal resolver_phase_muted - if resolver_phase_muted: - self._engine.set_mute_pipeline(False) - resolver_phase_muted = False - self._engine._queued_speech_mute_state = "idle" - - if is_dynamic_transfer: - self._engine.set_mute_pipeline(True) - resolver_phase_muted = True if is_dynamic_transfer and resolver.get("wait_message"): await self._engine.task.queue_frame( @@ -647,7 +635,6 @@ class CustomToolManager: persist_to_logs=True, ) ) - self._engine._queued_speech_mute_state = "waiting" try: resolved_transfer = await resolve_transfer_config( @@ -662,7 +649,6 @@ class CustomToolManager: destination = resolved_transfer.destination timeout_seconds = resolved_transfer.timeout_seconds except TransferResolutionError as e: - clear_transfer_setup_mute_state() validation_error_result = { "status": "failed", "message": "I'm sorry, but I couldn't find a valid destination for this transfer.", @@ -678,7 +664,6 @@ class CustomToolManager: resolved_transfer.source == "context_mapping" and not external_pbx_call ): - clear_transfer_setup_mute_state() await self._handle_transfer_result( { "status": "failed", @@ -702,7 +687,6 @@ class CustomToolManager: "action": "transfer_failed", "reason": "no_destination", } - clear_transfer_setup_mute_state() await self._handle_transfer_result( validation_error_result, function_call_params, properties ) @@ -720,11 +704,8 @@ class CustomToolManager: persist_to_logs=True, ) ) - self._engine._queued_speech_mute_state = "waiting" else: - played = await self._play_config_message(config) - if played: - self._engine._queued_speech_mute_state = "waiting" + await self._play_config_message(config) if external_pbx_call: workflow_configurations = ( @@ -742,7 +723,6 @@ class CustomToolManager: field_updates=field_updates, ) if external_result is not None: - clear_transfer_setup_mute_state() if external_result.get("status") == "success": self._engine._gathered_context[ "external_pbx_transferred" @@ -779,7 +759,6 @@ class CustomToolManager: "action": "transfer_failed", "reason": "provider_does_not_support_transfer", } - clear_transfer_setup_mute_state() await self._handle_transfer_result( validation_error_result, function_call_params, properties ) @@ -830,7 +809,6 @@ class CustomToolManager: except Exception as e: logger.error(f"Transfer provider failed: {e}") self._engine.set_mute_pipeline(False) - self._engine._queued_speech_mute_state = "idle" await call_transfer_manager.remove_transfer_context(transfer_id) provider_error_result = { "status": "failed", @@ -930,7 +908,6 @@ class CustomToolManager: f"Transfer call tool '{function_name}' execution failed: {e}" ) self._engine.set_mute_pipeline(False) - self._engine._queued_speech_mute_state = "idle" # Handle generic exception with user-friendly message exception_result = { diff --git a/api/tests/telephony/test_external_pbx_configuration.py b/api/tests/telephony/test_external_pbx_configuration.py index 1256bb53..5639e7c9 100644 --- a/api/tests/telephony/test_external_pbx_configuration.py +++ b/api/tests/telephony/test_external_pbx_configuration.py @@ -1,7 +1,6 @@ from unittest.mock import AsyncMock import pytest -from fastapi import HTTPException from api.routes import organization from api.services import tool_management @@ -43,42 +42,6 @@ def test_nested_masked_external_pbx_secrets_are_restored_on_update(): assert request["external_pbx"]["agent_api"]["password"] == "agent-secret" -@pytest.mark.asyncio -async def test_disabled_feature_allows_unchanged_telephony_configuration(monkeypatch): - monkeypatch.setattr( - organization, - "external_pbx_integrations_enabled", - AsyncMock(return_value=False), - ) - existing = _credentials() - - await organization._enforce_external_pbx_feature( - 7, - "ari", - _credentials(), - existing_credentials=existing, - ) - - -@pytest.mark.asyncio -async def test_disabled_feature_rejects_removing_telephony_configuration(monkeypatch): - monkeypatch.setattr( - organization, - "external_pbx_integrations_enabled", - AsyncMock(return_value=False), - ) - - with pytest.raises(HTTPException) as exc_info: - await organization._enforce_external_pbx_feature( - 7, - "ari", - {"external_pbx": None}, - existing_credentials=_credentials(), - ) - - assert exc_info.value.status_code == 403 - - @pytest.mark.asyncio async def test_disabled_feature_preserves_existing_tool_mapping(monkeypatch): monkeypatch.setattr( diff --git a/api/tests/test_custom_tools.py b/api/tests/test_custom_tools.py index 5ff818c3..659f77e0 100644 --- a/api/tests/test_custom_tools.py +++ b/api/tests/test_custom_tools.py @@ -1704,7 +1704,6 @@ class TestCustomToolManagerUnit: assert [ call.args[0] for call in mock_engine.set_mute_pipeline.call_args_list ] == [ - True, True, False, ] @@ -1716,8 +1715,8 @@ class TestCustomToolManagerUnit: assert "I will connect you with our Texas partner now." in spoken_texts @pytest.mark.asyncio - async def test_transfer_call_resolver_failure_resets_queued_speech_mute(self): - """Resolver failure after a wait message should not leave user input muted.""" + async def test_transfer_call_resolver_failure_does_not_toggle_pipeline_mute(self): + """Function-call muting covers resolver execution without pipeline state.""" from api.services.workflow.pipecat_engine_custom_tools import CustomToolManager mock_engine = Mock() @@ -1729,7 +1728,6 @@ class TestCustomToolManagerUnit: mock_engine.task = SimpleNamespace(queue_frame=AsyncMock()) mock_engine.set_mute_pipeline = Mock() mock_engine.end_call_with_reason = AsyncMock() - mock_engine._queued_speech_mute_state = "idle" manager = CustomToolManager(mock_engine) tool = MockToolModel( @@ -1792,13 +1790,7 @@ class TestCustomToolManagerUnit: assert result_received["status"] == "transfer_failed" assert result_received["reason"] == "no_destination" - assert mock_engine._queued_speech_mute_state == "idle" - assert [ - call.args[0] for call in mock_engine.set_mute_pipeline.call_args_list - ] == [ - True, - False, - ] + mock_engine.set_mute_pipeline.assert_not_called() @pytest.mark.asyncio async def test_transfer_call_propagates_provider_destination_error(self): diff --git a/api/tests/test_tool_schema.py b/api/tests/test_tool_schema.py index 76e638f4..dd354db7 100644 --- a/api/tests/test_tool_schema.py +++ b/api/tests/test_tool_schema.py @@ -66,6 +66,17 @@ def test_transfer_call_context_mapping_accepts_unique_routes(): assert config.context_mapping.routes[0].destination == "sales" +def test_transfer_call_context_mapping_rejects_blank_context_path(): + with pytest.raises(ValueError, match="context path cannot be blank"): + TransferCallConfig( + destination_source="context_mapping", + context_mapping={ + "context_path": " ", + "routes": [{"context_value": "yes", "destination": "sales"}], + }, + ) + + def test_transfer_call_context_mapping_rejects_duplicate_values_case_insensitively(): with pytest.raises(ValueError, match="must be unique"): TransferCallConfig( diff --git a/api/tests/test_workflow_configuration_policy.py b/api/tests/test_workflow_configuration_policy.py new file mode 100644 index 00000000..a6e7968f --- /dev/null +++ b/api/tests/test_workflow_configuration_policy.py @@ -0,0 +1,95 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from api.services.workflow import configuration_policy + + +def _stored_workflow(mappings: list[dict]): + return SimpleNamespace( + released_definition=SimpleNamespace( + workflow_configurations={"external_pbx_field_mappings": mappings} + ) + ) + + +@pytest.mark.asyncio +async def test_disabled_external_pbx_policy_preserves_hidden_mappings(monkeypatch): + mappings = [{"context_path": "qualified", "destination_field": "address3"}] + get_workflow = AsyncMock(return_value=_stored_workflow(mappings)) + get_draft = AsyncMock(return_value=None) + monkeypatch.setattr( + configuration_policy, + "external_pbx_integrations_enabled", + AsyncMock(return_value=False), + ) + monkeypatch.setattr(configuration_policy.db_client, "get_workflow", get_workflow) + monkeypatch.setattr(configuration_policy.db_client, "get_draft_version", get_draft) + + incoming = {"max_call_duration": 600} + prepared = await configuration_policy.apply_external_pbx_mapping_policy( + incoming, + workflow_id=12, + organization_id=7, + ) + + assert prepared == { + "max_call_duration": 600, + "external_pbx_field_mappings": mappings, + } + assert "external_pbx_field_mappings" not in incoming + get_workflow.assert_awaited_once_with(12, organization_id=7) + get_draft.assert_awaited_once_with(12) + + +@pytest.mark.asyncio +async def test_disabled_external_pbx_policy_rejects_mapping_changes(monkeypatch): + stored = [{"context_path": "qualified", "destination_field": "address3"}] + monkeypatch.setattr( + configuration_policy, + "external_pbx_integrations_enabled", + AsyncMock(return_value=False), + ) + monkeypatch.setattr( + configuration_policy.db_client, + "get_workflow", + AsyncMock(return_value=_stored_workflow(stored)), + ) + monkeypatch.setattr( + configuration_policy.db_client, + "get_draft_version", + AsyncMock(return_value=None), + ) + + with pytest.raises(configuration_policy.ExternalPBXConfigurationDisabledError): + await configuration_policy.apply_external_pbx_mapping_policy( + { + "external_pbx_field_mappings": [ + {"context_path": "qualified", "destination_field": "comments"} + ] + }, + workflow_id=12, + organization_id=7, + ) + + +@pytest.mark.asyncio +async def test_enabled_external_pbx_policy_does_not_load_workflow(monkeypatch): + get_workflow = AsyncMock() + monkeypatch.setattr( + configuration_policy, + "external_pbx_integrations_enabled", + AsyncMock(return_value=True), + ) + monkeypatch.setattr(configuration_policy.db_client, "get_workflow", get_workflow) + incoming = {"external_pbx_field_mappings": []} + + prepared = await configuration_policy.apply_external_pbx_mapping_policy( + incoming, + workflow_id=12, + organization_id=7, + ) + + assert prepared is incoming + get_workflow.assert_not_awaited() diff --git a/api/tests/test_workflow_configurations_schema.py b/api/tests/test_workflow_configurations_schema.py index 867f1763..185bf5bf 100644 --- a/api/tests/test_workflow_configurations_schema.py +++ b/api/tests/test_workflow_configurations_schema.py @@ -64,13 +64,23 @@ def test_cap_stays_within_concurrency_stale_timeout(): def test_external_pbx_field_mapping_is_validated(): config = WorkflowConfigurationDefaults( external_pbx_field_mappings=[ - {"context_path": "qualified", "destination_field": "address3"} + {"context_path": " qualified ", "destination_field": " address3 "} ] ) + assert config.external_pbx_field_mappings[0].context_path == "qualified" assert config.external_pbx_field_mappings[0].destination_field == "address3" +def test_external_pbx_field_mapping_rejects_blank_context_paths(): + with pytest.raises(ValidationError, match="context_path"): + WorkflowConfigurationDefaults( + external_pbx_field_mappings=[ + {"context_path": " ", "destination_field": "address3"} + ] + ) + + def test_external_pbx_field_mapping_rejects_invalid_field_names(): with pytest.raises(ValidationError, match="destination_field"): WorkflowConfigurationDefaults(