websocket-patterns

WebSocket Patterns

Endpoints

WS /api/v1/ws/workflow/{session_id}?token=JWT   # session-scoped updates
WS /api/v1/ws/user?token=JWT                    # all updates for a user

Token goes in the query string. This is the only place in the codebase where a JWT is allowed in a URL parameter — the WebSocket protocol has no concept of custom request headers, so there is no alternative. extract_token_from_request in utils/auth.py enforces this: query-param tokens are accepted only for paths starting with /api/ws/. All regular HTTP endpoints require Authorization: Bearer <token>.

Connection Limits

• Max 5 connections per user
• Max 3 connections per session

Complete Broadcast Function Reference

All broadcast functions live in api/websocket.py and follow the same signature:

• Always send to both session-scoped subscribers and general user subscribers.
• Always safe to await from any background task — they swallow send errors gracefully.

Main Workflow Broadcasts (called by workflows/job_application_workflow.py)

from api.websocket import (
    broadcast_agent_update,
    broadcast_phase_change,
    broadcast_workflow_complete,
    broadcast_workflow_error,
    broadcast_gate_decision,
)

await broadcast_agent_update(
    user_id=user_id,
    session_id=session_id,
    agent_name="job_analyzer",
    status="running",           # running | completed | failed
    message="Analyzing job posting...",
)

await broadcast_phase_change(
    user_id=user_id,
    session_id=session_id,
    phase="profile_matching",
    progress_percentage=60,
)

await broadcast_workflow_complete(user_id=user_id, session_id=session_id, result_summary=result)
await broadcast_workflow_error(user_id=user_id, session_id=session_id, error_message=error_msg, failed_agent="job_analyzer")
await broadcast_gate_decision(user_id=user_id, session_id=session_id, match_score=0.82, recommendation="Strong match")

Workflow Control Broadcasts (called by api/workflow.py background tasks)

from api.websocket import (
    broadcast_workflow_resumed,
    broadcast_document_generation_started,
)

# Sent at the start of _continue_workflow_background (after gate confirmation)
await broadcast_workflow_resumed(user_id=user_id, session_id=session_id)

# Sent at the start of _generate_documents_background
await broadcast_document_generation_started(user_id=user_id, session_id=session_id)