Merge branch 'main' into feat/speaches-integration

api/db/campaign_client.py

@@ -365,12 +365,29 @@ class CampaignClient(BaseDBClient):
             result = await session.execute(query)
             return list(result.scalars().all())
 
-    async def get_completed_runs_for_report(self, campaign_id: int) -> list:
+    async def get_completed_runs_for_report(
+        self,
+        campaign_id: int,
+        start_date: Optional[datetime] = None,
+        end_date: Optional[datetime] = None,
+    ) -> list:
         """Get completed workflow runs for campaign report CSV.
 
         Returns rows with only the columns needed for report generation.
         """
         async with self.async_session() as session:
+            conditions = [
+                WorkflowRunModel.campaign_id == campaign_id,
+                WorkflowRunModel.is_completed.is_(True),
+                WorkflowRunModel.cost_info["call_duration_seconds"]
+                .as_string()
+                .isnot(None),
+            ]
+            if start_date is not None:
+                conditions.append(WorkflowRunModel.created_at >= start_date)
+            if end_date is not None:
+                conditions.append(WorkflowRunModel.created_at <= end_date)
+
             query = (
                 select(
                     WorkflowRunModel.id,
@@ -381,13 +398,7 @@ class CampaignClient(BaseDBClient):
                     WorkflowRunModel.logs,
                     WorkflowRunModel.public_access_token,
                 )
-                .where(
-                    WorkflowRunModel.campaign_id == campaign_id,
-                    WorkflowRunModel.is_completed.is_(True),
-                    WorkflowRunModel.cost_info["call_duration_seconds"]
-                    .as_string()
-                    .isnot(None),
-                )
+                .where(*conditions)
                 .order_by(WorkflowRunModel.created_at.desc())
             )
             result = await session.execute(query)

api/db/workflow_client.py

@@ -465,7 +465,7 @@ class WorkflowClient(BaseDBClient):
                 return
 
             existing = workflow.call_disposition_codes or {}
-            codes = existing.get("disposition_codes", [])
+            codes = list(existing.get("disposition_codes", []))
             if disposition_code in codes:
                 return
 

api/pyproject.toml

@@ -1,5 +1,5 @@
 [project]
 name = "dograh-api"
-version = "1.19.1"
+version = "1.19.2"
 description = "Backend API for Dograh voice AI platform"
 requires-python = ">=3.12"

api/routes/campaign.py

@@ -706,13 +706,21 @@ async def get_campaign_source_download_url(
 async def download_campaign_report(
     campaign_id: int,
     user: UserModel = Depends(get_user),
+    start_date: Optional[datetime] = Query(
+        None, description="Filter runs created on or after this datetime (ISO 8601)"
+    ),
+    end_date: Optional[datetime] = Query(
+        None, description="Filter runs created on or before this datetime (ISO 8601)"
+    ),
 ) -> StreamingResponse:
     """Download a CSV report of completed campaign runs."""
     campaign = await db_client.get_campaign(campaign_id, user.selected_organization_id)
     if not campaign:
         raise HTTPException(status_code=404, detail="Campaign not found")
 
-    output, filename = await generate_campaign_report_csv(campaign_id)
+    output, filename = await generate_campaign_report_csv(
+        campaign_id, start_date=start_date, end_date=end_date
+    )
 
     return StreamingResponse(
         output,

api/routes/user.py

@@ -125,7 +125,11 @@ async def update_user_configurations(
 
     try:
         validator = UserConfigurationValidator()
-        await validator.validate(user_configurations)
+        await validator.validate(
+            user_configurations,
+            organization_id=user.selected_organization_id,
+            created_by=user.provider_id,
+        )
     except ValueError as e:
         raise HTTPException(status_code=422, detail=e.args[0])
 
@@ -163,7 +167,11 @@ async def validate_user_configurations(
     ):
         validator = UserConfigurationValidator()
         try:
-            status = await validator.validate(configurations)
+            status = await validator.validate(
+                configurations,
+                organization_id=user.selected_organization_id,
+                created_by=user.provider_id,
+            )
             await db_client.update_user_configuration_last_validated_at(user.id)
             return status
         except ValueError as e:

api/routes/webrtc_signaling.py

@@ -228,7 +228,7 @@ class SignalingManager:
                 {
                     "type": "error",
                     "payload": {
-                        "error_type": "quota_exceeded",
+                        "error_type": quota_result.error_code,
                         "message": quota_result.error_message,
                     },
                 }

api/services/campaign/report.py

@@ -1,6 +1,7 @@
 import csv
 import io
-from typing import Any, List
+from datetime import datetime
+from typing import Any, List, Optional
 
 from api.constants import BACKEND_API_ENDPOINT
 from api.db import db_client
@@ -27,12 +28,18 @@ def _collect_extracted_variable_keys(runs: List[Any]) -> list[str]:
     return list(keys)
 
 
-async def generate_campaign_report_csv(campaign_id: int) -> tuple[io.StringIO, str]:
+async def generate_campaign_report_csv(
+    campaign_id: int,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+) -> tuple[io.StringIO, str]:
     """Generate a CSV report for a campaign.
 
     Returns a tuple of (csv_output, filename).
     """
-    runs = await db_client.get_completed_runs_for_report(campaign_id)
+    runs = await db_client.get_completed_runs_for_report(
+        campaign_id, start_date=start_date, end_date=end_date
+    )
 
     # Collect dynamic extracted variable columns
     extracted_var_keys = _collect_extracted_variable_keys(runs)

api/services/configuration/check_validity.py

@@ -14,6 +14,12 @@ from api.schemas.user_configuration import (
 from api.services.configuration.registry import ServiceConfig, ServiceProviders
 from api.services.mps_service_key_client import mps_service_key_client
 
+AuthContext = TypedDict(
+    "AuthContext",
+    {"organization_id": Optional[int], "created_by": Optional[str]},
+    total=False,
+)
+
 
 class APIKeyStatus(TypedDict):
     model: str
@@ -43,7 +49,16 @@ class UserConfigurationValidator:
             ServiceProviders.SPEACHES.value: self._check_speaches_api_key,
         }
 
-    async def validate(self, configuration: UserConfiguration) -> APIKeyStatusResponse:
+    async def validate(
+        self,
+        configuration: UserConfiguration,
+        organization_id: Optional[int] = None,
+        created_by: Optional[str] = None,
+    ) -> APIKeyStatusResponse:
+        self._auth_context: AuthContext = {
+            "organization_id": organization_id,
+            "created_by": created_by,
+        }
         status_list = []
 
         status_list.extend(self._validate_service(configuration.llm, "llm"))
@@ -165,7 +180,12 @@ class UserConfigurationValidator:
                 "You provided a Dograh API key (dgr...) instead of a service key. "
                 "Please use a service key (mps...)."
             )
-        return mps_service_key_client.validate_service_key(api_key)
+        auth = getattr(self, "_auth_context", {})
+        return mps_service_key_client.validate_service_key(
+            api_key,
+            organization_id=auth.get("organization_id"),
+            created_by=auth.get("created_by"),
+        )
 
     def _check_sarvam_api_key(self, model: str, api_key: str) -> bool:
         return True

api/services/mps_service_key_client.py

@@ -276,7 +276,7 @@ class MPSServiceKeyClient:
                     "remaining_credits": data.get("remaining_credits", 0.0),
                 }
             else:
-                logger.error(
+                logger.warning(
                     f"Failed to check service key usage: {response.status_code} - {response.text}"
                 )
                 raise httpx.HTTPStatusError(
@@ -416,7 +416,12 @@ class MPSServiceKeyClient:
                     response=response,
                 )
 
-    def validate_service_key(self, service_key: str) -> bool:
+    def validate_service_key(
+        self,
+        service_key: str,
+        organization_id: Optional[int] = None,
+        created_by: Optional[str] = None,
+    ) -> bool:
         """
         Synchronously validate a Dograh service key by checking usage via MPS.
 
@@ -427,7 +432,7 @@ class MPSServiceKeyClient:
                 response = client.post(
                     f"{self.base_url}/api/v1/service-keys/usage",
                     json={"service_key": service_key},
-                    headers=self._get_headers(),
+                    headers=self._get_headers(organization_id, created_by),
                 )
                 return response.status_code == 200
         except Exception:

api/services/quota_service.py

@@ -20,6 +20,7 @@ class QuotaCheckResult:
 
     has_quota: bool
     error_message: str = ""
+    error_code: str = ""
 
 
 async def check_dograh_quota(user: UserModel) -> QuotaCheckResult:
@@ -76,6 +77,7 @@ async def check_dograh_quota(user: UserModel) -> QuotaCheckResult:
                     )
                     return QuotaCheckResult(
                         has_quota=False,
+                        error_code="quota_exceeded",
                         error_message=(
                             "You have exhausted your trial credits. "
                             "Please email founders@dograh.com for additional Dograh credits "
@@ -89,8 +91,16 @@ async def check_dograh_quota(user: UserModel) -> QuotaCheckResult:
                 )
             except Exception as e:
                 logger.error(f"Failed to check quota for Dograh key: {str(e)}")
+                error_str = str(e)
+                if "404" in error_str or "not found" in error_str.lower():
+                    return QuotaCheckResult(
+                        has_quota=False,
+                        error_code="invalid_service_key",
+                        error_message="You have invalid keys in your model configuration. Please validate the service keys.",
+                    )
                 return QuotaCheckResult(
                     has_quota=False,
+                    error_code="quota_check_failed",
                     error_message="Could not verify Dograh credits. Please try again.",
                 )
 

api/tests/test_add_call_disposition_code.py

@@ -0,0 +1,85 @@
+"""Test that add_call_disposition_code correctly persists changes.
+
+The bug: `codes` is a reference to the list inside the JSON column value.
+Calling `codes.append()` mutates the in-memory column value in-place.
+When SQLAlchemy compares old vs new on commit, it sees them as equal
+because the old value was already mutated — so the change is silently dropped.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from api.db.workflow_client import WorkflowClient
+
+
+def _make_workflow_stub(initial_disposition_codes):
+    """Create a mock workflow that behaves like a SQLAlchemy model instance.
+
+    Tracks attribute assignments so we can verify the new value is genuinely
+    different from the original (which is what SQLAlchemy needs to detect a change).
+    """
+    workflow = MagicMock()
+    # Store the initial value and track what gets assigned
+    workflow.call_disposition_codes = initial_disposition_codes
+    workflow._assigned_values = {}
+
+    original_setattr = type(workflow).__setattr__
+
+    def tracking_setattr(self, name, value):
+        if name == "call_disposition_codes":
+            self._assigned_values[name] = value
+        original_setattr(self, name, value)
+
+    type(workflow).__setattr__ = tracking_setattr
+    return workflow
+
+
+@pytest.fixture
+def client():
+    with patch("api.db.workflow_client.BaseDBClient.__init__", return_value=None):
+        c = WorkflowClient()
+        c.async_session = MagicMock()
+        return c
+
+
+def test_disposition_code_new_value_is_not_same_reference(client):
+    """The assigned list must NOT be the same object as the original.
+
+    If it is, SQLAlchemy won't detect the change because old == new
+    (the old was mutated in-place).
+    """
+    initial_codes = {"disposition_codes": ["existing_code"]}
+    original_list = initial_codes["disposition_codes"]
+
+    workflow = MagicMock()
+    workflow.call_disposition_codes = initial_codes
+
+    # Mock the session and query
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.first.return_value = workflow
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    mock_session.commit = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=False)
+
+    client.async_session = MagicMock(return_value=mock_session)
+
+    asyncio.get_event_loop().run_until_complete(
+        client.add_call_disposition_code(workflow_id=1, disposition_code="new_code")
+    )
+
+    # Verify the disposition code was added
+    assigned = workflow.call_disposition_codes
+    assert "new_code" in assigned["disposition_codes"]
+
+    # THE CRITICAL CHECK: the list inside the assigned value must be a *different*
+    # object from the original list. If it's the same object, SQLAlchemy's change
+    # detection won't work because the "old" value was mutated in-place.
+    assert assigned["disposition_codes"] is not original_list, (
+        "The assigned disposition_codes list is the same object as the original. "
+        "This means SQLAlchemy won't detect the change because the old value "
+        "was mutated in-place via list.append()."
+    )