810 lines
36 KiB
Python
810 lines
36 KiB
Python
# ============================================================
|
||
# 多智能体编排 Mesh（moduleId: core-agent-mesh, 可重生 ✅）
|
||
# 演示与编排层：按需创建智能体、挂 Goal、按能力边界分发任务、
|
||
# 智能体消息互通、RUNNING 超期自动重建智能体、Goal 看门狗。
|
||
# 诚实边界：任务通过统一工具运行时走真实业务管线（分解/试排/校验/报告），
|
||
# P2/P3 确认卡必须由人工审批；审批完成后恢复 Goal 分发。
|
||
# 落盘：<aps_home>/data/agent_mesh.json（可用 APS_MESH_PATH 覆盖）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from server.timeutil import fmt_dt
|
||
|
||
logger = logging.getLogger("aps.mesh")
|
||
|
||
# ---------------- 角色模板（能力边界） ----------------
|
||
ROLE_TEMPLATES: dict[str, dict[str, Any]] = {
|
||
"data-analyst": {
|
||
"label": "数据解析智能体",
|
||
"capabilities": ["folder.analyze", "data.analyze", "master.query", "plan.trace"],
|
||
"desc": "解析工程目录/表格，校验排产齐备度",
|
||
},
|
||
"mrp-planner": {
|
||
"label": "MRP分解智能体",
|
||
"capabilities": ["order.decompose", "mrp.release"],
|
||
"desc": "按 BOM/供需净算产出采购/委外建议并下达",
|
||
},
|
||
"scheduler": {
|
||
"label": "柔性排产智能体",
|
||
"capabilities": ["flex.schedule", "flex.capacity", "flex.reschedule"],
|
||
"desc": "能力池组虚拟产线，闭环求解排产",
|
||
},
|
||
"verifier": {
|
||
"label": "结果校验智能体",
|
||
"capabilities": ["readiness.query", "conflict.list", "plan.trace"],
|
||
"desc": "复核排产版本、冲突、齐备度与计划追溯证据",
|
||
},
|
||
"reporter": {
|
||
"label": "报告智能体",
|
||
"capabilities": ["report.generate"],
|
||
"desc": "生成排产分析报告",
|
||
},
|
||
}
|
||
|
||
# 看门狗系统智能体（内置，不可删）
|
||
WATCHDOG_AGENT_ID = "agent-watchdog"
|
||
WATCHDOG_NAME = "Goal看门狗"
|
||
|
||
# Goal 模板：APS 排产全流程（任务带能力意图与依赖）
|
||
GOAL_TEMPLATES: dict[str, dict[str, Any]] = {
|
||
"aps-full-flow": {
|
||
"label": "APS 排产全流程",
|
||
"tasks": [
|
||
{"key": "analyze", "title": "解析工程目录数据", "intent": "data.analyze",
|
||
"role": "data-analyst", "params": {}, "dependsOn": []},
|
||
{"key": "readiness", "title": "复查排产齐备度", "intent": "readiness.query",
|
||
"role": "verifier", "params": {}, "dependsOn": ["analyze"]},
|
||
{"key": "decompose", "title": "执行 MRP 分解", "intent": "order.decompose",
|
||
"role": "mrp-planner", "params": {}, "dependsOn": ["readiness"]},
|
||
{"key": "schedule", "title": "生成一版试排", "intent": "flex.schedule",
|
||
"role": "scheduler", "params": {}, "dependsOn": ["decompose"]},
|
||
{"key": "verify", "title": "核对冲突与计划追溯", "intent": "conflict.list",
|
||
"role": "verifier", "params": {}, "dependsOn": ["schedule"]},
|
||
{"key": "report", "title": "生成排产分析报告", "intent": "report.generate",
|
||
"role": "reporter", "params": {"reportType": "plan"}, "dependsOn": ["verify"]},
|
||
],
|
||
},
|
||
}
|
||
|
||
_DEFAULT_STALL_TIMEOUT_SEC = 45.0
|
||
_DEFAULT_MAX_RETRIES = 1
|
||
|
||
|
||
def _now() -> str:
|
||
return fmt_dt(datetime.now())
|
||
|
||
|
||
def _uid(prefix: str) -> str:
|
||
return f"{prefix}-{uuid.uuid4().hex[:8]}"
|
||
|
||
|
||
def _mesh_path() -> str:
|
||
override = os.environ.get("APS_MESH_PATH")
|
||
if override:
|
||
return override
|
||
try:
|
||
from server.aps_home import aps_home
|
||
return str(aps_home() / "data" / "agent_mesh.json")
|
||
except Exception:
|
||
return os.path.join("server", "data", "agent_mesh.json")
|
||
|
||
|
||
class MeshStore:
|
||
"""多智能体编排仓：线程安全 + 原子落盘。"""
|
||
|
||
def __init__(self, path: str | None = None) -> None:
|
||
self.path = path or _mesh_path()
|
||
self._lock = threading.RLock()
|
||
self.data: dict[str, Any] = self._load()
|
||
self._ensure_watchdog()
|
||
|
||
# ---------------- 持久化 ----------------
|
||
def _load(self) -> dict[str, Any]:
|
||
try:
|
||
with open(self.path, encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
if isinstance(data, dict):
|
||
data.setdefault("agents", [])
|
||
data.setdefault("goals", [])
|
||
data.setdefault("messages", [])
|
||
data.setdefault("watchdog", {})
|
||
return data
|
||
except (OSError, json.JSONDecodeError):
|
||
pass
|
||
return {"agents": [], "goals": [], "messages": [], "watchdog": {}}
|
||
|
||
def save(self) -> None:
|
||
with self._lock:
|
||
os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
|
||
fd, tmp = tempfile.mkstemp(dir=os.path.dirname(self.path) or ".", suffix=".tmp")
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
||
json.dump(self.data, fh, ensure_ascii=False, indent=1)
|
||
os.replace(tmp, self.path)
|
||
except OSError:
|
||
try:
|
||
os.unlink(tmp)
|
||
except OSError:
|
||
pass
|
||
|
||
def _ensure_watchdog(self) -> None:
|
||
with self._lock:
|
||
if not any(a.get("id") == WATCHDOG_AGENT_ID for a in self.data["agents"]):
|
||
self.data["agents"].insert(0, {
|
||
"id": WATCHDOG_AGENT_ID, "name": WATCHDOG_NAME, "role": "watchdog",
|
||
"label": "Goal看门狗（系统）", "capabilities": ["mesh.watchdog"],
|
||
"status": "IDLE", "system": True, "createdBy": "SYSTEM",
|
||
"createdAt": _now(), "heartbeatAt": _now(),
|
||
})
|
||
wd = self.data["watchdog"]
|
||
wd.setdefault("enabled", True)
|
||
wd.setdefault("stallTimeoutSec", _DEFAULT_STALL_TIMEOUT_SEC)
|
||
wd.setdefault("maxRetries", _DEFAULT_MAX_RETRIES)
|
||
wd.setdefault("lastTickAt", None)
|
||
wd.setdefault("alerts", [])
|
||
|
||
# ---------------- 智能体 ----------------
|
||
def create_agent(self, name: str, role: str, capabilities: list[str] | None = None,
|
||
*, created_by: str = "USER", parent_id: str | None = None,
|
||
goal_id: str | None = None) -> dict[str, Any]:
|
||
tpl = ROLE_TEMPLATES.get(role, {})
|
||
caps = capabilities if capabilities is not None else list(tpl.get("capabilities") or [])
|
||
agent = {
|
||
"id": _uid("agent"), "name": name.strip() or tpl.get("label") or role,
|
||
"role": role, "label": tpl.get("label") or role,
|
||
"capabilities": caps, "status": "IDLE", "system": False,
|
||
"createdBy": created_by, "parentId": parent_id,
|
||
"parentGoalId": goal_id,
|
||
"goalIds": [goal_id] if goal_id else [],
|
||
"createdAt": _now(), "heartbeatAt": _now(),
|
||
"currentTaskId": None,
|
||
}
|
||
with self._lock:
|
||
self.data["agents"].append(agent)
|
||
self.save()
|
||
return agent
|
||
|
||
def _capable_idle(self, intent: str, role: str | None = None) -> dict[str, Any] | None:
|
||
for agent in self.data["agents"]:
|
||
if agent.get("system"):
|
||
continue
|
||
if role and agent.get("role") != role:
|
||
continue
|
||
if agent.get("status") == "IDLE" and intent in (agent.get("capabilities") or []):
|
||
return agent
|
||
return None
|
||
|
||
def _agent_for_role(self, role: str | None) -> str | None:
|
||
if not role:
|
||
return None
|
||
agent = next(
|
||
(item for item in self.data["agents"]
|
||
if not item.get("system") and item.get("role") == role),
|
||
None,
|
||
)
|
||
return str(agent.get("id")) if agent else None
|
||
|
||
def _spawn_for_intent(self, intent: str, *, created_by: str, goal_id: str,
|
||
role: str | None = None) -> dict[str, Any] | None:
|
||
"""没有能执行该意图的空闲智能体时，按角色模板自动新建（允许新建智能体）。"""
|
||
for candidate_role, tpl in ROLE_TEMPLATES.items():
|
||
if role and candidate_role != role:
|
||
continue
|
||
if intent in (tpl.get("capabilities") or []):
|
||
seq = sum(1 for a in self.data["agents"] if a.get("role") == candidate_role) + 1
|
||
agent = self.create_agent(
|
||
f"{tpl['label']}-{seq}", candidate_role,
|
||
created_by=created_by, parent_id=None, goal_id=goal_id,
|
||
)
|
||
self.post_message(
|
||
from_agent_id=WATCHDOG_AGENT_ID, to_agent_id=None, goal_id=goal_id,
|
||
kind="INFO",
|
||
text=f"没有空闲的「{tpl['label']}」，已自动新建 {agent['name']} 接管 {intent}。",
|
||
)
|
||
return agent
|
||
return None
|
||
|
||
# ---------------- Goal / 任务 ----------------
|
||
def create_goal(self, title: str, *, template: str | None = "aps-full-flow",
|
||
tasks: list[dict[str, Any]] | None = None,
|
||
created_by: str = "USER", world_key: str | None = None,
|
||
session_id: str | None = None, project_id: str | None = None,
|
||
source_sha256: str | None = None,
|
||
run_id: str | None = None) -> dict[str, Any]:
|
||
tpl = GOAL_TEMPLATES.get(template or "", {})
|
||
raw_tasks = tasks if tasks is not None else tpl.get("tasks") or []
|
||
goal_id = _uid("goal")
|
||
run_id = str(run_id or _uid("run"))
|
||
task_rows = []
|
||
for spec in raw_tasks:
|
||
task_rows.append({
|
||
"id": _uid("task"), "goalId": goal_id,
|
||
"key": spec.get("key") or spec.get("intent"),
|
||
"title": spec.get("title") or spec.get("intent"),
|
||
"intent": spec["intent"], "role": spec.get("role"),
|
||
"params": dict(spec.get("params") or {}),
|
||
"dependsOn": list(spec.get("dependsOn") or []),
|
||
"status": "PENDING", "assigneeAgentId": None, "assigneeName": None,
|
||
"attempts": 0, "result": None, "error": None,
|
||
"startedAt": None, "finishedAt": None,
|
||
})
|
||
goal = {
|
||
"id": goal_id, "title": title.strip() or tpl.get("label") or "未命名 Goal",
|
||
"template": template, "status": "ACTIVE",
|
||
"createdBy": created_by, "worldKey": world_key,
|
||
"sessionId": session_id, "projectId": project_id,
|
||
"sourceSha256": source_sha256, "runId": run_id,
|
||
"createdAt": _now(), "tasks": task_rows,
|
||
"watchdog": {"stallTimeoutSec": float(self.data["watchdog"].get("stallTimeoutSec") or _DEFAULT_STALL_TIMEOUT_SEC),
|
||
"maxRetries": int(self.data["watchdog"].get("maxRetries") or _DEFAULT_MAX_RETRIES)},
|
||
}
|
||
with self._lock:
|
||
self.data["goals"].append(goal)
|
||
self.save()
|
||
sub_agents = self._ensure_goal_agents(goal)
|
||
goal["subAgentIds"] = [agent["id"] for agent in sub_agents]
|
||
self.save()
|
||
self.post_message(from_agent_id=WATCHDOG_AGENT_ID, to_agent_id=None, goal_id=goal_id,
|
||
run_id=run_id, kind="INFO",
|
||
text=(f"Goal「{goal['title']}」已挂载（run {run_id}），"
|
||
f"已拉起 {len(sub_agents)} 个子智能体，"
|
||
f"{len(task_rows)} 个任务进入看门狗监控。"))
|
||
return goal
|
||
|
||
def _ensure_goal_agents(self, goal: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Create one bounded sub-agent for every distinct task role in the Goal."""
|
||
roles = list(dict.fromkeys(
|
||
str(task.get("role") or "data-analyst") for task in goal.get("tasks") or []))
|
||
agents: list[dict[str, Any]] = []
|
||
for role in roles:
|
||
tpl = ROLE_TEMPLATES.get(role, {})
|
||
agent = next(
|
||
(item for item in self.data["agents"]
|
||
if not item.get("system") and item.get("role") == role
|
||
and goal["id"] in (item.get("goalIds") or [])),
|
||
None,
|
||
)
|
||
if agent is None:
|
||
seq = sum(1 for item in self.data["agents"] if item.get("role") == role) + 1
|
||
agent = self.create_agent(
|
||
f"{tpl.get('label') or role}-{seq}", role,
|
||
created_by=f"GOAL:{goal['id']}", goal_id=goal["id"],
|
||
)
|
||
elif goal["id"] not in (agent.get("goalIds") or []):
|
||
agent.setdefault("goalIds", []).append(goal["id"])
|
||
self.save()
|
||
agents.append(agent)
|
||
return agents
|
||
|
||
def get_goal(self, goal_id: str) -> dict[str, Any] | None:
|
||
return next((g for g in self.data["goals"] if g.get("id") == goal_id), None)
|
||
|
||
# ---------------- 消息总线 ----------------
|
||
def post_message(self, *, from_agent_id: str, to_agent_id: str | None,
|
||
goal_id: str | None, kind: str, text: str,
|
||
run_id: str | None = None,
|
||
evidence_refs: list[str] | None = None) -> dict[str, Any]:
|
||
msg = {
|
||
"id": _uid("msg"), "fromAgentId": from_agent_id,
|
||
"fromName": self._agent_name(from_agent_id),
|
||
"toAgentId": to_agent_id, "toName": self._agent_name(to_agent_id) if to_agent_id else "全体",
|
||
"goalId": goal_id, "runId": run_id, "kind": kind, "text": text,
|
||
"evidenceRefs": list(evidence_refs or []), "createdAt": _now(),
|
||
}
|
||
with self._lock:
|
||
self.data["messages"].append(msg)
|
||
self.data["messages"] = self.data["messages"][-300:]
|
||
self.save()
|
||
return msg
|
||
|
||
def _agent_name(self, agent_id: str | None) -> str:
|
||
if not agent_id:
|
||
return "?"
|
||
agent = next((a for a in self.data["agents"] if a.get("id") == agent_id), None)
|
||
return (agent or {}).get("name") or agent_id
|
||
|
||
# ---------------- 看门狗 ----------------
|
||
def watchdog_tick(self) -> dict[str, Any]:
|
||
"""扫描 RUNNING 超期任务：标记 STALLED → 自动重试/失败 → Goal 升级。幂等可频繁调用。"""
|
||
wd = self.data["watchdog"]
|
||
wd["lastTickAt"] = _now()
|
||
if not wd.get("enabled", True):
|
||
self.save()
|
||
return {"enabled": False, "stalled": 0}
|
||
stalled = 0
|
||
now = time.time()
|
||
for goal in self.data["goals"]:
|
||
if goal.get("status") not in ("ACTIVE", "ATTENTION"):
|
||
continue
|
||
gwd = goal.get("watchdog") or {}
|
||
timeout = float(gwd.get("stallTimeoutSec") or _DEFAULT_STALL_TIMEOUT_SEC)
|
||
max_retries = int(gwd.get("maxRetries") or _DEFAULT_MAX_RETRIES)
|
||
for task in goal.get("tasks") or []:
|
||
if task.get("status") != "RUNNING" or not task.get("startedAt"):
|
||
continue
|
||
try:
|
||
started = datetime.strptime(task["startedAt"], "%Y-%m-%d %H:%M:%S").timestamp()
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if now - started <= timeout:
|
||
continue
|
||
stalled += 1
|
||
attempts = int(task.get("attempts") or 0)
|
||
assignee = task.get("assigneeName") or "未分配"
|
||
if attempts <= max_retries:
|
||
task["status"] = "PENDING"
|
||
task["startedAt"] = None
|
||
task["assigneeAgentId"] = None
|
||
task["assigneeName"] = None
|
||
self.post_message(
|
||
from_agent_id=WATCHDOG_AGENT_ID, to_agent_id=None, goal_id=goal["id"],
|
||
kind="ALERT",
|
||
text=f"任务「{task['title']}」在 {assignee} 上超过 {timeout:.0f}s 未完成，已收回并重新排队（第 {attempts} 次重试）。")
|
||
else:
|
||
task["status"] = "FAILED"
|
||
task["error"] = f"看门狗：超过 {timeout:.0f}s 未完成且重试耗尽"
|
||
task["finishedAt"] = _now()
|
||
goal["status"] = "ATTENTION"
|
||
self.post_message(
|
||
from_agent_id=WATCHDOG_AGENT_ID, to_agent_id=None, goal_id=goal["id"],
|
||
kind="ALERT",
|
||
text=f"任务「{task['title']}」重试 {max_retries} 次仍超时，标记 FAILED，Goal「{goal['title']}」升级为 ATTENTION。")
|
||
if stalled:
|
||
wd.setdefault("alerts", []).append({"at": _now(), "stalled": stalled})
|
||
wd["alerts"] = wd["alerts"][-50:]
|
||
self.save()
|
||
return {"enabled": True, "stalled": stalled, "lastTickAt": wd["lastTickAt"]}
|
||
|
||
def configure_watchdog(self, *, stall_timeout_sec: float | None = None,
|
||
max_retries: int | None = None, enabled: bool | None = None) -> dict[str, Any]:
|
||
wd = self.data["watchdog"]
|
||
if stall_timeout_sec is not None:
|
||
wd["stallTimeoutSec"] = max(5.0, float(stall_timeout_sec))
|
||
if max_retries is not None:
|
||
wd["maxRetries"] = max(0, int(max_retries))
|
||
if enabled is not None:
|
||
wd["enabled"] = bool(enabled)
|
||
self.save()
|
||
return wd
|
||
|
||
# ---------------- 快照 ----------------
|
||
def reset(self) -> dict[str, Any]:
|
||
"""恢复起始状态：清空 Goal/消息/告警，只保留系统看门狗智能体。"""
|
||
with self._lock:
|
||
self.data["agents"] = [a for a in self.data["agents"] if a.get("system")]
|
||
self.data["goals"] = []
|
||
self.data["messages"] = []
|
||
self.data["watchdog"]["alerts"] = []
|
||
self.save()
|
||
return self.snapshot()
|
||
|
||
def snapshot(self) -> dict[str, Any]:
|
||
with self._lock:
|
||
return json.loads(json.dumps({
|
||
"agents": self.data["agents"],
|
||
"goals": self.data["goals"],
|
||
"messages": self.data["messages"][-100:],
|
||
"watchdog": self.data["watchdog"],
|
||
"roleTemplates": [
|
||
{"role": role, **{k: v for k, v in tpl.items()}}
|
||
for role, tpl in ROLE_TEMPLATES.items()
|
||
],
|
||
"goalTemplates": [
|
||
{"key": key, "label": tpl["label"], "taskCount": len(tpl["tasks"])}
|
||
for key, tpl in GOAL_TEMPLATES.items()
|
||
],
|
||
}))
|
||
|
||
|
||
_MESH: MeshStore | None = None
|
||
_MESH_LOCK = threading.Lock()
|
||
|
||
|
||
def get_mesh() -> MeshStore:
|
||
global _MESH
|
||
with _MESH_LOCK:
|
||
if _MESH is None:
|
||
_MESH = MeshStore()
|
||
return _MESH
|
||
|
||
|
||
# ---------------- 分发执行（真实业务管线） ----------------
|
||
_DISPATCH_THREADS: dict[str, threading.Thread] = {}
|
||
|
||
|
||
def dispatch_goal(goal_id: str, *, session_id: str | None = None) -> dict[str, Any]:
|
||
"""异步分发 Goal：按依赖序把任务分给有能力的空闲智能体，必要时自动新建。"""
|
||
mesh = get_mesh()
|
||
goal = mesh.get_goal(goal_id)
|
||
if goal is None:
|
||
raise ValueError(f"Goal 不存在：{goal_id}")
|
||
# Goal 绑定的项目世界与当前活动世界不一致时先切换，保证任务打在正确的数据边界上
|
||
goal_world = (goal.get("worldKey") or "").strip()
|
||
if goal_world:
|
||
try:
|
||
from server.state.store import get_store, switch_store
|
||
if getattr(get_store(), "world_key", "default") != goal_world:
|
||
switch_store(goal_world)
|
||
except Exception as exc: # noqa: BLE001 - 项目边界不可用必须失败关闭
|
||
logger.exception("mesh: switch store to %s failed", goal_world)
|
||
raise RuntimeError(f"Goal 绑定的项目世界不可用：{goal_world}") from exc
|
||
eff_session = session_id or (goal.get("sessionId") or "") or f"mesh-{goal_id}"
|
||
try:
|
||
from server.auth.context import get_identity
|
||
identity = get_identity()
|
||
except Exception: # noqa: BLE001 - 无请求身份时用系统兜底
|
||
identity = None
|
||
running = DispatchThread(goal_id=goal_id, session_id=eff_session,
|
||
world_key=goal_world or None, identity=identity)
|
||
if goal_id in _DISPATCH_THREADS and _DISPATCH_THREADS[goal_id].is_alive():
|
||
return {"message": "该 Goal 已在分发执行中", "goalId": goal_id, "alreadyRunning": True}
|
||
_DISPATCH_THREADS[goal_id] = running
|
||
running.start()
|
||
return {"message": f"Goal「{goal['title']}」已开始分发执行", "goalId": goal_id, "alreadyRunning": False}
|
||
|
||
|
||
def resume_goal_after_confirmation(
|
||
confirm_id: str, *, approve: bool, actor: str,
|
||
) -> dict[str, Any]:
|
||
"""把人工审批结果回写到 Mesh，并只在批准后恢复 Goal 分发。
|
||
|
||
该函数只消费 Mesh 自己登记过的 confirmationId；普通业务确认卡不会命中，
|
||
因此不会改变既有审批路径。重复调用通过任务内确认历史实现幂等。
|
||
"""
|
||
requested = str(confirm_id or "").strip()
|
||
if not requested:
|
||
return {"matched": False, "resolved": False, "resumed": False}
|
||
|
||
mesh = get_mesh()
|
||
actor_name = str(actor or "unknown")
|
||
matched_goal: dict[str, Any] | None = None
|
||
matched_task: dict[str, Any] | None = None
|
||
|
||
with mesh._lock:
|
||
for goal in mesh.data.get("goals") or []:
|
||
for task in goal.get("tasks") or []:
|
||
pending_ids = [str(item) for item in (task.get("confirmationIds") or [])]
|
||
resolutions = task.get("confirmationResolutions") or []
|
||
if requested in pending_ids:
|
||
if task.get("status") != "AWAITING_APPROVAL":
|
||
return {
|
||
"matched": True,
|
||
"resolved": False,
|
||
"resumed": False,
|
||
"reason": "TASK_NOT_AWAITING_APPROVAL",
|
||
"goalId": goal.get("id"),
|
||
"taskId": task.get("id"),
|
||
"taskStatus": task.get("status"),
|
||
}
|
||
matched_goal, matched_task = goal, task
|
||
break
|
||
if any(str(item.get("confirmId") or "") == requested for item in resolutions):
|
||
return {
|
||
"matched": True,
|
||
"resolved": True,
|
||
"resumed": False,
|
||
"reason": "ALREADY_RESOLVED",
|
||
"goalId": goal.get("id"),
|
||
"taskId": task.get("id"),
|
||
"taskStatus": task.get("status"),
|
||
"alreadyRunning": bool(
|
||
goal.get("id") in _DISPATCH_THREADS
|
||
and _DISPATCH_THREADS[goal["id"]].is_alive()
|
||
),
|
||
}
|
||
if matched_task is not None:
|
||
break
|
||
|
||
if matched_goal is None or matched_task is None:
|
||
return {"matched": False, "resolved": False, "resumed": False}
|
||
|
||
now = _now()
|
||
matched_task["confirmationIds"] = []
|
||
matched_task["finishedAt"] = now
|
||
resolutions = matched_task.setdefault("confirmationResolutions", [])
|
||
resolutions.append({
|
||
"confirmId": requested,
|
||
"approve": bool(approve),
|
||
"actor": actor_name,
|
||
"at": now,
|
||
})
|
||
if approve:
|
||
matched_task["status"] = "DONE"
|
||
matched_task["result"] = f"人工已批准 {requested} by {actor_name}"
|
||
matched_task["error"] = None
|
||
matched_goal["status"] = "ACTIVE"
|
||
mesh.post_message(
|
||
from_agent_id=WATCHDOG_AGENT_ID,
|
||
to_agent_id=mesh._agent_for_role(matched_task.get("role")),
|
||
goal_id=matched_goal.get("id"),
|
||
run_id=matched_goal.get("runId"),
|
||
kind="RESPONSE",
|
||
evidence_refs=[f"confirm:{requested}"],
|
||
text=(f"人工批准 {requested}（{actor_name}），任务"
|
||
f"「{matched_task.get('title')}」已完成，Goal 继续分发。"),
|
||
)
|
||
else:
|
||
matched_task["status"] = "FAILED"
|
||
matched_task["error"] = f"人工驳回 {requested} by {actor_name}"
|
||
matched_goal["status"] = "ATTENTION"
|
||
mesh.post_message(
|
||
from_agent_id=WATCHDOG_AGENT_ID,
|
||
to_agent_id=None,
|
||
goal_id=matched_goal.get("id"),
|
||
run_id=matched_goal.get("runId"),
|
||
kind="ALERT",
|
||
evidence_refs=[f"confirm:{requested}"],
|
||
text=(f"人工驳回 {requested}（{actor_name}），任务"
|
||
f"「{matched_task.get('title')}」已失败，Goal 停止分发。"),
|
||
)
|
||
mesh.save()
|
||
|
||
if not approve:
|
||
return {
|
||
"matched": True,
|
||
"resolved": True,
|
||
"resumed": False,
|
||
"goalId": matched_goal.get("id"),
|
||
"taskId": matched_task.get("id"),
|
||
"taskStatus": "FAILED",
|
||
}
|
||
|
||
dispatch_result = dispatch_goal(
|
||
str(matched_goal["id"]), session_id=matched_goal.get("sessionId"),
|
||
)
|
||
already_running = bool(dispatch_result.get("alreadyRunning"))
|
||
return {
|
||
"matched": True,
|
||
"resolved": True,
|
||
"resumed": True,
|
||
"started": not already_running,
|
||
"alreadyRunning": already_running,
|
||
"goalId": matched_goal.get("id"),
|
||
"taskId": matched_task.get("id"),
|
||
"taskStatus": "DONE",
|
||
"dispatch": dispatch_result,
|
||
}
|
||
|
||
|
||
class DispatchThread(threading.Thread):
|
||
def __init__(self, *, goal_id: str, session_id: str, world_key: str | None = None,
|
||
identity: Any = None) -> None:
|
||
super().__init__(daemon=True, name=f"mesh-dispatch-{goal_id}")
|
||
self.goal_id = goal_id
|
||
self.session_id = session_id
|
||
self.world_key = world_key
|
||
self.identity = identity
|
||
|
||
def run(self) -> None: # noqa: D102
|
||
mesh = get_mesh()
|
||
goal = mesh.get_goal(self.goal_id)
|
||
if goal is None:
|
||
return
|
||
token = None
|
||
if self.identity is not None:
|
||
try:
|
||
from server.auth.context import bind_identity
|
||
token = bind_identity(self.identity) # 项目仓按用户隔离：绑回创建时身份
|
||
except Exception: # noqa: BLE001
|
||
token = None
|
||
try:
|
||
while True:
|
||
task = self._next_ready(goal)
|
||
if task is None:
|
||
break
|
||
self._run_task(mesh, goal, task)
|
||
self._finish_goal(mesh, goal)
|
||
except Exception as exc: # noqa: BLE001 - 编排兜底：任何异常落 Goal，不炸线程
|
||
logger.exception("mesh dispatch failed")
|
||
goal["status"] = "ATTENTION"
|
||
mesh.post_message(from_agent_id=WATCHDOG_AGENT_ID, to_agent_id=None,
|
||
goal_id=self.goal_id, kind="ALERT",
|
||
text=f"分发执行异常：{exc}")
|
||
mesh.save()
|
||
finally:
|
||
if token is not None:
|
||
try:
|
||
from server.auth.context import reset_identity
|
||
reset_identity(token)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
def _next_ready(self, goal: dict[str, Any]) -> dict[str, Any] | None:
|
||
done_keys = {t["key"] for t in goal["tasks"] if t.get("status") == "DONE"}
|
||
for task in goal["tasks"]:
|
||
if task.get("status") == "PENDING" and all(dep in done_keys for dep in task.get("dependsOn") or []):
|
||
return task
|
||
# 还有 RUNNING（看门狗收回中会重新 PENDING）则等待
|
||
if any(t.get("status") == "RUNNING" for t in goal["tasks"]):
|
||
time.sleep(1.0)
|
||
return self._next_ready(goal)
|
||
return None
|
||
|
||
def _run_task(self, mesh: MeshStore, goal: dict[str, Any], task: dict[str, Any]) -> None:
|
||
intent = task["intent"]
|
||
role = task.get("role")
|
||
agent = mesh._capable_idle(intent, role) or mesh._spawn_for_intent(
|
||
intent, created_by=f"GOAL:{goal['id']}", goal_id=goal["id"], role=role)
|
||
if agent is None:
|
||
task["status"] = "FAILED"
|
||
task["error"] = f"没有任何角色模板能执行 {intent}"
|
||
task["finishedAt"] = _now()
|
||
goal["status"] = "ATTENTION"
|
||
mesh.save()
|
||
return
|
||
task["status"] = "RUNNING"
|
||
task["assigneeAgentId"] = agent["id"]
|
||
task["assigneeName"] = agent["name"]
|
||
task["attempts"] = int(task.get("attempts") or 0) + 1
|
||
task["startedAt"] = _now()
|
||
agent["status"] = "RUNNING"
|
||
agent["currentTaskId"] = task["id"]
|
||
agent["heartbeatAt"] = _now()
|
||
mesh.save()
|
||
mesh.post_message(from_agent_id=WATCHDOG_AGENT_ID, to_agent_id=agent["id"],
|
||
goal_id=goal["id"], run_id=goal.get("runId"), kind="REQUEST",
|
||
text=f"分发任务「{task['title']}」（{intent}），第 {task['attempts']} 次尝试。")
|
||
try:
|
||
text, blocks = _execute_intent(self.session_id, intent, task.get("params") or {}, agent,
|
||
world_key=self.world_key, goal=goal, task=task)
|
||
pending_cards = [
|
||
str((block.get("props") or {}).get("confirmId"))
|
||
for block in blocks or []
|
||
if (block.get("props") or {}).get("confirmId")
|
||
]
|
||
if pending_cards:
|
||
task["status"] = "AWAITING_APPROVAL"
|
||
task["confirmationIds"] = pending_cards
|
||
task["result"] = (text or "")[:600]
|
||
task["finishedAt"] = None
|
||
agent["status"] = "IDLE"
|
||
agent["currentTaskId"] = None
|
||
agent["heartbeatAt"] = _now()
|
||
goal["status"] = "ATTENTION"
|
||
mesh.save()
|
||
mesh.post_message(
|
||
from_agent_id=agent["id"], to_agent_id=WATCHDOG_AGENT_ID,
|
||
goal_id=goal["id"], run_id=goal.get("runId"), kind="ALERT",
|
||
evidence_refs=[f"confirm:{cid}" for cid in pending_cards],
|
||
text=(f"「{task['title']}」需要人工审批，执行智能体不会自动批准。"
|
||
f"待确认：{', '.join(pending_cards)}"),
|
||
)
|
||
return
|
||
# 排产/分解被阻断（fail-closed 文案）不算任务成功：落 FAILED 交看门狗升级
|
||
if "没有可下达的草稿建议单" in text and _released_supply_exists(self.world_key):
|
||
text = "供应建议单此前已下达（RELEASED 在账），本次幂等跳过。"
|
||
elif "被阻断" in text or "无法生成" in text or "没有可下达" in text:
|
||
raise RuntimeError(text.strip().splitlines()[0][:200])
|
||
task["status"] = "DONE"
|
||
task["result"] = (text or "")[:600]
|
||
task["finishedAt"] = _now()
|
||
agent["status"] = "IDLE"
|
||
agent["currentTaskId"] = None
|
||
agent["heartbeatAt"] = _now()
|
||
mesh.save()
|
||
next_task = self._peek_next_task(goal, task)
|
||
nxt = (next_task or {}).get("title")
|
||
summary_line = next(
|
||
(ln for ln in reversed((text or "").splitlines()) if ln.strip()), "")
|
||
next_agent = mesh._agent_for_role((next_task or {}).get("role"))
|
||
mesh.post_message(from_agent_id=agent["id"], to_agent_id=next_agent,
|
||
goal_id=goal["id"], run_id=goal.get("runId"), kind="RESPONSE",
|
||
evidence_refs=[f"mesh-task:{task['id']}"],
|
||
text=f"「{task['title']}」完成 ✅ {summary_line[:80]}"
|
||
+ (f"，@{nxt} 请接棒。" if nxt else "，Goal 任务全部就绪。"))
|
||
except Exception as exc: # noqa: BLE001 - 任务失败落状态，交给看门狗/重试
|
||
logger.exception("mesh task failed: %s", intent)
|
||
task["status"] = "FAILED"
|
||
task["error"] = str(exc)[:300]
|
||
task["finishedAt"] = _now()
|
||
agent["status"] = "IDLE"
|
||
agent["currentTaskId"] = None
|
||
goal["status"] = "ATTENTION"
|
||
mesh.save()
|
||
mesh.post_message(from_agent_id=agent["id"], to_agent_id=WATCHDOG_AGENT_ID,
|
||
goal_id=goal["id"], run_id=goal.get("runId"), kind="ALERT",
|
||
text=f"「{task['title']}」执行失败：{exc}")
|
||
|
||
@staticmethod
|
||
def _peek_next_task(goal: dict[str, Any], just_done: dict[str, Any]) -> dict[str, Any] | None:
|
||
done_keys = {t["key"] for t in goal["tasks"] if t.get("status") == "DONE"}
|
||
for task in goal["tasks"]:
|
||
if task.get("status") == "PENDING" and all(dep in done_keys for dep in task.get("dependsOn") or []):
|
||
return task
|
||
return None
|
||
|
||
def _finish_goal(self, mesh: MeshStore, goal: dict[str, Any]) -> None:
|
||
statuses = {t.get("status") for t in goal["tasks"]}
|
||
if statuses and statuses <= {"DONE"}:
|
||
goal["status"] = "ACHIEVED"
|
||
mesh.post_message(from_agent_id=WATCHDOG_AGENT_ID, to_agent_id=None,
|
||
goal_id=goal["id"], kind="INFO",
|
||
text=f"Goal「{goal['title']}」全部任务完成 ✅ 看门狗结案。")
|
||
elif goal.get("status") != "ATTENTION":
|
||
goal["status"] = "ATTENTION"
|
||
mesh.save()
|
||
|
||
|
||
def _execute_intent(session_id: str, intent: str, params: dict[str, Any],
|
||
agent: dict[str, Any], *, world_key: str | None = None,
|
||
goal: dict[str, Any] | None = None,
|
||
task: dict[str, Any] | None = None) -> tuple[str, list[dict[str, Any]]]:
|
||
"""通过真实业务管线执行意图（与对话同一入口 handle_intent）。
|
||
|
||
world_key 指定时先切到 Goal 绑定的项目世界：全局活动世界可能被
|
||
前端工作区同步抢占，必须在每个任务执行当下重新钉住数据边界。
|
||
"""
|
||
from server.agent_core.tool_runtime import run_tool_async
|
||
from server.contracts import IntentResult
|
||
from server.state.store import get_store, switch_store
|
||
|
||
if world_key and getattr(get_store(), "world_key", "default") != world_key:
|
||
switch_store(world_key)
|
||
store = get_store()
|
||
enriched = dict(params or {})
|
||
if goal:
|
||
enriched["_mesh"] = {
|
||
"goalId": goal.get("id"),
|
||
"runId": goal.get("runId"),
|
||
"worldKey": goal.get("worldKey"),
|
||
"sessionId": goal.get("sessionId"),
|
||
"sourceSha256": goal.get("sourceSha256"),
|
||
"projectId": goal.get("projectId"),
|
||
"fromAgentId": agent.get("id"),
|
||
"taskId": (task or {}).get("id"),
|
||
}
|
||
reply = asyncio.run(run_tool_async(
|
||
store, session_id,
|
||
IntentResult(intent=intent, params=enriched, confidence=1.0, source="LLM"),
|
||
actor=f"mesh:{agent['name']}",
|
||
))
|
||
blocks = [b.model_dump() for b in (reply.blocks or [])]
|
||
return reply.text or "", blocks
|
||
|
||
|
||
# ---------------- 看门狗后台循环 ----------------
|
||
_WATCHDOG_STARTED = False
|
||
|
||
|
||
def start_watchdog_loop(interval_sec: float = 3.0) -> None:
|
||
"""启动看门狗后台节拍（幂等）。桌面/服务启动时调用一次。"""
|
||
global _WATCHDOG_STARTED
|
||
if _WATCHDOG_STARTED:
|
||
return
|
||
_WATCHDOG_STARTED = True
|
||
|
||
def _loop() -> None:
|
||
while True:
|
||
try:
|
||
get_mesh().watchdog_tick()
|
||
except Exception: # noqa: BLE001 - 看门狗自身绝不能崩
|
||
logger.exception("mesh watchdog tick failed")
|
||
time.sleep(interval_sec)
|
||
|
||
threading.Thread(target=_loop, daemon=True, name="mesh-watchdog").start()
|
||
|
||
def _released_supply_exists(world_key: str | None) -> bool:
|
||
"""目标世界是否已有非草稿（已下达）采购/委外单：幂等判断依据。"""
|
||
try:
|
||
from server.state.store import get_store, switch_store
|
||
if world_key and getattr(get_store(), "world_key", "default") != world_key:
|
||
switch_store(world_key)
|
||
data = get_store().data
|
||
for row in (data.get("purchaseOrders") or []) + (data.get("outsourceOrders") or []):
|
||
if isinstance(row, dict) and row.get("status") and row.get("status") != "DRAFT":
|
||
return True
|
||
except Exception: # noqa: BLE001 - 判断失败保守返回 False（维持失败语义）
|
||
pass
|
||
return False
|