Skip to content

Commit 5ab5470

Browse files
author
yee.wang
committed
fix: stop streaming after delegated task tool calls
- enter external-wait state immediately when task tool-call is emitted - suppress same-turn assistant and stream events after delegated task calls - preserve only message_delta and result processing until the next round - cover stream_event and assistant fallback paths with regression tests - add trace-id debug logs to distinguish parent and subagent streams
1 parent 52aff38 commit 5ab5470

2 files changed

Lines changed: 246 additions & 20 deletions

File tree

‎src/qoder-language-model.ts‎

Lines changed: 48 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -53,9 +53,9 @@ configure({
5353

5454
// ── 调试日志 — 保存原始 stderr 引用，不受 SDK log filter 影响 ────────────────
5555
const _rawStderr = process.stderr.write.bind(process.stderr)
56-
function debugLog(msg: string): void {
56+
function debugLog(msg: string, traceId?: string): void {
5757
if (process.env.QODER_DEBUG === '1') {
58-
_rawStderr(`[QODER_DEBUG] ${msg}\n`)
58+
_rawStderr(`[QODER_DEBUG${traceId ? ` ${traceId}` : ''}] ${msg}\n`)
5959
}
6060
}
6161

@@ -514,9 +514,10 @@ export class QoderLanguageModel implements LanguageModelV2 {
514514
const cliPath = resolveQoderCLI()
515515
const qoderOptions = buildQoderQueryOptions(options, this.modelId, cliPath, this.providerOptions)
516516

517-
debugLog(`doStream() called, modelId=${this.modelId}, cliPath=${cliPath}`)
518-
debugLog(`doStream() prompt length=${typeof prompt === 'string' ? prompt.length : 'non-string'}`)
519-
debugLog(`doStream() qoderOptions.mcpServers=[${Object.keys(qoderOptions.mcpServers ?? {}).join(',')}]`)
517+
const streamTraceId = randomUUID().slice(0, 8)
518+
debugLog(`doStream() called, modelId=${this.modelId}, cliPath=${cliPath}`, streamTraceId)
519+
debugLog(`doStream() prompt length=${typeof prompt === 'string' ? prompt.length : 'non-string'}`, streamTraceId)
520+
debugLog(`doStream() qoderOptions.mcpServers=[${Object.keys(qoderOptions.mcpServers ?? {}).join(',')}]`, streamTraceId)
520521

521522
// opencode function 工具名集合（由 opencode 管理并执行，如 bash、read、context7_resolve-library-id 等）
522523
// 这些工具调用不带 providerExecuted，让 opencode 负责执行
@@ -529,7 +530,7 @@ export class QoderLanguageModel implements LanguageModelV2 {
529530
.filter((t) => t.type === 'function')
530531
.map((t) => normalizeToolName(t.name))
531532
)
532-
debugLog(`doStream() hasTools=${hasTools}, functionToolNames=[${[...functionToolNames].join(',')}]`)
533+
debugLog(`doStream() hasTools=${hasTools}, functionToolNames=[${[...functionToolNames].join(',')}]`, streamTraceId)
533534

534535
// 每次 query 独立的 AbortController，用于中断后台 qodercli 进程
535536
const abortController = new AbortController()
@@ -591,6 +592,10 @@ export class QoderLanguageModel implements LanguageModelV2 {
591592
// 典型值：tool_use / end_turn
592593
let lastAssistantStopReason: string | undefined
593594

595+
// 一旦 task 这类需要外部完成的 function tool 已经发出并收到 SDK 回放的 tool_result，
596+
// 当前轮后续 assistant/text/tool 输出都应被抑制；上层必须等真实工具结果后再发起下一轮。
597+
let suppressFurtherAssistantContent = false
598+
594599
try {
595600
// query() 是单次查询的最优路径（QoderAgentSDKClient 是双向交互会话，每次 connect() 冷启动更慢）
596601
const qoderQuery = query({ prompt, options: { ...qoderOptions, abortController } })
@@ -599,15 +604,26 @@ export class QoderLanguageModel implements LanguageModelV2 {
599604
for await (const msg of qoderQuery) {
600605
sdkMsgCount++
601606
const m = msg as Record<string, unknown>
602-
debugLog(`SDK msg #${sdkMsgCount}: type=${m.type}${m.subtype ? ` subtype=${m.subtype}` : ''}`)
607+
debugLog(`SDK msg #${sdkMsgCount}: type=${m.type}${m.subtype ? ` subtype=${m.subtype}` : ''}`, streamTraceId)
603608
if (m.type === 'system' && m.subtype === 'init') {
604609
const tools = Array.isArray(m.tools) ? m.tools.join(',') : ''
605610
const mcpServers = Array.isArray(m.mcp_servers) ? JSON.stringify(m.mcp_servers) : '[]'
606-
debugLog(`SDK init: tools=[${tools}] mcp_servers=${mcpServers}`)
611+
debugLog(`SDK init: tools=[${tools}] mcp_servers=${mcpServers}`, streamTraceId)
607612
}
608613

609614
// ── stream_event：增量文本 / 增量工具输入（流式 CLI 支持时） ──
610615
if (m.type === 'stream_event') {
616+
if (suppressFurtherAssistantContent) {
617+
const ev = (m as { event: Record<string, unknown> }).event
618+
if (ev.type === 'message_delta' && isRecord(ev.delta) && typeof ev.delta.stop_reason === 'string') {
619+
lastAssistantStopReason = ev.delta.stop_reason
620+
debugLog(`message_delta (suppressed): stop_reason=${ev.delta.stop_reason}`, streamTraceId)
621+
} else {
622+
debugLog(`suppressed stream_event after external-wait state: ${String(ev.type)}`, streamTraceId)
623+
}
624+
continue
625+
}
626+
611627
const ev = (m as { event: Record<string, unknown> }).event
612628

613629
if (ev.type === 'content_block_start' && isRecord(ev.content_block)) {
@@ -619,7 +635,7 @@ export class QoderLanguageModel implements LanguageModelV2 {
619635
const toolName = normalizeToolName(block.name)
620636
// normalizeToolName 统一处理：大小写 + AskUserQuestion→question + mcp__server__tool→server_tool
621637
const isProviderExecuted = hasTools && !functionToolNames.has(toolName)
622-
debugLog(`tool_use block_start: raw=${block.name} → normalized=${toolName}, inFunctionTools=${functionToolNames.has(toolName)}, isProviderExecuted=${isProviderExecuted}`)
638+
debugLog(`tool_use block_start: raw=${block.name} → normalized=${toolName}, inFunctionTools=${functionToolNames.has(toolName)}, isProviderExecuted=${isProviderExecuted}`, streamTraceId)
623639
streamToolBlocks.set(idx, { id: block.id, name: toolName, input: '', isProviderExecuted })
624640
if (!isProviderExecuted) {
625641
controller.enqueue({
@@ -681,8 +697,13 @@ export class QoderLanguageModel implements LanguageModelV2 {
681697
input: normalizedInput,
682698
} as LanguageModelV2StreamPart)
683699
emittedFunctionToolCall = true
684-
debugLog(`emitted tool-call to opencode: toolName=${toolBlock.name}, id=${toolBlock.id}`)
700+
debugLog(`emitted tool-call to opencode: toolName=${toolBlock.name}, id=${toolBlock.id}`, streamTraceId)
685701
toolBlock.input = normalizedInput
702+
703+
if (waitForExternalCompletionToolNames.has(toolBlock.name)) {
704+
suppressFurtherAssistantContent = true
705+
debugLog(`enter external-wait state immediately after tool-call for toolName=${toolBlock.name}, tool_use_id=${toolBlock.id}`, streamTraceId)
706+
}
686707
}
687708
pendingToolCalls.set(toolBlock.id, {
688709
toolName: toolBlock.name,
@@ -699,11 +720,16 @@ export class QoderLanguageModel implements LanguageModelV2 {
699720
}
700721
} else if (ev.type === 'message_delta' && isRecord(ev.delta) && typeof ev.delta.stop_reason === 'string') {
701722
lastAssistantStopReason = ev.delta.stop_reason
702-
debugLog(`message_delta: stop_reason=${ev.delta.stop_reason}`)
723+
debugLog(`message_delta: stop_reason=${ev.delta.stop_reason}`, streamTraceId)
703724
}
704725

705726
// ── assistant：完整消息块（CLI 不支持流式时走此路径） ──────────
706727
} else if (m.type === 'assistant') {
728+
if (suppressFurtherAssistantContent) {
729+
debugLog('suppressed assistant message after external-wait state', streamTraceId)
730+
continue
731+
}
732+
707733
const rawContent = (m.message as Record<string, unknown> | undefined)?.content
708734
const content = Array.isArray(rawContent) ? rawContent : []
709735
for (const block of content) {
@@ -747,6 +773,11 @@ export class QoderLanguageModel implements LanguageModelV2 {
747773
emittedFunctionToolCall = true
748774
// 更新 input 到 pendingToolCalls
749775
pendingToolCalls.set(block.id, { toolName, input: inputJson, isProviderExecuted })
776+
777+
if (waitForExternalCompletionToolNames.has(toolName)) {
778+
suppressFurtherAssistantContent = true
779+
debugLog(`enter external-wait state immediately after assistant tool-call for toolName=${toolName}, tool_use_id=${block.id}`, streamTraceId)
780+
}
750781
}
751782
}
752783
}
@@ -760,7 +791,7 @@ export class QoderLanguageModel implements LanguageModelV2 {
760791
if (block.type !== 'tool_result' || typeof block.tool_use_id !== 'string') continue
761792

762793
const toolCall = pendingToolCalls.get(block.tool_use_id)
763-
debugLog(`tool_result: tool_use_id=${block.tool_use_id}, found=${!!toolCall}, toolName=${toolCall?.toolName}, isProviderExecuted=${toolCall?.isProviderExecuted}`)
794+
debugLog(`tool_result: tool_use_id=${block.tool_use_id}, found=${!!toolCall}, toolName=${toolCall?.toolName}, isProviderExecuted=${toolCall?.isProviderExecuted}`, streamTraceId)
764795
if (!toolCall) continue
765796

766797
// CLI 内置工具（isProviderExecuted=true）的结果由 provider 自己消费，
@@ -778,9 +809,9 @@ export class QoderLanguageModel implements LanguageModelV2 {
778809
// ── result：会话结束 ────────────────────────────────────────
779810
} else if (m.type === 'result') {
780811
const outstandingToolCallCount = pendingToolCalls.size
781-
debugLog(`result: subtype=${m.subtype}, is_error=${m.is_error}, pendingToolCalls=${outstandingToolCallCount}, emittedFunctionToolCall=${emittedFunctionToolCall}, lastStopReason=${lastAssistantStopReason}`)
812+
debugLog(`result: subtype=${m.subtype}, is_error=${m.is_error}, pendingToolCalls=${outstandingToolCallCount}, emittedFunctionToolCall=${emittedFunctionToolCall}, lastStopReason=${lastAssistantStopReason}`, streamTraceId)
782813
if (outstandingToolCallCount > 0) {
783-
debugLog(`result: outstanding tools: ${[...pendingToolCalls.entries()].map(([id, t]) => `${t.toolName}(${id},prov=${t.isProviderExecuted})`).join(', ')}`)
814+
debugLog(`result: outstanding tools: ${[...pendingToolCalls.entries()].map(([id, t]) => `${t.toolName}(${id},prov=${t.isProviderExecuted})`).join(', ')}`, streamTraceId)
784815
}
785816

786817
// 关闭所有未关闭的文本块和推理块
@@ -819,7 +850,7 @@ export class QoderLanguageModel implements LanguageModelV2 {
819850
const hasOutstandingFunctionToolCalls = emittedFunctionToolCall && outstandingToolCallCount > 0
820851
const mappedFinishReason: LanguageModelV2FinishReason =
821852
hasOutstandingFunctionToolCalls ? 'tool-calls' : 'stop'
822-
debugLog(`finish: mappedFinishReason=${mappedFinishReason} (emittedFunctionToolCall=${emittedFunctionToolCall}, outstandingToolCallCount=${outstandingToolCallCount})`)
853+
debugLog(`finish: mappedFinishReason=${mappedFinishReason} (emittedFunctionToolCall=${emittedFunctionToolCall}, outstandingToolCallCount=${outstandingToolCallCount})`, streamTraceId)
823854
controller.enqueue({
824855
type: 'finish',
825856
finishReason: decorateFinishReason(mappedFinishReason),
@@ -840,14 +871,14 @@ export class QoderLanguageModel implements LanguageModelV2 {
840871
}
841872

842873
if (!hasFinish) {
843-
debugLog('stream ended without result message, emitting fallback finish=stop')
874+
debugLog('stream ended without result message, emitting fallback finish=stop', streamTraceId)
844875
controller.enqueue({
845876
type: 'finish',
846877
finishReason: decorateFinishReason('stop'),
847878
usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 },
848879
})
849880
}
850-
debugLog('stream complete, closing controller')
881+
debugLog('stream complete, closing controller', streamTraceId)
851882
cleanup()
852883
controller.close()
853884
} catch (err) {

0 commit comments

Comments
 (0)