feat: cap max_call_duration at 20 min via typed workflow_configurations request

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

api/routes/telephony.py

@@ -687,6 +687,7 @@ async def _handle_telephony_websocket(
             # WebSocket already closed, ignore
             pass
 
+
 @router.post("/inbound/run")
 async def handle_inbound_run(request: Request):
     """Workflow-agnostic inbound dispatcher.

api/routes/workflow.py

@@ -18,6 +18,7 @@ from api.db.workflow_template_client import WorkflowTemplateClient
 from api.enums import CallType, PostHogEvent, StorageBackend, WorkflowStatus
 from api.schemas.ai_model_configuration import OrganizationAIModelConfigurationV2
 from api.schemas.workflow import WorkflowRunResponseSchema
+from api.schemas.workflow_configurations import WorkflowConfigurationDefaults
 from api.sdk_expose import sdk_expose
 from api.services.auth.depends import get_user
 from api.services.configuration.ai_model_configuration import (
@@ -284,7 +285,10 @@ class UpdateWorkflowRequest(BaseModel):
     name: str | None = None
     workflow_definition: dict | None = None
     template_context_variables: dict | None = None
-    workflow_configurations: dict | None = None
+    # Typed so field constraints (e.g. the max_call_duration cap) are
+    # enforced by FastAPI; extra="allow" keeps passthrough keys like
+    # model_configuration_v2_override intact.
+    workflow_configurations: WorkflowConfigurationDefaults | None = None
 
 
 class WorkflowVersionResponse(BaseModel):
@@ -1039,7 +1043,13 @@ async def update_workflow(
 
         # Validate model overrides. v2 uses a complete workflow-level model
         # configuration; legacy v1 uses partial service overlays.
-        workflow_configurations = request.workflow_configurations
+        # exclude_unset keeps stored configs sparse: keys the request didn't
+        # send stay absent so runtime defaults keep applying to them.
+        workflow_configurations = (
+            request.workflow_configurations.model_dump(exclude_unset=True)
+            if request.workflow_configurations is not None
+            else None
+        )
         if workflow_configurations and workflow_configurations.get(
             WORKFLOW_MODEL_CONFIGURATION_V2_OVERRIDE_KEY
         ):

api/schemas/workflow_configurations.py

@@ -1,8 +1,12 @@
 from typing import Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 DEFAULT_MAX_CALL_DURATION_SECONDS = 300
+# Hard ceiling on configurable call duration. Must stay <= the concurrency
+# rate limiter's stale_call_timeout (20 min): a call running past that has
+# its slot purged as stale and the org concurrency limit under-counts.
+MAX_CALL_DURATION_SECONDS = 1200
 DEFAULT_MAX_USER_IDLE_TIMEOUT_SECONDS = 10.0
 DEFAULT_SMART_TURN_STOP_SECS = 2.0
 DEFAULT_TURN_START_STRATEGY = "default"
@@ -22,10 +26,24 @@ class AmbientNoiseConfigurationDefaults(BaseModel):
 class WorkflowConfigurationDefaults(BaseModel):
     model_config = ConfigDict(extra="allow")
 
+    @model_validator(mode="before")
+    @classmethod
+    def _treat_null_as_unset(cls, data):
+        # Stored configs (and older clients) carry explicit JSON nulls for
+        # keys the user never configured; dropping them lets the field
+        # defaults apply instead of failing validation.
+        if isinstance(data, dict):
+            return {k: v for k, v in data.items() if v is not None}
+        return data
+
     ambient_noise_configuration: AmbientNoiseConfigurationDefaults = Field(
         default_factory=AmbientNoiseConfigurationDefaults
     )
-    max_call_duration: int = DEFAULT_MAX_CALL_DURATION_SECONDS
+    max_call_duration: int = Field(
+        default=DEFAULT_MAX_CALL_DURATION_SECONDS,
+        gt=0,
+        le=MAX_CALL_DURATION_SECONDS,
+    )
     max_user_idle_timeout: float = DEFAULT_MAX_USER_IDLE_TIMEOUT_SECONDS
     smart_turn_stop_secs: float = DEFAULT_SMART_TURN_STOP_SECS
     turn_start_strategy: Literal["default", "min_words", "provisional_vad"] = (

api/tests/test_campaign_call_dispatcher.py

@@ -845,9 +845,7 @@ class TestAcquireConcurrentSlotScoping:
         )
 
     @pytest.mark.asyncio
-    async def test_no_campaign_max_concurrency_skips_scope(
-        self, mock_call_concurrency
-    ):
+    async def test_no_campaign_max_concurrency_skips_scope(self, mock_call_concurrency):
         dispatcher = CampaignCallDispatcher()
         campaign = self._campaign({})
 

api/tests/test_workflow_configurations_schema.py

@@ -0,0 +1,61 @@
+import pytest
+from pydantic import ValidationError
+
+from api.schemas.workflow_configurations import (
+    DEFAULT_MAX_CALL_DURATION_SECONDS,
+    MAX_CALL_DURATION_SECONDS,
+    WorkflowConfigurationDefaults,
+)
+
+
+def test_max_call_duration_default_within_bounds():
+    config = WorkflowConfigurationDefaults()
+    assert config.max_call_duration == DEFAULT_MAX_CALL_DURATION_SECONDS
+
+
+def test_max_call_duration_accepts_cap():
+    config = WorkflowConfigurationDefaults(max_call_duration=MAX_CALL_DURATION_SECONDS)
+    assert config.max_call_duration == MAX_CALL_DURATION_SECONDS
+
+
+def test_max_call_duration_rejects_over_cap():
+    with pytest.raises(ValidationError):
+        WorkflowConfigurationDefaults(max_call_duration=MAX_CALL_DURATION_SECONDS + 1)
+
+
+def test_max_call_duration_rejects_non_positive():
+    with pytest.raises(ValidationError):
+        WorkflowConfigurationDefaults(max_call_duration=0)
+
+
+def test_null_values_treated_as_unset():
+    """Stored configs / older clients send explicit JSON nulls for keys the
+    user never configured; they must validate as defaults, not fail."""
+    config = WorkflowConfigurationDefaults.model_validate(
+        {
+            "max_call_duration": None,
+            "turn_start_strategy": None,
+            "turn_start_min_words": None,
+        }
+    )
+    assert config.max_call_duration == DEFAULT_MAX_CALL_DURATION_SECONDS
+    # Nulls count as unset, so a sparse round-trip drops them entirely.
+    assert config.model_dump(exclude_unset=True) == {}
+
+
+def test_exclude_unset_round_trip_stays_sparse():
+    config = WorkflowConfigurationDefaults.model_validate(
+        {"max_call_duration": 600, "custom_extra_key": {"a": 1}}
+    )
+    assert config.model_dump(exclude_unset=True) == {
+        "max_call_duration": 600,
+        "custom_extra_key": {"a": 1},
+    }
+
+
+def test_cap_stays_within_concurrency_stale_timeout():
+    """A call outliving the rate limiter's stale window has its concurrency
+    slot purged mid-call, so the cap must never exceed it."""
+    from api.services.campaign.rate_limiter import rate_limiter
+
+    assert MAX_CALL_DURATION_SECONDS <= rate_limiter.stale_call_timeout