fix(chat): 节点失联时立即停止任务

src/server/workspaceRuntime.ts

@@ -356,6 +356,18 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
       this.live.delete(key)
       try {
         const current = this.getWork(owner, run.id)
+        if (current.cancelRequested && current.status === 'interrupted') {
+          resultMessage.status = 'interrupted'
+          resultMessage.error = '已停止'
+          this.store.atomic(() => {
+            this.store.saveMessage(owner, resultMessage)
+            this.saveWork(owner, current)
+            this.resolveInteractions(owner, current.id)
+          })
+          if (error) rejectTurn(error)
+          else resolveTurn(resultMessage)
+          return
+        }
         if(error&&!submitted&&current.status!=='interrupted'&&error instanceof HttpError&&RESOURCE_WAIT_CODES.has(error.code??'')){
           current.status='queued';current.resourceWait=true;current.error='等待执行节点或电脑资源';resultMessage.status='queued';resultMessage.visible=false;resultMessage.error=undefined
         }else if (error) {
@@ -402,6 +414,8 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
       finish(completedEvidence ? undefined : new Error('Hermes 连接断开'))
     gateway.onEvent = (frame: GatewayFrame) => {
       if (settled || !runtimeId || frame.session_id !== runtimeId) return
+      const currentTurn = this.getWork(owner, run.id)
+      if (currentTurn.cancelRequested || currentTurn.status === 'interrupted') return
       this.inspector?.record(owner,c.id,{direction:'event',method:frame.type,data:frame.payload,agentId:agent.id,runId:run.runId,taskId:run.conversationTaskId})
       const p = frame.payload ?? {},
         type = frame.type
@@ -814,16 +828,6 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
     const request = this.store.require<Run>(owner, 'run', work.runId).discussion ? `本次用户原始要求：\n${rootTrigger.content.slice(0, 16000)}\n\n` : ''
     return `${request}${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> {
     type Reply = WorkspaceInteraction & { answer?: string; responseState?: 'sending' | 'uncertain' | 'sent' }
     const interaction = this.store.require<Reply>(owner, 'interaction', id)
@@ -873,7 +877,10 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
     if (live?.taskId === work.id) {
       await live.gateway.rpc('session.interrupt', { session_id: live.runtimeId })
       const current = this.getWork(owner, work.id)
-      if (['complete', 'failed', 'interrupted'].includes(current.status)) return
+      if (['complete', 'failed', 'interrupted'].includes(current.status)) {
+        live.done(new Error('已停止'))
+        return
+      }
       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';message.error='已停止';this.store.saveMessage(owner,message)}

src/server/workspaceScheduler.ts

@@ -31,6 +31,9 @@ export interface Work {
   /** Revision captured when this turn receives a team-management grant. */
   teamManagementRevision?: number
   cancelRequested?: boolean
+  /** A locally interrupted turn still owes a remote interrupt confirmation. */
+  cleanupPending?: boolean
+  cleanupError?: string
   planned?: boolean
   silent?: boolean
   hadInteraction?: boolean
@@ -109,13 +112,23 @@ export abstract class WorkspaceScheduler {
       .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)
+    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 OR coalesce(json_extract(data,'$.cleanupPending'),0)=1)").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 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)
+      }
+    }
+  }
   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
@@ -295,6 +308,9 @@ export abstract class WorkspaceScheduler {
     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 (terminal(work.status) && work.cleanupPending) this.scheduleInterruptCleanup(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'
@@ -440,22 +456,62 @@ export abstract class WorkspaceScheduler {
     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 => {
+    for (const w of works) {
+      if (terminal(w.status)) continue
       const current = this.getWork(owner, w.id)
+      const wasQueued = current.status === 'queued'
       current.cancelRequested = true
-      if (current.status === 'queued') { current.status = 'interrupted'; this.saveWork(owner, current) }
-      else {
+      current.status = 'interrupted'
+      current.error = '已停止'
+      current.planned = true
+      current.cleanupPending = !wasQueued
+      if (current.currentMessageId) {
+        const message = this.store.require<Message>(owner, 'message', current.currentMessageId)
+        message.status = 'interrupted'
+        message.error = '已停止'
+        this.store.atomic(() => {
+          this.store.saveMessage(owner, message)
+          this.saveWork(owner, current)
+        })
+      } else this.saveWork(owner, current)
+      this.resolveInteractions(owner, current.id)
+      if (current.cleanupPending) this.scheduleInterruptCleanup(owner, current)
+    }
+    this.wake()
+  }
+  private scheduleInterruptCleanup(owner: string, work: Work): void {
+    if (this.closing || this.interrupting.has(work.id)) return
+    if ((this.recoverAfter.get(work.id) ?? 0) > Date.now()) { this.retrySoon(); return }
+    this.interrupting.add(work.id)
+    let succeeded = false
+    void this.interruptTurn(owner, work)
+      .then(() => {
+        succeeded = true
+        const current = this.getWork(owner, work.id)
+        current.cleanupPending = false
+        current.cleanupError = undefined
         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()
+      })
+      .catch(error => {
+        if (this.closing) return
+        const current = this.getWork(owner, work.id)
+        current.cleanupPending = true
+        current.cleanupError = 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)
+        if (succeeded) {
+          this.attempts.delete(work.id)
+          this.recoverAfter.delete(work.id)
         }
-      }
-    }))
-    this.wake()
+        this.wake()
+      })
+      .catch(() => this.retrySoon())
   }
   protected closeScheduler(): void {
     this.closing = true

tests/server/workspace.test.ts

@@ -1222,7 +1222,7 @@ it('serializes a session and stops only that task while the same Agent keeps rep
   expect(requests.filter(r=>r.method==='session.resume').map(r=>r.params.session_id)).toEqual([firstBinding.storedId])
   expect(pending[2]!.p.text).not.toContain('第二话题独立执行')
   await runtime.stopTask(owner,c.id,first.id)
-  expect(requests.filter(r=>r.method==='session.interrupt').map(r=>r.params.session_id)).toEqual([pending[2]!.p.session_id])
+  await vi.waitFor(()=>expect(requests.filter(r=>r.method==='session.interrupt').map(r=>r.params.session_id)).toEqual([pending[2]!.p.session_id]))
   expect(store.require<WorkspaceRun>(owner,'run',other.id).status).toBe('running')
   expect(store.require<any>(owner,'conversation-task',second.id).activeRunStatus).toBe('running')
   completeReply(pending[1]!.socket,pending[1]!.p,'第二话题独立完成')
@@ -1529,8 +1529,8 @@ it('retries a failed stop of a live task without falsely completing it or resubm
   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(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('interrupted')
+  await vi.waitFor(()=>expect(requests.filter(r=>r.method==='session.interrupt')).toHaveLength(2),{timeout:3000})
   expect(requests.filter(r=>r.method==='session.interrupt')).toHaveLength(2)
   expect(requests.filter(r=>r.method==='prompt.submit')).toHaveLength(1)
 })

tests/server/workspaceCollaboration.test.ts

@@ -17,6 +17,10 @@ let home: string, store: WorkspaceStore, runtime: FakeRuntime, uploads: UploadSt
 class FakeRuntime extends WorkspaceRuntime {
   calls: Array<{ agentId: string; work: Work }> = []
   reply: (agent: Agent, work: Work) => string = agent => `${agent.name} 的实质结果`
+  interrupt: (work: Work) => Promise<void> = work => {
+    work.status = 'interrupted'; this.saveWork('collaboration-owner', work)
+    return Promise.resolve()
+  }
   protected override async performTurn(user: string, c: Conversation, agent: Agent, work: Work): Promise<Message> {
     this.calls.push({ agentId: agent.id, work })
     const content = this.reply(agent, work)
@@ -25,7 +29,7 @@ class FakeRuntime extends WorkspaceRuntime {
     work.status = 'complete'; work.currentMessageId = message.id; this.saveWork(user, work)
     return message
   }
-  protected override async interruptTurn(user: string, work: Work): Promise<void> { work.status = 'interrupted'; this.saveWork(user, work) }
+  protected override async interruptTurn(user: string, work: Work): Promise<void> { await this.interrupt(work) }
 }
 beforeEach(() => {
   home = mkdtempSync(join(tmpdir(), 'yaoyao-collaboration-')); store = new WorkspaceStore(home); uploads = new UploadStore(home)
@@ -34,6 +38,26 @@ beforeEach(() => {
 afterEach(() => { runtime.close(); uploads.close(); store.close(); rmSync(home, { recursive: true, force: true }) })
 const bot = (name: string) => store.createAgent(owner, { name, profile: 'default' })
 const direct = (id: string) => store.list<Conversation>(owner, 'conversation').find(c => c.kind === 'direct' && c.memberIds[0] === id)!
+it('settles a stop locally while remote interruption is unavailable', async () => {
+  vi.spyOn(runtime, 'wake').mockImplementation(() => {})
+  const a = bot('停止挂起')
+  const run = runtime.send(owner, direct(a.id).id, { requestId: randomUUID(), content: '等待停止' })
+  const work = store.list<Work>(owner, 'turn').find(item => item.runId === run.id)!
+  work.status = 'running'
+  work.submitted = true
+  store.put(owner, 'turn', work.id, work)
+  let finishInterrupt!: () => void
+  runtime.interrupt = () => new Promise(resolve => { finishInterrupt = resolve })
+  await Promise.race([
+    runtime.stop(owner, run.id),
+    new Promise((_, reject) => setTimeout(() => reject(new Error('stop timed out')), 50)),
+  ])
+  expect(store.require<Run>(owner, 'run', run.id)).toMatchObject({ status: 'interrupted', stopRequested: true })
+  expect(store.require<Work>(owner, 'turn', work.id)).toMatchObject({ status: 'interrupted', cleanupPending: true })
+  expect(store.require<Conversation>(owner, 'conversation', run.conversationId).activeRunId).toBeUndefined()
+  finishInterrupt()
+  await vi.waitFor(() => expect(store.require<Work>(owner, 'turn', work.id).cleanupPending).toBe(false))
+})
 it('projects real deliveries into compact direction metadata for live events and history without rewriting raw messages', () => {
   vi.spyOn(runtime, 'wake').mockImplementation(() => {})
   const a = bot('竹儿'), b = bot('研究员')