fix: an ACP turn no longer dies on traffic this client does not know

The ACP transport failed the whole turn on anything it did not recognise: a
banner line on stdout, a notification with an unknown method, a session update
kind added by a newer Agent, or an unknown request. Every Agent upgrade that
adds a message therefore broke Meldwork until it was updated, and a single
large tool result (over 1 MiB on one line) failed the turn.

- stdout lines that are not JSON objects are ignored
- unknown notifications and session updates that fail validation are dropped
  before they reach the SDK, which logs rejected messages with their params
- unknown requests are forwarded with their params stripped, so the Agent gets
  a proper 'method not found' answer instead of waiting, and nothing
  secret-bearing is logged
- the per-line limit is 4 MiB (still bounded; total input and reply limits are
  unchanged)
- truncated JSON that starts like a message, an invalid response payload and
  invalid permission requests still fail closed

The Kimi 'unsafe protocol input' test described the old behaviour; it now
checks that malformed JSON still fails closed and that unknown traffic carrying
a secret neither fails the turn nor reaches a log. The new tolerance test
fails on the previous commit (LOCAL_AGENT_PROCESS_FAILED).

Checked with: cli-adapters-agent-protocols (69 pass), cli-native-acp-lifecycle
(15), cli-acp-tolerance, cli-adapters (64).
This commit is contained in:
Ryder Sun committed 2026-10-16 11:21:36 +08:00
1 parent a97afa9b5b
commit 462c1bd783
3 files changed
+144 -23

No files matched your search

+32 -3
View File
@@ -40,8 +40,9 @@ const ACP_VOLATILE_ENV_KEYS = new Set([
'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_LINE_BYTES = 4 * 1024 * 1024
const ACP_MAX_INPUT_BYTES = 16 * 1024 * 1024
const ACP_UNKNOWN_METHOD_NAME_LIMIT = 120
const ACP_MAX_REPLY_BYTES = 10 * 1024 * 1024
const ACP_PERSISTENT_RUNTIME_LIMIT = 16
const ACP_PERMISSION_OPTION_LIMIT = 16
@@ -105,6 +106,10 @@ function acpTransportError(diagnostic) {
return agentExecutionError('LOCAL_AGENT_PROCESS_FAILED', diagnostic)
}
// Returns the message to deliver, or null for traffic that is safe to ignore.
// A newer Agent can send updates and notifications this client does not know;
// failing the whole turn over them would break Meldwork on every Agent upgrade,
// so only messages that decide what is executed or approved stay strict.
function validateAcpInboundMessage(message, validators, replyState, protocolLabel) {
if (!message || typeof message !== 'object' || Array.isArray(message)
|| message.jsonrpc !== '2.0') {
@@ -116,7 +121,8 @@ function validateAcpInboundMessage(message, validators, replyState, protocolLabe
if (message.method === 'session/update' && !hasId) {
const parsed = validators.zSessionNotification.safeParse(message.params)
if (!parsed.success) {
throw acpTransportError(`${protocolLabel} returned invalid session update parameters.`)
replyState.ignoredMessages = (replyState.ignoredMessages || 0) + 1
return null
}
const update = parsed.data.update
if (replyState.collecting !== false
@@ -136,7 +142,23 @@ function validateAcpInboundMessage(message, validators, replyState, protocolLabe
}
return message
}
throw acpTransportError(`${protocolLabel} requested an unsupported client method.`)
// An unknown request still gets the connection's "method not found"
// answer, so the Agent is not left waiting. The SDK logs every request it
// rejects together with its parameters, so only the method name is
// forwarded. An unknown notification needs no answer and is dropped
// before it can reach the SDK's logging.
const validId = typeof message.id === 'string'
|| (Number.isInteger(message.id) && message.id >= 0)
if (hasId && validId) {
return {
jsonrpc: '2.0',
id: message.id,
method: message.method.slice(0, ACP_UNKNOWN_METHOD_NAME_LIMIT),
params: {},
}
}
replyState.ignoredMessages = (replyState.ignoredMessages || 0) + 1
return null
}
if (!Number.isInteger(message.id) || message.id < 0) {
@@ -232,6 +254,12 @@ function boundedAcpStream(
const parseLine = (line) => {
const text = line.toString('utf8').trim()
if (!text) return
// Some CLIs print banners or warnings on stdout next to the protocol.
// Anything that is not an object is not protocol traffic.
if (!text.startsWith('{')) {
replyState.ignoredMessages = (replyState.ignoredMessages || 0) + 1
return
}
let message
try {
message = JSON.parse(text)
@@ -241,6 +269,7 @@ function boundedAcpStream(
const validated = validateAcpInboundMessage(
message, validators, replyState, protocolLabel,
)
if (!validated) return
if (message.method === 'session/request_permission') {
replyState.pendingPermissionRequestIds.add(message.id)
} else if (message.method === 'session/update' && message.params?.update) {
@@ -0,0 +1,72 @@
const test = require('node:test')
const assert = require('node:assert/strict')
const fs = require('node:fs')
const os = require('node:os')
const path = require('node:path')
const adapters = require('../../../src/agents/cli/cli-adapters.cjs')
const {
executable,
readJsonWhenReady,
within,
} = require('../../support/cli-adapters-test-helpers.cjs')
function noisyHermes(directory) {
return executable(directory, 'hermes-noisy-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 input = readline.createInterface({ input: process.stdin })
const send = value => process.stdout.write(JSON.stringify(value) + '\\n')
process.stdout.write('Hermes 9.9.9 starting up...\\n')
input.on('line', (line) => {
let message
try { message = JSON.parse(line) } catch { return }
const sessionId = message.params?.sessionId || 'noisy-session'
if (message.id === 9001 && !message.method) {
fs.writeFileSync(process.env.MELDWORK_TEST_UNKNOWN_RESPONSE, JSON.stringify(message))
return
}
if (message.method === 'initialize') {
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/prompt') {
process.stdout.write('WARNING: update available (not protocol traffic)\\n')
send({ jsonrpc: '2.0', method: '_vendor/progress', params: { done: 1 } })
send({ jsonrpc: '2.0', method: 'session/update', params: { sessionId, update: { sessionUpdate: 'brand_new_update_kind', payload: 1 } } })
send({ jsonrpc: '2.0', id: 9001, method: 'vendor/unknown_request', params: { sessionId } })
send({ jsonrpc: '2.0', method: 'session/update', params: { sessionId, update: {
sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'acp answer despite the noise' },
} } })
setTimeout(() => send({ jsonrpc: '2.0', id: message.id, result: { stopReason: 'end_turn' } }), 50)
}
})
`)
}
test('ACP turns survive banners, unknown notifications, new update kinds and unknown requests', async (t) => {
const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-acp-tolerance-'))
t.after(async () => {
await adapters.shutdownAcpSessionRuntime?.()
fs.rmSync(directory, { recursive: true, force: true })
})
const unknownResponse = path.join(directory, 'unknown-response.json')
const agent = { kind: 'hermes', executable: noisyHermes(directory), name: 'Hermes' }
const result = await within(adapters.runAgent(agent, 'say something', directory, {
acpPersistenceKey: 'workspace:hermes:tolerance',
env: { MELDWORK_TEST_UNKNOWN_RESPONSE: unknownResponse },
}), 10000)
assert.equal(result.text, 'acp answer despite the noise')
// The Agent is told the method is unknown instead of being left waiting.
const response = await readJsonWhenReady(unknownResponse)
assert.equal(response.id, 9001)
assert.equal(response.error?.code, -32601)
})
@@ -859,7 +859,7 @@ input.on('line', (line) => {
)
})
test('Kimi ACP rejects unsafe protocol input without logging secret-bearing messages', async (t) => {
test('Kimi ACP fails closed on malformed JSON and ignores unknown traffic without logging secret-bearing messages', async (t) => {
const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'meldwork-kimi-acp-invalid-'))
const secret = 'private-kimi-transport-secret'
const cli = executable(directory, 'kimi-acp-invalid.cjs', `
@@ -878,10 +878,23 @@ input.on('line', (line) => {
if (process.env.MELDWORK_TEST_CASE === 'malformed') {
process.stdout.write('{"secret":"' + process.env.MELDWORK_TEST_SECRET + '"\\n')
} else {
// Traffic this client does not know: a banner, a notification and a
// request, each carrying the secret. None of it may fail the turn or
// reach a log.
process.stdout.write('banner ' + process.env.MELDWORK_TEST_SECRET + '\\n')
send({
jsonrpc: '2.0', method: 'unsupported/client_method',
params: { secret: process.env.MELDWORK_TEST_SECRET },
})
send({
jsonrpc: '2.0', id: 77, method: 'unsupported/client_request',
params: { secret: process.env.MELDWORK_TEST_SECRET },
})
send({ jsonrpc: '2.0', method: 'session/update', params: {
sessionId: 'kimi-invalid-session',
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'carried on' } },
} })
setTimeout(() => send({ jsonrpc: '2.0', id: message.id, result: { stopReason: 'end_turn' } }), 30)
}
}
})
@@ -889,27 +902,34 @@ input.on('line', (line) => {
t.after(() => fs.rmSync(directory, { recursive: true, force: true }))
const logged = []
const originalConsoleError = console.error
console.error = (...args) => logged.push(args.map(String).join(' '))
console.error = (...args) => logged.push(args.map(arg => (
typeof arg === 'string' ? arg : JSON.stringify(arg)
)).join(' '))
t.after(() => { console.error = originalConsoleError })
for (const testCase of ['malformed', 'unsupported']) {
await assert.rejects(
runAgent(
{ kind: 'kimi', executable: cli, name: 'Kimi' },
'validate transport',
directory,
{ env: { MELDWORK_TEST_CASE: testCase, MELDWORK_TEST_SECRET: secret } },
),
(error) => {
assert.equal(error.message, 'LOCAL_AGENT_PROCESS_FAILED')
assert.equal(error.diagnostic.includes(secret), false)
assert.equal(Object.prototype.propertyIsEnumerable.call(error, 'diagnostic'), false)
return true
},
)
}
await assert.rejects(
runAgent(
{ kind: 'kimi', executable: cli, name: 'Kimi' },
'validate transport',
directory,
{ env: { MELDWORK_TEST_CASE: 'malformed', MELDWORK_TEST_SECRET: secret } },
),
(error) => {
assert.equal(error.message, 'LOCAL_AGENT_PROCESS_FAILED')
assert.equal(error.diagnostic.includes(secret), false)
assert.equal(Object.prototype.propertyIsEnumerable.call(error, 'diagnostic'), false)
return true
},
)
assert.deepEqual(logged, [])
const survived = await runAgent(
{ kind: 'kimi', executable: cli, name: 'Kimi' },
'validate transport',
directory,
{ env: { MELDWORK_TEST_CASE: 'unknown', MELDWORK_TEST_SECRET: secret } },
)
assert.equal(survived.text, 'carried on')
assert.equal(logged.some(entry => entry.includes(secret)), false)
})
test('Kimi ACP bounds unframed input, cumulative reply text, and total protocol traffic', async (t) => {
@@ -929,7 +949,7 @@ input.on('line', (line) => {
} else if (message.method === 'session/prompt') {
const testCase = process.env.MELDWORK_TEST_CASE
if (testCase === 'line') {
process.stdout.write('x'.repeat(1024 * 1024 + 1))
process.stdout.write('x'.repeat(4 * 1024 * 1024 + 1))
return
}
const update = testCase === 'reply' ? 'agent_message_chunk' : 'agent_thought_chunk'