Merge remote-tracking branch 'origin/main' into review-560

# Conflicts:
#	api/services/telephony/ari_manager.py

api/db/workflow_run_client.py

@@ -229,6 +229,26 @@ class WorkflowRunClient(BaseDBClient):
             )
             return result.scalars().first()
 
+    async def get_workflow_run_configurations(
+        self, run_id: int, organization_id: int
+    ) -> dict:
+        """Load the immutable workflow configuration snapshot for one run."""
+
+        async with self.async_session() as session:
+            result = await session.execute(
+                select(WorkflowDefinitionModel.workflow_configurations)
+                .join(
+                    WorkflowRunModel,
+                    WorkflowRunModel.definition_id == WorkflowDefinitionModel.id,
+                )
+                .join(WorkflowModel, WorkflowRunModel.workflow_id == WorkflowModel.id)
+                .where(
+                    WorkflowRunModel.id == run_id,
+                    WorkflowModel.organization_id == organization_id,
+                )
+            )
+            return result.scalar_one_or_none() or {}
+
     async def get_organization_id_by_workflow_run_id(
         self, run_id: int | None
     ) -> int | None:

api/routes/campaign.py

@@ -696,8 +696,8 @@ async def update_campaign(
 @router.get("/{campaign_id}/runs")
 async def get_campaign_runs(
     campaign_id: int,
-    page: int = 1,
-    limit: int = 50,
+    page: int = Query(1, ge=1, description="Page number (starts from 1)"),
+    limit: int = Query(50, ge=1, le=100, description="Number of items per page"),
     filters: Optional[str] = Query(None, description="JSON-encoded filter criteria"),
     sort_by: Optional[str] = Query(
         None, description="Field to sort by (e.g., 'duration', 'created_at')"

api/routes/organization.py

@@ -1,4 +1,5 @@
-from typing import List, Optional
+from copy import deepcopy
+from typing import Any, List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 from loguru import logger
@@ -74,6 +75,7 @@ from api.services.organization_context import (
     get_organization_context,
 )
 from api.services.organization_preferences import (
+    external_pbx_integrations_enabled,
     get_organization_preferences,
     upsert_organization_preferences,
 )
@@ -101,14 +103,24 @@ def _sensitive_fields(provider_name: str) -> List[str]:
 
 def _mask_sensitive(provider_name: str, value: dict) -> dict:
     """Return a copy of ``value`` with sensitive fields masked for display."""
-    out = dict(value)
+    out = deepcopy(value)
     for field_name in _sensitive_fields(provider_name):
-        v = out.get(field_name)
+        v = _get_nested_field(out, field_name)
         if v:
-            out[field_name] = mask_key(v)
+            _set_nested_field(out, field_name, mask_key(str(v)))
     return out
 
 
+class TelephonyProviderUIOption(BaseModel):
+    value: str
+    label: str
+
+
+class TelephonyProviderUICondition(BaseModel):
+    field: str
+    equals: Any
+
+
 class TelephonyProviderUIField(BaseModel):
     """One form field on a telephony provider's configuration UI."""
 
@@ -119,6 +131,9 @@ class TelephonyProviderUIField(BaseModel):
     sensitive: bool
     description: Optional[str] = None
     placeholder: Optional[str] = None
+    options: Optional[List[TelephonyProviderUIOption]] = None
+    visible_when: Optional[TelephonyProviderUICondition] = None
+    section: Optional[str] = None
 
 
 class TelephonyProviderMetadata(BaseModel):
@@ -184,6 +199,9 @@ async def get_telephony_providers_metadata(user: UserModel = Depends(get_user)):
     if not user.selected_organization_id:
         raise HTTPException(status_code=400, detail="No organization selected")
 
+    external_pbx_enabled = await external_pbx_integrations_enabled(
+        user.selected_organization_id
+    )
     providers = []
     for spec in telephony_registry.all_specs():
         if spec.ui_metadata is None:
@@ -201,8 +219,30 @@ async def get_telephony_providers_metadata(user: UserModel = Depends(get_user)):
                         sensitive=f.sensitive,
                         description=f.description,
                         placeholder=f.placeholder,
+                        options=(
+                            [
+                                {"value": option.value, "label": option.label}
+                                for option in f.options
+                            ]
+                            if f.options
+                            else None
+                        ),
+                        visible_when=(
+                            {
+                                "field": f.visible_when.field,
+                                "equals": f.visible_when.equals,
+                            }
+                            if f.visible_when
+                            else None
+                        ),
+                        section=f.section,
                     )
                     for f in spec.ui_metadata.fields
+                    if not f.feature_gate
+                    or (
+                        f.feature_gate == "external_pbx_integrations"
+                        and external_pbx_enabled
+                    )
                 ],
                 docs_url=spec.ui_metadata.docs_url,
             )
@@ -496,24 +536,34 @@ async def get_model_configuration_preferences_legacy(
     return await get_preferences(user=user)
 
 
-@router.put(
-    "/model-configurations/preferences",
-    response_model=OrganizationPreferences,
-    include_in_schema=False,
-)
-async def save_model_configuration_preferences_legacy(
-    request: OrganizationPreferences,
-    user: UserModel = Depends(get_user_with_selected_organization),
-):
-    return await save_preferences(request=request, user=user)
-
-
 def preserve_masked_fields(provider: str, request_dict: dict, existing: dict):
     """If the client re-submitted a masked sensitive field, restore the original."""
     for field_name in _sensitive_fields(provider):
-        v = request_dict.get(field_name)
-        if v and is_mask_of(v, existing.get(field_name, "")):
-            request_dict[field_name] = existing[field_name]
+        v = _get_nested_field(request_dict, field_name)
+        existing_value = _get_nested_field(existing, field_name)
+        if v and is_mask_of(v, existing_value or ""):
+            _set_nested_field(request_dict, field_name, existing_value)
+
+
+def _get_nested_field(value: dict, dotted_path: str):
+    current = value
+    for part in dotted_path.split("."):
+        if not isinstance(current, dict):
+            return None
+        current = current.get(part)
+    return current
+
+
+def _set_nested_field(value: dict, dotted_path: str, field_value) -> None:
+    current = value
+    parts = dotted_path.split(".")
+    for part in parts[:-1]:
+        child = current.get(part)
+        if not isinstance(child, dict):
+            child = {}
+            current[part] = child
+        current = child
+    current[parts[-1]] = field_value
 
 
 def _credentials_from_payload(config: TelephonyConfigRequest) -> dict:

api/routes/tool.py

@@ -1,5 +1,6 @@
 """API routes for managing tools."""
 
+import time
 from typing import List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException
@@ -22,6 +23,8 @@ from api.schemas.tool import (
     ToolDefinition,
     ToolParameter,
     ToolResponse,
+    ToolTestRequest,
+    ToolTestResponse,
     TransferCallConfig,
     TransferCallToolDefinition,
     UpdateToolRequest,
@@ -33,11 +36,16 @@ from api.services.tool_management import (
     build_tool_response,
     create_tool_for_user,
     refresh_mcp_tool_for_user,
+    validate_external_pbx_tool_definition,
     validate_tool_credential_references,
 )
 from api.services.tool_management import (
     populate_discovered_tools as _populate_discovered_tools,
 )
+from api.services.workflow.tools.custom_tool import (
+    execute_http_tool,
+    serialize_query_params,
+)
 
 router = APIRouter(prefix="/tools")
 
@@ -56,6 +64,8 @@ __all__ = [
     "ToolDefinition",
     "ToolParameter",
     "ToolResponse",
+    "ToolTestRequest",
+    "ToolTestResponse",
     "TransferCallConfig",
     "TransferCallToolDefinition",
     "UpdateToolRequest",
@@ -195,6 +205,149 @@ async def refresh_mcp_tools(
         raise HTTPException(status_code=e.status_code, detail=e.message) from e
 
 
+@router.post("/{tool_uuid}/test")
+async def test_tool(
+    tool_uuid: str,
+    request: ToolTestRequest,
+    user: UserModel = Depends(get_user),
+) -> ToolTestResponse:
+    """Execute an HTTP API tool with sample LLM and preset parameters."""
+    if not user.selected_organization_id:
+        raise HTTPException(
+            status_code=400, detail="No organization selected for the user"
+        )
+
+    tool = await db_client.get_tool_by_uuid(
+        tool_uuid, user.selected_organization_id, include_archived=True
+    )
+
+    if not tool:
+        raise HTTPException(status_code=404, detail="Tool not found")
+
+    if tool.category != ToolCategory.HTTP_API.value:
+        raise HTTPException(status_code=400, detail="Only HTTP API tools can be tested")
+
+    tool_config = (
+        tool.definition.get("config", {}) if isinstance(tool.definition, dict) else {}
+    )
+    configured_method = tool_config.get("method", "?")
+    configured_url = tool_config.get("url", "?")
+
+    started_at = time.perf_counter()
+    result = await execute_http_tool(
+        tool,
+        request.llm_params,
+        preset_params=request.preset_params,
+        organization_id=user.selected_organization_id,
+        include_request_headers=True,
+    )
+    duration_ms = max(0, round((time.perf_counter() - started_at) * 1000))
+
+    status = result.get("status", "error")
+    status_code = result.get("status_code")
+    if status_code is not None and status_code >= 400:
+        status = "error"
+
+    hint = _hint_for_status_code(status_code, configured_method)
+
+    # Preset values take precedence over model-supplied values, matching live
+    # execution after configured preset templates have been resolved.
+    resolved_arguments = {**request.llm_params, **request.preset_params}
+
+    # Mirror execute_http_tool's own branch: POST/PUT/PATCH send the
+    # resolved arguments as a JSON body; GET/DELETE send them as query
+    # params. Never both.
+    request_body = None
+    request_params = None
+    if configured_method in ("POST", "PUT", "PATCH"):
+        request_body = resolved_arguments  # keep {} so preview matches wire request
+    elif resolved_arguments:
+        request_params = serialize_query_params(resolved_arguments)
+
+    return ToolTestResponse(
+        status=status,
+        status_code=status_code,
+        data=result.get("data"),
+        error=result.get("error"),
+        duration_ms=duration_ms,
+        hint=hint,
+        request_method=configured_method,
+        request_url=configured_url,
+        request_headers=result.get("request_headers", {}),
+        request_body=request_body,
+        request_params=request_params,
+    )
+
+
+def _hint_for_status_code(
+    status_code: Optional[int], configured_method: str
+) -> Optional[str]:
+    """Human-readable explanation for a status code a misconfigured tool
+    is likely to hit. Returns None for 2xx and any code not covered."""
+    if status_code == 400:
+        return (
+            "HTTP 400 Bad Request — the server rejected the request payload. "
+            "Verify the arguments/body match what this endpoint expects."
+        )
+    if status_code == 401:
+        return (
+            "HTTP 401 Unauthorized — the request wasn't authenticated. Check "
+            "the credential configured on the Authentication tab is present "
+            "and valid."
+        )
+    if status_code == 403:
+        return (
+            "HTTP 403 Forbidden — authenticated, but the configured "
+            "credential doesn't have permission for this endpoint/action."
+        )
+    if status_code == 404:
+        return (
+            f"HTTP 404 Not Found — verify the endpoint URL is correct and "
+            f"that {configured_method} is a valid method for it."
+        )
+    if status_code == 405:
+        return (
+            f"HTTP 405 Method Not Allowed — the endpoint rejected the "
+            f"configured method ({configured_method}). Verify the API expects "
+            f"{configured_method} for this URL."
+        )
+    if status_code == 408:
+        return (
+            "HTTP 408 Request Timeout — the endpoint didn't respond in time. "
+            "Check the endpoint is reachable, or increase Timeout (ms) if it's "
+            "just slow."
+        )
+    if status_code == 409:
+        return (
+            "HTTP 409 Conflict — the endpoint rejected the request due to a "
+            "conflicting resource state (e.g. duplicate create). Not "
+            "necessarily a configuration problem."
+        )
+    if status_code == 415:
+        return (
+            "HTTP 415 Unsupported Media Type — check the Content-Type header "
+            "matches the format this endpoint expects for the body."
+        )
+    if status_code == 422:
+        return (
+            "HTTP 422 Unprocessable Entity — the request was well-formed but "
+            "the payload's structure or field types don't match what this "
+            "endpoint expects. Compare your arguments against the API's "
+            "documented schema."
+        )
+    if status_code == 429:
+        return (
+            "HTTP 429 Too Many Requests — the endpoint is rate-limiting. Wait "
+            "and retry; not a configuration problem."
+        )
+    if status_code is not None and 500 <= status_code < 600:
+        return (
+            f"HTTP {status_code} — the endpoint itself errored. This is "
+            "likely an issue on the API's side, not your tool configuration."
+        )
+    return None
+
+
 @router.put("/{tool_uuid}")
 async def update_tool(
     tool_uuid: str,
@@ -223,6 +376,18 @@ async def update_tool(
     if request.definition:
         definition = request.definition.model_dump()
         try:
+            existing_tool = await db_client.get_tool_by_uuid(
+                tool_uuid,
+                user.selected_organization_id,
+                include_archived=True,
+            )
+            await validate_external_pbx_tool_definition(
+                definition,
+                organization_id=user.selected_organization_id,
+                existing_definition=(
+                    existing_tool.definition if existing_tool else None
+                ),
+            )
             await validate_tool_credential_references(
                 definition,
                 organization_id=user.selected_organization_id,

api/routes/workflow.py

@@ -44,6 +44,11 @@ from api.services.mps_service_key_client import mps_service_key_client
 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,6 +1056,16 @@ async def update_workflow(
             if request.workflow_configurations is not None
             else None
         )
+        try:
+            workflow_configurations = await apply_external_pbx_mapping_policy(
+                workflow_configurations,
+                workflow_id=workflow_id,
+                organization_id=user.selected_organization_id,
+            )
+        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
         ):
@@ -1407,8 +1422,8 @@ class WorkflowRunsResponse(BaseModel):
 @router.get("/{workflow_id}/runs")
 async def get_workflow_runs(
     workflow_id: int,
-    page: int = 1,
-    limit: int = 50,
+    page: int = Query(1, ge=1, description="Page number (starts from 1)"),
+    limit: int = Query(50, ge=1, le=100, description="Number of items per page"),
     filters: Optional[str] = Query(None, description="JSON-encoded filter criteria"),
     sort_by: Optional[str] = Query(
         None, description="Field to sort by (e.g., 'duration', 'created_at')"

api/schemas/organization_preferences.py

@@ -4,3 +4,4 @@ from pydantic import BaseModel
 class OrganizationPreferences(BaseModel):
     test_phone_number: str | None = None
     timezone: str | None = None
+    external_pbx_integrations_enabled: bool = False

api/schemas/tool.py

@@ -223,12 +223,70 @@ class HttpTransferResolverConfig(BaseModel):
         return v
 
 
+class ContextDestinationRoute(BaseModel):
+    """Map one gathered-context value to an external-PBX destination."""
+
+    context_value: str = Field(min_length=1, max_length=255)
+    destination: str = Field(min_length=1, max_length=255)
+
+    @field_validator("context_value", "destination")
+    @classmethod
+    def strip_non_empty(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped:
+            raise ValueError("mapping values cannot be blank")
+        return stripped
+
+
+class ContextDestinationMappingConfig(BaseModel):
+    """Resolve an external-PBX destination from gathered context."""
+
+    context_path: str = Field(
+        min_length=1,
+        max_length=255,
+        description=(
+            "Gathered-context path or extracted-variable name used for routing."
+        ),
+    )
+    routes: List[ContextDestinationRoute] = Field(min_length=1, max_length=100)
+    fallback_destination: Optional[str] = Field(
+        default=None,
+        max_length=255,
+        description="Optional provider-native fallback destination.",
+    )
+
+    @field_validator("context_path")
+    @classmethod
+    def strip_context_path(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped:
+            raise ValueError("context path cannot be blank")
+        return stripped
+
+    @field_validator("fallback_destination")
+    @classmethod
+    def normalize_fallback(cls, value: Optional[str]) -> Optional[str]:
+        if value is None:
+            return None
+        return value.strip() or None
+
+    @model_validator(mode="after")
+    def validate_unique_values(self):
+        values = [route.context_value.casefold() for route in self.routes]
+        if len(values) != len(set(values)):
+            raise ValueError("context mapping values must be unique")
+        return self
+
+
 class TransferCallConfig(BaseModel):
     """Configuration for Transfer Call tools."""
 
-    destination_source: Literal["static", "dynamic"] = Field(
+    destination_source: Literal["static", "dynamic", "context_mapping"] = Field(
         default="static",
-        description="Whether transfer destination is static/template or resolved by HTTP.",
+        description=(
+            "Whether the destination is static/template, resolved by HTTP, or "
+            "mapped from gathered context to an external-PBX destination."
+        ),
     )
     destination: str = Field(
         default="",
@@ -263,6 +321,10 @@ class TransferCallConfig(BaseModel):
         default=None,
         description="Optional resolver that determines transfer routing at call time.",
     )
+    context_mapping: Optional[ContextDestinationMappingConfig] = Field(
+        default=None,
+        description="Optional gathered-context to external-PBX destination mapping.",
+    )
 
     @model_validator(mode="after")
     def validate_destination_source_config(self):
@@ -270,6 +332,14 @@ class TransferCallConfig(BaseModel):
             raise ValueError(
                 "config.resolver is required when destination_source is dynamic"
             )
+        if (
+            self.destination_source == "context_mapping"
+            and self.context_mapping is None
+        ):
+            raise ValueError(
+                "config.context_mapping is required when destination_source is "
+                "context_mapping"
+            )
         return self
 
 
@@ -491,3 +561,32 @@ class McpRefreshResponse(BaseModel):
     tool_uuid: str
     discovered_tools: list = Field(default_factory=list)
     error: Optional[str] = None
+
+
+class ToolTestRequest(BaseModel):
+    """Request body for testing an HTTP API tool outside a live call."""
+
+    llm_params: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Values for parameters normally supplied by the model.",
+    )
+    preset_params: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Resolved values for parameters normally supplied from presets.",
+    )
+
+
+class ToolTestResponse(BaseModel):
+    """Result of testing an HTTP API tool."""
+
+    status: str
+    status_code: Optional[int] = None
+    data: Optional[Any] = None
+    error: Optional[str] = None
+    hint: Optional[str] = None
+    request_method: str
+    request_url: str
+    request_headers: Dict[str, str] = Field(default_factory=dict)
+    request_body: Optional[Dict[str, Any]] = None
+    request_params: Optional[Dict[str, Any]] = None
+    duration_ms: int

api/schemas/workflow_configurations.py

@@ -1,6 +1,6 @@
 from typing import Literal
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 DEFAULT_MAX_CALL_DURATION_SECONDS = 300
 # Hard ceiling on configurable call duration. Must stay <= the concurrency
@@ -16,6 +16,23 @@ DEFAULT_TURN_STOP_STRATEGY = "transcription"
 DEFAULT_CONTEXT_COMPACTION_ENABLED = False
 
 
+class ExternalPBXFieldMapping(BaseModel):
+    """Map one gathered-context value to a provider-native field."""
+
+    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", mode="before")
+    @classmethod
+    def strip_context_path(cls, value: object) -> object:
+        return value.strip() if isinstance(value, str) else value
+
+    @field_validator("destination_field", mode="before")
+    @classmethod
+    def strip_destination_field(cls, value: object) -> object:
+        return value.strip() if isinstance(value, str) else value
+
+
 class AmbientNoiseConfigurationDefaults(BaseModel):
     model_config = ConfigDict(extra="allow")
 
@@ -56,6 +73,10 @@ class WorkflowConfigurationDefaults(BaseModel):
     )
     dictionary: str = ""
     context_compaction_enabled: bool = DEFAULT_CONTEXT_COMPACTION_ENABLED
+    external_pbx_field_mappings: list[ExternalPBXFieldMapping] = Field(
+        default_factory=list,
+        max_length=100,
+    )
 
 
 def get_default_workflow_configurations() -> WorkflowConfigurationDefaults:

api/services/configuration/registry.py

@@ -584,6 +584,20 @@ class SarvamLLMConfiguration(BaseLLMConfiguration):
 
 
 OPENAI_REALTIME_MODELS = ["gpt-realtime-2"]
+# ISO 639-1 codes accepted by the Realtime API's input_audio_transcription.
+# Not exhaustive — the field allows custom input.
+OPENAI_REALTIME_LANGUAGES = [
+    "en",
+    "es",
+    "pt",
+    "fr",
+    "de",
+    "it",
+    "hi",
+    "ja",
+    "ko",
+    "zh",
+]
 OPENAI_REALTIME_VOICES = [
     "alloy",
     "ash",
@@ -618,6 +632,17 @@ class OpenAIRealtimeLLMConfiguration(BaseLLMConfiguration):
             "allow_custom_input": True,
         },
     )
+    language: str | None = Field(
+        default=None,
+        description=(
+            "ISO 639-1 language code for input audio transcription (e.g. 'pt', 'es'). "
+            "Improves transcription accuracy and latency. Leave unset to auto-detect."
+        ),
+        json_schema_extra={
+            "examples": OPENAI_REALTIME_LANGUAGES,
+            "allow_custom_input": True,
+        },
+    )
 
 
 GROK_REALTIME_MODELS = ["grok-voice-think-fast-1.0"]
@@ -1307,7 +1332,6 @@ class XAITTSConfiguration(BaseServiceConfiguration):
         description="BCP-47 language code for synthesis (e.g. 'en', 'fr', 'de'), or 'auto' for automatic language detection.",
         json_schema_extra={"allow_custom_input": True},
     )
-
     @computed_field
     @property
     def model(self) -> str:

api/services/organization_preferences.py

@@ -42,6 +42,16 @@ async def upsert_organization_preferences(
     return preferences
 
 
+async def external_pbx_integrations_enabled(
+    organization_id: int | None,
+    db=None,
+) -> bool:
+    """Return whether the organization opted into external-PBX integrations."""
+
+    preferences = await get_organization_preferences(organization_id, db=db)
+    return preferences.external_pbx_integrations_enabled
+
+
 async def _get_configuration(db, organization_id: int, key: str):
     row = db.get_configuration(organization_id, key)
     if isawaitable(row):

api/services/pipecat/audio_config.py

@@ -52,16 +52,6 @@ class AudioConfig:
             )
             self.pipeline_sample_rate = 16000
 
-        # Log configuration for auditing
-        logger.info(
-            f"AudioConfig initialized: "
-            f"transport_in={self.transport_in_sample_rate}Hz, "
-            f"transport_out={self.transport_out_sample_rate}Hz, "
-            f"vad={self.vad_sample_rate}Hz, "
-            f"pipeline={self.pipeline_sample_rate}Hz, "
-            f"buffer={self.buffer_size_seconds}s"
-        )
-
     @property
     def buffer_size_bytes(self) -> int:
         """Calculate buffer size in bytes based on pipeline sample rate."""

api/services/pipecat/run_pipeline.py

@@ -95,6 +95,9 @@ from pipecat.turns.user_start import (
     MinWordsUserTurnStartStrategy,
     ProvisionalVADUserTurnStartStrategy,
 )
+from pipecat.turns.user_start.transcription_user_turn_start_strategy import (
+    TranscriptionUserTurnStartStrategy,
+)
 from pipecat.turns.user_start.vad_user_turn_start_strategy import (
     VADUserTurnStartStrategy,
 )
@@ -162,7 +165,7 @@ def _create_non_realtime_user_turn_start_strategies(
         return [
             ProvisionalVADUserTurnStartStrategy(
                 pause_secs=_resolve_provisional_vad_pause_secs(run_configs)
-            )
+            ),
         ]
 
     if uses_external_turns:
@@ -172,7 +175,7 @@ def _create_non_realtime_user_turn_start_strategies(
         # confirms a real turn.
         return [ExternalUserTurnStartStrategy(enable_interruptions=True)]
 
-    return [VADUserTurnStartStrategy()]
+    return [TranscriptionUserTurnStartStrategy(), VADUserTurnStartStrategy()]
 
 
 def _create_non_realtime_user_turn_stop_strategies(

api/services/pipecat/service_factory.py

@@ -86,7 +86,7 @@ from pipecat.services.speechmatics.stt import (
     SpeechmaticsSTTService,
     SpeechmaticsSTTSettings,
 )
-from pipecat.services.xai.tts import XAIHttpTTSService, XAITTSSettings
+from pipecat.services.xai.tts import XAITTSService, XAIWebsocketTTSSettings
 from pipecat.transcriptions.language import Language
 from pipecat.utils.text.xml_function_tag_filter import XMLFunctionTagFilter
 
@@ -227,7 +227,6 @@ def create_stt_service(
         # Other models than flux
         # Use language from user config, defaulting to "multi" for multilingual support
         language = getattr(user_config.stt, "language", None) or "multi"
-        logger.debug(f"Using DeepGram Model - {user_config.stt.model}")
         return DeepgramSTTService(
             api_key=user_config.stt.api_key,
             settings=DeepgramSTTSettings(
@@ -818,11 +817,9 @@ def create_tts_service(
                 pipecat_language = Language(language_code)
             except ValueError:
                 pipecat_language = Language.EN
-        return XAIHttpTTSService(
+        return XAITTSService(
             api_key=user_config.tts.api_key,
-            sample_rate=audio_config.transport_out_sample_rate,
-            encoding="pcm",
-            settings=XAITTSSettings(
+            settings=XAIWebsocketTTSSettings(
                 voice=voice,
                 language=pipecat_language,
             ),
@@ -1009,6 +1006,13 @@ def create_realtime_llm_service(user_config, audio_config: "AudioConfig"):
             SessionProperties,
         )
 
+        # Pin the transcription language when configured. Without it the model
+        # auto-detects per utterance, which misfires on short/noisy telephony
+        # audio (e.g. Portuguese transcribed as English or Chinese).
+        transcription_kwargs = {}
+        if language:
+            transcription_kwargs["language"] = language
+
         return DograhOpenAIRealtimeLLMService(
             api_key=api_key,
             settings=DograhOpenAIRealtimeLLMService.Settings(
@@ -1016,7 +1020,9 @@ def create_realtime_llm_service(user_config, audio_config: "AudioConfig"):
                 session_properties=SessionProperties(
                     audio=AudioConfiguration(
                         input=AudioInput(
-                            transcription=InputAudioTranscription(),
+                            transcription=InputAudioTranscription(
+                                **transcription_kwargs
+                            ),
                         ),
                         output=AudioOutput(
                             voice=voice or "alloy",

api/services/telephony/ari_manager.py

@@ -30,8 +30,10 @@ from api.services.call_concurrency import (
     CallConcurrencyLimitError,
     call_concurrency,
 )
+from api.services.organization_preferences import external_pbx_integrations_enabled
 from api.services.quota_service import authorize_workflow_run_start
 from api.services.telephony.call_transfer_manager import get_call_transfer_manager
+from api.services.telephony.providers.ari.external_pbx import create_adapter
 from api.services.telephony.transfer_event_protocol import (
     TransferEvent,
     TransferEventType,
@@ -57,6 +59,7 @@ class ARIConnection:
         app_name: str,
         app_password: str,
         ws_client_name: str = "",
+        external_pbx_config: Optional[dict] = None,
     ):
         self.organization_id = organization_id
         self.telephony_configuration_id = telephony_configuration_id
@@ -64,6 +67,8 @@ class ARIConnection:
         self.app_name = app_name
         self.app_password = app_password
         self.ws_client_name = ws_client_name
+        self.external_pbx_config = external_pbx_config
+        self.external_pbx_adapter = create_adapter(external_pbx_config)
 
         self._ws: Optional[websockets.ClientConnection] = None
         self._task: Optional[asyncio.Task] = None
@@ -288,7 +293,7 @@ class ARIConnection:
         channel_state = channel.get("state", "unknown")
 
         # Log all events for each channel
-        logger.debug(
+        logger.trace(
             f"[ARI EVENT org={self.organization_id}] {event_type}: channel={channel_id}, state={channel_state}"
         )
 
@@ -414,7 +419,7 @@ class ARIConnection:
             )
 
         else:
-            logger.debug(
+            logger.trace(
                 f"[ARI org={self.organization_id}] Event: {event_type} "
                 f"channel={channel_id}"
             )
@@ -429,9 +434,9 @@ class ARIConnection:
             async with session.request(method, url, auth=auth, **kwargs) as response:
                 response_text = await response.text()
                 if response.status not in (200, 201, 204):
-                    logger.error(
+                    logger.warning(
                         f"[ARI org={self.organization_id}] REST API error: "
-                        f"{method} {path} -> {response.status}: {response_text}"
+                        f"{method} {path} {kwargs} -> {response.status}: {response_text}"
                     )
                     return {}
                 if response_text:
@@ -445,6 +450,41 @@ class ARIConnection:
         logger.info(f"[ARI org={self.organization_id}] Answered channel {channel_id}")
         return True
 
+    async def _get_channel_var(self, channel_id: str, variable: str) -> str:
+        """Read a channel variable/function via ARI. Returns '' if unset."""
+        result = await self._ari_request(
+            "GET", f"/channels/{channel_id}/variable", params={"variable": variable}
+        )
+        return (result or {}).get("value", "") or ""
+
+    async def _capture_external_pbx_call(
+        self, channel_id: str, channel_name: str = ""
+    ) -> Optional[dict]:
+        """Capture adapter-defined identity from inbound SIP headers."""
+        if self.external_pbx_adapter is None:
+            return None
+        # PJSIP_HEADER() only works on a PJSIP channel; on any other technology
+        # (Local, WebSocket, etc.) Asterisk returns a 500 ("This function
+        # requires a PJSIP channel"). Non-PJSIP legs carry no SIP headers to
+        # capture anyway, so skip the reads quietly instead of spamming errors.
+        if not channel_name.startswith("PJSIP/"):
+            logger.debug(
+                f"[ARI org={self.organization_id}] Skipping external PBX capture "
+                f"for non-PJSIP channel {channel_id} ({channel_name or 'unknown'})"
+            )
+            return None
+
+        async def read_header(name: str) -> str:
+            return await self._get_channel_var(channel_id, f"PJSIP_HEADER(read,{name})")
+
+        identity = await self.external_pbx_adapter.capture_call_identity(read_header)
+        if identity:
+            logger.info(
+                f"[ARI org={self.organization_id}] Captured "
+                f"{self.external_pbx_adapter.type} call identity for channel {channel_id} identity: {identity}"
+            )
+        return identity
+
     async def _create_external_media(
         self,
         workflow_id: str,
@@ -594,18 +634,11 @@ class ARIConnection:
 
             # 3. Create workflow run
             call_id = channel_id
-            workflow = await db_client.get_workflow(
-                inbound_workflow_id, organization_id=self.organization_id
-            )
-            if not workflow:
-                logger.error(
-                    f"[ARI org={self.organization_id}] Workflow "
-                    f"{inbound_workflow_id} not found"
-                )
-                await call_concurrency.release_slot(concurrency_slot)
-                await self._delete_channel(channel_id)
-                return
             run_inputs = await prepare_workflow_run_inputs(db_client, workflow)
+            # Capture the configured external PBX identity from SIP headers.
+            external_pbx_call = await self._capture_external_pbx_call(
+                channel_id, channel.get("name", "")
+            )
             workflow_run = await db_client.create_workflow_run(
                 name=f"ARI Inbound {caller_number}",
                 workflow_id=inbound_workflow_id,
@@ -618,6 +651,7 @@ class ARIConnection:
                     "direction": "inbound",
                     "provider": "ari",
                     "telephony_configuration_id": self.telephony_configuration_id,
+                    "external_pbx_call": external_pbx_call,
                 },
                 gathered_context={
                     "call_id": call_id,
@@ -1165,6 +1199,7 @@ class ARIManager:
             app_name = config["app_name"]
             app_password = config["app_password"]
             ws_client_name = config["ws_client_name"]
+            external_pbx_config = config.get("external_pbx")
 
             conn = ARIConnection(
                 org_id,
@@ -1173,6 +1208,7 @@ class ARIManager:
                 app_name,
                 app_password,
                 ws_client_name,
+                external_pbx_config,
             )
             key = conn.connection_key
 
@@ -1197,6 +1233,7 @@ class ARIManager:
                     or existing.app_name != app_name
                     or existing.app_password != app_password
                     or existing.ws_client_name != ws_client_name
+                    or existing.external_pbx_config != external_pbx_config
                 ):
                     logger.info(
                         f"[ARI Manager] Config {telephony_configuration_id} "
@@ -1234,6 +1271,11 @@ class ARIManager:
             app_name = credentials.get("app_name")
             app_password = credentials.get("app_password")
             ws_client_name = credentials.get("ws_client_name", "")
+            external_pbx = credentials.get("external_pbx")
+            if external_pbx and not await external_pbx_integrations_enabled(
+                row.organization_id
+            ):
+                external_pbx = None
 
             if not all([ari_endpoint, app_name, app_password]):
                 logger.warning(
@@ -1256,6 +1298,7 @@ class ARIManager:
                     "app_name": app_name,
                     "app_password": app_password,
                     "ws_client_name": ws_client_name,
+                    "external_pbx": external_pbx,
                 }
             )
 

api/services/telephony/base.py

@@ -446,3 +446,18 @@ class TelephonyProvider(ABC):
             True if provider supports call transfers, False otherwise
         """
         pass
+
+    async def transfer_external_pbx_call(
+        self,
+        *,
+        identity: Dict[str, Any],
+        destination: str,
+        field_updates: Optional[Dict[str, str]] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Handle an external-PBX-owned customer leg when one is present.
+
+        Providers without an external PBX return ``None`` so the ordinary
+        telephony transfer path continues unchanged.
+        """
+
+        return None

api/services/telephony/external_pbx.py

@@ -0,0 +1,43 @@
+"""Provider-neutral helpers for external-PBX workflow mappings."""
+
+from __future__ import annotations
+
+from typing import Any, Iterable, Mapping
+
+
+def _read_path(context: Mapping[str, Any], path: str) -> Any:
+    normalized = path.strip()
+    if normalized.startswith("gathered_context."):
+        normalized = normalized.removeprefix("gathered_context.")
+    current: Any = context
+    for part in normalized.split("."):
+        if not isinstance(current, Mapping):
+            return None
+        current = current.get(part)
+    if current is None and "." not in normalized:
+        extracted = context.get("extracted_variables")
+        if isinstance(extracted, Mapping):
+            current = extracted.get(normalized)
+    return current
+
+
+def resolve_external_pbx_field_mappings(
+    gathered_context: Mapping[str, Any] | None,
+    mappings: Iterable[Mapping[str, Any]] | None,
+) -> dict[str, str]:
+    """Return provider field -> non-empty gathered-context value."""
+
+    context = gathered_context or {}
+    resolved: dict[str, str] = {}
+    for mapping in mappings or []:
+        context_path = str(mapping.get("context_path", "")).strip()
+        destination_field = str(mapping.get("destination_field", "")).strip()
+        if not context_path or not destination_field:
+            continue
+        value = _read_path(context, context_path)
+        if value is None:
+            continue
+        text = str(value).strip()
+        if text:
+            resolved[destination_field] = text
+    return resolved

api/services/telephony/providers/ari/__init__.py

@@ -4,8 +4,10 @@ from typing import Any, Dict
 
 from api.services.telephony.registry import (
     ProviderSpec,
+    ProviderUICondition,
     ProviderUIField,
     ProviderUIMetadata,
+    ProviderUIOption,
     register,
 )
 
@@ -20,6 +22,7 @@ def _config_loader(value: Dict[str, Any]) -> Dict[str, Any]:
         "ari_endpoint": value.get("ari_endpoint"),
         "app_name": value.get("app_name"),
         "app_password": value.get("app_password"),
+        "external_pbx": value.get("external_pbx"),
         "from_numbers": value.get("from_numbers", []),
     }
 
@@ -58,6 +61,92 @@ _UI_METADATA = ProviderUIMetadata(
             type="string-array",
             description="SIP extensions/numbers for outbound calls",
         ),
+        ProviderUIField(
+            name="external_pbx.type",
+            label="External PBX Type",
+            type="select",
+            required=False,
+            description=(
+                "Enable PBX-specific call control for calls patched into Dograh "
+                "through this Asterisk configuration."
+            ),
+            options=[ProviderUIOption(value="vicidial", label="VICIdial")],
+            section="External PBX",
+            feature_gate="external_pbx_integrations",
+        ),
+        ProviderUIField(
+            name="external_pbx.agent_api.url",
+            label="Agent API URL",
+            type="text",
+            description="Full VICIdial remote-agent API URL, ending in agc/api.php",
+            placeholder="https://vici.example.com/agc/api.php",
+            visible_when=ProviderUICondition(
+                field="external_pbx.type", equals="vicidial"
+            ),
+            section="External PBX",
+            feature_gate="external_pbx_integrations",
+        ),
+        ProviderUIField(
+            name="external_pbx.agent_api.username",
+            label="Agent API User",
+            type="text",
+            sensitive=True,
+            visible_when=ProviderUICondition(
+                field="external_pbx.type", equals="vicidial"
+            ),
+            section="External PBX",
+            feature_gate="external_pbx_integrations",
+        ),
+        ProviderUIField(
+            name="external_pbx.agent_api.password",
+            label="Agent API Password",
+            type="password",
+            sensitive=True,
+            visible_when=ProviderUICondition(
+                field="external_pbx.type", equals="vicidial"
+            ),
+            section="External PBX",
+            feature_gate="external_pbx_integrations",
+        ),
+        ProviderUIField(
+            name="external_pbx.non_agent_api.url",
+            label="Non-Agent API URL",
+            type="text",
+            required=False,
+            description=(
+                "Optional. Required only when a workflow updates VICIdial lead fields."
+            ),
+            placeholder="https://vici.example.com/vicidial/non_agent_api.php",
+            visible_when=ProviderUICondition(
+                field="external_pbx.type", equals="vicidial"
+            ),
+            section="External PBX",
+            feature_gate="external_pbx_integrations",
+        ),
+        ProviderUIField(
+            name="external_pbx.non_agent_api.username",
+            label="Non-Agent API User",
+            type="text",
+            required=False,
+            sensitive=True,
+            visible_when=ProviderUICondition(
+                field="external_pbx.type", equals="vicidial"
+            ),
+            section="External PBX",
+            feature_gate="external_pbx_integrations",
+        ),
+        ProviderUIField(
+            name="external_pbx.non_agent_api.password",
+            label="Non-Agent API Password",
+            type="password",
+            required=False,
+            sensitive=True,
+            visible_when=ProviderUICondition(
+                field="external_pbx.type", equals="vicidial"
+            ),
+            section="External PBX",
+            feature_gate="external_pbx_integrations",
+        ),
     ],
 )
 

api/services/telephony/providers/ari/config.py

@@ -1,8 +1,69 @@
 """ARI (Asterisk REST Interface) telephony configuration schemas."""
 
-from typing import List, Literal
+from typing import List, Literal, Optional
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+
+class VicidialAgentAPIConfiguration(BaseModel):
+    """VICIdial remote-agent call-control API configuration."""
+
+    url: str = Field(..., min_length=1, description="Full URL to agc/api.php")
+    username: str = Field(..., min_length=1, description="VICIdial agent API user")
+    password: str = Field(..., min_length=1, description="VICIdial agent API password")
+    source: str = Field(default="dograh", description="VICIdial API source tag")
+
+    @field_validator("url")
+    @classmethod
+    def validate_http_url(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped.startswith(("http://", "https://")):
+            raise ValueError("VICIdial agent API URL must use http:// or https://")
+        return stripped
+
+
+class VicidialNonAgentAPIConfiguration(BaseModel):
+    """Optional VICIdial non-agent API configuration for lead updates."""
+
+    url: Optional[str] = Field(default=None, description="Full non_agent_api.php URL")
+    username: Optional[str] = Field(default=None, description="Non-agent API user")
+    password: Optional[str] = Field(default=None, description="Non-agent API password")
+    source: str = Field(default="dograh", description="Non-agent API source tag")
+
+    @field_validator("url")
+    @classmethod
+    def validate_http_url(cls, value: Optional[str]) -> Optional[str]:
+        if value is None:
+            return None
+        stripped = value.strip()
+        if stripped and not stripped.startswith(("http://", "https://")):
+            raise ValueError("VICIdial non-agent API URL must use http:// or https://")
+        return stripped or None
+
+    @model_validator(mode="after")
+    def validate_complete_credentials(self):
+        supplied = [self.url, self.username, self.password]
+        if any(supplied) and not all(supplied):
+            raise ValueError(
+                "VICIdial non-agent API URL, username, and password must be "
+                "configured together"
+            )
+        return self
+
+
+class VicidialExternalPBXConfiguration(BaseModel):
+    """External-PBX configuration used by the VICIdial strategy adapter."""
+
+    type: Literal["vicidial"] = Field(default="vicidial")
+    agent_api: VicidialAgentAPIConfiguration
+    non_agent_api: Optional[VicidialNonAgentAPIConfiguration] = None
+    timeout_seconds: int = Field(default=8, ge=1, le=30)
+
+    @model_validator(mode="after")
+    def drop_empty_non_agent_configuration(self):
+        if self.non_agent_api is not None and not self.non_agent_api.url:
+            self.non_agent_api = None
+        return self
 
 
 class ARIConfigurationRequest(BaseModel):
@@ -20,6 +81,10 @@ class ARIConfigurationRequest(BaseModel):
         default="",
         description="websocket_client.conf connection name for externalMedia (e.g., dograh_staging)",
     )
+    external_pbx: Optional[VicidialExternalPBXConfiguration] = Field(
+        default=None,
+        description="Optional external PBX connected through this Asterisk instance",
+    )
     from_numbers: List[str] = Field(
         default_factory=list,
         description="List of SIP extensions/numbers for outbound calls (optional)",
@@ -34,4 +99,5 @@ class ARIConfigurationResponse(BaseModel):
     app_name: str
     app_password: str  # Masked
     ws_client_name: str = ""
+    external_pbx: Optional[VicidialExternalPBXConfiguration] = None
     from_numbers: List[str]

api/services/telephony/providers/ari/external_pbx/__init__.py

@@ -0,0 +1,14 @@
+"""External-PBX adapter entrypoint for the ARI provider."""
+
+from .base import ExternalPBXAdapter, ExternalPBXResult
+from .registry import create_adapter, register_adapter, registered_adapter_types
+from .vicidial import VicidialAdapter
+
+register_adapter("vicidial", VicidialAdapter)
+
+__all__ = [
+    "ExternalPBXAdapter",
+    "ExternalPBXResult",
+    "create_adapter",
+    "registered_adapter_types",
+]

api/services/telephony/providers/ari/external_pbx/base.py

@@ -0,0 +1,44 @@
+"""Contracts for PBXs that hand a customer leg to Dograh through Asterisk."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Awaitable, Callable, Mapping
+
+HeaderReader = Callable[[str], Awaitable[str]]
+
+
+@dataclass(frozen=True)
+class ExternalPBXResult:
+    ok: bool
+    action: str
+    message: str
+
+
+class ExternalPBXAdapter(ABC):
+    """PBX-specific operations; ARI continues to own only Dograh's local leg."""
+
+    type: str
+
+    @abstractmethod
+    async def capture_call_identity(
+        self, read_header: HeaderReader
+    ) -> dict[str, str] | None:
+        """Read a stable upstream-call identity from inbound SIP headers."""
+
+    @abstractmethod
+    async def hangup(self, identity: Mapping[str, str]) -> ExternalPBXResult:
+        """Hang up the customer leg owned by the external PBX."""
+
+    @abstractmethod
+    async def transfer(
+        self, identity: Mapping[str, str], destination: str
+    ) -> ExternalPBXResult:
+        """Transfer the customer leg to a PBX-native destination."""
+
+    @abstractmethod
+    async def update_fields(
+        self, identity: Mapping[str, str], fields: Mapping[str, str]
+    ) -> ExternalPBXResult:
+        """Update provider-native fields associated with the call."""

api/services/telephony/providers/ari/external_pbx/registry.py

@@ -0,0 +1,34 @@
+"""Registry for drop-in external-PBX adapters."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from .base import ExternalPBXAdapter
+
+AdapterFactory = Callable[[dict[str, Any]], ExternalPBXAdapter]
+_FACTORIES: dict[str, AdapterFactory] = {}
+
+
+def register_adapter(pbx_type: str, factory: AdapterFactory) -> None:
+    normalized = pbx_type.strip().lower()
+    if not normalized:
+        raise ValueError("External PBX type cannot be empty")
+    if normalized in _FACTORIES:
+        raise ValueError(f"External PBX adapter already registered: {normalized}")
+    _FACTORIES[normalized] = factory
+
+
+def create_adapter(config: dict[str, Any] | None) -> ExternalPBXAdapter | None:
+    if not config:
+        return None
+    pbx_type = str(config.get("type", "")).strip().lower()
+    factory = _FACTORIES.get(pbx_type)
+    if factory is None:
+        raise ValueError(f"Unsupported external PBX type: {pbx_type or '<empty>'}")
+    return factory(config)
+
+
+def registered_adapter_types() -> tuple[str, ...]:
+    return tuple(sorted(_FACTORIES))

api/services/telephony/providers/ari/external_pbx/vicidial.py

@@ -0,0 +1,170 @@
+"""VICIdial implementation of external-PBX call and lead operations."""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Mapping
+
+import aiohttp
+from loguru import logger
+
+from .base import ExternalPBXAdapter, ExternalPBXResult, HeaderReader
+
+_LEAD_FIELD_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,63}$")
+_RESERVED_LEAD_FIELDS = frozenset({"source", "user", "pass", "function", "lead_id"})
+
+
+class VicidialAdapter(ExternalPBXAdapter):
+    type = "vicidial"
+
+    def __init__(self, config: dict[str, Any]):
+        agent_api = config.get("agent_api") or {}
+        non_agent_api = config.get("non_agent_api") or {}
+        self._agent_url = str(agent_api.get("url", "")).strip()
+        self._agent_user = str(agent_api.get("username", "")).strip()
+        self._agent_password = str(agent_api.get("password", ""))
+        self._agent_source = str(agent_api.get("source", "dograh")).strip()
+        self._non_agent_url = str(non_agent_api.get("url", "")).strip()
+        self._non_agent_user = str(non_agent_api.get("username", "")).strip()
+        self._non_agent_password = str(non_agent_api.get("password", ""))
+        self._non_agent_source = str(non_agent_api.get("source", "dograh")).strip()
+        self._timeout = aiohttp.ClientTimeout(
+            total=min(max(int(config.get("timeout_seconds", 8)), 1), 30)
+        )
+
+    async def capture_call_identity(
+        self, read_header: HeaderReader
+    ) -> dict[str, str] | None:
+        callerid = (await read_header("X-VICIDIAL-callerid")).strip()
+        if not callerid:
+            return None
+        return {
+            "type": self.type,
+            "callerid": callerid,
+            "agent_user": (await read_header("X-VICIDIAL-user")).strip(),
+            "lead_id": (await read_header("X-VICIDIAL-lead_id")).strip(),
+            "campaign_id": (await read_header("X-VICIDIAL-campaign_id")).strip(),
+            "ingroup_id": (await read_header("X-VICIDIAL-ingroup_id")).strip(),
+        }
+
+    async def _agent_call_control(
+        self, identity: Mapping[str, str], stage: str, **extra: str
+    ) -> ExternalPBXResult:
+        if not all([self._agent_url, self._agent_user, self._agent_password]):
+            return ExternalPBXResult(
+                False, stage.lower(), "Agent API is not configured"
+            )
+        if not identity.get("callerid") or not identity.get("agent_user"):
+            return ExternalPBXResult(
+                False, stage.lower(), "VICIdial call identity is incomplete"
+            )
+        params = {
+            "source": self._agent_source,
+            "user": self._agent_user,
+            "pass": self._agent_password,
+            "agent_user": identity["agent_user"],
+            "function": "ra_call_control",
+            "stage": stage,
+            "value": identity["callerid"],
+            **extra,
+        }
+        try:
+            async with aiohttp.ClientSession(timeout=self._timeout) as session:
+                async with session.get(self._agent_url, params=params) as response:
+                    response_text = (await response.text()).strip()
+                    ok = response.status == 200 and response_text.startswith("SUCCESS")
+            logger.info(
+                "[VICIdial] ra_call_control completed "
+                f"stage={stage} status={response.status} ok={ok}"
+            )
+            return ExternalPBXResult(
+                ok,
+                stage.lower(),
+                "VICIdial accepted the operation"
+                if ok
+                else "VICIdial rejected the operation",
+            )
+        except Exception as exc:
+            logger.error(f"[VICIdial] ra_call_control failed stage={stage}: {exc}")
+            return ExternalPBXResult(
+                False, stage.lower(), "VICIdial API request failed"
+            )
+
+    async def hangup(self, identity: Mapping[str, str]) -> ExternalPBXResult:
+        return await self._agent_call_control(identity, "HANGUP")
+
+    async def transfer(
+        self, identity: Mapping[str, str], destination: str
+    ) -> ExternalPBXResult:
+        choice = destination.strip()
+        if choice.lower() == "source":
+            choice = str(identity.get("ingroup_id", "")).strip()
+        if not choice:
+            return ExternalPBXResult(
+                False, "ingrouptransfer", "No VICIdial in-group was resolved"
+            )
+        return await self._agent_call_control(
+            identity, "INGROUPTRANSFER", ingroup_choices=choice
+        )
+
+    async def update_fields(
+        self, identity: Mapping[str, str], fields: Mapping[str, str]
+    ) -> ExternalPBXResult:
+        if not fields:
+            return ExternalPBXResult(True, "update_lead", "No lead fields configured")
+        if not all(
+            [self._non_agent_url, self._non_agent_user, self._non_agent_password]
+        ):
+            return ExternalPBXResult(
+                False, "update_lead", "Non-agent API is not configured"
+            )
+        lead_id = str(identity.get("lead_id", "")).strip()
+        if not lead_id:
+            return ExternalPBXResult(
+                False, "update_lead", "No VICIdial lead ID captured"
+            )
+
+        safe_fields: dict[str, str] = {}
+        for key, value in fields.items():
+            normalized = str(key).strip()
+            if (
+                not _LEAD_FIELD_RE.fullmatch(normalized)
+                or normalized.lower() in _RESERVED_LEAD_FIELDS
+            ):
+                logger.warning(
+                    f"[VICIdial] Ignoring invalid lead field name: {normalized!r}"
+                )
+                continue
+            safe_fields[normalized] = str(value)
+        if not safe_fields:
+            return ExternalPBXResult(
+                False, "update_lead", "No valid lead fields resolved"
+            )
+
+        params = {
+            **safe_fields,
+            "source": self._non_agent_source,
+            "user": self._non_agent_user,
+            "pass": self._non_agent_password,
+            "function": "update_lead",
+            "lead_id": lead_id,
+        }
+        try:
+            async with aiohttp.ClientSession(timeout=self._timeout) as session:
+                async with session.get(self._non_agent_url, params=params) as response:
+                    response_text = (await response.text()).strip()
+                    ok = response.status == 200 and response_text.startswith("SUCCESS")
+            logger.info(
+                "[VICIdial] update_lead completed "
+                f"status={response.status} ok={ok} field_count={len(safe_fields)}"
+            )
+            return ExternalPBXResult(
+                ok,
+                "update_lead",
+                "VICIdial lead updated" if ok else "VICIdial rejected the lead update",
+            )
+        except Exception as exc:
+            logger.error(f"[VICIdial] update_lead failed: {exc}")
+            return ExternalPBXResult(
+                False, "update_lead", "VICIdial API request failed"
+            )

api/services/telephony/providers/ari/provider.py

@@ -20,6 +20,7 @@ from api.services.telephony.base import (
     NormalizedInboundData,
     TelephonyProvider,
 )
+from api.services.telephony.providers.ari.external_pbx import create_adapter
 
 if TYPE_CHECKING:
     from fastapi import WebSocket
@@ -51,6 +52,7 @@ class ARIProvider(TelephonyProvider):
         self.app_name = config.get("app_name", "")
         self.app_password = config.get("app_password", "")
         self.from_numbers = config.get("from_numbers", [])
+        self.external_pbx_adapter = create_adapter(config.get("external_pbx"))
 
         if isinstance(self.from_numbers, str):
             self.from_numbers = [self.from_numbers]
@@ -363,6 +365,53 @@ class ARIProvider(TelephonyProvider):
         """ARI supports call transfers via bridge manipulation."""
         return True
 
+    async def transfer_external_pbx_call(
+        self,
+        *,
+        identity: Dict[str, Any],
+        destination: str,
+        field_updates: Optional[Dict[str, str]] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Delegate a PBX-owned customer leg to the configured adapter."""
+
+        adapter = self.external_pbx_adapter
+        if adapter is None or not identity:
+            return None
+        identity_type = identity.get("type") or identity.get("provider")
+        if identity_type != adapter.type:
+            logger.warning(
+                "[ARI External PBX] Captured identity does not match configured "
+                f"adapter: identity={identity_type!r} adapter={adapter.type!r}"
+            )
+            return {
+                "status": "failed",
+                "action": "external_pbx_transfer",
+                "message": "The external PBX call identity is invalid.",
+                "reason": "external_pbx_identity_mismatch",
+            }
+
+        update_result = None
+        if field_updates:
+            update_result = await adapter.update_fields(identity, field_updates)
+            if not update_result.ok:
+                logger.warning(
+                    "[ARI External PBX] Field update failed; continuing transfer "
+                    f"adapter={adapter.type} message={update_result.message}"
+                )
+
+        transfer_result = await adapter.transfer(identity, destination)
+        return {
+            "status": "success" if transfer_result.ok else "failed",
+            "action": "external_pbx_transfer",
+            "message": (
+                "Transferring your call now."
+                if transfer_result.ok
+                else "I'm sorry, I couldn't complete the transfer."
+            ),
+            "reason": None if transfer_result.ok else "external_pbx_transfer_failed",
+            "field_update_ok": update_result.ok if update_result else None,
+        }
+
     async def transfer_call(
         self,
         destination: str,

api/services/telephony/providers/ari/strategies.py

@@ -3,11 +3,14 @@
 This module contains the business logic for Asterisk ARI call operations.
 """
 
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any, Dict
 
 from loguru import logger
 from pipecat.serializers.call_strategies import HangupStrategy, TransferStrategy
 
+if TYPE_CHECKING:
+    from .external_pbx import ExternalPBXAdapter
+
 
 class ARIBridgeSwapStrategy(TransferStrategy):
     """Implements bridge swap transfer for Asterisk ARI.
@@ -200,6 +203,9 @@ class ARIBridgeSwapStrategy(TransferStrategy):
 class ARIHangupStrategy(HangupStrategy):
     """Implements hangup for Asterisk ARI channels."""
 
+    def __init__(self, external_pbx_adapter: "ExternalPBXAdapter | None" = None):
+        self._external_pbx_adapter = external_pbx_adapter
+
     async def execute_hangup(self, context: Dict[str, Any]) -> bool:
         """Hang up the Asterisk channel via ARI REST API."""
         try:
@@ -217,6 +223,10 @@ class ARIHangupStrategy(HangupStrategy):
                 )
                 return False
 
+            # The external PBX owns the real customer leg. End it before the
+            # local ARI leg so its conference teardown cannot race our SIP BYE.
+            await self._terminate_external_pbx_if_any(channel_id)
+
             endpoint = f"{ari_endpoint}/ari/channels/{channel_id}"
             auth = BasicAuth(app_name, app_password)
 
@@ -243,3 +253,76 @@ class ARIHangupStrategy(HangupStrategy):
         except Exception as e:
             logger.exception(f"Failed to hang up Asterisk channel: {e}")
             return False
+
+    async def _terminate_external_pbx_if_any(self, channel_id: str) -> None:
+        """If configured, hang up the external PBX's customer leg first.
+
+        Reuses the same channel->run lookup the transfer strategy uses
+        (Redis ``ari:channel:{id}`` -> run_id -> ``initial_context``). Best-effort:
+        never blocks dograh's own hangup if the upstream call fails.
+        """
+        redis = None
+        try:
+            import redis.asyncio as aioredis
+
+            from api.constants import REDIS_URL
+            from api.db import db_client
+
+            redis = aioredis.from_url(REDIS_URL, decode_responses=True)
+            run_id = await redis.get(f"ari:channel:{channel_id}")
+            if not run_id:
+                return
+            run = await db_client.get_workflow_run_by_id(int(run_id))
+            if not run:
+                return
+            identity = (run.initial_context or {}).get("external_pbx_call")
+            # Read the legacy key for calls created before this refactor.
+            identity = identity or (run.initial_context or {}).get("upstream_pbx")
+            # If the call was already transferred to the external PBX, the customer
+            # leg has moved on -- do NOT hang it up (that would drop the transferred
+            # customer); just let dograh's own legs tear down below.
+            transferred = (run.gathered_context or {}).get("external_pbx_transferred")
+            transferred = transferred or (run.gathered_context or {}).get(
+                "upstream_transferred"
+            )
+            if identity and not transferred and self._external_pbx_adapter:
+                from api.services.telephony.external_pbx import (
+                    resolve_external_pbx_field_mappings,
+                )
+
+                workflow_configurations = (
+                    await db_client.get_workflow_run_configurations(
+                        int(run_id), run.workflow.organization_id
+                    )
+                )
+                field_updates = resolve_external_pbx_field_mappings(
+                    run.gathered_context,
+                    workflow_configurations.get("external_pbx_field_mappings", []),
+                )
+                if field_updates:
+                    update_result = await self._external_pbx_adapter.update_fields(
+                        identity, field_updates
+                    )
+                    if not update_result.ok:
+                        logger.warning(
+                            "[ARI Hangup] External PBX field update failed; "
+                            f"continuing hangup: {update_result.message}"
+                        )
+                logger.info(
+                    "[ARI Hangup] External PBX call "
+                    f"({self._external_pbx_adapter.type}); "
+                    f"terminating customer leg via API before dropping dograh leg"
+                )
+                result = await self._external_pbx_adapter.hangup(identity)
+                if not result.ok:
+                    logger.warning(
+                        f"[ARI Hangup] External PBX rejected hangup: {result.message}"
+                    )
+        except Exception as e:
+            logger.error(f"[ARI Hangup] external PBX terminate check failed: {e}")
+        finally:
+            if redis is not None:
+                try:
+                    await redis.aclose()
+                except Exception as e:
+                    logger.warning(f"[ARI Hangup] failed to close Redis client: {e}")

api/services/telephony/providers/ari/transport.py

@@ -11,6 +11,7 @@ from api.services.pipecat.audio_mixer import build_audio_out_mixer
 from api.services.pipecat.transport_params import realtime_param_overrides
 from api.services.telephony.factory import load_credentials_for_transport
 
+from .external_pbx import create_adapter
 from .serializers import AsteriskFrameSerializer
 from .strategies import ARIBridgeSwapStrategy, ARIHangupStrategy
 
@@ -47,7 +48,9 @@ async def create_transport(
         app_name=app_name,
         app_password=app_password,
         transfer_strategy=ARIBridgeSwapStrategy(),
-        hangup_strategy=ARIHangupStrategy(),
+        hangup_strategy=ARIHangupStrategy(
+            external_pbx_adapter=create_adapter(config.get("external_pbx"))
+        ),
         params=AsteriskFrameSerializer.InputParams(
             asterisk_sample_rate=audio_config.transport_in_sample_rate,
             sample_rate=audio_config.pipeline_sample_rate,

api/services/telephony/registry.py

@@ -30,6 +30,22 @@ if TYPE_CHECKING:
     from api.services.telephony.base import TelephonyProvider
 
 
+@dataclass(frozen=True)
+class ProviderUIOption:
+    """One selectable value for a provider configuration field."""
+
+    value: str
+    label: str
+
+
+@dataclass(frozen=True)
+class ProviderUICondition:
+    """Display a field only when another form value matches ``equals``."""
+
+    field: str
+    equals: Any
+
+
 @dataclass(frozen=True)
 class ProviderUIField:
     """One form field for the telephony configuration UI.
@@ -46,6 +62,10 @@ class ProviderUIField:
     sensitive: bool = False  # If true, mask when displaying stored value
     description: Optional[str] = None
     placeholder: Optional[str] = None
+    options: Optional[List[ProviderUIOption]] = None
+    visible_when: Optional[ProviderUICondition] = None
+    section: Optional[str] = None
+    feature_gate: Optional[str] = None
 
 
 @dataclass(frozen=True)

api/services/tool_management.py

@@ -20,6 +20,7 @@ from api.schemas.tool import (
     McpRefreshResponse,
     ToolResponse,
 )
+from api.services.organization_preferences import external_pbx_integrations_enabled
 from api.services.posthog_client import capture_event
 from api.services.workflow.mcp_tool_session import discover_mcp_tools
 from api.services.workflow.tools.mcp_tool import (
@@ -119,6 +120,41 @@ async def validate_tool_credential_references(
             )
 
 
+async def validate_external_pbx_tool_definition(
+    definition: dict[str, Any],
+    *,
+    organization_id: int,
+    existing_definition: Optional[dict[str, Any]] = None,
+) -> None:
+    """Enforce the org feature gate for context-to-in-group routing."""
+
+    config = definition.get("config")
+    existing_config = (existing_definition or {}).get("config")
+    uses_external_pbx = (
+        isinstance(config, dict)
+        and config.get("destination_source") == "context_mapping"
+    )
+    existing_uses_external_pbx = (
+        isinstance(existing_config, dict)
+        and existing_config.get("destination_source") == "context_mapping"
+    )
+    if not uses_external_pbx and not existing_uses_external_pbx:
+        return
+    if await external_pbx_integrations_enabled(organization_id):
+        return
+    if isinstance(existing_config, dict) and existing_config == config:
+        # Preserve a hidden existing mapping while the feature is disabled.
+        return
+    raise ToolManagementError(
+        "external_pbx_feature_disabled",
+        (
+            "External PBX integrations are disabled for this organization. "
+            "Enable them in Platform Settings before configuring in-group routing."
+        ),
+        status_code=403,
+    )
+
+
 async def populate_discovered_tools(
     definition: dict[str, Any], *, organization_id: int
 ) -> dict[str, Any]:
@@ -168,6 +204,9 @@ async def create_tool_for_user(
         )
 
     definition = request.definition.model_dump()
+    await validate_external_pbx_tool_definition(
+        definition, organization_id=user.selected_organization_id
+    )
     await validate_tool_credential_references(
         definition, organization_id=user.selected_organization_id
     )

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

api/services/workflow/pipecat_engine.py

@@ -98,6 +98,10 @@ class PipecatEngine:
         self._gathered_context: dict = {}
         self._user_response_timeout_task: Optional[asyncio.Task] = None
         self._pending_extraction_tasks: set[asyncio.Task] = set()
+        # True once a final (synchronous) extraction has run, so the end-of-call
+        # and upstream-transfer paths don't redundantly re-extract the same
+        # terminal state.
+        self._final_extraction_done: bool = False
 
         # Will be set later in initialize() when we have
         # access to _context
@@ -507,6 +511,29 @@ class PipecatEngine:
                 f"Incomplete: {incomplete}"
             )
 
+    async def perform_final_variable_extraction(self) -> None:
+        """Flush in-flight + current-node variable extraction synchronously.
+
+        Awaits any background extractions still running from previous nodes,
+        then runs the current node's extraction inline so callers that need the
+        freshest extracted variables before acting can rely on them -- e.g.
+        end_call_with_reason before disposing the call, or an external-PBX
+        transfer that maps extracted variables into a provider lead update
+        call before handing the customer off.
+
+        Idempotent: only the first call does work. The external-PBX transfer
+        runs this just before forwarding update_lead, so the subsequent
+        end_call_with_reason would otherwise re-extract the same terminal state.
+        """
+        if self._final_extraction_done:
+            logger.debug("Final variable extraction already performed; skipping")
+            return
+        self._final_extraction_done = True
+        await self._await_pending_extractions()
+        await self._perform_variable_extraction_if_needed(
+            self._current_node, run_in_background=False
+        )
+
     async def _setup_llm_context(self, node: Node) -> None:
         """Common method to set up LLM context"""
         # Set OTel span name for tracing
@@ -742,13 +769,8 @@ class PipecatEngine:
             EndTaskReason.PIPELINE_ERROR.value,
             EndTaskReason.VOICEMAIL_DETECTED.value,
         ):
-            # Await any in-flight background extractions from previous nodes
-            await self._await_pending_extractions()
-
-            # Perform final variable extraction synchronously before ending
-            await self._perform_variable_extraction_if_needed(
-                self._current_node, run_in_background=False
-            )
+            # Flush in-flight + current-node extractions synchronously before ending
+            await self.perform_final_variable_extraction()
 
         frame_to_push = (
             CancelFrame(reason=reason) if abort_immediately else EndFrame(reason=reason)
@@ -766,6 +788,20 @@ class PipecatEngine:
                 call_tags.append(call_disposition)
             self._gathered_context["call_tags"] = call_tags
 
+        # Hangup strategies run while serializing the terminal frame. Persist
+        # the final extracted values first so external-PBX adapters can apply
+        # workflow lead-field mappings before terminating the customer leg.
+        try:
+            await db_client.update_workflow_run(
+                run_id=self._workflow_run_id,
+                gathered_context=self._gathered_context,
+            )
+        except Exception as exc:
+            # Call teardown must never be held hostage by an enrichment write.
+            logger.warning(
+                f"Could not persist final gathered context before hangup: {exc}"
+            )
+
         logger.debug(
             f"Finishing run with reason: {reason}, disposition: {call_disposition} "
             f"queueing frame {frame_to_push}"

api/services/workflow/pipecat_engine_custom_tools.py

@@ -24,6 +24,7 @@ from api.db import db_client
 from api.enums import ToolCategory, WorkflowRunMode
 from api.services.pipecat.audio_playback import play_audio, play_audio_loop
 from api.services.telephony.call_transfer_manager import get_call_transfer_manager
+from api.services.telephony.external_pbx import resolve_external_pbx_field_mappings
 from api.services.telephony.factory import get_telephony_provider_for_run
 from api.services.telephony.transfer_event_protocol import TransferContext
 from api.services.workflow.tools.calculator import get_calculator_tools, safe_calculator
@@ -609,22 +610,22 @@ class CustomToolManager:
                     )
                     return
 
+                external_pbx_call = (
+                    getattr(workflow_run, "initial_context", None) or {}
+                ).get("external_pbx_call")
+                # Compatibility for calls that started before the migration.
+                external_pbx_call = external_pbx_call or (
+                    getattr(workflow_run, "initial_context", None) or {}
+                ).get("upstream_pbx")
+                if external_pbx_call:
+                    # Context-to-in-group and lead-field mappings must see the
+                    # final conversation-derived values.
+                    await self._engine.perform_final_variable_extraction()
+
                 resolver = config.get("resolver") if isinstance(config, dict) else None
                 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(
@@ -634,7 +635,6 @@ class CustomToolManager:
                             persist_to_logs=True,
                         )
                     )
-                    self._engine._queued_speech_mute_state = "waiting"
 
                 try:
                     resolved_transfer = await resolve_transfer_config(
@@ -649,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.",
@@ -661,6 +660,25 @@ class CustomToolManager:
                     )
                     return
 
+                if (
+                    resolved_transfer.source == "context_mapping"
+                    and not external_pbx_call
+                ):
+                    await self._handle_transfer_result(
+                        {
+                            "status": "failed",
+                            "message": (
+                                "This call did not arrive through the configured "
+                                "external PBX."
+                            ),
+                            "action": "transfer_failed",
+                            "reason": "external_pbx_call_required",
+                        },
+                        function_call_params,
+                        properties,
+                    )
+                    return
+
                 # Validate destination phone number
                 if not destination or not destination.strip():
                     validation_error_result = {
@@ -669,12 +687,15 @@ 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
                     )
                     return
 
+                provider = await get_telephony_provider_for_run(
+                    workflow_run, organization_id
+                )
+
                 if resolved_transfer.message:
                     await self._engine.task.queue_frame(
                         TTSSpeakFrame(
@@ -683,15 +704,54 @@ 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 = (
+                        await db_client.get_workflow_run_configurations(
+                            self._engine._workflow_run_id, organization_id
+                        )
+                    )
+                    field_updates = resolve_external_pbx_field_mappings(
+                        self._engine._gathered_context,
+                        workflow_configurations.get("external_pbx_field_mappings", []),
+                    )
+                    external_result = await provider.transfer_external_pbx_call(
+                        identity=external_pbx_call,
+                        destination=destination,
+                        field_updates=field_updates,
+                    )
+                    if external_result is not None:
+                        if external_result.get("status") == "success":
+                            self._engine._gathered_context[
+                                "external_pbx_transferred"
+                            ] = True
+                            await db_client.update_workflow_run(
+                                run_id=self._engine._workflow_run_id,
+                                gathered_context={"external_pbx_transferred": True},
+                            )
+                            await function_call_params.result_callback(
+                                external_result, properties=properties
+                            )
+                            # Let VICIdial redirect the customer out of its
+                            # conference before Dograh tears down the local leg.
+                            await asyncio.sleep(4)
+                            await self._engine.end_call_with_reason(
+                                EndTaskReason.END_CALL_TOOL_REASON.value,
+                                abort_immediately=True,
+                            )
+                        else:
+                            await self._handle_transfer_result(
+                                {
+                                    **external_result,
+                                    "action": "transfer_failed",
+                                },
+                                function_call_params,
+                                properties,
+                            )
+                        return
 
-                provider = await get_telephony_provider_for_run(
-                    workflow_run, organization_id
-                )
                 if not provider.supports_transfers() or not provider.validate_config():
                     validation_error_result = {
                         "status": "failed",
@@ -699,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
                     )
@@ -750,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",
@@ -850,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 = {

api/services/workflow/tools/custom_tool.py

@@ -8,6 +8,7 @@ import httpx
 from loguru import logger
 
 from api.db import db_client
+from api.services.configuration.masking import mask_key
 from api.utils.credential_auth import build_auth_header
 from api.utils.template_renderer import render_template
 
@@ -21,6 +22,19 @@ TYPE_MAP = {
 }
 
 
+def serialize_query_params(arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """JSON-stringify dict/list values so they're safe to pass as query params.
+
+    httpx (and query strings in general) only support primitive param values.
+    Object/array-typed tool arguments must be serialized before going out as
+    GET/DELETE query params, otherwise httpx raises a TypeError.
+    """
+    return {
+        k: json.dumps(v) if isinstance(v, (dict, list)) else v
+        for k, v in arguments.items()
+    }
+
+
 def tool_to_function_schema(tool: Any) -> Dict[str, Any]:
     """Convert a ToolModel to an LLM function schema.
 
@@ -224,7 +238,9 @@ async def execute_http_tool(
     arguments: Dict[str, Any],
     call_context_vars: Optional[Dict[str, Any]] = None,
     gathered_context_vars: Optional[Dict[str, Any]] = None,
+    preset_params: Optional[Dict[str, Any]] = None,
     organization_id: Optional[int] = None,
+    include_request_headers: bool = False,
 ) -> Dict[str, Any]:
     """Execute an HTTP API tool.
 
@@ -233,7 +249,11 @@ async def execute_http_tool(
         arguments: Arguments passed by the LLM (parameter name -> value)
         call_context_vars: Initial context variables available at runtime
         gathered_context_vars: Variables extracted during the conversation
+        preset_params: Pre-resolved preset parameter values. Used by the test
+            endpoint; live calls omit this so configured templates are resolved.
         organization_id: Organization ID for credential lookup
+        include_request_headers: Include a client-safe header preview in the result.
+            Headers supplied by a stored credential are masked.
 
     Returns:
         Result dict with response data or error
@@ -248,7 +268,9 @@ async def execute_http_tool(
     # Get headers from config
     headers = dict(config.get("headers", {}) or {})
 
-    # Add auth header if credential is configured
+    # Add auth header if credential is configured. Keep track of which headers
+    # came from the credential so only those values are masked in test previews.
+    credential_headers: Dict[str, str] = {}
     credential_uuid = config.get("credential_uuid")
     if credential_uuid and organization_id:
         try:
@@ -256,8 +278,8 @@ async def execute_http_tool(
                 credential_uuid, organization_id
             )
             if credential:
-                auth_header = build_auth_header(credential)
-                headers.update(auth_header)
+                credential_headers = build_auth_header(credential)
+                headers.update(credential_headers)
                 logger.debug(f"Applied credential '{credential.name}' to tool request")
             else:
                 logger.warning(
@@ -266,17 +288,31 @@ async def execute_http_tool(
         except Exception as e:
             logger.error(f"Failed to fetch credential for tool '{tool.name}': {e}")
 
+    request_headers: Dict[str, str] = {}
+    if include_request_headers:
+        request_headers = {str(name): str(value) for name, value in headers.items()}
+        for header_name, header_value in credential_headers.items():
+            request_headers[header_name] = mask_key(str(header_value))
+
+    def build_result(result: Dict[str, Any]) -> Dict[str, Any]:
+        if include_request_headers:
+            return {**result, "request_headers": request_headers}
+        return result
+
     # Get timeout
     timeout_ms = config.get("timeout_ms", 5000)
     timeout_seconds = timeout_ms / 1000
 
-    try:
-        preset_arguments = _resolve_preset_parameters(
-            config, call_context_vars, gathered_context_vars
-        )
-    except ValueError as e:
-        logger.error(f"Custom tool '{tool.name}' preset parameter error: {e}")
-        return {"status": "error", "error": str(e)}
+    if preset_params is None:
+        try:
+            preset_arguments = _resolve_preset_parameters(
+                config, call_context_vars, gathered_context_vars
+            )
+        except ValueError as e:
+            logger.error(f"Custom tool '{tool.name}' preset parameter error: {e}")
+            return build_result({"status": "error", "error": str(e)})
+    else:
+        preset_arguments = dict(preset_params)
 
     resolved_arguments = {**(arguments or {}), **preset_arguments}
 
@@ -286,7 +322,7 @@ async def execute_http_tool(
     if method in ("POST", "PUT", "PATCH"):
         body = resolved_arguments
     elif method in ("GET", "DELETE") and resolved_arguments:
-        params = resolved_arguments
+        params = serialize_query_params(resolved_arguments)
 
     logger.info(
         f"Executing custom tool '{tool.name}' ({tool.tool_uuid}): {method} {url}"
@@ -322,23 +358,29 @@ async def execute_http_tool(
             logger.debug(
                 f"Custom tool '{tool.name}' completed with status {response.status_code}"
             )
-            return result
+            return build_result(result)
 
     except httpx.TimeoutException:
         logger.error(f"Custom tool '{tool.name}' timed out after {timeout_seconds}s")
-        return {
-            "status": "error",
-            "error": f"Request timed out after {timeout_seconds} seconds",
-        }
+        return build_result(
+            {
+                "status": "error",
+                "error": f"Request timed out after {timeout_seconds} seconds",
+            }
+        )
     except httpx.RequestError as e:
         logger.error(f"Custom tool '{tool.name}' request failed: {e}")
-        return {
-            "status": "error",
-            "error": f"Request failed: {str(e)}",
-        }
+        return build_result(
+            {
+                "status": "error",
+                "error": f"Request failed: {str(e)}",
+            }
+        )
     except Exception as e:
         logger.error(f"Custom tool '{tool.name}' execution failed: {e}")
-        return {
-            "status": "error",
-            "error": f"Tool execution failed: {str(e)}",
-        }
+        return build_result(
+            {
+                "status": "error",
+                "error": f"Tool execution failed: {str(e)}",
+            }
+        )

api/services/workflow/tools/transfer_resolver.py

@@ -11,6 +11,7 @@ import httpx
 from loguru import logger
 
 from api.db import db_client
+from api.services.organization_preferences import external_pbx_integrations_enabled
 from api.services.workflow.tools.custom_tool import _resolve_preset_parameters
 from api.utils.credential_auth import build_auth_header
 from api.utils.template_renderer import render_template
@@ -123,6 +124,70 @@ def _resolve_static_transfer(
     )
 
 
+def _context_value(
+    path: str,
+    call_context_vars: Optional[Dict[str, Any]],
+    gathered_context_vars: Optional[Dict[str, Any]],
+) -> Any:
+    initial = call_context_vars or {}
+    gathered = gathered_context_vars or {}
+    normalized = path.strip()
+    if normalized.startswith("initial_context."):
+        current: Any = initial
+        parts = normalized.removeprefix("initial_context.").split(".")
+    elif normalized.startswith("gathered_context."):
+        current = gathered
+        parts = normalized.removeprefix("gathered_context.").split(".")
+    else:
+        current = gathered
+        parts = normalized.split(".")
+
+    for part in parts:
+        if not isinstance(current, dict):
+            return None
+        current = current.get(part)
+    if current is None and "." not in normalized:
+        extracted = gathered.get("extracted_variables")
+        if isinstance(extracted, dict):
+            current = extracted.get(normalized)
+    return current
+
+
+def _resolve_context_mapping_transfer(
+    config: dict[str, Any],
+    call_context_vars: Optional[Dict[str, Any]],
+    gathered_context_vars: Optional[Dict[str, Any]],
+) -> ResolvedTransferConfig:
+    mapping = config.get("context_mapping")
+    if not isinstance(mapping, dict):
+        raise TransferResolutionError(
+            "invalid_context_mapping", "Transfer context mapping is missing"
+        )
+    path = str(mapping.get("context_path", "")).strip()
+    raw_value = _context_value(path, call_context_vars, gathered_context_vars)
+    match_value = "" if raw_value is None else str(raw_value).strip().casefold()
+    destination = ""
+    for route in mapping.get("routes") or []:
+        if not isinstance(route, dict):
+            continue
+        if str(route.get("context_value", "")).strip().casefold() == match_value:
+            destination = str(route.get("destination", "")).strip()
+            break
+    if not destination:
+        destination = str(mapping.get("fallback_destination") or "").strip()
+    if not destination:
+        raise TransferResolutionError(
+            "no_context_mapping_match",
+            f"No destination mapping matched gathered context path '{path}'",
+        )
+    return ResolvedTransferConfig(
+        destination=destination,
+        timeout_seconds=_base_timeout(config),
+        source="context_mapping",
+        metadata={"context_path": path, "matched": bool(match_value)},
+    )
+
+
 def _resolver_arguments(
     *,
     resolver: dict[str, Any],
@@ -276,6 +341,25 @@ async def resolve_transfer_config(
 ) -> ResolvedTransferConfig:
     """Resolve transfer destination and options for a transfer tool call."""
 
+    destination_source = config.get("destination_source", "static")
+    if destination_source == "context_mapping":
+        if not organization_id or not await external_pbx_integrations_enabled(
+            organization_id
+        ):
+            raise TransferResolutionError(
+                "external_pbx_feature_disabled",
+                "External PBX integrations are disabled for this organization",
+            )
+        resolved = _resolve_context_mapping_transfer(
+            config, call_context_vars, gathered_context_vars
+        )
+        logger.info(
+            "Transfer destination resolved from context mapping "
+            f"context_path={resolved.metadata.get('context_path')} "
+            f"source={resolved.source} destination={resolved.destination}"
+        )
+        return resolved
+
     resolver = config.get("resolver")
     if config.get("destination_source", "static") != "dynamic" or not isinstance(
         resolver, dict

api/tests/telephony/providers/ari/test_external_pbx.py

@@ -0,0 +1,218 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import redis.asyncio as aioredis
+
+from api.db import db_client
+from api.services.telephony.external_pbx import resolve_external_pbx_field_mappings
+from api.services.telephony.providers.ari.external_pbx import (
+    ExternalPBXResult,
+    create_adapter,
+)
+from api.services.telephony.providers.ari.strategies import ARIHangupStrategy
+from api.services.workflow.tools import transfer_resolver
+
+
+def _vicidial_config() -> dict:
+    return {
+        "type": "vicidial",
+        "agent_api": {
+            "url": "https://vici.example.com/agc/api.php",
+            "username": "agent-api-user",
+            "password": "secret",
+            "source": "dograh",
+        },
+        "non_agent_api": {
+            "url": "https://vici.example.com/vicidial/non_agent_api.php",
+            "username": "lead-api-user",
+            "password": "secret",
+            "source": "dograh",
+        },
+    }
+
+
+@pytest.mark.asyncio
+async def test_vicidial_adapter_captures_call_identity_from_headers():
+    adapter = create_adapter(_vicidial_config())
+    headers = {
+        "X-VICIDIAL-callerid": "M123",
+        "X-VICIDIAL-user": "remote-agent",
+        "X-VICIDIAL-lead_id": "42",
+        "X-VICIDIAL-campaign_id": "campaign",
+        "X-VICIDIAL-ingroup_id": "source-group",
+    }
+
+    async def read_header(name: str) -> str:
+        return headers.get(name, "")
+
+    identity = await adapter.capture_call_identity(read_header)
+
+    assert identity == {
+        "type": "vicidial",
+        "callerid": "M123",
+        "agent_user": "remote-agent",
+        "lead_id": "42",
+        "campaign_id": "campaign",
+        "ingroup_id": "source-group",
+    }
+
+
+@pytest.mark.asyncio
+async def test_vicidial_adapter_resolves_source_ingroup(monkeypatch):
+    adapter = create_adapter(_vicidial_config())
+    call_control = AsyncMock(
+        return_value=ExternalPBXResult(True, "ingrouptransfer", "ok")
+    )
+    monkeypatch.setattr(adapter, "_agent_call_control", call_control)
+
+    result = await adapter.transfer(
+        {"callerid": "M123", "agent_user": "agent", "ingroup_id": "support"},
+        "source",
+    )
+
+    assert result.ok is True
+    call_control.assert_awaited_once_with(
+        {"callerid": "M123", "agent_user": "agent", "ingroup_id": "support"},
+        "INGROUPTRANSFER",
+        ingroup_choices="support",
+    )
+
+
+def test_field_mapping_reads_extracted_variables_and_skips_empty_values():
+    fields = resolve_external_pbx_field_mappings(
+        {
+            "extracted_variables": {"qualified": "yes", "empty": "  "},
+            "call_disposition": "completed",
+        },
+        [
+            {"context_path": "qualified", "destination_field": "address3"},
+            {"context_path": "empty", "destination_field": "comments"},
+            {
+                "context_path": "call_disposition",
+                "destination_field": "status_notes",
+            },
+        ],
+    )
+
+    assert fields == {"address3": "yes", "status_notes": "completed"}
+
+
+@pytest.mark.asyncio
+async def test_context_mapping_resolves_ingroup_destination(monkeypatch):
+    monkeypatch.setattr(
+        transfer_resolver,
+        "external_pbx_integrations_enabled",
+        AsyncMock(return_value=True),
+    )
+
+    resolved = await transfer_resolver.resolve_transfer_config(
+        tool=SimpleNamespace(tool_uuid="tool-1"),
+        config={
+            "destination_source": "context_mapping",
+            "context_mapping": {
+                "context_path": "qualified",
+                "routes": [
+                    {"context_value": "YES", "destination": "sales"},
+                ],
+            },
+        },
+        arguments={},
+        call_context_vars={},
+        gathered_context_vars={"extracted_variables": {"qualified": " yes "}},
+        organization_id=7,
+        workflow_run_id=11,
+    )
+
+    assert resolved.destination == "sales"
+    assert resolved.source == "context_mapping"
+
+
+@pytest.mark.asyncio
+async def test_context_mapping_is_disabled_at_runtime(monkeypatch):
+    monkeypatch.setattr(
+        transfer_resolver,
+        "external_pbx_integrations_enabled",
+        AsyncMock(return_value=False),
+    )
+
+    with pytest.raises(
+        transfer_resolver.TransferResolutionError,
+        match="External PBX integrations are disabled",
+    ):
+        await transfer_resolver.resolve_transfer_config(
+            tool=SimpleNamespace(tool_uuid="tool-1"),
+            config={
+                "destination_source": "context_mapping",
+                "context_mapping": {
+                    "context_path": "qualified",
+                    "routes": [
+                        {"context_value": "yes", "destination": "sales"},
+                    ],
+                },
+            },
+            arguments={},
+            call_context_vars={},
+            gathered_context_vars={"qualified": "yes"},
+            organization_id=7,
+            workflow_run_id=11,
+        )
+
+
+@pytest.mark.asyncio
+async def test_hangup_strategy_updates_lead_before_customer_leg(monkeypatch):
+    redis = AsyncMock()
+    redis.get.return_value = "11"
+    monkeypatch.setattr(aioredis, "from_url", lambda *args, **kwargs: redis)
+    run = SimpleNamespace(
+        initial_context={
+            "external_pbx_call": {
+                "type": "vicidial",
+                "callerid": "M123",
+                "agent_user": "agent",
+                "lead_id": "42",
+            }
+        },
+        gathered_context={"extracted_variables": {"qualified": "yes"}},
+        workflow=SimpleNamespace(organization_id=7),
+    )
+    monkeypatch.setattr(
+        db_client, "get_workflow_run_by_id", AsyncMock(return_value=run)
+    )
+    monkeypatch.setattr(
+        db_client,
+        "get_workflow_run_configurations",
+        AsyncMock(
+            return_value={
+                "external_pbx_field_mappings": [
+                    {"context_path": "qualified", "destination_field": "address3"}
+                ]
+            }
+        ),
+    )
+    adapter = SimpleNamespace(
+        type="vicidial",
+        update_fields=AsyncMock(
+            return_value=ExternalPBXResult(True, "update_lead", "ok")
+        ),
+        hangup=AsyncMock(return_value=ExternalPBXResult(True, "hangup", "ok")),
+    )
+
+    await ARIHangupStrategy(adapter)._terminate_external_pbx_if_any("channel-1")
+
+    adapter.update_fields.assert_awaited_once_with(
+        run.initial_context["external_pbx_call"], {"address3": "yes"}
+    )
+    adapter.hangup.assert_awaited_once_with(run.initial_context["external_pbx_call"])
+    redis.aclose.assert_awaited_once_with()
+
+
+@pytest.mark.asyncio
+async def test_hangup_strategy_closes_redis_when_channel_has_no_run(monkeypatch):
+    redis = AsyncMock()
+    redis.get.return_value = None
+    monkeypatch.setattr(aioredis, "from_url", lambda *args, **kwargs: redis)
+
+    await ARIHangupStrategy()._terminate_external_pbx_if_any("missing-channel")
+
+    redis.aclose.assert_awaited_once_with()

api/tests/telephony/test_external_pbx_configuration.py

@@ -0,0 +1,80 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+from api.routes import organization
+from api.services import tool_management
+
+
+def _credentials(password: str = "agent-secret") -> dict:
+    return {
+        "ari_endpoint": "https://asterisk.example.com",
+        "app_name": "dograh",
+        "app_password": "ari-secret",
+        "external_pbx": {
+            "type": "vicidial",
+            "agent_api": {
+                "url": "https://vici.example.com/agc/api.php",
+                "username": "agent-user",
+                "password": password,
+            },
+        },
+    }
+
+
+def test_nested_external_pbx_secrets_are_masked_without_mutating_source():
+    credentials = _credentials()
+
+    masked = organization._mask_sensitive("ari", credentials)
+
+    assert masked["app_password"] != "ari-secret"
+    assert masked["external_pbx"]["agent_api"]["password"] != "agent-secret"
+    assert credentials["external_pbx"]["agent_api"]["password"] == "agent-secret"
+
+
+def test_nested_masked_external_pbx_secrets_are_restored_on_update():
+    existing = _credentials()
+    request = organization._mask_sensitive("ari", existing)
+
+    organization.preserve_masked_fields("ari", request, existing)
+
+    assert request["app_password"] == "ari-secret"
+    assert request["external_pbx"]["agent_api"]["password"] == "agent-secret"
+
+
+@pytest.mark.asyncio
+async def test_disabled_feature_preserves_existing_tool_mapping(monkeypatch):
+    monkeypatch.setattr(
+        tool_management,
+        "external_pbx_integrations_enabled",
+        AsyncMock(return_value=False),
+    )
+    definition = {
+        "type": "transfer_call",
+        "config": {
+            "destination_source": "context_mapping",
+            "context_mapping": {
+                "context_path": "qualified",
+                "routes": [{"context_value": "yes", "destination": "sales"}],
+            },
+        },
+    }
+
+    await tool_management.validate_external_pbx_tool_definition(
+        definition,
+        organization_id=7,
+        existing_definition=definition,
+    )
+
+    changed = {
+        "type": "transfer_call",
+        "config": {"destination_source": "static", "destination": "+15555550100"},
+    }
+    with pytest.raises(tool_management.ToolManagementError) as exc_info:
+        await tool_management.validate_external_pbx_tool_definition(
+            changed,
+            organization_id=7,
+            existing_definition=definition,
+        )
+
+    assert exc_info.value.status_code == 403

api/tests/test_custom_tools.py

@@ -30,6 +30,7 @@ from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.llm_service import FunctionCallParams
 
 from api.enums import WorkflowRunMode
+from api.services.configuration.masking import mask_key
 from api.services.workflow.pipecat_engine_custom_tools import get_function_schema
 from api.services.workflow.tools.custom_tool import (
     _coerce_parameter_value,
@@ -430,6 +431,7 @@ class TestExecuteHttpTool:
             assert result["status"] == "success"
             assert result["status_code"] == 201
             assert result["data"]["id"] == 123
+            assert "request_headers" not in result
 
     @pytest.mark.asyncio
     async def test_post_request_sends_nested_json_body(self):
@@ -546,6 +548,55 @@ class TestExecuteHttpTool:
             }
             assert result["status"] == "success"
 
+    @pytest.mark.asyncio
+    async def test_post_request_accepts_pre_resolved_preset_params(self):
+        """The test endpoint can supply preset values without call context."""
+        tool = MockToolModel(
+            tool_uuid="test-uuid-preset-override",
+            name="Create Lead",
+            description="Create a lead with caller context",
+            category="http_api",
+            definition={
+                "schema_version": 1,
+                "type": "http_api",
+                "config": {
+                    "method": "POST",
+                    "url": "https://api.example.com/leads",
+                    "timeout_ms": 5000,
+                    "preset_parameters": [
+                        {
+                            "name": "phone_number",
+                            "type": "string",
+                            "value_template": "{{initial_context.phone_number}}",
+                            "required": True,
+                        }
+                    ],
+                },
+            },
+        )
+
+        with patch(
+            "api.services.workflow.tools.custom_tool.httpx.AsyncClient"
+        ) as mock_client_class:
+            mock_client = AsyncMock()
+            mock_response = Mock()
+            mock_response.status_code = 201
+            mock_response.json.return_value = {"id": 123}
+            mock_client.request.return_value = mock_response
+            mock_client_class.return_value.__aenter__.return_value = mock_client
+
+            result = await execute_http_tool(
+                tool,
+                {"name": "John"},
+                preset_params={"phone_number": "+14155550123"},
+            )
+
+            assert mock_client.request.call_args.kwargs["json"] == {
+                "name": "John",
+                "phone_number": "+14155550123",
+            }
+            assert result["status"] == "success"
+
     @pytest.mark.asyncio
     async def test_missing_required_preset_parameter_returns_error(self):
         """Test that required preset parameters fail before the HTTP request."""
@@ -619,6 +670,52 @@ class TestExecuteHttpTool:
 
             assert result["status"] == "success"
 
+    @pytest.mark.asyncio
+    async def test_get_request_serializes_object_and_array_query_params(self):
+        """Object/array-typed arguments must be JSON-stringified for GET query
+        params — httpx raises a TypeError if a dict/list is passed as-is."""
+        tool = MockToolModel(
+            tool_uuid="test-uuid",
+            name="Search Users",
+            description="Search for users",
+            category="http_api",
+            definition={
+                "schema_version": 1,
+                "type": "http_api",
+                "config": {
+                    "method": "GET",
+                    "url": "https://api.example.com/users/search",
+                    "timeout_ms": 5000,
+                },
+            },
+        )
+
+        arguments = {
+            "source": "voice",
+            "metadata": {"campaign": "spring"},
+            "tags": ["a", "b"],
+        }
+
+        with patch(
+            "api.services.workflow.tools.custom_tool.httpx.AsyncClient"
+        ) as mock_client_class:
+            mock_client = AsyncMock()
+            mock_response = Mock()
+            mock_response.status_code = 200
+            mock_response.json.return_value = {"users": []}
+            mock_client.request.return_value = mock_response
+            mock_client_class.return_value.__aenter__.return_value = mock_client
+
+            result = await execute_http_tool(tool, arguments)
+
+            call_kwargs = mock_client.request.call_args.kwargs
+            assert call_kwargs["params"] == {
+                "source": "voice",
+                "metadata": '{"campaign": "spring"}',
+                "tags": '["a", "b"]',
+            }
+            assert result["status"] == "success"
+
     @pytest.mark.asyncio
     async def test_get_request_without_arguments_preserves_url_query_params(self):
         """Empty runtime args should not override query params already in the URL."""
@@ -762,11 +859,17 @@ class TestExecuteHttpTool:
             mock_client.request.return_value = mock_response
             mock_client_class.return_value.__aenter__.return_value = mock_client
 
-            await execute_http_tool(tool, {"data": "test"})
+            result = await execute_http_tool(
+                tool, {"data": "test"}, include_request_headers=True
+            )
 
             call_kwargs = mock_client.request.call_args.kwargs
             assert call_kwargs["headers"]["X-API-Key"] == "secret-key"
             assert call_kwargs["headers"]["X-Custom-Header"] == "custom-value"
+            assert result["request_headers"] == {
+                "X-API-Key": "secret-key",
+                "X-Custom-Header": "custom-value",
+            }
 
     @pytest.mark.asyncio
     async def test_request_includes_auth_header_from_credential(self):
@@ -807,7 +910,12 @@ class TestExecuteHttpTool:
             with patch("api.services.workflow.tools.custom_tool.db_client") as mock_db:
                 mock_db.get_credential_by_uuid = AsyncMock(return_value=mock_credential)
 
-                await execute_http_tool(tool, {"data": "test"}, organization_id=1)
+                result = await execute_http_tool(
+                    tool,
+                    {"data": "test"},
+                    organization_id=1,
+                    include_request_headers=True,
+                )
 
                 # Verify credential was fetched
                 mock_db.get_credential_by_uuid.assert_called_once_with(
@@ -819,6 +927,12 @@ class TestExecuteHttpTool:
                 assert (
                     call_kwargs["headers"]["Authorization"] == "Bearer my-secret-token"
                 )
+                assert result["request_headers"]["Authorization"] == mask_key(
+                    "Bearer my-secret-token"
+                )
+                assert (
+                    "my-secret-token" not in result["request_headers"]["Authorization"]
+                )
 
     @pytest.mark.asyncio
     async def test_no_credential_lookup_without_organization_id(self):
@@ -1704,7 +1818,6 @@ class TestCustomToolManagerUnit:
         assert [
             call.args[0] for call in mock_engine.set_mute_pipeline.call_args_list
         ] == [
-            True,
             True,
             False,
         ]
@@ -1716,8 +1829,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 +1842,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 +1904,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):

api/tests/test_mcp_tool_route.py

@@ -26,9 +26,15 @@ from api.routes.tool import (
     CreateToolRequest,
     McpToolConfig,
     McpToolDefinition,
+    ToolTestRequest,
+    ToolTestResponse,
     UpdateToolRequest,
     _populate_discovered_tools,
     refresh_mcp_tools,
+    router,
+)
+from api.routes.tool import (
+    test_tool as call_test_tool_route,
 )
 from api.services.workflow.tools.mcp_tool import (
     validate_mcp_definition,
@@ -420,6 +426,314 @@ def _mcp_tool_model(org_id=1):
     return t
 
 
+def _http_tool_model(method="GET"):
+    t = MagicMock()
+    t.tool_uuid = "tu-http"
+    t.name = "Mock HTTP"
+    t.category = "http_api"
+    t.definition = {
+        "schema_version": 1,
+        "type": "http_api",
+        "config": {"method": method, "url": "https://example.com/search"},
+    }
+    return t
+
+
+@pytest.mark.asyncio
+async def test_tool_executes_http_api_tool_with_llm_and_preset_params(monkeypatch):
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model()
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    executor = AsyncMock(
+        return_value={
+            "status": "success",
+            "status_code": 200,
+            "data": {"ok": True},
+        }
+    )
+    monkeypatch.setattr(tool_route, "execute_http_tool", executor)
+
+    resp = await call_test_tool_route(
+        "tu-http",
+        request=ToolTestRequest(
+            llm_params={"query": "cart"},
+            preset_params={
+                "customer_id": "c_123",
+                "sentiment": "cooperative",
+            },
+        ),
+        user=_fake_user(),
+    )
+
+    assert resp.status == "success"
+    assert resp.status_code == 200
+    assert resp.data == {"ok": True}
+    assert resp.error is None
+    executor.assert_awaited_once_with(
+        tool,
+        {"query": "cart"},
+        preset_params={
+            "customer_id": "c_123",
+            "sentiment": "cooperative",
+        },
+        organization_id=1,
+        include_request_headers=True,
+    )
+    assert resp.hint is None
+    assert resp.request_method == "GET"
+    assert resp.request_url == "https://example.com/search"
+    assert resp.request_headers == {}
+    assert resp.request_body is None
+    assert resp.request_params == {
+        "query": "cart",
+        "customer_id": "c_123",
+        "sentiment": "cooperative",
+    }
+
+
+@pytest.mark.asyncio
+async def test_tool_test_sets_request_body_for_post_method(monkeypatch):
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model(method="POST")
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    monkeypatch.setattr(
+        tool_route,
+        "execute_http_tool",
+        AsyncMock(
+            return_value={"status": "success", "status_code": 200, "data": {"id": 1}}
+        ),
+    )
+
+    resp = await call_test_tool_route(
+        "tu-http",
+        request=ToolTestRequest(llm_params={"name": "Ada"}),
+        user=_fake_user(),
+    )
+
+    assert resp.request_method == "POST"
+    assert resp.request_body == {"name": "Ada"}
+    assert resp.request_params is None
+
+
+@pytest.mark.asyncio
+async def test_tool_test_returns_masked_effective_request_headers(monkeypatch):
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model(method="POST")
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    monkeypatch.setattr(
+        tool_route,
+        "execute_http_tool",
+        AsyncMock(
+            return_value={
+                "status": "success",
+                "status_code": 200,
+                "data": {"ok": True},
+                "request_headers": {
+                    "X-Tenant": "acme",
+                    "Authorization": "****************oken",
+                },
+            }
+        ),
+    )
+
+    resp = await call_test_tool_route(
+        "tu-http", request=ToolTestRequest(), user=_fake_user()
+    )
+
+    assert resp.request_headers == {
+        "X-Tenant": "acme",
+        "Authorization": "****************oken",
+    }
+
+
+@pytest.mark.asyncio
+async def test_tool_test_request_body_includes_resolved_preset_parameters(
+    monkeypatch,
+):
+    """The Request preview includes direct preset values alongside LLM values."""
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model(method="POST")
+    tool.definition["config"]["preset_parameters"] = [
+        {
+            "name": "source",
+            "type": "string",
+            "value_template": "{{initial_context.metadata.channel}}",
+            "required": True,
+        }
+    ]
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    monkeypatch.setattr(
+        tool_route,
+        "execute_http_tool",
+        AsyncMock(
+            return_value={"status": "success", "status_code": 200, "data": {"ok": True}}
+        ),
+    )
+
+    resp = await call_test_tool_route(
+        "tu-http",
+        request=ToolTestRequest(
+            llm_params={"name": "Ada"},
+            preset_params={"source": "web_widget"},
+        ),
+        user=_fake_user(),
+    )
+
+    assert resp.request_body == {"name": "Ada", "source": "web_widget"}
+
+
+@pytest.mark.asyncio
+async def test_tool_test_no_arguments_post_shows_empty_body(monkeypatch):
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model(method="POST")
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    monkeypatch.setattr(
+        tool_route,
+        "execute_http_tool",
+        AsyncMock(return_value={"status": "success", "status_code": 200, "data": None}),
+    )
+
+    resp = await call_test_tool_route(
+        "tu-http", request=ToolTestRequest(), user=_fake_user()
+    )
+
+    # POST with no arguments sends json={} over the wire; preview must show {}
+    # so callers can distinguish an absent body from an empty one.
+    assert resp.request_body == {}
+    assert resp.request_params is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "status_code,expected_snippet",
+    [
+        (400, "HTTP 400 Bad Request"),
+        (401, "HTTP 401 Unauthorized"),
+        (403, "HTTP 403 Forbidden"),
+        (404, "HTTP 404 Not Found"),
+        (405, "HTTP 405 Method Not Allowed"),
+        (408, "HTTP 408 Request Timeout"),
+        (409, "HTTP 409 Conflict"),
+        (415, "HTTP 415 Unsupported Media Type"),
+        (422, "HTTP 422 Unprocessable Entity"),
+        (429, "HTTP 429 Too Many Requests"),
+        (500, "HTTP 500"),
+        (503, "HTTP 503"),
+    ],
+)
+async def test_tool_test_hint_for_status_code(
+    monkeypatch, status_code, expected_snippet
+):
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model(method="POST")
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    monkeypatch.setattr(
+        tool_route,
+        "execute_http_tool",
+        AsyncMock(
+            return_value={
+                "status": "error",
+                "status_code": status_code,
+                "error": "boom",
+            }
+        ),
+    )
+
+    resp = await call_test_tool_route(
+        "tu-http", request=ToolTestRequest(llm_params={"a": 1}), user=_fake_user()
+    )
+
+    assert resp.hint is not None
+    assert resp.hint.startswith(expected_snippet)
+
+
+@pytest.mark.asyncio
+async def test_tool_test_no_hint_on_success(monkeypatch):
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model(method="GET")
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    monkeypatch.setattr(
+        tool_route,
+        "execute_http_tool",
+        AsyncMock(return_value={"status": "success", "status_code": 200, "data": {}}),
+    )
+
+    resp = await call_test_tool_route(
+        "tu-http", request=ToolTestRequest(), user=_fake_user()
+    )
+
+    assert resp.hint is None
+
+
+@pytest.mark.asyncio
+async def test_tool_test_no_hint_for_uncovered_status_code(monkeypatch):
+    import api.routes.tool as tool_route
+
+    tool = _http_tool_model(method="GET")
+    monkeypatch.setattr(
+        tool_route.db_client, "get_tool_by_uuid", AsyncMock(return_value=tool)
+    )
+    monkeypatch.setattr(
+        tool_route,
+        "execute_http_tool",
+        AsyncMock(
+            return_value={"status": "error", "status_code": 418, "error": "teapot"}
+        ),
+    )
+
+    resp = await call_test_tool_route(
+        "tu-http", request=ToolTestRequest(), user=_fake_user()
+    )
+
+    assert resp.hint is None
+
+
+def test_tool_test_route_is_registered():
+    assert any(
+        route.path == "/tools/{tool_uuid}/test" and "POST" in route.methods
+        for route in router.routes
+    )
+
+
+@pytest.mark.asyncio
+async def test_tool_rejects_non_http_api_tool(monkeypatch):
+    import api.routes.tool as tool_route
+
+    monkeypatch.setattr(
+        tool_route.db_client,
+        "get_tool_by_uuid",
+        AsyncMock(return_value=_mcp_tool_model()),
+    )
+
+    with pytest.raises(HTTPException) as ei:
+        await call_test_tool_route(
+            "tu-mcp", request=ToolTestRequest(), user=_fake_user()
+        )
+
+    assert ei.value.status_code == 400
+
+
 @pytest.mark.asyncio
 async def test_refresh_success(monkeypatch):
     import api.services.tool_management as tool_svc
@@ -484,3 +798,41 @@ async def test_refresh_not_found_is_404(monkeypatch):
     with pytest.raises(HTTPException) as ei:
         await refresh_mcp_tools("nope", user=_fake_user())
     assert ei.value.status_code == 404
+
+
+def test_tool_test_response_has_hint_and_request_fields():
+    """ToolTestResponse must carry hint + request_method/url/body/params
+    so the frontend can show what was sent and why it may have failed."""
+    resp = ToolTestResponse(
+        status="error",
+        status_code=405,
+        data=None,
+        error="Method Not Allowed",
+        duration_ms=12,
+        hint="HTTP 405 Method Not Allowed — the endpoint rejected the configured method (POST).",
+        request_method="POST",
+        request_url="https://example.com/thing",
+        request_headers={"Authorization": "********oken"},
+        request_body={"a": 1},
+        request_params=None,
+    )
+    assert resp.hint.startswith("HTTP 405")
+    assert resp.request_method == "POST"
+    assert resp.request_url == "https://example.com/thing"
+    assert resp.request_headers == {"Authorization": "********oken"}
+    assert resp.request_body == {"a": 1}
+    assert resp.request_params is None
+
+
+def test_tool_test_response_request_fields_default_to_none_or_required():
+    """hint/request_body/request_params are optional; request_method/url are required."""
+    resp = ToolTestResponse(
+        status="success",
+        duration_ms=5,
+        request_method="GET",
+        request_url="https://example.com/thing",
+    )
+    assert resp.hint is None
+    assert resp.request_headers == {}
+    assert resp.request_body is None
+    assert resp.request_params is None

api/tests/test_run_list_pagination_bounds.py

@@ -0,0 +1,40 @@
+"""Pagination bounds for the workflow-run and campaign-run list endpoints.
+
+Regression for issue #553: `limit=0` raised an unhandled ZeroDivisionError
+(HTTP 500) in the `total_pages` computation, and negative `limit`/`page`
+produced nonsensical pagination. Both endpoints now validate the params
+(`limit` in [1, 100], `page` >= 1), matching the sibling list endpoints.
+"""
+
+import pytest
+
+
+async def _make_user(db_session, slug: str):
+    user, _ = await db_session.get_or_create_user_by_provider_id(f"{slug}_user")
+    org, _ = await db_session.get_or_create_organization_by_provider_id(
+        f"{slug}_org", user.id
+    )
+    await db_session.update_user_selected_organization(user.id, org.id)
+    return await db_session.get_user_by_id(user.id)
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/workflow/1/runs",
+        "/api/v1/campaign/1/runs",
+    ],
+)
+@pytest.mark.parametrize("query", ["limit=0", "limit=-5", "limit=101", "page=0"])
+async def test_run_list_rejects_out_of_range_pagination(
+    test_client_factory, db_session, path, query
+):
+    """Out-of-range limit/page is a 422 validation error, never a 500."""
+    user = await _make_user(db_session, "paginate_bounds")
+
+    async with test_client_factory(user) as client:
+        response = await client.get(f"{path}?{query}")
+
+    assert response.status_code == 422, (
+        f"{path}?{query} expected 422, got {response.status_code}: {response.text}"
+    )

api/tests/test_run_pipeline_realtime_turn_config.py

@@ -1,8 +1,11 @@
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import BotStartedSpeakingFrame, TranscriptionFrame
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_start import (
     ExternalUserTurnStartStrategy,
     MinWordsUserTurnStartStrategy,
     ProvisionalVADUserTurnStartStrategy,
+    TranscriptionUserTurnStartStrategy,
 )
 from pipecat.turns.user_start.vad_user_turn_start_strategy import (
     VADUserTurnStartStrategy,
@@ -134,14 +137,15 @@ def test_non_realtime_default_uses_external_start_for_external_turn_stt():
     assert strategies[0]._enable_interruptions is True
 
 
-def test_non_realtime_default_uses_vad_start_for_standard_stt():
+def test_non_realtime_default_uses_transcription_fallback_and_vad_for_standard_stt():
     strategies = _create_non_realtime_user_turn_start_strategies(
         {},
         uses_external_turns=False,
     )
 
-    assert len(strategies) == 1
-    assert isinstance(strategies[0], VADUserTurnStartStrategy)
+    assert len(strategies) == 2
+    assert isinstance(strategies[0], TranscriptionUserTurnStartStrategy)
+    assert isinstance(strategies[1], VADUserTurnStartStrategy)
 
 
 def test_non_realtime_can_use_min_words_start_strategy():
@@ -199,6 +203,28 @@ def test_non_realtime_provisional_vad_uses_configured_pause_secs():
     assert strategies[0]._pause_secs == 0.4
 
 
+async def test_non_realtime_provisional_vad_starts_on_transcript_without_vad():
+    strategies = _create_non_realtime_user_turn_start_strategies(
+        {"turn_start_strategy": "provisional_vad"},
+        uses_external_turns=False,
+    )
+    strategy = strategies[0]
+    turn_started = False
+
+    @strategy.event_handler("on_user_turn_started")
+    async def on_user_turn_started(strategy, params):
+        nonlocal turn_started
+        turn_started = True
+
+    await strategy.process_frame(BotStartedSpeakingFrame())
+    result = await strategy.process_frame(
+        TranscriptionFrame(text="Hello", user_id="user", timestamp="")
+    )
+
+    assert result == ProcessFrameResult.STOP
+    assert turn_started is True
+
+
 def test_non_realtime_uses_external_stop_for_external_turn_stt():
     strategies = _create_non_realtime_user_turn_stop_strategies(
         {},

api/tests/test_tool_schema.py

@@ -42,3 +42,50 @@ def test_transfer_call_dynamic_accepts_resolver_without_destination():
     assert config.destination_source == "dynamic"
     assert config.destination == ""
     assert config.resolver is not None
+
+
+def test_transfer_call_context_mapping_requires_mapping():
+    with pytest.raises(ValueError, match="context_mapping is required"):
+        TransferCallConfig(destination_source="context_mapping")
+
+
+def test_transfer_call_context_mapping_accepts_unique_routes():
+    config = TransferCallConfig(
+        destination_source="context_mapping",
+        context_mapping={
+            "context_path": "qualified",
+            "routes": [
+                {"context_value": "yes", "destination": "sales"},
+                {"context_value": "no", "destination": "support"},
+            ],
+            "fallback_destination": "source",
+        },
+    )
+
+    assert config.context_mapping is not None
+    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(
+            destination_source="context_mapping",
+            context_mapping={
+                "context_path": "qualified",
+                "routes": [
+                    {"context_value": "Yes", "destination": "sales"},
+                    {"context_value": "yes", "destination": "support"},
+                ],
+            },
+        )

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()

api/tests/test_workflow_configurations_schema.py

@@ -59,3 +59,32 @@ def test_cap_stays_within_concurrency_stale_timeout():
     from api.services.campaign.rate_limiter import rate_limiter
 
     assert MAX_CALL_DURATION_SECONDS <= rate_limiter.stale_call_timeout
+
+
+def test_external_pbx_field_mapping_is_validated():
+    config = WorkflowConfigurationDefaults(
+        external_pbx_field_mappings=[
+            {"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(
+            external_pbx_field_mappings=[
+                {"context_path": "qualified", "destination_field": "invalid-field"}
+            ]
+        )

api/tests/test_xai_tts_service_factory.py

@@ -24,10 +24,7 @@ def test_xai_tts_configuration_defaults():
     assert XAI_TTS_VOICES == ["eve", "ara", "leo", "rex", "sal"]
 
 
-@pytest.mark.parametrize("transport_out_sample_rate", [8000, 16000])
-def test_create_xai_tts_service_uses_pipeline_compatible_audio_format(
-    transport_out_sample_rate,
-):
+def test_create_xai_tts_service_uses_websocket_streaming():
     user_config = SimpleNamespace(
         tts=SimpleNamespace(
             provider=ServiceProviders.XAI.value,
@@ -38,20 +35,18 @@ def test_create_xai_tts_service_uses_pipeline_compatible_audio_format(
         )
     )
     audio_config = SimpleNamespace(
-        transport_out_sample_rate=transport_out_sample_rate,
+        transport_out_sample_rate=24000,
         transport_in_sample_rate=16000,
     )
 
-    with patch(
-        "api.services.pipecat.service_factory.XAIHttpTTSService"
-    ) as mock_service:
+    with patch("api.services.pipecat.service_factory.XAITTSService") as mock_service:
         create_tts_service(user_config, audio_config)
 
     assert mock_service.call_count == 1
     kwargs = mock_service.call_args.kwargs
     assert kwargs["api_key"] == "test-key"
-    assert kwargs["sample_rate"] == transport_out_sample_rate
-    assert kwargs["encoding"] == "pcm"
+    # Sample rate is resolved from the pipeline StartFrame, like other providers.
+    assert "sample_rate" not in kwargs
     assert kwargs["settings"].voice == "rex"
     assert kwargs["settings"].language == Language.EN
 
@@ -71,9 +66,7 @@ def test_create_xai_tts_service_converts_language():
         transport_in_sample_rate=16000,
     )
 
-    with patch(
-        "api.services.pipecat.service_factory.XAIHttpTTSService"
-    ) as mock_service:
+    with patch("api.services.pipecat.service_factory.XAITTSService") as mock_service:
         create_tts_service(user_config, audio_config)
 
     kwargs = mock_service.call_args.kwargs
@@ -95,9 +88,7 @@ def test_create_xai_tts_service_falls_back_to_english_for_unknown_language():
         transport_in_sample_rate=16000,
     )
 
-    with patch(
-        "api.services.pipecat.service_factory.XAIHttpTTSService"
-    ) as mock_service:
+    with patch("api.services.pipecat.service_factory.XAITTSService") as mock_service:
         create_tts_service(user_config, audio_config)
 
     kwargs = mock_service.call_args.kwargs
@@ -119,9 +110,7 @@ def test_create_xai_tts_service_preserves_auto_language():
         transport_in_sample_rate=16000,
     )
 
-    with patch(
-        "api.services.pipecat.service_factory.XAIHttpTTSService"
-    ) as mock_service:
+    with patch("api.services.pipecat.service_factory.XAITTSService") as mock_service:
         create_tts_service(user_config, audio_config)
 
     kwargs = mock_service.call_args.kwargs