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..dd4934d --- /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' + 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()) 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/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