feat(backend): push manager_worker coop-group events and structure ClawMind execution_graph

src/backend/src/agentclaw/community/adapters/http/task/translator.py

@@ -6,11 +6,12 @@
 """
 from __future__ import annotations
 
+import json
 from dataclasses import dataclass
 from typing import Any, Literal
 
 from agentclaw.community.core.errors import NotFound
-from agentclaw.community.core.task.domain.models import TaskCallbackData
+from agentclaw.community.core.task.domain.models import Status, TaskCallbackData
 from agentclaw.community.core.task.task_runner.callback_correlation import (
     CallbackCorrelationRegistry,
 )
@@ -148,13 +149,188 @@ def translate_claw_mind(raw: dict, disposition: Literal["start", "result"]) -> T
             "workflow_source": "claw_mind",
             "workflow_instance_id": (flow_runs.get("origin_session_id") or ""),
             "status": low_status,
-            "execution_graph": ext,
+            "execution_graph": _build_claw_mind_execution_graph(ext, run_status=low_status),
             "_raw_callback_body": raw,
             "result": result,
         }),
     )
 
 
+# ===== ClawMind ext_info → TaskExecutionGraph(graph_to_dict 形状)执行图快照 =====
+# adapter 层不 import repository serializers(避免跨层),此处手写与
+# core/task/repository/serializers.py:graph_to_dict 同型的 dict;若 graph_to_dict
+# 形状变更需同步。落 task_callback.execution_graph 只读投影。
+
+# 底层 status → TaskExecutionGraph 7 态:succeeded/done→DONE、failed→FAILED、
+# cancelled/aborted→CANCELLED、running/started→RUNNING,余缺省 PENDING。
+_CLAW_MIND_TO_TASK_STATUS: dict[str, Status] = {
+    "succeeded": Status.DONE, "completed": Status.DONE, "done": Status.DONE,
+    "node_succeeded": Status.DONE, "success": Status.DONE,
+    "failed": Status.FAILED, "node_failed": Status.FAILED,
+    "cancelled": Status.CANCELLED, "canceled": Status.CANCELLED, "aborted": Status.CANCELLED,
+    "running": Status.RUNNING, "started": Status.RUNNING, "in_progress": Status.RUNNING,
+    "active": Status.RUNNING,
+    "pending": Status.PENDING, "queued": Status.PENDING, "waiting": Status.PENDING,
+    "planning": Status.PLANNING,
+}
+
+
+def _claw_mind_status_to_task(low_status: Any) -> Status:
+    return _CLAW_MIND_TO_TASK_STATUS.get(str(low_status or "").lower(), Status.PENDING)
+
+
+def _parse_json(value: Any, default: Any = None) -> Any:
+    """容错解析 *_json 字段:dict 原样、str→ json.loads、None/异常 → default。"""
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except (ValueError, TypeError):
+            return default
+    return default
+
+
+def _parse_dict(value: Any) -> dict[str, Any]:
+    parsed = _parse_json(value, {})
+    return parsed if isinstance(parsed, dict) else {}
+
+
+def _to_ms(value: Any) -> int | None:
+    """ClawMind 秒级时间戳 → 毫秒(对齐 RuntimeInfo.start_time/end_time 约定)。
+    探测值 < 1e12 视为秒(×1000)、已毫秒保持;非法/None → None。"""
+    if value is None:
+        return None
+    try:
+        v = int(value)
+    except (TypeError, ValueError):
+        return None
+    return v * 1000 if v < 1_000_000_000_000 else v
+
+
+# 图级 extend_props 白名单(workflow 标识/运行指标);credentials_json/identity_key/
+# plugin_version 等密钥·摘要·版本,以及图级 node_count/succeeded_count(节点为权威源)均不入。
+_CLAW_MIND_GRAPH_KEEP = (
+    "workflow_id", "workflow_title", "flow_id", "origin_session_id",
+    "total_duration_ms", "total_token_usage", "triggered_by",
+    "current_phase", "started_at", "completed_at",
+)
+_CLAW_MIND_NODE_KEEP = (
+    "session_id", "session_key", "embedded_session_key",
+    "branch_id", "progress_message", "triggered_by",
+)
+
+
+def _build_claw_mind_execution_graph(ext: dict, *, run_status: Any) -> dict[str, Any] | None:
+    """ClawMind ext_info(flow_runs + node_executions)→ graph_to_dict 形状执行图快照。
+
+    - ``run_id`` = int(flow_runs.id)(非法 → 0);图级 status 由底层 status 映射 7 态;
+      ``output`` = 解析 flow_runs.result_json;
+    - extend_props 白名单取 flow_runs 的 workflow 标识/运行指标;
+    - nodes 取 node_executions:task_spec.metadata.title ← node_title(缺则 node_id),
+      run_info.{start,end}_time 秒→毫秒;output = 解析 output_json;token_usage/input/
+      system_context/timing/error 等富字段折叠进 run_info.extend_props;
+    - relations 由各节点 input_json.nodeOutputKeys(params 的兄弟字段)派生(多父 DAG),
+      两端须都在节点集内,过滤悬挂边(默认 DEPENDENCY)。
+    无 flow_runs 且无 node_executions → None。
+    """
+    flow_runs = ext.get("flow_runs") if isinstance(ext, dict) else None
+    flow_runs = flow_runs if isinstance(flow_runs, dict) else {}
+    node_execs = ext.get("node_executions") if isinstance(ext, dict) else None
+    node_execs = node_execs if isinstance(node_execs, list) else []
+    if not flow_runs and not node_execs:
+        return None
+
+    node_ids = {ne.get("node_id") for ne in node_execs
+                if isinstance(ne, dict) and ne.get("node_id")}
+
+    tasks: list[dict[str, Any]] = []
+    relations: list[dict[str, Any]] = []
+    for ne in node_execs:
+        if not isinstance(ne, dict) or not ne.get("node_id"):
+            continue
+        node_id = ne["node_id"]
+        status = _claw_mind_status_to_task(ne.get("status") or run_status)
+        input_doc = _parse_dict(ne.get("input_json"))
+        ik_raw = input_doc.get("nodeOutputKeys")
+        input_keys = ik_raw if isinstance(ik_raw, list) else []
+
+        ep: dict[str, Any] = {}
+        if ne.get("executor_type"):
+            ep["executor_type"] = ne["executor_type"]
+        if ne.get("attempt") is not None:
+            ep["attempt"] = ne["attempt"]
+        tok = _parse_dict(ne.get("token_usage_json"))
+        if tok:
+            ep["token_usage"] = tok
+        if input_doc:
+            ep["input"] = input_doc
+        sc = _parse_dict(ne.get("system_context_json"))
+        if sc:
+            ep["system_context"] = sc
+        if ne.get("duration_ms") is not None:
+            ep["duration_ms"] = ne["duration_ms"]
+        if ne.get("started_at") is not None:
+            ep["started_at"] = ne["started_at"]          # 原始秒
+        if ne.get("completed_at") is not None:
+            ep["completed_at"] = ne["completed_at"]
+        if ne.get("error_text"):
+            ep["error_text"] = ne["error_text"]
+        for k in _CLAW_MIND_NODE_KEEP:
+            if ne.get(k):
+                ep[k] = ne[k]
+
+        for src in input_keys:
+            if isinstance(src, str) and src in node_ids and src != node_id:
+                relations.append({"src_id": src, "dst_id": node_id,
+                                  "type": "DEPENDENCY", "extend_props": {}})
+
+        title = ne.get("node_title") or node_id
+        tasks.append({
+            "node_id": node_id,
+            "task_id": "",
+            "status": status.value,
+            "task_spec": {
+                "metadata": {"task_id": node_id, "title": title, "instruction": ""},
+                "context": {"background": "", "extend_props": {}},
+                "goal": {"objective": "", "acceptances": []},
+            },
+            "run_info": {
+                "run_mode": None,
+                "assignee": None,
+                "start_time": _to_ms(ne.get("started_at")),
+                "end_time": _to_ms(ne.get("completed_at")),
+                "output": _parse_dict(ne.get("output_json")),
+                "acceptance_result": None,
+                "extend_props": ep,
+            },
+        })
+
+    graph_ep: dict[str, Any] = {}
+    for k in _CLAW_MIND_GRAPH_KEEP:
+        if flow_runs.get(k) is not None:
+            graph_ep[k] = flow_runs[k]
+    graph_params = _parse_dict(flow_runs.get("params_json"))
+    if graph_params:
+        graph_ep["params"] = graph_params
+
+    try:
+        run_id = int(flow_runs["id"]) if flow_runs.get("id") is not None else 0
+    except (TypeError, ValueError):
+        run_id = 0
+
+    return {
+        "run_id": run_id,
+        "task_id": "",
+        "loop_round": 0,
+        "status": _claw_mind_status_to_task(run_status).value,
+        "output": _parse_dict(flow_runs.get("result_json")),
+        "extend_props": graph_ep,
+        "tasks": tasks,
+        "relations": relations,
+    }
+
+
 # ===== BCN(BCS Group)CloudEvent 回调解析(语雀《BCS Group 回调接入说明》) =====
 # 回调为 CloudEvent 信封:{event_id, event_type, source="bcs", scope{group_id,session_id,run_id},
 # stream, actor, data{...随 event_type 变}}。仅处理以下 5 个 state_machine 事件,其余返回 None(不处理)。

src/backend/src/agentclaw/community/core/task/task_center/engine.py

@@ -67,7 +67,7 @@ class ExecutionEngine:
     测试可经 facade/engine 子类覆写 ``_build_*`` 注入 stub 策略/投递(测试 seam)。"""
 
     def __init__(self, graph, *, bot=None, bcs=None, discover=None, bcn: BcnService | None = None,
-                 bcs_identity=None, api_base_url: str = "") -> None:
+                 bcs_identity=None, api_base_url: str = "", bot_token_provider=None) -> None:
         """graph: TaskGraphService;bot: OpenApiBotPort;bcs: BcsClientPort;discover: BotDiscoverServiceProtocol。
         端口由 DI 从配置注入(local/prod/double 只换端口实现,引擎代码不变)。prod 必传;测试子类覆写
         ``_build_*`` 注入 stub 策略/投递时可省略(走 super 路径默认 berth)。
@@ -83,6 +83,7 @@ def __init__(self, graph, *, bot=None, bcs=None, discover=None, bcn: BcnService
         self._bcn = bcn
         self._bcs_identity = bcs_identity
         self._api_base_url = api_base_url
+        self._bot_token_provider = bot_token_provider  # driver-bot session_token 取数(直读 bcs_bots);None→不发 Bearer
         self._bg_tasks: set[asyncio.Task] = set()
         self._locks: dict[str, threading.RLock] = {}
         self._locks_guard = threading.RLock()
@@ -139,6 +140,7 @@ def _build_executor(self):
             bot=self._bot, bcs=self._bcs, formatter=PromptFormatterImpl(),
             context=self, sink=self, poller=poller, identity_resolver=self._bcs_identity,
             graph=self._graph, api_base_url=self._api_base_url, bcn=self._bcn,
+            bot_token_provider=self._bot_token_provider,
         )
         import threading as _t
         self._poller_thread = _t.Thread(target=poller.run_poll_loop, daemon=True, name="task-exec-poller")

src/backend/src/agentclaw/community/core/task/task_center/task_service.py

@@ -39,6 +39,24 @@
 logger = logging.getLogger("task.service")
 
 
+def _resolve_coop_collab_mode(has_yaml: bool, group_kind: str | None) -> str:
+    """由 execution_config.group_kind(补充字段)+ 是否带 yaml 推导协作群 collab_mode(group_strategy)。
+
+    既有「有 yaml → state_machine」判断**不变**(group_kind 不介入);group_kind 只在无 yaml 时补充:
+    chat(新增)/manager_worker(默认)。state_machine 需 yaml 定义;未知值 → ValueError。
+    对齐 BCS collaboration.strategy 取值:chat / manager_worker / state_machine(参见语雀《BCS Group 回调接入说明》§4)。
+    """
+    if has_yaml:
+        return "state_machine"
+    if group_kind in ("chat", "manager_worker"):
+        return group_kind
+    if group_kind is None:
+        return "manager_worker"
+    if group_kind == "state_machine":
+        raise ValueError("group_kind=state_machine 需要 yaml 定义")
+    raise ValueError(f"未知 group_kind: {group_kind!r}")
+
+
 # TaskService 结构化实现 api.task.task_service.TaskServiceProtocol —— 依 api/README 四层
 # 契约,core/ 不 import api/(见 test_service_api_conformance.py:core 服务不继承 api Protocol,
 # 由 @runtime_checkable 的 isinstance/issubclass 做结构化一致性校验)。此处置空基类即可。
@@ -55,7 +73,8 @@ def __init__(self, graph, harness=None, *, bot=None, bcs=None, discover=None, bc
                  task_node_repo: TaskNodeRepositoryProtocol | None = None,
                  task_node_run_info_repo: TaskNodeRunInfoRepositoryProtocol | None = None,
                  bot_service=None,
-                 api_base_url: str | None = None) -> None:
+                 api_base_url: str | None = None,
+                 bot_token_provider=None) -> None:
         """graph: TaskGraphService;harness: TaskHarness | None(旁路复位,可选);
         bot/bcs/discover: 传输端口(DI 从配置注入 local/prod/double 实现传给引擎;省略=stub 路径/纯内核单测)。
         BBS 候选通过注入的 BcnService.list_bots_by_task_modes(复用统一 provider 身份)查询。
@@ -77,6 +96,7 @@ def __init__(self, graph, harness=None, *, bot=None, bcs=None, discover=None, bc
         self._callback_repo = callback_repo
         self._bot_service = bot_service
         self._api_base_url = api_base_url
+        self._bot_token_provider = bot_token_provider  # driver-bot session_token 取数(直读 bcs_bots);经 _build_engine 透传给 TaskExecutor
         self._engine = self._build_engine(bot=bot, bcs=bcs, discover=discover)
         # fire-and-forget 后台推进任务跟踪(防 GC + 异常可见 + drain seam)
         self._bg_tasks: set[asyncio.Task] = set()
@@ -96,6 +116,7 @@ def _build_engine(self, *, bot=None, bcs=None, discover=None) -> ExecutionEngine
             self._graph, bot=bot, bcs=bcs, discover=discover, bcn=self._bcn,
             bcs_identity=self._bcs_identity,
             api_base_url=self._api_base_url,
+            bot_token_provider=self._bot_token_provider,
         )
 
     @property
@@ -179,7 +200,7 @@ async def _run_yaml(self, task_id, request, task_info, run_id):
         _task_context = ((_ts.goal.objective or _ts.metadata.instruction or _ts.metadata.title) or "").strip()
         gf = GroupFormation(
             bot_ids=[request.owner_bot_id, *ec.get("participant_bot_ids", [])],
-            collab_mode="state_machine" if has_yaml else "manager_worker",
+            collab_mode=_resolve_coop_collab_mode(has_yaml, ec.get("group_kind")),
             group_name=ec.get("group_name", f"task-{task_id}"),
             members_info=[],
             extend_props={

src/backend/src/agentclaw/community/core/task/task_runner/integration/bcs_bot_token_provider.py

@@ -0,0 +1,77 @@
+"""BcsBotTokenProvider:driver-bot 的 BCS session_token 取数端口(core 只含中性端口 + 缓存包装 + 空实现)。
+
+为什么需要:BCS 建群(``POST /groups``)带 ``event_subscriptions`` 时走 ``require_human``,拒 Bot token;
+参考 ocb 把 driver-bot 的 session_token 作为 ``Authorization: Bearer`` 携带,让 BCS 把 caller 解析成
+driver/originator bot(带归属、统一个 caller 身份)。core 不挂厂商数据基建名;具体读法(DB 直读
+``bcs_bots.session_token`` 等)属 corp/数据源具体实现,放在 community/plugins(见 ``DbBcsBotTokenProvider``)。
+
+core 只暴露中性 ``BcsBotTokenProvider`` 端口 + ``CachingBcsBotTokenProvider`` 缓存包装 +
+``NullBcsBotTokenProvider``;联调/部署侧经 DI bind ``BcsBotTokenProvider`` 覆写默认提供方(见 task_module)。
+"""
+from __future__ import annotations
+
+import time
+from typing import Callable, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class BcsBotTokenProvider(Protocol):
+    """``bcs_bot_uuid -> session_token`` 解析端口(带缓存由实现负责)。"""
+
+    def get_token(self, bcs_bot_uuid: str) -> str | None: ...
+
+
+class NullBcsBotTokenProvider:
+    """无 token 实现(本地/singlebox/double/未配置):恒返回 None。
+
+    搭配建群不挂订阅/无 token 时走 no-sub 分支,无需 token;本实现下 ``caller_bot_token`` 不发,
+    行为同未配置 provider(向后兼容)。
+    """
+
+    def get_token(self, bcs_bot_uuid: str) -> str | None:
+        return None
+
+
+# 未命中时短缓存(秒):避免对 DB 反复打同一条不存在的 bot。对齐 ocb("查失败也缓存短 TTL")。
+_DEFAULT_FAIL_TTL_S: float = 60.0
+
+
+class CachingBcsBotTokenProvider:
+    """对 ``resolver`` 包一层进程内 TTL 缓存,命中/未命中分别缓存。
+
+    ``resolver`` 是真实查数闭包(``bcs_bot_uuid -> session_token 或 None``);prod 由 corp 覆写注入
+    直读 ``bcs_bots.session_token`` 的 resolver(放在 community/plugins),测试/本地注入桩。不把 token 明文写日志。
+
+    Args:
+        resolver: 真实查数闭包。
+        ttl_s: 命中缓存有效期(秒),默认 300(5 分钟,对齐 ocb)。
+        clock: 可注入的单调时钟(默认 ``time.monotonic``),便于测试不依赖真睡。
+    """
+
+    def __init__(
+        self,
+        resolver: Callable[[str], str | None],
+        *,
+        ttl_s: float = 300.0,
+        clock: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._resolver = resolver
+        self._ttl_s = ttl_s
+        self._clock = clock
+        # _cache: bot_uuid -> (token or "", expire_at)。空串哨兵表示一次未命中的短缓存。
+        self._cache: dict[str, tuple[str, float]] = {}
+
+    def get_token(self, bcs_bot_uuid: str) -> str | None:
+        now = self._clock()
+        cached = self._cache.get(bcs_bot_uuid)
+        if cached is not None:
+            token, expire_at = cached
+            if now < expire_at:
+                return token or None
+        token = self._resolver(bcs_bot_uuid)
+        if token:
+            self._cache[bcs_bot_uuid] = (token, now + self._ttl_s)
+            return token
+        # 未命中也短缓存,避免反复查库打爆 DB。
+        self._cache[bcs_bot_uuid] = ("", now + min(self._ttl_s, _DEFAULT_FAIL_TTL_S))
+        return None

src/backend/src/agentclaw/community/core/task/task_runner/integration/bcs_http_adapter.py

@@ -64,6 +64,7 @@ class BcsCreateGroupRequest:
     visibility: str | None = None
     opening_message: dict[str, Any] | None = None
     event_subscriptions: list[dict[str, Any]] | None = None    # 内联事件订阅(回调 webhook);BCS 把 CloudEvent 推到 sink.url
+    caller_bot_token: str | None = None                        # driver-bot 的 session token(直读 bcs_bots.session_token);参考 ocb:作为 Authorization: Bearer 做 caller 身份
 
 
 @dataclass
@@ -186,7 +187,13 @@ async def create_group(self, req: BcsCreateGroupRequest) -> BcsCreateGroupResult
             v = getattr(req, opt)
             if v is not None:
                 body[opt] = v
-        r = await self._req("POST", "/groups", json=body, idempotency_key=uuid.uuid4().hex)
+        # 参考 ocb(http_client.py:254-257):driver-bot 的 session token 经 Authorization: Bearer 做 caller 身份,
+        # BCS resolve_group_create_caller 据此把 caller 解析成 driver/originator bot(仅 HMAC X-ECB-* 无 caller)。
+        extra_headers: dict[str, str] | None = None
+        if req.caller_bot_token:
+            extra_headers = {"Authorization": f"Bearer {req.caller_bot_token}"}
+        r = await self._req("POST", "/groups", json=body, idempotency_key=uuid.uuid4().hex,
+                           extra_headers=extra_headers)
         data = r.json()
         return BcsCreateGroupResult(
             group_id=data["group_id"], session_id=data.get("session_id"),