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

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

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)

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:

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"

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},
         )
 

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),
                 },

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

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

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 - "

pipecat

@@ -1 +1 @@
-Subproject commit 3987090bfb3e1e4a0341f875c93ddee69a740d60
+Subproject commit e264bc3678b9466500f2284f1ca0f5f84dc7eaa8