Repository navigation
[llm-deepseek] 一个非法的流式工具入参会终止整轮（MALFORMED_RESPONSE，且不在默认重试白名单） #7581
CooperZhuang
started this conversation in
General
Replies: 1 comment 1 reply
Additional data point: a recoverable bracket slip in streamed tool input ends the whole turn with non-retryable MALFORMED_RESPONSESummaryWhen a DeepSeek Messages stream finishes normally ( In our case the stream was complete and the concatenation was correct. The model had dropped one closing Adding to this thread (same failure family as #8976, #9402, #10195 and #10406). This comment adds one concrete, repairable shape (a structural bracket slip, not a string-escape issue) and a confirmation on the current ReproductionAny tool whose input has nested arrays of objects works. Synthetic, sanitized example of the defect we observed (the outer array is never closed): {"items":[{"id":"a","query":{"top":3}}}
Our real occurrence: 978-character input, 360 Recoverability of the same input:
Current behavior{"type":"turn/end","data":{"reason":{"kind":"error","error":{"message":"DeepSeek Messages stream: tool input is invalid JSON","code":"MALFORMED_RESPONSE"}}}}
Expected behaviorA model-side formatting slip in tool arguments should degrade to a recoverable, per-call failure, not a terminal turn error. Either:
Root cause (master @ 687f282)
Inconsistent with the rest of the pipeline:
Proposed fix
We are glad to contribute a patch with a unit test (an SSE fixture for the bracket slip, an escape slip and an unrepairable input) for options 1 and 2. CONTRIBUTING says external PRs are not accepted right now, so we can publish it on a fork branch for maintainers to pick up, if that helps. Workaround available todayOverride the retry table on the DeepSeek provider config. It replaces the whole list, so include the defaults: retryPolicy:
mode: normal
retryableCodes: [EMPTY_RESPONSE, RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT, MALFORMED_RESPONSE]The failed attempt is not committed to history, so the step is resampled. The cost is one extra step of tokens, and the model still gets no feedback about the error. Environment
Related#8976 (retry-table patch on a fork), #9402, #10195, #10406, #9503 (survey of how other platforms repair), #9066 (pi-ai |
1 reply
Workaround plugin until
|
| case | exit | result |
|---|---|---|
baseline (plugin disabled), {"items":[{"id":"a","query":{"top":3}}} |
1 | turn_end error MALFORMED_RESPONSE (same as the report) |
| defaults, same input | 0 | repaired [insert ] @38] → finish converted → tool ran with {"items":[{"id":"a","query":{"top":3}}]} → completed |
defaults, unterminated string …"filter":"x eq '1'}} |
0 | passed-raw → isError ("Unterminated string in JSON at position 49" + tail) → model rewrote the call → tool ran → completed |
rawOnlyTools: [<tool>], bracket slip |
0 | not repaired → isError (position 38) → rewrite → completed |
| retry only (repair/passRaw off), resample valid | 0 | retry 1/1 → tool ran → completed |
| retry only, always invalid | 1 | one retry, then the original error ends the turn (bounded) |
Unit tests: 15/15 pass. They cover the repair cases (refuses unterminated strings, tail completion without closeTail, >maxEdits, {}, non-objects; leaves string content untouched), the middleware staying silent on other messages and on max-tokens, the retry budget resetting per turn, and clean listener load → disable → re-enable → dispose (1/1/1 → 0/0/0 → 1/1/0 → 0/0/0).
Limitations and risks
- The failed step's thinking signature and usage stay inside the adapter's
translate(), so they are omitted (never fabricated). The next request sends that thinking block unsigned, which may cost a cache hit, and token stats miss one step. - Structural repair can change meaning. A
]inserted at the wrong depth is still valid JSON. For write or side-effect tools, put them inrawOnlyTools(or setrepair.enabled: false) so they only ever get raw → isError and the model rewrites the call itself. - Matching uses the exact message text, so the plugin is fragile by design: if the text changes, it does nothing.
- Approval/permission prompts may fire for the raw call before the plugin's isError rejection.
- Chunks after the bad tool-call block are delayed until that stream's
finish.
The proper fix is still in translate.ts (repair at message_stop, or emit the raw arguments instead of throwing). Once upstream ships that, set enabled: false. MIT License.
index.js (plugin, MIT)
/**
* dsh-toolargs-repair — community stopgap for DeepSeek Harness.
*
* Problem: on the DeepSeek Messages route, translate() throws
* MALFORMED_RESPONSE ("DeepSeek Messages stream: tool input is invalid JSON")
* at message_stop when one streamed tool input is not valid JSON. The whole
* turn ends; the model never learns what went wrong. See discussions #7581,
* #8976, #9402, #10195, #10406.
*
* What this plugin does (all on public extension points, no monkey-patching):
* 1. `llm/stream` middleware: when a tool-call block-end carries arguments
* that are not a JSON object, the stream is held from that point until
* its finish chunk. Only if the finish is the exact DeepSeek
* MALFORMED_RESPONSE failure, the held tool calls are
* a) repaired conservatively (insert a skipped `]`/`}`, drop a trailing
* comma, at most 3 edits, result must be a non-empty plain object), or
* b) passed through with their raw text,
* and the terminal error is replaced by a normal `tool-calls` finish.
* Any other finish: the held chunks are released unchanged.
* 2. `tools/execute`: a raw (unrepaired) call flagged by step 1 gets an
* isError result with the parse position and a short tail of the input,
* so the model rewrites the call. Nothing is executed.
* 3. `agent/request-error` (last resort, off unless enabled): a capped
* resample for the same failure, counted per agent per turn.
*
* Every switch is in the Schemastery config. MIT License.
*/
import z from '@deepseek-ai/schemastery'
export const name = 'dsh-toolargs-repair'
export const inject = ['llm']
const DEFAULT_MESSAGES = [
'DeepSeek Messages stream: tool input is invalid JSON',
'DeepSeek Messages expected a JSON object'
]
export const Config = z.object({
enabled: z.boolean().default(true).description('Master switch.'),
repair: z.object({
enabled: z.boolean().default(true).description('Try a conservative structural repair before passing the raw text through.'),
maxEdits: z.natural().max(3).default(3).description('Maximum structural edits (1-3).'),
closeTail: z.boolean().default(false).description('Also append closers missing at the very end (looks like truncation; off by default).'),
rawOnlyTools: z.array(z.string()).default([]).description('Tool names that never get repaired (e.g. write/delete tools); they always go raw -> isError.')
}).description('Structural repair'),
passRaw: z.boolean().default(true).description('If repair is off or fails, keep the raw text and let the turn continue (the call is answered with isError).'),
toolError: z.boolean().default(true).description('Answer raw (non-object) calls flagged by this plugin with an isError tool result instead of running the tool.'),
retry: z.object({
enabled: z.boolean().default(false).description('Resample the step on the same failure (only reached when the stream fix did not handle it).'),
maxPerTurn: z.natural().max(5).default(1).description('Resample budget per agent per turn.')
}).description('Capped retry (last resort)'),
messages: z.array(z.string()).default(DEFAULT_MESSAGES).description('Exact MALFORMED_RESPONSE messages to act on. Anything else is left alone.'),
logRawChars: z.natural().default(0).description('Include up to N chars of the raw arguments in the log (0 = length only).')
})
/**
* Conservative structural repair. Fixes only bracket slips outside strings:
* a closer that skips an open container (insert the skipped closers), a
* trailing comma before a closer, and (optionally) closers missing at the end.
* Never invents keys or values, never touches string content.
*/
export function repairToolJson(raw, { maxEdits = 3, closeTail = false } = {}) {
if (typeof raw !== 'string' || raw.trim() === '') return { ok: false, reason: 'empty' }
const out = [], stack = [], edits = []
let inString = false, escape = false
for (let i = 0; i < raw.length; i++) {
const ch = raw[i]
if (inString) {
out.push(ch)
if (escape) escape = false
else if (ch === '\\') escape = true
else if (ch === '"') inString = false
continue
}
if (ch === '"') { inString = true; out.push(ch); continue }
if (ch === '{') { stack.push('}'); out.push(ch); continue }
if (ch === '[') { stack.push(']'); out.push(ch); continue }
if (ch === '}' || ch === ']') {
let k = out.length - 1
while (k >= 0 && /\s/.test(out[k])) k--
if (k >= 0 && out[k] === ',') { out.splice(k, 1); edits.push({ at: i, op: 'drop ,' }) }
if (stack.length === 0) return { ok: false, reason: `unexpected ${ch} at ${i}` }
if (stack[stack.length - 1] !== ch) {
const depth = stack.lastIndexOf(ch)
if (depth < 0) return { ok: false, reason: `mismatched ${ch} at ${i}` }
while (stack.length - 1 > depth) { const c = stack.pop(); out.push(c); edits.push({ at: i, op: `insert ${c}` }) }
}
stack.pop(); out.push(ch); continue
}
out.push(ch)
}
if (inString) return { ok: false, reason: 'unterminated string' }
if (stack.length && !closeTail) return { ok: false, reason: `${stack.length} container(s) unclosed at end` }
while (stack.length) { const c = stack.pop(); out.push(c); edits.push({ at: raw.length, op: `append ${c}` }) }
if (edits.length === 0) return { ok: false, reason: 'no structural defect found' }
if (edits.length > maxEdits) return { ok: false, reason: `too many edits (${edits.length} > ${maxEdits})` }
const text = out.join('')
let value
try { value = JSON.parse(text) } catch (e) { return { ok: false, reason: `still invalid: ${e.message}` } }
if (!isPlainObject(value)) return { ok: false, reason: 'result is not a plain object' }
if (Object.keys(value).length === 0) return { ok: false, reason: 'result is {}' }
return { ok: true, text, edits }
}
function isPlainObject(v) { return typeof v === 'object' && v !== null && !Array.isArray(v) }
function isObjectArgs(raw) {
if (!raw) return true // empty input maps to {} downstream; upstream accepts it
try { return isPlainObject(JSON.parse(raw)) } catch { return false }
}
function diagnose(raw) {
try { const v = JSON.parse(raw); return isPlainObject(v) ? undefined : `parsed as ${Array.isArray(v) ? 'an array' : typeof v}, not an object` } catch (e) { return e.message }
}
export function apply(ctx, config) {
const log = ctx.logger(name)
if (!config.enabled) return log.info('disabled by config')
const messages = new Set(config.messages)
const matches = (f) => f?.code === 'MALFORMED_RESPONSE' && messages.has(f.message)
const flagged = new Map() // callId -> raw text, for calls passed through unrepaired (bounded)
const note = (kind, data) => {
log.info(`${kind} ${JSON.stringify(data)}`)
ctx.emit('toolargs-repair/applied', { kind, ...data })
}
const rawInfo = (raw) => config.logRawChars ? { raw: raw.length > config.logRawChars ? raw.slice(0, config.logRawChars) + '…' : raw } : {}
if (config.repair.enabled || config.passRaw) {
ctx.on('llm/stream', (options, next) => (async function* () {
let held = null // chunks held since the first invalid tool-call block-end
for await (const chunk of next()) {
if (!held && chunk.type === 'block-end' && chunk.block?.type === 'tool-call' && !isObjectArgs(chunk.block.arguments)) held = []
if (!held) { yield chunk; continue }
if (chunk.type !== 'finish') { held.push(chunk); continue }
if (chunk.reason?.kind !== 'error' || !matches(chunk.reason.failure)) { yield* held; yield chunk; held = null; continue }
// Exact upstream signature: fix the held tool calls, or give up untouched.
const fixed = []
let unhandled = false
for (const c of held) {
if (c.type !== 'block-end' || c.block?.type !== 'tool-call' || isObjectArgs(c.block.arguments)) { fixed.push(c); continue }
const raw = c.block.arguments, tool = c.block.name, callId = c.block.id
const r = config.repair.enabled && !config.repair.rawOnlyTools.includes(tool) ? repairToolJson(raw, config.repair) : { ok: false, reason: config.repair.enabled ? 'rawOnlyTools' : 'repair disabled' }
if (r.ok) {
note('repaired', { model: options.model, tool, callId, rawLength: raw.length, edits: r.edits, ...rawInfo(raw) })
fixed.push({ ...c, block: { ...c.block, arguments: r.text } })
} else if (config.passRaw) {
note('passed-raw', { model: options.model, tool, callId, rawLength: raw.length, reason: r.reason, diagnostic: diagnose(raw), ...rawInfo(raw) })
flagged.set(callId, raw); if (flagged.size > 256) flagged.delete(flagged.keys().next().value)
fixed.push(c)
} else { unhandled = true; break }
}
const original = held
held = null
if (unhandled) { yield* original; yield chunk; continue } // leave the failure to llm-retry / the retry fallback
yield* fixed
note('finish-converted', { model: options.model, from: chunk.reason.failure.message })
// Thinking signature and usage of this step stay inside the adapter; they are omitted, never fabricated.
yield { type: 'finish', reason: { kind: 'tool-calls' } }
}
if (held) yield* held // stream ended without finish: release untouched
})())
}
if (config.toolError) {
ctx.on('tools/execute', (exec, next) => {
if (!flagged.has(exec.callId) || isPlainObject(exec.arguments)) return next()
const raw = flagged.get(exec.callId); flagged.delete(exec.callId)
const diagnostic = diagnose(raw) ?? 'arguments are not a JSON object'
const tail = raw.length > 120 ? `…${raw.slice(-120)}` : raw
const message = `Tool call rejected before execution: the arguments are not a valid JSON object (${diagnostic}). Nothing was executed. The arguments ended with: ${tail} — re-send the complete call with valid JSON; check that every [ and { is closed in the right order.`
note('tool-error', { tool: exec.name, callId: exec.callId, diagnostic })
return Promise.resolve({ content: [{ type: 'text', text: `Error: ${message}` }], isError: true, error: { message, info: { name: 'ToolArgsError', code: 'INVALID_ARGS', rawLength: raw.length } } })
})
}
if (config.retry.enabled) {
const budget = new WeakMap() // agent -> { turn, used }; resets when the turn number changes
ctx.on('agent/request-error', (payload, next) => {
if (!matches(payload.failure)) return next()
const key = payload.agent ?? ctx
let b = budget.get(key)
if (!b || b.turn !== payload.turn) budget.set(key, b = { turn: payload.turn, used: 0 })
if (b.used >= config.retry.maxPerTurn) return next()
b.used++
note('retry', { turn: payload.turn, step: payload.step, attempt: b.used, maxPerTurn: config.retry.maxPerTurn })
return Promise.resolve({ kind: 'retry' })
})
}
}patch.example.yml
# Usage: dsh --patch ./patch.example.yml (works with any profile, e.g. --profile headless)
# `name` is a module specifier: an absolute path to index.js, or a package name if you install it.
- insert:
- id: toolargs-repair
name: /path/to/dsh-toolargs-repair/index.js
config:
enabled: true
repair:
enabled: true # conservative structural repair first
maxEdits: 3 # 1-3
closeTail: false # do not complete input that just stops (looks like truncation)
rawOnlyTools: [] # e.g. [write_file, delete_record]: never repair, always raw -> isError
passRaw: true # unrepairable -> keep raw text, turn continues
toolError: true # raw call -> isError tool result (parse position + tail)
retry:
enabled: false # last resort: resample the step
maxPerTurn: 1
logRawChars: 0 # 0 = log only the length of the raw argumentstest/fake-deepseek.mjs (fake SSE server)
// Minimal fake of the DeepSeek Messages (Anthropic-style) SSE endpoint. Test only.
import http from 'node:http'
import fs from 'node:fs'
const scenario = process.env.FAKE_SCENARIO || 'bad'
const badArgs = fs.readFileSync(process.env.FAKE_ARGS_FILE, 'utf8')
const goodArgs = '{"items":[{"id":"a","query":{"top":3}}]}'
let n = 0, toolTurns = 0
const sse = (res, events) => { res.writeHead(200, { 'content-type': 'text/event-stream' }); for (const e of events) res.write(`event: ${e.type}\ndata: ${JSON.stringify(e)}\n\n`); res.end() }
const toolTurn = (args) => [
{ type: 'message_start', message: { id: `msg_${n}`, type: 'message', role: 'assistant', content: [], usage: { input_tokens: 100, output_tokens: 0 } } },
{ type: 'content_block_start', index: 0, content_block: { type: 'thinking', thinking: '' } },
{ type: 'content_block_delta', index: 0, delta: { type: 'thinking_delta', thinking: 'Need the batch.' } },
{ type: 'content_block_delta', index: 0, delta: { type: 'signature_delta', signature: 'sig-0001' } },
{ type: 'content_block_stop', index: 0 },
{ type: 'content_block_start', index: 1, content_block: { type: 'tool_use', id: `toolu_${n}`, name: 'batch_lookup', input: {} } },
...args.match(/[\s\S]{1,7}/g).map((c) => ({ type: 'content_block_delta', index: 1, delta: { type: 'input_json_delta', partial_json: c } })),
{ type: 'content_block_stop', index: 1 },
{ type: 'message_delta', delta: { stop_reason: 'tool_use' }, usage: { output_tokens: 42 } },
{ type: 'message_stop' }
]
const textTurn = (text) => [
{ type: 'message_start', message: { id: `msg_${n}`, type: 'message', role: 'assistant', content: [], usage: { input_tokens: 200, output_tokens: 0 } } },
{ type: 'content_block_start', index: 0, content_block: { type: 'text', text: '' } },
{ type: 'content_block_delta', index: 0, delta: { type: 'text_delta', text } },
{ type: 'content_block_stop', index: 0 },
{ type: 'message_delta', delta: { stop_reason: 'end_turn' }, usage: { output_tokens: 5 } },
{ type: 'message_stop' }
]
http.createServer((req, res) => {
let body = ''
req.on('data', (d) => body += d)
req.on('end', () => {
if (!req.url.endsWith('/messages')) { res.writeHead(404); return res.end('{}') }
n++
const json = JSON.parse(body)
if (!json.tools?.length) return sse(res, textTurn('Title')) // side requests (e.g. session title)
const last = json.messages.at(-1)
const results = (Array.isArray(last?.content) ? last.content : []).filter((b) => b.type === 'tool_result')
fs.appendFileSync(process.env.FAKE_LOG, JSON.stringify({ n, results }) + '\n')
if (results.length) {
const text = JSON.stringify(results[0].content)
if (results[0].is_error === true) return sse(res, toolTurn(goodArgs)) // the model rewrites the call
return sse(res, textTurn(`final answer after tool: ${text.slice(0, 120)}`))
}
toolTurns++
if (scenario === 'retry' && toolTurns > 1) return sse(res, toolTurn(goodArgs)) // resample is valid
return sse(res, toolTurn(badArgs))
})
}).listen(Number(process.env.FAKE_PORT), '127.0.0.1', () => console.log('fake up'))test/sample-tool.mjs (synthetic batch tool)
// Test-only tool with a nested batch schema (synthetic).
import fs from 'node:fs'
export const name = 'sample-batch-tool'
export const inject = ['tools']
const OUTPUT = { type: 'object', properties: { content: { type: 'array', items: {} }, isError: { type: 'boolean' } }, required: ['content', 'isError'], additionalProperties: false }
export function apply(ctx) {
ctx.tools.register({
name: 'batch_lookup',
description: 'Look up a batch of records.',
parameters: { type: 'object', properties: { items: { type: 'array', items: { type: 'object', properties: { id: { type: 'string' }, query: { type: 'object' } }, required: ['id'] } } }, required: ['items'] },
output: { schema: OUTPUT, render: (_a, v) => v.content },
execute: async (args) => {
fs.appendFileSync(process.env.TEST_LOG, JSON.stringify({ executedWith: args }) + '\n')
return { content: [{ type: 'text', text: `batch_lookup ok: ${args.items.length} item(s)` }], isError: false }
}
})
ctx.on('toolargs-repair/applied', (e) => fs.appendFileSync(process.env.TEST_LOG, JSON.stringify({ plugin: e }) + '\n'))
}test/run-e2e.sh
#!/bin/bash
# End-to-end run of the real `dsh` CLI against the fake SSE server.
# usage: NODE=<node 24 binary> DSH_BIN=<path to @deepseek-ai/dsh/lib/bin.js> ./run-e2e.sh <label> <fake-scenario> <fixture> <plugin-config-yaml|off>
set -u
cd "$(dirname "$0")"
L=$1; SC=$2; ARGS=$3; CFG=$4
OUT=out/$L; rm -rf "$OUT"; mkdir -p "$OUT" home
PORT=$((18700 + RANDOM % 200))
PLUGIN="$(cd .. && pwd)/index.js"
if [ "$CFG" = off ]; then P=" disabled: true"; else P=" config: $CFG"; fi
cat > "$OUT/patch.yml" <<YML
- insert:
- id: sample-batch-tool
name: $PWD/sample-tool.mjs
- id: toolargs-repair
name: $PLUGIN
$P
YML
FAKE_SCENARIO=$SC FAKE_ARGS_FILE=$PWD/fixtures/$ARGS FAKE_LOG=$PWD/$OUT/fake.jsonl FAKE_PORT=$PORT "$NODE" fake-deepseek.mjs > "$OUT/fake.out" 2>&1 &
FP=$!; sleep 0.5
DSH_HOME=$PWD/home DEEPSEEK_API_KEY=sk-test DEEPSEEK_BASE_URL=http://127.0.0.1:$PORT TEST_LOG=$PWD/$OUT/test.jsonl \
timeout 120 "$NODE" "$DSH_BIN" --profile headless --patch "$PWD/$OUT/patch.yml" --json "look up the batch" > "$OUT/stdout.jsonl" 2> "$OUT/stderr.txt"
echo "exit=$?" > "$OUT/exit.txt"
kill $FP 2>/dev/nulltest/unit.mjs
// Unit + lifecycle checks (no network): node unit.mjs
import assert from 'node:assert/strict'
import { Context } from '@deepseek-ai/cordis'
import * as plugin from '../index.js'
const { repairToolJson, Config } = plugin
const ok = (name, fn) => Promise.resolve().then(fn).then(() => console.log('PASS', name), (e) => { console.log('FAIL', name, e.message); process.exitCode = 1 })
// 1. repair function
await ok('bracket slip repaired with 1 edit', () => {
const r = repairToolJson('{"items":[{"id":"a","query":{"top":3}}}')
assert.equal(r.ok, true); assert.equal(r.text, '{"items":[{"id":"a","query":{"top":3}}]}'); assert.equal(r.edits.length, 1)
})
await ok('trailing comma dropped', () => assert.equal(repairToolJson('{"a":[1,2,],"b":1}').text, '{"a":[1,2],"b":1}'))
await ok('unterminated string refused', () => assert.equal(repairToolJson(`{"q":{"f":"x eq '1'}}`).ok, false))
await ok('missing tail closer refused by default', () => assert.equal(repairToolJson('{"a":[1,2]').ok, false))
await ok('missing tail closer with closeTail', () => assert.equal(repairToolJson('{"a":[1,2]', { closeTail: true }).text, '{"a":[1,2]}'))
await ok('more than maxEdits refused', () => assert.equal(repairToolJson('{"a":[{"b":[{"c":[1}}}}}', { maxEdits: 3 }).ok, false))
await ok('{} never produced', () => assert.equal(repairToolJson('{,}').ok, false))
await ok('non-object refused', () => assert.equal(repairToolJson('[{"a":1}}]').ok, false))
await ok('string content untouched', () => assert.equal(repairToolJson('{"s":"]}","a":[1}').text, '{"s":"]}","a":[1]}'))
await ok('maxEdits capped at 3 by schema', () => assert.throws(() => Config({ repair: { maxEdits: 4 } })))
// 2. middleware + lifecycle on a bare Context
const ctx = new Context()
ctx.provide('llm'); ctx.set('llm', { stub: true })
const hooks = (n) => (ctx.events._hooks[n] || []).length
const names = ['llm/stream', 'tools/execute', 'agent/request-error']
const counts = () => names.map((n) => `${n}=${hooks(n)}`).join(' ')
const fiber = ctx.plugin(plugin, Config({ retry: { enabled: true, maxPerTurn: 1 } }))
await fiber; await new Promise((r) => setTimeout(r, 30))
const loaded = counts()
const bad = { type: 'block-end', index: 1, block: { type: 'tool-call', id: 'c1', name: 't', arguments: '{"items":[{"id":"a"}}' } }
const errFinish = (message) => ({ type: 'finish', reason: { kind: 'error', failure: { code: 'MALFORMED_RESPONSE', message } } })
const run = async (chunks) => { const out = []; for await (const c of ctx.waterfall(null, 'llm/stream', { model: 'm' }, () => (async function* () { yield* chunks })())) out.push(c); return out }
await ok('exact signature: repaired + finish converted', async () => {
const out = await run([bad, errFinish('DeepSeek Messages stream: tool input is invalid JSON')])
assert.equal(out[0].block.arguments, '{"items":[{"id":"a"}]}'); assert.deepEqual(out[1].reason, { kind: 'tool-calls' })
})
await ok('other message: chunks released unchanged (silent)', async () => {
const out = await run([bad, errFinish('something else')])
assert.equal(out[0].block.arguments, bad.block.arguments); assert.equal(out[1].reason.kind, 'error')
})
await ok('normal finish (e.g. max-tokens): unchanged', async () => {
const out = await run([bad, { type: 'finish', reason: { kind: 'max-tokens' } }])
assert.equal(out[0].block.arguments, bad.block.arguments); assert.equal(out[1].reason.kind, 'max-tokens')
})
await ok('retry budget resets per turn', async () => {
const agent = {}
const f = { code: 'MALFORMED_RESPONSE', message: 'DeepSeek Messages stream: tool input is invalid JSON' }
const call = (turn, step) => ctx.waterfall(null, 'agent/request-error', { agent, turn, step, failure: f }, async () => undefined)
assert.deepEqual(await call(1, 1), { kind: 'retry' }); assert.equal(await call(1, 2), undefined)
assert.deepEqual(await call(2, 1), { kind: 'retry' }); assert.equal(await call(2, 1), undefined)
assert.equal(await ctx.waterfall(null, 'agent/request-error', { agent, turn: 3, step: 1, failure: { code: 'TIMEOUT', message: 'x' } }, async () => undefined), undefined)
})
fiber.update?.(Config({ enabled: false })); await new Promise((r) => setTimeout(r, 30))
const disabled = counts()
fiber.update?.(Config({})); await new Promise((r) => setTimeout(r, 30))
const reenabled = counts()
await fiber.dispose(); await new Promise((r) => setTimeout(r, 30))
const disposed = counts()
await ok(`lifecycle: loaded[${loaded}] disabled[${disabled}] re-enabled(default, retry off)[${reenabled}] disposed[${disposed}]`, () => {
assert.equal(loaded, 'llm/stream=1 tools/execute=1 agent/request-error=1')
assert.equal(disabled, 'llm/stream=0 tools/execute=0 agent/request-error=0')
assert.equal(reenabled, 'llm/stream=1 tools/execute=1 agent/request-error=0')
assert.equal(disposed, 'llm/stream=0 tools/execute=0 agent/request-error=0')
})
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
包：
@deepseek-ai/dsh-llm-deepseek0.1.7-alpha.2（涉及dsh-llm、dsh-agent-loop、dsh-tools）路由：
deepseek-official（DeepSeek Messages）· 模型deepseek-flash环境： Windows 11，Node 24，web profile
摘要
只要有一次工具调用的流式
input_json_delta累积出来不是 JSON 对象，整轮就会以MALFORMED_RESPONSE: DeepSeek Messages stream: tool input is invalid JSON终止，该轮已经完成的每一步都被丢弃。我认为这个校验不该卡在流式层：同一个字段的另外两个消费方都是刻意宽容的。模型侧的一次格式打滑，应该退化成一次工具报错，而不是干掉整轮。
实际表现
模型在为
ask_user_question生成参数时，在 JSON 字符串值里写了一个未转义的 ASCII 双引号（跨两个 delta 下发）：{"questions": [{"description": "相当于回到"什么都不设"的原厂行为。"}]}会话日志先记录
assistant/attempt（完整原始流已保留），紧接着是：{"type":"turn/end","data":{"turn":2,"reason":{"kind":"error","error":{"message":"DeepSeek Messages stream: tool input is invalid JSON","code":"MALFORMED_RESPONSE"}}}}这是模型侧的格式问题，不是链路损坏——流本身是完整的，而且该 block 在抛错之前已经以
block-end正常下发。它同样不可重试：dsh-llm的DEFAULT_RETRYABLE_CODES只有EMPTY_RESPONSE, RATE_LIMIT, SERVER, TIMEOUT, TRANSPORT，所以一份坏文档就要赔上整轮，连一次重试都没有。为什么说这个校验是「异类」
同一个字段的另外两个消费方都是有意为之的宽容：
dsh-agent-loop的parseArguments()（lib/index.js:534）——「解析模型参数，把非法 JSON 保留为文本，空输入映射为{}」。无法解析的参数会流到工具层，工具层抛出ToolArgsError（INVALID_ARGS），注册表再经toolErrorResult把它转成isError结果（dsh-toolslib/index.js:3344-3348）——这就是一次普通工具报错，模型下一步可以自行纠正。llm-deepseek的toolInput()（lib/index.js:1308），用于把 assistant 历史重新序列化——「Messages 无法表示的历史参数使用空 input；durable 内容保持不变」。无法解析或非对象的历史在线上降级为{}，而会话记录保留原文。只有
message_stop处的校验（lib/index.js:1729-1738，即lib/types/translate.js/src/types/translate.ts区域）把它当成致命错误。同一处的第二种情况
同一个循环还会拒绝「能解析但不是对象」的 JSON：
object(parsed)会抛MALFORMED_RESPONSE: DeepSeek Messages expected a JSON object，例如整个工具入参是["a"]或"text"。结果同样是整轮终止。（max_tokens截断本来就是豁免的，这更让人觉得这条严格路径并非有意设计。）复现
确定性、可离线，不需要真实模型。用本地 HTTP 服务返回一段合成的 Messages SSE，然后驱动
DeepSeekAdapter.stream()：在 0.1.7-alpha.2 上跑四个用例：
MALFORMED_RESPONSE: tool input is invalid JSONblock-end.arguments["a"]）MALFORMED_RESPONSE: expected a JSON objectusage→finish）max_tokens截断的入参建议修复
把该校验改成 advisory，与上面两条宽容通路一致：
非法文档随后会原样到达工具层，产生
invalid arguments: "arguments" must be an object，模型用正确转义重发——也就是本来设计好的自我纠正路径。相关（可能是有意为之）
MALFORMED_RESPONSE不在DEFAULT_RETRYABLE_CODES中。由于这类失败属于 provider 输出问题而非传输问题（与被重试的EMPTY_RESPONSE同类），也许它也值得重试——不过上面的修复才是真正保证前进，而不是赌重新采样。我目前在本机的临时缓解：打上上述补丁，并把
MALFORMED_RESPONSE加进该路由的retryPolicy.retryableCodes。如果你们需要，我可以开 PR。All reactions