workflow适配器修改&并行需求适配-agent-runtime-AtomGit

Mapplications/a2a_service/.env.example+3-3

	@@ -22,7 +22,7 @@ BOOTSTRAP_LOCK_TTL_SEC=180
22	BOOTSTRAP_WAIT_TIMEOUT_SEC=300	22	BOOTSTRAP_WAIT_TIMEOUT_SEC=300
23	BOOTSTRAP_POLL_INTERVAL_SEC=1.0	23	BOOTSTRAP_POLL_INTERVAL_SEC=1.0
24		24
25	-# ── Rate Limit（入口限流）─────────────────────────────────────────────────	25	+# ── Rate Limit（入口限流），需要根据模型实际并发能力配置────────────────────────────────────────
26	RATE_LIMIT_MAX_REQUESTS=5	26	RATE_LIMIT_MAX_REQUESTS=5
27	RATE_LIMIT_WINDOW_SECONDS=120	27	RATE_LIMIT_WINDOW_SECONDS=120
28	GLOBAL_RATE_LIMIT_MAX_REQUESTS=100	28	GLOBAL_RATE_LIMIT_MAX_REQUESTS=100
	@@ -30,8 +30,8 @@ GLOBAL_RATE_LIMIT_WINDOW_SECONDS=30
30		30
31	# ── VersatileAdapter（内部 A2A 服务地址）─────────────────────────────────────	31	# ── VersatileAdapter（内部 A2A 服务地址）─────────────────────────────────────
32	VERSATILE_ADAPTER_URL=http://localhost:8091	32	VERSATILE_ADAPTER_URL=http://localhost:8091
33	-# a2a_service → VersatileAdapter 的 HTTP 超时（秒），与versatile_adapter中VERSATILE_TIMEOUT 默认值对齐	33	+# a2a_service → VersatileAdapter 的 HTTP 超时（秒），与versatile_adapter中VERSATILE_TIMEOUT 默认值对齐，需要<=工小智前端超时时延57s
34	-VERSATILE_ADAPTER_TIMEOUT=600	34	+VERSATILE_ADAPTER_TIMEOUT=57
35		35
36	# ── 并行子 Agent / 多工作流（对齐 TECH §5.1）─────────────────────────────────	36	# ── 并行子 Agent / 多工作流（对齐 TECH §5.1）─────────────────────────────────
37	# 注：子 Agent url 不再由框架配置（P-006）——由 Agent 自管、随派发请求下传，框架不读取。	37	# 注：子 Agent url 不再由框架配置（P-006）——由 Agent 自管、随派发请求下传，框架不读取。

Mapplications/a2a_service/config.py+2-2

	@@ -48,8 +48,8 @@ class Settings(BaseSettings):
48		48
49	# ── VersatileAdapter（内部 A2A 服务地址）────────────────────────────────	49	# ── VersatileAdapter（内部 A2A 服务地址）────────────────────────────────
50	versatile_adapter_url: Optional[str] = None	50	versatile_adapter_url: Optional[str] = None
51	- # a2a_service → VersatileAdapter 的 HTTP 超时（秒），与versatile_adapter中VERSATILE_TIMEOUT 默认值对齐	51	+ # a2a_service → VersatileAdapter 的 HTTP 超时（秒），与versatile_adapter中VERSATILE_TIMEOUT 默认值对齐，需要<=工小智前端超时时延57s
52	- versatile_adapter_timeout: int = 600	52	+ versatile_adapter_timeout: int = 57
53	# VA 流中携带工作流最终结果的 QA 节点名称（node_type=="QA" 且 node_name==此值）	53	# VA 流中携带工作流最终结果的 QA 节点名称（node_type=="QA" 且 node_name==此值）
54	va_workflow_result_node: Optional[str] = None	54	va_workflow_result_node: Optional[str] = None
55		55

Mapplications/a2a_service/orchestrator/executor.py+108-36

	@@ -25,6 +25,7 @@ from typing import Any, AsyncGenerator, Optional
25		25
26	import httpx	26	import httpx
27	from a2a.client import Client, ClientFactory	27	from a2a.client import Client, ClientFactory
		28	+from a2a.client.errors import A2AClientError
28	from a2a.server.agent_execution import AgentExecutor, RequestContext	29	from a2a.server.agent_execution import AgentExecutor, RequestContext
29	from a2a.server.context import ServerCallContext	30	from a2a.server.context import ServerCallContext
30	from a2a.server.events import EventQueue	31	from a2a.server.events import EventQueue
	@@ -197,6 +198,7 @@ class _VaRequestPayload:
197	conv_id: str = ""	198	conv_id: str = ""
198	trace_id: str = ""	199	trace_id: str = ""
199	agent_id: str = ""	200	agent_id: str = ""
		201	+ target: Optional[dict] = None
200		202
201		203
202	@dataclass(frozen=True)	204	@dataclass(frozen=True)
	@@ -644,8 +646,11 @@ class Executor(AgentExecutor):
644	text_part.text = payload.query	646	text_part.text = payload.query
645		647
646	data_struct = Struct()	648	data_struct = Struct()
		649	+ target = dict(payload.target or {})
		650	+ target.setdefault("conversation_id", payload.conv_id)
647	data_struct.update(	651	data_struct.update(
648	{	652	{
		653	+ "target": target,
649	"headers": payload.headers,	654	"headers": payload.headers,
650	"body": payload.body,	655	"body": payload.body,
651	"params": payload.params or {},	656	"params": payload.params or {},
	@@ -674,30 +679,9 @@ class Executor(AgentExecutor):
674	return stream_resp.status_update	679	return stream_resp.status_update
675	return None	680	return None
676		681
677	- def _extract_upstream_error(self, event: TaskArtifactUpdateEvent) -> Optional[dict]:
678	- """识别 VA 上游错误终态帧。
679	-
680	- AgentEngine ``versatile_proxy.py:336`` 把 ``event=='exception'`` 视为
681	- workflow_complete 终态。这里把 ``event in ("error", "exception")`` 都识别
682	- 为终态错误，返回 data 部分（含 code/message）；非错误帧返回 None。
683	-
684	- 识别后 Executor 应把 Task 标 FAILED 并清空 ``va_task_id``，避免下次请求被
685	- 当成 cascade 续轮、用 stale task_id 调 VA 锁死 conversation。
686	- """
687	- for part in event.artifact.parts:
688	- if part.WhichOneof("content") != "data":
689	- continue
690	- frame = MessageToDict(part.data)
691	- if not isinstance(frame, dict):
692	- continue
693	- if frame.get("event") in ("error", "exception"):
694	- inner = frame.get("data")
695	- return inner if isinstance(inner, dict) else {}
696	- return None
697	-
698	@staticmethod	682	@staticmethod
699	def _format_upstream_error(err: dict) -> str:	683	def _format_upstream_error(err: dict) -> str:
700	- """把上游 error/exception 的 data 拼成可展示的错误描述字符串。"""	684	+ """把 VA FAILED 状态携带的 upstream_error 拼成可展示的错误描述字符串。"""
701	code = err.get("code")	685	code = err.get("code")
702	message = err.get("message") or err.get("msg") or ""	686	message = err.get("message") or err.get("msg") or ""
703	if code:	687	if code:
	@@ -760,7 +744,7 @@ class Executor(AgentExecutor):
760	VA 侧在 updater.complete(message) 时通过 Part 的 metadata {"vatype": "workflow_result"} 标识，	744	VA 侧在 updater.complete(message) 时通过 Part 的 metadata {"vatype": "workflow_result"} 标识，
761	DataPart 为纯文本 string_value。	745	DataPart 为纯文本 string_value。
762	"""	746	"""
763	- if not event.status or not event.status.message:	747	+ if not event.status or not event.status.HasField("message"):
764	return None	748	return None
765	for part in event.status.message.parts:	749	for part in event.status.message.parts:
766	if not part.HasField("text") or not part.HasField("metadata"):	750	if not part.HasField("text") or not part.HasField("metadata"):
	@@ -770,6 +754,29 @@ class Executor(AgentExecutor):
770	return part.text	754	return part.text
771	return None	755	return None
772		756
		757	+ def _extract_failed_error(self, event: TaskStatusUpdateEvent) -> Optional[dict]:
		758	+ """从 VA Sidecar 的 FAILED 状态事件 message 中提取 upstream_error。"""
		759	+ if not event.status or not event.status.HasField("message"):
		760	+ return None
		761	+ for part in event.status.message.parts:
		762	+ if not part.HasField("text") or not part.HasField("metadata"):
		763	+ continue
		764	+ meta = MessageToDict(part.metadata)
		765	+ if meta.get("vatype") != "upstream_error":
		766	+ continue
		767	+ raw_text = part.text or ""
		768	+ if not raw_text:
		769	+ continue
		770	+ try:
		771	+ parsed = json.loads(raw_text)
		772	+ except (ValueError, TypeError):
		773	+ return {"message": raw_text}
		774	+ if isinstance(parsed, dict):
		775	+ inner = parsed.get("data") if isinstance(parsed.get("data"), dict) else parsed
		776	+ return inner
		777	+ return {"message": raw_text}
		778	+ return None
		779	+
773	@staticmethod	780	@staticmethod
774	def _extract_data_proxy_frames(event: TaskArtifactUpdateEvent) -> list[dict]:	781	def _extract_data_proxy_frames(event: TaskArtifactUpdateEvent) -> list[dict]:
775	"""从 VA artifact 取出 ``vatype=data_proxy`` 的 text Part，json.loads 还原为结构化帧列表。	782	"""从 VA artifact 取出 ``vatype=data_proxy`` 的 text Part，json.loads 还原为结构化帧列表。
	@@ -889,6 +896,7 @@ class Executor(AgentExecutor):
889	conv_id=conv_id,	896	conv_id=conv_id,
890	trace_id=trace_id,	897	trace_id=trace_id,
891	agent_id=agent_id,	898	agent_id=agent_id,
		899	+ target={"type": "workflow", "intent": effective_intent},
892	)	900	)
893	)	901	)
894		902
	@@ -962,7 +970,7 @@ class Executor(AgentExecutor):
962	logger.debug("[Executor] VA TaskStatusUpdateEvent(COMPLETED)")	970	logger.debug("[Executor] VA TaskStatusUpdateEvent(COMPLETED)")
963	elif event.status.state == TASK_STATE_FAILED:	971	elif event.status.state == TASK_STATE_FAILED:
964	if upstream_error is None:	972	if upstream_error is None:
965	- upstream_error = {"message": "VA 任务异常终止"}	973	+ upstream_error = self._extract_failed_error(event) or {"message": "VA 任务异常终止"}
966	logger.debug("[Executor] VA TaskStatusUpdateEvent(FAILED)")	974	logger.debug("[Executor] VA TaskStatusUpdateEvent(FAILED)")
967		975
968	except Exception as e:	976	except Exception as e:
	@@ -1034,6 +1042,10 @@ class Executor(AgentExecutor):
1034	# 当前轮输入能透传给下游工作流，而不是回退到首轮缓存 body。	1042	# 当前轮输入能透传给下游工作流，而不是回退到首轮缓存 body。
1035	body = dict(original_body)	1043	body = dict(original_body)
1036	body["stream"] = True	1044	body["stream"] = True
		1045	+ input_section = body.get("input") if isinstance(body.get("input"), dict) else {}
		1046	+ custom_data = body.get("custom_data") if isinstance(body.get("custom_data"), dict) else {}
		1047	+ custom_inputs = custom_data.get("inputs") if isinstance(custom_data.get("inputs"), dict) else {}
		1048	+ routed_intent = input_section.get("intent") or custom_inputs.get("intent") or ""
1037		1049
1038	# 在 a2a 续轮调用侧记录 Versatile 前后 Tag 日志	1050	# 在 a2a 续轮调用侧记录 Versatile 前后 Tag 日志
1039	versatile_call_id = str(uuid.uuid4())	1051	versatile_call_id = str(uuid.uuid4())
	@@ -1052,6 +1064,7 @@ class Executor(AgentExecutor):
1052	conv_id=conv_id,	1064	conv_id=conv_id,
1053	trace_id=trace_id,	1065	trace_id=trace_id,
1054	agent_id=agent_id,	1066	agent_id=agent_id,
		1067	+ target={"type": "workflow", "intent": routed_intent} if routed_intent else None,
1055	)	1068	)
1056	)	1069	)
1057		1070
	@@ -1105,7 +1118,7 @@ class Executor(AgentExecutor):
1105	logger.debug("[Executor] VA 续轮 TaskStatusUpdateEvent(COMPLETED)")	1118	logger.debug("[Executor] VA 续轮 TaskStatusUpdateEvent(COMPLETED)")
1106	elif event.status.state == TASK_STATE_FAILED:	1119	elif event.status.state == TASK_STATE_FAILED:
1107	if upstream_error is None:	1120	if upstream_error is None:
1108	- upstream_error = {"message": "VA 任务异常终止"}	1121	+ upstream_error = self._extract_failed_error(event) or {"message": "VA 任务异常终止"}
1109	logger.debug("[Executor] VA 续轮 TaskStatusUpdateEvent(FAILED)")	1122	logger.debug("[Executor] VA 续轮 TaskStatusUpdateEvent(FAILED)")
1110		1123
1111	except Exception as e:	1124	except Exception as e:
	@@ -1357,6 +1370,8 @@ class Executor(AgentExecutor):
1357		1370
1358	content = ""	1371	content = ""
1359	child_task_id = ""	1372	child_task_id = ""
		1373	+ terminal_status: Optional[str] = None # 子 Agent 终态 FAILED/CANCELED（问题 1）
		1374	+ terminal_error = ""
1360	try:	1375	try:
1361	async for frame in self._drive_sub_agent(	1376	async for frame in self._drive_sub_agent(
1362	spec, sub_conv_id, child_path, turn_ctx, cancel_event	1377	spec, sub_conv_id, child_path, turn_ctx, cancel_event
	@@ -1372,6 +1387,10 @@ class Executor(AgentExecutor):
1372	if ftype == "__completed__":	1387	if ftype == "__completed__":
1373	content = frame.get("content", "")	1388	content = frame.get("content", "")
1374	continue	1389	continue
		1390	+ if ftype == "__terminal__":
		1391	+ terminal_status = frame.get("status", "failed")
		1392	+ terminal_error = frame.get("error", "")
		1393	+ continue
1375	# report 帧：已盖章（更深层 sub_task）透传 / 否则盖章为本节点 agent 帧	1394	# report 帧：已盖章（更深层 sub_task）透传 / 否则盖章为本节点 agent 帧
1376	if frame.get("type") == "sub_task":	1395	if frame.get("type") == "sub_task":
1377	await self._emit_sub_task(	1396	await self._emit_sub_task(
	@@ -1393,7 +1412,19 @@ class Executor(AgentExecutor):
1393	error=str(exc), child_task_id=child_task_id,	1412	error=str(exc), child_task_id=child_task_id,
1394	)	1413	)
1395		1414
1396	- if cancel_event.is_set():	1415	+ if terminal_status == "failed":
		1416	+ err = terminal_error or "子 Agent 终态异常"
		1417	+ await self._emit_sub_task(
		1418	+ turn_ctx, child_path, "agent",
		1419	+ {"event": "node_end", "status": "failed", "error": err},
		1420	+ )
		1421	+ logger.warning(f"[Executor] 子 Agent 终态 FAILED：entity={spec.entity_id}, error={err}")
		1422	+ return SubAgentResult(
		1423	+ entity_id=spec.entity_id, status="failed",
		1424	+ error=err, child_task_id=child_task_id,
		1425	+ )
		1426	+
		1427	+ if cancel_event.is_set() or terminal_status == "cancelled":
1397	await self._emit_sub_task(	1428	await self._emit_sub_task(
1398	turn_ctx, child_path, "agent",	1429	turn_ctx, child_path, "agent",
1399	{"event": "node_end", "status": "cancelled", "reason": "用户取消"},	1430	{"event": "node_end", "status": "cancelled", "reason": "用户取消"},
	@@ -1517,10 +1548,21 @@ class Executor(AgentExecutor):
1517	if frame is not None:	1548	if frame is not None:
1518	yield frame	1549	yield frame
1519	elif kind == "status_update":	1550	elif kind == "status_update":
1520	- if _is_completed_status(stream_resp.status_update):	1551	+ su = stream_resp.status_update
		1552	+ if _is_completed_status(su):
1521	yield {	1553	yield {
1522	"type": "__completed__",	1554	"type": "__completed__",
1523	- "content": extract_content(stream_resp.status_update),	1555	+ "content": extract_content(su),
		1556	+ }
		1557	+ return
		1558	+ state = su.status.state if (su and su.status) else None
		1559	+ if state in (TASK_STATE_FAILED, TASK_STATE_CANCELED):
		1560	+ # 子 Agent 终态异常：不能落到流末被当成 done 静默上报（问题 1）。
		1561	+ # error 取 status.message 文本（基础设施级失败时承载错因）。
		1562	+ yield {
		1563	+ "type": "__terminal__",
		1564	+ "status": "failed" if state == TASK_STATE_FAILED else "cancelled",
		1565	+ "error": extract_content(su),
1524	}	1566	}
1525	return	1567	return
1526	return # 流正常结束	1568	return # 流正常结束
	@@ -1533,7 +1575,14 @@ class Executor(AgentExecutor):
1533	yield {"type": "__completed__", **final}	1575	yield {"type": "__completed__", **final}
1534	return	1576	return
1535	raise	1577	raise
1536	- except (httpx.RemoteProtocolError, httpx.ReadError, OSError) as exc:	1578	+ except (httpx.RemoteProtocolError, httpx.ReadError, OSError, A2AClientError) as exc:
		1579	+ # a2a-sdk 把 httpx 传输错误（RequestError 系）统一包成 A2AClientError，
		1580	+ # 故必须显式纳入；仅当底层 __cause__ 为传输类错误时才视为可恢复瞬断走重连，
		1581	+ # 否则（HTTP 4xx/5xx、SSE 协议错等非传输 A2AClientError）按真失败 re-raise。
		1582	+ if isinstance(exc, A2AClientError) and not isinstance(
		1583	+ getattr(exc, "__cause__", None), (httpx.RequestError, OSError)
		1584	+ ):
		1585	+ raise
1537	if child_task_id is None:	1586	if child_task_id is None:
1538	raise # 首帧前断线，无 task_id 可重连/查询 → failed（wait_for 兜底）	1587	raise # 首帧前断线，无 task_id 可重连/查询 → failed（wait_for 兜底）
1539	if retry_count >= max_retries:	1588	if retry_count >= max_retries:
	@@ -1628,7 +1677,17 @@ class Executor(AgentExecutor):
1628	return {"workflow_id": spec.workflow_id, "status": "cancelled",	1677	return {"workflow_id": spec.workflow_id, "status": "cancelled",
1629	"result": final_result, "error": "", "elapsed_ms": elapsed_ms}	1678	"result": final_result, "error": "", "elapsed_ms": elapsed_ms}
1630		1679
1631	- node_end_result = dict(final_result) if isinstance(final_result, dict) else {"value": final_result}	1680	+ if final_result is None:
		1681	+ # VA 流未给出任何终态（异常截断 / VA 崩溃）→ 不能当 done 静默上报（问题 2）。
		1682	+ await self._emit_sub_task(
		1683	+ turn_ctx, wf_path, "workflow",
		1684	+ {"event": "node_end", "status": "failed", "error": "VA 工作流未返回终态结果"},
		1685	+ )
		1686	+ logger.warning(f"[Executor] 工作流无终态结果：workflow={spec.workflow_id}")
		1687	+ return {"workflow_id": spec.workflow_id, "status": "failed",
		1688	+ "result": None, "error": "VA 工作流未返回终态结果", "elapsed_ms": elapsed_ms}
		1689	+
		1690	+ node_end_result = dict(final_result) # None 已在上方拦截，此处必为 dict
1632	node_end_result["elapsed_ms"] = elapsed_ms	1691	node_end_result["elapsed_ms"] = elapsed_ms
1633	await self._emit_sub_task(	1692	await self._emit_sub_task(
1634	turn_ctx, wf_path, "workflow",	1693	turn_ctx, wf_path, "workflow",
	@@ -1657,22 +1716,29 @@ class Executor(AgentExecutor):
1657	params = cached.get("params", {})	1716	params = cached.get("params", {})
1658	trace_id = cached.get("trace_id", "")	1717	trace_id = cached.get("trace_id", "")
1659		1718
		1719	+ # 与单调路径 _call_versatile_adapter 同构：先做推荐入口临时改写，
		1720	+ # 再用改写后的 query/intent 同时覆盖 body 入参与 target（二者必须一致，否则下游
		1721	+ # 工作流入参与路由 intent 对不上）。
		1722	+ effective_intent, effective_query = _rewrite_recommend_delegate(
		1723	+ delegate.intent, delegate.task_description,
		1724	+ )
		1725	+
1660	# 用 delegate 的 query/intent 覆盖 body（与 _call_versatile_adapter 同构）	1726	# 用 delegate 的 query/intent 覆盖 body（与 _call_versatile_adapter 同构）
1661	input_section = dict(body.get("input") or {})	1727	input_section = dict(body.get("input") or {})
1662	- input_section["query"] = delegate.task_description	1728	+ input_section["query"] = effective_query
1663	- input_section["intent"] = delegate.intent	1729	+ input_section["intent"] = effective_intent
1664	body["input"] = input_section	1730	body["input"] = input_section
1665	custom_data = dict(body.get("custom_data") or {})	1731	custom_data = dict(body.get("custom_data") or {})
1666	custom_inputs = dict(custom_data.get("inputs") or {})	1732	custom_inputs = dict(custom_data.get("inputs") or {})
1667	- custom_inputs["query"] = delegate.task_description	1733	+ custom_inputs["query"] = effective_query
1668	- custom_inputs["intent"] = delegate.intent	1734	+ custom_inputs["intent"] = effective_intent
1669	custom_data["inputs"] = custom_inputs	1735	custom_data["inputs"] = custom_inputs
1670	body["custom_data"] = custom_data	1736	body["custom_data"] = custom_data
1671	body["stream"] = True	1737	body["stream"] = True
1672		1738
1673	request = self._build_va_message(	1739	request = self._build_va_message(
1674	_VaRequestPayload(	1740	_VaRequestPayload(
1675	- query=delegate.task_description,	1741	+ query=effective_query,
1676	headers=headers,	1742	headers=headers,
1677	body=body,	1743	body=body,
1678	params=params,	1744	params=params,
	@@ -1680,6 +1746,9 @@ class Executor(AgentExecutor):
1680	conv_id=conv_id,	1746	conv_id=conv_id,
1681	trace_id=trace_id,	1747	trace_id=trace_id,
1682	agent_id=delegate.target_agent or "",	1748	agent_id=delegate.target_agent or "",
		1749	+ # 路由只用 intent：wf_path[-1] 是模型在单次并行调用内生成的局部唯一 id，
		1750	+ # 非 VA 可识别的真实 workflow_id，不能进 target（否则有撞 id 误路由风险）。
		1751	+ target={"type": "workflow", "intent": effective_intent} if effective_intent else None,
1683	)	1752	)
1684	)	1753	)
1685		1754
	@@ -1705,7 +1774,10 @@ class Executor(AgentExecutor):
1705	final_result = {"workflow_result": self._extract_workflow_result(event)}	1774	final_result = {"workflow_result": self._extract_workflow_result(event)}
1706	break	1775	break
1707	elif state == TASK_STATE_FAILED:	1776	elif state == TASK_STATE_FAILED:
1708	- raise RuntimeError(f"VA 工作流异常终止：path={list(wf_path)}")	1777	+ err = self._extract_failed_error(event) or {"message": "VA 工作流异常终止"}
		1778	+ raise RuntimeError(
		1779	+ f"VA 工作流异常终止：path={list(wf_path)}，{self._format_upstream_error(err)}"
		1780	+ )
1709	elif state == TASK_STATE_INPUT_REQUIRED:	1781	elif state == TASK_STATE_INPUT_REQUIRED:
1710	# 本需求声明不支持中断（PENDING P-011）；并行 fire-and-gather 无法续轮 →	1782	# 本需求声明不支持中断（PENDING P-011）；并行 fire-and-gather 无法续轮 →
1711	# 防御性当异常失败，不静默（同事确认本场景不应出现该终态）。	1783	# 防御性当异常失败，不静默（同事确认本场景不应出现该终态）。

Mapplications/a2a_service/tests/framework_parallel/_helpers.py+13-0

	@@ -46,6 +46,7 @@ from a2a.types.a2a_pb2 import (
46	ROLE_AGENT,	46	ROLE_AGENT,
47	TASK_STATE_CANCELED,	47	TASK_STATE_CANCELED,
48	TASK_STATE_COMPLETED,	48	TASK_STATE_COMPLETED,
		49	+ TASK_STATE_FAILED,
49	TASK_STATE_WORKING,	50	TASK_STATE_WORKING,
50	)	51	)
51	from google.protobuf.json_format import MessageToDict	52	from google.protobuf.json_format import MessageToDict
	@@ -127,6 +128,18 @@ def sr_completed_text(text: str) -> StreamResponse:
127	return StreamResponse(status_update=status_completed_text(text))	128	return StreamResponse(status_update=status_completed_text(text))
128		129
129		130
		131	+def sr_failed(text: str = "") -> StreamResponse:
		132	+ """子 Agent FAILED 终态帧（error 文本走 status.message）。"""
		133	+ parts = [Part(text=text)] if text else []
		134	+ msg = Message(role=ROLE_AGENT, message_id="m", parts=parts) if parts else None
		135	+ su = TaskStatusUpdateEvent(
		136	+ task_id="t", context_id="c",
		137	+ status=TaskStatus(state=TASK_STATE_FAILED, message=msg) if msg
		138	+ else TaskStatus(state=TASK_STATE_FAILED),
		139	+ )
		140	+ return StreamResponse(status_update=su)
		141	+
		142	+
130	# ════════════════════════════════════════════════════════════════════	143	# ════════════════════════════════════════════════════════════════════
131	# 假异步流 / 假子 Agent 客户端	144	# 假异步流 / 假子 Agent 客户端
132	# ════════════════════════════════════════════════════════════════════	145	# ════════════════════════════════════════════════════════════════════

Mapplications/a2a_service/tests/framework_parallel/test_dispatch.py+35-0

	@@ -177,6 +177,41 @@ async def test_run_sub_agent_failure_marks_failed():
177	assert envs[-1]["data"] == {"event": "node_end", "status": "failed", "error": "接口报错"}	177	assert envs[-1]["data"] == {"event": "node_end", "status": "failed", "error": "接口报错"}
178		178
179		179
		180	+async def test_run_sub_agent_terminal_failed_not_silent_done():
		181	+ """问题 1：__terminal__(failed) 终态 → 结果 failed，不落空内容 done。"""
		182	+ executor = make_executor(sub_agent_client=MagicMock())
		183	+ _set_fake_drive(executor, frames=[
		184	+ {"type": "__task_created__", "task_id": "child-1"},
		185	+ {"type": "__terminal__", "status": "failed", "error": "子Agent内部异常"},
		186	+ ])
		187	+ ctx = make_turn_ctx()
		188	+
		189	+ result = await executor._run_sub_agent(_specs("A")[0], ctx, asyncio.Event(), ("A",))
		190	+
		191	+ assert result.status == "failed"
		192	+ assert "子Agent内部异常" in result.error
		193	+ assert result.content == ""
		194	+ envs = collect_sub_tasks(ctx.event_queue)
		195	+ assert envs[-1]["data"]["event"] == "node_end"
		196	+ assert envs[-1]["data"]["status"] == "failed"
		197	+
		198	+
		199	+async def test_run_sub_agent_terminal_cancelled_marks_cancelled():
		200	+ """问题 1：__terminal__(cancelled) 终态 → 结果 cancelled。"""
		201	+ executor = make_executor(sub_agent_client=MagicMock())
		202	+ _set_fake_drive(executor, frames=[
		203	+ {"type": "__task_created__", "task_id": "child-1"},
		204	+ {"type": "__terminal__", "status": "cancelled", "error": ""},
		205	+ ])
		206	+ ctx = make_turn_ctx()
		207	+
		208	+ result = await executor._run_sub_agent(_specs("A")[0], ctx, asyncio.Event(), ("A",))
		209	+
		210	+ assert result.status == "cancelled"
		211	+ envs = collect_sub_tasks(ctx.event_queue)
		212	+ assert envs[-1]["data"]["status"] == "cancelled"
		213	+
		214	+
180	async def test_run_sub_agent_cancelled_when_token_set():	215	async def test_run_sub_agent_cancelled_when_token_set():
181	executor = make_executor(sub_agent_client=MagicMock())	216	executor = make_executor(sub_agent_client=MagicMock())
182	_set_fake_drive(executor, frames=[]) # 无帧，循环后检测 cancel	217	_set_fake_drive(executor, frames=[]) # 无帧，循环后检测 cancel

Mapplications/a2a_service/tests/framework_parallel/test_drive_sub_agent.py+22-0

	@@ -29,6 +29,7 @@ from tests.framework_parallel._helpers import (
29	make_turn_ctx,	29	make_turn_ctx,
30	sr_artifact,	30	sr_artifact,
31	sr_completed_text,	31	sr_completed_text,
		32	+ sr_failed,
32	sr_task,	33	sr_task,
33	task_completed_text,	34	task_completed_text,
34	)	35	)
	@@ -71,6 +72,27 @@ def test_streamresponse_is_not_task_instance():
71	assert not isinstance(sr, Task)	72	assert not isinstance(sr, Task)
72		73
73		74
		75	+# ── 问题 1：子 Agent FAILED 终态 → __terminal__，不可落流末当 done ────────────
		76	+
		77	+
		78	+async def test_failed_status_yields_terminal_not_silent_done():
		79	+ """子 Agent 返回 FAILED status_update → 产出 __terminal__(failed)，error 取 message 文本。"""
		80	+ client = FakeSubAgentClient(send=[FakeAsyncStream([
		81	+ sr_task("child-1"),
		82	+ sr_failed("子Agent内部异常"),
		83	+ ])])
		84	+ executor = make_executor(sub_agent_client=client)
		85	+
		86	+ frames = await _drive(executor)
		87	+
		88	+ assert frames[0] == {"type": "__task_created__", "task_id": "child-1"}
		89	+ assert frames[-1] == {
		90	+ "type": "__terminal__", "status": "failed", "error": "子Agent内部异常",
		91	+ }
		92	+ # 关键：未产出 __completed__（不会被上层当成功）
		93	+ assert all(f.get("type") != "__completed__" for f in frames)
		94	+
		95	+
74	# ── RECONN-03：断连期间已完成 → resubscribe 抛错 → tasks/get 回退 ────────────	96	# ── RECONN-03：断连期间已完成 → resubscribe 抛错 → tasks/get 回退 ────────────
75		97
76		98

Mapplications/a2a_service/tests/framework_parallel/test_workflows.py+91-1

	@@ -14,9 +14,12 @@ from __future__ import annotations
14		14
15	import asyncio	15	import asyncio
16		16
17	-from common.events import MultiDelegateRequest, WorkflowSpec	17	+from google.protobuf.json_format import MessageToDict
		18	+
		19	+from common.events import DelegateRequest, MultiDelegateRequest, WorkflowSpec
18		20
19	from tests.framework_parallel._helpers import (	21	from tests.framework_parallel._helpers import (
		22	+ FakeAsyncStream,
20	collect_sub_tasks,	23	collect_sub_tasks,
21	make_executor,	24	make_executor,
22	make_turn_ctx,	25	make_turn_ctx,
	@@ -127,6 +130,22 @@ async def test_run_one_workflow_failure_isolated():
127	assert envs[-1]["sub_task_path"] == ["A", "wf:c"]	130	assert envs[-1]["sub_task_path"] == ["A", "wf:c"]
128		131
129		132
		133	+async def test_run_one_workflow_no_terminal_result_marks_failed():
		134	+ """问题 2：VA 流未给出任何终态（final_result is None）→ 判 failed，不静默 done。"""
		135	+ executor = make_executor()
		136	+ _set_fake_va(executor, returns=None)
		137	+ ctx = make_turn_ctx(sub_task_path=("A",))
		138	+
		139	+ result = await executor._run_one_workflow(_wfs("wf:n")[0], ctx, asyncio.Event())
		140	+
		141	+ assert result["status"] == "failed"
		142	+ assert result["result"] is None
		143	+ assert "未返回终态" in result["error"]
		144	+ envs = collect_sub_tasks(ctx.event_queue)
		145	+ assert envs[-1]["data"]["event"] == "node_end"
		146	+ assert envs[-1]["data"]["status"] == "failed"
		147	+
		148	+
130	async def test_run_one_workflow_timeout():	149	async def test_run_one_workflow_timeout():
131	executor = make_executor(workflow_timeout_seconds=0.05)	150	executor = make_executor(workflow_timeout_seconds=0.05)
132	_set_fake_va(executor, returns={"url": "late"}, sleep=1)	151	_set_fake_va(executor, returns={"url": "late"}, sleep=1)
	@@ -152,3 +171,74 @@ async def test_run_one_workflow_cancelled():
152	assert result["status"] == "cancelled"	171	assert result["status"] == "cancelled"
153	envs = collect_sub_tasks(ctx.event_queue)	172	envs = collect_sub_tasks(ctx.event_queue)
154	assert envs[-1]["data"]["status"] == "cancelled"	173	assert envs[-1]["data"]["status"] == "cancelled"
		174	+
		175	+
		176	+# ════════════════════════════════════════════════════════════════════
		177	+# 并行委托 → VA 请求构造：intent 改写对齐 + target 仅用 intent 路由（决策 a/b）
		178	+# 这两个用例不 stub _drive_workflow_va，直接驱动真实函数体并捕获发往 VA 的请求。
		179	+# ════════════════════════════════════════════════════════════════════
		180	+
		181	+
		182	+def _capture_va_request(executor) -> dict:
		183	+ """让真实 _drive_workflow_va 跑起来：捕获发往 VA 的 SendMessageRequest，返回空流。"""
		184	+ captured: dict = {}
		185	+
		186	+ def send_message(request):
		187	+ captured["request"] = request
		188	+ return FakeAsyncStream([]) # 空流 → async for 直接结束，请求已在迭代前构造
		189	+
		190	+ executor._va_client.send_message = send_message
		191	+ return captured
		192	+
		193	+
		194	+def _va_data_part(request) -> dict:
		195	+ """从 SendMessageRequest 的 DataPart 还原 dict（target/headers/body/params/...）。"""
		196	+ for part in request.message.parts:
		197	+ if part.WhichOneof("content") == "data":
		198	+ return MessageToDict(part.data)
		199	+ return {}
		200	+
		201	+
		202	+def _va_text_part(request) -> str:
		203	+ for part in request.message.parts:
		204	+ if part.WhichOneof("content") == "text":
		205	+ return part.text
		206	+ return ""
		207	+
		208	+
		209	+async def test_drive_workflow_va_rewrites_intent_and_omits_workflow_id():
		210	+ """决策 a：推荐入口改写生效，body 入参与 target 的 intent 一致；
		211	+ 决策 b：target 只用 intent 路由，不含模型生成的局部 workflow_id（wf_path[-1]）。"""
		212	+ executor = make_executor()
		213	+ captured = _capture_va_request(executor)
		214	+ ctx = make_turn_ctx(sub_task_path=("A",))
		215	+ delegate = DelegateRequest(intent="理财推荐", task_description="推荐理财产品")
		216	+
		217	+ await executor._drive_workflow_va(delegate, ctx, ("A", "wf-1"), asyncio.Event())
		218	+
		219	+ data = _va_data_part(captured["request"])
		220	+ # 决策 a：理财推荐 → 理财选品购买；query → 请推荐低风险理财产品（body 与 target 一致）
		221	+ assert data["target"]["intent"] == "理财选品购买"
		222	+ assert data["body"]["input"]["intent"] == "理财选品购买"
		223	+ assert data["body"]["input"]["query"] == "请推荐低风险理财产品"
		224	+ assert data["body"]["custom_data"]["inputs"]["intent"] == "理财选品购买"
		225	+ assert _va_text_part(captured["request"]) == "请推荐低风险理财产品"
		226	+ # 决策 b：target 不含 workflow_id（wf_path[-1]="wf-1" 仅用于节点盖章，不进路由）
		227	+ assert data["target"]["type"] == "workflow"
		228	+ assert "workflow_id" not in data["target"]
		229	+
		230	+
		231	+async def test_drive_workflow_va_passthrough_intent_no_rewrite():
		232	+ """非改写 intent：body 与 target 用原始 intent；仍不含 workflow_id，
		233	+ 且 _build_va_message 注入 conversation_id。"""
		234	+ executor = make_executor()
		235	+ captured = _capture_va_request(executor)
		236	+ ctx = make_turn_ctx(conv_id="c", sub_task_path=("A",))
		237	+ delegate = DelegateRequest(intent="转账", task_description="给张三转100")
		238	+
		239	+ await executor._drive_workflow_va(delegate, ctx, ("A", "wf-9"), asyncio.Event())
		240	+
		241	+ data = _va_data_part(captured["request"])
		242	+ assert data["target"] == {"type": "workflow", "intent": "转账", "conversation_id": "c"}
		243	+ assert data["body"]["input"]["intent"] == "转账"
		244	+ assert data["body"]["input"]["query"] == "给张三转100"

Mapplications/a2a_service/tests/integration/test_versatile_delegate.py+114-1

	@@ -25,6 +25,8 @@ from __future__ import annotations
25		25
26	import asyncio	26	import asyncio
27	import json	27	import json
		28	+import sys
		29	+from pathlib import Path
28	from types import SimpleNamespace	30	from types import SimpleNamespace
29	from unittest.mock import AsyncMock, MagicMock	31	from unittest.mock import AsyncMock, MagicMock
30		32
	@@ -58,7 +60,7 @@ from common.events import (
58	ToolStatusEvent,	60	ToolStatusEvent,
59	)	61	)
60	from config import get_settings	62	from config import get_settings
61	-from orchestrator.executor import Executor, _TurnContext	63	+from orchestrator.executor import Executor, _TurnContext, _VaRequestPayload
62		64
63		65
64	CONV_ID = "conv-delegate-1"	66	CONV_ID = "conv-delegate-1"
	@@ -241,6 +243,80 @@ def _make_executor_with_va_stream(va_events: list) -> Executor:
241	return Executor(va_client=va_client, redis=redis, task_store=task_store)	243	return Executor(va_client=va_client, redis=redis, task_store=task_store)
242		244
243		245
		246	+def test_build_va_message_packs_target_for_workflow_routing():
		247	+ executor = _make_executor_with_va_stream([])
		248	+ request = executor._build_va_message(
		249	+ _VaRequestPayload(
		250	+ query="查理财",
		251	+ headers={},
		252	+ body={"custom_data": {}},
		253	+ params={},
		254	+ conv_id=CONV_ID,
		255	+ target={
		256	+ "type": "workflow",
		257	+ "intent": "理财推荐",
		258	+ "workflow_id": "wf_wealth",
		259	+ },
		260	+ )
		261	+ )
		262	+
		263	+ data_part = next(p for p in request.message.parts if p.WhichOneof("content") == "data")
		264	+ carried = MessageToDict(data_part.data)
		265	+ assert carried["target"] == {
		266	+ "type": "workflow",
		267	+ "conversation_id": CONV_ID,
		268	+ "intent": "理财推荐",
		269	+ "workflow_id": "wf_wealth",
		270	+ }
		271	+
		272	+
		273	+def test_a2a_message_target_routes_to_va_workflow_adapter(tmp_path):
		274	+ va_root = Path(__file__).resolve().parents[3] / "versatile_adapter"
		275	+ if str(va_root) not in sys.path:
		276	+ sys.path.insert(0, str(va_root))
		277	+
		278	+ from dispatcher.runner import VersatileAdapterRunner
		279	+
		280	+ config_path = tmp_path / "versatile_proxy.yaml"
		281	+ config_path.write_text(
		282	+ """
		283	+adapters:
		284	+ - name: default_controller
		285	+ type: controller
		286	+ url_template: "http://mock-host/v1/agents/agent-a/conversations/{conversation_id}"
		287	+ - name: wf_wealth
		288	+ type: workflow
		289	+ url_template: "http://mock-host/v1/workflows/{workflow_id}/conversations/{conversation_id}"
		290	+ workflow_id: wf_wealth
		291	+ intent: "理财推荐"
		292	+""",
		293	+ encoding="utf-8",
		294	+ )
		295	+ executor = _make_executor_with_va_stream([])
		296	+ request = executor._build_va_message(
		297	+ _VaRequestPayload(
		298	+ query="查理财",
		299	+ headers={},
		300	+ body={"custom_data": {}},
		301	+ params={},
		302	+ conv_id=CONV_ID,
		303	+ target={
		304	+ "type": "workflow",
		305	+ "intent": "理财推荐",
		306	+ "workflow_id": "wf_wealth",
		307	+ },
		308	+ )
		309	+ )
		310	+
		311	+ data_part = next(p for p in request.message.parts if p.WhichOneof("content") == "data")
		312	+ target = MessageToDict(data_part.data)["target"]
		313	+ runner = VersatileAdapterRunner(config_path=config_path)
		314	+ cfg = runner._match_workflow(target)
		315	+
		316	+ assert cfg is not None
		317	+ assert cfg.name == "wf_wealth"
		318	+
		319	+
244	def _is_event_with_state(event, state) -> bool:	320	def _is_event_with_state(event, state) -> bool:
245	"""判断事件是否是 ``TaskStatusUpdateEvent`` 且处于指定状态。"""	321	"""判断事件是否是 ``TaskStatusUpdateEvent`` 且处于指定状态。"""
246	return (	322	return (
	@@ -621,6 +697,43 @@ async def test_va_failed_event_does_not_emit_input_required(monkeypatch):
621	assert len(input_required) == 0, "VA FAILED 路径不应发出 INPUT_REQUIRED 状态事件"	697	assert len(input_required) == 0, "VA FAILED 路径不应发出 INPUT_REQUIRED 状态事件"
622		698
623		699
		700	+@pytest.mark.asyncio
		701	+async def test_va_failed_event_with_plain_text_error_is_forwarded(monkeypatch):
		702	+ """VA FAILED 携带非 JSON upstream_error 文本时，应原样作为错误详情透传。"""
		703	+ event = _va_failed_event(error_payload=None)
		704	+ message = Message(
		705	+ role=ROLE_AGENT,
		706	+ message_id="va-msg-failed-text",
		707	+ task_id="va-task-1",
		708	+ context_id=CONV_ID,
		709	+ parts=[_text_part("上游服务不可用", vatype="upstream_error")],
		710	+ )
		711	+ event.status.message.CopyFrom(message)
		712	+ executor, _task, _task_store = _make_executor_with_real_task([event])
		713	+
		714	+ async def fake_agent_stream(**kwargs):
		715	+ yield DelegateRequest(intent="查", task_description="查")
		716	+
		717	+ monkeypatch.setattr("orchestrator.executor.agent_stream", fake_agent_stream)
		718	+
		719	+ queue = EventQueue()
		720	+ await executor.run_agent(
		721	+ _make_turn_ctx(queue),
		722	+ query="x",
		723	+ original_body={},
		724	+ cascade_result=None,
		725	+ )
		726	+
		727	+ enqueued = _drain_queue(queue)
		728	+ failed_events = [e for e in enqueued if _is_event_with_state(e, TASK_STATE_FAILED)]
		729	+ assert len(failed_events) == 1
		730	+ text_chunks = [
		731	+ p.text for p in failed_events[0].status.message.parts
		732	+ if p.WhichOneof("content") == "text"
		733	+ ]
		734	+ assert "上游服务不可用" in text_chunks
		735	+
		736	+
624	@pytest.mark.asyncio	737	@pytest.mark.asyncio
625	async def test_va_failed_event_without_payload_falls_back_to_generic_message(monkeypatch):	738	async def test_va_failed_event_without_payload_falls_back_to_generic_message(monkeypatch):
626	"""VA FAILED 事件 status.message 为空时，应使用兜底通用错误文案。"""	739	"""VA FAILED 事件 status.message 为空时，应使用兜底通用错误文案。"""

Mapplications/versatile_adapter/.env.example+16-1

	@@ -2,9 +2,16 @@
2	ADAPTER_APP_NAME=VersatileAdapter	2	ADAPTER_APP_NAME=VersatileAdapter
3		3
4	# ── Versatile 低码平台 ────────────────────────────────────────────────────────	4	# ── Versatile 低码平台 ────────────────────────────────────────────────────────
		5	+# 说明：以下 VERSATILE_URL_TEMPLATE / VERSATILE_TIMEOUT /
		6	+# VERSATILE_HEADERS_TEMPLATE / VERSATILE_WORKFLOW_RESULT_NODE 等配置，
		7	+# 仅在未提供或未成功加载 /etc/edpagent/config/versatile_proxy.yaml 时，
		8	+# 用于自动生成唯一的 default_controller 兜底配置。
		9	+# 正式部署、多 adapter 路由、workflow intent/workflow_id 匹配等场景，
		10	+# 请优先使用 versatile_proxy.yaml；YAML 存在且 adapters 非空时，以 YAML 为准。
5	# {conversation_id} 在运行时替换为实际会话 ID	11	# {conversation_id} 在运行时替换为实际会话 ID
6	VERSATILE_URL_TEMPLATE=	12	VERSATILE_URL_TEMPLATE=
7	-VERSATILE_TIMEOUT=600	13	+# versatile请求超时时延，需要<=工小智前端超时时延57s
		14	+VERSATILE_TIMEOUT=57
8		15
9	# 调用 Versatile 时无条件注入的默认请求头（JSON）。默认仅含 Accept/stream；需要	16	# 调用 Versatile 时无条件注入的默认请求头（JSON）。默认仅含 Accept/stream；需要
10	# Session 鉴权（Cookie:AGENT_SID 等）的环境取消下面那行的 # 注释 + 替换 token 即可。	17	# Session 鉴权（Cookie:AGENT_SID 等）的环境取消下面那行的 # 注释 + 替换 token 即可。
	@@ -32,3 +39,11 @@ va_workflow_result_node=
32	# ── Log ───────────────────────────────────────────────────────────────────────	39	# ── Log ───────────────────────────────────────────────────────────────────────
33	ADAPTER_LOG_LEVEL=DEBUG	40	ADAPTER_LOG_LEVEL=DEBUG
34	ADAPTER_LOG_FILE=logs/versatile_adapter.log	41	ADAPTER_LOG_FILE=logs/versatile_adapter.log
		42	+
		43	+# ── Redis（会话状态）─────────────────────────────────────────────────────────
		44	+REDIS_HOST=localhost
		45	+REDIS_PORT=6379
		46	+REDIS_DB=0
		47	+REDIS_PASSWORD=
		48	+REDIS_SESSION_TTL=180
		49	+

Mapplications/versatile_adapter/a2a_facade/executor.py+5-2

	@@ -89,12 +89,15 @@ class A2aVersatileExecutor(AgentExecutor):
89		89
90	elif event.execution_completed is not None:	90	elif event.execution_completed is not None:
91	is_failed = event.execution_completed.is_failed	91	is_failed = event.execution_completed.is_failed
92	- if event.execution_completed.result:	92	+ if event.execution_completed.error_message:
		93	+ part = self._make_text_part(event.execution_completed.error_message, "upstream_error")
		94	+ message = new_message(parts=[part])
		95	+ elif event.execution_completed.result:
93	part = self._make_text_part(event.execution_completed.result, "workflow_result")	96	part = self._make_text_part(event.execution_completed.result, "workflow_result")
94	message = new_message(parts=[part])	97	message = new_message(parts=[part])
95		98
96	if is_failed:	99	if is_failed:
97	- await updater.failed()	100	+ await updater.failed(message)
98	logger.info(f"[A2aVA] 流异常结束(failed): conv_id={context.context_id}, task_id={context.task_id}")	101	logger.info(f"[A2aVA] 流异常结束(failed): conv_id={context.context_id}, task_id={context.task_id}")
99	else:	102	else:
100	await updater.complete(message)	103	await updater.complete(message)

Mapplications/versatile_adapter/adapters/versatile_controller.py+8-4

	@@ -10,6 +10,7 @@ End/exception 终态检测、GXZQAResponseNode 过滤。
10	from __future__ import annotations	10	from __future__ import annotations
11		11
12	import json as _json	12	import json as _json
		13	+import re
13	from typing import Optional	14	from typing import Optional
14		15
15	from loguru import logger	16	from loguru import logger
	@@ -46,15 +47,16 @@ class VersatileController(VersatileProxy):
46	self._workflow_result_node = workflow_result_node	47	self._workflow_result_node = workflow_result_node
47		48
48	def _process_chunk(self, chunk: str, ctx: VersatileStreamCtx) -> list[AdapterEvent]:	49	def _process_chunk(self, chunk: str, ctx: VersatileStreamCtx) -> list[AdapterEvent]:
49	- if '"node_type":"End"' in chunk:	50	+ if re.search(r'"node_type"\s:\s"End"', chunk):
50	logger.debug(f"[VersatileController] End 节点，yield data_proxy")	51	logger.debug(f"[VersatileController] End 节点，yield data_proxy")
51	ctx.completed = True	52	ctx.completed = True
52	return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]	53	return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]
53		54
54	- if '"event":"exception"' in chunk:	55	+ if re.search(r'"event"\s:\s"exception"', chunk):
55	logger.debug(f"[VersatileController] exception 帧，yield data_proxy")	56	logger.debug(f"[VersatileController] exception 帧，yield data_proxy")
56	ctx.completed = True	57	ctx.completed = True
57	ctx.is_failed = True	58	ctx.is_failed = True
		59	+ ctx.error_message = chunk
58	return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]	60	return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]
59		61
60	if self._workflow_result_node and f'"node_name":"{self._workflow_result_node}"' in chunk:	62	if self._workflow_result_node and f'"node_name":"{self._workflow_result_node}"' in chunk:
	@@ -78,9 +80,11 @@ class VersatileController(VersatileProxy):
78	if not ctx.completed:	80	if not ctx.completed:
79	return [AdapterEvent(execution_input_required=ExecutionInputRequiredContent())]	81	return [AdapterEvent(execution_input_required=ExecutionInputRequiredContent())]
80		82
81	- if ctx.execution_result:	83	+ if ctx.is_failed or ctx.execution_result:
82	return [AdapterEvent(execution_completed=ExecutionCompletedContent(	84	return [AdapterEvent(execution_completed=ExecutionCompletedContent(
83	- is_failed=ctx.is_failed, result=ctx.execution_result	85	+ is_failed=ctx.is_failed,
		86	+ result=ctx.execution_result or "",
		87	+ error_message=ctx.error_message,
84	))]	88	))]
85		89
86	return []	90	return []

Mapplications/versatile_adapter/adapters/versatile_proxy.py+3-7

	@@ -9,7 +9,6 @@ VersatileProxy — HTTP+SSE 流式调用基类。
9	"""	9	"""
10	from __future__ import annotations	10	from __future__ import annotations
11		11
12	-import json as _json
13	from typing import AsyncGenerator, Optional	12	from typing import AsyncGenerator, Optional
14		13
15	from abc import abstractmethod	14	from abc import abstractmethod
	@@ -26,12 +25,13 @@ from adapters.base_adapter import BaseAdapter
26	class VersatileStreamCtx:	25	class VersatileStreamCtx:
27	"""SSE 行循环中累积的可变状态，贯穿 _process_line → _process_chunk → _on_stream_end。"""	26	"""SSE 行循环中累积的可变状态，贯穿 _process_line → _process_chunk → _on_stream_end。"""
28		27
29	- __slots__ = ("completed", "is_failed", "execution_result")	28	+ __slots__ = ("completed", "is_failed", "execution_result", "error_message")
30		29
31	def __init__(self) -> None:	30	def __init__(self) -> None:
32	self.completed: bool = False	31	self.completed: bool = False
33	self.is_failed: bool = False	32	self.is_failed: bool = False
34	self.execution_result: str \| None = None	33	self.execution_result: str \| None = None
		34	+ self.error_message: str = ""
35		35
36		36
37	class VersatileProxy(BaseAdapter):	37	class VersatileProxy(BaseAdapter):
	@@ -154,11 +154,7 @@ class VersatileProxy(BaseAdapter):
154	for key, value in request.headers.items():	154	for key, value in request.headers.items():
155	cmd += f" -H '{key}: {value}'"	155	cmd += f" -H '{key}: {value}'"
156	if body:	156	if body:
157	- try:	157	+ cmd += f" -d '{body.decode('utf-8', errors='replace')}'"
158	- json_body = _json.loads(body.decode("utf-8"))
159	- cmd += f" -d '{_json.dumps(json_body, ensure_ascii=False)}'"
160	- except Exception:
161	- cmd += f" -d '{body.decode('utf-8', errors='replace')}'"
162	banner_start = f"{'='20} Proxy Request (Stream) Start {'='20}"	158	banner_start = f"{'='20} Proxy Request (Stream) Start {'='20}"
163	banner_end = f"{'='20} Proxy Request (Stream) End {'='20}"	159	banner_end = f"{'='20} Proxy Request (Stream) End {'='20}"
164	logger.info("[VersatileProxy] {}", banner_start)	160	logger.info("[VersatileProxy] {}", banner_start)

Mapplications/versatile_adapter/adapters/versatile_workflow.py+46-4

	@@ -10,20 +10,24 @@ HTTP 流断流由调用方（executor）通过流结束触发 complete()。
10	"""	10	"""
11	from __future__ import annotations	11	from __future__ import annotations
12		12
		13	+import json as _json
		14	+import re
13	from typing import Optional	15	from typing import Optional
14		16
		17	+from loguru import logger
		18	+
15	from adapters.versatile_proxy import VersatileProxy, VersatileStreamCtx	19	from adapters.versatile_proxy import VersatileProxy, VersatileStreamCtx
16	from event.events import (	20	from event.events import (
17	AdapterEvent,	21	AdapterEvent,
18	DataProxyContent,	22	DataProxyContent,
		23	+ ExecutionCompletedContent,
		24	+ ExecutionInputRequiredContent,
19	)	25	)
20		26
21		27
22	class VersatileWorkflow(VersatileProxy):	28	class VersatileWorkflow(VersatileProxy):
23	"""低码工作流协议适配器。"""	29	"""低码工作流协议适配器。"""
24		30
25	- _SKIP_TYPES = frozenset({"finish", "runCompleted", "dialogId"})
26	-
27	def __init__(	31	def __init__(
28	self,	32	self,
29	url_template: str,	33	url_template: str,
	@@ -31,18 +35,56 @@ class VersatileWorkflow(VersatileProxy):
31	timeout: int = 600,	35	timeout: int = 600,
32	headers_template: Optional[dict] = None,	36	headers_template: Optional[dict] = None,
33	forward_header_whitelist: Optional[set[str]] = None,	37	forward_header_whitelist: Optional[set[str]] = None,
		38	+ workflow_result_node: Optional[str] = None,
34	) -> None:	39	) -> None:
35	super().__init__(url_template, timeout, headers_template, forward_header_whitelist)	40	super().__init__(url_template, timeout, headers_template, forward_header_whitelist)
36	self._workflow_id = workflow_id	41	self._workflow_id = workflow_id
		42	+ self._workflow_result_node = workflow_result_node
37		43
38	def _build_url(self, conv_id: str) -> str:	44	def _build_url(self, conv_id: str) -> str:
39	return self._url_template.format(conversation_id=conv_id, workflow_id=self._workflow_id)	45	return self._url_template.format(conversation_id=conv_id, workflow_id=self._workflow_id)
40		46
41	def _process_chunk(self, chunk: str, ctx: VersatileStreamCtx) -> list[AdapterEvent]:	47	def _process_chunk(self, chunk: str, ctx: VersatileStreamCtx) -> list[AdapterEvent]:
42	- for t in self._SKIP_TYPES:	48	+ if re.search(r'"type"\s:\s"(?:finish\|runCompleted)"', chunk):
43	- if f'"type":"{t}"' in chunk:	49	+ ctx.completed = True
		50	+ return []
		51	+
		52	+ if re.search(r'"type"\s:\s"dialogId"', chunk):
		53	+ return []
		54	+
		55	+ if re.search(r'"event"\s:\s"exception"', chunk):
		56	+ logger.debug(f"[VersatileWorkflow] exception 帧，yield data_proxy")
		57	+ ctx.completed = True
		58	+ ctx.is_failed = True
		59	+ ctx.error_message = chunk
		60	+ return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]
		61	+
		62	+ if self._workflow_result_node and f'"node_name":"{self._workflow_result_node}"' in chunk:
		63	+ try:
		64	+ parsed = _json.loads(chunk)
		65	+ except Exception:
		66	+ logger.warning(f"[VersatileWorkflow] 无法解析 workflow_result 行: {chunk!r:.80}")
		67	+ return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]
		68	+ data = (parsed.get("custom_rsp_data") or parsed).get("data") or {}
		69	+ if isinstance(data, dict) and data.get("node_type") == "QA":
		70	+ text = data.get("text", "") or ""
		71	+ if not text:
		72	+ return []
		73	+ logger.debug(f"[VersatileWorkflow] workflow_result: {text!r:.60}")
		74	+ ctx.execution_result = text
44	return []	75	return []
		76	+
45	return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]	77	return [AdapterEvent(data_proxy=DataProxyContent(raw_data=chunk))]
46		78
47	def _on_stream_end(self, ctx: VersatileStreamCtx) -> list[AdapterEvent]:	79	def _on_stream_end(self, ctx: VersatileStreamCtx) -> list[AdapterEvent]:
		80	+ if not ctx.completed:
		81	+ return [AdapterEvent(execution_input_required=ExecutionInputRequiredContent())]
		82	+
		83	+ if ctx.is_failed or ctx.execution_result:
		84	+ return [AdapterEvent(execution_completed=ExecutionCompletedContent(
		85	+ is_failed=ctx.is_failed,
		86	+ result=ctx.execution_result or "",
		87	+ error_message=ctx.error_message,
		88	+ ))]
		89	+
48	return []	90	return []

Mapplications/versatile_adapter/dispatcher/runner.py+37-4

	@@ -11,6 +11,7 @@ VersatileAdapterRunner — 配置驱动的动态路由层。
11	"""	11	"""
12	from __future__ import annotations	12	from __future__ import annotations
13		13
		14	+import os
14	from pathlib import Path	15	from pathlib import Path
15	from typing import AsyncGenerator, Optional	16	from typing import AsyncGenerator, Optional
16		17
	@@ -23,7 +24,34 @@ from config import get_settings
23	from event.events import AdapterEvent	24	from event.events import AdapterEvent
24		25
25		26
26	-_DEFAULT_CONFIG_PATH = Path("/etc/edpagent/config/versatile_proxy.yaml")	27	+# 配置文件加载优先级：
		28	+# 1. VersatileAdapterRunner(config_path=...) 显式参数
		29	+# 2. 环境变量 VERSATILE_PROXY_CONFIG_PATH
		30	+# 3. 部署默认路径 /etc/edpagent/config/versatile_proxy.yaml
		31	+# 4. 当前 versatile_adapter 目录下的 versatile_proxy.yaml（便于本地开发）
		32	+_DEPLOY_DEFAULT_CONFIG_PATH = Path("/etc/edpagent/config/versatile_proxy.yaml")
		33	+_LOCAL_DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent.parent / "versatile_proxy.yaml"
		34	+
		35	+
		36	+def _resolve_default_config_path() -> Path:
		37	+ env_path = os.environ.get("VERSATILE_PROXY_CONFIG_PATH")
		38	+ if env_path:
		39	+ return Path(env_path)
		40	+ if _DEPLOY_DEFAULT_CONFIG_PATH.exists():
		41	+ return _DEPLOY_DEFAULT_CONFIG_PATH
		42	+ return _LOCAL_DEFAULT_CONFIG_PATH
		43	+
		44	+
		45	+def _merge_workflow_defaults(raw: dict, workflow_defaults: dict) -> dict:
		46	+ if raw.get("type") != "workflow":
		47	+ return raw
		48	+
		49	+ merged = {workflow_defaults, raw}
		50	+ default_headers = workflow_defaults.get("headers_template") or {}
		51	+ adapter_headers = raw.get("headers_template") or {}
		52	+ if default_headers or adapter_headers:
		53	+ merged["headers_template"] = {default_headers, adapter_headers}
		54	+ return merged
27		55
28		56
29	class _VersatileAdapterConfig:	57	class _VersatileAdapterConfig:
	@@ -54,7 +82,7 @@ class VersatileAdapterRunner:
54	"""配置驱动的动态路由 Runner。"""	82	"""配置驱动的动态路由 Runner。"""
55		83
56	def __init__(self, config_path: Optional[Path] = None) -> None:	84	def __init__(self, config_path: Optional[Path] = None) -> None:
57	- path = config_path or _DEFAULT_CONFIG_PATH	85	+ path = config_path or _resolve_default_config_path()
58	self._adapters = self._load_config(path)	86	self._adapters = self._load_config(path)
59	if not self._adapters:	87	if not self._adapters:
60	self._adapters = self._build_from_settings()	88	self._adapters = self._build_from_settings()
	@@ -70,9 +98,13 @@ class VersatileAdapterRunner:
70	logger.warning(f"[VersatileAdapterRunner] 配置文件不存在: {path}，将从 Settings 生成")	98	logger.warning(f"[VersatileAdapterRunner] 配置文件不存在: {path}，将从 Settings 生成")
71	return []	99	return []
72	with open(path, encoding="utf-8") as f:	100	with open(path, encoding="utf-8") as f:
73	- raw = yaml.safe_load(f)	101	+ raw = yaml.safe_load(f) or {}
		102	+ workflow_defaults = raw.get("workflow_defaults", {})
74	adapters_raw = raw.get("adapters", [])	103	adapters_raw = raw.get("adapters", [])
75	- return [_VersatileAdapterConfig(b) for b in adapters_raw]	104	+ return [
		105	+ _VersatileAdapterConfig(_merge_workflow_defaults(b, workflow_defaults))
		106	+ for b in adapters_raw
		107	+ ]
76		108
77	@staticmethod	109	@staticmethod
78	def _build_from_settings() -> list[_VersatileAdapterConfig]:	110	def _build_from_settings() -> list[_VersatileAdapterConfig]:
	@@ -118,6 +150,7 @@ class VersatileAdapterRunner:
118	timeout=cfg.timeout,	150	timeout=cfg.timeout,
119	headers_template=cfg.headers_template,	151	headers_template=cfg.headers_template,
120	forward_header_whitelist=whitelist,	152	forward_header_whitelist=whitelist,
		153	+ workflow_result_node=cfg.workflow_result_node,
121	)	154	)
122	return VersatileController(	155	return VersatileController(
123	url_template=cfg.url_template,	156	url_template=cfg.url_template,

Mapplications/versatile_adapter/event/events.py+3-2

	@@ -28,9 +28,10 @@ class ExecutionInputRequiredContent(BaseModel, frozen=True):
28		28
29		29
30	class ExecutionCompletedContent(BaseModel, frozen=True):	30	class ExecutionCompletedContent(BaseModel, frozen=True):
31	- """终态信号：任务完成并携带工作流结果。"""	31	+ """终态信号：任务完成并携带工作流结果或错误详情。"""
32	is_failed: bool = False	32	is_failed: bool = False
33	- result: str	33	+ result: str = ""
		34	+ error_message: str = ""
34		35
35		36
36	class AdapterEvent(BaseModel):	37	class AdapterEvent(BaseModel):

Mapplications/versatile_adapter/tests/conftest.py+2-0

	@@ -39,6 +39,7 @@ adapters:
39	timeout: 30	39	timeout: 30
40	workflow_id: wf_knowledge_qa	40	workflow_id: wf_knowledge_qa
41	intent: knowledge_qa	41	intent: knowledge_qa
		42	+ workflow_result_node: WorkflowQAResponseNode
42	forward_header_whitelist:	43	forward_header_whitelist:
43	- x-trace-id	44	- x-trace-id
44		45
	@@ -48,6 +49,7 @@ adapters:
48	timeout: 30	49	timeout: 30
49	workflow_id: wf_wealth	50	workflow_id: wf_wealth
50	intent: "理财推荐"	51	intent: "理财推荐"
		52	+ workflow_result_node: WealthQAResponseNode
51	"""	53	"""
52		54
53		55

Mapplications/versatile_adapter/tests/test_events_and_card.py+1-0

	@@ -47,6 +47,7 @@ class TestAdapterEvent:
47	)	47	)
48	)	48	)
49	assert e.execution_completed.is_failed is True	49	assert e.execution_completed.is_failed is True
		50	+ assert e.execution_completed.error_message == ""
50		51
51	@staticmethod	52	@staticmethod
52	def test_execution_input_required_event():	53	def test_execution_input_required_event():

Mapplications/versatile_adapter/tests/test_runner_routing.py+135-0

	@@ -10,16 +10,150 @@ from __future__ import annotations
10		10
11	import pytest	11	import pytest
12		12
		13	+import dispatcher.runner as runner_module
13	from adapters.versatile_controller import VersatileController	14	from adapters.versatile_controller import VersatileController
14	from adapters.versatile_workflow import VersatileWorkflow	15	from adapters.versatile_workflow import VersatileWorkflow
15	from dispatcher.runner import VersatileAdapterRunner	16	from dispatcher.runner import VersatileAdapterRunner
16		17
17		18
18	# ════════════════════════════════════════════════════════════════════	19	# ════════════════════════════════════════════════════════════════════
		20	+# 默认配置路径解析
		21	+# ════════════════════════════════════════════════════════════════════
		22	+
		23	+
		24	+def test_resolve_default_config_path_uses_env(monkeypatch, tmp_path):
		25	+ custom_path = tmp_path / "custom.yaml"
		26	+ monkeypatch.setenv("VERSATILE_PROXY_CONFIG_PATH", str(custom_path))
		27	+
		28	+ assert runner_module._resolve_default_config_path() == custom_path
		29	+
		30	+
		31	+def test_resolve_default_config_path_uses_deploy_when_exists(monkeypatch, tmp_path):
		32	+ deploy_path = tmp_path / "deploy.yaml"
		33	+ local_path = tmp_path / "local.yaml"
		34	+ deploy_path.write_text("adapters: []", encoding="utf-8")
		35	+ monkeypatch.delenv("VERSATILE_PROXY_CONFIG_PATH", raising=False)
		36	+ monkeypatch.setattr(runner_module, "_DEPLOY_DEFAULT_CONFIG_PATH", deploy_path)
		37	+ monkeypatch.setattr(runner_module, "_LOCAL_DEFAULT_CONFIG_PATH", local_path)
		38	+
		39	+ assert runner_module._resolve_default_config_path() == deploy_path
		40	+
		41	+
		42	+def test_resolve_default_config_path_falls_back_to_local(monkeypatch, tmp_path):
		43	+ deploy_path = tmp_path / "missing.yaml"
		44	+ local_path = tmp_path / "local.yaml"
		45	+ monkeypatch.delenv("VERSATILE_PROXY_CONFIG_PATH", raising=False)
		46	+ monkeypatch.setattr(runner_module, "_DEPLOY_DEFAULT_CONFIG_PATH", deploy_path)
		47	+ monkeypatch.setattr(runner_module, "_LOCAL_DEFAULT_CONFIG_PATH", local_path)
		48	+
		49	+ assert runner_module._resolve_default_config_path() == local_path
		50	+
		51	+
19	# YAML 加载	52	# YAML 加载
20	# ════════════════════════════════════════════════════════════════════	53	# ════════════════════════════════════════════════════════════════════
21		54
22		55
		56	+def test_explicit_config_path_takes_priority_over_env(monkeypatch, write_yaml, tmp_path):
		57	+ env_path = tmp_path / "env.yaml"
		58	+ monkeypatch.setenv("VERSATILE_PROXY_CONFIG_PATH", str(env_path))
		59	+
		60	+ runner = VersatileAdapterRunner(config_path=write_yaml())
		61	+
		62	+ assert [a.name for a in runner._adapters] == [
		63	+ "default_controller",
		64	+ "wf_knowledge_qa",
		65	+ "wf_wealth",
		66	+ ]
		67	+
		68	+
		69	+def test_workflow_defaults_are_inherited_by_workflow_adapter(write_yaml):
		70	+ runner = VersatileAdapterRunner(config_path=write_yaml("""
		71	+workflow_defaults:
		72	+ url_template: "http://mock-host/v1/workflows/{workflow_id}/conversations/{conversation_id}"
		73	+ timeout: 45
		74	+ headers_template:
		75	+ Accept: "text/event-stream"
		76	+ forward_header_whitelist:
		77	+ - x-user-id
		78	+adapters:
		79	+ - name: default_controller
		80	+ type: controller
		81	+ url_template: "http://mock-host/v1/agents/agent-a/conversations/{conversation_id}"
		82	+ - name: wf_wealth
		83	+ type: workflow
		84	+ workflow_id: wf_wealth
		85	+ intent: "理财推荐"
		86	+"""))
		87	+
		88	+ workflow = next(a for a in runner._adapters if a.name == "wf_wealth")
		89	+ assert workflow.url_template == "http://mock-host/v1/workflows/{workflow_id}/conversations/{conversation_id}"
		90	+ assert workflow.timeout == 45
		91	+ assert workflow.headers_template == {"Accept": "text/event-stream"}
		92	+ assert workflow.forward_header_whitelist == {"x-user-id"}
		93	+
		94	+
		95	+def test_workflow_adapter_overrides_workflow_defaults(write_yaml):
		96	+ runner = VersatileAdapterRunner(config_path=write_yaml("""
		97	+workflow_defaults:
		98	+ url_template: "http://default/workflows/{workflow_id}/conversations/{conversation_id}"
		99	+ timeout: 600
		100	+ headers_template:
		101	+ Accept: "text/event-stream"
		102	+ X-App-Code: "common"
		103	+ forward_header_whitelist:
		104	+ - x-user-id
		105	+adapters:
		106	+ - name: default_controller
		107	+ type: controller
		108	+ url_template: "http://mock-host/v1/agents/agent-a/conversations/{conversation_id}"
		109	+ - name: wf_special
		110	+ type: workflow
		111	+ workflow_id: wf_special
		112	+ intent: special
		113	+ url_template: "http://special/workflows/{workflow_id}/conversations/{conversation_id}"
		114	+ timeout: 120
		115	+ headers_template:
		116	+ X-App-Code: "special"
		117	+ X-Scene: "special-flow"
		118	+ forward_header_whitelist:
		119	+ - authorization
		120	+"""))
		121	+
		122	+ workflow = next(a for a in runner._adapters if a.name == "wf_special")
		123	+ assert workflow.url_template == "http://special/workflows/{workflow_id}/conversations/{conversation_id}"
		124	+ assert workflow.timeout == 120
		125	+ assert workflow.headers_template == {
		126	+ "Accept": "text/event-stream",
		127	+ "X-App-Code": "special",
		128	+ "X-Scene": "special-flow",
		129	+ }
		130	+ assert workflow.forward_header_whitelist == {"authorization"}
		131	+
		132	+
		133	+def test_workflow_defaults_do_not_apply_to_controller(write_yaml):
		134	+ runner = VersatileAdapterRunner(config_path=write_yaml("""
		135	+workflow_defaults:
		136	+ url_template: "http://default/workflows/{workflow_id}/conversations/{conversation_id}"
		137	+ timeout: 45
		138	+ headers_template:
		139	+ X-App-Code: "workflow-default"
		140	+adapters:
		141	+ - name: default_controller
		142	+ type: controller
		143	+ url_template: "http://controller/conversations/{conversation_id}"
		144	+ timeout: 60
		145	+ - name: wf_wealth
		146	+ type: workflow
		147	+ workflow_id: wf_wealth
		148	+ intent: "理财推荐"
		149	+"""))
		150	+
		151	+ controller = runner._controller_cfg
		152	+ assert controller.url_template == "http://controller/conversations/{conversation_id}"
		153	+ assert controller.timeout == 60
		154	+ assert controller.headers_template == {}
		155	+
		156	+
23	def test_load_yaml_creates_all_adapters(write_yaml):	157	def test_load_yaml_creates_all_adapters(write_yaml):
24	"""YAML 中 3 个 adapter 都被加载，名称类型正确。"""	158	"""YAML 中 3 个 adapter 都被加载，名称类型正确。"""
25	runner = VersatileAdapterRunner(config_path=write_yaml())	159	runner = VersatileAdapterRunner(config_path=write_yaml())
	@@ -109,6 +243,7 @@ def test_create_workflow_adapter_instance(runner):
109	assert isinstance(adapter, VersatileWorkflow)	243	assert isinstance(adapter, VersatileWorkflow)
110	# 内部字段	244	# 内部字段
111	assert adapter._workflow_id == "wf_knowledge_qa"	245	assert adapter._workflow_id == "wf_knowledge_qa"
		246	+ assert adapter._workflow_result_node == "WorkflowQAResponseNode"
112	assert adapter._timeout == 30	247	assert adapter._timeout == 30
113		248
114		249

Mapplications/versatile_adapter/tests/test_runner_stream.py+35-3

	@@ -189,8 +189,7 @@ async def test_controller_flow_exception_marks_failed(write_yaml):
189	sse_lines = [	189	sse_lines = [
190	'data: {"event":"exception","data":{"message":"运行时错误"}}',	190	'data: {"event":"exception","data":{"message":"运行时错误"}}',
191	]	191	]
192	- # 注意：workflow_result_node 未命中 → ctx.execution_result=None	192	+ # workflow_result_node 未命中，但 exception 会在流结束时产出 failed completed 事件
193	- # _on_stream_end: completed=True, execution_result=None → 返回空 []
194	with _patch_httpx(sse_lines):	193	with _patch_httpx(sse_lines):
195	events = []	194	events = []
196	async for ev in runner.run_async(	195	async for ev in runner.run_async(
	@@ -201,10 +200,15 @@ async def test_controller_flow_exception_marks_failed(write_yaml):
201	):	200	):
202	events.append(ev)	201	events.append(ev)
203		202
204	- # exception 帧本身作为 data_proxy 转发	203	+ # exception 帧本身作为 data_proxy 转发，流结束补 failed completed
205	data_events = [e for e in events if e.data_proxy is not None]	204	data_events = [e for e in events if e.data_proxy is not None]
206	assert len(data_events) == 1	205	assert len(data_events) == 1
207	assert '"event":"exception"' in data_events[0].data_proxy.raw_data	206	assert '"event":"exception"' in data_events[0].data_proxy.raw_data
		207	+ completed = [e for e in events if e.execution_completed is not None]
		208	+ assert len(completed) == 1
		209	+ assert completed[0].execution_completed.is_failed is True
		210	+ assert completed[0].execution_completed.result == ""
		211	+ assert '"event":"exception"' in completed[0].execution_completed.error_message
208		212
209		213
210	# ════════════════════════════════════════════════════════════════════	214	# ════════════════════════════════════════════════════════════════════
	@@ -239,6 +243,34 @@ async def test_workflow_intent_matched_skips_filtered_types(write_yaml):
239	assert len(data_events) == 2	243	assert len(data_events) == 2
240		244
241		245
		246	+@pytest.mark.asyncio
		247	+async def test_workflow_flow_with_workflow_result_node(write_yaml):
		248	+ """workflow adapter：命中自身 workflow_result_node 时提取结果并在 finish 后 completed。"""
		249	+ runner = VersatileAdapterRunner(config_path=write_yaml())
		250	+ sse_lines = [
		251	+ 'data: {"type":"text","data":{"content":"processing"}}',
		252	+ 'data: {"data":{"node_type":"QA","node_name":"WorkflowQAResponseNode","text":"工作流答案"}}',
		253	+ 'data: {"type":"finish","data":{"content":""}}',
		254	+ ]
		255	+
		256	+ with _patch_httpx(sse_lines):
		257	+ events = []
		258	+ async for ev in runner.run_async(
		259	+ target={"intent": "knowledge_qa", "conversation_id": "c-2"},
		260	+ headers={},
		261	+ params={},
		262	+ body={"custom_data": {}},
		263	+ ):
		264	+ events.append(ev)
		265	+
		266	+ data_events = [e for e in events if e.data_proxy is not None]
		267	+ assert len(data_events) == 1
		268	+ completed = [e for e in events if e.execution_completed is not None]
		269	+ assert len(completed) == 1
		270	+ assert completed[0].execution_completed.is_failed is False
		271	+ assert completed[0].execution_completed.result == "工作流答案"
		272	+
		273	+
242	@pytest.mark.asyncio	274	@pytest.mark.asyncio
243	async def test_workflow_id_matched_uses_workflow_url(write_yaml):	275	async def test_workflow_id_matched_uses_workflow_url(write_yaml):
244	"""workflow adapter：按 workflow_id 匹配；URL 应包含 wf_wealth。"""	276	"""workflow adapter：按 workflow_id 匹配；URL 应包含 wf_wealth。"""

Mapplications/versatile_adapter/tests/test_versatile_proxy.py+109-4

	@@ -231,9 +231,21 @@ class TestControllerProcessChunk:
231	events = ctrl._process_chunk(chunk, ctx)	231	events = ctrl._process_chunk(chunk, ctx)
232	assert ctx.completed is True	232	assert ctx.completed is True
233	assert ctx.is_failed is True	233	assert ctx.is_failed is True
		234	+ assert ctx.error_message == chunk
234	# exception 帧也需要转发前端	235	# exception 帧也需要转发前端
235	assert len(events) == 1	236	assert len(events) == 1
236		237
		238	+ @staticmethod
		239	+ def test_error_event_passes_through_without_terminating():
		240	+ ctrl = VersatileController("http://h/")
		241	+ ctx = VersatileStreamCtx()
		242	+ chunk = '{"event": "error", "data": {"message": "err"}}'
		243	+ events = ctrl._process_chunk(chunk, ctx)
		244	+ assert ctx.completed is False
		245	+ assert ctx.is_failed is False
		246	+ assert ctx.error_message == ""
		247	+ assert len(events) == 1
		248	+
237	@staticmethod	249	@staticmethod
238	def test_workflow_result_node_compact_json_also_matched():	250	def test_workflow_result_node_compact_json_also_matched():
239	"""紧凑格式（separators=(",", ":")）同样被命中。"""	251	"""紧凑格式（separators=(",", ":")）同样被命中。"""
	@@ -247,15 +259,30 @@ class TestControllerProcessChunk:
247	assert events == []	259	assert events == []
248	assert ctx.execution_result == "compact"	260	assert ctx.execution_result == "compact"
249		261
		262	+ @staticmethod
		263	+ def test_workflow_result_node_decodes_json_escaped_text():
		264	+ """workflow_result text 通过 JSON 解析提取，需还原引号、换行和 Unicode 转义。"""
		265	+ ctrl = VersatileController("http://h/", workflow_result_node="GXZQAResponseNode")
		266	+ ctx = VersatileStreamCtx()
		267	+ expected = '他说"你好"\n下一行'
		268	+ chunk = json.dumps(
		269	+ {"data": {"node_type": "QA", "node_name": "GXZQAResponseNode", "text": expected}},
		270	+ separators=(",", ":"),
		271	+ )
		272	+ events = ctrl._process_chunk(chunk, ctx)
		273	+ assert events == []
		274	+ assert ctx.execution_result == expected
		275	+
250	@staticmethod	276	@staticmethod
251	def test_workflow_result_node_unparseable_json_falls_through():	277	def test_workflow_result_node_unparseable_json_falls_through():
252	- """识别到 node_name 但 JSON 无法解析 → fall back 到 data_proxy 透传。"""	278	+ """识别到 node_name 但 JSON 无法解析时，fall back 到 data_proxy 透传。"""
253	ctrl = VersatileController("http://h/", workflow_result_node="GXZQAResponseNode")	279	ctrl = VersatileController("http://h/", workflow_result_node="GXZQAResponseNode")
254	ctx = VersatileStreamCtx()	280	ctx = VersatileStreamCtx()
255	chunk = '{not-valid-json "node_name":"GXZQAResponseNode"'	281	chunk = '{not-valid-json "node_name":"GXZQAResponseNode"'
256	events = ctrl._process_chunk(chunk, ctx)	282	events = ctrl._process_chunk(chunk, ctx)
257	assert len(events) == 1	283	assert len(events) == 1
258	assert events[0].data_proxy is not None	284	assert events[0].data_proxy is not None
		285	+ assert ctx.execution_result is None
259		286
260	@staticmethod	287	@staticmethod
261	def test_other_qa_node_passes_through():	288	def test_other_qa_node_passes_through():
	@@ -327,6 +354,18 @@ class TestControllerOnStreamEnd:
327	assert events[0].execution_completed.is_failed is True	354	assert events[0].execution_completed.is_failed is True
328	assert events[0].execution_completed.result == "exception-detail"	355	assert events[0].execution_completed.result == "exception-detail"
329		356
		357	+ @staticmethod
		358	+ def test_completed_failed_no_result_uses_error_message(ctrl):
		359	+ ctx = VersatileStreamCtx()
		360	+ ctx.completed = True
		361	+ ctx.is_failed = True
		362	+ ctx.error_message = "raw-error"
		363	+ events = ctrl._on_stream_end(ctx)
		364	+ assert len(events) == 1
		365	+ assert events[0].execution_completed.is_failed is True
		366	+ assert events[0].execution_completed.result == ""
		367	+ assert events[0].execution_completed.error_message == "raw-error"
		368	+
330	@staticmethod	369	@staticmethod
331	def test_completed_no_result_yields_nothing(ctrl):	370	def test_completed_no_result_yields_nothing(ctrl):
332	"""End 节点完成但未提取到 workflow_result：不产 execution_completed。"""	371	"""End 节点完成但未提取到 workflow_result：不产 execution_completed。"""
	@@ -352,6 +391,7 @@ class TestWorkflowProcessChunk:
352	ctx = VersatileStreamCtx()	391	ctx = VersatileStreamCtx()
353	chunk = f'{{"type":"{skip_type}","data":{{"content":"x"}}}}'	392	chunk = f'{{"type":"{skip_type}","data":{{"content":"x"}}}}'
354	assert wf._process_chunk(chunk, ctx) == []	393	assert wf._process_chunk(chunk, ctx) == []
		394	+ assert ctx.completed is (skip_type in {"finish", "runCompleted"})
355		395
356	@staticmethod	396	@staticmethod
357	@pytest.mark.parametrize("kept_type", ["rawData", "nodeType", "text", "answer", "message"])	397	@pytest.mark.parametrize("kept_type", ["rawData", "nodeType", "text", "answer", "message"])
	@@ -371,9 +411,73 @@ class TestWorkflowProcessChunk:
371	assert wf._process_chunk(chunk, ctx) == []	411	assert wf._process_chunk(chunk, ctx) == []
372		412
373	@staticmethod	413	@staticmethod
374	- def test_workflow_on_stream_end_yields_nothing(wf):	414	+ def test_workflow_on_stream_end_not_completed_yields_input_required(wf):
375	- """VersatileWorkflow._on_stream_end 始终返回 []（不产 execution_input_required）。"""	415	+ ctx = VersatileStreamCtx()
376	- assert wf._on_stream_end(VersatileStreamCtx()) == []	416	+ events = wf._on_stream_end(ctx)
		417	+ assert len(events) == 1
		418	+ assert events[0].execution_input_required is not None
		419	+
		420	+ @staticmethod
		421	+ def test_workflow_on_stream_end_completed_no_result_yields_nothing(wf):
		422	+ ctx = VersatileStreamCtx()
		423	+ ctx.completed = True
		424	+ assert wf._on_stream_end(ctx) == []
		425	+
		426	+ @staticmethod
		427	+ def test_workflow_error_event_passes_through_without_terminating(wf):
		428	+ ctx = VersatileStreamCtx()
		429	+ chunk = '{"event":"error","data":{"message":"workflow failed"}}'
		430	+ events = wf._process_chunk(chunk, ctx)
		431	+ assert ctx.completed is False
		432	+ assert ctx.is_failed is False
		433	+ assert ctx.error_message == ""
		434	+ assert len(events) == 1
		435	+
		436	+ completed = wf._on_stream_end(ctx)
		437	+ assert len(completed) == 1
		438	+ assert completed[0].execution_input_required is not None
		439	+
		440	+ @staticmethod
		441	+ def test_workflow_result_node_extracted():
		442	+ wf = VersatileWorkflow(
		443	+ "http://h/{workflow_id}/{conversation_id}",
		444	+ "wf-1",
		445	+ workflow_result_node="WorkflowQAResponseNode",
		446	+ )
		447	+ ctx = VersatileStreamCtx()
		448	+ chunk = json.dumps({
		449	+ "data": {
		450	+ "node_type": "QA",
		451	+ "node_name": "WorkflowQAResponseNode",
		452	+ "text": "workflow answer",
		453	+ }
		454	+ }, separators=(",", ":"))
		455	+ assert wf._process_chunk(chunk, ctx) == []
		456	+ assert ctx.execution_result == "workflow answer"
		457	+
		458	+ ctx.completed = True
		459	+ completed = wf._on_stream_end(ctx)
		460	+ assert len(completed) == 1
		461	+ assert completed[0].execution_completed.result == "workflow answer"
		462	+
		463	+ @staticmethod
		464	+ def test_workflow_result_node_decodes_json_escaped_text():
		465	+ wf = VersatileWorkflow(
		466	+ "http://h/{workflow_id}/{conversation_id}",
		467	+ "wf-1",
		468	+ workflow_result_node="WorkflowQAResponseNode",
		469	+ )
		470	+ ctx = VersatileStreamCtx()
		471	+ expected = '他说"你好"\n下一行'
		472	+ chunk = json.dumps({
		473	+ "data": {
		474	+ "node_type": "QA",
		475	+ "node_name": "WorkflowQAResponseNode",
		476	+ "text": expected,
		477	+ }
		478	+ }, separators=(",", ":"))
		479	+ assert wf._process_chunk(chunk, ctx) == []
		480	+ assert ctx.execution_result == expected
377		481
378		482
379	# ════════════════════════════════════════════════════════════════════	483	# ════════════════════════════════════════════════════════════════════
	@@ -388,6 +492,7 @@ class TestStreamCtx:
388	assert ctx.completed is False	492	assert ctx.completed is False
389	assert ctx.is_failed is False	493	assert ctx.is_failed is False
390	assert ctx.execution_result is None	494	assert ctx.execution_result is None
		495	+ assert ctx.error_message == ""
391		496
392	@staticmethod	497	@staticmethod
393	def test_multiple_chunks_accumulate():	498	def test_multiple_chunks_accumulate():

Mapplications/versatile_adapter/versatile_proxy.yaml.example+52-18

	@@ -3,8 +3,21 @@
3	# 部署路径（默认）：/etc/edpagent/config/versatile_proxy.yaml	3	# 部署路径（默认）：/etc/edpagent/config/versatile_proxy.yaml
4	# 若该文件不存在，Runner 会自动从 Settings（.env / 环境变量）生成唯一的 controller 配置。	4	# 若该文件不存在，Runner 会自动从 Settings（.env / 环境变量）生成唯一的 controller 配置。
5	#	5	#
6	-# adapters 列表可包含多个 adapter，运行时按 route 动态匹配：	6	+# 配置文件加载优先级：
7	-# - route 含 workflow_id / intent 且匹配到 workflow 配置 → VersatileWorkflow	7	+# 1. VersatileAdapterRunner(config_path=...) 显式参数
		8	+# 2. 环境变量 VERSATILE_PROXY_CONFIG_PATH
		9	+# 3. 部署默认路径 /etc/edpagent/config/versatile_proxy.yaml
		10	+# 4. 当前 versatile_adapter 目录下的 versatile_proxy.yaml（便于本地开发）
		11	+#
		12	+# workflow_defaults 仅对 type: workflow 的 adapter 生效：
		13	+# - workflow adapter 未配置的字段会从 workflow_defaults 继承
		14	+# - workflow adapter 显式配置优先于 workflow_defaults
		15	+# - headers_template 会浅合并，adapter 同名 header 覆盖默认值
		16	+# - forward_header_whitelist 不合并；adapter 配置后整体覆盖默认白名单
		17	+# - controller 通常只有一个，请在 controller adapter 内完整配置
		18	+#
		19	+# adapters 列表可包含多个 adapter，运行时按 target 动态匹配：
		20	+# - target 含 workflow_id / intent 且匹配到 workflow 配置 → VersatileWorkflow
8	# - 否则 → VersatileController（使用第一个 type=controller 的配置）	21	# - 否则 → VersatileController（使用第一个 type=controller 的配置）
9	#	22	#
10	# 各字段说明：	23	# 各字段说明：
	@@ -14,10 +27,24 @@
14	# timeout — HTTP 流式请求超时（秒），默认 600	27	# timeout — HTTP 流式请求超时（秒），默认 600
15	# headers_template — 调用 Versatile 时注入的默认请求头	28	# headers_template — 调用 Versatile 时注入的默认请求头
16	# forward_header_whitelist — 仅转发白名单中的外部请求头（小写）；不设置则转发全部	29	# forward_header_whitelist — 仅转发白名单中的外部请求头（小写）；不设置则转发全部
17	-# workflow_result_node — （仅 controller）SSE 流中匹配 node_type=="QA" 且	30	+# workflow_result_node — （controller/workflow 可选）SSE 流中匹配 node_type=="QA" 且
18	-# node_name==此值的帧，提取 text 作为 workflow_result	31	+# node_name==此值的帧，提取 text 作为 workflow_result；
19	-# workflow_id — （仅 workflow）route 中 workflow_id 精确匹配时命中	32	+# adapter 级配置，每个 adapter 可不同
20	-# intent — （仅 workflow）route 中 intent 匹配时命中	33	+# workflow_id — （仅 workflow）target 中 workflow_id 精确匹配时命中
		34	+# intent — （仅 workflow）target 中 intent 匹配时命中
		35	+
		36	+workflow_defaults:
		37	+ url_template: "https://versatile.example.com/workflows/{workflow_id}/conversations/{conversation_id}"
		38	+ timeout: 600
		39	+ headers_template:
		40	+ Accept: "application/json, text/event-stream"
		41	+ stream: "true"
		42	+ forward_header_whitelist:
		43	+ - x-user-id
		44	+ - x-project-id
		45	+ - cust-token
		46	+ - cust-userid
		47	+ - cookie
21		48
22	adapters:	49	adapters:
23	- name: default_controller	50	- name: default_controller
	@@ -28,21 +55,28 @@ adapters:
28	Accept: "application/json, text/event-stream"	55	Accept: "application/json, text/event-stream"
29	stream: "true"	56	stream: "true"
30	forward_header_whitelist:	57	forward_header_whitelist:
31	- - x-user-id	58	+ - x-user-id
32	- - x-project-id	59	+ - x-project-id
33	- - cust-token	60	+ - cust-token
34	- - cust-userid	61	+ - cust-userid
35	- - cookie	62	+ - cookie
36	workflow_result_node: ""	63	workflow_result_node: ""
37		64
38	- name: versatile_workflow_1	65	- name: versatile_workflow_1
39	type: workflow	66	type: workflow
40	intent: "knowledge_qa"	67	intent: "knowledge_qa"
41	- url_template: "https://versatile.example.com/workflows/{workflow_id}/conversations/{conversation_id}"
42	- timeout: 600
43	- headers_template:
44	- Accept: "application/json, text/event-stream"
45	- stream: "true"
46	- forward_header_whitelist:
47	- - x-trace-id
48	workflow_id: "wf_abc123"	68	workflow_id: "wf_abc123"
		69	+ workflow_result_node: "WorkflowQAResponseNode"
		70	+
		71	+ - name: versatile_workflow_2
		72	+ type: workflow
		73	+ intent: "special_flow"
		74	+ workflow_id: "wf_special"
		75	+ url_template: "https://special.example.com/workflows/{workflow_id}/conversations/{conversation_id}"
		76	+ timeout: 120
		77	+ headers_template:
		78	+ X-App-Code: "special"
		79	+ forward_header_whitelist:
		80	+ - authorization
		81	+ - x-tenant-id
		82	+ workflow_result_node: "SpecialQAResponseNode"