发布 v0.3.8：完整迁移团队协作调度与执行恢复

README.md

@@ -1,6 +1,6 @@
 # 夭夭 Web
 
-当前发布版本：**v0.3.7**，配套 iOS **1.3（172）**。详见 [发布说明](docs/releases/v0.3.7.md)。
+当前发布版本：**v0.3.8**，配套 iOS **1.3（173）**。详见 [发布说明](docs/releases/v0.3.8.md)。
 
 夭夭 Web 为网页和 iOS 提供统一的 Agent、群聊、文件库及语音配置服务。Hermes 通过标准 9119 HTTP/WebSocket 接口执行任务；无需安装 Yaoyao 插件。
 

docs/releases/v0.3.8.md

@@ -0,0 +1,19 @@
+# v0.3.8
+
+将原团队协作调度迁入 Web，以持久化的成员任务和批次替换简化队列。
+
+- 管理员模式下，用户消息先由管理员处理。管理员可以按依赖逐个委派，或一次委派多名成员并行执行；整批成功、失败或中断后只复核一次，复核不额外消耗轮数。
+- 自由模式恢复自动参与接力；无关的自动成员可以静默退出，管理员必须公开处理。支持默认 3 轮、1～100 轮和不限轮数。
+- 执行中继续接收消息并持久化排队。同一群成员串行、每群最多 3 个执行、全局最多 4 个执行；排队消息不会提前进入上一轮上下文。
+- 支持整轮停止和单成员停止；停止未确认时继续核对，避免误报已停止。并行审批与澄清绑定各自成员任务，防止串线和重复提交。
+- 保存批次与后续调度计划，重启后核对原执行，不重复提交提示词或重复复核。现有 v0.3.x 未完成任务向新任务记录转换。
+- 上下文按触发点有界读取，保留首尾、思考摘要、工具名称、附件与失败摘要；隔离其他聊天及缺失历史记录。
+- Web 与 iOS 继续使用原有聊天组件，成员头像独立显示运行/等待状态，设置中提供成员停止与不限轮数。
+
+配套 iOS 1.3（173）。逐项迁移证据见 [协作迁移对照](../workspace-orchestration-parity.md)。
+
+## 验证
+
+- Web 85 个测试文件、488 项测试通过，类型检查和生产构建通过。
+- 浏览器端到端流程通过，覆盖管理员委派、运行中排队、历史、成员、模板及升级前的原有流程。
+- iOS 协议与头像状态测试、完整界面验收通过。

docs/workspace-chat.md

@@ -29,9 +29,9 @@
 
 ## 执行
 
-用户消息先持久化，再创建运行。每聊天一次只能有一个运行；群内按已记录的队列逐个执行。每个群成员有独立上游会话，聊天之间不复用会话。
+用户消息先持久化，再创建根运行和成员任务；执行期间的新消息持久化排队。每个群成员有独立上游会话，同一群成员串行，每群最多 3 个执行、全局最多 4 个执行。管理员模式按用户消息顺序协调，同批成员可以并行。
 
-明确 @ 优先；否则管理员模式由管理员开始，自由模式按自动回复成员开始，无人启用则由管理员响应。管理员委派后由成员执行并返回管理员；自由模式按成员 @ 接力。默认最多 3 轮，范围 1～100，达到上限即结束自动接力。引用和代码中的 @ 不触发协作。
+管理员模式始终先由管理员处理，成员整批终结后返回一次管理员复核。自由模式中明确 @ 优先，无明确 @ 时由管理员和自动参与成员响应；后续无明确 @ 时可按自动参与配置接力。自动成员可静默退出，管理员必须公开处理。默认最多 3 轮，支持 1～100 及 -1（不限），管理员复核不额外增加轮数。达到上限给出提示并结束自动接力；引用和代码中的 @ 不触发协作。
 
 每次成员执行前读取最新角色及群规则，附加到本次输入，不修改基础 Profile。网页和手机只展示用户原始消息，不展示执行提示词。
 

docs/workspace-orchestration-parity.md

@@ -0,0 +1,23 @@
+# 团队协作迁移验收清单
+
+对照迁移前 v0.2.31 的 `group_store.py`、`group_orchestrator.py` 与主持流程契约测试，将调度行为迁入 Web。保持当前创建的 Agent、统一聊天列表、可变群成员和管理员保护，不恢复旧主题接口、插件依赖或名称兼容层。
+
+- 管理员模式：所有用户消息先交管理员；管理员可串行委派或并行委派一批成员，整批终结后只复核一次。
+- 成员成功、失败或中断均形成可复核结果；管理员复核不增加轮数。
+- 自由模式：显式提及优先，无提及按自动参与配置接力；无关自动成员可静默退出，管理员必须公开答复。
+- 默认 3 轮，支持 1～100 与 -1（不限）；只在自动接力边界执行上限判断。
+- 运行中用户消息先持久化再排队；同一成员串行、每群最多 3 个执行、全局最多 4 个执行。
+- 成员任务、批次与后续调度计划持久化；重启后核对已有执行，不重发未确认的提示词，也不重复复核。
+- 上下文按照触发点截取，保留附件、失败结果和省略提示；不把后排用户请求串入先前任务。
+- 支持整轮停止与单成员停止；审批、澄清、超时/断线与取消结果绑定具体成员任务。
+- Web/iOS 保持原有聊天布局，用成员头像表现并发状态；恢复排队发送和不限轮数设置。
+
+## 已完成的验证
+
+`tests/server/workspace.test.ts` 覆盖管理员入口、串行依赖、并行批次、失败/中断回流、复核不占轮数、自由讨论静默退出、公开答复兜底、排队与上下文隔离、全局/群并发限制、不限轮数、并发审批、事务回滚、批次恢复、长历史分页核对、旧执行迁入与停止失败重试。
+
+`tests/e2e/workspace-chat.spec.ts` 通过真实 Web 界面验证管理员先处理后委派，执行中发送的消息持久化排队，前一轮结束后自动处理后一条消息，并验证模板、成员管理和历史同步。
+
+iOS 的 `testTeamWorkspaceStaysInsideOriginalDrawer` 验证原有页面结构、群内成员管理、模板分工、执行中发送并排队、完成后继续回复与不限轮数设置。成员并发头像状态另有 `WorkspaceChatTests` 回归。
+
+验收使用独立标准 Hermes 协议测试服务，不依赖旧插件，不修改生产用户聊天记录。实际远端模型和那台机器的部署状态需以其运行日志为准。

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "hermes-yaoyao",
-  "version": "0.3.7",
+  "version": "0.3.8",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "hermes-yaoyao",
-      "version": "0.3.7",
+      "version": "0.3.8",
       "license": "BSL-1.1",
       "dependencies": {
         "@koa/bodyparser": "^5.0.0",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hermes-yaoyao",
-  "version": "0.3.7",
+  "version": "0.3.8",
   "private": true,
   "license": "BSL-1.1",
   "description": "夭夭 Web：直连 Hermes 9119 的聊天、团队、文件库与产物工作台",

release.json

@@ -1,6 +1,6 @@
 {
   "schemaVersion": 1,
-  "releaseVersion": "0.3.7",
-  "webVersion": "0.3.7",
-  "gitTag": "v0.3.7"
+  "releaseVersion": "0.3.8",
+  "webVersion": "0.3.8",
+  "gitTag": "v0.3.8"
 }

src/client/components/composer/ComposerShell.vue

@@ -89,7 +89,7 @@ let errorTimer: number | undefined
 const canSubmit = computed(() => !props.disabled && !props.sending && (text.value.trim().length > 0 || attachments.value.length > 0))
 // A running group still accepts new room messages. Only an ordinary session
 // turns its primary send affordance into an interrupt control.
-const showStop = computed(() => props.streaming && (props.stopWhileRunning || (props.mode === 'chat' && !canSubmit.value)))
+const showStop = computed(() => props.streaming && !text.value.trim() && !attachments.value.length && (props.stopWhileRunning || props.mode === 'chat'))
 const compactModel = computed(() => props.modelLabel.split('/').filter(Boolean).at(-1) || props.modelLabel)
 const contextPercent = computed(() => props.contextLimit > 0 ? Math.min(100, Math.round(props.contextUsed / props.contextLimit * 100)) : 0)
 const hasContext = computed(() => props.contextLimit > 0)

src/client/components/workspace/ConversationList.vue

@@ -18,7 +18,7 @@ const rows = computed<SidebarItem[]>(() => props.conversations.filter(c => c.arc
     meta: formatMessageTime(c.lastMessageAt ?? c.createdAt),
     avatarKind: c.kind === 'direct' ? 'agent' : 'team',
     avatarState: workspaceAvatarState(c, c.memberIds[0] || ''),
-    unread: Math.max(0, c.lastSeq - c.readSeq), status: c.activeRunId ? 'working' : undefined })))
+    unread: c.unreadCount ?? Math.max(0, c.lastSeq - c.readSeq), status: c.activeRunId ? 'working' : undefined })))
 function openMenu(id: string, event: MouseEvent) {
   menuId.value = id
   menuPosition.value = { left: `${Math.max(8, Math.min(event.clientX, window.innerWidth - 170))}px`, top: `${Math.max(8, Math.min(event.clientY, window.innerHeight - 100))}px` }

src/client/components/workspace/viewModels.ts

@@ -365,7 +365,7 @@ export function artifactToUi(item: ConversationArtifact): UiLibraryItem {
 
 /** Adapt the Web-owned transcript to the existing chat presentation. */
 export function workspaceMessagesToUi(messages: import('@shared/workspace').WorkspaceMessage[]): UiMessage[] {
-  return messages.map(message => ({
+  return messages.filter(message => message.visible !== false).map(message => ({
     id: message.id, role: message.role, author: message.agentName,
     profile: message.agentId, createdAt: message.createdAt,
     content: message.content.replace(/(!?\[[^\]]*\])\(<?([^)>]+)>?\)/g, (whole, label: string, path: string) => {
@@ -400,6 +400,8 @@ export function workspaceAvatarMembers(memberIds: string[], agents: import('@sha
 
 
 export function workspaceAvatarState(conversation: import('@shared/workspace').WorkspaceConversation | undefined, agentId: string): 'idle' | 'working' | 'waiting' {
+  const state = conversation?.activeAgentStates?.[agentId]
+  if (state) return state === 'running' ? 'working' : 'waiting'
   if (!conversation?.activeRunId || conversation.activeAgentId !== agentId) return 'idle'
   return conversation.activeRunStatus === 'waiting' || conversation.activeRunStatus === 'uncertain' ? 'waiting' : 'working'
 }

src/client/views/ConversationsView.vue

@@ -174,6 +174,7 @@ async function load(id = selected.value, append = false) {
       run: Run | null
       interactions: Interaction[]
       context: Record<string, unknown> | null
+      hiddenMessageIds?: string[]
     }>(`/api/app/conversations/${id}`)
     if (disposed || own !== generation) return
     const atBottom = timeline.value?.isFollowingBottom() ?? true
@@ -183,6 +184,7 @@ async function load(id = selected.value, append = false) {
           (a, b) => a.seq - b.seq,
         )
       : r.messages
+    messages.value = messages.value.filter(m => m.visible !== false && !r.hiddenMessageIds?.includes(m.id))
     run.value = r.run
     interactions.value = r.interactions
     context.value = r.context
@@ -359,7 +361,7 @@ async function upload(event: Event) {
 }
 async function send() {
   const c = active.value
-  if (!c || busy.value || c.activeRunId || (!text.value.trim() && !files.value.length)) return
+  if (!c || busy.value || (!text.value.trim() && !files.value.length)) return
   busy.value = true
   error.value = ''
   const fingerprint = JSON.stringify([c.id, text.value, mentions.value, files.value.map(f => f.id)])
@@ -414,6 +416,13 @@ async function action(operation: 'pin' | 'archive', id = active.value?.id) {
     error.value = String(e)
   }
 }
+async function stopMember(id: string) {
+  if (!active.value) return
+  try {
+    await apiRequest(`/api/app/conversations/${active.value.id}/agents/${id}/stop`, { method: 'POST', body: {} })
+    await load(); await refresh()
+  } catch (cause) { error.value = cause instanceof Error ? cause.message : '停止失败' }
+}
 async function control(action: 'stop' | 'reconcile') {
   if (!run.value) return
   try {
@@ -615,6 +624,7 @@ onBeforeUnmount(() => {
             >至少需要两个 Agent 才能创建群聊。</small
           >
           <small v-if="dialog === 'editGroup'">可增减成员，最多 8 位。当前管理员不能移除；如需移除，请先更换管理员并保存。</small>
+          <button v-for="a in agents.filter(a => active?.activeAgentStates?.[a.id])" :key="`stop:${a.id}`" type="button" @click="stopMember(a.id)">停止 {{ a.name }}</button>
         </fieldset>
         <fieldset v-if="dialog === 'editGroup' && Object.keys(form.memberRoles).length">
           <legend>角色分工</legend>
@@ -646,7 +656,8 @@ onBeforeUnmount(() => {
             >
           </fieldset>
           <label
-            >自动协作轮数<input
+            ><input type="checkbox" :checked="form.maxReplyRounds === -1" @change="form.maxReplyRounds = ($event.target as HTMLInputElement).checked ? -1 : 3" />不限制自动协作轮数</label>
+          <label v-if="form.maxReplyRounds !== -1">自动协作轮数<input
               v-model.number="form.maxReplyRounds"
               type="number"
               min="1"

src/server/workspaceMentions.ts

@@ -0,0 +1,22 @@
+import type { WorkspaceAgent as Agent } from '../shared/workspace.js'
+
+export function mentionedAgents(text: string, agents: Array<Pick<Agent, 'id' | 'name'>>): string[] {
+  if (!text.includes('@')) return []
+  const plain = text
+    .replace(/<quoted_message\b[^>]*>[\s\S]*?<\/quoted_message>/gi, '')
+    .replace(/```[\s\S]*?(?:```|$)|~~~[\s\S]*?(?:~~~|$)/g, '')
+    .replace(/`[^`]*`/g, '')
+    .replace(/https?:\/\/\S+/g, '')
+    .replace(/[\p{L}\p{N}._%+-]{1,64}@[\p{L}\p{N}.-]{1,255}\.[\p{L}]{2,63}/gu, '')
+    .split('\n').filter(line => !line.trimStart().startsWith('>')).join('\n')
+  const names = [...agents].sort((a, b) => b.name.length - a.name.length)
+  const alternatives = [...names.map(a => a.name.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')), 'all', '所有人']
+  const pattern = new RegExp(`(?<![A-Za-z0-9_.%+/@-])@(${alternatives.join('|')})(?![A-Za-z0-9_])`, 'giu')
+  const result = new Set<string>()
+  for (const match of plain.matchAll(pattern)) {
+    const name = match[1]!
+    if (/^(all|所有人)$/i.test(name)) { for (const a of agents) result.add(a.id) }
+    else { const a = agents.find(a => a.name.toLocaleLowerCase() === name.toLocaleLowerCase()); if (a) result.add(a.id) }
+  }
+  return [...result]
+}

src/server/workspaceRoutes.ts

@@ -118,6 +118,7 @@ export function workspaceRouter(
     ctx.body = {
       conversation: store.conversationSummary(user, conversation),
       messages: store.messages(user, conversation.id),
+      hiddenMessageIds: store.hiddenMessageIds(user, conversation.id),
       run: conversation.activeRunId
         ? store.require<WorkspaceRun>(user, 'run', conversation.activeRunId)
         : null,
@@ -127,8 +128,15 @@ export function workspaceRouter(
       context: store.get(user, 'context', conversation.id) ?? null,
     }
   })
-  router.patch('/api/app/conversations/:id', (ctx) => {
-    ctx.body = { conversation: store.updateConversation(owner(ctx), ctx.params.id, body(ctx)) }
+  router.patch('/api/app/conversations/:id', async (ctx) => {
+    const user = owner(ctx), conversation = store.updateConversation(user, ctx.params.id, body(ctx))
+    if (conversation.archived) await runtime.stopConversation(user, conversation.id)
+    runtime.wake()
+    ctx.body = { conversation: store.require(user, 'conversation', conversation.id) }
+  })
+  router.post('/api/app/conversations/:id/agents/:agentId/stop', async (ctx) => {
+    await runtime.stopAgent(owner(ctx), ctx.params.id, ctx.params.agentId)
+    ctx.body = { ok: true }
   })
   router.get('/api/app/conversations/:id/messages', (ctx) => {
     ctx.body = {

src/server/workspaceRuntime.ts

@@ -13,17 +13,11 @@ import type {
   WorkspaceInteraction,
 } from '../shared/workspace.js'
 
-type Run = WorkspaceRun & {
-  queue: string[]
-  next: string[]
-  hostReturn: boolean
-  currentMessageId?: string
-  turnConfiguration?: {
-    mode: 'host' | 'free'
-    administratorId: string
-    members: Array<Pick<Agent, 'id' | 'name'>>
-  }
-}
+import { WorkspaceScheduler, type Work, NO_REPLY, HOST_FALLBACK } from './workspaceScheduler.js'
+import { mentionedAgents } from './workspaceMentions.js'
+export { mentionedAgents } from './workspaceMentions.js'
+type Run = WorkspaceRun
+
 export interface WorkspaceBinding {
   id: string
   nodeId: string
@@ -33,12 +27,15 @@ export interface WorkspaceBinding {
   aliases: string[]
   runId: string
   messageId: string
+  taskId?: string
+  contextSeq?: number
 }
 interface LiveTurn {
   gateway: WorkspaceGateway
   runtimeId: string
   runId: string
   agentId: string
+  taskId: string
   done(error?: Error): void
 }
 export const sendInput = z
@@ -50,81 +47,14 @@ export const sendInput = z
   })
   .strict()
   .refine((b) => b.content.trim() || b.fileIds.length, '请输入消息或添加附件')
-export function mentionedAgents(text: string, agents: Array<Pick<Agent, 'id' | 'name'>>): string[] {
-  const plain = text
-    .replace(/<quoted_message\b[^>]*>[\s\S]*?<\/quoted_message>/gi, '')
-    .replace(/```[\s\S]*?(?:```|$)|~~~[\s\S]*?(?:~~~|$)/g, '')
-    .replace(/`[^`]*`/g, '')
-    .replace(/https?:\/\/\S+/g, '')
-    .replace(/[\p{L}\p{N}._%+-]+@[\p{L}\p{N}.-]+\.[\p{L}]{2,}/gu, '')
-    .split('\n').filter(line => !line.trimStart().startsWith('>')).join('\n')
-  const names = [...agents].sort((a, b) => b.name.length - a.name.length)
-  const alternatives = [...names.map(a => a.name.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')), 'all', '所有人']
-  const pattern = new RegExp(`(?<![A-Za-z0-9_.%+/@-])@(${alternatives.join('|')})(?![A-Za-z0-9_])`, 'giu')
-  const result = new Set<string>()
-  for (const match of plain.matchAll(pattern)) {
-    const name = match[1]!
-    if (/^(all|所有人)$/i.test(name)) { for (const a of agents) result.add(a.id) }
-    else { const a = agents.find(a => a.name.toLocaleLowerCase() === name.toLocaleLowerCase()); if (a) result.add(a.id) }
-  }
-  return [...result]
-}
-export class WorkspaceRuntime {
+export class WorkspaceRuntime extends WorkspaceScheduler {
   private live = new Map<string, LiveTurn>()
-  private executing = new Set<string>()
-  private closing = false
-  private recoveryTimers = new Map<string, ReturnType<typeof setTimeout>>()
-  private recoveryAttempts = new Map<string, number>()
-  onMessage: (owner: string, message: Message) => Promise<void> = async () => {}
-  onNotify: (
-    owner: string,
-    c: Conversation,
-    run: Run,
-    message?: Message,
-    interaction?: WorkspaceInteraction,
-  ) => void = () => {}
-  private notify(
-    owner: string,
-    c: Conversation,
-    run: Run,
-    message?: Message,
-    interaction?: WorkspaceInteraction,
-  ): void {
-    try {
-      this.onNotify(owner, c, run, message, interaction)
-    } catch {
-      /* Optional push failures never change the durable execution result. */
-    }
-  }
-  constructor(
-    readonly store: WorkspaceStore,
-    readonly nodes: WorkspaceNodes,
-    readonly uploads: UploadStore,
-    readonly userActive: (owner: string) => boolean = () => true,
-  ) {
-    // An admitted but unacknowledged turn may already have executed a tool.
-    for (const owner of store.owners())
-      for (const run of store.list<Run>(owner, 'run')) {
-        if (['running', 'waiting'].includes(run.status)) {
-          run.status = 'uncertain'
-          run.error = '服务已重启，请核对运行状态'
-          store.saveRun(owner, run)
-        }
-      }
-  }
-  start(): void {
-    for (const owner of this.store.owners())
-      for (const run of this.store.list<Run>(owner, 'run')) {
-        if (run.status === 'queued') void this.execute(owner, run.id)
-        else if (run.status === 'uncertain') this.scheduleReconcile(owner, run.id)
-      }
-  }
+  constructor(store: WorkspaceStore, readonly nodes: WorkspaceNodes, readonly uploads: UploadStore, userActive: (owner: string) => boolean = () => true) { super(store, userActive) }
   send(owner: string, conversationId: string, input: unknown): Run {
     const body = parse(sendInput, input)
     const result = this.store.command(owner, body.requestId, { conversationId, ...body }, () => {
       const c = this.store.require<Conversation>(owner, 'conversation', conversationId)
       if (c.archived) throw new HttpError(409, '聊天已归档', 'conversation_archived')
-      if (c.activeRunId) throw new HttpError(409, '请等待当前回复结束或先停止', 'conversation_busy')
       if (body.mentionIds.some((a) => !c.memberIds.includes(a)))
         throw new HttpError(400, '只能 @ 群内成员', 'invalid_mentions')
       const records = this.uploads.records(body.fileIds, owner)
@@ -133,17 +63,9 @@ export class WorkspaceRuntime {
         agents = c.memberIds.map((id) => this.store.require<Agent>(owner, 'agent', id))
       const mentions = [
         ...new Set(
-          body.mentionIds.length ? body.mentionIds : mentionedAgents(body.content, agents),
+          [...body.mentionIds, ...mentionedAgents(body.content, agents)],
         ),
       ]
-      const queue =
-        c.kind === 'direct'
-          ? [...c.memberIds]
-          : mentions.length
-            ? mentions
-            : c.mode === 'free' && c.autoReplyIds.length
-              ? [...c.autoReplyIds]
-              : [c.administratorId]
       const message: Message = {
         id: randomUUID(),
         conversationId,
@@ -179,141 +101,32 @@ export class WorkspaceRuntime {
         conversationId,
         messageId: message.id,
         mentionIds: mentions,
-        activeAgentId: queue[0],
+        activeAgentId: c.administratorId,
         status: 'queued',
         round: 0,
-        queue,
-        next: [],
-        hostReturn: c.mode === 'host' && queue.some((id) => id !== c.administratorId),
         createdAt: now,
         updatedAt: now,
       }
-      this.store.saveRun(owner, run)
+      this.admit(owner, run, c, message)
       this.uploads.markReferenced(body.fileIds, owner)
       return run
     })
-    if (this.store.require<Run>(owner, 'run', result.id).status === 'queued')
-      void this.execute(owner, result.id)
+    this.wake()
     return this.store.require<Run>(owner, 'run', result.id)
   }
-  private async execute(owner: string, id: string): Promise<void> {
-    if (this.executing.has(id) || this.closing) return
-    this.executing.add(id)
-    try {
-      while (!this.closing) {
-        const run = this.store.require<Run>(owner, 'run', id)
-        if (!['queued', 'running'].includes(run.status)) break
-        if (!this.userActive(owner)) throw new HttpError(403, '账号已停用', 'account_disabled')
-        const c = this.store.require<Conversation>(owner, 'conversation', run.conversationId)
-        // A current reply may finish, but removed members must not start another turn.
-        run.queue = run.queue.filter(agentId => c.memberIds.includes(agentId))
-        run.next = run.next.filter(agentId => c.memberIds.includes(agentId))
-        if (!run.queue.length) {
-          if (
-            c.kind === 'direct' ||
-            run.round + 1 >= c.maxReplyRounds ||
-            (!run.next.length && !run.hostReturn)
-          ) {
-            run.status = 'complete'
-            this.store.saveRun(owner, run)
-            this.notify(
-              owner,
-              c,
-              run,
-              this.store
-                .messages(owner, c.id)
-                .filter((m) => m.runId === id && m.role === 'assistant')
-                .at(-1),
-            )
-            break
-          }
-          run.round += 1
-          run.queue = run.hostReturn ? [c.administratorId] : [...new Set(run.next)]
-          run.hostReturn = false
-          run.next = []
-          this.store.saveRun(owner, run)
-        }
-        const agent = this.store.require<Agent>(owner, 'agent', run.queue[0]!)
-        run.turnConfiguration = {
-          mode: c.mode,
-          administratorId: c.administratorId,
-          members: c.memberIds.map((id) => {
-            const a = this.store.require<Agent>(owner, 'agent', id)
-            return { id: a.id, name: a.name }
-          }),
-        }
-        run.status = 'running'
-        run.activeAgentId = agent.id
-        this.store.saveRun(owner, run)
-        const message = await this.turn(owner, c, agent, run)
-        await this.onMessage(owner, message).catch(() => {})
-        if (this.closing) return
-        this.advance(owner, run.id, agent.id, message)
-      }
-    } catch (error) {
-      if (!this.closing) {
-        const run = this.store.require<Run>(owner, 'run', id)
-        if (!['interrupted', 'uncertain'].includes(run.status)) {
-          run.status = 'failed'
-          run.error = (error instanceof Error ? error.message : '执行失败').slice(0, 1000)
-          this.store.saveRun(owner, run)
-          this.store.saveMessage(owner, {
-            id:randomUUID(), conversationId:run.conversationId, seq:0, role:'system',
-            content:`执行失败：${run.error}`, reasoning:'', runId:run.id, status:'failed',
-            attachments:[], tools:[], createdAt:Date.now(),
-          })
-          this.notify(owner, this.store.require(owner, 'conversation', run.conversationId), run)
-        }
-      }
-    } finally {
-      this.executing.delete(id)
-      this.scheduleReconcile(owner, id)
-    }
-  }
-  private advance(owner: string, runId: string, agentId: string, message: Message): void {
-    this.store.atomic(() => {
-      const run = this.store.require<Run>(owner, 'run', runId)
-      if (run.status === 'interrupted') return
-      const c = this.store.require<Conversation>(owner, 'conversation', run.conversationId)
-      const configuration = run.turnConfiguration
-      if (!configuration) throw new Error('运行配置缺失，已停止自动协作')
-      if (configuration.mode === 'free' || agentId === configuration.administratorId)
-        run.next.push(
-          ...mentionedAgents(message.content, configuration.members).filter(
-            id => id !== agentId && c.memberIds.includes(id),
-          ),
-        )
-      if (
-        c.kind === 'group' &&
-        configuration.mode === 'host' &&
-        agentId !== configuration.administratorId
-      )
-        run.hostReturn = true
-      if (run.queue[0] === agentId) run.queue.shift()
-      run.queue = run.queue.filter(id => c.memberIds.includes(id))
-      run.next = run.next.filter(id => c.memberIds.includes(id))
-      run.activeAgentId = run.queue[0]
-      run.currentMessageId = undefined
-      run.error = undefined
-      run.status = 'running'
-      this.store.saveRun(owner, run)
-    })
-  }
-  private async turn(
+  protected async performTurn(
     owner: string,
     c: Conversation,
     agent: Agent,
-    run: Run,
+    run: Work,
     recovering = false,
   ): Promise<Message> {
     const key = `${c.id}:${agent.id}`,
       target = this.nodes.target(owner, agent.nodeId)
     const gateway = new WorkspaceGateway(target)
-    run.activeAgentId = agent.id
-    this.store.saveRun(owner, run)
     let binding = this.store.get<WorkspaceBinding>(owner, 'binding', key)
     let message =
-      recovering && run.currentMessageId
+      run.currentMessageId
         ? this.store.require<Message>(owner, 'message', run.currentMessageId)
         : undefined
     if (!message) {
@@ -327,7 +140,9 @@ export class WorkspaceRuntime {
         content: '',
         reasoning: '',
         status: 'streaming',
-        runId: run.id,
+        runId: run.runId,
+        taskId: run.id,
+        visible: run.replyMode !== 'automatic' || run.requiredReply,
         tools: [],
         attachments: [],
         createdAt: Date.now(),
@@ -335,7 +150,7 @@ export class WorkspaceRuntime {
       run.currentMessageId = message.id
       this.store.atomic(() => {
         this.store.saveMessage(owner, message!)
-        this.store.saveRun(owner, run)
+        this.saveWork(owner, run)
       })
     }
     const resultMessage = message
@@ -356,23 +171,46 @@ export class WorkspaceRuntime {
       if (settled) return
       settled = true
       this.live.delete(key)
-      if (error) {
-        const current = this.store.require<Run>(owner, 'run', run.id)
-        resultMessage.status =
-          current.status === 'interrupted' ? 'interrupted' : submitted ? 'uncertain' : 'failed'
-        if (submitted && current.status !== 'interrupted') {
-          current.status = 'uncertain'
-          current.error = `执行状态待确认：${error.message.slice(0, 500)}（正在自动核对，不会重复提交）`
-          this.store.saveRun(owner, current)
+      try {
+        const current = this.getWork(owner, run.id)
+        if (error) {
+          current.status = current.status === 'interrupted' ? 'interrupted' : submitted ? 'uncertain' : 'failed'
+          current.error = error.message.slice(0, 1000)
+          resultMessage.status = current.status
+          resultMessage.error = current.error
+          const optionalUnpublished = current.replyMode === 'automatic' && !current.requiredReply && resultMessage.visible === false
+          resultMessage.visible = !optionalUnpublished && (submitted || current.status !== 'interrupted')
+          if (!resultMessage.content.trim() && current.status === 'failed') resultMessage.content = `执行失败：${current.error}`
+        } else {
+          const silent = (resultMessage.content.trim() === NO_REPLY || !resultMessage.content.trim()) && current.replyMode === 'automatic' && (!current.requiredReply || current.hadInteraction) && !resultMessage.tools.length && !resultMessage.attachments.length
+          current.silent = silent
+          if (current.replyMode === 'automatic') {
+            resultMessage.content = resultMessage.content.replace(/\[\[YAOYAO_[A-Z0-9_]*(?:\]\])?/g, '').trim()
+            resultMessage.reasoning = resultMessage.reasoning.replace(/\[\[YAOYAO_[A-Z0-9_]*(?:\]\])?/g, '')
+          }
+          if (current.requiredReply && !silent && !resultMessage.content) resultMessage.content = HOST_FALLBACK
+          if (!silent && !resultMessage.content && !resultMessage.attachments.length && !resultMessage.tools.length) {
+            current.status = 'failed'; current.error = '未返回有效回复'; resultMessage.content = '执行失败：未返回有效回复'
+          } else { current.status = 'complete'; current.error = undefined }
+          resultMessage.visible = !silent
+          if (silent) resultMessage.reasoning = ''
+          resultMessage.status = current.status
+          resultMessage.error = current.error
         }
-        this.store.saveMessage(owner, resultMessage)
-        rejectTurn(error)
-      } else {
-        resultMessage.status = 'complete'
-        this.store.saveMessage(owner, resultMessage)
-        resolveTurn(resultMessage)
-      }
-      gateway.close()
+        if (current.status !== 'uncertain') {
+          for (const tool of resultMessage.tools) {
+            if (!/complete|error|failed/.test(String(tool.status))) tool.status = current.status === 'complete' ? 'tool.complete' : 'tool.error'
+          }
+        }
+        this.store.atomic(() => {
+          this.store.saveMessage(owner, resultMessage)
+          this.saveWork(owner, current)
+          if (current.status !== 'uncertain') this.resolveInteractions(owner, current.id)
+        })
+        if (error) rejectTurn(error)
+        else resolveTurn(resultMessage)
+      } catch (failure) { rejectTurn(failure instanceof Error ? failure : new Error('无法保存执行状态')) }
+      finally { gateway.close() }
     }
     gateway.onDisconnect = () =>
       finish(completedEvidence ? undefined : new Error('Hermes 连接断开'))
@@ -380,12 +218,16 @@ export class WorkspaceRuntime {
       if (settled || !runtimeId || frame.session_id !== runtimeId) return
       const p = frame.payload ?? {},
         type = frame.type
-      if (type === 'message.delta') resultMessage.content += String(p.text ?? p.delta ?? '')
+      if (type === 'message.delta') {
+        resultMessage.content += String(p.text ?? p.delta ?? '')
+        if (resultMessage.content.trim() && !NO_REPLY.startsWith(resultMessage.content.trim()) && !resultMessage.content.trim().startsWith('[[YAOYAO_')) resultMessage.visible = true
+      }
       else if (type === 'reasoning.delta')
         resultMessage.reasoning += String(p.text ?? p.delta ?? '')
       else if (type === 'message.interim') {
         resultMessage.content += '\n\n'
       } else if (type.startsWith('tool.')) {
+        resultMessage.visible = true
         const id = String(p.tool_id ?? p.id ?? ''),
           index = resultMessage.tools.findIndex((t) => t.id === id)
         const value = { ...p, id, status: type }
@@ -397,10 +239,15 @@ export class WorkspaceRuntime {
         )
       ) {
         const upstreamId = String(p.request_id ?? p.requestId ?? p.id ?? '')
-        const interaction: WorkspaceInteraction = {
+        const previous = this.store.list<WorkspaceInteraction>(owner, 'interaction').find(i => {
+          const b = this.store.get<{ taskId: string; upstreamId: string }>(owner, 'interaction-binding', i.id)
+          return b?.taskId === run.id && b.upstreamId === upstreamId
+        })
+        if (previous?.resolved) return
+        const interaction: WorkspaceInteraction = previous ?? {
           id: randomUUID(),
           conversationId: c.id,
-          runId: run.id,
+          runId: run.runId,
           agentId: agent.id,
           kind: type.startsWith('approval') ? 'approval' : 'clarification',
           message: String(p.question ?? p.message ?? p.prompt ?? '需要确认'),
@@ -408,18 +255,20 @@ export class WorkspaceRuntime {
           resolved: false,
         }
         this.store.put(owner, 'interaction', interaction.id, interaction)
-        this.store.put(owner, 'interaction-binding', interaction.id, { key, upstreamId })
+        this.store.put(owner, 'interaction-binding', interaction.id, { key, upstreamId, taskId: run.id })
         this.store.event(owner, 'interaction.changed', interaction, c.id)
-        const current = this.store.require<Run>(owner, 'run', run.id)
+        const current = this.getWork(owner, run.id)
+        current.hadInteraction = true
         current.status = 'waiting'
-        this.store.saveRun(owner, current)
-        this.notify(owner, c, current, undefined, interaction)
+        this.saveWork(owner, current)
+        this.notify(owner, c, this.store.require<Run>(owner, 'run', run.runId), undefined, interaction)
       } else if (['message.complete', 'run.completed'].includes(type)) {
         if (typeof p.text === 'string') resultMessage.content = p.text
+        if (typeof p.reasoning === 'string') resultMessage.reasoning = p.reasoning
         if (p.status === 'interrupted') {
-          const current = this.store.require<Run>(owner, 'run', run.id)
+          const current = this.getWork(owner, run.id)
           current.status = 'interrupted'
-          this.store.saveRun(owner, current)
+          this.saveWork(owner, current)
           finish(new Error('已停止'))
         } else if (p.error || p.status === 'failed') {
           submitted = false
@@ -522,42 +371,49 @@ export class WorkspaceRuntime {
             storedId,
           ]),
         ],
-        runId: run.id,
+        runId: run.runId,
+        taskId: run.id,
+        contextSeq: binding?.contextSeq ?? 0,
         messageId: resultMessage.id,
       }
       this.store.put(owner, 'binding', key, binding)
-      this.live.set(key, { gateway, runtimeId, runId: run.id, agentId: agent.id, done: finish })
-      if (this.closing || this.store.require<Run>(owner, 'run', run.id).status === 'interrupted')
+      this.live.set(key, { gateway, runtimeId, runId: run.runId, taskId: run.id, agentId: agent.id, done: finish })
+      if (this.closing || this.getWork(owner, run.id).cancelRequested || this.getWork(owner, run.id).status === 'interrupted')
         throw new Error('运行已停止')
       if (recovering) {
         if (opened.running) {
-          const current = this.store.require<Run>(owner, 'run', run.id)
+          const current = this.getWork(owner, run.id)
           current.status = 'running'
-          this.store.saveRun(owner, current)
+          this.saveWork(owner, current)
         } else {
-          const response = await target.session.request(
-            `/api/sessions/${encodeURIComponent(storedId)}/messages`,
-            {
-              search: new URLSearchParams({
-                profile: agent.profile,
-                limit: '500',
-                order: 'latest',
-                include_compacted: 'true',
-              }),
-            },
-          )
-          if (response.status !== 200) throw new Error('无法核对历史')
-          const history = JSON.parse(response.body.toString()).messages ?? []
-          const marker = `[yaoyao-run:${run.id}:${resultMessage.id}]`
-          const index = history.findLastIndex(
-            (m: any) => m.role === 'user' && String(m.content ?? m.text).includes(marker),
-          )
-          const after = index >= 0 ? history.slice(index + 1) : []
-          const nextUser = after.findIndex((m: any) => m.role === 'user')
-          const answer = (nextUser >= 0 ? after.slice(0, nextUser) : after)
-            .filter((m: any) => m.role === 'assistant' && (m.content || m.text)).at(-1)
-          if (!answer) throw new Error('无法确认本轮是否执行完成，请检查后停止此轮')
+          const marker = `[yaoyao-run:${run.runId}:${resultMessage.id}]`
+          let answer: { content?: string; text?: string } | undefined
+          let found = false, seenLastEvent = false
+          for (let page = 0; page < 20 && !found; page++) {
+            const response = await target.session.request(`/api/sessions/${encodeURIComponent(storedId)}/messages`, {
+              search: new URLSearchParams({ profile: agent.profile, limit: '500', offset: String(page * 500), order: 'latest', include_compacted: 'true' }),
+              maxResponseBytes: 8 * 1024 * 1024,
+            })
+            if (response.status !== 200) throw new Error('无法核对历史')
+            const history = JSON.parse(response.body.toString()).messages
+            if (!Array.isArray(history)) throw new Error('历史响应无效')
+            for (const item of [...history].reverse()) {
+              if (item.role === 'user') {
+                if (String(item.content ?? item.text).includes(marker)) { found = true; break }
+                answer = undefined; seenLastEvent = false
+              } else if (!seenLastEvent && ['assistant', 'tool'].includes(item.role)) {
+                seenLastEvent = true
+                const calls = item.tool_calls
+                const hasCalls = Array.isArray(calls) ? calls.length > 0 : !!calls && calls !== '[]'
+                if (item.role === 'assistant' && !hasCalls && !item.function_call && item.finish_reason !== 'tool_calls' && (item.content || item.text)) answer = item
+              }
+            }
+            if (history.length < 500) break
+          }
+          if (!found || !answer) throw new Error('无法确认本轮的最终回复，正在核对原执行；不会重复提交')
           resultMessage.content = String(answer.content ?? answer.text)
+          binding.contextSeq = Math.max(binding.contextSeq ?? 0, run.contextThroughSeq ?? 0)
+          this.store.put(owner, 'binding', key, binding)
           finish()
         }
       } else {
@@ -592,30 +448,34 @@ export class WorkspaceRuntime {
           c.kind === 'group'
             ? `你正在群聊「${c.name}」发言。群成员：${members.map((a) => `@${a.name} (id=${a.id})`).join('、')}。\n群规则：${c.instructions}\n${c.mode === 'host' ? (agent.id === c.administratorId ? '你是管理员。必要时用精确 @成员名称 委派工作；收到结果后复核并给用户结论。任务完成时不要继续 @。' : '执行当前委派任务。公开给出结果，由管理员复核；不要安排其他成员。') : '按自己的职责回复，只在需要协作时 @成员。不要重复已完成的工作。'}`
             : '',
+          run.requiredReply ? '你必须公开处理本次消息，直接回答、委派或澄清；禁止静默。管理员可按依赖一次 @一人，也可同时 @多人并行执行，整批结束后系统统一交回复核。' : run.replyMode === 'automatic' ? `你按自动参与配置收到消息。若与职责无关，禁止调用工具、禁止 @，完整答复只能是 ${NO_REPLY}。有关时正常回答。` : '',
+          `本轮用户指定成员：${this.store.require<Run>(owner, 'run', run.runId).mentionIds.map(id => members.find(a => a.id === id)).filter(Boolean).map(a => '@' + a!.name).join('、') || '未指定'}`,
           '角色规则不赋予额外工具权限；仍遵守基础 Hermes 的工具和安全约束。',
-          `[yaoyao-run:${run.id}:${resultMessage.id}]`,
+          `[yaoyao-run:${run.runId}:${resultMessage.id}]`,
         ]
           .filter(Boolean)
           .join('\n\n')
-        const text =
-          c.kind === 'group'
-            ? this.store
-                .messages(owner, c.id)
-              .filter((m) => !['queued','streaming'].includes(m.status))
-                .map(
-                  (m) => `${m.role === 'user' ? '用户' : (m.agentName ?? 'Agent')}：${m.content}`,
-                )
-                .join('\n\n')
-            : trigger.content
+        const text = c.kind === 'group' ? this.contextText(owner, c, agent, run, binding.contextSeq ?? 0) : trigger.content
+        const admission = this.getWork(owner, run.id)
+        if (!this.store.require<Conversation>(owner, 'conversation', c.id).memberIds.includes(agent.id)) {
+          admission.status = 'interrupted'; admission.error = '执行前成员已移除'; this.saveWork(owner, admission)
+          throw new Error('执行前成员已移除')
+        }
+        if (admission.cancelRequested || admission.status === 'interrupted') throw new Error('运行已停止')
+        admission.contextThroughSeq = run.contextThroughSeq ?? run.triggerSeq
+        admission.submitted = true
+        this.saveWork(owner, admission)
         submitted = true
         try {
           await gateway.rpc('prompt.submit', {
             session_id: runtimeId,
             text: `${rules}\n\n${text}\n${attachmentRefs.join('\n')}`,
           })
+          binding.contextSeq = Math.max(binding.contextSeq ?? 0, admission.contextThroughSeq)
+          this.store.put(owner, 'binding', key, binding)
         } catch (error) {
           // A JSON-RPC error is a definitive rejection, not a lost receipt.
-          if (error instanceof HttpError && error.code === 'gateway_rejected') submitted = false
+          if (error instanceof HttpError && error.code === 'gateway_rejected') { submitted = false; const current = this.getWork(owner, run.id); current.submitted = false; this.saveWork(owner, current) }
           throw error
         }
       }
@@ -625,121 +485,109 @@ export class WorkspaceRuntime {
       return await completion
     }
   }
+  private contextText(owner: string, c: Conversation, agent: Agent, work: Work, after: number): string {
+    const rootTrigger = this.store.require<Message>(owner, 'message', work.messageId)
+    const roots = new Map(this.store.list<Run>(owner, 'run').filter(r => r.conversationId === c.id).map(r => [r.id, this.store.get<Message>(owner, 'message', r.messageId)?.seq ?? 0]))
+    const history = this.store.messages(owner, c.id, Number.MAX_SAFE_INTEGER, Number.MAX_SAFE_INTEGER, true)
+    const unfinished = history.find(m => m.seq > after && m.seq <= work.triggerSeq && (['queued', 'streaming', 'uncertain'].includes(m.status) || (m.runId && (roots.get(m.runId) ?? 0) > rootTrigger.seq)))
+    work.contextThroughSeq = unfinished ? unfinished.seq - 1 : work.triggerSeq
+    const eligible = history
+      .filter(m => m.seq > after && m.seq <= work.triggerSeq && m.visible !== false && ['complete', 'failed', 'interrupted'].includes(m.status)
+        && !(m.role === 'assistant' && m.agentId === agent.id) && (!m.runId || (roots.get(m.runId) ?? 0) <= rootTrigger.seq))
+    const selected: string[] = []
+    let size = 0, omitted = 0
+    for (const message of [...eligible].reverse()) {
+      const files = message.attachments.map(f => `[附件 ${f.name}](${f.sourcePath || `/api/app/files/${f.id}/download`})`).join('\n')
+      const content = message.content.length > 24_000 ? message.content.slice(0, 12_000) + '\n[内容过长，保留首尾片段]\n' + message.content.slice(-12_000) : message.content
+      const reasoning = message.reasoning ? `\n思考摘要：${message.reasoning.slice(0, 2000)}` : ''
+      const tools = [...new Set(message.tools.map(t => String(t.name ?? t.tool_name ?? '')).filter(name => /^[A-Za-z0-9][A-Za-z0-9_.-]{0,119}$/.test(name)))].slice(0, 16)
+      const row = `${message.role === 'user' ? '用户' : message.agentName ?? '系统'}（${message.status}）：${content}${reasoning}${tools.length ? `\n使用工具：${tools.join('、')}` : ''}${message.error ? `\n错误：${message.error}` : ''}${files ? '\n' + files : ''}`
+      if (selected.length >= 50 || size + row.length > 30_000) { omitted++; continue }
+      selected.unshift(row); size += row.length
+    }
+    // Preserve a large trigger as a bounded excerpt rather than silently losing it.
+    if (!selected.length && eligible.length) selected.push(eligible.at(-1)!.content.slice(-30_000))
+    const batch = work.reviewOf ? this.works(owner, work.runId).filter(t => t.batchId === work.reviewOf).map(t => `${this.store.get<Agent>(owner, 'agent', t.agentId)?.name ?? t.agentId}：${t.status}${t.error ? `；${t.error}` : ''}`).join('\n') : ''
+    return `${batch ? `本批次执行结果：\n${batch}\n\n` : ''}${omitted ? `较早上下文有 ${omitted} 条因长度限制省略，请勿假定已完整读取。\n\n` : ''}${selected.join('\n\n')}`
+  }
+  private resolveInteractions(owner: string, taskId: string): void {
+    for (const interaction of this.store.list<WorkspaceInteraction>(owner, 'interaction')) {
+      const binding = this.store.get<{ taskId?: string }>(owner, 'interaction-binding', interaction.id)
+      if (binding?.taskId === taskId && !interaction.resolved) {
+        interaction.resolved = true
+        this.store.put(owner, 'interaction', interaction.id, interaction)
+        this.store.event(owner, 'interaction.changed', interaction, interaction.conversationId)
+      }
+    }
+  }
   async respond(owner: string, id: string, answer: string): Promise<void> {
-    const interaction = this.store.require<WorkspaceInteraction>(owner, 'interaction', id)
-    if (interaction.resolved) return
-    const b = this.store.require<{ key: string; upstreamId: string }>(
-        owner,
-        'interaction-binding',
-        id,
-      ),
-      live = this.live.get(b.key)
-    if (!live || live.runId !== interaction.runId)
-      throw new HttpError(409, '请先恢复此轮连接', 'interaction_offline')
-    if (
-      interaction.kind === 'approval' &&
-      !['once', 'always', 'deny', 'allow', 'approve'].includes(answer)
-    )
-      throw new HttpError(400, '审批选项无效', 'invalid_approval')
-    await live.gateway.rpc(
-      interaction.kind === 'approval' ? 'approval.respond' : 'clarify.respond',
-      {
-        session_id: live.runtimeId,
-        request_id: b.upstreamId,
+    type Reply = WorkspaceInteraction & { answer?: string; responseState?: 'sending' | 'uncertain' | 'sent' }
+    const interaction = this.store.require<Reply>(owner, 'interaction', id)
+    if (interaction.resolved) {
+      if (interaction.answer && interaction.answer !== answer) throw new HttpError(409, '该请求已使用其他答复完成', 'interaction_answer_conflict')
+      return
+    }
+    if (interaction.responseState === 'sending' || interaction.responseState === 'uncertain') throw new HttpError(409, '答复状态待确认，请核对原请求', 'interaction_uncertain')
+    const binding = this.store.require<{ key: string; upstreamId: string; taskId: string }>(owner, 'interaction-binding', id)
+    const live = this.live.get(binding.key)
+    if (!live || live.taskId !== binding.taskId || live.runId !== interaction.runId) throw new HttpError(409, '请先恢复此轮连接', 'interaction_offline')
+    if (interaction.kind === 'approval' && !['once', 'session', 'always', 'deny', 'allow', 'approve'].includes(answer)) throw new HttpError(400, '审批选项无效', 'invalid_approval')
+    interaction.answer = answer; interaction.responseState = 'sending'
+    this.store.put(owner, 'interaction', id, interaction)
+    try {
+      await live.gateway.rpc(interaction.kind === 'approval' ? 'approval.respond' : 'clarify.respond', {
+        session_id: live.runtimeId, request_id: binding.upstreamId,
         ...(interaction.kind === 'approval' ? { choice: answer } : { answer }),
-      },
-    )
-    interaction.resolved = true
+      })
+    } catch (error) {
+      const latest = this.store.require<Reply>(owner, 'interaction', id)
+      if (!latest.resolved) {
+        latest.responseState = error instanceof HttpError && error.code === 'gateway_rejected' ? undefined : 'uncertain'
+        this.store.put(owner, 'interaction', id, latest)
+      }
+      throw error
+    }
+    interaction.resolved = true; interaction.responseState = 'sent'
     this.store.put(owner, 'interaction', id, interaction)
     this.store.event(owner, 'interaction.changed', interaction, interaction.conversationId)
-    const run = this.store.require<Run>(owner, 'run', interaction.runId)
-    run.status = 'running'
-    this.store.saveRun(owner, run)
+    const current = this.getWork(owner, binding.taskId)
+    if (current.status === 'waiting') { current.status = 'running'; this.saveWork(owner, current) }
   }
-  async stop(owner: string, id: string): Promise<void> {
-    const run = this.store.require<Run>(owner, 'run', id)
-    if (['complete', 'failed', 'interrupted'].includes(run.status)) return
-    let live = [...this.live.values()].find((t) => t.runId === id)
-    if (live) {
+  protected async interruptTurn(owner: string, work: Work): Promise<void> {
+    const key = `${work.conversationId}:${work.agentId}`
+    const live = this.live.get(key)
+    if (live?.taskId === work.id) {
       await live.gateway.rpc('session.interrupt', { session_id: live.runtimeId })
-      run.status = 'interrupted'
-      this.store.saveRun(owner, run)
+      const current = this.getWork(owner, work.id)
+      if (['complete', 'failed', 'interrupted'].includes(current.status)) return
+      current.status = 'interrupted'; current.error = '已停止'
+      this.saveWork(owner, current)
       live.done(new Error('已停止'))
     } else {
-      const bindings = this.store
-        .list<WorkspaceBinding>(owner, 'binding')
-        .filter((b) => b.runId === id)
-      for (const b of bindings) {
-        const g = new WorkspaceGateway(this.nodes.target(owner, b.nodeId))
+      const binding = this.store.get<WorkspaceBinding>(owner, 'binding', key)
+      if (work.submitted) {
+        if (!binding || binding.taskId !== work.id) throw new Error('无法确认待停止成员的会话身份')
+        const gateway = new WorkspaceGateway(this.nodes.target(owner, binding.nodeId))
         try {
-          await g.connect()
-          const s = await g.rpc('session.resume', {
-            profile: b.profile,
-            session_id: b.storedId,
-            close_on_disconnect: false,
-            omit_messages: true,
-          })
-          if (s.running) await g.rpc('session.interrupt', { session_id: s.session_id })
-        } finally {
-          g.close()
-        }
+          await gateway.connect()
+          const opened = await gateway.rpc('session.resume', { profile: binding.profile, session_id: binding.storedId, omit_messages: true, close_on_disconnect: false })
+          if (opened.running) await gateway.rpc('session.interrupt', { session_id: opened.session_id })
+        } finally { gateway.close() }
+      }
+      const current = this.getWork(owner, work.id)
+      current.status = 'interrupted'; current.error = '已停止'
+      this.saveWork(owner, current)
+      if (current.currentMessageId) {
+        const message = this.store.require<Message>(owner, 'message', current.currentMessageId)
+        message.status = 'interrupted'; this.store.saveMessage(owner, message)
       }
-      run.status = 'interrupted'
-      this.store.saveRun(owner, run)
-    }
-    for (const i of this.store
-      .list<WorkspaceInteraction>(owner, 'interaction')
-      .filter((i) => i.runId === id && !i.resolved)) {
-      i.resolved = true
-      this.store.put(owner, 'interaction', i.id, i)
-      this.store.event(owner, 'interaction.changed', i, i.conversationId)
-    }
-  }
-  private scheduleReconcile(owner: string, id: string): void {
-    if (this.closing || !this.userActive(owner) || this.recoveryTimers.has(id)) return
-    const run = this.store.get<Run>(owner, 'run', id)
-    if (run?.status !== 'uncertain' || !run.currentMessageId || !run.queue.length) {
-      this.recoveryAttempts.delete(id)
-      return
-    }
-    const attempt = this.recoveryAttempts.get(id) ?? 0
-    this.recoveryAttempts.set(id, attempt + 1)
-    const timer = setTimeout(() => {
-      this.recoveryTimers.delete(id)
-      if (this.executing.has(id)) { this.scheduleReconcile(owner, id); return }
-      void this.reconcile(owner, id).catch(() => {})
-    }, Math.min(30_000, 500 * 2 ** Math.min(attempt, 6)))
-    timer.unref()
-    this.recoveryTimers.set(id, timer)
-  }
-  async reconcile(owner: string, id: string): Promise<void> {
-    if (this.executing.has(id) || this.closing || !this.userActive(owner)) return
-    const run = this.store.require<Run>(owner, 'run', id)
-    if (run.status !== 'uncertain' || !run.currentMessageId || !run.queue.length) return
-    const timer = this.recoveryTimers.get(id)
-    if (timer) clearTimeout(timer)
-    this.recoveryTimers.delete(id)
-    this.executing.add(id)
-    try {
-      const agent = this.store.require<Agent>(owner, 'agent', run.queue[0]!),
-        c = this.store.require<Conversation>(owner, 'conversation', run.conversationId)
-      const message = await this.turn(owner, c, agent, run, true)
-      this.advance(owner, id, agent.id, message)
-    } finally {
-      this.executing.delete(id)
-      this.scheduleReconcile(owner, id)
     }
-    await this.execute(owner, id)
+    this.resolveInteractions(owner, work.id)
   }
   close(): void {
-    this.closing = true
-    for (const timer of this.recoveryTimers.values()) clearTimeout(timer)
-    this.recoveryTimers.clear()
-    this.recoveryAttempts.clear()
-    for (const t of this.live.values()) {
-      t.gateway.close()
-      t.done(new Error('服务关闭'))
-    }
+    this.closeScheduler()
+    for (const live of [...this.live.values()]) live.done(new Error('服务关闭'))
     this.live.clear()
   }
+
 }

src/server/workspaceScheduler.ts

@@ -0,0 +1,328 @@
+import { randomUUID } from 'node:crypto'
+import { WorkspaceStore } from './workspaceStore.js'
+import { HttpError } from './errors.js'
+import { mentionedAgents } from './workspaceMentions.js'
+import type { WorkspaceAgent as Agent, WorkspaceConversation as Conversation, WorkspaceMessage as Message, WorkspaceRun as Run, WorkspaceInteraction } from '../shared/workspace.js'
+
+export const NO_REPLY = '[[YAOYAO_NO_REPLY_V1]]'
+export const HOST_FALLBACK = '我还不能确定你希望我处理什么，请补充具体目标、范围，或明确需要我协调的 Agent。'
+const terminal = (status: string) => ['complete', 'failed', 'interrupted'].includes(status)
+export interface Work {
+  id: string
+  runId: string
+  conversationId: string
+  agentId: string
+  messageId: string
+  triggerSeq: number
+  depth: number
+  batchId: string
+  reviewOf?: string
+  replyMode: 'mentioned' | 'automatic'
+  requiredReply: boolean
+  status: Run['status']
+  currentMessageId?: string
+  submitted?: boolean
+  cancelRequested?: boolean
+  planned?: boolean
+  silent?: boolean
+  hadInteraction?: boolean
+  error?: string
+  contextThroughSeq?: number
+  turnConfiguration?: { mode: 'host' | 'free'; administratorId: string; members: Array<Pick<Agent, 'id' | 'name'>> }
+  createdAt: number
+}
+
+/** Durable per-member work and cascade ledger; Hermes transport lives in WorkspaceRuntime. */
+export abstract class WorkspaceScheduler {
+  protected closing = false
+  protected executing = new Set<string>()
+  private wakePending = false
+  private pumping = false
+  private retryTimer?: ReturnType<typeof setTimeout>
+  private recoverAfter = new Map<string, number>()
+  private attempts = new Map<string, number>()
+  private interrupting = new Set<string>()
+  onMessage: (owner: string, message: Message) => Promise<void> = async () => {}
+  onNotify: (owner: string, c: Conversation, run: Run, message?: Message, interaction?: WorkspaceInteraction) => void = () => {}
+  constructor(readonly store: WorkspaceStore, readonly userActive: (owner: string) => boolean = () => true) {
+    for (const owner of store.owners()) {
+      // Forward the already-admitted v0.3.x turn without replaying its prompt.
+      for (const root of store.list<Run & { queue?: string[]; next?: string[]; hostReturn?: boolean; round: number; currentMessageId?: string; turnConfiguration?: Work['turnConfiguration'] }>(owner, 'run')) {
+        if (terminal(root.status) || this.works(owner, root.id).length) continue
+        const c = store.require<Conversation>(owner, 'conversation', root.conversationId)
+        const trigger = store.require<Message>(owner, 'message', root.messageId)
+        const continuation = root.queue?.length === 0
+        const ids = continuation ? root.hostReturn ? [c.administratorId] : root.next ?? [] : root.queue ?? [root.activeAgentId ?? c.administratorId]
+        if (!ids.length && root.currentMessageId) {
+          const agentId = store.get<Message>(owner, 'message', root.currentMessageId)?.agentId ?? root.activeAgentId
+          if (agentId) ids.push(agentId)
+        }
+        if (!ids.length) { root.status = root.stopRequested ? 'interrupted' : 'complete'; store.saveRun(owner, root); continue }
+        const last = store.messages(owner, c.id).filter(m => m.runId === root.id).at(-1)
+        for (const [index, id] of ids.entries()) {
+          const work = this.enqueue(owner, root, id, root.round + (continuation && !root.hostReturn ? 1 : 0), root.id, last?.seq ?? trigger.seq, 'mentioned', id === c.administratorId)
+          if (index === 0 && root.currentMessageId) {
+            Object.assign(work, { currentMessageId: root.currentMessageId, submitted: true, status: 'uncertain', turnConfiguration: root.turnConfiguration })
+            store.put(owner, 'turn', work.id, work)
+            const key = `${c.id}:${id}`
+            const binding = store.get<{ runId: string; messageId: string; taskId?: string }>(owner, 'binding', key)
+            if (binding?.runId === root.id && binding.messageId === root.currentMessageId) {
+              binding.taskId = work.id; store.put(owner, 'binding', key, binding)
+            }
+          }
+        }
+      }
+      for (const work of this.works(owner)) {
+        if (['running', 'waiting'].includes(work.status)) {
+          work.status = work.submitted ? 'uncertain' : 'queued'
+          work.error = work.submitted ? '服务已重启，正在核对原执行' : undefined
+          store.put(owner, 'turn', work.id, work)
+        }
+      }
+    }
+  }
+  protected notify(owner: string, c: Conversation, run: Run, message?: Message, interaction?: WorkspaceInteraction): void {
+    try { this.onNotify(owner, c, run, message, interaction) } catch { /* Push cannot change execution. */ }
+  }
+  protected abstract performTurn(owner: string, c: Conversation, agent: Agent, work: Work, recovering: boolean): Promise<Message>
+  protected abstract interruptTurn(owner: string, work: Work): Promise<void>
+  protected works(owner: string, runId?: string): Work[] {
+    return this.store.db.prepare(`SELECT data FROM workspace_entities WHERE owner=? AND kind='turn' ${runId ? "AND json_extract(data,'$.runId')=?" : ''}`).all(...(runId ? [owner, runId] : [owner]))
+      .map(row => JSON.parse(String(row.data)) as Work)
+  }
+  private pendingWorks(owner: string): Work[] {
+    return this.store.db.prepare("SELECT data FROM workspace_entities WHERE owner=? AND kind='turn' AND (json_extract(data,'$.status') NOT IN ('complete','failed','interrupted') OR coalesce(json_extract(data,'$.planned'),0)=0)").all(owner)
+      .map(row => JSON.parse(String(row.data)) as Work)
+  }
+  protected getWork(owner: string, id: string): Work { return this.store.require(owner, 'turn', id) }
+  protected saveWork(owner: string, work: Work): void {
+    this.store.atomic(() => { this.store.put(owner, 'turn', work.id, work); this.updateRoot(owner, work.runId) })
+  }
+  protected enqueue(owner: string, root: Run, agentId: string, depth: number, batchId: string, triggerSeq: number, replyMode: Work['replyMode'], requiredReply: boolean, reviewOf?: string): Work {
+    const existing = this.works(owner, root.id).find(w => reviewOf ? w.reviewOf === reviewOf : w.agentId === agentId && w.depth === depth && !w.reviewOf)
+    if (existing) return existing
+    const work: Work = { id: randomUUID(), runId: root.id, conversationId: root.conversationId, agentId, messageId: root.messageId, depth, batchId, triggerSeq, replyMode, requiredReply, reviewOf, status: 'queued', createdAt: Date.now() }
+    this.store.put(owner, 'turn', work.id, work)
+    return work
+  }
+  protected admit(owner: string, root: Run, c: Conversation, trigger: Message): void {
+    this.store.put(owner, 'run', root.id, root)
+    const ids = c.kind === 'direct' || c.mode === 'host' ? [c.administratorId]
+      : root.mentionIds.length ? root.mentionIds : [...new Set([c.administratorId, ...c.autoReplyIds])]
+    for (const id of ids) this.enqueue(owner, root, id, 0, root.id, trigger.seq,
+      c.kind === 'direct' ? 'mentioned' : c.mode === 'host' || !root.mentionIds.length ? 'automatic' : 'mentioned', c.kind === 'group' && id === c.administratorId && (c.mode === 'host' || !root.mentionIds.length))
+    this.updateRoot(owner, root.id)
+  }
+  start(): void { this.wake() }
+  wake(): void {
+    if (this.closing || this.wakePending) return
+    this.wakePending = true
+    queueMicrotask(() => { this.wakePending = false; if (!this.closing) this.pump() })
+  }
+  private retrySoon(): void {
+    if (this.closing || this.retryTimer) return
+    this.retryTimer = setTimeout(() => { this.retryTimer = undefined; this.wake() }, 500)
+    this.retryTimer.unref()
+  }
+  private updateRoot(owner: string, id: string): void {
+    const root = this.store.require<Run>(owner, 'run', id), tasks = this.works(owner, id)
+    const active = tasks.filter(t => !terminal(t.status))
+    const done = !active.length && tasks.every(t => t.planned)
+    const wasTerminal = terminal(root.status)
+    root.round = Math.max(0, ...tasks.map(t => t.depth))
+    root.activeAgentId = active.find(t => t.status !== 'queued')?.agentId ?? active[0]?.agentId
+    if (done) {
+      const interrupted = root.stopRequested || tasks.some(t => t.requiredReply && t.status === 'interrupted') || (tasks.length > 0 && tasks.every(t => t.status === 'interrupted'))
+      const failed = tasks.some(t => t.status === 'failed' && (t.requiredReply || this.store.require<Conversation>(owner, 'conversation', root.conversationId).kind === 'direct'))
+      root.status = interrupted ? 'interrupted' : failed ? 'failed' : 'complete'
+    } else {
+      root.status = active.some(t => t.status === 'uncertain') ? 'uncertain' : active.some(t => t.status === 'running') ? 'running' : active.some(t => t.status === 'waiting') ? 'waiting' : tasks.some(t => t.status !== 'queued') ? 'running' : 'queued'
+    }
+    root.error = tasks.find(t => t.status === 'uncertain' || t.status === 'failed')?.error
+    this.store.saveRun(owner, root)
+    if (done && !wasTerminal && root.status !== 'interrupted') {
+      const c = this.store.require<Conversation>(owner, 'conversation', root.conversationId)
+      if (root.status === 'failed' && c.kind === 'direct') this.store.saveMessage(owner, { id: randomUUID(), conversationId: c.id, seq: 0, role: 'system', content: `执行失败：${root.error ?? '运行失败'}`, reasoning: '', status: 'failed', runId: root.id, attachments: [], tools: [], createdAt: Date.now() })
+      this.notify(owner, c, root, this.store.messages(owner, c.id).filter(m => m.runId === id && m.visible !== false).at(-1))
+    }
+  }
+  private plan(owner: string, id: string): void {
+    this.store.atomic(() => {
+      const work = this.getWork(owner, id)
+      if (!terminal(work.status) || work.planned) return
+      const root = this.store.require<Run>(owner, 'run', work.runId), c = this.store.require<Conversation>(owner, 'conversation', work.conversationId)
+      work.planned = true
+      this.store.put(owner, 'turn', work.id, work)
+      if (root.stopRequested || c.archived || c.kind === 'direct') { this.updateRoot(owner, root.id); return }
+      const config = work.turnConfiguration ?? { mode: c.mode, administratorId: c.administratorId, members: c.memberIds.map(id => this.store.require<Agent>(owner, 'agent', id)) }
+      const message = work.currentMessageId ? this.store.get<Message>(owner, 'message', work.currentMessageId) : undefined
+      if (config.mode === 'host' && work.agentId !== config.administratorId) {
+        const batch = this.works(owner, root.id).filter(t => t.batchId === work.batchId && !t.reviewOf)
+        if (batch.every(t => terminal(t.status))) {
+          const through = Math.max(work.triggerSeq, ...batch.map(t => t.currentMessageId ? this.store.get<Message>(owner, 'message', t.currentMessageId)?.seq ?? 0 : 0))
+          this.enqueue(owner, root, c.administratorId, work.depth, `review:${work.batchId}`, through, 'automatic', true, work.batchId)
+        }
+      } else if (work.status === 'complete' && !work.silent && message) {
+        const nextDepth = work.depth + 1
+        const mentions = mentionedAgents(message.content, config.members).filter(id => id !== work.agentId && c.memberIds.includes(id))
+        const targets = mentions.length ? mentions : config.mode === 'free' ? c.autoReplyIds.filter(id => id !== work.agentId && c.memberIds.includes(id)) : []
+        if (c.maxReplyRounds === -1 || nextDepth < c.maxReplyRounds) {
+          for (const target of targets) this.enqueue(owner, root, target, nextDepth, work.id, message.seq, mentions.includes(target) ? 'mentioned' : 'automatic', false)
+        } else if (targets.length && !this.store.get(owner, 'limit-notice', root.id)) {
+          this.store.put(owner, 'limit-notice', root.id, { id: root.id })
+          this.store.saveMessage(owner, { id: randomUUID(), conversationId: c.id, seq: 0, role: 'system', content: '已达到自动协作轮数上限，本轮不再自动分派。', reasoning: '', status: 'complete', runId: root.id, attachments: [], tools: [], createdAt: Date.now() })
+        }
+      }
+      this.updateRoot(owner, root.id)
+    })
+  }
+  private pump(): void {
+    if (this.pumping || this.closing) return
+    this.pumping = true
+    try {
+      let all = this.store.owners().flatMap(owner => this.pendingWorks(owner).map(work => ({ owner, work })))
+      for (const { owner, work } of all) {
+        if (['running', 'waiting'].includes(work.status) && !this.executing.has(work.id)) {
+          work.status = work.submitted ? 'uncertain' : 'queued'
+          this.saveWork(owner, work)
+        }
+      }
+      for (const { owner, work } of all) if (terminal(work.status) && !work.planned && !this.executing.has(work.id)) this.plan(owner, work.id)
+      all = this.store.owners().flatMap(owner => this.pendingWorks(owner).map(work => ({ owner, work })))
+      all.sort((a, b) => {
+        const am = this.store.require<Message>(a.owner, 'message', a.work.messageId), bm = this.store.require<Message>(b.owner, 'message', b.work.messageId)
+        return am.createdAt - bm.createdAt || (a.work.conversationId === b.work.conversationId ? am.seq - bm.seq : 0) || a.work.createdAt - b.work.createdAt || a.work.id.localeCompare(b.work.id)
+      })
+      for (const { owner, work: candidate } of all) {
+        const work = this.getWork(owner, candidate.id)
+        if (work.cancelRequested && work.status === 'uncertain' && this.executing.has(work.id)) {
+          if ((this.recoverAfter.get(work.id) ?? 0) > Date.now()) { this.retrySoon(); continue }
+          if (!this.interrupting.has(work.id)) {
+            this.interrupting.add(work.id)
+            void this.interruptTurn(owner, work).catch(error => {
+              if (this.closing) return
+              const current = this.getWork(owner, work.id)
+              if (terminal(current.status)) return
+              current.status = 'uncertain'; current.error = `停止状态待确认：${error instanceof Error ? error.message : '连接中断'}`
+              this.saveWork(owner, current)
+              const attempt = this.attempts.get(work.id) ?? 0
+              this.attempts.set(work.id, attempt + 1)
+              this.recoverAfter.set(work.id, Date.now() + Math.min(30_000, 500 * 2 ** Math.min(attempt, 6)))
+              this.retrySoon()
+            }).finally(() => { this.interrupting.delete(work.id); this.wake() }).catch(() => this.retrySoon())
+          }
+          continue
+        }
+        if (this.executing.size >= 4 || !['queued', 'uncertain'].includes(work.status) || this.executing.has(work.id)) continue
+        if (!this.userActive(owner)) { this.retrySoon(); continue }
+        const c = this.store.require<Conversation>(owner, 'conversation', work.conversationId), root = this.store.require<Run>(owner, 'run', work.runId)
+        if (work.status === 'queued' && c.mode === 'host' && work.depth === 0 && work.batchId === root.id && work.requiredReply && !work.currentMessageId) {
+          work.agentId = c.administratorId
+        }
+        if (work.status === 'queued' && (c.archived || root.stopRequested || !c.memberIds.includes(work.agentId))) {
+          work.status = 'interrupted'; work.error = '执行前成员已移除或聊天已停止'; this.saveWork(owner, work); this.wake(); continue
+        }
+        const occupied = all.map(entry => this.getWork(entry.owner, entry.work.id)).filter(t => (['running', 'waiting', 'uncertain'].includes(t.status) || this.executing.has(t.id)) && t.id !== work.id)
+        if (occupied.some(t => t.conversationId === c.id && t.agentId === work.agentId)) continue
+        const inRoom = occupied.filter(t => t.conversationId === c.id)
+        if (work.status === 'queued' && (occupied.length >= 4 || inRoom.length >= 3)) continue
+        if (work.status === 'uncertain' && (this.recoverAfter.get(work.id) ?? 0) > Date.now()) { this.retrySoon(); continue }
+        if (c.mode === 'host' || c.kind === 'direct') {
+          if (inRoom.some(t => c.kind === 'direct' || t.runId !== work.runId || t.batchId !== work.batchId || t.agentId === c.administratorId || work.agentId === c.administratorId)) continue
+        }
+        this.executing.add(work.id)
+        const recovering = work.status === 'uncertain'
+        if (!recovering) {
+          work.status = 'running'
+          if (c.kind === 'group' && c.mode === 'host' && work.agentId === c.administratorId) { work.requiredReply = true; work.replyMode = 'automatic' }
+          work.turnConfiguration = { mode: c.mode, administratorId: c.administratorId, members: c.memberIds.map(id => { const a = this.store.require<Agent>(owner, 'agent', id); return { id, name: a.name } }) }
+          this.saveWork(owner, work)
+        }
+        void this.executeWork(owner, work, recovering).catch(() => this.retrySoon())
+      }
+    } catch { this.retrySoon() } finally { this.pumping = false }
+  }
+  private async executeWork(owner: string, work: Work, recovering: boolean): Promise<void> {
+    try {
+      if (work.cancelRequested) await this.interruptTurn(owner, work)
+      else {
+        const c = this.store.require<Conversation>(owner, 'conversation', work.conversationId)
+        const agent = this.store.require<Agent>(owner, 'agent', work.agentId)
+        const message = await this.performTurn(owner, c, agent, work, recovering)
+        if (message.visible !== false) await this.onMessage(owner, message).catch(() => {})
+      }
+    } catch (error) {
+      if (this.closing) return
+      const current = this.getWork(owner, work.id)
+      if (!terminal(current.status) && current.status !== 'uncertain') {
+        current.status = current.submitted ? 'uncertain' : 'failed'
+        current.error = error instanceof Error ? error.message : '执行失败'
+        this.saveWork(owner, current)
+      }
+      if (current.status === 'failed' && !current.currentMessageId) {
+        current.currentMessageId = randomUUID()
+        this.store.saveMessage(owner, { id: current.currentMessageId, conversationId: current.conversationId, seq: 0, role: 'assistant', agentId: current.agentId, agentName: this.store.get<Agent>(owner, 'agent', current.agentId)?.name, content: `执行失败：${current.error}`, reasoning: '', status: 'failed', error: current.error, taskId: current.id, runId: current.runId, attachments: [], tools: [], createdAt: Date.now() })
+        this.saveWork(owner, current)
+      }
+    } finally {
+      this.executing.delete(work.id)
+      if (!this.closing) {
+        const current = this.getWork(owner, work.id)
+        if (current.status === 'uncertain') {
+          const attempt = this.attempts.get(work.id) ?? 0
+          this.attempts.set(work.id, attempt + 1)
+          this.recoverAfter.set(work.id, Date.now() + Math.min(30_000, 500 * 2 ** Math.min(attempt, 6)))
+          this.retrySoon()
+        } else { this.attempts.delete(work.id); this.recoverAfter.delete(work.id) }
+        this.wake()
+      }
+    }
+  }
+  async reconcile(owner: string, id: string): Promise<void> {
+    this.store.require<Run>(owner, 'run', id)
+    for (const w of this.works(owner, id)) this.recoverAfter.delete(w.id)
+    this.wake()
+  }
+  async stop(owner: string, id: string): Promise<void> {
+    const root = this.store.require<Run>(owner, 'run', id)
+    if (terminal(root.status)) return
+    root.stopRequested = true
+    this.store.saveRun(owner, root)
+    await this.cancelWorks(owner, this.works(owner, id))
+  }
+  async stopAgent(owner: string, conversationId: string, agentId: string): Promise<void> {
+    const c = this.store.require<Conversation>(owner, 'conversation', conversationId)
+    if (!c.memberIds.includes(agentId)) throw new HttpError(400, '只能停止群内成员', 'invalid_members')
+    await this.cancelWorks(owner, this.works(owner).filter(w => w.conversationId === conversationId && w.agentId === agentId))
+  }
+  async stopConversation(owner: string, id: string): Promise<void> {
+    this.store.require<Conversation>(owner, 'conversation', id)
+    const roots = this.store.list<Run>(owner, 'run').filter(r => r.conversationId === id && !terminal(r.status))
+    this.store.atomic(() => { for (const root of roots) { root.stopRequested = true; this.store.saveRun(owner, root) } })
+    await this.cancelWorks(owner, this.works(owner).filter(w => roots.some(r => r.id === w.runId)))
+  }
+  private async cancelWorks(owner: string, works: Work[]): Promise<void> {
+    await Promise.all(works.filter(w => !terminal(w.status)).map(async w => {
+      const current = this.getWork(owner, w.id)
+      current.cancelRequested = true
+      if (current.status === 'queued') { current.status = 'interrupted'; this.saveWork(owner, current) }
+      else {
+        this.saveWork(owner, current)
+        try { await this.interruptTurn(owner, current) }
+        catch (error) {
+          const latest = this.getWork(owner, current.id)
+          if (terminal(latest.status)) return
+          latest.status = 'uncertain'; latest.error = `停止状态待确认：${error instanceof Error ? error.message : '连接中断'}`
+          this.saveWork(owner, latest); this.recoverAfter.delete(latest.id); this.retrySoon()
+        }
+      }
+    }))
+    this.wake()
+  }
+  protected closeScheduler(): void {
+    this.closing = true
+    if (this.retryTimer) clearTimeout(this.retryTimer)
+    this.recoverAfter.clear()
+    this.attempts.clear()
+  }
+}

src/server/workspaceStore.ts

@@ -59,7 +59,7 @@ export const groupInput = z
     administratorId: z.string().uuid(),
     mode: z.enum(['host', 'free']).default('host'),
     autoReplyIds: z.array(z.string().uuid()).max(8).default([]),
-    maxReplyRounds: z.number().int().min(1).max(100).default(3),
+    maxReplyRounds: z.union([z.literal(-1), z.number().int().min(1).max(100)]).default(3),
   })
   .strict()
 export const conversationPatch = z
@@ -72,7 +72,7 @@ export const conversationPatch = z
     administratorId: z.string().uuid().optional(),
     mode: z.enum(['host', 'free']).optional(),
     autoReplyIds: z.array(z.string().uuid()).max(8).optional(),
-    maxReplyRounds: z.number().int().min(1).max(100).optional(),
+    maxReplyRounds: z.union([z.literal(-1), z.number().int().min(1).max(100)]).optional(),
     archived: z.boolean().optional(),
     pinned: z.boolean().optional(),
   })
@@ -101,6 +101,9 @@ export class WorkspaceStore {
       CREATE TABLE IF NOT EXISTS workspace_events(seq INTEGER PRIMARY KEY AUTOINCREMENT,owner TEXT NOT NULL,type TEXT NOT NULL,conversation_id TEXT,data TEXT NOT NULL);
       CREATE INDEX IF NOT EXISTS workspace_events_owner ON workspace_events(owner,seq);
       CREATE TABLE IF NOT EXISTS workspace_commands(owner TEXT NOT NULL,id TEXT NOT NULL,fingerprint TEXT NOT NULL,result TEXT NOT NULL,PRIMARY KEY(owner,id));`)
+    this.db.exec(`CREATE INDEX IF NOT EXISTS workspace_turn_run ON workspace_entities(owner, json_extract(data,'$.runId')) WHERE kind='turn';
+      CREATE INDEX IF NOT EXISTS workspace_turn_status ON workspace_entities(owner, json_extract(data,'$.status'), json_extract(data,'$.planned')) WHERE kind='turn';
+      CREATE INDEX IF NOT EXISTS workspace_turn_conversation ON workspace_entities(owner, json_extract(data,'$.conversationId')) WHERE kind='turn';`)
     this.changes.setMaxListeners(0)
   }
   get<T>(owner: string, kind: string, id: string): T | undefined {
@@ -342,11 +345,12 @@ export class WorkspaceStore {
     conversationId: string,
     before = Number.MAX_SAFE_INTEGER,
     limit = 100,
+    includeHidden = false,
   ): Message[] {
     this.require(owner, 'conversation', conversationId)
     return this.db
       .prepare(
-        `SELECT data FROM workspace_entities WHERE owner=? AND kind='message' AND json_extract(data,'$.conversationId')=? AND json_extract(data,'$.seq')<? ORDER BY json_extract(data,'$.seq') DESC LIMIT ?`,
+        `SELECT data FROM workspace_entities WHERE owner=? AND kind='message' AND json_extract(data,'$.conversationId')=? AND json_extract(data,'$.seq')<? ${includeHidden ? '' : "AND coalesce(json_extract(data,'$.visible'),1) != 0"} ORDER BY json_extract(data,'$.seq') DESC LIMIT ?`,
       )
       .all(owner, conversationId, before, limit)
       .map((r) => JSON.parse(String(r.data)) as Message)
@@ -355,11 +359,15 @@ export class WorkspaceStore {
   saveMessage(owner: string, message: Message): void {
     this.atomic(() => {
       const c = this.require<Conversation>(owner, 'conversation', message.conversationId)
+      const previous = this.get<Message>(owner, 'message', message.id)
+      if (previous?.visible === false && message.visible !== false) message.seq = 0
       if (!message.seq) message.seq = c.lastSeq + 1
       c.lastSeq = Math.max(c.lastSeq, message.seq)
-      c.updatedAt = Date.now()
-      c.lastMessageAt = Math.max(c.lastMessageAt ?? 0, message.createdAt)
-      c.preview = notificationPlainText(message.content, { maximum: 160, fallback: '' })
+      if (message.visible !== false && (message.content.trim() || message.attachments.length)) {
+        c.updatedAt = Date.now()
+        c.lastMessageAt = Math.max(c.lastMessageAt ?? 0, message.createdAt)
+        c.preview = notificationPlainText(message.content, { maximum: 160, fallback: '' })
+      }
       this.put(owner, 'message', message.id, message)
       this.put(owner, 'conversation', c.id, c)
       this.event(owner, 'message.changed', message, c.id)
@@ -373,21 +381,31 @@ export class WorkspaceStore {
     const member = conversation.kind === 'direct' ? this.get<Agent>(owner, 'agent', conversation.memberIds[0] ?? '') : undefined
     return {
       ...conversation,
+      unreadCount: this.messages(owner, conversation.id, Number.MAX_SAFE_INTEGER, Number.MAX_SAFE_INTEGER).filter(m => m.seq > conversation.readSeq).length,
       ...(member ? { avatar: this.agentSummary(member).avatar } : {}),
       lastMessageAt: conversation.lastMessageAt
         ?? (conversation.lastSeq > 0 ? this.messages(owner, conversation.id, Number.MAX_SAFE_INTEGER, 1).at(-1)?.createdAt : undefined)
         ?? conversation.createdAt,
     }
   }
+  hiddenMessageIds(owner: string, conversationId: string): string[] {
+    return this.messages(owner, conversationId, Number.MAX_SAFE_INTEGER, Number.MAX_SAFE_INTEGER, true).filter(m => m.visible === false).map(m => m.id)
+  }
   saveRun(owner: string, run: Run): void {
     this.atomic(() => {
       const c = this.require<Conversation>(owner, 'conversation', run.conversationId)
-      const terminal = ['complete', 'failed', 'interrupted'].includes(run.status)
-      c.activeRunId = terminal ? undefined : run.id
-      c.activeAgentId = terminal ? undefined : run.activeAgentId
-      c.activeRunStatus = terminal ? undefined : run.status
       run.updatedAt = Date.now()
       this.put(owner, 'run', run.id, run)
+      const roots = this.list<Run>(owner, 'run').filter(r => r.conversationId === c.id && !['complete', 'failed', 'interrupted'].includes(r.status))
+        .sort((a, b) => this.require<Message>(owner, 'message', a.messageId).seq - this.require<Message>(owner, 'message', b.messageId).seq)
+      const current = roots[0]
+      c.activeRunId = current?.id
+      c.activeAgentId = current?.activeAgentId
+      c.activeRunStatus = current?.status
+      const tasks = this.db.prepare("SELECT data FROM workspace_entities WHERE owner=? AND kind='turn' AND json_extract(data,'$.conversationId')=? AND json_extract(data,'$.status') IN ('running','waiting','uncertain')").all(owner, c.id)
+        .map(row => JSON.parse(String(row.data)) as { agentId: string; status: string })
+      c.activeAgentStates = Object.fromEntries(tasks.map(t => [t.agentId, t.status as 'running' | 'waiting' | 'uncertain']))
+      c.queuedMessageCount = roots.filter(r => r.status === 'queued').length
       this.put(owner, 'conversation', c.id, c)
       this.event(owner, 'run.changed', run, c.id)
       this.event(owner, 'conversation.changed', c, c.id)

src/shared/workspace.ts

@@ -36,6 +36,9 @@ export interface WorkspaceConversation {
   activeRunId?: string
   activeAgentId?: string
   activeRunStatus?: WorkspaceRun['status']
+  activeAgentStates?: Record<string, 'running' | 'waiting' | 'uncertain'>
+  queuedMessageCount?: number
+  unreadCount?: number
   createdAt: number
   updatedAt: number
 }
@@ -61,6 +64,9 @@ export interface WorkspaceMessage {
   content: string
   reasoning: string
   runId?: string
+  taskId?: string
+  visible?: boolean
+  error?: string
   status: 'queued' | 'streaming' | 'complete' | 'failed' | 'interrupted' | 'uncertain'
   attachments: WorkspaceFile[]
   tools: Array<Record<string, unknown>>
@@ -75,6 +81,7 @@ export interface WorkspaceRun {
   status: 'queued' | 'running' | 'waiting' | 'complete' | 'failed' | 'interrupted' | 'uncertain'
   round: number
   error?: string
+  stopRequested?: boolean
   createdAt: number
   updatedAt: number
 }

tests/e2e/workspace-chat.spec.ts

@@ -74,12 +74,12 @@ test('created roles and editable teams share a durable chat list without plugin
   await dialog.getByRole('button', { name: '取消', exact: true }).click()
   await page.getByPlaceholder('输入消息，Enter 发送，Shift + Enter 换行').fill('请@开发工程师给出实现结果')
   await page.getByRole('button', { name: '发送消息', exact: true }).click()
-  await expect(page.locator('.message--assistant')).toHaveCount(2)
-  await expect(page.locator('.message--assistant').first()).toContainText('我是开发工程师')
+  await expect(page.locator('.message--assistant')).toHaveCount(3)
+  await expect(page.locator('.message--assistant').nth(1)).toContainText('我是开发工程师')
   await expect(page.locator('.run-status')).toHaveCount(0)
   const second = await context.newPage()
   await second.goto(page.url())
-  await expect(second.locator('.message--assistant')).toHaveCount(2)
+  await expect(second.locator('.message--assistant')).toHaveCount(3)
   await page.screenshot({ path: testInfo.outputPath('mixed-chats.png'), fullPage: true })
   const groupURL=page.url()
   await page.getByRole('button', { name: '聊天设置', exact: true }).click()
@@ -87,7 +87,7 @@ test('created roles and editable teams share a durable chat list without plugin
   await expect(dialog.getByRole('checkbox', { name: '产品经理', exact: true })).toBeDisabled()
   await dialog.getByRole('button', { name: '保存', exact: true }).click()
   await expect(groupRow.locator('.team-avatar__member')).toHaveCount(1)
-  await expect(page.locator('.message--assistant')).toHaveCount(2)
+  await expect(page.locator('.message--assistant')).toHaveCount(3)
   await expect(second.locator('.desktop-sidebar .sidebar-item').filter({ hasText: '产品开发团队' }).locator('.team-avatar__member')).toHaveCount(1)
   await page.reload()
   await page.getByRole('button', { name: '聊天设置', exact: true }).click()
@@ -105,8 +105,8 @@ test('created roles and editable teams share a durable chat list without plugin
   await expect(groupRow.locator('[aria-label="新成员 的头像"]')).toHaveCount(1)
   await page.getByPlaceholder('输入消息，Enter 发送，Shift + Enter 换行').fill('请@新成员继续')
   await page.getByRole('button', { name: '发送消息', exact: true }).click()
-  await expect(page.locator('.message--assistant')).toHaveCount(4)
-  await expect(page.locator('.message--assistant').nth(2)).toContainText('我是新成员')
+  await expect(page.locator('.message--assistant')).toHaveCount(6)
+  await expect(page.locator('.message--assistant').nth(4)).toContainText('我是新成员')
   await page.getByRole('button',{name:'文件库',exact:true}).click()
   await page.getByRole('button',{name:'查看来源',exact:true}).first().click()
   await expect(page).toHaveURL(groupURL)
@@ -153,5 +153,17 @@ test('created roles and editable teams share a durable chat list without plugin
   await page.getByPlaceholder('输入消息，Enter 发送，Shift + Enter 换行').fill('请汇总设计方案')
   await page.getByRole('button', { name: '发送消息', exact: true }).click()
   await expect(page.locator('.message--assistant')).toContainText('我是产品经理')
+  const composer = page.getByPlaceholder('输入消息，Enter 发送，Shift + Enter 换行')
+  await composer.fill('[hold-workspace] 正在执行的任务')
+  await page.getByRole('button', { name: '发送消息', exact: true }).click()
+  await expect(page.getByRole('button', { name: '停止生成', exact: true })).toBeVisible()
+  await composer.fill('后续排队任务')
+  await expect(page.getByRole('button', { name: '发送消息', exact: true })).toBeEnabled()
+  await page.getByRole('button', { name: '发送消息', exact: true }).click()
+  await expect(page.locator('.message--user').last()).toContainText('后续排队任务')
+  const detail = await (await page.request.get(new URL(page.url()).pathname.replace('/conversations/', '/api/app/conversations/'))).json()
+  expect(detail.conversation.queuedMessageCount).toBe(1)
+  await page.request.post('http://127.0.0.1:19120/__release')
+  await expect.poll(async () => (await (await page.request.get(`/api/app/conversations/${detail.conversation.id}`)).json()).conversation.activeRunId).toBeUndefined()
   expect(pluginCalls).toEqual([])
 })

tests/fixtures/workspace-server.ts

@@ -23,10 +23,17 @@ const sessions = new Map<
   }
 >()
 const calls: Array<{ method: string; params: Record<string, unknown> }> = []
+const heldReplies: Array<() => void> = []
 const upstream = createServer((req, res) => {
   const url = new URL(req.url || '/', `http://127.0.0.1:${upstreamPort}`)
   res.setHeader('content-type', 'application/json')
   const send = (v: unknown) => res.end(JSON.stringify(v))
+  if (url.pathname === '/__release' && req.method === 'POST') {
+    const held = heldReplies.splice(0)
+    for (const complete of held) complete()
+    send({ released: held.length })
+    return
+  }
   if (url.pathname.includes('/plugins/')) {
     res.statusCode = 404
     send({ error: 'This fixture has no plugins' })
@@ -145,10 +152,12 @@ wss.on('connection', (socket) => {
       })
       respond({ status: 'streaming' })
       const name = /你是 (.*?)。/.exec(f.params.text)?.[1] || '助手'
-      const text = `我是${name}。已按角色规则处理这条消息。\n\n- 会话独立保存\n- 可以继续交流\n\n[报告](/tmp/workspace-report.txt)`
+      const requested = /本轮用户指定成员：([^\n]*)/.exec(f.params.text)?.[1] ?? ''
+      const delegates = f.params.text.includes('你是管理员。') && !f.params.text.includes('本批次执行结果：') && requested.startsWith('@')
+      const text = `我是${name}。已按角色规则处理这条消息。\n\n- 会话独立保存\n- 可以继续交流\n\n[报告](/tmp/workspace-report.txt)${delegates ? `\n请${requested}处理任务。` : ''}`
       event('message.start', {}, f.params.session_id)
       setTimeout(() => event('message.delta', { text: text.slice(0, 12) }, f.params.session_id), 80)
-      setTimeout(() => {
+      const complete = () => {
         stored!.running = false
         stored!.messages.push({
           id: randomUUID(),
@@ -157,7 +166,9 @@ wss.on('connection', (socket) => {
           timestamp: Date.now() / 1000,
         })
         event('message.complete', { text, status: 'complete' }, f.params.session_id)
-      }, 250)
+      }
+      if (f.params.text.includes('[hold-workspace]')) heldReplies.push(complete)
+      else setTimeout(complete, 250)
       return
     }
     if (f.method === 'session.interrupt' && stored) stored.running = false

tests/server/workspace.test.ts

@@ -25,7 +25,9 @@ let requests: Array<{ method: string; params: Record<string, any> }>,
   reply: (socket: WebSocket, params: Record<string, any>) => void
 let nodes: WorkspaceNodes
 let rejectPrompt: string | undefined
-let recoveryHistory: Array<{ role: string; content: string }>
+let rejectedInterrupts: number
+let recoveryHistory: Array<{ role: string; content: string; tool_calls?: unknown }>
+let storedByRuntime: Map<string, string>, recoveryByStored: Map<string, Array<{ role: string; content: string }>>
 const owner = 'first-user',
   other = 'second-user'
 beforeEach(async () => {
@@ -34,7 +36,9 @@ beforeEach(async () => {
   uploads = new UploadStore(home)
   requests = []
   recoveryHistory = []
+  storedByRuntime = new Map(); recoveryByStored = new Map()
   rejectPrompt = undefined
+  rejectedInterrupts = 0
   server = new WebSocketServer({ port: 0, host: '127.0.0.1' })
   await new Promise<void>((resolve) => server.once('listening', resolve))
   const address = server.address() as { port: number }
@@ -43,9 +47,9 @@ beforeEach(async () => {
     client: { directAgent: undefined },
     session: {
       webSocketCredential: async () => ({ name: 'ticket', value: 'test' }),
-      request: async () => ({
+      request: async (path: string, options?: {search?: URLSearchParams}) => ({
         status: 200,
-        body: Buffer.from(JSON.stringify({ messages: recoveryHistory })),
+        body: Buffer.from(JSON.stringify({ messages: (() => { const all = recoveryByStored.get(decodeURIComponent(path.split('/')[3]!)) ?? recoveryHistory; const offset = Number(options?.search?.get('offset') ?? 0); return all.slice(Math.max(0,all.length-offset-500), all.length-offset) })() })),
       }),
     },
   }
@@ -72,25 +76,17 @@ beforeEach(async () => {
       const frame = JSON.parse(String(data))
       requests.push(frame)
       const respond = (result: unknown) => socket.send(JSON.stringify({ id: frame.id, result }))
-      if (frame.method === 'session.create')
-        respond({
-          session_id: randomUUID(),
-          stored_session_id: randomUUID(),
-          info: { profile_name: frame.params.profile },
-          running: false,
-        })
-      else if (frame.method === 'session.resume')
-        respond({
-          session_id: randomUUID(),
-          stored_session_id: frame.params.session_id,
-          running: false,
-          info: { profile_name: frame.params.profile },
-        })
+      if (frame.method === 'session.create' || frame.method === 'session.resume') {
+        const runtimeId = randomUUID(), storedId = frame.method === 'session.resume' ? frame.params.session_id : randomUUID()
+        storedByRuntime.set(runtimeId, storedId)
+        respond({ session_id: runtimeId, stored_session_id: storedId, running: false, info: { profile_name: frame.params.profile } })
+      }
       else if (frame.method === 'prompt.submit') {
         if (rejectPrompt) { socket.send(JSON.stringify({id:frame.id,error:{code:4006,message:rejectPrompt}})); return }
         respond({ status: 'streaming' })
         reply(socket, frame.params)
       } else if (frame.method === 'session.usage') respond({ context_used: 400, context_max: 1000 })
+      else if (frame.method === 'session.interrupt' && rejectedInterrupts-- > 0) socket.send(JSON.stringify({id:frame.id,error:{code:500,message:'暂时不能中断'}}))
       else respond({ ok: true, status: 'interrupted' })
     })
   })
@@ -331,26 +327,26 @@ describe('Web-owned workspace', () => {
   it('skips removed queued members and accepts a newly added member on the next request', async () => {
     const a = agent('管理员'), b = agent('待回复'), added = agent('新成员')
     const c = store.createGroup(owner, { name: '队列维护', memberIds: [a.id, b.id], administratorId: a.id, mode: 'free', autoReplyIds: [a.id, b.id] })
-    let turns = 0
     reply = (socket, p) => {
-      if (turns++ === 0) store.updateConversation(owner, c.id, { memberIds: [a.id, added.id] })
       socket.send(JSON.stringify({ method: 'event', params: { type: 'message.complete', session_id: p.session_id, payload: { text: '请@待回复继续' } } }))
     }
-    await finished(runtime.send(owner, c.id, { requestId: randomUUID(), content: '开始' }).id)
+    const queued = runtime.send(owner, c.id, { requestId: randomUUID(), content: '开始' })
+    store.updateConversation(owner, c.id, { memberIds: [a.id, added.id] })
+    await finished(queued.id)
     expect(store.messages(owner, c.id).filter(m => m.role === 'assistant').map(m => m.agentId)).toEqual([a.id])
     await finished(runtime.send(owner, c.id, { requestId: randomUUID(), content: '请@新成员继续' }).id)
-    expect(store.messages(owner, c.id).filter(m => m.role === 'assistant').map(m => m.agentId)).toEqual([a.id, added.id])
+    expect(store.messages(owner, c.id).filter(m => m.role === 'assistant').map(m => m.agentId)).toEqual([a.id, added.id, a.id])
   })
   it('finishes an in-flight removed member reply without scheduling it again', async () => {
     const a = agent('管理员'), b = agent('正在回复')
     const c = store.createGroup(owner, { name: '回复中移除', memberIds: [a.id, b.id], administratorId: a.id })
     let turns = 0
     reply = (socket, p) => {
-      if (turns++ === 0) store.updateConversation(owner, c.id, { memberIds: [a.id] })
-      socket.send(JSON.stringify({ method: 'event', params: { type: 'message.complete', session_id: p.session_id, payload: { text: '请@正在回复继续' } } }))
+      if (turns++ === 1) store.updateConversation(owner, c.id, { memberIds: [a.id] })
+      socket.send(JSON.stringify({ method: 'event', params: { type: 'message.complete', session_id: p.session_id, payload: { text: turns === 1 ? '请@正在回复继续' : '已完成' } } }))
     }
     await finished(runtime.send(owner, c.id, { requestId: randomUUID(), content: '@正在回复 开始' }).id)
-    expect(store.messages(owner, c.id).filter(m => m.role === 'assistant').map(m => m.agentId)).toEqual([b.id, a.id])
+    expect(store.messages(owner, c.id).filter(m => m.role === 'assistant').map(m => m.agentId)).toEqual([a.id, b.id, a.id])
     expect(() => runtime.send(owner, c.id, { requestId: randomUUID(), content: '继续', mentionIds: [b.id] })).toThrow('只能 @ 群内成员')
   })
   it('dispatches a request exactly once across retries and rejects payload reuse', async () => {
@@ -453,11 +449,10 @@ describe('Web-owned workspace', () => {
     await vi.waitFor(() =>
       expect(store.require<WorkspaceRun>(owner, 'run', run.id).status).toBe('uncertain'),
     )
-    expect(() => runtime.send(owner, c.id, { requestId: randomUUID(), content: '重试' })).toThrow(
-      '当前回复',
-    )
+    const queued = runtime.send(owner, c.id, { requestId: randomUUID(), content: '后续请求' })
+    expect(queued.status).toBe('queued')
     expect(requests.filter((r) => r.method === 'prompt.submit')).toHaveLength(1)
-    await runtime.stop(owner, run.id)
+    await runtime.stopConversation(owner, c.id)
     expect(
       store.require<WorkspaceConversation>(owner, 'conversation', c.id).activeRunId,
     ).toBeUndefined()
@@ -525,7 +520,7 @@ it('publishes only the executing member and clears avatar activity when stopped'
   reply=(socket,p)=>socket.send(JSON.stringify({method:'event',params:{type:'approval.request',session_id:p.session_id,payload:{request_id:'approval-avatar',message:'需要确认'}}}))
   const run=runtime.send(owner,c.id,{requestId:randomUUID(),content:'处理任务',mentionIds:[b.id]})
   await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',run.id).status).toBe('waiting'))
-  expect(store.require<WorkspaceConversation>(owner,'conversation',c.id)).toMatchObject({activeAgentId:b.id,activeRunStatus:'waiting'})
+  expect(store.require<WorkspaceConversation>(owner,'conversation',c.id)).toMatchObject({activeAgentId:a.id,activeRunStatus:'waiting'})
   await runtime.stop(owner,run.id)
   const stopped=store.require<WorkspaceConversation>(owner,'conversation',c.id)
   expect(stopped.activeAgentId).toBeUndefined()
@@ -589,3 +584,285 @@ it('recognizes mentions inside Chinese sentences and ignores quoted code and ema
   expect(mentionedAgents('mail@Ann.com https://site/@竹儿 `@瑶儿` <quoted_message>@瑶儿</quoted_message>',roster)).toEqual([])
   expect(mentionedAgents('@Anna',roster)).toEqual([])
 })
+
+function completeReply(socket: WebSocket, p: Record<string, any>, text: string, status = 'complete') {
+  socket.send(JSON.stringify({ method: 'event', params: { type: 'message.complete', session_id: p.session_id, payload: { text, status, ...(status === 'failed' ? { error: text } : {}) } } }))
+}
+const speaker = (p: Record<string, any>) => /^你是 (.*?)。/.exec(p.text)?.[1]
+
+it('starts with the administrator, executes a batch concurrently, and reviews failures once without spending a round', async () => {
+  const a = agent('管理员'), b = agent('成员乙'), c = agent('成员丙')
+  const group = store.createGroup(owner, { name: '并行验收', memberIds: [a.id,b.id,c.id], administratorId: a.id, maxReplyRounds: 2 })
+  const pending = new Map<string, { socket: WebSocket; p: Record<string, any> }>()
+  let admin = 0, review = ''
+  reply = (socket,p) => {
+    if (speaker(p) === a.name) { if (admin++ === 0) completeReply(socket,p,'请@成员乙检查接口，@成员丙检查数据'); else { review=p.text; completeReply(socket,p,'汇总完成') } }
+    else pending.set(speaker(p)!, {socket,p})
+  }
+  const root = runtime.send(owner,group.id,{requestId:randomUUID(),content:'@成员乙 请处理'})
+  await vi.waitFor(()=>expect(pending.size).toBe(2))
+  expect(admin).toBe(1)
+  expect(store.require<WorkspaceConversation>(owner,'conversation',group.id).activeAgentStates).toEqual({[b.id]:'running',[c.id]:'running'})
+  completeReply(pending.get(c.name)!.socket,pending.get(c.name)!.p,'数据检查失败','failed')
+  await vi.waitFor(()=>expect(store.messages(owner,group.id).some(m=>m.status==='failed')).toBe(true))
+  expect(admin).toBe(1)
+  completeReply(pending.get(b.name)!.socket,pending.get(b.name)!.p,'接口正常')
+  await finished(root.id)
+  expect(admin).toBe(2)
+  expect(review).toContain('数据检查失败')
+  expect(review).toContain('接口正常')
+  expect(store.require<WorkspaceRun>(owner,'run',root.id).round).toBe(1)
+})
+
+it('allows successive dependent delegations and does not charge administrator reviews against the limit', async () => {
+  const a=agent('负责人'),b=agent('先做'),c=agent('后做')
+  const g=store.createGroup(owner,{name:'串行依赖',memberIds:[a.id,b.id,c.id],administratorId:a.id,maxReplyRounds:3})
+  const order:string[]=[];let admin=0
+  reply=(socket,p)=>{const name=speaker(p)!;order.push(name);completeReply(socket,p,name===a.name ? ['@先做 第一步','@后做 第二步','完成'][admin++]! : '结果已提交')}
+  await finished(runtime.send(owner,g.id,{requestId:randomUUID(),content:'执行'}).id)
+  expect(order).toEqual([a.name,b.name,a.name,c.name,a.name])
+})
+
+it('continues free discussion automatically and hides irrelevant automatic replies', async () => {
+  const a=agent('公开管理员'),b=agent('自动成员')
+  const g=store.createGroup(owner,{name:'自由讨论',memberIds:[a.id,b.id],administratorId:a.id,mode:'free',autoReplyIds:[b.id],maxReplyRounds:3})
+  reply=(socket,p)=>completeReply(socket,p,speaker(p)===a.name?'我来回答':'[[YAOYAO_NO_REPLY_V1]]')
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'一个问题'})
+  await finished(root.id)
+  expect(requests.filter(r=>r.method==='prompt.submit'&&speaker(r.params)===b.name)).toHaveLength(2)
+  expect(store.messages(owner,g.id).map(m=>m.content)).toEqual(['一个问题','我来回答'])
+  expect(store.conversationSummary(owner,store.require(owner,'conversation',g.id)).unreadCount).toBe(2)
+  expect(store.hiddenMessageIds(owner,g.id)).toHaveLength(2)
+})
+
+it('requires a public administrator reply when the model tries to remain silent', async () => {
+  const a=agent('必须答复'),b=agent('旁观者'),g=store.createGroup(owner,{name:'公开答复',memberIds:[a.id,b.id],administratorId:a.id})
+  reply=(socket,p)=>completeReply(socket,p,'[[YAOYAO_NO_REPLY_V1]]')
+  await finished(runtime.send(owner,g.id,{requestId:randomUUID(),content:'帮助我'}).id)
+  const answer=store.messages(owner,g.id).at(-1)!
+  expect(answer.content).toContain('请补充具体目标')
+  expect(answer.content).not.toContain('YAOYAO_')
+})
+
+it('queues later user requests durably without leaking them into the preceding collaboration', async () => {
+  const a=agent('排队管理员'),b=agent('排队执行者'),g=store.createGroup(owner,{name:'消息排队',memberIds:[a.id,b.id],administratorId:a.id})
+  let first: {socket:WebSocket;p:Record<string,any>}|undefined
+  let admin=0;const prompts:string[]=[]
+  reply=(socket,p)=>{prompts.push(p.text);if(speaker(p)===a.name&&admin++===0)first={socket,p};else completeReply(socket,p,'已完成')}
+  const one=runtime.send(owner,g.id,{requestId:randomUUID(),content:'第一条任务'})
+  await vi.waitFor(()=>expect(first).toBeDefined())
+  const two=runtime.send(owner,g.id,{requestId:randomUUID(),content:'之后才处理的秘密标记'})
+  expect(two.status).toBe('queued')
+  expect(store.require<WorkspaceConversation>(owner,'conversation',g.id).activeRunId).toBe(one.id)
+  completeReply(first!.socket,first!.p,'请@排队执行者继续')
+  await finished(one.id);await finished(two.id)
+  expect(prompts).toHaveLength(4)
+  expect(prompts.slice(0,3).every(p=>!p.includes('之后才处理的秘密标记'))).toBe(true)
+  expect(prompts[3]).toContain('之后才处理的秘密标记')
+})
+
+it('stops one member while keeping its sibling running and returns the interrupted result for review', async () => {
+  const a=agent('停止管理员'),b=agent('停止乙'),c=agent('继续丙'),g=store.createGroup(owner,{name:'独立停止',memberIds:[a.id,b.id,c.id],administratorId:a.id})
+  const pending=new Map<string,{socket:WebSocket;p:Record<string,any>}>();let admin=0,review=''
+  reply=(socket,p)=>{if(speaker(p)===a.name){if(admin++===0)completeReply(socket,p,'@停止乙 @继续丙');else {review=p.text;completeReply(socket,p,'已汇总')}}else pending.set(speaker(p)!,{socket,p})}
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'开始'})
+  await vi.waitFor(()=>expect(pending.size).toBe(2))
+  await runtime.stopAgent(owner,g.id,b.id)
+  expect(admin).toBe(1)
+  expect(store.require<WorkspaceConversation>(owner,'conversation',g.id).activeAgentStates?.[c.id]).toBe('running')
+  completeReply(pending.get(c.name)!.socket,pending.get(c.name)!.p,'丙完成')
+  await finished(root.id)
+  expect(review).toContain('interrupted')
+  expect(review).toContain('丙完成')
+})
+
+it('recovers a partially completed parallel batch after restart without resubmitting members or duplicating review', async () => {
+  const a=agent('恢复管理员'),b=agent('恢复乙'),c=agent('恢复丙'),g=store.createGroup(owner,{name:'批次恢复',memberIds:[a.id,b.id,c.id],administratorId:a.id})
+  let admin=0;const pending=new Map<string,{socket:WebSocket;p:Record<string,any>}>()
+  reply=(socket,p)=>{if(speaker(p)===a.name)completeReply(socket,p,admin++===0?'@恢复乙 @恢复丙':'复核完成');else pending.set(speaker(p)!,{socket,p})}
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'并行执行'})
+  await vi.waitFor(()=>expect(pending.size).toBe(2))
+  completeReply(pending.get(b.name)!.socket,pending.get(b.name)!.p,'乙已完成')
+  const last=pending.get(c.name)!
+  recoveryByStored.set(storedByRuntime.get(last.p.session_id)!,[{role:'user',content:last.p.text},{role:'assistant',content:'丙已完成'}])
+  last.socket.terminate()
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('uncertain'))
+  runtime.close();runtime=new WorkspaceRuntime(store,nodes,uploads);runtime.start()
+  await finished(root.id)
+  expect(admin).toBe(2)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(4)
+  expect(store.list<any>(owner,'turn').filter(t=>t.reviewOf)).toHaveLength(1)
+})
+
+it('persists the next delegation when restarted between reply completion and cascade planning', async () => {
+  const a=agent('记录管理员'),b=agent('记录成员'),g=store.createGroup(owner,{name:'计划恢复',memberIds:[a.id,b.id],administratorId:a.id})
+  let admin=0
+  reply=(socket,p)=>completeReply(socket,p,speaker(p)===a.name&&admin++===0?'@记录成员 执行':'完成')
+  runtime.onMessage=async()=>{runtime.close()}
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'执行一次'})
+  await vi.waitFor(()=>expect(store.list<any>(owner,'turn')[0]?.status).toBe('complete'))
+  runtime=new WorkspaceRuntime(store,nodes,uploads);runtime.start()
+  await finished(root.id)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(3)
+})
+
+it('bounds long context while preserving the task beginning and ending', async () => {
+  const a=agent('长上下文'),b=agent('辅助'),g=store.createGroup(owner,{name:'上下文边界',memberIds:[a.id,b.id],administratorId:a.id})
+  const text='任务开始标记'+ '文'.repeat(60_000)+'任务结束标记'
+  await finished(runtime.send(owner,g.id,{requestId:randomUUID(),content:text}).id)
+  const prompt=requests.find(r=>r.method==='prompt.submit')!.params.text
+  expect(prompt).toContain('任务开始标记');expect(prompt).toContain('任务结束标记')
+  expect(prompt).toContain('保留首尾片段');expect(prompt.length).toBeLessThan(32_000)
+})
+
+it('enforces three tasks per group and four globally while preserving queued work', async () => {
+  const agents=['并发甲','并发乙','并发丙','并发丁','其他群'].map(n=>agent(n))
+  const group=store.createGroup(owner,{name:'并发限制',memberIds:agents.slice(0,4).map(a=>a.id),administratorId:agents[0]!.id,mode:'free',autoReplyIds:agents.slice(0,4).map(a=>a.id),maxReplyRounds:1})
+  const pending=new Map<string,{socket:WebSocket;p:Record<string,any>}>()
+  reply=(socket,p)=>pending.set(speaker(p)!,{socket,p})
+  const one=runtime.send(owner,group.id,{requestId:randomUUID(),content:'并发测试'})
+  const two=runtime.send(owner,direct(agents[4]!.id).id,{requestId:randomUUID(),content:'另一群'})
+  await vi.waitFor(()=>expect(pending.size).toBe(4))
+  expect(Object.keys(store.require<WorkspaceConversation>(owner,'conversation',group.id).activeAgentStates!)).toHaveLength(3)
+  expect(store.list<any>(owner,'turn').filter(t=>t.status==='running')).toHaveLength(4)
+  const released=[...pending.values()].find(v=>speaker(v.p)!==agents[4]!.name)!
+  completeReply(released.socket,released.p,'完成')
+  await vi.waitFor(()=>expect(pending.size).toBe(5))
+  expect(store.list<any>(owner,'turn').filter(t=>t.status==='running').length).toBeLessThanOrEqual(4)
+  for(const v of pending.values()) if(v!==released) completeReply(v.socket,v.p,'完成')
+  await finished(one.id);await finished(two.id)
+})
+
+it('supports unlimited rounds beyond the finite limit and still permits stopping', async () => {
+  const a=agent('不限甲'),b=agent('不限乙'),g=store.createGroup(owner,{name:'不限轮次',memberIds:[a.id,b.id],administratorId:a.id,mode:'free',autoReplyIds:[a.id],maxReplyRounds:-1})
+  let count=0
+  reply=(socket,p)=>{
+    if(count++===0){const work=store.list<any>(owner,'turn').find(t=>t.status==='running')!;work.depth=100;store.put(owner,'turn',work.id,work);completeReply(socket,p,'继续讨论')}
+    else completeReply(socket,p,'[[YAOYAO_NO_REPLY_V1]]')
+  }
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'@不限乙 开始'})
+  await finished(root.id)
+  expect(count).toBe(2)
+  expect(store.require<WorkspaceRun>(owner,'run',root.id).round).toBe(101)
+  expect(()=>store.updateConversation(owner,g.id,{maxReplyRounds:0})).toThrow()
+})
+
+it('keeps parallel approval cards scoped to their member and deduplicates replayed requests and answers', async () => {
+  const a=agent('审批管理员'),b=agent('审批乙'),c=agent('审批丙'),g=store.createGroup(owner,{name:'并发审批',memberIds:[a.id,b.id,c.id],administratorId:a.id})
+  let admin=0
+  server.on('connection',socket=>socket.on('message',raw=>{const f=JSON.parse(String(raw));if(f.method==='approval.respond')completeReply(socket,f.params,'已确认')}))
+  reply=(socket,p)=>{
+    if(speaker(p)===a.name){completeReply(socket,p,admin++===0?'@审批乙 @审批丙':'汇总');return}
+    const frame=JSON.stringify({method:'event',params:{type:'approval.request',session_id:p.session_id,payload:{request_id:'same-request',message:'是否继续'}}})
+    socket.send(frame);socket.send(frame)
+  }
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'开始'})
+  await vi.waitFor(()=>expect(store.list<WorkspaceInteraction>(owner,'interaction').filter(i=>!i.resolved)).toHaveLength(2))
+  const interactions=store.list<WorkspaceInteraction>(owner,'interaction')
+  expect(store.require<WorkspaceConversation>(owner,'conversation',g.id).activeAgentStates).toEqual({[b.id]:'waiting',[c.id]:'waiting'})
+  await runtime.respond(owner,interactions[0]!.id,'once')
+  await runtime.respond(owner,interactions[0]!.id,'once')
+  await expect(runtime.respond(owner,interactions[0]!.id,'deny')).rejects.toThrow('其他答复')
+  expect(admin).toBe(1)
+  await runtime.respond(owner,interactions[1]!.id,'once')
+  await finished(root.id)
+  expect(requests.filter(r=>r.method==='approval.respond')).toHaveLength(2)
+  expect(admin).toBe(2)
+})
+
+it('retries a rolled-back cascade plan without losing or duplicating delegation', async () => {
+  const a=agent('事务管理员'),b=agent('事务成员'),g=store.createGroup(owner,{name:'调度事务',memberIds:[a.id,b.id],administratorId:a.id})
+  const original=store.put.bind(store);let failed=false,admin=0
+  store.put=(user,kind,id,value)=>{if(!failed&&kind==='turn'&&(value as any).agentId===b.id){failed=true;throw new Error('模拟一次写入失败')}original(user,kind,id,value)}
+  reply=(socket,p)=>completeReply(socket,p,speaker(p)===a.name&&admin++===0?'@事务成员 继续':'完成')
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'开始'})
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('complete'),{timeout:4000})
+  expect(failed).toBe(true)
+  expect(store.list<any>(owner,'turn').filter(t=>t.agentId===b.id)).toHaveLength(1)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(3)
+})
+
+it('upgrades an admitted v0.3.7 execution without submitting its prompt again', async () => {
+  const a=agent('升级执行'),c=direct(a.id)
+  reply=(socket,p)=>{recoveryHistory=[{role:'user',content:p.text},{role:'assistant',content:'升级前已完成'}];socket.terminate()}
+  const root=runtime.send(owner,c.id,{requestId:randomUUID(),content:'只执行一次'})
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('uncertain'))
+  runtime.close()
+  const old=store.list<any>(owner,'turn')[0]!
+  for(const task of store.list<any>(owner,'turn'))store.remove(owner,'turn',task.id)
+  const legacy=store.require<any>(owner,'run',root.id)
+  Object.assign(legacy,{queue:[a.id],next:[],hostReturn:false,currentMessageId:old.currentMessageId,turnConfiguration:old.turnConfiguration})
+  store.put(owner,'run',root.id,legacy)
+  const binding=store.get<any>(owner,'binding',`${c.id}:${a.id}`)!
+  delete binding.taskId;store.put(owner,'binding',binding.id,binding)
+  runtime=new WorkspaceRuntime(store,nodes,uploads);runtime.start()
+  await finished(root.id)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(1)
+  expect(store.messages(owner,c.id).at(-1)?.content).toBe('升级前已完成')
+})
+
+it('does not read unrelated or orphaned historical runs when projecting a group prompt', async () => {
+  const a=agent('上下文隔离'),b=agent('组内成员'),g=store.createGroup(owner,{name:'干净群',memberIds:[a.id,b.id],administratorId:a.id})
+  store.put(owner,'run','historical-orphan',{id:'historical-orphan',conversationId:'unrelated-conversation',messageId:'missing-history',mentionIds:[],status:'complete',round:0,createdAt:1,updatedAt:1})
+  await finished(runtime.send(owner,g.id,{requestId:randomUUID(),content:'当前群任务'}).id)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(1)
+})
+
+it('pages backwards to find a long-running turn marker and never mistakes a tool call for a final answer', async () => {
+  const a=agent('分页恢复'),c=direct(a.id)
+  reply=(socket,p)=>{
+    recoveryHistory=[{role:'user',content:p.text},...Array.from({length:600},()=>({role:'tool',content:'工具结果'})),{role:'assistant',content:'最终结果'}]
+    socket.terminate()
+  }
+  await finished(runtime.send(owner,c.id,{requestId:randomUUID(),content:'长时间执行'}).id)
+  expect(store.messages(owner,c.id).at(-1)?.content).toBe('最终结果')
+  let submitted=''
+  reply=(socket,p)=>{submitted=p.text;recoveryHistory=[{role:'user',content:submitted},{role:'assistant',content:'准备调用工具',tool_calls:[{id:'tool'}]},{role:'tool',content:'工具输出'}];socket.terminate()}
+  const root=runtime.send(owner,c.id,{requestId:randomUUID(),content:'不要把过程当结论'})
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('uncertain'))
+  await runtime.reconcile(owner,root.id)
+  await new Promise(resolve=>setTimeout(resolve,100))
+  expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('uncertain')
+  recoveryHistory.push({role:'assistant',content:'真正完成'})
+  await runtime.reconcile(owner,root.id)
+  await finished(root.id)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(2)
+})
+
+it('keeps explicit replies visible and emits one notice when the configured cascade limit is reached', async () => {
+  const a=agent('轮数管理员'),b=agent('轮数成员'),g=store.createGroup(owner,{name:'有限轮次',memberIds:[a.id,b.id],administratorId:a.id,maxReplyRounds:2})
+  reply=(socket,p)=>completeReply(socket,p,speaker(p)===a.name?'@轮数成员 继续':'已完成本步')
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'开始'})
+  await finished(root.id)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(3)
+  expect(store.messages(owner,g.id).filter(m=>m.content.includes('轮数上限'))).toHaveLength(1)
+  store.updateConversation(owner,g.id,{mode:'free',autoReplyIds:[]})
+  reply=(socket,p)=>completeReply(socket,p,'[[YAOYAO_NO_REPLY_V1]]')
+  await finished(runtime.send(owner,g.id,{requestId:randomUUID(),content:'@轮数成员 显示这个字符串'}).id)
+  expect(store.messages(owner,g.id).at(-1)?.content).toBe('[[YAOYAO_NO_REPLY_V1]]')
+})
+
+it('retries a failed stop of a live task without falsely completing it or resubmitting work', async () => {
+  const a=agent('停止恢复'),c=direct(a.id)
+  reply=()=>{}
+  const root=runtime.send(owner,c.id,{requestId:randomUUID(),content:'等待停止'})
+  await vi.waitFor(()=>expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(1))
+  rejectedInterrupts=1
+  await runtime.stop(owner,root.id)
+  expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('uncertain')
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('interrupted'),{timeout:3000})
+  expect(requests.filter(r=>r.method==='session.interrupt')).toHaveLength(2)
+  expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(1)
+})
+
+it('marks a stopped administrator as interrupted without sending a false completion notification', async () => {
+  const a=agent('管理员停止'),b=agent('普通成员'),g=store.createGroup(owner,{name:'停止管理员',memberIds:[a.id,b.id],administratorId:a.id})
+  reply=()=>{}
+  const notifications: string[]=[]
+  runtime.onNotify=(_owner,_c,run)=>notifications.push(run.status)
+  const root=runtime.send(owner,g.id,{requestId:randomUUID(),content:'等待'})
+  await vi.waitFor(()=>expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(1))
+  await runtime.stopAgent(owner,g.id,a.id)
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('interrupted'))
+  expect(notifications).toEqual([])
+})