Phase 4.2+4.3: YAML workflow spec + ACA protocol abstraction (v2026.07.19.10)

VERSION

@@ -1 +1 @@
-v2026.07.19.9
+v2026.07.19.10

backend/python/app/api/admin/agents_admin.py

@@ -654,3 +654,131 @@ async def set_agent_orchestration(
     agent.orchestration = og
     db.commit()
     return {"agent_id": agent_id, "orchestration": agent.orchestration, "updated": True}
+
+
+# ============================================================
+# Phase 4.2: YAML workflow import/export/validate/templates
+# ============================================================
+
+import yaml as _yaml
+import os as _os
+from app.core.yaml_workflow import validate_yaml, yaml_to_orchestration, orchestration_to_yaml
+from app.core.orchestrator import validate_graph
+
+_YTMPL = _os.path.join(_os.path.dirname(_os.path.dirname(_os.path.dirname(__file__))), "workflow_templates")
+
+
+class YAMLBody(BaseModel):
+    yaml_text: str
+
+
+@router.post("/orchestration/yaml/validate", summary="校验 YAML 工作流")
+async def yaml_validate(req: YAMLBody, admin_user=Depends(get_current_admin)):
+    try:
+        data = _yaml.safe_load(req.yaml_text)
+    except _yaml.YAMLError as e:
+        return {"ok": False, "errors": [f"YAML parse: {str(e)}"], "warnings": []}
+    r = validate_yaml(data)
+    if r["ok"]:
+        try:
+            validate_graph(yaml_to_orchestration(data))
+        except Exception as e:
+            r["ok"] = False; r["errors"].append(f"Graph fail: {str(e)}")
+    return r
+
+
+@router.post("/{agent_id}/orchestration/yaml/import", summary="YAML 导入编排")
+async def yaml_import(agent_id: str, req: YAMLBody, admin_user=Depends(get_current_admin), db=Depends(get_db)):
+    ag = db.query(MarketApp).filter(MarketApp.id == agent_id, MarketApp.type == "agent").first()
+    if not ag:
+        raise HTTPException(status_code=404, detail="Agent not found")
+    try:
+        data = _yaml.safe_load(req.yaml_text)
+    except _yaml.YAMLError as e:
+        raise HTTPException(status_code=400, detail=f"YAML parse: {str(e)}")
+    v = validate_yaml(data)
+    if not v["ok"]:
+        raise HTTPException(status_code=400, detail="YAML invalid: " + "; ".join(v["errors"]))
+    orch = yaml_to_orchestration(data)
+    try:
+        validate_graph(orch)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"Graph invalid: {str(e)}")
+    ag.orchestration = orch
+    ag.updated_at = datetime.now()
+    db.commit()
+    return {"agent_id": agent_id, "orchestration": ag.orchestration, "imported": True}
+
+
+@router.post("/{agent_id}/orchestration/yaml/export", summary="编排导出 YAML")
+async def yaml_export(agent_id: str, admin_user=Depends(get_current_admin), db=Depends(get_db)):
+    ag = db.query(MarketApp).filter(MarketApp.id == agent_id, MarketApp.type == "agent").first()
+    if not ag:
+        raise HTTPException(status_code=404, detail="Agent not found")
+    if not ag.orchestration:
+        raise HTTPException(status_code=400, detail="No orchestration")
+    yaml_str = _yaml.dump(orchestration_to_yaml(ag.orchestration), allow_unicode=True, default_flow_style=False, sort_keys=False)
+    return {"agent_id": agent_id, "yaml": yaml_str, "version": "1.0"}
+
+
+@router.get("/orchestration/yaml/templates", summary="内置模板")
+async def yaml_templates(admin_user=Depends(get_current_admin)):
+    tmpls = []
+    if _os.path.isdir(_YTMPL):
+        for fp in sorted(_os.listdir(_YTMPL)):
+            if fp.endswith((".yaml", ".yml")):
+                try:
+                    with open(_os.path.join(_YTMPL, fp), encoding="utf-8") as f:
+                        txt = f.read()
+                        d = _yaml.safe_load(txt) or {}
+                        tmpls.append({"filename": fp, "name": d.get("name", fp),
+                                      "description": d.get("description", ""),
+                                      "category": d.get("category", "general"), "yaml": txt})
+                except Exception:
+                    tmpls.append({"filename": fp, "name": fp, "error": True})
+    return {"templates": tmpls, "count": len(tmpls)}
+
+
+# ============================================================
+# Phase 4.3: ACA 协议抽象 — protocols / tools / health
+# ============================================================
+
+from app.core.aca import registry as _aca_registry, init_aca_registry
+
+# 懒初始化：首次调用 ACA 端点时确保 registry 已填充
+_aca_initialized = False
+
+def _ensure_aca():
+    global _aca_initialized
+    if not _aca_initialized:
+        init_aca_registry()
+        _aca_initialized = True
+
+
+@router.get("/aca/protocols", summary="ACA 已注册协议")
+async def aca_protocols(admin_user=Depends(get_current_admin)):
+    _ensure_aca()
+    return [
+        {"name": a.name, "version": a.version, "capabilities": a.capabilities}
+        for a in _aca_registry.list_all()
+    ]
+
+
+@router.get("/aca/protocols/{name}/tools", summary="协议工具列表")
+async def aca_protocol_tools(name: str, admin_user=Depends(get_current_admin)):
+    _ensure_aca()
+    adapter = _aca_registry.get(name)
+    if not adapter:
+        raise HTTPException(status_code=404, detail=f"协议 {name} 未注册")
+    tools = await adapter.list_tools()
+    return {"protocol": name, "tools": tools, "count": len(tools)}
+
+
+@router.post("/aca/protocols/{name}/health", summary="协议健康检查")
+async def aca_protocol_health(name: str, admin_user=Depends(get_current_admin)):
+    _ensure_aca()
+    adapter = _aca_registry.get(name)
+    if not adapter:
+        raise HTTPException(status_code=404, detail=f"协议 {name} 未注册")
+    ok = await adapter.health_check()
+    return {"protocol": name, "healthy": ok}  

backend/python/app/api/admin/market_admin.py

@@ -183,6 +183,11 @@ async def get_all_apps(
                 "install_count": a.install_count,
                 "created_at": a.created_at.isoformat() if a.created_at else None,
                 "updated_at": a.updated_at.isoformat() if a.updated_at else None,
+                # Phase 4.1: 安检/版本化/绿标
+                "version": a.version or "1.0.0",
+                "security_scan": a.security_scan,
+                "verified": bool(a.verified),
+                "dependencies": a.dependencies,
             }
             for a in apps
         ],

backend/python/app/core/aca.py

@@ -0,0 +1,293 @@
+"""Phase 4.3: ACA (Agent Communication Abstraction) 协议抽象层.
+
+设计目标：
+- 定义统一对外工具协议接口，隔离 MCP / A2A / 自定义协议的具体实现。
+- 防止协议绑架：未来 MCP 升级或 A2A 替代 MCP 时，只需增删适配器，不改业务代码。
+- 每个协议适配器被注册为 "protocol adapter"，按名称发现。
+
+接口：
+    ACAProtocol
+        name: str
+        version: str
+        capabilities: list[str]
+        async list_tools() -> list[dict]       # [{name, description, inputSchema}]
+        async call_tool(tool_name, arguments) -> str
+        async health_check() -> bool
+
+    ProtocolRegistry
+        register(adapter: ACAProtocol)
+        get(name: str) -> ACAProtocol
+        list_all() -> list[ACAProtocol]
+        async list_all_tools() -> list[dict]   # 聚合所有协议的 tools，带 protocol tag
+        async call_tool(protocol_name, tool_name, args) -> str
+
+使用示例:
+    from app.core.aca import registry
+    tools = await registry.list_all_tools()
+    # tools: [{"name":"search","protocol":"mcp","description":"..."}, ...]
+    result = await registry.call_tool("mcp", "search", {"query":"..."})
+"""
+from __future__ import annotations
+
+import abc
+import asyncio
+import json
+import logging
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger("aiflow.aca")
+
+# ── 协议接口 ─────────────────────────────────────────────────────────────────
+
+class ACAProtocol(abc.ABC):
+    name: str = ""
+    version: str = "1.0"
+    capabilities: list[str] = []
+
+    @abc.abstractmethod
+    async def list_tools(self) -> list[dict]:
+        """返回 [{name, description, inputSchema}]。"""
+
+    @abc.abstractmethod
+    async def call_tool(self, tool_name: str, arguments: dict) -> str:
+        """调用工具并返回文本结果。"""
+
+    async def health_check(self) -> bool:
+        """健康检查，默认 True。"""
+        return True
+
+    def __repr__(self):
+        return f"<ACA {self.name} v{self.version}>"
+
+
+# ── 注册表 ───────────────────────────────────────────────────────────────────
+
+class ProtocolRegistry:
+    """全局协议注册表，单例。"""
+
+    def __init__(self):
+        self._adapters: dict[str, ACAProtocol] = {}
+
+    def register(self, adapter: ACAProtocol):
+        if not adapter.name:
+            raise ValueError("protocol adapter must have a name")
+        self._adapters[adapter.name] = adapter
+        logger.info("ACA registry: registered %s", adapter)
+
+    def get(self, name: str) -> ACAProtocol | None:
+        return self._adapters.get(name)
+
+    def list_all(self) -> list[ACAProtocol]:
+        return list(self._adapters.values())
+
+    async def list_all_tools(self) -> list[dict]:
+        """聚合所有已注册协议的 tools，每个 tool 标记 protocol。"""
+        tasks = [self._list_with_name(a) for a in self._adapters.values()]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        flat: list[dict] = []
+        for rlist in results:
+            if isinstance(rlist, list):
+                flat.extend(rlist)
+        return flat
+
+    async def _list_with_name(self, adapter: ACAProtocol) -> list[dict]:
+        try:
+            tools = await adapter.list_tools()
+            for t in tools:
+                t["protocol"] = adapter.name
+            return tools
+        except Exception as e:
+            logger.warning("ACA: list_tools failed for %s: %s", adapter.name, e)
+            return []
+
+    async def call_tool(self, protocol_name: str, tool_name: str, arguments: dict) -> str:
+        adapter = self._adapters.get(protocol_name)
+        if not adapter:
+            raise ValueError(f"协议 {protocol_name} 未注册")
+        return await adapter.call_tool(tool_name, arguments)
+
+
+# 全局单例
+registry = ProtocolRegistry()
+
+
+# ── MCP 适配器 ───────────────────────────────────────────────────────────────
+
+class McpProtocolAdapter(ACAProtocol):
+    """包装平台现有的 MCP 客户端 (app.mcp_client)，提供统一协议接口。"""
+
+    def __init__(self, server_configs: list[dict] = None):
+        self.name = "mcp"
+        self.version = "1.0"
+        self.capabilities = ["tools", "resources", "prompts"]
+        self._server_configs = server_configs or []
+        self._tools_cache: list[dict] = []
+        self._cache_ts: float = 0
+
+    async def list_tools(self) -> list[dict]:
+        # Use cached tools if available (TTL 30s)
+        import time
+        if self._tools_cache and (time.time() - self._cache_ts) < 30:
+            return [dict(t) for t in self._tools_cache]
+        from app.mcp_client import list_mcp_tools
+        all_tools: list[dict] = []
+        for server in self._server_configs:
+            try:
+                tools = await list_mcp_tools(server)
+                for t in tools:
+                    t["_mcp_server_id"] = server.get("id", "")
+                    t["_mcp_server_name"] = server.get("name", "")
+                all_tools.extend(tools)
+            except Exception as e:
+                logger.warning("MCP list_tools failed for %s: %s", server.get("name"), e)
+        self._tools_cache = [dict(t) for t in all_tools]
+        self._cache_ts = time.time()
+        return list(self._tools_cache)
+
+    async def call_tool(self, tool_name: str, arguments: dict) -> str:
+        from app.mcp_client import call_mcp_tool
+        # Find the server that has this tool (check cache first)
+        for t in self._tools_cache:
+            if t.get("name") == tool_name:
+                sid = t.get("_mcp_server_id", "")
+                server = next((s for s in self._server_configs if s.get("id") == sid), None)
+                if server:
+                    return await call_mcp_tool(server, tool_name, arguments)
+        # Fallback: try first available server
+        if self._server_configs:
+            return await call_mcp_tool(self._server_configs[0], tool_name, arguments)
+        return json.dumps({"error": "No MCP server configured for tool: " + tool_name})
+
+    async def health_check(self) -> bool:
+        # Check if any MCP server is connected
+        try:
+            tools = await self.list_tools()
+            return len(tools) > 0
+        except Exception:
+            return False
+
+
+# ── A2A 适配器 ───────────────────────────────────────────────────────────────
+
+class A2aProtocolAdapter(ACAProtocol):
+    """A2A (Agent-to-Agent) 协议适配器。
+
+    通过标准 JSON-RPC 与远端 Agent 通信。端点需实现:
+    - POST /a2a/tools  → 返回工具列表
+    - POST /a2a/call   → 调用工具
+
+    A2A 规范参考: agentcommunicationprotocol.dev
+    """
+
+    def __init__(self, endpoint: str, api_key: str = None, timeout: float = 30):
+        self.name = "a2a"
+        self.version = "0.9"
+        self.capabilities = ["tools"]
+        self.endpoint = endpoint.rstrip("/")
+        self.api_key = api_key
+        self.timeout = timeout
+
+    async def list_tools(self) -> list[dict]:
+        async with httpx.AsyncClient(timeout=self.timeout, trust_env=False) as client:
+            headers = self._headers()
+            resp = await client.post(f"{self.endpoint}/a2a/tools", headers=headers, json={})
+        if resp.status_code != 200:
+            logger.warning("A2A list_tools %s HTTP %s: %s", self.endpoint, resp.status_code, resp.text[:200])
+            return []
+        data = resp.json()
+        return data.get("tools") or data.get("data", {}).get("tools") or []
+
+    async def call_tool(self, tool_name: str, arguments: dict) -> str:
+        async with httpx.AsyncClient(timeout=self.timeout, trust_env=False) as client:
+            headers = self._headers()
+            body = {
+                "jsonrpc": "2.0",
+                "method": "tools/call",
+                "params": {"name": tool_name, "arguments": arguments},
+                "id": 1,
+            }
+            resp = await client.post(f"{self.endpoint}/a2a/call", headers=headers, json=body)
+        if resp.status_code != 200:
+            return json.dumps({"error": f"A2A HTTP {resp.status_code}: {resp.text[:300]}"})
+        data = resp.json()
+        if "error" in data:
+            return json.dumps({"error": data["error"]})
+        result = data.get("result")
+        if isinstance(result, dict):
+            return result.get("content", json.dumps(result))
+        return str(result) if result is not None else ""
+
+    async def health_check(self) -> bool:
+        try:
+            async with httpx.AsyncClient(timeout=10, trust_env=False) as client:
+                headers = self._headers()
+                resp = await client.get(f"{self.endpoint}/a2a/health", headers=headers)
+            return resp.status_code == 200
+        except Exception:
+            return False
+
+    def _headers(self) -> dict:
+        h = {"Content-Type": "application/json"}
+        if self.api_key:
+            h["Authorization"] = f"Bearer {self.api_key}"
+        return h
+
+
+# ── 启动时初始化 ─────────────────────────────────────────────────────────────
+
+def init_aca_registry():
+    """在应用启动时调用，注册所有协议适配器到全局 registry。"""
+    from app.database import SessionLocal
+    from app.models import MCPServer
+
+    # 注册 MCP 适配器（从 DB 读取所有 enabled MCP server 配置）
+    try:
+        db = SessionLocal()
+        servers = db.query(MCPServer).filter(MCPServer.is_enabled == True).all()
+        server_configs = []
+        for s in servers:
+            server_configs.append({
+                "id": s.id,
+                "name": s.name,
+                "transport_type": s.transport_type,
+                "command": s.command,
+                "args": json.loads(s.args) if isinstance(s.args, str) else s.args or [],
+                "env_vars": json.loads(s.env_vars) if isinstance(s.env_vars, str) else s.env_vars or {},
+                "url": s.url,
+                "headers": json.loads(s.headers) if isinstance(s.headers, str) else s.headers or {},
+            })
+        db.close()
+        if server_configs:
+            registry.register(McpProtocolAdapter(server_configs))
+            logger.info("ACA: MCP adapter registered with %d servers", len(server_configs))
+    except Exception as e:
+        logger.warning("ACA: failed to init MCP adapter: %s", e)
+
+    # 注册 A2A 适配器（从配置或 DB 读取 A2A endpoint）
+    # A2A endpoint 可以从 site_settings 读取，暂无则留空（以后台配置驱动）
+    a2a_endpoint = _get_a2a_config()
+    if a2a_endpoint:
+        registry.register(A2aProtocolAdapter(
+            endpoint=a2a_endpoint.get("endpoint", ""),
+            api_key=a2a_endpoint.get("api_key"),
+        ))
+        logger.info("ACA: A2A adapter registered at %s", a2a_endpoint.get("endpoint"))
+
+
+def _get_a2a_config() -> dict | None:
+    """从 site_settings 读取 A2A 全局配置。"""
+    try:
+        from app.database import SessionLocal
+        from app.models import SiteSetting
+        db = SessionLocal()
+        setting = db.query(SiteSetting).filter(SiteSetting.key == "aca_a2a_endpoint").first()
+        db.close()
+        if setting and setting.value:
+            val = json.loads(setting.value) if isinstance(setting.value, str) else setting.value
+            if isinstance(val, dict) and val.get("endpoint"):
+                return val
+    except Exception:
+        pass
+    return None

backend/python/app/core/yaml_workflow.py

@@ -0,0 +1,302 @@
+"""Phase 4.2: QihuFlow 工作流可移植描述语言 (YAML v1.0)
+
+提供 YAML 工作流文本 ↔ orchestrator JSON 的双向转换与校验。
+
+YAML 规范 v1.0 示例:
+
+    version: "1.0"
+    name: "Research & Code"
+    description: "Research then generate code"
+    output_node: merge
+
+    nodes:
+      - id: research
+        type: llm
+        on_error: continue
+        config:
+          model: auto
+          system_prompt: "You are a researcher."
+      - id: coder
+        type: local
+        agent_id: "uuid-here"
+        config:
+          skill_ids: ["search", "code-run"]
+      - id: merge
+        type: reduce
+        config:
+          strategy: concat
+
+    edges:
+      - from: research
+        to: coder
+      - from: coder
+        to: merge
+
+config 字段映射（YAML config ↔ orchestrator JSON）:
+
+    YAML config key                 JSON field
+    ─────────────────────────────────────────────
+    model                           model_id
+    system_prompt                   system_prompt
+    skill_ids                       skill_ids
+    knowledge_base_ids              knowledge_base_ids
+    endpoint                        agent_endpoint
+    api_key                         api_key
+    strategy                        strategy (reduce)
+    max_tokens                      max_tokens
+"""
+from __future__ import annotations
+
+import re
+import json
+from typing import Any
+
+SPEC_VERSION = "1.0"
+
+# YAML config → JSON field mapping
+_CONFIG_MAP = {
+    "model":             "model_id",
+    "system_prompt":     "system_prompt",
+    "skill_ids":         "skill_ids",
+    "knowledge_base_ids": "knowledge_base_ids",
+    "endpoint":          "agent_endpoint",
+    "api_key":           "api_key",
+    "strategy":          "strategy",
+    "max_tokens":        "max_tokens",
+    "description":       "description",
+}
+
+# reverse map: JSON → YAML config keys to nest
+_REVERSE_CONFIG = {
+    "model_id", "system_prompt", "skill_ids", "knowledge_base_ids",
+    "agent_endpoint", "api_key", "strategy", "max_tokens",
+}
+
+VALID_NODE_TYPES = {"llm", "local", "external", "reduce"}
+VALID_ON_ERROR = {"halt", "continue"}
+VALID_REDUCE_STRATEGY = {"concat", "first", "last"}
+
+
+# ── validation ──────────────────────────────────────────────────────────────
+
+def validate_yaml(data: Any) -> dict:
+    """校验 YAML 工作流结构，返回 {ok, errors:[...], warnings:[...]}。
+
+    data 应为解析后的 YAML 字典（由调用方用 yaml.safe_load 解析）。
+    """
+    errors: list = []
+    warnings: list = []
+
+    if not isinstance(data, dict):
+        return {"ok": False, "errors": ["YAML 顶层必须为对象"], "warnings": []}
+
+    version = data.get("version")
+    if not version or version != SPEC_VERSION:
+        warnings.append(f"version 应为 \"{SPEC_VERSION}\"，当前为 \"{version}\"")
+
+    nodes = data.get("nodes")
+    if not isinstance(nodes, list) or not nodes:
+        errors.append("nodes 必须为非空数组")
+        return {"ok": False, "errors": errors, "warnings": warnings}
+
+    node_ids: set = set()
+    for i, node in enumerate(nodes):
+        if not isinstance(node, dict):
+            errors.append(f"节点 #{i} 必须为对象")
+            continue
+        nid = node.get("id")
+        if not nid or not isinstance(nid, str):
+            errors.append(f"节点 #{i} 缺少字符串 id")
+            continue
+        if nid in node_ids:
+            errors.append(f"节点 id 重复: {nid}")
+        node_ids.add(nid)
+
+        ntype = node.get("type")
+        if ntype not in VALID_NODE_TYPES:
+            errors.append(f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(sorted(VALID_NODE_TYPES))})")
+
+        on_error = node.get("on_error", "continue")
+        if on_error and on_error not in VALID_ON_ERROR:
+            errors.append(f"节点 {nid} on_error 非法: {on_error}")
+
+        # type-specific checks
+        if ntype in ("local", "external"):
+            if not node.get("agent_id"):
+                errors.append(f"节点 {nid} 类型 {ntype} 必须指定 agent_id")
+
+        # config validation
+        cfg = node.get("config", {})
+        if isinstance(cfg, dict):
+            strategy = cfg.get("strategy")
+            if strategy and strategy not in VALID_REDUCE_STRATEGY:
+                errors.append(f"节点 {nid} config.strategy 非法: {strategy}")
+
+    edges = data.get("edges", [])
+    if not isinstance(edges, list):
+        errors.append("edges 必须为数组")
+    else:
+        for i, edge in enumerate(edges):
+            if not isinstance(edge, dict):
+                errors.append(f"边 #{i} 必须为对象")
+                continue
+            f = edge.get("from")
+            t = edge.get("to")
+            if not f or not t:
+                errors.append(f"边 #{i} 缺少 from/to")
+            if f and f not in node_ids:
+                errors.append(f"边 #{i} from={f} 指向不存在的节点")
+            if t and t not in node_ids:
+                errors.append(f"边 #{i} to={t} 指向不存在的节点")
+            if f and t and f == t:
+                errors.append(f"边 #{i} 存在自环: {f}")
+
+    output_node = data.get("output_node")
+    if output_node and output_node not in node_ids:
+        warnings.append(f"output_node={output_node} 不存在于节点列表中，将自动检测出度为零的节点")
+
+    # cycle detection (topological sort)
+    if node_ids and not errors:
+        indeg = {nid: 0 for nid in node_ids}
+        adj = {nid: [] for nid in node_ids}
+        for e in edges:
+            if isinstance(e, dict):
+                f, t = e.get("from"), e.get("to")
+                if f in adj and t in adj:
+                    adj[f].append(t)
+                    indeg[t] += 1
+        queue = [nid for nid in node_ids if indeg[nid] == 0]
+        processed = 0
+        while queue:
+            n = queue.pop(0)
+            processed += 1
+            for nxt in adj[n]:
+                indeg[nxt] -= 1
+                if indeg[nxt] == 0:
+                    queue.append(nxt)
+        if processed != len(node_ids):
+            errors.append("检测到环路，无法进行拓扑排序")
+
+    return {
+        "ok": len(errors) == 0,
+        "errors": errors,
+        "warnings": warnings,
+    }
+
+
+# ── YAML → orchestrator JSON ─────────────────────────────────────────────────
+
+def yaml_to_orchestration(data: dict) -> dict:
+    """将 YAML 工作流对象转换为 orchestrator.run_graph() 接受的 JSON 格式。"""
+    result: dict = {"nodes": [], "edges": []}
+
+    result["name"] = data.get("name")
+    result["description"] = data.get("description")
+    if data.get("output_node"):
+        result["output_node"] = data["output_node"]
+
+    for raw_node in data.get("nodes", []):
+        if not isinstance(raw_node, dict):
+            continue
+        node: dict = {
+            "id": raw_node.get("id", ""),
+            "type": raw_node.get("type", "llm"),
+            "description": raw_node.get("description", ""),
+            "on_error": raw_node.get("on_error", "continue"),
+        }
+
+        # agent_id for local/external
+        if raw_node.get("agent_id"):
+            node["agent_id"] = raw_node["agent_id"]
+
+        # flatten config into top-level keys
+        config = raw_node.get("config", {})
+        if isinstance(config, dict):
+            for yaml_key, json_key in _CONFIG_MAP.items():
+                if yaml_key in config and config[yaml_key] is not None:
+                    node[json_key] = config[yaml_key]
+
+        # also carry over any extra keys from raw_node (preserve flexibility)
+        for k, v in raw_node.items():
+            if k not in ("id", "type", "config", "description", "on_error", "agent_id"):
+                if k in ("skill_ids", "knowledge_base_ids", "model_id", "system_prompt",
+                          "agent_endpoint", "api_key", "strategy", "max_tokens"):
+                    node[k] = v
+
+        result["nodes"].append(node)
+
+    for edge in data.get("edges", []):
+        if isinstance(edge, dict):
+            result["edges"].append({
+                "from": edge.get("from", ""),
+                "to": edge.get("to", ""),
+            })
+
+    return result
+
+
+# ── orchestrator JSON → YAML ─────────────────────────────────────────────────
+
+def orchestration_to_yaml(data: dict) -> dict:
+    """将 orchestrator JSON 转换为 YAML 规范对象（可再 yaml.dump 输出）。"""
+    result: dict = {"version": SPEC_VERSION}
+
+    if data.get("name"):
+        result["name"] = data["name"]
+    if data.get("description"):
+        result["description"] = data["description"]
+    if data.get("output_node"):
+        result["output_node"] = data["output_node"]
+
+    result["nodes"] = []
+    for json_node in data.get("nodes", []):
+        if not isinstance(json_node, dict):
+            continue
+        node: dict = {
+            "id": json_node.get("id", ""),
+            "type": json_node.get("type", "llm"),
+        }
+        if json_node.get("description"):
+            node["description"] = json_node["description"]
+        if json_node.get("on_error"):
+            node["on_error"] = json_node["on_error"]
+        if json_node.get("agent_id"):
+            node["agent_id"] = json_node["agent_id"]
+
+        # nest known config keys into config sub-object
+        config: dict = {}
+        for yaml_key, json_key in _CONFIG_MAP.items():
+            if json_key in json_node and json_node[json_key] is not None:
+                config[yaml_key] = json_node[json_key]
+
+        # also preserve any extra keys not in the map
+        for k, v in json_node.items():
+            if k not in ("id", "type", "description", "on_error", "agent_id") \
+               and k not in _REVERSE_CONFIG:
+                if v is not None:
+                    config[k] = v
+
+        if config:
+            node["config"] = config
+
+        result["nodes"].append(node)
+
+    result["edges"] = []
+    for edge in data.get("edges", []):
+        if isinstance(edge, dict):
+            result["edges"].append({
+                "from": edge.get("from", ""),
+                "to": edge.get("to", ""),
+            })
+
+    return result
+
+
+# ── YAML 模板参数替换 ────────────────────────────────────────────────────────
+
+def apply_template_params(yaml_text: str, params: dict) -> str:
+    """替换 YAML 模板中的 ${param} 占位符。"""
+    def _replacer(m: re.Match) -> str:
+        key = m.group(1)
+        return str(params.get(key, m.group(0)))
+    return re.sub(r"\$\{(\w+)\}", _replacer, yaml_text)

backend/python/app/workflow_templates/code-review.yaml

@@ -0,0 +1,35 @@
+version: "1.0"
+name: "Code Review Pipeline"
+description: "Generate code → review → merge. A standard code-review workflow."
+category: development
+
+nodes:
+  - id: code_gen
+    type: llm
+    description: "Generate code based on requirements"
+    on_error: halt
+    config:
+      model: auto
+      system_prompt: "You are a senior software engineer. Write clean, production-ready code based on the user's requirements. Include comments and error handling."
+
+  - id: review
+    type: llm
+    description: "Review generated code for bugs and style"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a code reviewer. Analyze the provided code for bugs, style issues, performance problems, and security vulnerabilities. Provide specific, actionable feedback."
+
+  - id: merge
+    type: reduce
+    description: "Combine code with review feedback"
+    config:
+      strategy: concat
+
+edges:
+  - from: code_gen
+    to: review
+  - from: code_gen
+    to: merge
+  - from: review
+    to: merge

backend/python/app/workflow_templates/multi-agent-debate.yaml

@@ -0,0 +1,35 @@
+version: "1.0"
+name: "Multi-Agent Debate"
+description: "Two agents debate a topic from opposing perspectives, then a judge agent synthesizes the conclusion."
+category: creativity
+
+nodes:
+  - id: proponent
+    type: llm
+    description: "Argue in favor of the proposition"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a skilled debater. Argue strongly IN FAVOR of the given proposition. Use logic, evidence, and persuasive reasoning. Be passionate but fair."
+
+  - id: opponent
+    type: llm
+    description: "Argue against the proposition"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a skilled debater. Argue strongly AGAINST the given proposition. Use logic, evidence, and persuasive reasoning to find flaws in the proponent's case."
+
+  - id: judge
+    type: llm
+    description: "Evaluate both sides and reach a conclusion"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are an impartial judge. Evaluate both arguments fairly. Identify the strongest points from each side. Provide a balanced conclusion and recommendation."
+
+edges:
+  - from: proponent
+    to: judge
+  - from: opponent
+    to: judge

backend/python/app/workflow_templates/rag-search.yaml

@@ -0,0 +1,25 @@
+version: "1.0"
+name: "RAG Search Pipeline"
+description: "Search knowledge base → augment with context → generate answer. Standard RAG pipeline."
+category: productivity
+
+nodes:
+  - id: search
+    type: llm
+    description: "Generate search queries and retrieve relevant context"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a search specialist. For the given question, identify 2-3 key search points. Then search the provided knowledge base for relevant information. Return the most relevant passages with source references."
+
+  - id: answer
+    type: llm
+    description: "Generate final answer with citations"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a knowledge assistant. Using the search results provided, generate a comprehensive answer. Always cite your sources. If information is insufficient, clearly state what is missing."
+
+edges:
+  - from: search
+    to: answer

backend/python/app/workflow_templates/research-assistant.yaml

@@ -0,0 +1,25 @@
+version: "1.0"
+name: "Research & Answer"
+description: "Research topic → generate answer. Uses two LLM nodes for research and synthesis."
+category: productivity
+
+nodes:
+  - id: research
+    type: llm
+    description: "Search and gather information on the topic"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a thorough researcher. Search for key facts, data points, and relevant context about the given topic. Provide a comprehensive summary with citations."
+
+  - id: answer
+    type: llm
+    description: "Synthesize research into final answer"
+    on_error: continue
+    config:
+      model: auto
+      system_prompt: "You are a professional writer. Based on the research summary provided, craft a clear, well-structured answer for the user. Include key findings and actionable insights."
+
+edges:
+  - from: research
+    to: answer

backend/python/tests/test_42_43.py

@@ -0,0 +1,92 @@
+"""Phase 4.2+4.3: YAML workflow + ACA protocol registry tests."""
+import pytest
+from app.core.yaml_workflow import validate_yaml, yaml_to_orchestration, orchestration_to_yaml
+
+
+# ── YAML workflow tests ──────────────────────────────────────────────────────
+
+VALID_YAML = """
+version: "1.0"
+name: "Test"
+nodes:
+  - id: a
+    type: llm
+    config:
+      model: auto
+      system_prompt: "You are a helper."
+  - id: b
+    type: llm
+  - id: c
+    type: reduce
+    config:
+      strategy: concat
+edges:
+  - from: a
+    to: b
+  - from: b
+    to: c
+  - from: a
+    to: c
+"""
+
+
+def test_validate_valid():
+    import yaml
+    data = yaml.safe_load(VALID_YAML)
+    r = validate_yaml(data)
+    assert r["ok"], r["errors"]
+    assert r["errors"] == []
+
+
+def test_validate_invalid_type():
+    import yaml
+    data = yaml.safe_load("version: '1.0'\nnodes:\n  - id: x\n    type: bogus")
+    r = validate_yaml(data)
+    assert not r["ok"]
+    assert any("bogus" in e for e in r["errors"])
+
+
+def test_validate_cycle():
+    import yaml
+    data = yaml.safe_load("version: '1.0'\nnodes:\n  - id: a\n    type: llm\n  - id: b\n    type: llm\nedges:\n  - from: a\n    to: b\n  - from: b\n    to: a")
+    r = validate_yaml(data)
+    assert not r["ok"]
+    assert any("环" in e for e in r["errors"])
+
+
+def test_yaml_to_json_roundtrip():
+    import yaml
+    data = yaml.safe_load(VALID_YAML)
+    orch = yaml_to_orchestration(data)
+    assert len(orch["nodes"]) == 3
+    assert len(orch["edges"]) == 3
+    # config flattening
+    node_a = next(n for n in orch["nodes"] if n["id"] == "a")
+    assert node_a.get("model_id") == "auto"
+    assert node_a.get("system_prompt") == "You are a helper."
+
+
+def test_json_to_yaml_to_json():
+    """Roundtrip: JSON → YAML obj → JSON should preserve key fields."""
+    orch_json = {
+        "nodes": [
+            {"id": "x", "type": "llm", "model_id": "auto"},
+            {"id": "y", "type": "reduce", "strategy": "concat"},
+        ],
+        "edges": [{"from": "x", "to": "y"}],
+    }
+    yaml_obj = orchestration_to_yaml(orch_json)
+    assert yaml_obj["version"] == "1.0"
+    # config nesting
+    node_x = next(n for n in yaml_obj["nodes"] if n["id"] == "x")
+    assert node_x.get("config", {}).get("model") == "auto"
+    node_y = next(n for n in yaml_obj["nodes"] if n["id"] == "y")
+    assert node_y.get("config", {}).get("strategy") == "concat"
+
+
+def test_validate_missing_agent_id():
+    import yaml
+    data = yaml.safe_load("version: '1.0'\nnodes:\n  - id: x\n    type: local\n  - id: y\n    type: llm\nedges:\n  - from: x\n    to: y")
+    r = validate_yaml(data)
+    assert not r["ok"]
+    assert any("agent_id" in e for e in r["errors"])