fix(desktop): recover stale agent sessions in group runs

Recover reused ACP sessions that refuse before producing output so generic workspace recovery can establish a fresh session. Keep Pi processes alive through internal tool turns until agent_end, and extract V4 natural discussion decision methods into a dedicated module without changing the runner API or persisted structures. Verified with focused CLI, natural discussion, workspace recovery, syntax, diff, and full desktop test suites.

desktop/src/agents/cli/cli-acp-runner.cjs

@@ -886,6 +886,11 @@ async function runAcpTurn(runtime, prompt, options, spec, profile) {
         ? 'timeout'
         : normalized.failure?.category === 'cancellation' ? 'stopped' : 'failed',
     })
+    if (promptStarted && options.sessionRef
+        && runtime.replyState.bytes === 0
+        && normalized.message === 'LOCAL_AGENT_REFUSED') {
+      throw agentExecutionError('LOCAL_AGENT_SESSION_INVALID')
+    }
     if (!promptStarted && normalized.message !== 'LOCAL_AGENT_SESSION_INVALID') {
       allowAcpSetupFallback(normalized)
     }

desktop/src/agents/cli/cli-adapters.cjs

@@ -314,7 +314,7 @@ async function runAgent(agent, prompt, workdir, options = {}) {
       if (terminalObserved || !event || typeof event !== 'object') return
       const terminal = (
         (['workbuddy', 'claude', 'qwen'].includes(agent.kind) && event.type === 'result')
-        || (agent.kind === 'pi' && ['turn_end', 'agent_end'].includes(event.type))
+        || (agent.kind === 'pi' && event.type === 'agent_end')
         || (agent.kind === 'codex' && event.type === 'turn.completed')
         || (['mimo', 'opencode'].includes(agent.kind) && event.type === 'step_finish')
         || (agent.kind === 'gemini' && event.type === 'result')

desktop/src/workspace/local-workspace-auto-runner-natural-decisions.cjs

@@ -0,0 +1,124 @@
+const { parseTaskDecision } = require('../collaboration/task-decision.cjs')
+
+const v4NaturalDecisionMethods = {
+  v4NaturalRouteDecision(decision, activeKinds) {
+    const participants = Array.isArray(activeKinds) ? activeKinds : []
+    if (decision == null) return { status: 'none', kinds: [] }
+    let parsed
+    try { parsed = parseTaskDecision(decision) }
+    catch { return { status: 'invalid', kinds: [] } }
+    const requested = parsed.nextKinds || []
+    if (requested.some(kind => !participants.includes(kind))) return { status: 'invalid', kinds: [] }
+    if (!requested.length) return { status: 'none', kinds: [] }
+    return { status: 'valid', kinds: participants.filter(kind => requested.includes(kind)) }
+  },
+
+  async v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) {
+    const messages = this.v4NaturalDiscussionRoundMessages(group, controller, threadRootId, round)
+    const owner = this.v4NaturalOwner(group, controller, threadRootId)
+    const routes = messages.map(message => this.v4NaturalRouteDecision(
+      message.trace?.context?.taskDecision, activeKinds,
+    ))
+    if (routes.some(route => route.status === 'invalid')) {
+      this.addMessage(group.id, 'system', 'Discussion stopped without a valid task completion decision.',
+        '', threadRootId, { key: 'system.autoTaskDecisionMissing' })
+      return { status: 'partial' }
+    }
+    const nextKinds = this.v4NaturalNextKinds(group, controller, threadRootId, round, activeKinds)
+    if (nextKinds.length) return { nextKinds, ownerReview: false }
+    if (!activeKinds.includes(owner)) return { status: 'partial' }
+    const ownerMessage = messages.find(message => message.agentKind === owner)
+    const decision = ownerMessage?.trace?.context?.taskDecision
+    if (decision?.handoffTo && !activeKinds.includes(decision.handoffTo)) return { status: 'partial' }
+    if (decision?.status === 'continue') return { nextKinds: [owner], ownerReview: false }
+    const sawAllContributions = messages.length === 1
+      || (controller.discussionStyle === 'sequential' && messages.at(-1)?.agentKind === owner)
+    if (!ownerMessage || !sawAllContributions) return { nextKinds: [owner], ownerReview: true }
+    if (!decision) {
+      this.addMessage(group.id, 'system', 'Discussion stopped without a valid task completion decision.',
+        '', threadRootId, { key: 'system.autoTaskDecisionMissing' })
+      return { status: 'partial' }
+    }
+    if (decision.status === 'completed') return { status: controller.failedKinds.length ? 'partial' : 'completed' }
+    if (decision.status === 'needs-human' && this.requestTaskDecision) {
+      const response = await this.requestTaskDecision(group, controller, ownerMessage)
+      if (response?.status === 'approved') return { nextKinds: [owner], ownerReview: true }
+      if (response?.status === 'rejected') {
+        controller.stopReason = 'human_gate_rejected'
+        return { status: 'stopped' }
+      }
+    }
+    this.addMessage(group.id, 'system', decision.reason, '', threadRootId, {
+      key: decision.status === 'needs-human' ? 'system.autoTaskNeedsHuman' : 'system.autoTaskBlocked',
+      params: { reason: decision.reason },
+    })
+    return { status: 'partial' }
+  },
+
+  v4NaturalNextKinds(group, controller, threadRootId, round, activeKinds) {
+    const current = this.v4NaturalDiscussionRoundMessages(group, controller, threadRootId, round)
+    const selected = new Set(current.flatMap(message => (
+      this.v4NaturalRouteDecision(message.trace?.context?.taskDecision, activeKinds).kinds
+        .filter(kind => kind !== message.agentKind)
+    )))
+    if (!selected.size) return []
+    return activeKinds.filter(kind => selected.has(kind))
+  },
+
+  v4NaturalOwner(group, controller, threadRootId) {
+    let owner = controller.targetKinds[0] || ''
+    for (const message of this.state().messages) {
+      if (message.groupId !== group.id || message.threadRootId !== threadRootId
+          || message.role !== 'agent' || message.agentKind !== owner
+          || message.trace?.phase !== 'discussion'
+          || !this.v4NaturalMessageMatchesBinding(message, controller)) continue
+      const decision = message.trace?.context?.taskDecision
+      if (decision?.status === 'continue' && decision.handoffTo
+          && controller.targetKinds.includes(decision.handoffTo)) owner = decision.handoffTo
+    }
+    return owner
+  },
+
+  v4NaturalDiscussionIsRepeating(group, controller, threadRootId, round, activeKinds) {
+    const window = Math.max(4, activeKinds.length * 2)
+    if (round < window + 2) return false
+    const firstRound = round - window + 1
+    const previous = new Map()
+    const observedRounds = new Set()
+    const messages = this.state().messages
+    for (let index = messages.length - 1; index >= 0; index -= 1) {
+      const message = messages[index]
+      const messageRound = Number(message.trace?.round) || 0
+      if (message.groupId !== group.id || message.threadRootId !== threadRootId
+          || message.role !== 'agent' || message.trace?.phase !== 'discussion'
+          || message.trace?.runId !== controller.runId || messageRound < 2
+          || messageRound > round || !this.v4NaturalMessageMatchesBinding(message, controller)) continue
+      const contribution = this.v4NaturalMessageContent(message)
+      if (contribution.partial) return false
+      const content = contribution.text
+        .replace(/@[A-Za-z0-9][A-Za-z0-9_-]*/gu, '')
+        .replace(/\s+/gu, ' ').trim()
+      if (previous.has(message.agentKind) && previous.get(message.agentKind) !== content) return false
+      if (messageRound >= firstRound) {
+        observedRounds.add(messageRound)
+        previous.set(message.agentKind, content)
+      } else {
+        previous.delete(message.agentKind)
+        if (observedRounds.size === window && previous.size === 0) return true
+      }
+    }
+    return false
+  },
+
+  v4NaturalStopRepeatingDiscussion(group, controller, threadRootId, round, activeKinds) {
+    if (!this.v4NaturalDiscussionIsRepeating(group, controller, threadRootId, round, activeKinds)) return false
+    this.addMessage(
+      group.id, 'system',
+      'Automatic discussion stopped because repeated handoffs produced no new contribution.',
+      '', threadRootId, { key: 'system.autoDiscussionStalled' },
+    )
+    return true
+  },
+}
+
+module.exports = { v4NaturalDecisionMethods }

desktop/src/workspace/local-workspace-auto-runner.cjs

@@ -73,6 +73,7 @@ const {
   unauthorizedFailure,
 } = require('./local-workspace-auto-runner-support.cjs')
 const { traceCapsuleFromAgentRun } = require('../runs/run-harness-normalization.cjs')
+const { v4NaturalDecisionMethods } = require('./local-workspace-auto-runner-natural-decisions.cjs')
 const { v4OutcomeMethods } = require('./local-workspace-auto-runner-v4-outcomes.cjs')
 
 class LocalWorkspaceAutoRunner {
@@ -2450,20 +2451,6 @@ class LocalWorkspaceAutoRunner {
     ].filter(Boolean).join('\n\n')
   }
 
-  v4NaturalRouteDecision(decision, activeKinds) {
-    const participants = Array.isArray(activeKinds) ? activeKinds : []
-    if (decision == null) return { status: 'none', kinds: [] }
-    let parsed
-    try { parsed = parseTaskDecision(decision) }
-    catch { return { status: 'invalid', kinds: [] } }
-    const requested = parsed.nextKinds || []
-    if (requested.some(kind => !participants.includes(kind))) {
-      return { status: 'invalid', kinds: [] }
-    }
-    if (!requested.length) return { status: 'none', kinds: [] }
-    return { status: 'valid', kinds: participants.filter(kind => requested.includes(kind)) }
-  }
-
   v4NaturalMessageMatchesBinding(message, controller) {
     const phase = String(message?.trace?.phase || '')
     const kind = String(message?.agentKind || '')
@@ -2730,122 +2717,6 @@ class LocalWorkspaceAutoRunner {
     ))
   }
 
-  v4NaturalOwner(group, controller, threadRootId) {
-    let owner = controller.targetKinds[0] || ''
-    for (const message of this.state().messages) {
-      if (message.groupId !== group.id || message.threadRootId !== threadRootId
-          || message.role !== 'agent' || message.agentKind !== owner
-          || message.trace?.phase !== 'discussion'
-          || !this.v4NaturalMessageMatchesBinding(message, controller)) continue
-      const decision = message.trace?.context?.taskDecision
-      if (decision?.status === 'continue' && decision.handoffTo
-          && controller.targetKinds.includes(decision.handoffTo)) owner = decision.handoffTo
-    }
-    return owner
-  }
-
-  async v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) {
-    const messages = this.v4NaturalDiscussionRoundMessages(group, controller, threadRootId, round)
-    const owner = this.v4NaturalOwner(group, controller, threadRootId)
-    const routes = messages.map(message => this.v4NaturalRouteDecision(
-      message.trace?.context?.taskDecision, activeKinds,
-    ))
-    if (routes.some(route => route.status === 'invalid')) {
-      this.addMessage(group.id, 'system', 'Discussion stopped without a valid task completion decision.',
-        '', threadRootId, { key: 'system.autoTaskDecisionMissing' })
-      return { status: 'partial' }
-    }
-    const nextKinds = this.v4NaturalNextKinds(group, controller, threadRootId, round, activeKinds)
-    if (nextKinds.length) return { nextKinds, ownerReview: false }
-    if (!activeKinds.includes(owner)) return { status: 'partial' }
-    const ownerMessage = messages.find(message => message.agentKind === owner)
-    const decision = ownerMessage?.trace?.context?.taskDecision
-    if (decision?.handoffTo && !activeKinds.includes(decision.handoffTo)) {
-      return { status: 'partial' }
-    }
-    if (decision?.status === 'continue') return { nextKinds: [owner], ownerReview: false }
-    const sawAllContributions = messages.length === 1
-      || (controller.discussionStyle === 'sequential' && messages.at(-1)?.agentKind === owner)
-    if (!ownerMessage || !sawAllContributions) return { nextKinds: [owner], ownerReview: true }
-    if (!decision) {
-      this.addMessage(group.id, 'system', 'Discussion stopped without a valid task completion decision.',
-        '', threadRootId, { key: 'system.autoTaskDecisionMissing' })
-      return { status: 'partial' }
-    }
-    if (decision.status === 'completed') {
-      return { status: controller.failedKinds.length ? 'partial' : 'completed' }
-    }
-    if (decision.status === 'needs-human' && this.requestTaskDecision) {
-      const response = await this.requestTaskDecision(group, controller, ownerMessage)
-      if (response?.status === 'approved') return { nextKinds: [owner], ownerReview: true }
-      if (response?.status === 'rejected') {
-        controller.stopReason = 'human_gate_rejected'
-        return { status: 'stopped' }
-      }
-    }
-    this.addMessage(group.id, 'system', decision.reason, '', threadRootId, {
-      key: decision.status === 'needs-human' ? 'system.autoTaskNeedsHuman' : 'system.autoTaskBlocked',
-      params: { reason: decision.reason },
-    })
-    return { status: 'partial' }
-  }
-
-  v4NaturalNextKinds(group, controller, threadRootId, round, activeKinds) {
-    const current = this.v4NaturalDiscussionRoundMessages(group, controller, threadRootId, round)
-    const selected = new Set(current.flatMap(message => (
-      this.v4NaturalRouteDecision(message.trace?.context?.taskDecision, activeKinds).kinds
-        .filter(kind => kind !== message.agentKind)
-    )))
-    if (!selected.size) return []
-    return activeKinds.filter(kind => selected.has(kind))
-  }
-
-  v4NaturalDiscussionIsRepeating(group, controller, threadRootId, round, activeKinds) {
-    const window = Math.max(4, activeKinds.length * 2)
-    if (round < window + 2) return false
-    const firstRound = round - window + 1
-    const previous = new Map()
-    const observedRounds = new Set()
-    const messages = this.state().messages
-    // Inspect only the recent window and each speaker's preceding reply, including after recovery.
-    for (let index = messages.length - 1; index >= 0; index -= 1) {
-      const message = messages[index]
-      const messageRound = Number(message.trace?.round) || 0
-      if (message.groupId !== group.id || message.threadRootId !== threadRootId
-          || message.role !== 'agent' || message.trace?.phase !== 'discussion'
-          || message.trace?.runId !== controller.runId || messageRound < 2
-          || messageRound > round || !this.v4NaturalMessageMatchesBinding(message, controller)) continue
-      const contribution = this.v4NaturalMessageContent(message)
-      if (contribution.partial) return false
-      const content = contribution.text
-        .replace(/@[A-Za-z0-9][A-Za-z0-9_-]*/gu, '')
-        .replace(/\s+/gu, ' ').trim()
-      if (previous.has(message.agentKind) && previous.get(message.agentKind) !== content) {
-        return false
-      }
-      if (messageRound >= firstRound) {
-        observedRounds.add(messageRound)
-        previous.set(message.agentKind, content)
-      } else {
-        previous.delete(message.agentKind)
-        if (observedRounds.size === window && previous.size === 0) return true
-      }
-    }
-    return false
-  }
-
-  v4NaturalStopRepeatingDiscussion(group, controller, threadRootId, round, activeKinds) {
-    if (!this.v4NaturalDiscussionIsRepeating(group, controller, threadRootId, round, activeKinds)) {
-      return false
-    }
-    this.addMessage(
-      group.id, 'system',
-      'Automatic discussion stopped because repeated handoffs produced no new contribution.',
-      '', threadRootId, { key: 'system.autoDiscussionStalled' },
-    )
-    return true
-  }
-
   async runV4SequentialDiscussion(group, controller, threadRootId, context, input) {
     const {
       targetKinds, activeKinds: initialActiveKinds, writerKind, batchId,
@@ -6138,6 +6009,10 @@ class LocalWorkspaceAutoRunner {
   }
 }
 
-Object.assign(LocalWorkspaceAutoRunner.prototype, v4OutcomeMethods)
+Object.assign(
+  LocalWorkspaceAutoRunner.prototype,
+  v4NaturalDecisionMethods,
+  v4OutcomeMethods,
+)
 
 module.exports = { LocalWorkspaceAutoRunner }

desktop/test/agents/cli/cli-adapters-agent-protocols.test.cjs

@@ -1891,6 +1891,36 @@ process.stdout.write(JSON.stringify({ type: 'agent_end' }) + '\\n')
   assert.equal(result.outcome, 'completed')
 })
 
+test('Pi keeps running after a tool turn until its agent loop ends', async (t) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-pi-tool-turn-'))
+  t.after(() => fs.rmSync(directory, { recursive: true, force: true }))
+  const cli = executable(directory, 'pi-tool-turn.cjs', `
+const send = value => process.stdout.write(JSON.stringify(value) + '\\n')
+send({ type: 'session', id: 'pi-tool-turn-session' })
+send({ type: 'message_update', assistantMessageEvent: {
+  type: 'text_delta', delta: 'Preparing evidence. ',
+} })
+send({ type: 'turn_end', message: { role: 'assistant', stopReason: 'toolUse' } })
+setTimeout(() => {
+  send({ type: 'message_update', assistantMessageEvent: {
+    type: 'text_delta', delta: 'Final answer.',
+  } })
+  send({ type: 'agent_end', messages: [{ role: 'assistant', stopReason: 'stop' }] })
+}, 400)
+`)
+
+  const result = await runAgent(
+    { kind: 'pi', executable: cli, name: 'Pi Agent' },
+    'research the current market',
+    directory,
+  )
+  assert.deepEqual(result, {
+    text: 'Preparing evidence. Final answer.',
+    sessionRef: 'pi-tool-turn-session',
+    outcome: 'completed',
+  })
+})
+
 test('runAgent strips partial Pi protocol and Harness record echoes', async (t) => {
   const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-pi-partial-echo-'))
   t.after(() => fs.rmSync(directory, { recursive: true, force: true }))

desktop/test/agents/cli/cli-adapters.test.cjs

@@ -1195,6 +1195,39 @@ input.on('line', (line) => {
   assert.equal(answerEvents.at(-1).replace, true)
 })
 
+test('a reused ACP session that refuses before any output is recoverable as stale', async (t) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-acp-stale-refusal-'))
+  const cli = executable(directory, 'acp-stale-refusal.cjs', `
+const readline = require('node:readline')
+const input = readline.createInterface({ input: process.stdin })
+const send = value => process.stdout.write(JSON.stringify(value) + '\\n')
+input.on('close', () => process.exit(0))
+input.on('line', line => {
+  const message = JSON.parse(line)
+  if (message.method === 'initialize') {
+    send({ jsonrpc: '2.0', id: message.id, result: { protocolVersion: 1 } })
+  } else if (message.method === 'session/resume') {
+    send({ jsonrpc: '2.0', id: message.id, result: {} })
+  } else if (message.method === 'session/set_mode') {
+    send({ jsonrpc: '2.0', id: message.id, result: {} })
+  } else if (message.method === 'session/prompt') {
+    send({ jsonrpc: '2.0', id: message.id, result: { stopReason: 'refusal' } })
+  }
+})
+`)
+  t.after(() => fs.rmSync(directory, { recursive: true, force: true }))
+
+  await assert.rejects(
+    runAgent(
+      { kind: 'hermes', executable: cli, name: 'Hermes' },
+      'continue the group task',
+      directory,
+      { sessionRef: 'stale-session', sessionTransport: 'acp', hermesAcpAvailable: true },
+    ),
+    error => error.message === 'LOCAL_AGENT_SESSION_INVALID',
+  )
+})
+
 test('ACP terminal stop reasons preserve exact outcome and failure semantics', async (t) => {
   const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-acp-stop-reasons-'))
   const cli = executable(directory, 'acp-stop-reasons.cjs', `