diff --git a/desktop/src/agents/cli/cli-acp-runner.cjs b/desktop/src/agents/cli/cli-acp-runner.cjs index 7864131..9342063 100644 --- a/desktop/src/agents/cli/cli-acp-runner.cjs +++ b/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) } diff --git a/desktop/src/agents/cli/cli-adapters.cjs b/desktop/src/agents/cli/cli-adapters.cjs index 08a19a9..51a975f 100644 --- a/desktop/src/agents/cli/cli-adapters.cjs +++ b/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') diff --git a/desktop/src/workspace/local-workspace-auto-runner-natural-decisions.cjs b/desktop/src/workspace/local-workspace-auto-runner-natural-decisions.cjs new file mode 100644 index 0000000..15ee9a3 --- /dev/null +++ b/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 } diff --git a/desktop/src/workspace/local-workspace-auto-runner.cjs b/desktop/src/workspace/local-workspace-auto-runner.cjs index 278e082..ef72eea 100644 --- a/desktop/src/workspace/local-workspace-auto-runner.cjs +++ b/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 } diff --git a/desktop/test/agents/cli/cli-adapters-agent-protocols.test.cjs b/desktop/test/agents/cli/cli-adapters-agent-protocols.test.cjs index 5f0b4c1..c166e2e 100644 --- a/desktop/test/agents/cli/cli-adapters-agent-protocols.test.cjs +++ b/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 })) diff --git a/desktop/test/agents/cli/cli-adapters.test.cjs b/desktop/test/agents/cli/cli-adapters.test.cjs index 9f94618..605228a 100644 --- a/desktop/test/agents/cli/cli-adapters.test.cjs +++ b/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', `