"""Workflow API routes - CRUD, execution, approval, and real-time progress"""

from __future__ import annotations

import asyncio
import hmac
import json
import logging
import re
import uuid
from datetime import datetime, timezone
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect, Security
from fastapi.security import APIKeyHeader, APIKeyQuery

from agentkit.orchestrator.workflow_schema import (
    ApproveRequest,
    CreateWorkflowRequest,
    ExecuteWorkflowRequest,
    WorkflowDefinition,
    WorkflowExecution,
    WorkflowStage,
    WorkflowSummary,
)

logger = logging.getLogger(__name__)

router = APIRouter(tags=["workflows"])

# ---------------------------------------------------------------------------
# API Key Authentication
# ---------------------------------------------------------------------------

_api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
_api_key_query = APIKeyQuery(name="api_key", auto_error=False)


async def _verify_api_key(
    request: Request,
    api_key_header: str | None = Security(_api_key_header),
    api_key_query: str | None = Security(_api_key_query),
) -> None:
    """Verify API key for REST endpoints. Raises HTTPException if invalid."""
    configured_api_key: str | None = None
    if hasattr(request.app.state, "server_config") and request.app.state.server_config:
        configured_api_key = request.app.state.server_config.api_key
    if configured_api_key is None and hasattr(request.app.state, "api_key"):
        configured_api_key = request.app.state.api_key

    # If no API key is configured, allow all requests (backwards compat)
    if configured_api_key is None:
        return

    provided = api_key_header or api_key_query
    if not hmac.compare_digest((provided or "").encode(), configured_api_key.encode()):
        raise HTTPException(
            status_code=401,
            detail="Invalid or missing API key. Provide via X-API-Key header or api_key query parameter.",
        )


# ---------------------------------------------------------------------------
# In-memory Workflow Store
# ---------------------------------------------------------------------------


class WorkflowStore:
    """In-memory workflow store with async-safe mutation methods."""

    def __init__(self, max_workflows: int = 500, max_executions: int = 1000):
        self._workflows: dict[str, WorkflowDefinition] = {}
        self._executions: dict[str, WorkflowExecution] = {}
        self._max_workflows = max_workflows
        self._max_executions = max_executions
        self._approval_events: dict[str, asyncio.Event] = {}  # key: f"{execution_id}:{stage_name}"
        self._running_tasks: dict[str, asyncio.Task] = {}
        self._execution_locks: dict[str, asyncio.Lock] = {}
        self._lock = asyncio.Lock()

    def _evict_execution(self, execution_id: str) -> None:
        """Remove execution and its associated approval events."""
        self._executions.pop(execution_id, None)
        keys_to_remove = [k for k in self._approval_events if k.startswith(f"{execution_id}:")]
        for k in keys_to_remove:
            event = self._approval_events.pop(k, None)
            if event is not None:
                event.set()  # Wake any waiting coroutine

    async def save(self, workflow: WorkflowDefinition) -> WorkflowDefinition:
        async with self._lock:
            workflow.updated_at = datetime.now(timezone.utc).isoformat()
            self._workflows[workflow.workflow_id] = workflow
            # Evict oldest if over limit
            if len(self._workflows) > self._max_workflows:
                oldest_id = min(
                    self._workflows, key=lambda k: self._workflows[k].updated_at
                )
                del self._workflows[oldest_id]
            return workflow

    def get(self, workflow_id: str) -> WorkflowDefinition | None:
        return self._workflows.get(workflow_id)

    def list(self, limit: int = 50) -> list[WorkflowSummary]:
        sorted_wf = sorted(
            self._workflows.values(),
            key=lambda w: w.updated_at,
            reverse=True,
        )
        summaries = []
        for w in sorted_wf[:limit]:
            summaries.append(
                WorkflowSummary(
                    workflow_id=w.workflow_id,
                    name=w.name,
                    version=w.version,
                    stage_count=len(w.stages),
                    trigger_count=len(w.triggers),
                    created_at=w.created_at,
                    updated_at=w.updated_at,
                )
            )
        return summaries

    async def delete(self, workflow_id: str) -> bool:
        async with self._lock:
            if workflow_id in self._workflows:
                del self._workflows[workflow_id]
                return True
            return False

    async def create_execution(self, workflow_id: str) -> WorkflowExecution:
        async with self._lock:
            execution = WorkflowExecution(
                execution_id=str(uuid.uuid4()),
                workflow_id=workflow_id,
                status="pending",
                started_at=datetime.now(timezone.utc).isoformat(),
            )
            self._executions[execution.execution_id] = execution
            # Evict oldest if over limit
            if len(self._executions) > self._max_executions:
                oldest_id = min(
                    self._executions,
                    key=lambda k: self._executions[k].started_at or "",
                )
                self._evict_execution(oldest_id)
            return execution

    def get_execution(self, execution_id: str) -> WorkflowExecution | None:
        return self._executions.get(execution_id)

    async def update_execution(self, execution_id: str, **kwargs: Any) -> WorkflowExecution:
        async with self._lock:
            execution = self._executions.get(execution_id)
            if execution is None:
                raise KeyError(f"Execution '{execution_id}' not found")
            for key, value in kwargs.items():
                if hasattr(execution, key):
                    setattr(execution, key, value)
            return execution

    def get_execution_lock(self, execution_id: str) -> asyncio.Lock:
        """Get or create a per-execution lock for approve/cancel serialization."""
        if execution_id not in self._execution_locks:
            self._execution_locks[execution_id] = asyncio.Lock()
        return self._execution_locks[execution_id]


# Module-level singleton
_workflow_store = WorkflowStore()

# WebSocket subscribers for real-time execution progress (keyed by execution_id)
_ws_subscribers: dict[str, set[WebSocket]] = {}
_ws_subscribers_lock = asyncio.Lock()


async def _ws_subscribe(execution_id: str, ws: WebSocket) -> None:
    async with _ws_subscribers_lock:
        if execution_id not in _ws_subscribers:
            _ws_subscribers[execution_id] = set()
        _ws_subscribers[execution_id].add(ws)


async def _ws_unsubscribe(execution_id: str, ws: WebSocket) -> None:
    async with _ws_subscribers_lock:
        if execution_id in _ws_subscribers:
            _ws_subscribers[execution_id].discard(ws)
            if not _ws_subscribers[execution_id]:
                del _ws_subscribers[execution_id]


def _get_store(request: Request) -> WorkflowStore:
    """Get the workflow store from app state or use the module-level singleton."""
    store = getattr(request.app.state, "workflow_store", None)
    return store if store is not None else _workflow_store


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------


def _validate_workflow_stages(stages: list[WorkflowStage]) -> None:
    """Validate workflow stages for missing dependencies and circular deps."""
    stage_names = {s.name for s in stages}
    for stage in stages:
        for dep in stage.depends_on:
            if dep not in stage_names:
                raise HTTPException(
                    status_code=400,
                    detail=f"阶段 '{stage.name}' 依赖了不存在的阶段 '{dep}'",
                )

    # Check for circular dependencies
    in_degree: dict[str, int] = {s.name: 0 for s in stages}
    dependents: dict[str, list[str]] = {s.name: [] for s in stages}
    for s in stages:
        for dep in s.depends_on:
            in_degree[s.name] += 1
            dependents[dep].append(s.name)

    remaining = set(in_degree.keys())
    while remaining:
        current_level = [name for name in remaining if in_degree[name] == 0]
        if not current_level:
            raise HTTPException(
                status_code=400,
                detail="工作流存在循环依赖",
            )
        for name in current_level:
            remaining.remove(name)
            for dep in dependents[name]:
                in_degree[dep] -= 1


# ---------------------------------------------------------------------------
# Workflow execution engine (simplified)
# ---------------------------------------------------------------------------


async def _execute_workflow(
    workflow: WorkflowDefinition,
    execution: WorkflowExecution,
    variables: dict[str, Any],
    store: WorkflowStore | None = None,
) -> None:
    """Execute a workflow by running its stages in topological order."""
    _store = store or _workflow_store
    execution.status = "running"
    await _store.update_execution(execution.execution_id, status="running")

    # Topological sort
    stage_map = {s.name: s for s in workflow.stages}
    in_degree: dict[str, int] = {s.name: 0 for s in workflow.stages}
    dependents: dict[str, list[str]] = {s.name: [] for s in workflow.stages}
    for s in workflow.stages:
        for dep in s.depends_on:
            in_degree[s.name] += 1
            dependents[dep].append(s.name)

    ordered: list[str] = []
    remaining = set(in_degree.keys())
    while remaining:
        current_level = [name for name in remaining if in_degree[name] == 0]
        if not current_level:
            execution.status = "failed"
            execution.error = "循环依赖"
            execution.completed_at = datetime.now(timezone.utc).isoformat()
            await _store.update_execution(
                execution.execution_id,
                status="failed",
                error="循环依赖",
                completed_at=execution.completed_at,
            )
            return
        for name in sorted(current_level):
            ordered.append(name)
            remaining.remove(name)
            for dep in dependents[name]:
                in_degree[dep] -= 1

    # Execute stages in order
    for stage_name in ordered:
        stage = stage_map[stage_name]
        execution.current_stage = stage_name
        await _store.update_execution(
            execution.execution_id,
            current_stage=stage_name,
        )

        # Notify WebSocket subscribers
        await _broadcast_ws({
            "event": "stage_started",
            "execution_id": execution.execution_id,
            "stage": stage_name,
        }, execution_id=execution.execution_id)

        try:
            if stage.type == "approval":
                # Pause execution and wait for approval via asyncio.Event
                event_key = f"{execution.execution_id}:{stage_name}"
                approval_event = asyncio.Event()
                _store._approval_events[event_key] = approval_event

                execution.status = "paused"
                execution.current_stage = stage_name
                await _store.update_execution(
                    execution.execution_id,
                    status="paused",
                    current_stage=stage_name,
                )
                await _broadcast_ws({
                    "event": "approval_required",
                    "execution_id": execution.execution_id,
                    "stage": stage_name,
                }, execution_id=execution.execution_id)

                # Wait for approval with timeout
                try:
                    approval_timeout = stage.config.get("approval_timeout", 3600)
                    await asyncio.wait_for(approval_event.wait(), timeout=approval_timeout)
                    # Check if execution was cancelled/rejected while waiting
                    if execution.status == "cancelled":
                        await _broadcast_ws({
                            "event": "stage_failed",
                            "execution_id": execution.execution_id,
                            "stage": stage_name,
                            "error": "Approval rejected",
                        }, execution_id=execution.execution_id)
                        return
                    # Approval was granted — the /approve endpoint already set stage_results
                    # Only update status to running if not already set
                    if execution.status != "running":
                        execution.status = "running"
                        await _store.update_execution(
                            execution.execution_id,
                            status="running",
                        )
                except asyncio.TimeoutError:
                    execution.stage_results[stage_name] = {
                        "status": "timeout",
                        "approver": "none",
                        "comment": "审批超时",
                    }
                    execution.status = "failed"
                    execution.error = f"Approval timeout for stage {stage_name}"
                    execution.completed_at = datetime.now(timezone.utc).isoformat()
                    await _store.update_execution(
                        execution.execution_id,
                        status="failed",
                        error=execution.error,
                        completed_at=execution.completed_at,
                        stage_results=execution.stage_results,
                    )
                    await _broadcast_ws({
                        "event": "stage_failed",
                        "execution_id": execution.execution_id,
                        "stage": stage_name,
                        "error": "Approval timeout",
                    }, execution_id=execution.execution_id)
                    return
                finally:
                    _store._approval_events.pop(event_key, None)
            elif stage.type == "condition":
                # Evaluate condition expression
                condition_expr = stage.config.get("expression", "")
                result = _evaluate_condition(condition_expr, variables)
                execution.stage_results[stage_name] = {
                    "status": "completed",
                    "condition_result": result,
                }
                await _store.update_execution(
                    execution.execution_id,
                    stage_results=execution.stage_results,
                )
            else:
                # Skill or parallel stage - simulate execution
                execution.stage_results[stage_name] = {
                    "status": "completed",
                    "output": {"dry_run": True, "action": stage.action},
                }
                await _store.update_execution(
                    execution.execution_id,
                    stage_results=execution.stage_results,
                )

            await _broadcast_ws({
                "event": "stage_completed",
                "execution_id": execution.execution_id,
                "stage": stage_name,
            }, execution_id=execution.execution_id)

        except Exception as e:
            execution.stage_results[stage_name] = {
                "status": "failed",
                "error": str(e),
            }
            execution.status = "failed"
            execution.error = f"阶段 '{stage_name}' 执行失败: {e}"
            execution.completed_at = datetime.now(timezone.utc).isoformat()
            await _store.update_execution(
                execution.execution_id,
                status="failed",
                error=execution.error,
                completed_at=execution.completed_at,
                stage_results=execution.stage_results,
            )
            await _broadcast_ws({
                "event": "stage_failed",
                "execution_id": execution.execution_id,
                "stage": stage_name,
                "error": str(e),
            }, execution_id=execution.execution_id)
            return

    execution.status = "completed"
    execution.completed_at = datetime.now(timezone.utc).isoformat()
    execution.current_stage = None
    await _store.update_execution(
        execution.execution_id,
        status="completed",
        completed_at=execution.completed_at,
        current_stage=None,
    )
    await _broadcast_ws({
        "event": "execution_completed",
        "execution_id": execution.execution_id,
    }, execution_id=execution.execution_id)


_SAFE_VAR_PATTERN = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
_SAFE_OPERATORS = {"==", "!=", ">", "<", ">=", "<="}


def _evaluate_condition(expression: str, variables: dict[str, Any]) -> bool:
    """Evaluate a condition expression safely."""
    expression = expression.strip()
    if not expression:
        return True

    # Try each operator (longer operators first to avoid partial matches)
    for op in sorted(_SAFE_OPERATORS, key=len, reverse=True):
        if op in expression:
            parts = expression.split(op, 1)
            if len(parts) != 2:
                continue
            left = parts[0].strip()
            right = parts[1].strip()

            # Validate variable names
            if left and not _SAFE_VAR_PATTERN.match(left):
                raise ValueError(f"Invalid variable name in condition: {left}")

            if left and left not in variables:
                # Missing variable: treat as None/empty — condition evaluates to False
                left_val = None
            else:
                left_val = variables.get(left, left)
            # Strip quotes from right side if present
            if right.startswith('"') and right.endswith('"'):
                right_val = right[1:-1]
            elif right.startswith("'") and right.endswith("'"):
                right_val = right[1:-1]
            elif right and _SAFE_VAR_PATTERN.match(right):
                right_val = variables.get(right, right)
            else:
                right_val = right

            # Compare based on operator
            if op == "==":
                return str(left_val) == str(right_val)
            if op == "!=":
                return str(left_val) != str(right_val)
            try:
                left_num = float(left_val)
                right_num = float(right_val)
            except (ValueError, TypeError):
                return False
            if op == ">":
                return left_num > right_num
            if op == "<":
                return left_num < right_num
            if op == ">=":
                return left_num >= right_num
            if op == "<=":
                return left_num <= right_num

    # Boolean check for variable existence
    if _SAFE_VAR_PATTERN.match(expression):
        return bool(variables.get(expression))

    raise ValueError(f"Invalid condition expression: {expression}")


async def _broadcast_ws(message: dict[str, Any], execution_id: str | None = None) -> None:
    """Broadcast a message to WebSocket subscribers for a specific execution."""
    async with _ws_subscribers_lock:
        targets = set()
        if execution_id and execution_id in _ws_subscribers:
            targets = set(_ws_subscribers[execution_id])  # snapshot
    disconnected = []
    for ws in targets:
        try:
            await ws.send_json(message)
        except Exception:
            disconnected.append(ws)
    if disconnected:
        async with _ws_subscribers_lock:
            for ws in disconnected:
                for eid in list(_ws_subscribers.keys()):
                    _ws_subscribers[eid].discard(ws)
                    if not _ws_subscribers[eid]:
                        del _ws_subscribers[eid]


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------


@router.get("/workflows")
async def list_workflows(request: Request, limit: int = 50, _auth: None = Depends(_verify_api_key)):
    """List all workflows."""
    store = _get_store(request)
    summaries = store.list(limit=limit)
    return {"workflows": [s.model_dump() for s in summaries], "total": len(summaries)}


@router.post("/workflows", status_code=201)
async def create_workflow(request: Request, body: CreateWorkflowRequest, _auth: None = Depends(_verify_api_key)):
    """Create a new workflow."""
    store = _get_store(request)
    _validate_workflow_stages(body.stages)

    workflow = WorkflowDefinition(
        workflow_id=str(uuid.uuid4()),
        name=body.name,
        stages=body.stages,
        triggers=body.triggers,
        variables_schema=body.variables_schema,
        output_schema=body.output_schema,
    )
    saved = await store.save(workflow)
    return saved.model_dump()


@router.get("/workflows/{workflow_id}")
async def get_workflow(request: Request, workflow_id: str, _auth: None = Depends(_verify_api_key)):
    """Get a workflow by ID."""
    store = _get_store(request)
    workflow = store.get(workflow_id)
    if workflow is None:
        raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在")
    return workflow.model_dump()


@router.put("/workflows/{workflow_id}")
async def update_workflow(
    request: Request, workflow_id: str, body: CreateWorkflowRequest,
    _auth: None = Depends(_verify_api_key),
):
    """Update an existing workflow."""
    store = _get_store(request)
    existing = store.get(workflow_id)
    if existing is None:
        raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在")

    _validate_workflow_stages(body.stages)

    existing.name = body.name
    existing.stages = body.stages
    existing.triggers = body.triggers
    existing.variables_schema = body.variables_schema
    existing.output_schema = body.output_schema
    existing.version += 1
    saved = await store.save(existing)
    return saved.model_dump()


@router.delete("/workflows/{workflow_id}")
async def delete_workflow(request: Request, workflow_id: str, _auth: None = Depends(_verify_api_key)):
    """Delete a workflow."""
    store = _get_store(request)
    deleted = await store.delete(workflow_id)
    if not deleted:
        raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在")
    return {"message": "已删除"}


@router.post("/workflows/{workflow_id}/execute")
async def execute_workflow(
    request: Request, workflow_id: str, body: ExecuteWorkflowRequest
):
    """Execute a workflow."""
    store = _get_store(request)
    workflow = store.get(workflow_id)
    if workflow is None:
        raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在")

    execution = await store.create_execution(workflow_id)
    execution.variables = body.variables

    # Start execution in background
    task = asyncio.create_task(
        _execute_workflow(workflow, execution, body.variables, store=store)
    )
    store._running_tasks[execution.execution_id] = task
    task.add_done_callback(lambda t: store._running_tasks.pop(execution.execution_id, None))

    return {
        "execution_id": execution.execution_id,
        "workflow_id": workflow_id,
        "status": execution.status,
    }


@router.get("/workflows/executions/{execution_id}")
async def get_execution(request: Request, execution_id: str):
    """Get execution status."""
    store = _get_store(request)
    execution = store.get_execution(execution_id)
    if execution is None:
        raise HTTPException(
            status_code=404, detail=f"执行记录 '{execution_id}' 不存在"
        )
    return execution.model_dump()


@router.post("/workflows/executions/{execution_id}/approve")
async def approve_execution(
    request: Request, execution_id: str, body: ApproveRequest,
    _auth: None = Depends(_verify_api_key),
):
    """Approve a paused approval node."""
    store = _get_store(request)
    execution = store.get_execution(execution_id)
    if execution is None:
        raise HTTPException(
            status_code=404, detail=f"执行记录 '{execution_id}' 不存在"
        )

    exec_lock = store.get_execution_lock(execution_id)
    async with exec_lock:
        # Re-fetch execution after acquiring lock
        execution = store.get_execution(execution_id)
        if execution is None:
            raise HTTPException(
                status_code=404, detail=f"执行记录 '{execution_id}' 不存在"
            )
        if execution.status != "paused":
            raise HTTPException(
                status_code=400, detail="当前执行状态不是等待审批"
            )

        if body.approved:
            if execution.current_stage:
                execution.stage_results[execution.current_stage] = {
                    "status": "approved",
                    "approver": "user",
                    "comment": body.comment,
                }
            execution.status = "running"
            await store.update_execution(
                execution.execution_id,
                status="running",
                stage_results=execution.stage_results,
            )
            # Resume the waiting execution by setting the approval event
            stage_name = execution.current_stage
            if stage_name:
                event_key = f"{execution_id}:{stage_name}"
                if event_key in store._approval_events:
                    store._approval_events[event_key].set()
        else:
            execution.status = "cancelled"
            execution.completed_at = datetime.now(timezone.utc).isoformat()
            if execution.current_stage:
                execution.stage_results[execution.current_stage] = {
                    "status": "rejected",
                    "approver": "user",
                    "comment": body.comment,
                }
            await store.update_execution(
                execution.execution_id,
                status="cancelled",
                completed_at=execution.completed_at,
                stage_results=execution.stage_results,
            )
            # Set the approval event so the waiting coroutine can observe the cancelled state
            stage_name = execution.current_stage
            if stage_name:
                event_key = f"{execution_id}:{stage_name}"
                if event_key in store._approval_events:
                    store._approval_events[event_key].set()

    return execution.model_dump()


@router.post("/workflows/executions/{execution_id}/cancel")
async def cancel_execution(request: Request, execution_id: str, _auth: None = Depends(_verify_api_key)):
    """Cancel a running execution."""
    store = _get_store(request)
    execution = store.get_execution(execution_id)
    if execution is None:
        raise HTTPException(
            status_code=404, detail=f"执行记录 '{execution_id}' 不存在"
        )

    exec_lock = store.get_execution_lock(execution_id)
    async with exec_lock:
        # Re-fetch execution after acquiring lock
        execution = store.get_execution(execution_id)
        if execution is None:
            raise HTTPException(
                status_code=404, detail=f"执行记录 '{execution_id}' 不存在"
            )
        if execution.status not in ("running", "paused", "pending"):
            raise HTTPException(
                status_code=400, detail="当前执行状态无法取消"
            )

        execution.status = "cancelled"
        execution.completed_at = datetime.now(timezone.utc).isoformat()
        await store.update_execution(
            execution.execution_id,
            status="cancelled",
            completed_at=execution.completed_at,
        )
        # Set any pending approval event so a paused workflow can observe the cancelled state
        if hasattr(execution, "current_stage") and execution.current_stage:
            event_key = f"{execution_id}:{execution.current_stage}"
            if event_key in store._approval_events:
                store._approval_events[event_key].set()

    return execution.model_dump()


@router.websocket("/workflows/ws")
async def workflow_websocket(websocket: WebSocket):
    """Real-time workflow execution progress WebSocket."""
    # Authentication
    configured_api_key: str | None = None
    if hasattr(websocket.app.state, "server_config") and websocket.app.state.server_config:
        configured_api_key = websocket.app.state.server_config.api_key
    if configured_api_key is None and hasattr(websocket.app.state, "api_key"):
        configured_api_key = websocket.app.state.api_key

    if configured_api_key:
        provided = websocket.query_params.get("api_key")
        if not hmac.compare_digest((provided or "").encode(), configured_api_key.encode()):
            await websocket.close(code=4001, reason="Invalid or missing api_key")
            return

    await websocket.accept()

    # Determine execution_id from query params; may be None for backward compat
    execution_id = websocket.query_params.get("execution_id")
    subscribed_execution_id: str | None = None

    if execution_id:
        await _ws_subscribe(execution_id, websocket)
        subscribed_execution_id = execution_id

    try:
        while True:
            try:
                raw = await asyncio.wait_for(websocket.receive_text(), timeout=120.0)
                # Handle subscription messages
                try:
                    msg = json.loads(raw)
                    if isinstance(msg, dict) and msg.get("type") == "subscribe":
                        new_eid = msg.get("execution_id")
                        if new_eid:
                            # Unsubscribe from previous if any
                            if subscribed_execution_id:
                                await _ws_unsubscribe(subscribed_execution_id, websocket)
                            await _ws_subscribe(new_eid, websocket)
                            subscribed_execution_id = new_eid
                except (json.JSONDecodeError, KeyError):
                    pass
            except asyncio.TimeoutError:
                await websocket.close(code=1000, reason="Heartbeat timeout")
                return
            # Keep connection alive - messages are primarily server-push
    except WebSocketDisconnect:
        logger.debug("Workflow WebSocket disconnected")
    except Exception as e:
        logger.error(f"Workflow WebSocket error: {e}")
    finally:
        if subscribed_execution_id:
            await _ws_unsubscribe(subscribed_execution_id, websocket)
