From c99bd29ef18467ffa80d6da13f5f23ee7d0bbddf Mon Sep 17 00:00:00 2001 From: Sabiha Khan <87858386+chewwbaka@users.noreply.github.com> Date: Thu, 11 Dec 2025 15:42:28 +0530 Subject: [PATCH] fix: call_id and stream_id for vobiz pipeline, add workflow run state (#78) * fix: add workflow run state for pipeline * fix: call and stream id for vobiz pipeline --- ...6841e6_add_state_field_to_workflow_runs.py | 62 +++++++++++++++++++ api/db/models.py | 8 ++- api/db/workflow_run_client.py | 3 + api/enums.py | 6 ++ api/routes/telephony.py | 26 +++++++- api/services/campaign/call_dispatcher.py | 3 +- api/services/pipecat/event_handlers.py | 2 + api/services/pipecat/transport_setup.py | 11 ++-- .../telephony/providers/vobiz_provider.py | 30 +++++++-- pipecat | 2 +- 10 files changed, 139 insertions(+), 14 deletions(-) create mode 100644 api/alembic/versions/49a8fe6841e6_add_state_field_to_workflow_runs.py diff --git a/api/alembic/versions/49a8fe6841e6_add_state_field_to_workflow_runs.py b/api/alembic/versions/49a8fe6841e6_add_state_field_to_workflow_runs.py new file mode 100644 index 0000000..e7c7356 --- /dev/null +++ b/api/alembic/versions/49a8fe6841e6_add_state_field_to_workflow_runs.py @@ -0,0 +1,62 @@ +"""add_state_field_to_workflow_runs + +Revision ID: 49a8fe6841e6 +Revises: a188ff90e76f +Create Date: 2025-12-10 17:34:31.232048 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = '49a8fe6841e6' +down_revision: Union[str, None] = 'a188ff90e76f' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Create the workflow_run_state enum type + workflow_run_state_enum = sa.Enum( + 'initialized', 'running', 'completed', + name='workflow_run_state' + ) + workflow_run_state_enum.create(op.get_bind()) + + # Add the state column to workflow_runs table (nullable first) + op.add_column( + 'workflow_runs', + sa.Column( + 'state', + sa.Enum('initialized', 'running', 'completed', name='workflow_run_state'), + nullable=True + ) + ) + + # Set appropriate state values for existing records + # Completed workflows should be marked as 'completed' + # Non-completed workflows should be marked as 'initialized' + op.execute(""" + UPDATE workflow_runs + SET state = CASE + WHEN is_completed = true THEN 'completed'::workflow_run_state + ELSE 'initialized'::workflow_run_state + END + """) + + # Now make the column non-nullable with 'initialized' as default for new records + op.alter_column( + 'workflow_runs', + 'state', + nullable=False, + server_default='initialized' + ) + + +def downgrade() -> None: + # Drop the state column + op.drop_column('workflow_runs', 'state') + + # Drop the enum type + sa.Enum(name='workflow_run_state').drop(op.get_bind()) diff --git a/api/db/models.py b/api/db/models.py index 47fdd98..b196915 100644 --- a/api/db/models.py +++ b/api/db/models.py @@ -19,7 +19,7 @@ from sqlalchemy import ( ) from sqlalchemy.orm import declarative_base, relationship -from ..enums import IntegrationAction, WorkflowRunMode, WorkflowStatus +from ..enums import IntegrationAction, WorkflowRunMode, WorkflowRunState, WorkflowStatus Base = declarative_base() @@ -314,6 +314,12 @@ class WorkflowRunModel(Base): Enum(*[mode.value for mode in WorkflowRunMode], name="workflow_run_mode"), nullable=False, ) + state = Column( + Enum(*[state.value for state in WorkflowRunState], name="workflow_run_state"), + nullable=False, + default=WorkflowRunState.INITIALIZED.value, + server_default=text("'initialized'::workflow_run_state"), + ) is_completed = Column(Boolean, default=False) recording_url = Column(String, nullable=True) transcript_url = Column(String, nullable=True) diff --git a/api/db/workflow_run_client.py b/api/db/workflow_run_client.py index fcd4f90..7179ae4 100644 --- a/api/db/workflow_run_client.py +++ b/api/db/workflow_run_client.py @@ -306,6 +306,7 @@ class WorkflowRunClient(BaseDBClient): initial_context: dict | None = None, gathered_context: dict | None = None, logs: dict | None = None, + state: str | None = None, ) -> WorkflowRunModel: async with self.async_session() as session: result = await session.execute( @@ -337,6 +338,8 @@ class WorkflowRunClient(BaseDBClient): run.logs = {**run.logs, **logs} if is_completed: run.is_completed = is_completed + if state: + run.state = state try: await session.commit() except Exception as e: diff --git a/api/enums.py b/api/enums.py index a9e9465..4510c76 100644 --- a/api/enums.py +++ b/api/enums.py @@ -54,6 +54,12 @@ class StorageBackend(Enum): return cls.MINIO +class WorkflowRunState(Enum): + INITIALIZED = "initialized" # Workflow run created, ready for connection + RUNNING = "running" # Websocket connected and pipeline active + COMPLETED = "completed" # Workflow run finished + + class WorkflowRunStatus(Enum): # historical modes VOICE = "VOICE" diff --git a/api/routes/telephony.py b/api/routes/telephony.py index e60a9c9..e400256 100644 --- a/api/routes/telephony.py +++ b/api/routes/telephony.py @@ -14,6 +14,7 @@ from starlette.responses import HTMLResponse from api.db import db_client from api.db.models import UserModel +from api.enums import WorkflowRunState from api.services.auth.depends import get_user from api.services.campaign.call_dispatcher import campaign_call_dispatcher from api.services.campaign.campaign_event_publisher import get_campaign_event_publisher @@ -228,6 +229,14 @@ async def websocket_endpoint( await websocket.close(code=4404, reason="Workflow not found") return + # Check workflow run state - only allow 'initialized' state + if workflow_run.state != WorkflowRunState.INITIALIZED.value: + logger.warning( + f"Workflow run {workflow_run_id} not in initialized state: {workflow_run.state}" + ) + await websocket.close(code=4409, reason="Workflow run not available for connection") + return + # Extract provider type from workflow run context provider_type = None if workflow_run.gathered_context: @@ -256,6 +265,16 @@ async def websocket_endpoint( await websocket.close(code=4400, reason="Provider mismatch") return + # Set workflow run state to 'running' before starting the pipeline + await db_client.update_workflow_run( + run_id=workflow_run_id, + state=WorkflowRunState.RUNNING.value + ) + + logger.info( + f"[run {workflow_run_id}] Set workflow run state to 'running' for {provider_type} provider" + ) + # Delegate to provider-specific handler await provider.handle_websocket( websocket, workflow_id, user_id, workflow_run_id @@ -362,7 +381,11 @@ async def _process_status_update( await campaign_call_dispatcher.release_call_slot(workflow_run_id) # Mark workflow run as completed - await db_client.update_workflow_run(run_id=workflow_run_id, is_completed=True) + await db_client.update_workflow_run( + run_id=workflow_run_id, + is_completed=True, + state=WorkflowRunState.COMPLETED.value + ) elif status.status in ["failed", "busy", "no-answer", "canceled"]: logger.warning( @@ -396,6 +419,7 @@ async def _process_status_update( await db_client.update_workflow_run( run_id=workflow_run_id, is_completed=True, + state=WorkflowRunState.COMPLETED.value, gathered_context={"call_tags": call_tags}, ) diff --git a/api/services/campaign/call_dispatcher.py b/api/services/campaign/call_dispatcher.py index 7675de9..d32bb32 100644 --- a/api/services/campaign/call_dispatcher.py +++ b/api/services/campaign/call_dispatcher.py @@ -7,7 +7,7 @@ from loguru import logger from api.db import db_client from api.db.models import QueuedRunModel, WorkflowRunModel -from api.enums import OrganizationConfigurationKey +from api.enums import OrganizationConfigurationKey, WorkflowRunState from api.services.campaign.rate_limiter import rate_limiter from api.services.telephony.base import TelephonyProvider from api.services.telephony.factory import get_telephony_provider @@ -277,6 +277,7 @@ class CampaignCallDispatcher: await db_client.update_workflow_run( run_id=workflow_run.id, is_completed=True, + state=WorkflowRunState.COMPLETED.value, gathered_context={ "error": str(e), }, diff --git a/api/services/pipecat/event_handlers.py b/api/services/pipecat/event_handlers.py index 5d201d3..0c8fdb8 100644 --- a/api/services/pipecat/event_handlers.py +++ b/api/services/pipecat/event_handlers.py @@ -1,6 +1,7 @@ from loguru import logger from api.db import db_client +from api.enums import WorkflowRunState from api.services.campaign.call_dispatcher import campaign_call_dispatcher from api.services.pipecat.audio_config import AudioConfig from api.services.pipecat.audio_transcript_buffers import ( @@ -176,6 +177,7 @@ def register_task_event_handler( usage_info=usage_info, gathered_context=gathered_context, is_completed=True, + state=WorkflowRunState.COMPLETED.value, ) # Release concurrent slot for campaign calls diff --git a/api/services/pipecat/transport_setup.py b/api/services/pipecat/transport_setup.py index 4137cce..b1ffd25 100644 --- a/api/services/pipecat/transport_setup.py +++ b/api/services/pipecat/transport_setup.py @@ -21,6 +21,7 @@ from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnal from pipecat.audio.vad.silero import SileroVADAnalyzer, VADParams from pipecat.serializers.plivo import PlivoFrameSerializer from pipecat.serializers.twilio import TwilioFrameSerializer +from pipecat.serializers.vobiz import VobizFrameSerializer from pipecat.serializers.vonage import VonageFrameSerializer from pipecat.transports.base_transport import TransportParams from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection @@ -256,20 +257,20 @@ async def create_vobiz_transport( turn_analyzer = create_turn_analyzer(workflow_run_id, audio_config) - # Use PlivoFrameSerializer for Vobiz (Plivo-compatible protocol) - serializer = PlivoFrameSerializer( + # Use VobizFrameSerializer for Vobiz WebSocket protocol + serializer = VobizFrameSerializer( stream_id=stream_id, call_id=call_id, auth_id=auth_id, auth_token=auth_token, - params=PlivoFrameSerializer.InputParams( - plivo_sample_rate=8000, # Vobiz uses MULAW at 8kHz + params=VobizFrameSerializer.InputParams( + vobiz_sample_rate=8000, # Vobiz uses MULAW at 8kHz sample_rate=audio_config.pipeline_sample_rate, ), ) logger.debug( - f"[run {workflow_run_id}] PlivoFrameSerializer created for Vobiz - " + f"[run {workflow_run_id}] VobizFrameSerializer created for Vobiz - " f"transport_rate=8000Hz, pipeline_rate={audio_config.pipeline_sample_rate}Hz" ) diff --git a/api/services/telephony/providers/vobiz_provider.py b/api/services/telephony/providers/vobiz_provider.py index a38093f..4619048 100644 --- a/api/services/telephony/providers/vobiz_provider.py +++ b/api/services/telephony/providers/vobiz_provider.py @@ -2,6 +2,7 @@ Vobiz implementation of the TelephonyProvider interface. """ +import json import random from typing import TYPE_CHECKING, Any, Dict, List, Optional @@ -292,16 +293,35 @@ class VobizProvider(TelephonyProvider): workflow_run_id: int, ) -> None: """ - Handle Vobiz WebSocket connection using Plivo-compatible protocol. + Handle Vobiz WebSocket connection using Vobiz WebSocket protocol. - Uses workflow_run_id as stream/call identifiers and delegates - message handling to PlivoFrameSerializer. + Extracts stream_id and call_id from the start event and delegates + message handling to VobizFrameSerializer. """ from api.services.pipecat.run_pipeline import run_pipeline_vobiz + + first_msg = await websocket.receive_text() + start_msg = json.loads(first_msg) + logger.debug(f"Received the first message: {start_msg}") + + # Validate that this is a start event + if start_msg.get("event") != "start": + logger.error(f"Expected 'start' event, got: {start_msg.get('event')}") + await websocket.close(code=4400, reason="Expected start event") + return + + logger.debug(f"Vobiz WebSocket connected for workflow_run {workflow_run_id}") try: - stream_id = f"vobiz-stream-{workflow_run_id}" - call_id = f"vobiz-call-{workflow_run_id}" + # Extract stream_id and call_id from the start event + start_data = start_msg.get("start", {}) + stream_id = start_data.get("streamId") + call_id = start_data.get("callId") + + if not stream_id or not call_id: + logger.error(f"Missing streamId or callId in start event: {start_data}") + await websocket.close(code=4400, reason="Missing streamId or callId") + return logger.info( f"[run {workflow_run_id}] Starting Vobiz WebSocket handler - " diff --git a/pipecat b/pipecat index 3987090..e264bc3 160000 --- a/pipecat +++ b/pipecat @@ -1 +1 @@ -Subproject commit 3987090bfb3e1e4a0341f875c93ddee69a740d60 +Subproject commit e264bc3678b9466500f2284f1ca0f5f84dc7eaa8