"""
Agent 可视化编排 API（管理员）
Phase 3.6: Agent Visual Orchestration - DAG Workflow Editor
"""
from fastapi import APIRouter, Depends, HTTPException, Query, Body, WebSocket, WebSocketDisconnect
from sqlalchemy.orm import Session
from sqlalchemy import or_, func
from pydantic import BaseModel
from typing import Optional, List, Any
import json
import uuid
import asyncio
from datetime import datetime

from ...database import get_db, SessionLocal
from ...models import MarketApp, EvolutionSkill, AIModel, KnowledgeBase
from ...auth import get_current_admin
from ...middleware.log_service import LogService

router = APIRouter()


# ============================================================
# Pydantic models
# ============================================================

class WorkflowNode(BaseModel):
    id: str
    type: str  # start / llm / knowledge / skill / condition / http / code / agent / end
    position: dict  # {x, y}
    data: dict  # {label, config: {...}}


class WorkflowEdge(BaseModel):
    id: str
    source: str
    target: str
    sourceHandle: Optional[str] = None
    targetHandle: Optional[str] = None


class WorkflowCreate(BaseModel):
    name: str
    description: Optional[str] = None
    agent_id: Optional[str] = None
    nodes: List[WorkflowNode] = []
    edges: List[WorkflowEdge] = []
    variables: Optional[dict] = None


class WorkflowUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    agent_id: Optional[str] = None
    nodes: Optional[List[WorkflowNode]] = None
    edges: Optional[List[WorkflowEdge]] = None
    variables: Optional[dict] = None
    status: Optional[str] = None  # draft / published / archived


class WorkflowExecuteRequest(BaseModel):
    input_text: str = ""
    variables: Optional[dict] = None


# ============================================================
# Node type definitions (templates)
# ============================================================

NODE_TEMPLATES = {
    "start": {
        "type": "start",
        "label": "开始",
        "icon": "play",
        "color": "#3B6D11",
        "description": "工作流入口节点",
        "config_schema": {
            "input_variable": {"type": "string", "label": "输入变量名", "default": "input"}
        }
    },
    "llm": {
        "type": "llm",
        "label": "LLM 调用",
        "icon": "brain",
        "color": "#534AB7",
        "description": "调用大语言模型生成回复",
        "config_schema": {
            "model_id": {"type": "select", "label": "模型", "options_from": "models"},
            "system_prompt": {"type": "textarea", "label": "系统提示词", "default": "你是一个有用的AI助手。"},
            "user_prompt": {"type": "textarea", "label": "用户提示词模板", "default": "{input}", "help": "可用变量: {input}, {prev_output}"},
            "temperature": {"type": "number", "label": "温度", "default": 0.7, "min": 0, "max": 2},
            "max_tokens": {"type": "number", "label": "最大Token", "default": 2048}
        }
    },
    "knowledge": {
        "type": "knowledge",
        "label": "知识库检索",
        "icon": "book",
        "color": "#185FA5",
        "description": "从知识库检索相关文档",
        "config_schema": {
            "kb_id": {"type": "select", "label": "知识库", "options_from": "knowledge_bases"},
            "query_template": {"type": "textarea", "label": "查询模板", "default": "{input}"},
            "top_k": {"type": "number", "label": "返回数量", "default": 3}
        }
    },
    "skill": {
        "type": "skill",
        "label": "Skill 调用",
        "icon": "zap",
        "color": "#0F6E56",
        "description": "执行已注册的 Skill",
        "config_schema": {
            "skill_id": {"type": "select", "label": "Skill", "options_from": "skills"},
            "params_template": {"type": "textarea", "label": "参数模板 (JSON)", "default": "{}"}
        }
    },
    "condition": {
        "type": "condition",
        "label": "条件分支",
        "icon": "git-branch",
        "color": "#BA7517",
        "description": "根据条件选择分支",
        "config_schema": {
            "condition_expr": {"type": "textarea", "label": "条件表达式", "default": "len(input) > 0", "help": "Python表达式, 返回True走sourceHandle=true, False走false"}
        }
    },
    "http": {
        "type": "http",
        "label": "HTTP 请求",
        "icon": "globe",
        "color": "#D4537E",
        "description": "发送 HTTP 请求",
        "config_schema": {
            "url": {"type": "string", "label": "URL", "default": "https://api.example.com"},
            "method": {"type": "select", "label": "方法", "options": ["GET", "POST", "PUT", "DELETE"], "default": "GET"},
            "headers": {"type": "textarea", "label": "请求头 (JSON)", "default": "{}"},
            "body_template": {"type": "textarea", "label": "请求体模板", "default": ""}
        }
    },
    "code": {
        "type": "code",
        "label": "代码执行",
        "icon": "terminal",
        "color": "#5F5E5A",
        "description": "执行 Python 代码片段",
        "config_schema": {
            "code": {"type": "textarea", "label": "Python 代码", "default": "result = input\noutput = result", "help": "可用变量: input, prev_output, context. 必须设置 output 变量"}
        }
    },
    "agent": {
        "type": "agent",
        "label": "Agent 节点",
        "icon": "bot",
        "color": "#7C3AED",
        "description": "调用平台 AI Agent — 支持并行多 Agent 协作",
        "config_schema": {
            "agent_id": {"type": "select", "label": "选择 Agent", "options_from": "agents"},
            "prompt": {"type": "textarea", "label": "提示词模板", "default": "{input}", "help": "可用变量: {input}, {prev_output}, {context} 中的变量"},
            "system_prompt": {"type": "textarea", "label": "系统提示词 (覆盖Agent默认)", "default": "", "help": "留空则使用Agent默认系统提示词"},
            "temperature": {"type": "number", "label": "温度 (覆盖默认)", "default": 0.7, "min": 0, "max": 2},
            "max_tokens": {"type": "number", "label": "最大Token", "default": 4096},
            "output_variable": {"type": "string", "label": "输出变量名", "default": "", "help": "将Agent输出存入context.{变量名}，供后续节点使用"}
        }
    },
    "end": {
        "type": "end",
        "label": "结束",
        "icon": "stop",
        "color": "#A32D2D",
        "description": "工作流出口节点",
        "config_schema": {
            "output_variable": {"type": "string", "label": "输出变量名", "default": "output"}
        }
    }
}


# ============================================================
# Helpers
# ============================================================

def _get_workflow_table():
    """Get the agent_workflows table via raw SQL (avoid model changes)"""
    from sqlalchemy import text
    return text


def workflow_to_dict(row) -> dict:
    """Convert a workflow DB row (dict) to response dict"""
    return {
        "id": row.get("id"),
        "name": row.get("name"),
        "description": row.get("description"),
        "agent_id": row.get("agent_id"),
        "nodes": json.loads(row["nodes"]) if isinstance(row.get("nodes"), str) else (row.get("nodes") or []),
        "edges": json.loads(row["edges"]) if isinstance(row.get("edges"), str) else (row.get("edges") or []),
        "variables": json.loads(row["variables"]) if isinstance(row.get("variables"), str) else (row.get("variables") or {}),
        "status": row.get("status", "draft"),
        "version": row.get("version", 1),
        "created_by": row.get("created_by"),
        "created_at": row["created_at"].isoformat() if row.get("created_at") else None,
        "updated_at": row["updated_at"].isoformat() if row.get("updated_at") else None,
    }


def _execute_raw(db, sql, params=None):
    """Execute raw SQL and return rows as dicts"""
    from sqlalchemy import text
    result = db.execute(text(sql), params or {})
    if result.returns_rows:
        cols = result.keys()
        return [dict(zip(cols, row)) for row in result.fetchall()]
    return []


# ============================================================
# Endpoints
# ============================================================

@router.get("/templates", summary="获取节点模板")
async def get_node_templates(
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取所有可用的节点类型模板和配置 schema"""
    # Fetch options for select fields
    options = {}

    # Models
    models = db.query(AIModel).filter(AIModel.is_enabled == True).all()
    options["models"] = [{"value": m.id, "label": f"{getattr(m, 'display_name', None) or m.name}"} for m in models]

    # Knowledge bases
    kbs = db.query(KnowledgeBase).filter(KnowledgeBase.status == "active").all()
    options["knowledge_bases"] = [{"value": kb.id, "label": kb.name} for kb in kbs]

    # Skills
    skills = db.query(EvolutionSkill).filter(EvolutionSkill.state == "active").all()
    options["skills"] = [{"value": s.id, "label": s.skill_name} for s in skills]

    # Agents (for multi-agent orchestration - Phase G3)
    agents = db.query(MarketApp).filter(MarketApp.type == "agent", MarketApp.status == "published").all()
    options["agents"] = [{"value": a.id, "label": a.name} for a in agents]

    return {
        "code": 0,
        "data": {
            "node_templates": NODE_TEMPLATES,
            "options": options,
        }
    }


@router.get("/", summary="获取工作流列表")
async def list_workflows(
    keyword: str = Query(None),
    status: str = Query(None),
    page: int = Query(1, ge=1),
    page_size: int = Query(15, ge=1, le=100),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取所有工作流列表（分页）"""
    conditions = []
    params = {}

    if keyword:
        conditions.append("(name LIKE :kw OR description LIKE :kw)")
        params["kw"] = f"%{keyword}%"
    if status:
        conditions.append("status = :status")
        params["status"] = status

    where_clause = " AND ".join(conditions) if conditions else "1=1"

    # Count
    count_sql = f"SELECT COUNT(*) as total FROM agent_workflows WHERE {where_clause}"
    count_rows = _execute_raw(db, count_sql, params)
    total = count_rows[0]["total"] if count_rows else 0

    # Query
    offset = (page - 1) * page_size
    query_sql = f"""
        SELECT * FROM agent_workflows
        WHERE {where_clause}
        ORDER BY updated_at DESC
        LIMIT :limit OFFSET :offset
    """
    params["limit"] = page_size
    params["offset"] = offset
    rows = _execute_raw(db, query_sql, params)

    return {
        "code": 0,
        "data": {
            "total": total,
            "page": page,
            "page_size": page_size,
            "items": [workflow_to_dict(r) for r in rows],
        }
    }


@router.get("/stats", summary="工作流统计")
async def get_workflow_stats(
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取工作流统计数据"""
    rows = _execute_raw(db, "SELECT status, COUNT(*) as cnt FROM agent_workflows GROUP BY status")
    stats = {r["status"]: r["cnt"] for r in rows}
    total = sum(stats.values())

    return {
        "code": 0,
        "data": {
            "total": total,
            "draft": stats.get("draft", 0),
            "published": stats.get("published", 0),
            "archived": stats.get("archived", 0),
        }
    }


@router.get("/{workflow_id}", summary="获取工作流详情")
async def get_workflow(
    workflow_id: str,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取单个工作流详情"""
    rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
    if not rows:
        raise HTTPException(status_code=404, detail="工作流不存在")

    result = workflow_to_dict(rows[0])

    # Get agent info if linked
    if result.get("agent_id"):
        agent = db.query(MarketApp).filter(MarketApp.id == result["agent_id"]).first()
        if agent:
            result["agent_name"] = agent.name

    return {"code": 0, "data": result}


@router.post("/", summary="创建工作流")
async def create_workflow(
    req: WorkflowCreate,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """创建新的工作流"""
    from sqlalchemy import text

    wf_id = str(uuid.uuid4())
    now = datetime.now()

    # Validate: must have exactly one start node
    start_nodes = [n for n in req.nodes if n.type == "start"]
    if len(start_nodes) == 0 and len(req.nodes) > 0:
        raise HTTPException(status_code=400, detail="工作流必须包含一个开始节点")
    if len(start_nodes) > 1:
        raise HTTPException(status_code=400, detail="工作流只能有一个开始节点")

    sql = text("""
        INSERT INTO agent_workflows (id, name, description, agent_id, nodes, edges, variables, status, version, created_by, created_at, updated_at)
        VALUES (:id, :name, :description, :agent_id, :nodes, :edges, :variables, 'draft', 1, :created_by, :created_at, :updated_at)
    """)

    db.execute(sql, {
        "id": wf_id,
        "name": req.name,
        "description": req.description or "",
        "agent_id": req.agent_id,
        "nodes": json.dumps([n.dict() for n in req.nodes], ensure_ascii=False),
        "edges": json.dumps([e.dict() for e in req.edges], ensure_ascii=False),
        "variables": json.dumps(req.variables or {}, ensure_ascii=False),
        "created_by": admin_user.id,
        "created_at": now,
        "updated_at": now,
    })
    db.commit()

    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="create_workflow",
        target_type="workflow",
        target_id=wf_id,
        detail=f"创建工作流: {req.name}",
    )

    # Return created
    rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": wf_id})
    return {"code": 0, "message": "工作流创建成功", "data": workflow_to_dict(rows[0])}


@router.put("/{workflow_id}", summary="更新工作流")
async def update_workflow(
    workflow_id: str,
    req: WorkflowUpdate,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """更新工作流"""
    from sqlalchemy import text

    rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
    if not rows:
        raise HTTPException(status_code=404, detail="工作流不存在")

    existing = rows[0]
    sets = []
    params = {"id": workflow_id, "updated_at": datetime.now()}

    if req.name is not None:
        sets.append("name = :name")
        params["name"] = req.name
    if req.description is not None:
        sets.append("description = :description")
        params["description"] = req.description
    if req.agent_id is not None:
        sets.append("agent_id = :agent_id")
        params["agent_id"] = req.agent_id
    if req.nodes is not None:
        sets.append("nodes = :nodes")
        params["nodes"] = json.dumps([n.dict() for n in req.nodes], ensure_ascii=False)
    if req.edges is not None:
        sets.append("edges = :edges")
        params["edges"] = json.dumps([e.dict() for e in req.edges], ensure_ascii=False)
    if req.variables is not None:
        sets.append("variables = :variables")
        params["variables"] = json.dumps(req.variables, ensure_ascii=False)
    if req.status is not None:
        sets.append("status = :status")
        params["status"] = req.status

    if sets:
        sets.append("version = version + 1")
        sql = text(f"UPDATE agent_workflows SET {', '.join(sets)} WHERE id = :id")
        db.execute(sql, params)
        db.commit()

    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="update_workflow",
        target_type="workflow",
        target_id=workflow_id,
        detail=f"更新工作流: {existing.get('name', '')}",
    )

    rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
    return {"code": 0, "message": "工作流已更新", "data": workflow_to_dict(rows[0])}


@router.delete("/{workflow_id}", summary="删除工作流")
async def delete_workflow(
    workflow_id: str,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """删除工作流"""
    from sqlalchemy import text

    rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
    if not rows:
        raise HTTPException(status_code=404, detail="工作流不存在")

    wf_name = rows[0].get("name", "")
    db.execute(text("DELETE FROM agent_workflows WHERE id = :id"), {"id": workflow_id})
    db.commit()

    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="delete_workflow",
        target_type="workflow",
        target_id=workflow_id,
        detail=f"删除工作流: {wf_name}",
    )

    return {"code": 0, "message": "工作流已删除"}


@router.put("/{workflow_id}/toggle", summary="发布/归档工作流")
async def toggle_workflow(
    workflow_id: str,
    is_active: bool = Query(..., description="True=发布, False=归档"),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """发布或归档工作流"""
    from sqlalchemy import text

    rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
    if not rows:
        raise HTTPException(status_code=404, detail="工作流不存在")

    new_status = "published" if is_active else "archived"
    db.execute(
        text("UPDATE agent_workflows SET status = :status, updated_at = :now WHERE id = :id"),
        {"status": new_status, "now": datetime.now(), "id": workflow_id}
    )
    db.commit()

    return {"code": 0, "message": f"工作流已{'发布' if is_active else '归档'}"}


@router.post("/{workflow_id}/execute", summary="执行工作流")
async def execute_workflow(
    workflow_id: str,
    req: WorkflowExecuteRequest,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """执行工作流 — 按 DAG 拓扑顺序处理节点，同层节点并行执行 (Phase G3)"""
    rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
    if not rows:
        raise HTTPException(status_code=404, detail="工作流不存在")

    wf = workflow_to_dict(rows[0])
    nodes = wf["nodes"]
    edges = wf["edges"]

    if not nodes:
        raise HTTPException(status_code=400, detail="工作流没有节点")

    # Build structures
    node_map = {n["id"]: n for n in nodes}
    edge_map = {}
    reverse_edge_map = {}  # target -> [sources]
    for e in edges:
        src = e["source"]
        tgt = e["target"]
        if src not in edge_map:
            edge_map[src] = []
        edge_map[src].append({"target": tgt, "sourceHandle": e.get("sourceHandle")})
        if tgt not in reverse_edge_map:
            reverse_edge_map[tgt] = []
        reverse_edge_map[tgt].append(src)

    # Find start nodes
    start_node = None
    for n in nodes:
        if n["type"] == "start":
            start_node = n["id"]
            break
    if not start_node and nodes:
        start_node = nodes[0]["id"]

    # Topological sort to assign levels
    in_degree = {n["id"]: len(reverse_edge_map.get(n["id"], [])) for n in nodes}
    levels = {}
    queue = [start_node]
    levels[start_node] = 0

    while queue:
        nid = queue.pop(0)
        level = levels[nid]
        for edge in edge_map.get(nid, []):
            tgt = edge["target"]
            if tgt in node_map:
                new_level = level + 1
                if tgt not in levels or new_level > levels[tgt]:
                    levels[tgt] = new_level
                in_degree[tgt] -= 1
                if in_degree[tgt] <= 0:
                    queue.append(tgt)

    # Group nodes by level
    max_level = max(levels.values()) if levels else 0
    level_groups = {i: [] for i in range(max_level + 1)}
    for nid, lvl in levels.items():
        level_groups[lvl].append(nid)

    # Context
    context = {
        "input": req.input_text,
        "variables": req.variables or {},
        "prev_output": "",
        "results": {},
        "log": [],
    }

    final_output = ""
    abort = False

    # Execute level by level
    for lvl in range(max_level + 1):
        if abort:
            break

        level_nodes = level_groups[lvl]
        if not level_nodes:
            continue

        # Execute nodes in this level
        tasks = []
        for nid in level_nodes:
            node = node_map.get(nid)
            if not node:
                continue
            task = _execute_node_async(node, context, db, edge_map, node_map, workflow_id)
            tasks.append(task)

        # Parallel execution for same-level nodes
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for res in results:
            if isinstance(res, Exception):
                context["log"].append({
                    "node_id": "unknown",
                    "node_label": "unknown",
                    "node_type": "unknown",
                    "success": False,
                    "error": str(res),
                })
                abort = True
                continue

            if res is None:
                continue

            nid = res["node_id"]
            context["results"][nid] = res["result"]

            if res["success"]:
                context["prev_output"] = res["result"].get("output", "")
                context["log"].append({
                    "node_id": nid,
                    "node_label": res["node_label"],
                    "node_type": res["node_type"],
                    "success": True,
                    "output_preview": str(res["result"].get("output", ""))[:200],
                })

                if res["node_type"] == "end":
                    final_output = res["result"].get("output", "")
                    abort = True
                    break

                # Handle condition node branching — update context for next level
                if res["node_type"] == "condition" and "next_nodes" in res:
                    # The condition routing is applied by _execute_node_async
                    pass
            else:
                context["log"].append({
                    "node_id": nid,
                    "node_label": res["node_label"],
                    "node_type": res["node_type"],
                    "success": False,
                    "error": res.get("error", "未知错误"),
                })
                abort = True
                break

    if not final_output:
        final_output = context["prev_output"]

    return {
        "code": 0,
        "data": {
            "output": final_output,
            "log": context["log"],
            "results": context["results"],
        }
    }


@router.websocket("/{workflow_id}/execute-stream")
async def execute_workflow_stream(
    websocket: WebSocket,
    workflow_id: str,
):
    """WebSocket 端点 — 实时流式推送工作流执行进度 (Phase G3)"""
    await websocket.accept()

    try:
        # Receive execution params
        data = await websocket.receive_json()
        input_text = data.get("input_text", "")

        # Create DB session
        db = SessionLocal()
        try:
            rows = _execute_raw(db, "SELECT * FROM agent_workflows WHERE id = :id", {"id": workflow_id})
            if not rows:
                await websocket.send_json({"type": "workflow_error", "data": {"output": "工作流不存在"}})
                await websocket.close()
                return

            wf = workflow_to_dict(rows[0])
            nodes = wf["nodes"]
            edges = wf["edges"]

            if not nodes:
                await websocket.send_json({"type": "workflow_error", "data": {"output": "工作流没有节点"}})
                await websocket.close()
                return

            # Build structures
            node_map = {n["id"]: n for n in nodes}
            edge_map = {}
            reverse_edge_map = {}
            for e in edges:
                src = e["source"]
                tgt = e["target"]
                if src not in edge_map:
                    edge_map[src] = []
                edge_map[src].append({"target": tgt, "sourceHandle": e.get("sourceHandle")})
                if tgt not in reverse_edge_map:
                    reverse_edge_map[tgt] = []
                reverse_edge_map[tgt].append(src)

            # Find start
            start_node = None
            for n in nodes:
                if n["type"] == "start":
                    start_node = n["id"]
                    break
            if not start_node and nodes:
                start_node = nodes[0]["id"]

            # Topo sort
            in_degree = {n["id"]: len(reverse_edge_map.get(n["id"], [])) for n in nodes}
            levels = {start_node: 0}
            queue = [start_node]
            while queue:
                nid = queue.pop(0)
                lvl = levels[nid]
                for edge in edge_map.get(nid, []):
                    tgt = edge["target"]
                    if tgt in node_map:
                        nl = lvl + 1
                        if tgt not in levels or nl > levels[tgt]:
                            levels[tgt] = nl
                        in_degree[tgt] -= 1
                        if in_degree[tgt] <= 0:
                            queue.append(tgt)

            max_level = max(levels.values()) if levels else 0
            level_groups = {i: [] for i in range(max_level + 1)}
            for nid, lvl in levels.items():
                level_groups[lvl].append(nid)

            context = {
                "input": input_text,
                "variables": {},
                "prev_output": "",
                "results": {},
                "log": [],
            }

            final_output = ""
            abort = False

            for lvl in range(max_level + 1):
                if abort:
                    break

                level_nodes = level_groups[lvl]
                if not level_nodes:
                    continue

                tasks_list = []
                for nid in level_nodes:
                    node = node_map.get(nid)
                    if not node:
                        continue

                    # Send node_start
                    await websocket.send_json({
                        "type": "node_start",
                        "node_id": nid,
                        "node_type": node["type"],
                        "node_label": node.get("data", {}).get("label", node["type"]),
                    })

                    task = _execute_node_async(node, context, db, edge_map, node_map, workflow_id)
                    tasks_list.append(task)

                results = await asyncio.gather(*tasks_list, return_exceptions=True)

                for res in results:
                    if isinstance(res, Exception):
                        await websocket.send_json({
                            "type": "node_error",
                            "node_id": "unknown",
                            "node_label": "unknown",
                            "node_type": "unknown",
                            "error": str(res),
                        })
                        abort = True
                        continue

                    if res is None:
                        continue

                    nid = res["node_id"]
                    context["results"][nid] = res["result"]

                    if res["success"]:
                        context["prev_output"] = res["result"].get("output", "")

                        await websocket.send_json({
                            "type": "node_complete",
                            "node_id": nid,
                            "node_label": res["node_label"],
                            "node_type": res["node_type"],
                            "output_preview": str(res["result"].get("output", ""))[:200],
                        })

                        if res["node_type"] == "end":
                            final_output = res["result"].get("output", "")
                            abort = True
                            break
                    else:
                        await websocket.send_json({
                            "type": "node_error",
                            "node_id": nid,
                            "node_label": res["node_label"],
                            "node_type": res["node_type"],
                            "error": res.get("error", "未知错误"),
                        })
                        abort = True
                        break

            if not final_output:
                final_output = context["prev_output"]

            await websocket.send_json({
                "type": "workflow_complete",
                "data": {
                    "output": final_output,
                    "log": context["log"],
                }
            })

        finally:
            db.close()

    except WebSocketDisconnect:
        pass
    except Exception as e:
        try:
            await websocket.send_json({
                "type": "workflow_error",
                "data": {"output": f"执行失败: {str(e)}"}
            })
        except Exception:
            pass
    finally:
        try:
            await websocket.close()
        except Exception:
            pass


def _execute_node(node_type: str, config: dict, context: dict, db: Session) -> dict:
    """Execute a single node and return result"""

    if node_type == "start":
        return {"output": context["input"]}

    elif node_type == "llm":
        return _execute_llm_node(config, context, db)

    elif node_type == "knowledge":
        return _execute_knowledge_node(config, context, db)

    elif node_type == "skill":
        return _execute_skill_node(config, context, db)

    elif node_type == "condition":
        return {"output": context["prev_output"]}

    elif node_type == "http":
        return _execute_http_node(config, context)

    elif node_type == "code":
        return _execute_code_node(config, context)

    elif node_type == "agent":
        return _execute_agent_node(config, context, db)

    elif node_type == "end":
        var_name = config.get("output_variable", "output")
        return {"output": context.get(var_name, context.get("prev_output", ""))}

    return {"output": ""}


async def _execute_node_async(node: dict, context: dict, db: Session, edge_map: dict, node_map: dict, workflow_id: str) -> Optional[dict]:
    """Execute a single node async, returning structured result for parallel orchestration"""
    nid = node["id"]
    node_type = node["type"]
    node_config = node.get("data", {}).get("config", {})
    node_label = node.get("data", {}).get("label", node_type)

    try:
        # Publish node start event
        from app.api.admin.acp_gateway import publish_workflow_event
        publish_workflow_event(workflow_id, nid, {
            "event": "node_start",
            "node_type": node_type,
            "node_label": node_label,
        })

        if node_type == "agent":
            # Async agent invocation — may take time
            result = await _execute_agent_node_async(node_config, context, db, workflow_id, nid)

            # Store output in context variable if specified
            output_var = node_config.get("output_variable", "")
            if output_var and result.get("output"):
                context[output_var] = result["output"]

            publish_workflow_event(workflow_id, nid, {
                "event": "node_complete",
                "node_type": node_type,
                "node_label": node_label,
                "output_preview": str(result.get("output", ""))[:200],
            })
        else:
            # Sync node execution (run in thread)
            result = await asyncio.to_thread(_execute_node, node_type, node_config, context, db)
            publish_workflow_event(workflow_id, nid, {
                "event": "node_complete",
                "node_type": node_type,
                "node_label": node_label,
                "output_preview": str(result.get("output", ""))[:200],
            })

        return {
            "node_id": nid,
            "node_label": node_label,
            "node_type": node_type,
            "success": True,
            "result": result,
        }

    except Exception as e:
        from app.api.admin.acp_gateway import publish_workflow_event
        publish_workflow_event(workflow_id, nid, {
            "event": "node_error",
            "node_type": node_type,
            "node_label": node_label,
            "error": str(e),
        })
        return {
            "node_id": nid,
            "node_label": node_label,
            "node_type": node_type,
            "success": False,
            "error": str(e),
        }


async def _execute_agent_node_async(config: dict, context: dict, db: Session, workflow_id: str, node_id: str) -> dict:
    """Execute agent node via ACP Gateway — async"""
    from app.api.admin.acp_gateway import get_acp_gateway

    agent_id = config.get("agent_id")
    if not agent_id:
        return {"output": "[Agent] 未选择 Agent", "error": "no agent_id"}

    # Build prompt from template
    prompt = config.get("prompt", "{input}")
    prompt = prompt.replace("{input}", str(context.get("input", "")))
    prompt = prompt.replace("{prev_output}", str(context.get("prev_output", "")))
    # Substitute context variables
    for key, val in context.items():
        if isinstance(val, str):
            prompt = prompt.replace("{" + key + "}", val)

    system_prompt = config.get("system_prompt", "")
    temperature = float(config.get("temperature", 0.7))
    max_tokens = int(config.get("max_tokens", 4096))

    gateway = get_acp_gateway()

    # Set notification handler for Redis EventBus
    from app.api.admin.acp_gateway import publish_workflow_event
    gateway.set_notification_handler(
        lambda data: publish_workflow_event(workflow_id, node_id, data)
    )

    result = await gateway.invoke_agent(
        agent_id=agent_id,
        prompt=prompt,
        system_prompt=system_prompt,
        temperature=temperature,
        max_tokens=max_tokens,
        context=context,
        db_session=db,
    )

    return result


def _execute_llm_node(config: dict, context: dict, db: Session) -> dict:
    """Execute LLM call"""
    import httpx

    model_id = config.get("model_id")
    if not model_id:
        return {"output": "[LLM] 未配置模型", "error": "no model_id"}

    model = db.query(AIModel).filter(AIModel.id == model_id).first()
    if not model:
        return {"output": "[LLM] 模型不存在", "error": "model not found"}

    system_prompt = config.get("system_prompt", "你是一个有用的AI助手。")
    user_prompt = config.get("user_prompt", "{input}")

    # Template substitution
    user_prompt = user_prompt.replace("{input}", str(context.get("input", "")))
    user_prompt = user_prompt.replace("{prev_output}", str(context.get("prev_output", "")))

    api_key = model.api_key or ""
    base_url = model.api_base or "https://api.siliconflow.cn/v1"
    model_name = model.model_id or model.name

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model_name,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        "temperature": float(config.get("temperature", 0.7)),
        "max_tokens": int(config.get("max_tokens", 2048)),
    }

    try:
        with httpx.Client(timeout=60) as client:
            resp = client.post(f"{base_url}/chat/completions", headers=headers, json=payload)
            resp.raise_for_status()
            data = resp.json()
            output = data["choices"][0]["message"]["content"]
            return {"output": output, "usage": data.get("usage", {})}
    except Exception as e:
        return {"output": f"[LLM] 调用失败: {str(e)}", "error": str(e)}


def _execute_knowledge_node(config: dict, context: dict, db: Session) -> dict:
    """Execute knowledge base search"""
    kb_id = config.get("kb_id")
    if not kb_id:
        return {"output": "[KB] 未配置知识库", "error": "no kb_id"}

    query = config.get("query_template", "{input}")
    query = query.replace("{input}", str(context.get("input", "")))
    query = query.replace("{prev_output}", str(context.get("prev_output", "")))

    top_k = int(config.get("top_k", 3))

    # Use BM25 search (same as chat.py)
    try:
        from ...models import KnowledgeDocumentChunk
        chunks = db.query(KnowledgeDocumentChunk).filter(
            KnowledgeDocumentChunk.knowledge_base_id == kb_id
        ).limit(100).all()

        if not chunks:
            return {"output": "[KB] 知识库无文档", "results": []}

        # Simple keyword matching (BM25 simplified)
        query_words = set(query.lower().split())
        scored = []
        for chunk in chunks:
            content = (chunk.content or "").lower()
            score = sum(1 for w in query_words if w in content)
            if score > 0:
                scored.append((score, chunk.content[:500]))

        scored.sort(key=lambda x: -x[0])
        results = [s[1] for s in scored[:top_k]]

        if not results:
            results = [c.content[:500] for c in chunks[:top_k]]

        combined = "\n\n---\n\n".join(results)
        return {"output": combined, "results": results}
    except Exception as e:
        return {"output": f"[KB] 检索失败: {str(e)}", "error": str(e)}


def _execute_skill_node(config: dict, context: dict, db: Session) -> dict:
    """Execute skill call"""
    skill_id = config.get("skill_id")
    if not skill_id:
        return {"output": "[Skill] 未配置 Skill", "error": "no skill_id"}

    skill = db.query(EvolutionSkill).filter(EvolutionSkill.id == skill_id).first()
    if not skill:
        return {"output": "[Skill] Skill 不存在", "error": "skill not found"}

    # For now, return skill info (actual execution would call the skill handler)
    return {
        "output": f"[Skill] {skill.skill_name} 已调用",
        "skill_id": skill_id,
        "skill_name": skill.skill_name,
    }


def _execute_agent_node(config: dict, context: dict, db: Session) -> dict:
    """Execute agent node (sync wrapper) — calls the async version via asyncio.run"""
    agent_id = config.get("agent_id")
    if not agent_id:
        return {"output": "[Agent] 未选择 Agent", "error": "no agent_id"}

    prompt = config.get("prompt", "{input}")
    prompt = prompt.replace("{input}", str(context.get("input", "")))
    prompt = prompt.replace("{prev_output}", str(context.get("prev_output", "")))

    system_prompt = config.get("system_prompt", "")
    temperature = float(config.get("temperature", 0.7))
    max_tokens = int(config.get("max_tokens", 4096))

    from .acp_gateway import get_acp_gateway

    gateway = get_acp_gateway()

    try:
        import asyncio
        loop = asyncio.get_event_loop()
        if loop.is_running():
            # In async context — create new loop
            new_loop = asyncio.new_event_loop()
            result = new_loop.run_until_complete(
                gateway.invoke_agent(
                    agent_id=agent_id,
                    prompt=prompt,
                    system_prompt=system_prompt,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    context=context,
                    db_session=db,
                )
            )
            new_loop.close()
        else:
            result = loop.run_until_complete(
                gateway.invoke_agent(
                    agent_id=agent_id,
                    prompt=prompt,
                    system_prompt=system_prompt,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    context=context,
                    db_session=db,
                )
            )
        return result
    except Exception as e:
        return {"output": f"[Agent] 调用失败: {str(e)}", "error": str(e)}


def _execute_http_node(config: dict, context: dict) -> dict:
    """Execute HTTP request"""
    import httpx

    url = config.get("url", "")
    method = config.get("method", "GET")
    headers_str = config.get("headers", "{}")
    body_template = config.get("body_template", "")

    # Template substitution
    url = url.replace("{input}", str(context.get("input", "")))
    body_template = body_template.replace("{input}", str(context.get("input", "")))
    body_template = body_template.replace("{prev_output}", str(context.get("prev_output", "")))

    try:
        headers = json.loads(headers_str) if headers_str else {}
    except Exception:
        headers = {}

    try:
        with httpx.Client(timeout=30) as client:
            if method == "GET":
                resp = client.get(url, headers=headers)
            elif method == "POST":
                resp = client.post(url, headers=headers, content=body_template)
            elif method == "PUT":
                resp = client.put(url, headers=headers, content=body_template)
            elif method == "DELETE":
                resp = client.delete(url, headers=headers)
            else:
                return {"output": f"[HTTP] 不支持的方法: {method}"}

            return {"output": resp.text[:2000], "status_code": resp.status_code}
    except Exception as e:
        return {"output": f"[HTTP] 请求失败: {str(e)}", "error": str(e)}


def _execute_code_node(config: dict, context: dict) -> dict:
    """Execute Python code (sandboxed)"""
    code = config.get("code", "output = input")

    # Very basic sandbox — restrict builtins
    safe_builtins = {
        "len": len, "str": str, "int": int, "float": float,
        "bool": bool, "list": list, "dict": dict, "set": set,
        "range": range, "enumerate": enumerate, "zip": zip,
        "min": min, "max": max, "sum": sum, "abs": abs,
        "round": round, "sorted": sorted, "reversed": reversed,
        "print": print, "json": json, "True": True, "False": False, "None": None,
    }

    local_vars = {
        "input": context.get("input", ""),
        "prev_output": context.get("prev_output", ""),
        "context": context,
        "output": "",
    }

    try:
        exec(code, {"__builtins__": safe_builtins}, local_vars)
        return {"output": str(local_vars.get("output", ""))}
    except Exception as e:
        return {"output": f"[Code] 执行失败: {str(e)}", "error": str(e)}