diff --git a/desktop/src/agents/cli/cli-acp-runner.cjs b/desktop/src/agents/cli/cli-acp-runner.cjs index d018536..8bb4369 100644 --- a/desktop/src/agents/cli/cli-acp-runner.cjs +++ b/desktop/src/agents/cli/cli-acp-runner.cjs @@ -25,6 +25,21 @@ const { createAcpOutboundPayload } = require('../../collaboration/outbound-paylo const { redactSecrets } = require('../../security/secret-redaction.cjs') const ACP_CANCEL_GRACE_MS = 250 +// After a stop or timeout the Agent gets this long to answer the cancelled +// prompt. A clean cancellation keeps the long-lived process and its native +// session; only an Agent that ignores cancel is torn down. +const ACP_CANCEL_SETTLE_MS = 3000 +const ACP_INITIALIZE_TIMEOUT_MS = 60 * 1000 +const ACP_STDERR_TAIL_BYTES = 64 * 1024 +// Process identity must not change when only volatile shell values change. +// Version managers such as fnm put a per-shell directory on PATH, so a fresh +// login-shell capture used to restart the ACP process and drop its session. +const ACP_VOLATILE_ENV_KEYS = new Set([ + 'PATH', 'Path', 'TERM', 'TERM_PROGRAM', 'TERM_PROGRAM_VERSION', 'TERM_SESSION_ID', + 'SHLVL', 'PWD', 'OLDPWD', '_', 'TMPDIR', 'COLUMNS', 'LINES', 'SHELL', + 'FNM_MULTISHELL_PATH', 'FNM_DIR', 'FNM_VERSION_FILE_STRATEGY', 'FNM_LOGLEVEL', + 'FNM_NODE_DIST_MIRROR', 'FNM_ARCH', 'FNM_COREPACK_ENABLED', 'FNM_RESOLVE_ENGINES', +]) const ACP_MAX_LINE_BYTES = 1024 * 1024 const ACP_MAX_INPUT_BYTES = 16 * 1024 * 1024 const ACP_MAX_REPLY_BYTES = 10 * 1024 * 1024 @@ -568,7 +583,9 @@ function withAcpRuntimeLock(key, callback) { } function runtimeSignature(agent, workdir, spec, profile, childEnv) { - const environment = Object.keys(childEnv).sort().map(key => [key, childEnv[key]]) + const environment = Object.keys(childEnv).sort() + .filter(key => !ACP_VOLATILE_ENV_KEYS.has(key)) + .map(key => [key, childEnv[key]]) return createHash('sha256').update(JSON.stringify({ kind: agent.kind, executable: agent.executable, @@ -681,8 +698,14 @@ async function createAcpRuntime(input) { }) child.stderr.on('data', (chunk) => { try { runtime.activeTurn?.noteActivity?.() } catch { /* activity is best-effort */ } + // Keep only recent stderr. A long-lived process used to keep the first + // megabyte of its whole lifetime, so a new failure was classified from + // stale early lines (for example an old auth warning). + runtime.stderr.push(chunk) runtime.stderrBytes += chunk.length - if (runtime.stderrBytes <= 1024 * 1024) runtime.stderr.push(chunk) + while (runtime.stderrBytes > ACP_STDERR_TAIL_BYTES && runtime.stderr.length > 1) { + runtime.stderrBytes -= runtime.stderr.shift().length + } }) runtime.observedStdout = createObservedAcpInput( child.stdout, runtime.replyState, protocolLabel, @@ -741,10 +764,30 @@ async function createAcpRuntime(input) { ) runtime.connection = new ClientSideConnection(() => client, stream) acpInitializingRuntimes.add(runtime) + const abortSignal = input.options?.signal + const initializeTimeoutMs = Number.isFinite(input.options?.acpInitializeTimeoutMs) + && input.options.acpInitializeTimeoutMs > 0 + ? input.options.acpInitializeTimeoutMs + : ACP_INITIALIZE_TIMEOUT_MS + let initializeTimer = null + let abortListener = null try { + // initialize used to wait forever while holding the per-conversation + // lock, leaving the Agent unusable in that conversation until restart. const initialized = await Promise.race([ runtime.connection.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }), runtime.failurePromise, + new Promise((_, reject) => { + initializeTimer = setTimeout(() => reject(agentExecutionError( + 'LOCAL_AGENT_PROTOCOL_UNAVAILABLE', + `${protocolLabel} did not answer the ACP initialize request.`, + )), initializeTimeoutMs) + }), + new Promise((_, reject) => { + abortListener = () => reject(agentExecutionError('LOCAL_AGENT_EXECUTION_STOPPED')) + if (abortSignal?.aborted) abortListener() + else abortSignal?.addEventListener('abort', abortListener, { once: true }) + }), ]) runtime.agentCapabilities = initialized?.agentCapabilities && typeof initialized.agentCapabilities === 'object' @@ -752,8 +795,11 @@ async function createAcpRuntime(input) { : {} } catch (error) { await closeAcpRuntime(runtime) + if (abortSignal?.aborted) throw agentExecutionError('LOCAL_AGENT_EXECUTION_STOPPED') throw allowAcpSetupFallback(acpProtocolError(error, childEnv)) } finally { + clearTimeout(initializeTimer) + if (abortListener) abortSignal?.removeEventListener('abort', abortListener) acpInitializingRuntimes.delete(runtime) } return runtime @@ -801,6 +847,8 @@ async function runAcpTurn(runtime, prompt, options, spec, profile) { pendingPermissionCallbacks: new Set(), } runtime.activeTurn = turn + runtime.stderr = [] + runtime.stderrBytes = 0 runtime.replyState.bytes = 0 runtime.replyState.inputBytes = 0 runtime.replyState.observedInputBytes = 0 @@ -817,6 +865,8 @@ async function runAcpTurn(runtime, prompt, options, spec, profile) { runtime.publicSessionRef = visibleSessionRef } let promptStarted = false + let promptRequest = null + let caughtError = null let abortRequested = false let timeoutRequested = false let cancelPromise = Promise.resolve() @@ -881,10 +931,12 @@ async function runAcpTurn(runtime, prompt, options, spec, profile) { let promptResult try { promptStarted = true - promptResult = await operation(runtime.connection.prompt({ + promptRequest = Promise.resolve(runtime.connection.prompt({ sessionId: protocolSessionRef, prompt: [{ type: 'text', text: prompt }, ...imageBlocks], })) + promptRequest.catch(() => {}) + promptResult = await operation(promptRequest) } finally { runtime.replyState.collecting = false } @@ -927,6 +979,7 @@ async function runAcpTurn(runtime, prompt, options, spec, profile) { if (!promptStarted && normalized.message !== 'LOCAL_AGENT_SESSION_INVALID') { allowAcpSetupFallback(normalized) } + caughtError = normalized throw normalized } finally { clearTimeout(timeout) @@ -939,7 +992,25 @@ async function runAcpTurn(runtime, prompt, options, spec, profile) { turn.pendingPermissionCallbacks.clear() runtime.activeTurn = null runtime.lastUsedAt = Date.now() - if (abortRequested) await settleWithin(cancelPromise, ACP_CANCEL_GRACE_MS) + if (abortRequested) { + await settleWithin(cancelPromise, ACP_CANCEL_GRACE_MS) + // A user stop or watchdog timeout used to destroy the process and its + // native session. Keep both when the Agent honours the cancellation. + let promptSettled = false + // Disposable runtimes are always closed after the turn; only a + // persistent runtime is worth waiting for. + if (promptRequest && !acpDisposableRuntimes.has(runtime)) { + await settleWithin( + promptRequest.then(() => { promptSettled = true }, () => { promptSettled = true }), + ACP_CANCEL_SETTLE_MS, + ) + } + if (caughtError && promptSettled && !runtime.closed && !runtime.failureError) { + Object.defineProperty(caughtError, 'acpRuntimeReusable', { + value: true, enumerable: false, configurable: true, + }) + } + } } } @@ -1003,8 +1074,10 @@ async function runPersistentAcpAgent(agent, prompt, workdir, options, spec, prof try { return await runAcpTurn(runtime, prompt, options, spec, profile) } catch (error) { - acpSessionRuntimes.delete(key) - await closeAcpRuntime(runtime) + if (error?.acpRuntimeReusable !== true) { + acpSessionRuntimes.delete(key) + await closeAcpRuntime(runtime) + } throw error } finally { runtime.active = false diff --git a/desktop/test/agents/cli/cli-native-acp-lifecycle.test.cjs b/desktop/test/agents/cli/cli-native-acp-lifecycle.test.cjs index ee3b958..eebf20f 100644 --- a/desktop/test/agents/cli/cli-native-acp-lifecycle.test.cjs +++ b/desktop/test/agents/cli/cli-native-acp-lifecycle.test.cjs @@ -221,6 +221,129 @@ input.on('line', (line) => { assert.match(fallback.text, /^legacy chat --quiet --image \/diagram\.png/) }) +function resilientHermes(directory) { + return executable(directory, 'hermes-resilient-acp.cjs', ` +const fs = require('node:fs') +const readline = require('node:readline') +if (process.argv[2] !== 'acp') { + process.stdout.write('legacy fallback') + process.exit(0) +} +const countFile = process.env.MELDWORK_TEST_SPAWN_COUNT_FILE +const spawnCount = (fs.existsSync(countFile) ? Number(fs.readFileSync(countFile, 'utf8')) : 0) + 1 +fs.writeFileSync(countFile, String(spawnCount)) +const input = readline.createInterface({ input: process.stdin }) +const send = value => process.stdout.write(JSON.stringify(value) + '\\n') +let pendingSlowPrompt = null +input.on('line', (line) => { + const message = JSON.parse(line) + const sessionId = message.params?.sessionId || 'hermes-resilient-session' + if (message.method === 'initialize') { + if (process.env.MELDWORK_TEST_HANG_FIRST_INIT === '1' && spawnCount === 1) return + send({ jsonrpc: '2.0', id: message.id, result: { protocolVersion: 1 } }) + } else if (message.method === 'session/new') { + send({ jsonrpc: '2.0', id: message.id, result: { sessionId } }) + } else if (message.method === 'session/set_mode') { + send({ jsonrpc: '2.0', id: message.id, result: {} }) + } else if (message.method === 'session/cancel') { + if (pendingSlowPrompt) { + send({ jsonrpc: '2.0', id: pendingSlowPrompt, result: { stopReason: 'cancelled' } }) + pendingSlowPrompt = null + } + } else if (message.method === 'session/prompt') { + const text = message.params?.prompt?.[0]?.text || '' + if (text === 'slow prompt') { + pendingSlowPrompt = message.id + fs.writeFileSync(process.env.MELDWORK_TEST_PROMPT_FILE, 'received') + return + } + send({ jsonrpc: '2.0', method: 'session/update', params: { sessionId, update: { + sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'acp answer ' + spawnCount }, + } } }) + send({ jsonrpc: '2.0', id: message.id, result: { stopReason: 'end_turn' } }) + } +}) +`) +} + +test('a hanging ACP initialize times out and releases the conversation lock', async (t) => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-acp-init-timeout-')) + const spawnCountFile = path.join(directory, 'spawn-count.txt') + t.after(async () => { + await adapters.shutdownAcpSessionRuntime?.() + fs.rmSync(directory, { recursive: true, force: true }) + }) + const agent = { kind: 'hermes', executable: resilientHermes(directory), name: 'Hermes' } + const env = { MELDWORK_TEST_SPAWN_COUNT_FILE: spawnCountFile, MELDWORK_TEST_HANG_FIRST_INIT: '1' } + const first = await within(adapters.runAgent(agent, 'first prompt', directory, { + acpPersistenceKey: 'workspace:hermes:init-timeout', acpInitializeTimeoutMs: 100, env, + }), 5000) + // The stuck initialize no longer blocks: the turn falls back instead of hanging. + assert.equal(first.text, 'legacy fallback') + const second = await within(adapters.runAgent(agent, 'second prompt', directory, { + acpPersistenceKey: 'workspace:hermes:init-timeout', acpInitializeTimeoutMs: 100, env, + }), 5000) + assert.equal(second.text, 'acp answer 2') +}) + +test('a stopped ACP turn keeps the persistent process and its session when cancel is honoured', async (t) => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-acp-cancel-keep-')) + const spawnCountFile = path.join(directory, 'spawn-count.txt') + const promptFile = path.join(directory, 'prompt.txt') + t.after(async () => { + await adapters.shutdownAcpSessionRuntime?.() + fs.rmSync(directory, { recursive: true, force: true }) + }) + const agent = { kind: 'hermes', executable: resilientHermes(directory), name: 'Hermes' } + const env = { MELDWORK_TEST_SPAWN_COUNT_FILE: spawnCountFile, MELDWORK_TEST_PROMPT_FILE: promptFile } + const sessionRefs = [] + const controller = new AbortController() + const stopped = adapters.runAgent(agent, 'slow prompt', directory, { + acpPersistenceKey: 'workspace:hermes:cancel-keep', + env, + signal: controller.signal, + onSessionRef: sessionRef => sessionRefs.push(sessionRef), + }).then(() => null, error => error) + await within(readWhenReady(promptFile), 5000) + controller.abort() + const error = await within(stopped, 5000) + assert.equal(error?.message, 'LOCAL_AGENT_EXECUTION_STOPPED') + + const next = await within(adapters.runAgent(agent, 'next prompt', directory, { + acpPersistenceKey: 'workspace:hermes:cancel-keep', + sessionRef: sessionRefs.at(-1), + sessionTransport: 'acp', + env, + }), 5000) + assert.equal(next.text, 'acp answer 1') + assert.equal(next.sessionRef, sessionRefs.at(-1)) + assert.equal(fs.readFileSync(spawnCountFile, 'utf8'), '1') +}) + +test('a changed PATH between turns keeps the persistent ACP process and session', async (t) => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-acp-path-churn-')) + const spawnCountFile = path.join(directory, 'spawn-count.txt') + t.after(async () => { + await adapters.shutdownAcpSessionRuntime?.() + fs.rmSync(directory, { recursive: true, force: true }) + }) + const agent = { kind: 'hermes', executable: resilientHermes(directory), name: 'Hermes' } + const shellPath = shell => [`/tmp/fnm_multishells/${shell}/bin`, '/usr/bin', '/bin'].join(path.delimiter) + const first = await adapters.runAgent(agent, 'first prompt', directory, { + acpPersistenceKey: 'workspace:hermes:path-churn', + env: { MELDWORK_TEST_SPAWN_COUNT_FILE: spawnCountFile, PATH: shellPath('1001_1') }, + }) + const second = await adapters.runAgent(agent, 'second prompt', directory, { + acpPersistenceKey: 'workspace:hermes:path-churn', + sessionRef: first.sessionRef, + sessionTransport: 'acp', + env: { MELDWORK_TEST_SPAWN_COUNT_FILE: spawnCountFile, PATH: shellPath('1002_2') }, + }) + assert.equal(second.sessionRef, first.sessionRef) + assert.equal(second.text, 'acp answer 1') + assert.equal(fs.readFileSync(spawnCountFile, 'utf8'), '1') +}) + test('ACP closes a start-only tool with the partial terminal outcome', async (t) => { const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-acp-start-only-tool-')) const cli = executable(directory, 'acp-start-only-tool.cjs', `