apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-06-07 07:55:16 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: number pool initialization in multi telephony setup

Browse source 
If there are multiple telephony configurations, the form number should be initialized from the campaigns given telephonic configuration rather than the organization default telephonic configuration.
This commit is contained in:
Abhishek Kumar 2026-05-08 14:48:53 +05:30
parent 81a363b06e
commit 6d93be3ef6
31 changed files with 1105 additions and 238 deletions
Download patch file Download diff file Expand all files Collapse all files

3
api/.env.example
View file

@ -31,7 +31,8 @@ MINIO_BUCKET=voice-audio
MINIO_SECURE=false
# Tracing and Analytics using Langfuse
ENABLE_TRACING=false
# Credentials can be set here or per-organization via the UI at /settings.
# Tracing is automatically active when credentials are available.
# LANGFUSE_SECRET_KEY="sk-lf-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
# LANGFUSE_PUBLIC_KEY="pk-lf-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
# LANGFUSE_HOST="https://cloud.langfuse.com"

3
api/constants.py
View file

@ -13,9 +13,6 @@ FILLER_SOUND_PROBABILITY = 0.0
VOICEMAIL_RECORDING_DURATION = 5.0
# Configuration constants
ENABLE_TRACING = os.getenv("ENABLE_TRACING", "false").lower() == "true"
# Langfuse Configuration
LANGFUSE_HOST = os.getenv("LANGFUSE_HOST")
LANGFUSE_PUBLIC_KEY = os.getenv("LANGFUSE_PUBLIC_KEY")

42
api/db/filters.py
View file

@ -3,7 +3,7 @@
from datetime import datetime
from typing import Any, Dict, List, Optional
from sqlalchemy import Float, Integer, and_, cast, func
from sqlalchemy import Float, Integer, Text, and_, cast, func
from sqlalchemy.dialects.postgresql import JSONB
from api.db.models import WorkflowRunModel
@ -48,8 +48,10 @@ ATTRIBUTE_FIELD_MAPPING = {
"tokenUsage": "cost_info.total_cost_usd",
"runId": "id",
"workflowId": "workflow_id",
"campaignId": "campaign_id",
"callTags": "gathered_context.call_tags",
"phoneNumber": "initial_context.phone",
"callerNumber": "initial_context.caller_number",
"calledNumber": "initial_context.called_number",
}
@ -69,7 +71,8 @@ def apply_workflow_run_filters(
- runId: Filter by workflow run ID (exact match)
- workflowId: Filter by workflow ID (exact match)
- callTags: Filter by gathered_context.call_tags (array of strings)
- phoneNumber: Filter by initial_context.phone (text search)
- callerNumber: Filter by initial_context.caller_number (text search)
- calledNumber: Filter by initial_context.called_number (text search)
Args:
base_query: The base SQLAlchemy query to apply filters to
@ -119,6 +122,12 @@ def apply_workflow_run_filters(
WorkflowRunModel.workflow_id == value["value"]
)
elif filter_type == "number" and field == "campaign_id":
if value.get("value") is not None:
filter_conditions.append(
WorkflowRunModel.campaign_id == value["value"]
)
elif filter_type == "dateRange" and field == "created_at":
# Same as attribute-based dateRange
if value.get("from"):
@ -169,15 +178,30 @@ def apply_workflow_run_filters(
call_tags = gathered_context_jsonb.op("->")("call_tags")
filter_conditions.append(call_tags.op("@>")(func.cast(tags, JSONB)))
elif filter_type == "text" and field == "initial_context.phone":
# Filter by phone number (contains search)
elif filter_type == "text" and field == "initial_context.caller_number":
phone = value.get("value", "").strip()
if phone:
# Use ->> operator for compatibility with all PostgreSQL versions
# Cast ->> result to Text so .contains() emits LIKE,
# not the JSONB @> operator (the default for untyped exprs).
filter_conditions.append(
cast(WorkflowRunModel.initial_context, JSONB)
.op("->>")("phone")
.contains(phone)
cast(
cast(WorkflowRunModel.initial_context, JSONB).op("->>")(
"caller_number"
),
Text,
).contains(phone)
)
elif filter_type == "text" and field == "initial_context.called_number":
phone = value.get("value", "").strip()
if phone:
filter_conditions.append(
cast(
cast(WorkflowRunModel.initial_context, JSONB).op("->>")(
"called_number"
),
Text,
).contains(phone)
)
elif filter_type == "numberRange":

82
api/db/organization_usage_client.py
View file

@ -245,7 +245,7 @@ class OrganizationUsageClient(BaseDBClient):
limit: int = 50,
offset: int = 0,
filters: Optional[list[dict]] = None,
) -> tuple[list[dict], int]:
) -> tuple[list[dict], int, float, int]:
"""Get paginated workflow runs with usage for an organization."""
async with self.async_session() as session:
query = (
@ -267,7 +267,15 @@ class OrganizationUsageClient(BaseDBClient):
# Only allow specific filters for usage history endpoint
# This ensures security and prevents unexpected filter attributes
allowed_filters = {"duration", "dispositionCode", "phoneNumber"}
allowed_filters = {
"duration",
"dispositionCode",
"callerNumber",
"calledNumber",
"runId",
"workflowId",
"campaignId",
}
sanitized_filters = []
if filters:
@ -315,13 +323,15 @@ class OrganizationUsageClient(BaseDBClient):
total_tokens += dograh_tokens
total_duration_seconds += int(round(call_duration))
# Extract phone number from initial_context based on call_type.
ic = run.initial_context or {}
caller_number = ic.get("caller_number")
called_number = ic.get("called_number") or ic.get("phone_number")
# DEPRECATED: phone_number — use caller_number/called_number.
# Inbound runs only have caller_number/called_number; the
# caller_number is the customer. Outbound runs use the
# phone_number key written by the dispatchers.
ic = run.initial_context or {}
if run.call_type == "inbound":
phone_number = ic.get("caller_number")
phone_number = caller_number
else:
phone_number = ic.get("phone_number")
@ -341,6 +351,8 @@ class OrganizationUsageClient(BaseDBClient):
"recording_url": run.recording_url,
"transcript_url": run.transcript_url,
"phone_number": phone_number,
"caller_number": caller_number,
"called_number": called_number,
"call_type": run.call_type,
"disposition": disposition,
"initial_context": run.initial_context,
@ -355,6 +367,66 @@ class OrganizationUsageClient(BaseDBClient):
return formatted_runs, total_count, total_tokens, total_duration_seconds
async def get_usage_runs_for_report(
self,
organization_id: int,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
filters: Optional[list[dict]] = None,
) -> list:
"""Get filtered runs for an organization-scoped usage CSV report.
Mirrors the filter allowlist used by `get_usage_history`, but selects
only the columns needed by `build_run_report_csv` and returns every
matching run (no pagination).
"""
async with self.async_session() as session:
query = (
select(
WorkflowRunModel.id,
WorkflowRunModel.workflow_id,
WorkflowRunModel.definition_id,
WorkflowRunModel.campaign_id,
WorkflowRunModel.created_at,
WorkflowRunModel.initial_context,
WorkflowRunModel.gathered_context,
WorkflowRunModel.cost_info,
WorkflowRunModel.public_access_token,
)
.join(WorkflowModel, WorkflowRunModel.workflow_id == WorkflowModel.id)
.join(UserModel, WorkflowModel.user_id == UserModel.id)
.where(
UserModel.selected_organization_id == organization_id,
WorkflowRunModel.cost_info.isnot(None),
)
.order_by(WorkflowRunModel.created_at.desc())
)
if start_date:
query = query.where(WorkflowRunModel.created_at >= start_date)
if end_date:
query = query.where(WorkflowRunModel.created_at <= end_date)
allowed_filters = {
"duration",
"dispositionCode",
"callerNumber",
"calledNumber",
"runId",
"workflowId",
"campaignId",
}
sanitized_filters = []
if filters:
for filter_item in filters:
if filter_item.get("attribute") in allowed_filters:
sanitized_filters.append(filter_item)
query = apply_workflow_run_filters(query, sanitized_filters)
result = await session.execute(query)
return list(result.all())
async def get_daily_usage_breakdown(
self,
organization_id: int,

2
api/routes/campaign.py
View file

@ -15,11 +15,11 @@ from api.db import db_client
from api.db.models import UserModel
from api.enums import OrganizationConfigurationKey
from api.services.auth.depends import get_user
from api.services.campaign.report import generate_campaign_report_csv
from api.services.campaign.runner import campaign_runner_service
from api.services.campaign.source_sync import CampaignSourceSyncService
from api.services.campaign.source_sync_factory import get_sync_service
from api.services.quota_service import check_dograh_quota
from api.services.reports import generate_campaign_report_csv
from api.services.storage import storage_fs
router = APIRouter(prefix="/campaign")

104
api/routes/organization_usage.py
View file

@ -3,14 +3,16 @@ from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from loguru import logger
from pydantic import BaseModel
from pydantic import BaseModel, Field
from api.constants import DEPLOYMENT_MODE
from api.db import db_client
from api.db.models import UserModel
from api.services.auth.depends import get_user
from api.services.mps_service_key_client import mps_service_key_client
from api.services.reports import generate_usage_runs_report_csv
router = APIRouter(prefix="/organizations")
@ -47,7 +49,13 @@ class WorkflowRunUsageResponse(BaseModel):
call_duration_seconds: int
recording_url: Optional[str] = None
transcript_url: Optional[str] = None
phone_number: Optional[str] = None
phone_number: Optional[str] = Field(
default=None,
deprecated=True,
description="Deprecated. Use caller_number and called_number instead.",
)
caller_number: Optional[str] = None
called_number: Optional[str] = None
call_type: Optional[str] = None
disposition: Optional[str] = None
initial_context: Optional[Dict[str, Any]] = None
@ -129,13 +137,55 @@ async def get_mps_credits(user: UserModel = Depends(get_user)):
raise HTTPException(status_code=500, detail=str(e))
FILTERS_DESCRIPTION = """\
JSON-encoded array of filter objects. Each object has the shape:
```json
{ "attribute": "<name>", "type": "<type>", "value": <value> }
```
Supported `attribute` / `type` / `value` combinations:
| attribute | type | value shape | matches |
|-----------------|---------------|----------------------------------------------|------------------------------------------------------|
| `runId` | `number` | `{ "value": 12345 }` | exact run id |
| `workflowId` | `number` | `{ "value": 42 }` | exact agent (workflow) id |
| `campaignId` | `number` | `{ "value": 7 }` | exact campaign id |
| `callerNumber` | `text` | `{ "value": "415555" }` | substring match on `initial_context.caller_number` |
| `calledNumber` | `text` | `{ "value": "9911848" }` | substring match on `initial_context.called_number` |
| `dispositionCode` | `multiSelect` | `{ "codes": ["XFER", "DNC"] }` | any of the codes in `gathered_context.mapped_call_disposition` |
| `duration` | `numberRange` | `{ "min": 60, "max": 300 }` | call duration (seconds), inclusive bounds |
Unknown attributes and unsupported `type` values are silently ignored.
Date filtering on this endpoint is done via the dedicated `start_date` / `end_date` query params, not via a `dateRange` filter object.
"""
@router.get("/usage/runs", response_model=UsageHistoryResponse)
async def get_usage_history(
start_date: Optional[str] = Query(None, description="ISO format date string"),
end_date: Optional[str] = Query(None, description="ISO format date string"),
start_date: Optional[str] = Query(
None,
description="ISO 8601 date-time string (UTC). Lower bound (inclusive) on `created_at`.",
examples=["2026-04-01T00:00:00Z"],
),
end_date: Optional[str] = Query(
None,
description="ISO 8601 date-time string (UTC). Upper bound (inclusive) on `created_at`.",
examples=["2026-05-01T00:00:00Z"],
),
page: int = Query(1, ge=1),
limit: int = Query(50, ge=1, le=100),
filters: Optional[str] = Query(None, description="JSON string of filters"),
filters: Optional[str] = Query(
None,
description=FILTERS_DESCRIPTION,
examples=[
'[{"attribute":"callerNumber","type":"text","value":{"value":"415555"}}]',
'[{"attribute":"campaignId","type":"number","value":{"value":7}},'
'{"attribute":"duration","type":"numberRange","value":{"min":60,"max":300}}]',
'[{"attribute":"dispositionCode","type":"multiSelect","value":{"codes":["XFER","DNC"]}}]',
],
),
user: UserModel = Depends(get_user),
):
"""Get paginated workflow runs with usage for the organization."""
@ -185,6 +235,50 @@ async def get_usage_history(
raise HTTPException(status_code=500, detail=str(e))
@router.get("/usage/runs/report")
async def download_usage_runs_report(
start_date: Optional[str] = Query(
None,
description="ISO 8601 date-time string (UTC). Lower bound (inclusive) on `created_at`.",
),
end_date: Optional[str] = Query(
None,
description="ISO 8601 date-time string (UTC). Upper bound (inclusive) on `created_at`.",
),
filters: Optional[str] = Query(
None,
description=FILTERS_DESCRIPTION,
),
user: UserModel = Depends(get_user),
) -> StreamingResponse:
"""Download a CSV of runs matching the same filters as `/usage/runs`."""
if not user.selected_organization_id:
raise HTTPException(status_code=400, detail="No organization selected")
start_dt = datetime.fromisoformat(start_date) if start_date else None
end_dt = datetime.fromisoformat(end_date) if end_date else None
parsed_filters = None
if filters:
try:
parsed_filters = json.loads(filters)
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="Invalid filters format")
output, filename = await generate_usage_runs_report_csv(
user.selected_organization_id,
start_date=start_dt,
end_date=end_dt,
filters=parsed_filters,
)
return StreamingResponse(
output,
media_type="text/csv",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.get("/usage/daily-breakdown", response_model=DailyUsageBreakdownResponse)
async def get_daily_usage_breakdown(
days: int = Query(7, ge=1, le=30, description="Number of days to include"),

21
api/routes/telephony.py
View file

@ -55,6 +55,9 @@ class InitiateCallRequest(BaseModel):
# Optional explicit telephony config to use for the test call. If omitted,
# falls back to the user's per-user default (when set), then the org default.
telephony_configuration_id: int | None = None
# Optional caller-ID phone number to dial out from. Must belong to the
# resolved telephony configuration; otherwise the provider picks one.
from_phone_number_id: int | None = None
@router.post(
@ -173,11 +176,29 @@ async def initiate_call(
keywords = {"workflow_id": request.workflow_id, "user_id": user.id}
# Resolve optional caller-ID. The config has already been validated against
# the user's organization, so filtering by config_id is sufficient for
# tenant isolation.
from_number: str | None = None
if request.from_phone_number_id is not None:
if telephony_configuration_id is None:
raise HTTPException(
status_code=400,
detail="from_phone_number_id_requires_telephony_configuration",
)
phone_row = await db_client.get_phone_number_for_config(
request.from_phone_number_id, telephony_configuration_id
)
if not phone_row or not phone_row.is_active:
raise HTTPException(status_code=400, detail="from_phone_number_not_found")
from_number = phone_row.address_normalized
# Initiate call via provider
result = await provider.initiate_call(
to_number=phone_number,
webhook_url=webhook_url,
workflow_run_id=workflow_run_id,
from_number=from_number,
**keywords,
)

2
api/routes/workflow.py
View file

@ -19,7 +19,6 @@ from api.enums import CallType, PostHogEvent, StorageBackend
from api.schemas.workflow import WorkflowRunResponseSchema
from api.sdk_expose import sdk_expose
from api.services.auth.depends import get_user
from api.services.campaign.report import generate_workflow_report_csv
from api.services.configuration.check_validity import UserConfigurationValidator
from api.services.configuration.masking import (
mask_workflow_definition,
@ -28,6 +27,7 @@ from api.services.configuration.masking import (
from api.services.configuration.resolve import resolve_effective_config
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.dto import ReactFlowDTO, sanitize_workflow_definition
from api.services.workflow.duplicate import duplicate_workflow

59
api/services/campaign/campaign_call_dispatcher.py
View file

@ -106,7 +106,9 @@ class CampaignCallDispatcher:
provider = await self.get_provider_for_campaign(campaign)
if provider.from_numbers:
await rate_limiter.initialize_from_number_pool(
campaign.organization_id, provider.from_numbers
campaign.organization_id,
provider.from_numbers,
telephony_configuration_id=campaign.telephony_configuration_id,
)
except Exception as e:
logger.warning(f"Failed to initialize from_number pool: {e}")
@ -210,8 +212,13 @@ class CampaignCallDispatcher:
provider = await self.get_provider_for_campaign(campaign)
workflow_run_mode = provider.PROVIDER_NAME
# Acquire a unique from_number from the pool
from_number = await self.acquire_from_number(campaign.organization_id)
# Acquire a unique from_number from the pool scoped to this campaign's
# telephony configuration so orgs with multiple configs don't leak
# caller IDs across configs.
from_number = await self.acquire_from_number(
campaign.organization_id,
telephony_configuration_id=campaign.telephony_configuration_id,
)
if from_number is None:
# Release concurrent slot before raising
await rate_limiter.release_concurrent_slot(
@ -257,7 +264,10 @@ class CampaignCallDispatcher:
# Store from_number mapping for cleanup on call completion
await rate_limiter.store_workflow_from_number_mapping(
workflow_run.id, campaign.organization_id, from_number
workflow_run.id,
campaign.organization_id,
from_number,
telephony_configuration_id=campaign.telephony_configuration_id,
)
except Exception as e:
# Release slot and from_number on error
@ -266,7 +276,9 @@ class CampaignCallDispatcher:
)
if from_number:
await rate_limiter.release_from_number(
campaign.organization_id, from_number
campaign.organization_id,
from_number,
telephony_configuration_id=campaign.telephony_configuration_id,
)
raise
@ -364,8 +376,10 @@ class CampaignCallDispatcher:
workflow_run.id
)
if from_number_mapping:
fn_org_id, fn_number = from_number_mapping
await rate_limiter.release_from_number(fn_org_id, fn_number)
fn_org_id, fn_number, fn_tcid = from_number_mapping
await rate_limiter.release_from_number(
fn_org_id, fn_number, telephony_configuration_id=fn_tcid
)
await rate_limiter.delete_workflow_from_number_mapping(workflow_run.id)
raise
@ -464,23 +478,24 @@ class CampaignCallDispatcher:
await asyncio.sleep(1)
async def acquire_from_number(
self, organization_id: int, timeout: float = 600
self,
organization_id: int,
telephony_configuration_id: int | None,
timeout: float = 600,
) -> Optional[str]:
"""
Acquire a from_number from the pool with retry.
Acquire a from_number from the (org, telephony config) pool with retry.
Waits up to timeout seconds, polling every 1s.
Args:
organization_id: ID of the organization for which to acquire the from_number.
timeout: Maximum time in seconds to wait for a from_number before giving up.
Returns:
The acquired phone number as a string, or None if timeout is exceeded.
"""
wait_start = time.time()
while True:
from_number = await rate_limiter.acquire_from_number(organization_id)
from_number = await rate_limiter.acquire_from_number(
organization_id, telephony_configuration_id
)
if from_number:
return from_number
@ -488,13 +503,15 @@ class CampaignCallDispatcher:
if wait_time > timeout:
logger.warning(
f"From number pool exhausted for org {organization_id} "
f"after waiting {wait_time:.1f}s"
f"config {telephony_configuration_id} after waiting "
f"{wait_time:.1f}s"
)
return None
logger.debug(
f"All from_numbers in use for org {organization_id}, "
f"waited {wait_time:.1f}s, retrying..."
f"All from_numbers in use for org {organization_id} "
f"config {telephony_configuration_id}, waited {wait_time:.1f}s, "
"retrying..."
)
await asyncio.sleep(1)
@ -515,13 +532,15 @@ class CampaignCallDispatcher:
)
slot_released = True
# Release from_number back to pool
# Release from_number back to its (org, telephony config) pool
from_number_mapping = await rate_limiter.get_workflow_from_number_mapping(
workflow_run_id
)
if from_number_mapping:
fn_org_id, fn_number = from_number_mapping
fn_success = await rate_limiter.release_from_number(fn_org_id, fn_number)
fn_org_id, fn_number, fn_tcid = from_number_mapping
fn_success = await rate_limiter.release_from_number(
fn_org_id, fn_number, telephony_configuration_id=fn_tcid
)
if fn_success:
await rate_limiter.delete_workflow_from_number_mapping(workflow_run_id)
logger.info(

74
api/services/campaign/rate_limiter.py
View file

@ -247,22 +247,31 @@ class RateLimiter:
# ======== FROM NUMBER POOL METHODS ========
@staticmethod
def _from_number_pool_key(
organization_id: int, telephony_configuration_id: int | None
) -> str:
return f"from_number_pool:{organization_id}:{telephony_configuration_id}"
async def initialize_from_number_pool(
self, organization_id: int, from_numbers: list[str]
self,
organization_id: int,
from_numbers: list[str],
telephony_configuration_id: int | None,
) -> bool:
"""
Initialize the from_number pool for an organization.
Initialize the from_number pool for an organization + telephony config.
Uses ZADD NX so it won't overwrite numbers that are already in use.
Args:
organization_id: The organization ID
from_numbers: List of phone numbers to add to the pool
Pools are scoped per (organization_id, telephony_configuration_id) so
that orgs with multiple telephony configurations do not leak caller IDs
across configs.
"""
if not from_numbers:
return False
redis_client = await self._get_redis()
key = f"from_number_pool:{organization_id}"
key = self._from_number_pool_key(organization_id, telephony_configuration_id)
try:
# ZADD NX: only add members that don't already exist (preserves in-use scores)
@ -274,15 +283,18 @@ class RateLimiter:
logger.error(f"Error initializing from_number pool: {e}")
return False
async def acquire_from_number(self, organization_id: int) -> Optional[str]:
async def acquire_from_number(
self, organization_id: int, telephony_configuration_id: int | None
) -> Optional[str]:
"""
Atomically acquire an available from_number from the pool.
Atomically acquire an available from_number from the pool for the given
(organization_id, telephony_configuration_id).
Cleans stale entries (score > 0 and older than 30 min) before acquiring.
Returns the phone number if available, None if all numbers are in use.
"""
redis_client = await self._get_redis()
key = f"from_number_pool:{organization_id}"
key = self._from_number_pool_key(organization_id, telephony_configuration_id)
now = time.time()
stale_cutoff = now - self.stale_call_timeout
@ -321,16 +333,21 @@ class RateLimiter:
logger.error(f"Error acquiring from_number: {e}")
return None
async def release_from_number(self, organization_id: int, from_number: str) -> bool:
async def release_from_number(
self,
organization_id: int,
from_number: str,
telephony_configuration_id: int | None,
) -> bool:
"""
Release a from_number back to the pool by setting its score to 0.
Harmless if already released (score already 0).
Release a from_number back to its (org, telephony config) pool by
setting its score to 0. Harmless if already released (score already 0).
"""
if not from_number:
return False
redis_client = await self._get_redis()
key = f"from_number_pool:{organization_id}"
key = self._from_number_pool_key(organization_id, telephony_configuration_id)
lua_script = """
local key = KEYS[1]
@ -356,19 +373,33 @@ class RateLimiter:
return False
async def store_workflow_from_number_mapping(
self, workflow_run_id: int, organization_id: int, from_number: str
self,
workflow_run_id: int,
organization_id: int,
from_number: str,
telephony_configuration_id: int | None,
) -> bool:
"""
Store the mapping between workflow_run_id and its from_number.
Used for cleanup when calls complete.
Store the mapping between workflow_run_id and its from_number, plus
the telephony_configuration_id so cleanup can release back to the
correct pool.
"""
redis_client = await self._get_redis()
mapping_key = f"workflow_from_number:{workflow_run_id}"
try:
# Redis hashes can't store None — use empty string sentinel for legacy
# campaigns whose telephony_configuration_id has not been backfilled.
tcid_value = (
"" if telephony_configuration_id is None else telephony_configuration_id
)
await redis_client.hset(
mapping_key,
mapping={"org_id": organization_id, "from_number": from_number},
mapping={
"org_id": organization_id,
"from_number": from_number,
"telephony_configuration_id": tcid_value,
},
)
await redis_client.expire(mapping_key, 1800) # 30 min TTL
return True
@ -378,10 +409,11 @@ class RateLimiter:
async def get_workflow_from_number_mapping(
self, workflow_run_id: int
) -> Optional[tuple[int, str]]:
) -> Optional[tuple[int, str, int | None]]:
"""
Get the from_number mapping for a workflow run.
Returns (organization_id, from_number) tuple or None if not found.
Returns (organization_id, from_number, telephony_configuration_id) or
None if not found. telephony_configuration_id is None for legacy entries.
"""
redis_client = await self._get_redis()
mapping_key = f"workflow_from_number:{workflow_run_id}"
@ -389,7 +421,9 @@ class RateLimiter:
try:
mapping = await redis_client.hgetall(mapping_key)
if mapping and "org_id" in mapping and "from_number" in mapping:
return (int(mapping["org_id"]), mapping["from_number"])
raw_tcid = mapping.get("telephony_configuration_id", "")
tcid = int(raw_tcid) if raw_tcid not in (None, "") else None
return (int(mapping["org_id"]), mapping["from_number"], tcid)
return None
except Exception as e:
logger.error(f"Error getting workflow from_number mapping: {e}")

5
api/services/pipecat/pipeline_builder.py
View file

@ -2,9 +2,6 @@ import os
from loguru import logger
from api.constants import (
ENABLE_TRACING,
)
from api.services.pipecat.audio_config import AudioConfig
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineParams, PipelineTask
@ -178,7 +175,7 @@ def create_pipeline_task(pipeline, workflow_run_id, audio_config: AudioConfig =
task = PipelineTask(
pipeline,
params=pipeline_params,
enable_tracing=ENABLE_TRACING,
enable_tracing=True,
enable_rtvi=False,
conversation_id=f"{workflow_run_id}",
)

15
api/services/pipecat/tracing_config.py
View file

@ -6,7 +6,6 @@ from opentelemetry.sdk.trace import SpanProcessor
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
from api.constants import (
ENABLE_TRACING,
LANGFUSE_HOST,
LANGFUSE_PUBLIC_KEY,
LANGFUSE_SECRET_KEY,
@ -138,10 +137,12 @@ class _OrgRoutingExporter(SpanExporter):
def ensure_tracing() -> bool:
"""Initialize OTEL tracing if enabled. Returns True if tracing is available.
"""Initialize OTEL tracing. Returns True once the routing exporter is set up.
Installs an ``_OrgRoutingExporter`` so that spans can be routed to
org-specific Langfuse projects at export time.
org-specific Langfuse projects at export time. Spans without a matching
exporter (no env-var defaults, no registered org) are silently dropped, so
this is safe to call unconditionally.
Idempotent — safe to call from both the pipeline process and the ARQ worker.
"""
@ -149,9 +150,6 @@ def ensure_tracing() -> bool:
if _tracing_initialized:
return True
if not ENABLE_TRACING:
return False
# Build the default exporter from env-var credentials (may be None)
default_exporter = None
if all([LANGFUSE_HOST, LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY]):
@ -162,11 +160,6 @@ def ensure_tracing() -> bool:
endpoint=f"{LANGFUSE_HOST}/api/public/otel/v1/traces",
headers={"Authorization": f"Basic {langfuse_auth}"},
)
else:
logger.warning(
"ENABLE_TRACING is true but default Langfuse credentials are not configured. "
"Only org-level credentials will be used."
)
_org_routing_exporter = _OrgRoutingExporter(default_exporter)
setup_tracing(service_name="dograh-pipeline", exporter=_org_routing_exporter)

14
api/services/reports/__init__.py
View file

@ -1,3 +1,15 @@
from .daily_report import DailyReportService
from .run_report import (
build_run_report_csv,
generate_campaign_report_csv,
generate_usage_runs_report_csv,
generate_workflow_report_csv,
)
__all__ = ["DailyReportService"]
__all__ = [
"DailyReportService",
"build_run_report_csv",
"generate_campaign_report_csv",
"generate_usage_runs_report_csv",
"generate_workflow_report_csv",
]

40
api/services/campaign/report.py → api/services/reports/run_report.py
View file

@ -1,6 +1,13 @@
"""CSV reports built from completed workflow runs.
Shared by campaign-, workflow-, and organization-usage-scoped reports.
The DB client supplies the row set; this module owns the column layout
so every endpoint emits the same shape.
"""
import csv
import io
from datetime import datetime
from datetime import UTC, datetime
from typing import Any, List, Optional
from api.constants import BACKEND_API_ENDPOINT
@ -25,11 +32,8 @@ def _collect_extracted_variable_keys(runs: List[Any]) -> list[str]:
return list(keys)
def _build_run_report_csv(runs: List[Any]) -> io.StringIO:
"""Build a CSV from completed workflow runs.
Shared between campaign-scoped and workflow-scoped reports.
"""
def build_run_report_csv(runs: List[Any]) -> io.StringIO:
"""Build a CSV from completed workflow runs."""
extracted_var_keys = _collect_extracted_variable_keys(runs)
output = io.StringIO()
@ -98,7 +102,7 @@ async def generate_campaign_report_csv(
runs = await db_client.get_completed_runs_for_report(
campaign_id=campaign_id, start_date=start_date, end_date=end_date
)
return _build_run_report_csv(runs), f"campaign_{campaign_id}_report.csv"
return build_run_report_csv(runs), f"campaign_{campaign_id}_report.csv"
async def generate_workflow_report_csv(
@ -110,4 +114,24 @@ async def generate_workflow_report_csv(
runs = await db_client.get_completed_runs_for_report(
workflow_id=workflow_id, start_date=start_date, end_date=end_date
)
return _build_run_report_csv(runs), f"workflow_{workflow_id}_report.csv"
return build_run_report_csv(runs), f"workflow_{workflow_id}_report.csv"
async def generate_usage_runs_report_csv(
organization_id: int,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
filters: Optional[list[dict]] = None,
) -> tuple[io.StringIO, str]:
"""Generate a CSV report for runs visible on the org-wide usage page.
Honors the same date / filter inputs as the `/usage/runs` listing.
"""
runs = await db_client.get_usage_runs_for_report(
organization_id,
start_date=start_date,
end_date=end_date,
filters=filters,
)
timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
return build_run_report_csv(runs), f"usage_runs_{timestamp}.csv"

350
api/tests/test_from_number_pool_isolation.py Normal file
View file

@ -0,0 +1,350 @@
"""
Tests verifying that the from_number pool isolates numbers per
telephony_configuration_id within an organization.
When an org has multiple telephony configurations (each with its own pool of
caller IDs), a campaign pinned to config A must never be handed a from_number
that belongs to config B. Otherwise the call is placed via provider A using a
DID owned by config B and either fails or originates from the wrong number.
These tests cover both:
- The rate_limiter, which owns the Redis-backed pool.
- The CampaignCallDispatcher, which must thread the campaign's
telephony_configuration_id through acquire / release / mapping calls.
"""
import os
import uuid
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from api.services.campaign.campaign_call_dispatcher import CampaignCallDispatcher
from api.services.campaign.rate_limiter import RateLimiter
def _unique_id() -> int:
"""A stable-but-unique positive int derived from a uuid for keying tests."""
return uuid.uuid4().int % 10_000_000
@pytest.fixture
async def isolated_rate_limiter():
"""A RateLimiter wired to the same Redis as production but using unique ids
per test, with cleanup of any keys it touched."""
rl = RateLimiter()
redis_client = await rl._get_redis()
created_keys: list[str] = []
original_eval = redis_client.eval
original_zadd = redis_client.zadd
async def tracking_eval(script, numkeys, *args, **kwargs):
if numkeys >= 1:
created_keys.append(args[0])
return await original_eval(script, numkeys, *args, **kwargs)
async def tracking_zadd(name, *args, **kwargs):
created_keys.append(name)
return await original_zadd(name, *args, **kwargs)
redis_client.eval = tracking_eval # type: ignore[assignment]
redis_client.zadd = tracking_zadd # type: ignore[assignment]
yield rl
redis_client.eval = original_eval # type: ignore[assignment]
redis_client.zadd = original_zadd # type: ignore[assignment]
if created_keys:
await redis_client.delete(*set(created_keys))
await rl.close()
# ---------------------------------------------------------------------------
# Rate limiter pool isolation
# ---------------------------------------------------------------------------
class TestRateLimiterFromNumberPoolIsolation:
"""The rate_limiter pool keys must include telephony_configuration_id."""
@pytest.mark.skipif(
"REDIS_URL" not in os.environ,
reason="Requires Redis (set REDIS_URL via .env.test)",
)
@pytest.mark.asyncio
async def test_acquire_only_returns_numbers_for_requested_config(
self, isolated_rate_limiter
):
rl = isolated_rate_limiter
org_id = _unique_id()
config_a = _unique_id()
config_b = _unique_id()
numbers_a = [f"+1555111{i:04d}" for i in range(3)]
numbers_b = [f"+1555222{i:04d}" for i in range(3)]
await rl.initialize_from_number_pool(
org_id, numbers_a, telephony_configuration_id=config_a
)
await rl.initialize_from_number_pool(
org_id, numbers_b, telephony_configuration_id=config_b
)
# Drain and cycle config_a's pool many times; the acquire should never
# hand out a config_b number.
seen: set[str] = set()
for _ in range(20):
n = await rl.acquire_from_number(
org_id, telephony_configuration_id=config_a
)
if n is None:
break
seen.add(n)
await rl.release_from_number(org_id, n, telephony_configuration_id=config_a)
assert seen == set(numbers_a), (
f"Expected only config_a numbers, but acquire returned: {seen}. "
f"Cross-config leak: {seen - set(numbers_a)}"
)
@pytest.mark.skipif(
"REDIS_URL" not in os.environ,
reason="Requires Redis (set REDIS_URL via .env.test)",
)
@pytest.mark.asyncio
async def test_release_returns_number_to_owning_config_pool(
self, isolated_rate_limiter
):
rl = isolated_rate_limiter
org_id = _unique_id()
config_a = _unique_id()
config_b = _unique_id()
numbers_a = ["+15551110001", "+15551110002"]
numbers_b = ["+15552220001", "+15552220002"]
await rl.initialize_from_number_pool(
org_id, numbers_a, telephony_configuration_id=config_a
)
await rl.initialize_from_number_pool(
org_id, numbers_b, telephony_configuration_id=config_b
)
# Acquire all of config_a's numbers (none released).
first = await rl.acquire_from_number(
org_id, telephony_configuration_id=config_a
)
second = await rl.acquire_from_number(
org_id, telephony_configuration_id=config_a
)
assert {first, second} == set(numbers_a)
# config_a is now exhausted — config_b is fully untouched.
# Acquiring for config_a must return None, NOT spill into config_b.
none_for_a = await rl.acquire_from_number(
org_id, telephony_configuration_id=config_a
)
assert none_for_a is None, (
f"Pool for config_a is exhausted but acquire returned {none_for_a} — "
"this indicates a cross-config leak."
)
# config_b's pool is fully available.
b_acquired = []
for _ in range(2):
n = await rl.acquire_from_number(
org_id, telephony_configuration_id=config_b
)
assert n is not None
b_acquired.append(n)
assert set(b_acquired) == set(numbers_b)
@pytest.mark.skipif(
"REDIS_URL" not in os.environ,
reason="Requires Redis (set REDIS_URL via .env.test)",
)
@pytest.mark.asyncio
async def test_workflow_from_number_mapping_round_trips_config(
self, isolated_rate_limiter
):
"""The mapping stored at dispatch must include the config so cleanup
can release back to the correct pool."""
rl = isolated_rate_limiter
workflow_run_id = _unique_id()
org_id = _unique_id()
config_id = _unique_id()
from_number = "+15553330001"
await rl.store_workflow_from_number_mapping(
workflow_run_id,
org_id,
from_number,
telephony_configuration_id=config_id,
)
mapping = await rl.get_workflow_from_number_mapping(workflow_run_id)
assert mapping is not None
# Tuple shape: (org_id, from_number, telephony_configuration_id)
assert len(mapping) == 3, (
f"mapping must include telephony_configuration_id, got: {mapping}"
)
assert mapping == (org_id, from_number, config_id)
await rl.delete_workflow_from_number_mapping(workflow_run_id)
# ---------------------------------------------------------------------------
# Dispatcher: must thread telephony_configuration_id end-to-end
# ---------------------------------------------------------------------------
def _make_campaign(
*,
organization_id: int,
telephony_configuration_id: int,
workflow_id: int = 1,
campaign_id: int = 99,
) -> SimpleNamespace:
return SimpleNamespace(
id=campaign_id,
organization_id=organization_id,
workflow_id=workflow_id,
created_by=1,
telephony_configuration_id=telephony_configuration_id,
rate_limit_per_second=100,
processed_rows=0,
orchestrator_metadata={},
)
def _make_queued_run(
*,
queued_run_id: int = 1,
phone_number: str = "+15559990001",
) -> SimpleNamespace:
return SimpleNamespace(
id=queued_run_id,
source_uuid=f"src-{queued_run_id}",
context_variables={"phone_number": phone_number},
)
class TestDispatcherThreadsTelephonyConfig:
"""The dispatcher must pass telephony_configuration_id when acquiring,
storing the mapping, and releasing the from_number."""
@pytest.mark.asyncio
async def test_dispatch_call_acquires_from_number_for_campaign_config(self):
org_id = 7
config_id = 4242
campaign = _make_campaign(
organization_id=org_id, telephony_configuration_id=config_id
)
queued_run = _make_queued_run()
provider = MagicMock()
provider.PROVIDER_NAME = "twilio"
provider.WEBHOOK_ENDPOINT = "twilio/voice"
provider.from_numbers = ["+15551110001"]
provider.initiate_call = AsyncMock(
return_value=SimpleNamespace(call_id="call-1", provider_metadata={})
)
workflow_run = SimpleNamespace(id=555, logs={})
dispatcher = CampaignCallDispatcher()
with (
patch.object(
dispatcher,
"get_provider_for_campaign",
AsyncMock(return_value=provider),
),
patch(
"api.services.campaign.campaign_call_dispatcher.db_client"
) as mock_db,
patch(
"api.services.campaign.campaign_call_dispatcher.rate_limiter"
) as mock_rl,
patch(
"api.services.campaign.campaign_call_dispatcher.get_backend_endpoints",
AsyncMock(return_value=("https://example.com", None)),
),
):
mock_db.get_workflow_by_id = AsyncMock(return_value=SimpleNamespace(id=1))
mock_db.create_workflow_run = AsyncMock(return_value=workflow_run)
mock_db.update_workflow_run = AsyncMock()
mock_rl.acquire_from_number = AsyncMock(return_value="+15551110001")
mock_rl.release_from_number = AsyncMock()
mock_rl.release_concurrent_slot = AsyncMock()
mock_rl.store_workflow_slot_mapping = AsyncMock()
mock_rl.store_workflow_from_number_mapping = AsyncMock()
await dispatcher.dispatch_call(queued_run, campaign, slot_id="slot-1")
# acquire_from_number on rate_limiter must be called with the
# campaign's telephony_configuration_id.
assert mock_rl.acquire_from_number.await_count == 1
call = mock_rl.acquire_from_number.await_args
kwargs = call.kwargs
args = call.args
received_config = kwargs.get("telephony_configuration_id") or (
args[1] if len(args) > 1 else None
)
assert received_config == config_id, (
"dispatch_call must pass campaign.telephony_configuration_id "
f"({config_id}) to rate_limiter.acquire_from_number, got "
f"args={args}, kwargs={kwargs}"
)
# The workflow→from_number mapping must also remember the config so
# the cleanup path can release back to the right pool.
assert mock_rl.store_workflow_from_number_mapping.await_count == 1
store_call = mock_rl.store_workflow_from_number_mapping.await_args
store_kwargs = store_call.kwargs
store_args = store_call.args
stored_config = store_kwargs.get("telephony_configuration_id") or (
store_args[3] if len(store_args) > 3 else None
)
assert stored_config == config_id, (
"store_workflow_from_number_mapping must persist the "
f"telephony_configuration_id ({config_id}); got args={store_args}, "
f"kwargs={store_kwargs}"
)
@pytest.mark.asyncio
async def test_release_call_slot_uses_stored_telephony_config(self):
"""When a call completes, release_call_slot must release the from_number
to the same telephony config it was acquired from."""
org_id = 7
config_id = 4242
from_number = "+15551110001"
workflow_run_id = 555
dispatcher = CampaignCallDispatcher()
with patch(
"api.services.campaign.campaign_call_dispatcher.rate_limiter"
) as mock_rl:
mock_rl.get_workflow_slot_mapping = AsyncMock(return_value=None)
mock_rl.get_workflow_from_number_mapping = AsyncMock(
return_value=(org_id, from_number, config_id)
)
mock_rl.release_from_number = AsyncMock(return_value=True)
mock_rl.delete_workflow_from_number_mapping = AsyncMock(return_value=True)
await dispatcher.release_call_slot(workflow_run_id)
assert mock_rl.release_from_number.await_count == 1
call = mock_rl.release_from_number.await_args
args = call.args
kwargs = call.kwargs
released_config = kwargs.get("telephony_configuration_id") or (
args[2] if len(args) > 2 else None
)
assert released_config == config_id, (
"release_call_slot must pass telephony_configuration_id "
f"({config_id}) so the number is returned to its pool; got "
f"args={args}, kwargs={kwargs}"
)

7
docker-compose.yaml
View file

@ -110,8 +110,9 @@ services:
# FastAPI workers count
FASTAPI_WORKERS: 1
# Langfuse
ENABLE_TRACING: "false"
# Langfuse — credentials can be set here or per-organization via the UI
# at /settings. Tracing is automatically active when credentials are
# available; uncomment to set defaults for all organizations.
# LANGFUSE_SECRET_KEY: ""
# LANGFUSE_PUBLIC_KEY: ""
# LANGFUSE_HOST: ""
@ -220,4 +221,4 @@ volumes:
networks:
app-network:
driver: bridge
driver: bridge

11
docs/api-reference/calls/list-runs.mdx Normal file
View file

@ -0,0 +1,11 @@
---
title: "List Runs with Usage"
description: "Paginated list of workflow runs across the organization with usage and cost"
openapi: "GET /api/v1/organizations/usage/runs"
---
Returns a paginated list of runs across all agents in your organization, including duration, cost, and usage details for each run.
Use `start_date` and `end_date` (ISO 8601) to scope the window, and `page` / `limit` to paginate. Pass `filters` as a JSON-encoded string to narrow results further.
To fetch the full transcript or recording for a specific run, use [Retrieve Call Details](/api-reference/calls/get-run).

2
docs/api-reference/openapi.json
View file

File diff suppressed because one or more lines are too long

2
docs/configurations/tracing.mdx
View file

@ -119,7 +119,7 @@ We provide seamless integration with Langfuse for tracing if you want to use you
Once enabled, traces will be available for every completed call in Dograh.
<Note>
For self-hosted deployments, you can also configure Langfuse via [environment variables](/developer/environment-variables#tracing-langfuse) (`ENABLE_TRACING`, `LANGFUSE_SECRET_KEY`, `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_HOST`) if you prefer. The UI settings take precedence over environment variables.
For self-hosted deployments, you can also configure Langfuse via [environment variables](/developer/environment-variables#tracing-langfuse) (`LANGFUSE_SECRET_KEY`, `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_HOST`) as a default for all organizations. Tracing activates automatically once credentials are available — no separate enable flag is required. Per-organization credentials set in the UI take precedence over environment variables.
</Note>
## Quick Reference

3
docs/developer/environment-variables.mdx
View file

@ -23,7 +23,6 @@ The relevant required variables for each mode are noted in the descriptions belo
| `ENVIRONMENT` | `local` | Runtime environment. Affects logging and behaviour. One of `local`, `production`, `test` |
| `DEPLOYMENT_MODE` | `oss` | Deployment mode. Use `oss` for self-hosted |
| `AUTH_PROVIDER` | `local` | Authentication provider. Use `local` for OSS |
| `ENABLE_TRACING` | `false` | Enable distributed tracing via [Langfuse](/configurations/tracing) |
---
@ -105,7 +104,7 @@ Dograh uses **MinIO by default**, which is bundled with the self-hosted deployme
| `LANGFUSE_PUBLIC_KEY` | `null` | Langfuse public key |
| `LANGFUSE_SECRET_KEY` | `null` | Langfuse secret key |
Set `ENABLE_TRACING=true` alongside these to activate LLM call tracing. See the [Tracing guide](/configurations/tracing) for setup instructions.
Tracing activates automatically as soon as credentials are available — either via these environment variables (applied to all organizations) or per-organization in the UI under **Platform Settings**. If neither is set, spans are dropped silently. See the [Tracing guide](/configurations/tracing) for setup instructions.
---

1
docs/docs.json
View file

@ -203,6 +203,7 @@
"api-reference/calls",
"api-reference/calls/trigger",
"api-reference/calls/get-run",
"api-reference/calls/list-runs",
"api-reference/calls/download",
"api-reference/calls/inbound"
]

2
pipecat

@ -1 +1 @@
Subproject commit 97b3b041bda0099dbe48b6f20daf49ce113711f3
Subproject commit 0d0e3b3bc0bc03f3d3c167dc609ea24eb22e72a0

7
sdk/python/src/dograh_sdk/_generated_models.py
View file

@ -1,6 +1,6 @@
# generated by datamodel-codegen:
# filename: dograh-openapi-XXXXXX.json.x7226hNjaY
# timestamp: 2026-05-07T08:13:29+00:00
# filename: dograh-openapi-XXXXXX.json.k2orBArfVN
# timestamp: 2026-05-08T08:26:28+00:00
from __future__ import annotations
@ -111,6 +111,9 @@ class InitiateCallRequest(BaseModel):
telephony_configuration_id: Annotated[
int | None, Field(title='Telephony Configuration Id')
] = None
from_phone_number_id: Annotated[int | None, Field(title='From Phone Number Id')] = (
None
)
class NodeCategory(Enum):

2
sdk/typescript/src/_generated_models.ts
View file

@ -432,6 +432,8 @@ export interface components {
phone_number?: string | null;
/** Telephony Configuration Id */
telephony_configuration_id?: number | null;
/** From Phone Number Id */
from_phone_number_id?: number | null;
};
/**
* NodeCategory

159
ui/src/app/usage/page.tsx
View file

@ -1,11 +1,12 @@
"use client";
import { ChevronLeft, ChevronRight, Globe } from 'lucide-react';
import { ChevronLeft, ChevronRight, Download, Globe } from 'lucide-react';
import { useRouter, useSearchParams } from 'next/navigation';
import { useCallback, useEffect, useId, useState } from 'react';
import TimezoneSelect, { type ITimezoneOption } from 'react-timezone-select';
import { toast } from 'sonner';
import { getDailyUsageBreakdownApiV1OrganizationsUsageDailyBreakdownGet, getMpsCreditsApiV1OrganizationsUsageMpsCreditsGet, getUsageHistoryApiV1OrganizationsUsageRunsGet } from '@/client/sdk.gen';
import { downloadUsageRunsReportApiV1OrganizationsUsageRunsReportGet, getDailyUsageBreakdownApiV1OrganizationsUsageDailyBreakdownGet, getMpsCreditsApiV1OrganizationsUsageMpsCreditsGet, getUsageHistoryApiV1OrganizationsUsageRunsGet } from '@/client/sdk.gen';
import type { DailyUsageBreakdownResponse, MpsCreditsResponse, UsageHistoryResponse, WorkflowRunUsageResponse } from '@/client/types.gen';
import { DailyUsageTable } from '@/components/DailyUsageTable';
import { FilterBuilder } from '@/components/filters/FilterBuilder';
@ -49,15 +50,21 @@ export default function UsagePage() {
return pageParam ? parseInt(pageParam, 10) : 1;
});
const [isExecutingFilters, setIsExecutingFilters] = useState(false);
const [isDownloadingReport, setIsDownloadingReport] = useState(false);
// Daily usage breakdown state (only for paid orgs)
const [dailyUsage, setDailyUsage] = useState<DailyUsageBreakdownResponse | null>(null);
const [isLoadingDaily, setIsLoadingDaily] = useState(false);
// Initialize filters from URL
// Initialize filters from URL. `activeFilters` tracks the in-progress
// edits in the FilterBuilder; `appliedFilters` is what's actually been
// committed via Apply (and what drives fetching + the download button).
const [activeFilters, setActiveFilters] = useState<ActiveFilter[]>(() => {
return decodeFiltersFromURL(searchParams, usageFilterAttributes);
});
const [appliedFilters, setAppliedFilters] = useState<ActiveFilter[]>(() => {
return decodeFiltersFromURL(searchParams, usageFilterAttributes);
});
// Media preview dialog
const mediaPreview = MediaPreviewDialog();
@ -83,51 +90,50 @@ export default function UsagePage() {
}
}, [auth.isAuthenticated]);
// Translate the FilterBuilder state into the query-param shape the
// backend expects. Shared between the listing fetch and the CSV export
// so they stay in lockstep.
const buildUsageQueryParams = (filters?: ActiveFilter[]) => {
let filterParam: string | undefined;
let startDate = '';
let endDate = '';
if (filters && filters.length > 0) {
const dateRangeFilter = filters.find(f => f.attribute.id === 'dateRange');
if (dateRangeFilter && dateRangeFilter.value) {
const dateValue = dateRangeFilter.value as DateRangeValue;
if (dateValue.from) startDate = dateValue.from.toISOString();
if (dateValue.to) endDate = dateValue.to.toISOString();
}
const otherFilters = filters.filter(f => f.attribute.id !== 'dateRange');
if (otherFilters.length > 0) {
const filterData = otherFilters.map(filter => ({
attribute: filter.attribute.id,
type: filter.attribute.type,
value: filter.value,
}));
filterParam = JSON.stringify(filterData);
}
}
return {
...(startDate && { start_date: startDate }),
...(endDate && { end_date: endDate }),
...(filterParam && { filters: filterParam }),
};
};
// Fetch usage history
const fetchUsageHistory = useCallback(async (page: number, filters?: ActiveFilter[]) => {
if (!auth.isAuthenticated) return;
setIsLoadingHistory(true);
try {
let filterParam = undefined;
let startDate = '';
let endDate = '';
if (filters && filters.length > 0) {
// Extract date range filter if present
const dateRangeFilter = filters.find(f => f.attribute.id === 'dateRange');
if (dateRangeFilter && dateRangeFilter.value) {
const dateValue = dateRangeFilter.value as DateRangeValue;
if (dateValue.from) {
// The dates are already in the user's local timezone
// Convert to UTC ISO string for the backend
startDate = dateValue.from.toISOString();
}
if (dateValue.to) {
// Convert to UTC ISO string for the backend
endDate = dateValue.to.toISOString();
}
}
// Process other filters (excluding dateRange)
const otherFilters = filters.filter(f => f.attribute.id !== 'dateRange');
if (otherFilters.length > 0) {
const filterData = otherFilters.map(filter => ({
attribute: filter.attribute.id,
type: filter.attribute.type,
value: filter.value,
}));
filterParam = JSON.stringify(filterData);
}
}
const response = await getUsageHistoryApiV1OrganizationsUsageRunsGet({
query: {
page,
limit: 50,
...(startDate && { start_date: startDate }),
...(endDate && { end_date: endDate }),
...(filterParam && { filters: filterParam })
...buildUsageQueryParams(filters),
},
});
@ -161,6 +167,37 @@ export default function UsagePage() {
}
}, [auth.isAuthenticated, organizationPricing]);
// Download a CSV of all runs matching the current filters.
const handleDownloadReport = async () => {
if (!auth.isAuthenticated) return;
setIsDownloadingReport(true);
try {
const response = await downloadUsageRunsReportApiV1OrganizationsUsageRunsReportGet({
query: buildUsageQueryParams(appliedFilters),
parseAs: 'blob',
});
if (response.data) {
const blob = response.data as Blob;
const url = window.URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
a.download = 'usage_runs_report.csv';
document.body.appendChild(a);
a.click();
a.remove();
window.URL.revokeObjectURL(url);
} else {
toast.error('Failed to download report');
}
} catch (error) {
console.error('Failed to download usage report:', error);
toast.error('Failed to download report');
} finally {
setIsDownloadingReport(false);
}
};
// Handle timezone change
const handleTimezoneChange = async (timezone: ITimezoneOption | string) => {
setSelectedTimezone(timezone);
@ -195,9 +232,9 @@ export default function UsagePage() {
useEffect(() => {
if (auth.isAuthenticated) {
fetchMpsCredits();
fetchUsageHistory(currentPage, activeFilters);
fetchUsageHistory(currentPage, appliedFilters);
}
}, [auth.isAuthenticated, currentPage, activeFilters, fetchUsageHistory, fetchMpsCredits]);
}, [auth.isAuthenticated, currentPage, appliedFilters, fetchUsageHistory, fetchMpsCredits]);
// Fetch daily usage when organizationPricing becomes available
useEffect(() => {
@ -229,6 +266,7 @@ export default function UsagePage() {
const handleApplyFilters = useCallback(async () => {
setIsExecutingFilters(true);
setCurrentPage(1); // Reset to first page when applying filters
setAppliedFilters(activeFilters);
updateUrlParams({ page: 1, filters: activeFilters });
await fetchUsageHistory(1, activeFilters);
setIsExecutingFilters(false);
@ -241,6 +279,8 @@ export default function UsagePage() {
const handleClearFilters = useCallback(async () => {
setIsExecutingFilters(true);
setCurrentPage(1);
setActiveFilters([]);
setAppliedFilters([]);
updateUrlParams({ page: 1, filters: [] }); // Clear filters from URL
await fetchUsageHistory(1, []); // Fetch all runs without filters
setIsExecutingFilters(false);
@ -249,8 +289,8 @@ export default function UsagePage() {
// Handle page change
const handlePageChange = (newPage: number) => {
setCurrentPage(newPage);
updateUrlParams({ page: newPage, filters: activeFilters });
fetchUsageHistory(newPage, activeFilters);
updateUrlParams({ page: newPage, filters: appliedFilters });
fetchUsageHistory(newPage, appliedFilters);
};
// Handle row click to navigate to workflow run
@ -289,8 +329,8 @@ export default function UsagePage() {
<div>
<div className="flex justify-between items-start">
<div>
<h1 className="text-3xl font-bold mb-2">Usage Dashboard</h1>
<p className="text-muted-foreground">Monitor your Dograh Token usage and quota</p>
<h1 className="text-3xl font-bold mb-2">Agent Runs</h1>
<p className="text-muted-foreground">See all your Agent Runs across all Voice Agents. You can use filters to filter out required Agent Runs.</p>
</div>
<div className="flex items-center gap-2">
<Globe className="h-4 w-4 text-muted-foreground" />
@ -419,7 +459,7 @@ export default function UsagePage() {
)}
{/* Filter Builder */}
<div className="mb-6">
<div className="mb-6 space-y-3">
<FilterBuilder
availableAttributes={usageFilterAttributes}
activeFilters={activeFilters}
@ -428,6 +468,19 @@ export default function UsagePage() {
onClearFilters={handleClearFilters}
isExecuting={isExecutingFilters}
/>
{appliedFilters.length > 0 && (
<div className="flex justify-end">
<Button
variant="outline"
size="sm"
onClick={handleDownloadReport}
disabled={isDownloadingReport}
>
<Download className="h-4 w-4 mr-2" />
{isDownloadingReport ? 'Preparing...' : 'Download Filtered Results'}
</Button>
</div>
)}
</div>
{/* Usage History */}
@ -435,9 +488,9 @@ export default function UsagePage() {
<CardHeader>
<div className="flex justify-between items-start">
<div className="space-y-1.5">
<CardTitle>Usage History</CardTitle>
<CardTitle>All Runs</CardTitle>
<CardDescription>
View detailed usage by workflow run
Every agent run across your organization, with usage details
</CardDescription>
</div>
</div>
@ -463,7 +516,7 @@ export default function UsagePage() {
<TableHead className="font-semibold">Date</TableHead>
<TableHead className="font-semibold text-right">Duration</TableHead>
<TableHead className="font-semibold text-right">
{organizationPricing?.price_per_second_usd ? 'Cost (USD)' : 'Dograh Tokens'}
{organizationPricing?.price_per_second_usd ? 'Cost (USD)' : 'Tokens'}
</TableHead>
<TableHead className="font-semibold">Actions</TableHead>
</TableRow>
@ -490,7 +543,9 @@ export default function UsagePage() {
)}
</TableCell>
<TableCell className="text-sm">
{run.phone_number || '-'}
{(run.call_type === 'inbound'
? run.caller_number
: run.called_number) || '-'}
</TableCell>
<TableCell>
{run.disposition ? (
@ -526,7 +581,7 @@ export default function UsagePage() {
</div>
{/* Summary */}
{activeFilters.length > 0 && (
{appliedFilters.length > 0 && (
<div className="mt-4 p-3 bg-muted rounded-md">
<p className="text-sm text-muted-foreground">
Total for filtered period: <span className="font-semibold text-foreground">
@ -570,7 +625,7 @@ export default function UsagePage() {
)}
</>
) : (
<p className="text-center py-8 text-muted-foreground">No usage history found</p>
<p className="text-center py-8 text-muted-foreground">No runs found</p>
)}
</CardContent>
</Card>

79
ui/src/app/workflow/[workflowId]/components/PhoneCallDialog.tsx
View file

@ -9,9 +9,10 @@ import { PhoneInput } from 'react-international-phone';
import {
initiateCallApiV1TelephonyInitiateCallPost,
listPhoneNumbersApiV1OrganizationsTelephonyConfigsConfigIdPhoneNumbersGet,
listTelephonyConfigurationsApiV1OrganizationsTelephonyConfigsGet
} from '@/client/sdk.gen';
import type { TelephonyConfigurationListItem } from '@/client/types.gen';
import type { PhoneNumberResponse, TelephonyConfigurationListItem } from '@/client/types.gen';
import { Button } from "@/components/ui/button";
import {
Dialog,
@ -58,6 +59,9 @@ export const PhoneCallDialog = ({
const [sipMode, setSipMode] = useState(() => /^(PJSIP|SIP)\//i.test(userConfig?.test_phone_number || ""));
const [telephonyConfigs, setTelephonyConfigs] = useState<TelephonyConfigurationListItem[]>([]);
const [selectedConfigId, setSelectedConfigId] = useState<string>("");
const [fromPhoneNumbers, setFromPhoneNumbers] = useState<PhoneNumberResponse[]>([]);
const [selectedFromPhoneNumberId, setSelectedFromPhoneNumberId] = useState<string>("");
const [loadingPhoneNumbers, setLoadingPhoneNumbers] = useState(false);
// Check telephony configuration when dialog opens
useEffect(() => {
@ -102,9 +106,49 @@ export const PhoneCallDialog = ({
setNeedsConfiguration(null);
setTelephonyConfigs([]);
setSelectedConfigId("");
setFromPhoneNumbers([]);
setSelectedFromPhoneNumberId("");
}
}, [open]);
// Fetch phone numbers whenever the selected telephony configuration changes.
useEffect(() => {
if (!open || !selectedConfigId) {
setFromPhoneNumbers([]);
setSelectedFromPhoneNumberId("");
return;
}
let cancelled = false;
const fetchPhoneNumbers = async () => {
setLoadingPhoneNumbers(true);
try {
const response = await listPhoneNumbersApiV1OrganizationsTelephonyConfigsConfigIdPhoneNumbersGet({
path: { config_id: Number(selectedConfigId) },
});
if (cancelled) return;
const all = response.data?.phone_numbers ?? [];
const active = all.filter((p) => p.is_active);
setFromPhoneNumbers(active);
const defaultPhone = active.find((p) => p.is_default_caller_id) ?? active[0];
setSelectedFromPhoneNumberId(defaultPhone ? String(defaultPhone.id) : "");
} catch (err) {
if (cancelled) return;
console.error("Failed to load phone numbers for config:", err);
setFromPhoneNumbers([]);
setSelectedFromPhoneNumberId("");
} finally {
if (!cancelled) setLoadingPhoneNumbers(false);
}
};
fetchPhoneNumbers();
return () => {
cancelled = true;
};
}, [open, selectedConfigId]);
// Keep phoneNumber in sync with userConfig when dialog opens
useEffect(() => {
if (open) {
@ -148,6 +192,7 @@ export const PhoneCallDialog = ({
workflow_id: workflowId,
phone_number: phoneNumber,
telephony_configuration_id: selectedConfigId ? Number(selectedConfigId) : null,
from_phone_number_id: selectedFromPhoneNumberId ? Number(selectedFromPhoneNumberId) : null,
},
});
@ -230,6 +275,38 @@ export const PhoneCallDialog = ({
</Select>
</div>
)}
{selectedConfigId && (
<div className="flex flex-col gap-1.5">
<Label htmlFor="from-phone-number">Caller ID (from)</Label>
{loadingPhoneNumbers ? (
<div className="flex items-center text-sm text-muted-foreground">
<Loader2 className="h-4 w-4 animate-spin mr-2" />
Loading phone numbers...
</div>
) : fromPhoneNumbers.length > 0 ? (
<Select
value={selectedFromPhoneNumberId}
onValueChange={setSelectedFromPhoneNumberId}
>
<SelectTrigger id="from-phone-number" className="w-full">
<SelectValue placeholder="Select a phone number" />
</SelectTrigger>
<SelectContent>
{fromPhoneNumbers.map((phone) => (
<SelectItem key={phone.id} value={String(phone.id)}>
{phone.label ? `${phone.label} — ${phone.address}` : phone.address}
{phone.is_default_caller_id ? " — default" : ""}
</SelectItem>
))}
</SelectContent>
</Select>
) : (
<div className="text-xs text-muted-foreground">
No phone numbers in this configuration. The provider will pick one automatically.
</div>
)}
</div>
)}
{sipMode ? (
<Input
value={phoneNumber}

4
ui/src/client/index.ts
View file

File diff suppressed because one or more lines are too long

32
ui/src/client/sdk.gen.ts
View file

File diff suppressed because one or more lines are too long

175
ui/src/client/types.gen.ts
View file

@ -1995,6 +1995,10 @@ export type InitiateCallRequest = {
* Telephony Configuration Id
*/
telephony_configuration_id?: number | null;
/**
* From Phone Number Id
*/
from_phone_number_id?: number | null;
};
/**
@ -3683,34 +3687,6 @@ export type TransferCallConfig = {
timeout?: number;
};
/**
* TransferCallRequest
*
* Request model for initiating a call transfer.
*/
export type TransferCallRequest = {
/**
* Destination
*/
destination: string;
/**
* Organization Id
*/
organization_id: number;
/**
* Transfer Id
*/
transfer_id: string;
/**
* Conference Name
*/
conference_name: string;
/**
* Timeout
*/
timeout?: number | null;
};
/**
* TransferCallToolDefinition
*
@ -4625,8 +4601,20 @@ export type WorkflowRunUsageResponse = {
transcript_url?: string | null;
/**
* Phone Number
*
* Deprecated. Use caller_number and called_number instead.
*
* @deprecated
*/
phone_number?: string | null;
/**
* Caller Number
*/
caller_number?: string | null;
/**
* Called Number
*/
called_number?: string | null;
/**
* Call Type
*/
@ -4882,33 +4870,6 @@ export type HandleInboundTelephonyApiV1TelephonyInboundWorkflowIdPostResponses =
200: unknown;
};
export type InitiateCallTransferApiV1TelephonyCallTransferPostData = {
body: TransferCallRequest;
path?: never;
query?: never;
url: '/api/v1/telephony/call-transfer';
};
export type InitiateCallTransferApiV1TelephonyCallTransferPostErrors = {
/**
* Not found
*/
404: unknown;
/**
* Validation Error
*/
422: HttpValidationError;
};
export type InitiateCallTransferApiV1TelephonyCallTransferPostError = InitiateCallTransferApiV1TelephonyCallTransferPostErrors[keyof InitiateCallTransferApiV1TelephonyCallTransferPostErrors];
export type InitiateCallTransferApiV1TelephonyCallTransferPostResponses = {
/**
* Successful Response
*/
200: unknown;
};
export type CompleteTransferFunctionCallApiV1TelephonyTransferResultTransferIdPostData = {
body?: never;
path: {
@ -9559,13 +9520,13 @@ export type GetUsageHistoryApiV1OrganizationsUsageRunsGetData = {
/**
* Start Date
*
* ISO format date string
* ISO 8601 date-time string (UTC). Lower bound (inclusive) on `created_at`.
*/
start_date?: string | null;
/**
* End Date
*
* ISO format date string
* ISO 8601 date-time string (UTC). Upper bound (inclusive) on `created_at`.
*/
end_date?: string | null;
/**
@ -9579,7 +9540,28 @@ export type GetUsageHistoryApiV1OrganizationsUsageRunsGetData = {
/**
* Filters
*
* JSON string of filters
* JSON-encoded array of filter objects. Each object has the shape:
*
* ```json
* { "attribute": "<name>", "type": "<type>", "value": <value> }
* ```
*
* Supported `attribute` / `type` / `value` combinations:
*
* | attribute | type | value shape | matches |
* |-----------------|---------------|----------------------------------------------|------------------------------------------------------|
* | `runId` | `number` | `{ "value": 12345 }` | exact run id |
* | `workflowId` | `number` | `{ "value": 42 }` | exact agent (workflow) id |
* | `campaignId` | `number` | `{ "value": 7 }` | exact campaign id |
* | `callerNumber` | `text` | `{ "value": "415555" }` | substring match on `initial_context.caller_number` |
* | `calledNumber` | `text` | `{ "value": "9911848" }` | substring match on `initial_context.called_number` |
* | `dispositionCode` | `multiSelect` | `{ "codes": ["XFER", "DNC"] }` | any of the codes in `gathered_context.mapped_call_disposition` |
* | `duration` | `numberRange` | `{ "min": 60, "max": 300 }` | call duration (seconds), inclusive bounds |
*
* Unknown attributes and unsupported `type` values are silently ignored.
*
* Date filtering on this endpoint is done via the dedicated `start_date` / `end_date` query params, not via a `dateRange` filter object.
*
*/
filters?: string | null;
};
@ -9608,6 +9590,83 @@ export type GetUsageHistoryApiV1OrganizationsUsageRunsGetResponses = {
export type GetUsageHistoryApiV1OrganizationsUsageRunsGetResponse = GetUsageHistoryApiV1OrganizationsUsageRunsGetResponses[keyof GetUsageHistoryApiV1OrganizationsUsageRunsGetResponses];
export type DownloadUsageRunsReportApiV1OrganizationsUsageRunsReportGetData = {
body?: never;
headers?: {
/**
* Authorization
*/
authorization?: string | null;
/**
* X-Api-Key
*/
'X-API-Key'?: string | null;
};
path?: never;
query?: {
/**
* Start Date
*
* ISO 8601 date-time string (UTC). Lower bound (inclusive) on `created_at`.
*/
start_date?: string | null;
/**
* End Date
*
* ISO 8601 date-time string (UTC). Upper bound (inclusive) on `created_at`.
*/
end_date?: string | null;
/**
* Filters
*
* JSON-encoded array of filter objects. Each object has the shape:
*
* ```json
* { "attribute": "<name>", "type": "<type>", "value": <value> }
* ```
*
* Supported `attribute` / `type` / `value` combinations:
*
* | attribute | type | value shape | matches |
* |-----------------|---------------|----------------------------------------------|------------------------------------------------------|
* | `runId` | `number` | `{ "value": 12345 }` | exact run id |
* | `workflowId` | `number` | `{ "value": 42 }` | exact agent (workflow) id |
* | `campaignId` | `number` | `{ "value": 7 }` | exact campaign id |
* | `callerNumber` | `text` | `{ "value": "415555" }` | substring match on `initial_context.caller_number` |
* | `calledNumber` | `text` | `{ "value": "9911848" }` | substring match on `initial_context.called_number` |
* | `dispositionCode` | `multiSelect` | `{ "codes": ["XFER", "DNC"] }` | any of the codes in `gathered_context.mapped_call_disposition` |
* | `duration` | `numberRange` | `{ "min": 60, "max": 300 }` | call duration (seconds), inclusive bounds |
*
* Unknown attributes and unsupported `type` values are silently ignored.
*
* Date filtering on this endpoint is done via the dedicated `start_date` / `end_date` query params, not via a `dateRange` filter object.
*
*/
filters?: string | null;
};
url: '/api/v1/organizations/usage/runs/report';
};
export type DownloadUsageRunsReportApiV1OrganizationsUsageRunsReportGetErrors = {
/**
* Not found
*/
404: unknown;
/**
* Validation Error
*/
422: HttpValidationError;
};
export type DownloadUsageRunsReportApiV1OrganizationsUsageRunsReportGetError = DownloadUsageRunsReportApiV1OrganizationsUsageRunsReportGetErrors[keyof DownloadUsageRunsReportApiV1OrganizationsUsageRunsReportGetErrors];
export type DownloadUsageRunsReportApiV1OrganizationsUsageRunsReportGetResponses = {
/**
* Successful Response
*/
200: unknown;
};
export type GetDailyUsageBreakdownApiV1OrganizationsUsageDailyBreakdownGetData = {
body?: never;
headers?: {

2
ui/src/components/layout/AppSidebar.tsx
View file

@ -166,7 +166,7 @@ export function AppSidebar() {
const observeSection = [
{
title: "Usage",
title: "Agent Runs",
url: "/usage",
icon: TrendingUp,
},

39
ui/src/lib/filterAttributes.ts
View file

@ -85,14 +85,32 @@ export const baseFilterAttributes: Record<string, Omit<FilterAttribute, "id">> =
step: 1,
},
},
phoneNumber: {
callerNumber: {
type: "text",
label: "Phone Number",
label: "Caller Number",
config: {
placeholder: "Enter phone number (partial match)",
placeholder: "Enter caller number (partial match)",
maxLength: 20,
},
},
calledNumber: {
type: "text",
label: "Called Number",
config: {
placeholder: "Enter called number (partial match)",
maxLength: 20,
},
},
campaignId: {
type: "number",
label: "Campaign ID",
config: {
placeholder: "Enter campaign ID",
min: 1,
max: 9999999,
step: 1,
},
},
};
// Helper function to create filter attributes with proper IDs
@ -135,7 +153,8 @@ export const superadminFilterAttributes = createFilterAttributes(
"dateRange",
"runId",
"workflowId",
"phoneNumber",
"callerNumber",
"calledNumber",
"dispositionCode",
"status",
"duration",
@ -153,9 +172,19 @@ export const usageFilterAttributes = createFilterAttributes(
"dateRange",
"duration",
"dispositionCode",
"phoneNumber",
"callerNumber",
"calledNumber",
"runId",
"workflowId",
"campaignId",
],
{
runId: {
label: "Run ID",
},
workflowId: {
label: "Agent ID",
},
dateRange: {
label: "Date Range",
config: {