fix: add workflow run state for 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'
+            ELSE 'initialized'
+        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/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