支持主持人并列调度多个成员

hermes-plugins/yaoyao/dashboard/group_orchestrator.py

@@ -440,7 +440,8 @@ def build_run_prompt(
     if envelope["run"]["requiredReply"] and host_flow:
         automatic_rule = (
             "你持有本话题当前唯一发言令牌，必须公开处理。请根据最新上下文动态决定下一步："
-            "若要委派，只能使用成员列表中的一个精确 @显示名；禁止 @all，禁止一次 @ 多人；"
+            "有依赖关系时只 @下一位成员；任务相互独立时可以同时使用多个精确 @显示名 或 @all，"
+            "系统会并列执行并在整批结束后统一交还给你复核；"
             "若任务已经完成，直接给出最终答复且不要 @任何成员；也可以发起澄清请求。"
             "禁止保持静默，禁止输出任何 YAOYAO_NO_REPLY 标记。"
         )
@@ -461,7 +462,7 @@ def build_run_prompt(
             "完成后系统会把结果交还主持人复核并决定下一步。"
         )
     collaboration_rule = (
-        "主持流程中只有主持人可以调度下一位成员。"
+        "主持流程中只有主持人可以调度成员；成员可以按依赖串行，也可以成批并列执行。"
         if host_flow
         else "仅在确实需要成员协作时，使用成员列表中的精确 @显示名 或 @all 发起下一轮。"
     )

hermes-plugins/yaoyao/dashboard/group_store.py

@@ -97,6 +97,7 @@ _CASCADE_OPERATIONS = frozenset(
 _CASCADE_PARSE_VERSION = 3
 _MAX_CASCADE_PAGE_SIZE = 32
 _MAX_CASCADE_PLAN_BYTES = 16 * 1024
+_HOST_FLOW_BARRIER_CONTENT = "internal.host-flow.parallel-barrier"
 _TOPIC_TITLE_LENGTH = 120
 _TOPIC_PREVIEW_LENGTH = 240
 _DEFAULT_SESSION_CONFIGURATION_JSON = (
@@ -1958,11 +1959,16 @@ class GroupStore:
                 next_mode = self._orchestration_mode(command["orchestrationMode"])
                 if next_mode != current_room["orchestration_mode"]:
                     active = connection.execute(
-                        """SELECT 1 FROM group_agent_runs
-                        WHERE room_id = ?
-                          AND status IN ('queued', 'running', 'awaiting_input')
+                        """SELECT 1 FROM group_agent_runs AS run
+                        LEFT JOIN group_idempotency AS cascade
+                          ON cascade.request_id = 'internal:cascade:' || run.id
+                        WHERE run.room_id = ?
+                          AND (
+                            run.status IN ('queued', 'running', 'awaiting_input')
+                            OR cascade.operation = ?
+                          )
                         LIMIT 1""",
-                        (canonical_room_id,),
+                        (canonical_room_id, _CASCADE_PENDING_OPERATION),
                     ).fetchone()
                     if active is not None:
                         raise GroupConflictError(
@@ -2981,6 +2987,22 @@ class GroupStore:
                       WHERE occupied_topic.topic_id = candidate.topic_id
                         AND occupied_topic.status IN ('running', 'awaiting_input')
                     )
+                    OR (
+                      agent.is_host = 0
+                      AND NOT EXISTS (
+                        SELECT 1 FROM group_agent_runs AS occupied_topic
+                        JOIN group_agents AS occupied_topic_agent
+                          ON occupied_topic_agent.id = occupied_topic.agent_id
+                        WHERE occupied_topic.topic_id = candidate.topic_id
+                          AND occupied_topic.status IN ('running', 'awaiting_input')
+                          AND (
+                            occupied_topic.root_message_id != candidate.root_message_id
+                            OR occupied_topic.trigger_message_id != candidate.trigger_message_id
+                            OR occupied_topic.depth != candidate.depth
+                            OR occupied_topic_agent.is_host = 1
+                          )
+                      )
+                    )
                   )
                 ORDER BY CASE WHEN room.orchestration_mode = 'host'
                               THEN root_message.seq ELSE trigger_message.seq END ASC,
@@ -3392,7 +3414,7 @@ class GroupStore:
             raise ValueError("A runtime-less settle only supports failed")
         with self.write_transaction() as connection:
             run = self._run_row(connection, canonical_run_id)
-            self._active_room(connection, run["room_id"])
+            room = self._active_room(connection, run["room_id"])
             agent = self._enabled_agent(connection, run["room_id"], run["agent_id"])
             response = self._message_row(connection, run["response_message_id"])
             self._validate_run_message_state(run, response)
@@ -3485,7 +3507,9 @@ class GroupStore:
                     ),
                     created_at=now,
                 )
-            if outcome == "completed":
+            if outcome == "completed" or (
+                room["orchestration_mode"] == "host" and not agent["is_host"]
+            ):
                 self._record_cascade_plan(
                     connection,
                     source=run,
@@ -3542,8 +3566,8 @@ class GroupStore:
         )
         if source_agent is None:
             raise GroupStoreError("Cascade source Agent is missing")
-        if room["orchestration_mode"] == "host" and source_message["visible"]:
-            if source_agent["is_host"]:
+        if room["orchestration_mode"] == "host":
+            if source_agent["is_host"] and source_message["visible"]:
                 target_ids, warnings = self._plan_cascade_mentions(
                     source_message["content"],
                     agents,
@@ -3551,14 +3575,7 @@ class GroupStore:
                     include_automatic=False,
                 )
                 explicit_ids = list(target_ids)
-                if len(target_ids) > 1:
-                    warnings.append({
-                        "code": "mention",
-                        "message": "主持流程每轮只调度一位成员，已采用第一个 @ 目标",
-                    })
-                    target_ids = target_ids[:1]
-                    explicit_ids = explicit_ids[:1]
-            else:
+            elif not source_agent["is_host"]:
                 host = next(
                     (agent for agent in agents if agent["is_host"] and agent["enabled"]),
                     None,
@@ -3567,6 +3584,9 @@ class GroupStore:
                     raise GroupStoreError("Host flow has no enabled host Agent")
                 target_ids, warnings = [str(host["id"])], []
                 explicit_ids = []
+            else:
+                target_ids, warnings = [], []
+                explicit_ids = []
         elif source_message["visible"]:
             target_ids, warnings = self._plan_cascade_mentions(
                 source_message["content"],
@@ -3616,6 +3636,36 @@ class GroupStore:
             ),
         )
 
+    def _record_host_flow_terminal_plan_if_needed(
+        self,
+        connection: sqlite3.Connection,
+        run: sqlite3.Row,
+        created_at: float,
+    ) -> None:
+        room = connection.execute(
+            "SELECT orchestration_mode, archived FROM group_rooms WHERE id = ?",
+            (run["room_id"],),
+        ).fetchone()
+        agent = connection.execute(
+            "SELECT is_host, enabled FROM group_agents WHERE id = ? AND room_id = ?",
+            (run["agent_id"], run["room_id"]),
+        ).fetchone()
+        if (
+            room is None
+            or room["archived"]
+            or room["orchestration_mode"] != "host"
+            or agent is None
+            or not agent["enabled"]
+            or agent["is_host"]
+        ):
+            return
+        self._record_cascade_plan(
+            connection,
+            source=run,
+            source_message=self._message_row(connection, run["response_message_id"]),
+            created_at=created_at,
+        )
+
     @classmethod
     def _plan_cascade_mentions(
         cls,
@@ -4058,25 +4108,31 @@ class GroupStore:
             if source is None:
                 raise GroupStoreError("Stored cascade source is missing")
             room = connection.execute(
-                "SELECT archived FROM group_rooms WHERE id = ?",
+                "SELECT archived, orchestration_mode FROM group_rooms WHERE id = ?",
                 (source["room_id"],),
             ).fetchone()
             source_agent = connection.execute(
-                "SELECT enabled FROM group_agents WHERE id = ? AND room_id = ?",
+                "SELECT enabled, is_host FROM group_agents WHERE id = ? AND room_id = ?",
                 (source["agent_id"], source["room_id"]),
             ).fetchone()
             if room is None:
                 raise GroupStoreError("Stored cascade ownership is corrupt")
-            if source["status"] != "completed":
-                raise GroupStoreError("Stored cascade source is not completed")
             source_message = self._message_row(
                 connection, source["response_message_id"]
             )
-            if (
-                source_message["id"] != plan["sourceMessageId"]
-                or source_message["status"] != "completed"
-            ):
+            host_flow_member_terminal = (
+                room["orchestration_mode"] == "host"
+                and source_agent is not None
+                and not source_agent["is_host"]
+                and source["status"] in {"completed", "failed", "interrupted"}
+            )
+            if source["status"] != "completed" and not host_flow_member_terminal:
+                raise GroupStoreError("Stored cascade source is not terminal")
+            if source_message["id"] != plan["sourceMessageId"]:
                 raise GroupStoreError("Stored cascade message is corrupt")
+            self._validate_run_message_status(
+                str(source["status"]), str(source_message["status"])
+            )
             if room["archived"] or source_agent is None or not source_agent["enabled"]:
                 result = self._cascade_summary([], [], state="discarded")
                 connection.execute(
@@ -4220,6 +4276,83 @@ class GroupStore:
             warning_texts.append("Cascade depth limit reached")
             canonical_agent_ids = []
         source_message = self._message_row(connection, source["response_message_id"])
+        trigger_message_id = str(source["response_message_id"])
+        if host_review:
+            current_host = next(
+                (agent for agent in room_agents if agent["is_host"] and agent["enabled"]),
+                None,
+            )
+            if current_host is None:
+                raise GroupStoreError("Host flow has no enabled host Agent")
+            canonical_agent_ids = [str(current_host["id"])]
+            reply_modes = {str(current_host["id"]): "automatic"}
+            unfinished_sibling = connection.execute(
+                """SELECT 1 FROM group_agent_runs
+                WHERE topic_id = ? AND root_message_id = ? AND depth = ?
+                  AND trigger_message_id = ? AND id != ?
+                  AND status NOT IN ('completed', 'failed', 'interrupted')
+                LIMIT 1""",
+                (
+                    source["topic_id"],
+                    source["root_message_id"],
+                    source["depth"],
+                    source["trigger_message_id"],
+                    source["id"],
+                ),
+            ).fetchone()
+            if unfinished_sibling is not None:
+                return {"runs": [], "systemMessages": []}
+            existing_review = connection.execute(
+                """SELECT 1 FROM group_agent_runs
+                WHERE topic_id = ? AND root_message_id = ? AND depth = ?
+                  AND agent_id = ? LIMIT 1""",
+                (
+                    source["topic_id"],
+                    source["root_message_id"],
+                    next_depth,
+                    current_host["id"],
+                ),
+            ).fetchone()
+            if existing_review is not None:
+                return {"runs": [], "systemMessages": []}
+            barrier = connection.execute(
+                """SELECT id FROM group_messages
+                WHERE room_id = ? AND topic_id = ? AND root_message_id = ?
+                  AND reply_to_message_id = ? AND sender_kind = 'system'
+                  AND visible = 0 AND content = ? AND status = 'completed'
+                ORDER BY seq ASC LIMIT 1""",
+                (
+                    source["room_id"],
+                    source["topic_id"],
+                    source["root_message_id"],
+                    source["trigger_message_id"],
+                    _HOST_FLOW_BARRIER_CONTENT,
+                ),
+            ).fetchone()
+            if barrier is None:
+                barrier_id = str(uuid.uuid4())
+                connection.execute(
+                    """INSERT INTO group_messages
+                    (id, room_id, topic_id, sender_kind, sender_id, sender_name,
+                     root_message_id, reply_to_message_id, client_message_id,
+                     content, reasoning, tool_state_json, status, error, visible,
+                     created_at, updated_at)
+                    VALUES (?, ?, ?, 'system', 'system', '系统', ?, ?, NULL,
+                            ?, '', '[]', 'completed', '', 0, ?, ?)""",
+                    (
+                        barrier_id,
+                        source["room_id"],
+                        source["topic_id"],
+                        source["root_message_id"],
+                        source["trigger_message_id"],
+                        _HOST_FLOW_BARRIER_CONTENT,
+                        now,
+                        now,
+                    ),
+                )
+                trigger_message_id = barrier_id
+            else:
+                trigger_message_id = str(barrier["id"])
         explicit_ids, _ = self._plan_cascade_mentions(
             source_message["content"],
             room_agents,
@@ -4276,7 +4409,7 @@ class GroupStore:
                     agent_id,
                     agent["display_name"],
                     source["root_message_id"],
-                    source["response_message_id"],
+                    trigger_message_id,
                     int(visible),
                     now,
                     now,
@@ -4298,7 +4431,7 @@ class GroupStore:
                     source["room_id"],
                     source["topic_id"],
                     agent_id,
-                    source["response_message_id"],
+                    trigger_message_id,
                     response_id,
                     source["root_message_id"],
                     next_depth,
@@ -4330,6 +4463,31 @@ class GroupStore:
                 created_at=now,
             )
             created_runs.append(run)
+        if (
+            room["orchestration_mode"] == "host"
+            and source_agent["is_host"]
+            and created_runs
+        ):
+            barrier_id = str(uuid.uuid4())
+            connection.execute(
+                """INSERT INTO group_messages
+                (id, room_id, topic_id, sender_kind, sender_id, sender_name,
+                 root_message_id, reply_to_message_id, client_message_id,
+                 content, reasoning, tool_state_json, status, error, visible,
+                 created_at, updated_at)
+                VALUES (?, ?, ?, 'system', 'system', '系统', ?, ?, NULL,
+                        ?, '', '[]', 'completed', '', 0, ?, ?)""",
+                (
+                    barrier_id,
+                    source["room_id"],
+                    source["topic_id"],
+                    source["root_message_id"],
+                    source["response_message_id"],
+                    _HOST_FLOW_BARRIER_CONTENT,
+                    now,
+                    now,
+                ),
+            )
         system_messages: list[dict[str, object]] = []
         for warning_text in list(dict.fromkeys(warning_texts))[:MAX_AGENTS_PER_ROOM]:
             duplicate = connection.execute(
@@ -4468,8 +4626,8 @@ class GroupStore:
         )
         with self.write_transaction() as connection:
             run = self._run_row(connection, canonical_run_id)
-            self._active_room(connection, run["room_id"])
-            self._enabled_agent(connection, run["room_id"], run["agent_id"])
+            room = self._active_room(connection, run["room_id"])
+            agent = self._enabled_agent(connection, run["room_id"], run["agent_id"])
             response = self._message_row(connection, run["response_message_id"])
             if (
                 response["room_id"] != run["room_id"]
@@ -4539,6 +4697,17 @@ class GroupStore:
                             message,
                             created_at=now,
                         )
+            if status == "completed" or (
+                room["orchestration_mode"] == "host" and not agent["is_host"]
+            ):
+                self._record_cascade_plan(
+                    connection,
+                    source=self._run_row(connection, run["id"]),
+                    source_message=self._message_row(
+                        connection, run["response_message_id"]
+                    ),
+                    created_at=now,
+                )
             return result
 
     def bind_run_runtime(
@@ -5297,6 +5466,9 @@ class GroupStore:
                 runtime_session_ids=run_runtime_ids,
                 run=changed,
             )
+            self._record_host_flow_terminal_plan_if_needed(
+                connection, self._run_row(connection, run["id"]), now
+            )
             changed_ids.append(run["id"])
         return {"runIds": changed_ids, "runtimeSessionIds": runtime_ids}
 
@@ -5363,6 +5535,9 @@ class GroupStore:
                     runtime_session_ids=runtime_ids,
                     run=changed,
                 )
+                self._record_host_flow_terminal_plan_if_needed(
+                    connection, self._run_row(connection, run["id"]), now
+                )
                 changed_ids.append(run["id"])
             return changed_ids
 

hermes-plugins/yaoyao/dashboard/tests/test_group_host_contract.py

@@ -748,20 +748,78 @@ class GroupHostContractTests(unittest.TestCase):
                 store,
                 room["id"],
                 agents["host"]["id"],
-                content="@成员一 先处理，@成员二 等待",
+                content="@成员一 调研前端，@成员二 调研后端，并列处理",
             )
             delegated = store.complete_cascade(source)
-            self.assertEqual(delegated["runCount"], 1)
-            delegated_run = store.get_run(delegated["runIds"][0])
-            self.assertEqual(delegated_run["agentId"], agents["one"]["id"])
+            self.assertEqual(delegated["runCount"], 2)
+            delegated_runs = [
+                store.get_run(run_id) for run_id in delegated["runIds"]
+            ]
+            self.assertEqual(
+                {run["agentId"] for run in delegated_runs},
+                {agents["one"]["id"], agents["two"]["id"]},
+            )
+            self.assertEqual(
+                {run["triggerMessageId"] for run in delegated_runs},
+                {store.get_run(source)["responseMessageId"]},
+            )
             with store.connection() as connection:
-                required_reply = connection.execute(
-                    "SELECT required_reply FROM group_agent_runs WHERE id = ?",
-                    (delegated_run["id"],),
-                ).fetchone()[0]
-            self.assertEqual(required_reply, 0)
-            self.assertEqual(delegated_run["depth"], 1)
-            self.assertEqual(delegated["systemMessageCount"], 1)
+                required_replies = connection.execute(
+                    """SELECT required_reply FROM group_agent_runs
+                    WHERE id IN (?, ?)""",
+                    tuple(delegated["runIds"]),
+                ).fetchall()
+            self.assertTrue(all(row[0] == 0 for row in required_replies))
+            self.assertTrue(all(run["depth"] == 1 for run in delegated_runs))
+            self.assertEqual(delegated["systemMessageCount"], 0)
+
+            first_parallel = store.claim_next_runnable_run()
+            second_parallel = store.claim_next_runnable_run()
+            self.assertEqual(
+                {first_parallel["id"], second_parallel["id"]},
+                set(delegated["runIds"]),
+            )
+            self.assertIsNone(store.claim_next_runnable_run())
+            later = self._send(store, room["id"], "这是后续独立请求", [])
+
+            store.upsert_agent_message(
+                first_parallel["id"],
+                content="并列结果一",
+                reasoning="",
+                tool_state=[],
+                status="completed",
+            )
+            store.transition_run(first_parallel["id"], "completed")
+            waiting = store.complete_cascade(first_parallel["id"])
+            self.assertEqual(waiting["runCount"], 0)
+
+            store.upsert_agent_message(
+                second_parallel["id"],
+                content="并列结果二",
+                reasoning="",
+                tool_state=[],
+                status="completed",
+            )
+            store.transition_run(second_parallel["id"], "completed")
+            with self.assertRaisesRegex(
+                STORE_MODULE.GroupConflictError, "runs are active"
+            ):
+                store.update_room(
+                    room["id"],
+                    {"requestId": new_id(), "orchestrationMode": "free"},
+                )
+            joined = store.complete_cascade(second_parallel["id"])
+            self.assertEqual(joined["runCount"], 1)
+            review = store.get_run(joined["runIds"][0])
+            self.assertEqual(review["agentId"], agents["host"]["id"])
+            self.assertEqual(review["depth"], 1)
+            claimed_review = store.claim_next_runnable_run()
+            self.assertEqual(claimed_review["id"], review["id"])
+            projection = store.read_run_projection(review["id"])
+            contents = [message["content"] for message in projection["messages"]]
+            self.assertIn("并列结果一", contents)
+            self.assertIn("并列结果二", contents)
+            self.assertNotIn(later["message"]["content"], contents)
 
         with tempfile.TemporaryDirectory() as directory:
             store = GroupStore(Path(directory) / "group.db")
@@ -800,7 +858,7 @@ class GroupHostContractTests(unittest.TestCase):
                 projection, room["agents"]
             )
             self.assertIn("当前唯一发言令牌", prompt)
-            self.assertIn("禁止 @all", prompt)
+            self.assertIn("并列执行", prompt)
 
     def test_v10_migration_adds_free_orchestration_mode(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
@@ -828,6 +886,58 @@ class GroupHostContractTests(unittest.TestCase):
                 migrated.get_room(room["id"])["orchestrationMode"], "free"
             )
 
+    def test_host_flow_parallel_barrier_returns_failures_to_host(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = GroupStore(Path(directory) / "group.db")
+            store.initialize()
+            room = store.create_room({
+                "requestId": new_id(),
+                "name": "并列失败群",
+                "cwd": "",
+                "orchestrationMode": "host",
+                "agents": [
+                    {"profile": "host", "displayName": "主持人", "isHost": True},
+                    {"profile": "one", "displayName": "成员一"},
+                    {"profile": "two", "displayName": "成员二"},
+                ],
+            })
+            agents = {agent["profile"]: agent for agent in room["agents"]}
+            source = self._insert_completed_source(
+                store,
+                room["id"],
+                agents["host"]["id"],
+                content="@成员一 检查接口，@成员二 检查数据",
+            )
+            delegated = store.complete_cascade(source)
+            first = store.claim_next_runnable_run()
+            second = store.claim_next_runnable_run()
+            self.assertEqual(
+                {first["id"], second["id"]}, set(delegated["runIds"])
+            )
+
+            store.upsert_agent_message(
+                first["id"],
+                content="接口检查完成",
+                reasoning="",
+                tool_state=[],
+                status="completed",
+            )
+            store.transition_run(first["id"], "completed")
+            self.assertEqual(store.complete_cascade(first["id"])["runCount"], 0)
+
+            store.transition_run(second["id"], "failed", error="数据检查失败")
+            joined = store.complete_cascade(second["id"])
+            self.assertEqual(joined["runCount"], 1)
+            review = store.claim_next_runnable_run()
+            projection = store.read_run_projection(review["id"])
+            self.assertEqual(review["agentId"], agents["host"]["id"])
+            self.assertTrue(
+                any(
+                    message["summary"] == "数据检查失败"
+                    for message in projection["messages"]
+                )
+            )
+
     @staticmethod
     def _send(
         store: GroupStore, room_id: str, content: str, mention_ids: list[str]

src/client/components/groups/CreateGroupDialog.vue

@@ -80,7 +80,7 @@ function create() {
             </select>
           </label>
           <label v-if="hostFlowEnabled" class="host-picker">
-            <span>协作模式<small>主持流程由主持人逐步选择下一位成员，避免多人并发造成上下文分叉。</small></span>
+            <span>协作模式<small>主持人可按依赖串行调度，也可一次 @ 多人并列执行；整批结束后统一复核。</small></span>
             <select v-model="orchestrationMode" aria-label="协作模式">
               <option value="free">自由讨论</option>
               <option value="host">主持流程</option>

src/client/components/groups/GroupManager.vue

@@ -216,7 +216,7 @@ function statusLabel(status: GroupAgent['status']): string {
       <label><span>名称</span><input v-model="form.name" maxlength="80" @change="saveRoom" /></label>
       <label class="rounds"><span>最多回复轮数<small>-1 表示无限</small></span><input v-model.number="form.replyRounds" type="number" min="-1" max="100" aria-label="最多回复轮数" @change="saveRoom" /></label>
       <label v-if="hostFlowEnabled" class="flow-mode">
-        <span>协作模式<small>主持流程会逐步调度，每轮只允许一位成员执行。</small></span>
+        <span>协作模式<small>主持人可按依赖逐步调度，也可一次 @ 多人并列执行。</small></span>
         <select v-model="form.orchestrationMode" aria-label="协作模式" :disabled="busy || hasActiveAgents" :title="hasActiveAgents ? '请等待当前回复完成或先中断 Agent' : ''" @change="saveRoom">
           <option value="free">自由讨论</option>
           <option value="host">主持流程</option>