mirror of
https://github.com/Ryder-Sun/Meldwork.git
synced 2026-11-07 06:56:03 -10:00
fix: keep long-lived ACP Agents usable across stops, restarts and hangs
Persistent ACP runtimes (Hermes, Kimi, MiMo, OpenCode) were fragile: - any turn error, including a user stop or watchdog timeout, destroyed the process and its native session; Hermes rejects cross-process resume, so the conversation then fell back to the silent legacy CLI; - the process identity hashed the whole child environment, so a fresh login-shell PATH (fnm uses a per-shell directory) restarted the process and invalidated the stored session; - initialize had no timeout and ignored stop while holding the conversation lock, leaving that Agent stuck until the app restarted; - stderr kept the first megabyte of the process lifetime, so new failures were classified from stale lines. Now a stopped or timed-out turn waits up to 3 s for the Agent to answer the cancelled prompt and keeps the process and session when it does; volatile shell values (PATH, TERM, TMPDIR, fnm variables, ...) no longer change the process identity; initialize times out after 60 s (settable per call) and honours stop without falling back; stderr keeps only the current turn's last 64 KB. Disposable runtimes still close after every turn. Checks: cli-native-acp-lifecycle with three new fake-ACP tests (hanging initialize releases the lock, a cancelled turn reuses one process and session, a changed PATH keeps the process), plus the existing lifecycle, adapter and protocol suites (185 pass). Full desktop suite: 1528 pass.
This commit is contained in:
1 parent
28fb01eff5
commit
6522d357e0
2 files changed
+202
-6
No files matched your search
@@ -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
|
||||
|
||||
@@ -221,6 +221,129 @@ input.on('line', (line) => {
|
||||
assert.match(fallback.text, /^legacy chat --quiet --image <dir>\/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', `
|
||||
|
||||
Reference in new issue
Block a user