diff --git a/README.md b/README.md index 10b31d5..fc187e9 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,9 @@ OUTPUT_THINK=true # 是否输出思考过程 (true/false) LEGACY_REASONING_IN_CONTENT=false # 推理输出格式，false=reasoning_content字段，true=旧版并入content (true/false) SIMPLE_MODEL_MAP=false # 简化模型映射 (true/false) MODELS_CACHE_TTL=3600 # 模型列表缓存有效期（秒），0=永不过期 +AGENT_TURN_MAX_ATTEMPTS=3 # 单个 Agent 回合生成有效工具调用/最终态的最大尝试数（2-6） +AGENT_CONTEXT_FILE_THRESHOLD_BYTES=92160 # 超过阈值时外置完整 Agent 上下文 +AGENT_CONTEXT_LIVE_PROMPT_BYTES=49152 # 外置后仍内联保留的关键任务状态大小 # 🌐 代理与反代配置 QWEN_CHAT_PROXY_URL= # 自定义 Chat API 反代URL (默认: https://chat.qwen.ai) @@ -131,8 +134,9 @@ CACHE_MODE=default # 图片缓存模式 (default/file) | `LEGACY_REASONING_IN_CONTENT` | 推理输出格式。默认 `false`=推理走独立的 `reasoning_content` 字段；`true`=旧版行为（`` 并入 `content`） | `true` 或 `false` | | `SIMPLE_MODEL_MAP` | 简化模型映射，只返回基础模型不包含变体 | `true` 或 `false` | | `MODELS_CACHE_TTL` | 模型列表缓存有效期（秒），过期后下次请求自动向上游刷新；`0` 表示永不过期 | `3600` | +| `AGENT_TURN_MAX_ATTEMPTS` | 工具请求在一次 HTTP 回合内生成有效 `tool_calls`、明确完成态或阻塞态的最大尝试数；范围 2–6，耗尽后非流式请求返回 HTTP 429/503，SSE 请求返回显式错误帧，绝不伪装成正常 `stop` | `3` | | `AGENT_CONTEXT_FILE_THRESHOLD_BYTES` | Agent 请求体超过此大小时，将完整工具定义和历史自动外置为 Qwen 文本文档，避免触发约 128 KiB 的 WAF 限制 | `92160`（90 KiB） | -| `AGENT_CONTEXT_LIVE_PROMPT_BYTES` | 上下文外置后，实时请求中保留的工具协议和当前回合最大大小 | `49152`（48 KiB） | +| `AGENT_CONTEXT_LIVE_PROMPT_BYTES` | 上下文外置后，实时请求中保留的工具协议、system/developer 指令、原始任务、最近工具进度和当前结果的最大大小 | `49152`（48 KiB） | | `QWEN_CHAT_PROXY_URL` | 自定义 Chat API 反代地址 | `https://your-proxy.com` | | `QWEN_CLI_PROXY_URL` | 自定义 CLI API 反代地址 | `https://your-cli-proxy.com` | | `PROXY_URL` | 出站请求代理地址，支持 HTTP/HTTPS/SOCKS5 | `http://127.0.0.1:7890` | @@ -508,10 +512,14 @@ Authorization: Bearer sk-your-api-key - 流式输出按 OpenAI 规范分片：先发 `function.name + 空 arguments` 头块，随后多个 `arguments` 切片 - 历史消息中的 `assistant.tool_calls` 与 `role:"tool"` 自动折叠回链，`tool_call_id` 精确关联 - `tool_choice` 全四态：`"auto"` / `"required"` / `{type:"function",function:{name:"..."}}` / `"none"` -- `tool_choice="required"` 或指定函数时，若首次未触发工具调用，自动追加强约束提示重试一次 -- 当上游只返回思考、没有正文或工具调用时自动补偿重试一次，避免 Agent 收到空结束态而提前停止 +- 携带工具的请求启用严格三态回合门禁：未完成必须返回真实 `tool_calls`；只有显式确认全部完成或确实阻塞时才能返回 `stop` +- 每个上游生成尝试的裸正文、工具调用和结束状态保持门禁隔离；安全 thinking 与合法 `agent_final` / `agent_blocked` 包装体内的正式正文都会实时显示，但只有闭标签验证通过后才发送 `finish_reason=stop` +- 默认最多执行 3 次协议纠正（可用 `AGENT_TURN_MAX_ATTEMPTS` 配置为 2–6）；耗尽后非流式请求返回真正的 HTTP 429/503，已经建立的 SSE 则发送标准错误帧和 `[DONE]`，两者都不会伪造 `finish_reason=stop` +- 流式请求立即建立真正的 SSE，按上游节奏分别发送 `reasoning_content` 和正式 `content`，并用 SSE 注释帧保活；非流式门禁仍使用 HTTP `102 Processing` 保活，以保留最终真实状态码 +- 协议纠正会复用同一个 Qwen `chatId`，并使用主回答（`response_index=0`）的 `response_id` 作为下一次 `parentId`，避免一次回合制造多个 `New chat` +- thinking 通道若只包含一个完整、合法的工具调用块，也会安全转换为标准 OpenAI `tool_calls`，不会因“只有思考”而中断 Agent - Qwen Web 以干净 HTTP EOF 正常结束时会正确映射为 `stop` / `tool_calls`；只有连接重置等真实传输异常才返回流错误 -- Agent 请求体超过安全阈值时，完整工具定义和历史会通过 Qwen 官方文件接口外置，当前回合仍留在实时提示中，避免长工具循环撞上 WAF/captcha +- Agent 请求体超过安全阈值时，完整上下文会通过 Qwen 官方文件接口外置，同时内联保留 system/developer 指令、原始任务、最近工具链和当前结果；附件失败时也使用同样的关键状态压缩策略 - Qwen 返回 HTTP 200 的 WAF/captcha 业务帧时，会显式返回 `upstream_waf_challenge`，不再伪装为空成功或普通 502 > 对 Codex、Claude Code、OpenClaw 等长时间运行的 Agent，建议保持默认的 90 KiB / 48 KiB 阈值。若反代还会附加较大的请求头或正文，可适当下调 `AGENT_CONTEXT_FILE_THRESHOLD_BYTES`。 diff --git a/src/config/index.js b/src/config/index.js index 71797e9..eb1ff5e 100644 --- a/src/config/index.js +++ b/src/config/index.js @@ -53,6 +53,12 @@ const config = { // chat 请求重试配置（运行时可被 web UI 覆盖，见 src/utils/data-persistence.js#loadSettings） chatRetryCount: Math.max(0, parseInt(process.env.CHAT_RETRY_COUNT, 10) || 1), chatRetryBackoffMs: Math.max(0, parseInt(process.env.CHAT_RETRY_BACKOFF_MS, 10) || 400), + // Agent 回合协议纠正次数。这里是一次 HTTP 回合内的上游生成尝试总数， + // 与传输错误重试分开；耗尽后必须显式失败，绝不能伪装成 finish_reason=stop。 + agentTurnMaxAttempts: Math.min( + 6, + Math.max(2, parseInt(process.env.AGENT_TURN_MAX_ATTEMPTS, 10) || 3) + ), // chat.qwen.ai 的 WAF 会在 JSON 请求体接近 128 KiB 时返回 captcha。 // 提前把 Agent 全量历史外置成文本文档，给协议头和当前回合留出安全余量。 agentContextFileThresholdBytes: Math.max( diff --git a/src/controllers/chat.js b/src/controllers/chat.js index 131114f..8427f9d 100644 --- a/src/controllers/chat.js +++ b/src/controllers/chat.js @@ -13,6 +13,7 @@ const config = require('../config/index.js') const { logger } = require('../utils/logger') const { createUpstreamDeltaNormalizer } = require('../utils/chat-helpers.js') const { assertNoUpstreamFailure } = require('../utils/upstream-error.js') +const { runOpenAIAgentTurn } = require('../utils/openai-agent-runtime.js') const normalizeOpenAIFinishReason = (upstreamReason, hasToolCalls, upstreamCompleted) => { if (hasToolCalls) return 'tool_calls' @@ -38,6 +39,7 @@ const writeOpenAIStreamError = (res, message, code = 'upstream_incomplete') => { } })}\n\n`) res.write('data: [DONE]\n\n') + if (typeof res.flush === 'function') res.flush() res.end() } @@ -51,8 +53,9 @@ const setResponseHeaders = (res, stream) => { if (stream) { res.set({ 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', + 'Cache-Control': 'no-cache, no-transform', 'Connection': 'keep-alive', + 'X-Accel-Buffering': 'no', }) } else { res.set({ @@ -165,7 +168,322 @@ const attributeChatUsage = (account, usage) => { } } +const writeOpenAIHttpError = (res, error = {}) => { + const status = Number(error.status) || 502 + const message = error.message || '上游未能生成有效响应' + const code = error.code || 'upstream_error' + if (res.headersSent) { + if (!res.writableEnded) writeOpenAIStreamError(res, message, code) + return + } + res.status(status) + res.set({ 'Content-Type': 'application/json' }) + res.json({ + error: { + message, + type: status === 429 ? 'rate_limit_error' : 'upstream_error', + code + } + }) +} + +const runWithProcessingHeartbeat = async (res, work, intervalMs = 15000) => { + if (typeof res?.writeProcessing !== 'function') return work() + const heartbeatMs = Math.max(1, Number(intervalMs) || 15000) + const heartbeat = setInterval(() => { + if (res.headersSent || res.writableEnded || res.destroyed) return + try { + // 102 是临时响应，不会提交最终状态码/响应头。这样既能保持长 thinking + // 连接活跃，又能在门禁耗尽时返回真正的 HTTP 429/503。 + res.writeProcessing() + } catch (_) { + // 某些 HTTP/2/反代适配器不实现临时响应；跳过即可，不能改发 SSE 注释。 + } + }, heartbeatMs) + heartbeat.unref?.() + try { + return await work() + } finally { + clearInterval(heartbeat) + } +} + +const runWithSSEHeartbeat = async (res, work, intervalMs = 15000) => { + const heartbeatMs = Math.max(1, Number(intervalMs) || 15000) + const heartbeat = setInterval(() => { + if (res.writableEnded || res.destroyed) return + try { + // SSE 已经提交 200 响应后，用注释帧保活。注释不会进入 OpenAI delta， + // 但能阻止反代在长 thinking 或纠正 attempt 期间把连接判为空闲。 + res.write(': qwen2api-agent-keepalive\n\n') + if (typeof res.flush === 'function') res.flush() + } catch (_) { + // 客户端断开会由后续流消费/写入路径统一收敛。 + } + }, heartbeatMs) + heartbeat.unref?.() + try { + return await work() + } finally { + clearInterval(heartbeat) + } +} + +const normalizeAgentUsage = (attempt, requestBody, completionText) => { + let usage = { ...(attempt?.totalTokens || {}) } + if (!usage.prompt_tokens && !usage.completion_tokens) { + usage = createUsageObject(requestBody?.messages || [], completionText, null) + } + usage.prompt_tokens = Math.max(0, Number(usage.prompt_tokens) || 0) + usage.completion_tokens = Math.max(0, Number(usage.completion_tokens) || 0) + usage.total_tokens = usage.prompt_tokens + usage.completion_tokens + return usage +} + +const prepareAgentOutput = async (attempt, enableThinking, enableWebSearch) => { + let reasoning = String(attempt?.reasoning || '') + let content = attempt?.toolCalls?.length > 0 ? '' : String(attempt?.visibleText || '') + + if (attempt?.webSearchInfo) { + const table = await accountManager.generateMarkdownTable(attempt.webSearchInfo, config.searchInfoMode) + if (enableThinking && reasoning) reasoning = `${table}\n\n${reasoning}` + else if (enableWebSearch && config.searchInfoMode === 'text') { + content = `${content}${content ? '\n\n' : ''}---\n${table}` + } + } + + if (config.legacyReasoningInContent && reasoning) { + content = `\n\n${reasoning}\n\n${content ? `\n${content}` : ''}` + reasoning = '' + } + return { reasoning, content } +} + +const handleOpenAIAgentStream = async ( + res, + response, + enableThinking, + enableWebSearch, + requestBody, + options +) => { + setResponseHeaders(res, true) + const messageId = generateUUID() + const created = Math.round(Date.now() / 1000) + let firstDelta = true + const writeDelta = (delta) => { + if (!delta || Object.keys(delta).length === 0) return + const normalizedDelta = firstDelta ? { role: 'assistant', ...delta } : delta + firstDelta = false + res.write(`data: ${JSON.stringify({ + id: `chatcmpl-${messageId}`, + object: 'chat.completion.chunk', + created, + choices: [{ index: 0, delta: normalizedDelta, finish_reason: null }] + })}\n\n`) + if (typeof res.flush === 'function') res.flush() + } + + // 立即提交标准 SSE 首帧，不能等整个上游 attempt 收完后才让客户端看到响应。 + // 裸正文/工具调用仍由下方门禁缓冲；安全思考与已确认进入 final/blocked + // 包装体的正式正文会按上游节奏增量输出。 + if (typeof res.flushHeaders === 'function') res.flushHeaders() + writeDelta({ role: 'assistant' }) + + const liveReasoningByAttempt = new Map() + const onReasoningDelta = enableThinking && !config.legacyReasoningInContent + ? async (text, metadata = {}) => { + const attemptNumber = Math.max(1, Number(metadata.attemptNumber) || 1) + if (!liveReasoningByAttempt.has(attemptNumber)) { + liveReasoningByAttempt.set(attemptNumber, '') + } + if (text) { + liveReasoningByAttempt.set( + attemptNumber, + `${liveReasoningByAttempt.get(attemptNumber)}${text}` + ) + writeDelta({ reasoning_content: text }) + } + } + : null + const onContentDelta = !config.legacyReasoningInContent + ? async (text) => { + if (text) writeDelta({ content: text }) + } + : null + + let runtime + try { + runtime = await runWithSSEHeartbeat( + res, + () => runOpenAIAgentTurn(response, { + ...options, + requestBody, + sendChatRequest: options.sendChatRequest || sendChatRequest, + on_reasoning_delta: onReasoningDelta, + on_content_delta: onContentDelta + }), + options.agent_processing_heartbeat_ms + ) + } catch (error) { + logger.error('OpenAI Agent 回合处理失败', 'AGENT', '', error) + writeOpenAIHttpError(res, { + status: 502, + message: error.publicMessage || '上游 Agent 回合处理失败', + code: error.code || 'upstream_stream_error' + }) + return + } + if (!runtime.ok) { + writeOpenAIHttpError(res, runtime.error) + return + } + + const { attempt, finishReason } = runtime + const output = await prepareAgentOutput(attempt, enableThinking, enableWebSearch) + let bufferedReasoning = output.reasoning + const acceptedReasoningWasStreamed = liveReasoningByAttempt.has(runtime.attempts) + const rawAcceptedReasoning = String(attempt.reasoning || '') + if (acceptedReasoningWasStreamed && rawAcceptedReasoning && bufferedReasoning.endsWith(rawAcceptedReasoning)) { + // 已实时发送的安全思考不能在门禁通过后再重复回放。若前面附加了搜索表格， + // 只补发这一段派生前缀。 + bufferedReasoning = bufferedReasoning.slice(0, -rawAcceptedReasoning.length) + } + + let bufferedContent = output.content + const streamedVisibleText = String(attempt.streamedVisibleText || '') + const acceptedVisibleText = String(attempt.visibleText || '') + if ( + streamedVisibleText && + acceptedVisibleText.startsWith(streamedVisibleText) && + bufferedContent.startsWith(acceptedVisibleText) + ) { + // 正式回复的包装体已经按上游节奏实时发送，只补发极少数尚未发送的尾部， + // 以及门禁通过后追加的搜索信息等派生内容。 + bufferedContent = `${acceptedVisibleText.slice(streamedVisibleText.length)}${bufferedContent.slice(acceptedVisibleText.length)}` + } else if (streamedVisibleText && finishReason !== 'stop' && attempt.streamedControlState?.opened) { + // length/content_filter 等中止发生在包装闭合前时，不能把带控制标签的原始 + // 缓冲再次作为正文回放；已发送的安全正文由对应 finish_reason 正常收尾。 + bufferedContent = '' + } + + if (bufferedReasoning) writeDelta({ reasoning_content: bufferedReasoning }) + if (bufferedContent) writeDelta({ content: bufferedContent }) + + const ARG_CHUNK_SIZE = 32 + for (const call of attempt.toolCalls || []) { + writeDelta({ + tool_calls: [{ + index: call.index, + id: call.id, + type: 'function', + function: { name: call.function.name, arguments: '' } + }] + }) + const args = call.function.arguments || '{}' + for (let offset = 0; offset < args.length; offset += ARG_CHUNK_SIZE) { + writeDelta({ + tool_calls: [{ + index: call.index, + function: { arguments: args.slice(offset, offset + ARG_CHUNK_SIZE) } + }] + }) + } + } + const completionText = `${output.reasoning}${output.content}${JSON.stringify(attempt.toolCalls || [])}` + const usage = normalizeAgentUsage(attempt, requestBody, completionText) + attributeChatUsage(options.currentAccount, usage) + res.write(`data: ${JSON.stringify({ + id: `chatcmpl-${messageId}`, + object: 'chat.completion.chunk', + created, + choices: [{ index: 0, delta: {}, finish_reason: finishReason }] + })}\n\n`) + res.write(`data: ${JSON.stringify({ + id: `chatcmpl-${messageId}`, + object: 'chat.completion.chunk', + created, + choices: [], + usage + })}\n\n`) + res.write('data: [DONE]\n\n') + if (typeof res.flush === 'function') res.flush() + res.end() +} + +const handleOpenAIAgentNonStream = async ( + res, + response, + enableThinking, + enableWebSearch, + model, + requestBody, + options +) => { + let runtime + try { + runtime = await runWithProcessingHeartbeat( + res, + () => runOpenAIAgentTurn(response, { + ...options, + requestBody, + sendChatRequest: options.sendChatRequest || sendChatRequest + }), + options.agent_processing_heartbeat_ms + ) + } catch (error) { + logger.error('OpenAI 非流式 Agent 回合处理失败', 'AGENT', '', error) + writeOpenAIHttpError(res, { + status: 502, + message: error.publicMessage || '上游 Agent 回合处理失败', + code: error.code || 'upstream_error' + }) + return + } + if (!runtime.ok) { + writeOpenAIHttpError(res, runtime.error) + return + } + + setResponseHeaders(res, false) + const { attempt, finishReason } = runtime + const output = await prepareAgentOutput(attempt, enableThinking, enableWebSearch) + const assistantMessage = { + role: 'assistant', + content: output.content || (attempt.toolCalls.length > 0 ? null : '') + } + if (output.reasoning) assistantMessage.reasoning_content = output.reasoning + if (attempt.toolCalls.length > 0) { + assistantMessage.tool_calls = attempt.toolCalls.map(call => ({ + id: call.id, + type: 'function', + function: { ...call.function } + })) + } + const completionText = `${output.reasoning}${output.content}${JSON.stringify(attempt.toolCalls || [])}` + const usage = normalizeAgentUsage(attempt, requestBody, completionText) + attributeChatUsage(options.currentAccount, usage) + res.json({ + id: `chatcmpl-${generateUUID()}`, + object: 'chat.completion', + created: Math.round(Date.now() / 1000), + model, + choices: [{ index: 0, message: assistantMessage, finish_reason: finishReason }], + usage + }) +} + const handleStreamResponse = async (res, response, enable_thinking, enable_web_search, requestBody = null, options = {}) => { + if (options.has_tools && options.strict_agent_turn !== false) { + return handleOpenAIAgentStream( + res, + response, + enable_thinking, + enable_web_search, + requestBody, + options + ) + } try { const message_id = generateUUID() let web_search_info = null @@ -657,6 +975,17 @@ const handleStreamResponse = async (res, response, enable_thinking, enable_web_s * @param {boolean} [options.has_tools] - 是否启用工具调用解析 */ const handleNonStreamResponse = async (res, response, enable_thinking, enable_web_search, model, requestBody = null, options = {}) => { + if (options.has_tools && options.strict_agent_turn !== false) { + return handleOpenAIAgentNonStream( + res, + response, + enable_thinking, + enable_web_search, + model, + requestBody, + options + ) + } try { let fullContent = '' let fullReasoning = '' // 新版模式下累积的推理内容（reasoning_content） @@ -1010,7 +1339,12 @@ const handleChatCompletion = async (req, res) => { has_tools: req.has_tools, tool_choice: req.tool_choice, allowed_tool_names: req.allowed_tool_names, - currentAccount: response_data.currentAccount + currentAccount: response_data.currentAccount, + upstream_request_body: response_data.requestBody, + upstream_context: { + chatId: response_data.chatId, + parentId: response_data.parentId + } }) } else { setResponseHeaders(res, false) @@ -1018,7 +1352,12 @@ const handleChatCompletion = async (req, res) => { has_tools: req.has_tools, tool_choice: req.tool_choice, allowed_tool_names: req.allowed_tool_names, - currentAccount: response_data.currentAccount + currentAccount: response_data.currentAccount, + upstream_request_body: response_data.requestBody, + upstream_context: { + chatId: response_data.chatId, + parentId: response_data.parentId + } }) } diff --git a/src/middlewares/chat-middleware.js b/src/middlewares/chat-middleware.js index 0540a46..a3824fd 100644 --- a/src/middlewares/chat-middleware.js +++ b/src/middlewares/chat-middleware.js @@ -1,8 +1,42 @@ const { generateUUID } = require('../utils/tools.js') const { isChatType, isThinkingEnabled, parserModel, parserMessages } = require('../utils/chat-helpers.js') const { buildToolSystemPrompt, foldToolMessages } = require('../utils/tool-prompt.js') +const { buildAgentTurnDirective } = require('../utils/agent-turn.js') const { logger } = require('../utils/logger') +const shouldEnableToolRuntime = (tools, chatType, toolChoice) => ( + Array.isArray(tools) && + tools.length > 0 && + chatType === 't2t' && + toolChoice !== 'none' +) + +const AGENT_CURRENT_MESSAGE_MARKER = '# Current message' + +const ensureAgentCurrentEnvelope = (content, role = 'user') => { + const wrap = (text) => { + const value = String(text || '') + if (value.includes('# Conversation history (JSONL)') || value.includes(AGENT_CURRENT_MESSAGE_MARKER)) { + return value + } + return `${AGENT_CURRENT_MESSAGE_MARKER}\n${JSON.stringify({ role, content: value })}` + } + + if (typeof content === 'string') return wrap(content) + if (!Array.isArray(content)) return content + + let wrapped = false + const result = content.map(item => { + if (!wrapped && item?.type === 'text') { + wrapped = true + return { ...item, text: wrap(item.text) } + } + return item + }) + if (!wrapped) result.unshift({ type: 'text', text: wrap('') }) + return result +} + /** * 处理聊天请求体的中间件 * 解析和转换请求参数为内部格式 @@ -74,7 +108,11 @@ const processRequestBody = async (req, res, next) => { // 处理 tools 参数 : 通过提示词为网页版模型注入工具调用能力 const chatType = isChatType(model) - const hasTools = Array.isArray(tools) && tools.length > 0 && chatType === 't2t' + // OpenAI 允许请求同时携带 tools 和 tool_choice="none"。这种请求必须走普通 + // 文本完成路径，不能注入工具协议或启用严格 Agent 回合门禁。 + const hasTools = shouldEnableToolRuntime(tools, chatType, tool_choice) + const originalLastMessage = Array.isArray(messages) ? messages[messages.length - 1] : null + const afterToolResult = ['tool', 'function'].includes(String(originalLastMessage?.role || '').toLowerCase()) let preparedMessages = messages let toolSystemPrompt = '' if (hasTools) { @@ -104,6 +142,10 @@ const processRequestBody = async (req, res, next) => { // 工具提示词拼接到用户消息内容上 if (hasTools && toolSystemPrompt) { + body.messages[0].content = ensureAgentCurrentEnvelope( + body.messages[0].content, + lastMessage.role || 'user' + ) const msgContent = body.messages[0].content if (typeof msgContent === 'string') { body.messages[0].content = `${toolSystemPrompt}\n\n${msgContent}` @@ -115,6 +157,19 @@ const processRequestBody = async (req, res, next) => { msgContent.unshift({ type: 'text', text: toolSystemPrompt }) } } + + const turnDirective = buildAgentTurnDirective({ afterToolResult }) + const directedContent = body.messages[0].content + if (typeof directedContent === 'string') { + body.messages[0].content = `${directedContent}\n\n${turnDirective}` + } else if (Array.isArray(directedContent)) { + const textIdx = directedContent.findIndex(c => c?.type === 'text') + if (textIdx >= 0) { + directedContent[textIdx].text = `${directedContent[textIdx].text || ''}\n\n${turnDirective}` + } else { + directedContent.unshift({ type: 'text', text: turnDirective }) + } + } } // 保存完整消息历史供下游使用（用于多轮对话上下文） @@ -145,5 +200,7 @@ const processRequestBody = async (req, res, next) => { } module.exports = { - processRequestBody + processRequestBody, + shouldEnableToolRuntime, + ensureAgentCurrentEnvelope } diff --git a/src/utils/agent-turn.js b/src/utils/agent-turn.js new file mode 100644 index 0000000..12318b3 --- /dev/null +++ b/src/utils/agent-turn.js @@ -0,0 +1,235 @@ +const AGENT_FINAL_OPEN = '' +const AGENT_FINAL_CLOSE = '' +const AGENT_BLOCKED_OPEN = '' +const AGENT_BLOCKED_CLOSE = '' + +const escapeRegExp = (value) => String(value).replace(/[.*+?^${}()|[\]\\]/g, '\\$&') + +const unwrapExactTag = (value, openTag, closeTag) => { + const pattern = new RegExp( + `^\\s*${escapeRegExp(openTag)}([\\s\\S]*?)${escapeRegExp(closeTag)}\\s*$`, + 'i' + ) + const matched = String(value || '').match(pattern) + return matched ? matched[1].trim() : null +} + +/** + * Agent 请求的可见输出必须明确声明本回合是“已完成”还是“需要用户输入”。 + * 工具调用由 tool-prompt 解析器先行抽取，因此这里仅处理剩余文本。 + */ +const parseAgentControlText = (value) => { + const raw = String(value || '') + const trimmed = raw.trim() + if (!trimmed) return { kind: 'empty', text: '' } + + const finalText = unwrapExactTag(trimmed, AGENT_FINAL_OPEN, AGENT_FINAL_CLOSE) + if (finalText !== null) return { kind: 'final', text: finalText } + + const blockedText = unwrapExactTag(trimmed, AGENT_BLOCKED_OPEN, AGENT_BLOCKED_CLOSE) + if (blockedText !== null) return { kind: 'blocked', text: blockedText } + + if (/<\/?agent_(?:final|blocked)>/i.test(trimmed)) { + return { kind: 'invalid_control', text: trimmed } + } + return { kind: 'bare', text: trimmed } +} + +/** + * 增量识别严格 Agent 的 final/blocked 包装。只有开标签位于首个非空白位置时 + * 才开放正文；闭标签及其可能跨 chunk 的前缀始终留在缓冲区。 + * + * 正文采用与 parseAgentControlText 相同的 trim 语义：丢弃包装内的首尾空白， + * 中间空白仍按原顺序增量输出。完整合法性最终仍由 parseAgentControlText 判定。 + */ +const createAgentControlStreamParser = () => { + const modes = [ + { kind: 'final', open: AGENT_FINAL_OPEN, close: AGENT_FINAL_CLOSE }, + { kind: 'blocked', open: AGENT_BLOCKED_OPEN, close: AGENT_BLOCKED_CLOSE } + ] + let state = 'prefix' + let mode = null + let pending = '' + let bodyStarted = false + let trailingWhitespace = '' + let emittedText = false + let invalid = false + + const createResult = () => ({ + textDelta: '', + kind: mode?.kind || null, + opened: false, + closed: state === 'closed' && !invalid, + invalid + }) + + const appendBodyText = (value, final, result) => { + let text = String(value || '') + if (!bodyStarted) { + text = text.replace(/^\s+/, '') + if (!text) { + if (final) trailingWhitespace = '' + return + } + bodyStarted = true + } + + const combined = `${trailingWhitespace}${text}` + const trailing = combined.match(/\s+$/)?.[0] || '' + const safe = trailing ? combined.slice(0, -trailing.length) : combined + if (safe) { + result.textDelta += safe + emittedText = true + } + trailingWhitespace = final ? '' : trailing + } + + const splitClosePrefix = (value, closeTag) => { + const lower = value.toLowerCase() + const close = closeTag.toLowerCase() + const maxLength = Math.min(lower.length, close.length - 1) + for (let length = maxLength; length > 0; length--) { + if (close.startsWith(lower.slice(-length))) { + return { + safe: value.slice(0, -length), + remainder: value.slice(-length) + } + } + } + return { safe: value, remainder: '' } + } + + const processBody = (result) => { + const closeTag = mode.close + const closeIndex = pending.toLowerCase().indexOf(closeTag.toLowerCase()) + if (closeIndex !== -1) { + appendBodyText(pending.slice(0, closeIndex), true, result) + pending = pending.slice(closeIndex + closeTag.length) + state = 'closed' + result.closed = true + if (pending.trim()) { + invalid = true + state = 'invalid' + result.invalid = true + result.closed = false + } + return + } + + const split = splitClosePrefix(pending, closeTag) + pending = split.remainder + appendBodyText(split.safe, false, result) + } + + const processPrefix = (result) => { + const leadingLength = pending.match(/^\s*/)?.[0].length || 0 + const candidate = pending.slice(leadingLength) + if (!candidate) return + const lowerCandidate = candidate.toLowerCase() + const matchedMode = modes.find(item => lowerCandidate.startsWith(item.open.toLowerCase())) + if (matchedMode) { + mode = matchedMode + pending = candidate.slice(matchedMode.open.length) + state = 'body' + result.kind = mode.kind + result.opened = true + processBody(result) + return + } + + const isOpenPrefix = modes.some(item => item.open.toLowerCase().startsWith(lowerCandidate)) + if (!isOpenPrefix) { + invalid = true + state = 'invalid' + result.invalid = true + } + } + + const push = (chunk) => { + const result = createResult() + if (typeof chunk !== 'string' || chunk.length === 0 || state === 'invalid') return result + pending += chunk + if (state === 'prefix') processPrefix(result) + else if (state === 'body') processBody(result) + else if (state === 'closed' && pending.trim()) { + invalid = true + state = 'invalid' + result.invalid = true + result.closed = false + } + result.kind = mode?.kind || result.kind + return result + } + + const flush = () => { + const result = createResult() + if (state !== 'closed' || pending.trim()) { + invalid = true + state = 'invalid' + result.invalid = true + result.closed = false + } + result.kind = mode?.kind || null + return result + } + + return { + push, + flush, + getState: () => ({ + kind: mode?.kind || null, + opened: mode !== null, + closed: state === 'closed' && !invalid, + invalid, + hasEmittedText: emittedText + }) + } +} + +const buildAgentTurnDirective = ({ afterToolResult = false } = {}) => { + const continuation = afterToolResult + ? 'The current message is a tool result from the same unfinished task. It is evidence to inspect, not a new task and not a reason to stop after one action.' + : 'Treat this request as one step in an Agent task. Recover the original acceptance criteria from the conversation before deciding whether the task is complete.' + + return [ + '# Agent loop control (highest-priority output contract)', + continuation, + 'The client executes tools and automatically sends each tool result back in the next request. Keep that loop alive until the original task is genuinely complete.', + 'Before responding, check the original request, every claimed deliverable, failures in tool results, and whether verification is still missing.', + 'Your entire visible response MUST be exactly one of these modes:', + '1. If any action, inspection, edit, command, test, retry, or verification remains: emit one or more valid `...` blocks and no prose.', + `2. Only when every requested outcome is complete and supported by tool-result evidence: emit ${AGENT_FINAL_OPEN}a concise final report${AGENT_FINAL_CLOSE}.`, + `3. Only when progress is impossible without new user input or authority: emit ${AGENT_BLOCKED_OPEN}the exact blocker and required input${AGENT_BLOCKED_CLOSE}.`, + 'Bare prose, a plan, a progress update, hidden reasoning without visible output, or a claim such as “done” without the completion wrapper is an invalid Agent turn and will be regenerated.', + 'Never use the completion wrapper merely because one tool call finished. If verification has not run or any requested work remains, call the next tool.' + ].join('\n') +} + +const buildAgentRetryHint = (reason = 'incomplete') => { + const reasonText = { + empty: 'The previous attempt ended without a visible answer or executable tool call.', + bare: 'The previous attempt returned bare prose without declaring a verified final result or emitting the next tool call.', + invalid_control: 'The previous attempt used a malformed or mixed Agent completion wrapper.', + invalid_tool_call: 'The previous attempt contained an invalid, truncated, or unknown tool call.', + required_tool: 'The previous attempt violated tool_choice and did not call the required tool.' + }[reason] || 'The previous attempt did not produce a valid Agent turn.' + + return [ + '# Agent turn recovery', + reasonText, + 'Continue the SAME original task. Re-check its acceptance criteria and the latest tool result.', + `If work remains, output only valid \`...\` blocks. If and only if all work is verified complete, output ${AGENT_FINAL_OPEN}the final report${AGENT_FINAL_CLOSE}.`, + `If user input is strictly required, output ${AGENT_BLOCKED_OPEN}the blocker${AGENT_BLOCKED_CLOSE}. Do not output bare planning prose.` + ].join('\n') +} + +module.exports = { + AGENT_FINAL_OPEN, + AGENT_FINAL_CLOSE, + AGENT_BLOCKED_OPEN, + AGENT_BLOCKED_CLOSE, + parseAgentControlText, + createAgentControlStreamParser, + buildAgentTurnDirective, + buildAgentRetryHint +} diff --git a/src/utils/openai-agent-runtime.js b/src/utils/openai-agent-runtime.js new file mode 100644 index 0000000..113a3c3 --- /dev/null +++ b/src/utils/openai-agent-runtime.js @@ -0,0 +1,463 @@ +const { isJson } = require('./tools.js') +const { + parseToolCallsFromText, + createToolCallStreamParser, + createNativeToolCallAccumulator +} = require('./tool-prompt.js') +const { consumeSSEStream, createUpstreamResponseFilter } = require('./sse.js') +const { createUpstreamDeltaNormalizer } = require('./chat-helpers.js') +const { assertNoUpstreamFailure } = require('./upstream-error.js') +const { + parseAgentControlText, + createAgentControlStreamParser, + buildAgentRetryHint +} = require('./agent-turn.js') +const config = require('../config/index.js') +const { logger } = require('./logger.js') + +const NON_RETRYABLE_FINISH_REASONS = new Set([ + 'length', + 'max_tokens', + 'content_filter', + 'refusal' +]) + +const normalizeCreatedMetadata = (payload) => { + const created = payload?.['response.created'] || payload?.response?.created + if (!created || typeof created !== 'object') return null + return { + chatId: created.chat_id || created.chatId || null, + parentId: created.parent_id || created.parentId || null, + responseId: created.response_id || created.responseId || null, + responseIndex: created.response_index ?? created.responseIndex ?? null + } +} + +const imageMarkdownFromDelta = (delta) => { + const result = [] + for (const item of delta?.extra?.image_list || []) { + if (item?.image) result.push(`![image](${item.image})`) + } + return result +} + +/** + * 完整消费一次 Qwen 上游 attempt。裸正文与工具调用始终留在门禁内；调用方可 + * 实时接收安全思考，以及已经进入 final/blocked 包装体的正式正文增量。 + * 每次调用都新建 parser/filter/accumulator，失败 attempt 不会污染下一次。 + */ +const collectOpenAIAgentAttempt = async (upstreamResponse, options = {}) => { + const hasTools = options.has_tools !== false + const allowedToolNames = options.allowed_tool_names || [] + const normalizeDelta = createUpstreamDeltaNormalizer() + const acceptUpstreamFrame = createUpstreamResponseFilter() + const nativeTools = hasTools + ? createNativeToolCallAccumulator({ allowedToolNames }) + : null + const reasoningStreamParser = typeof options.on_reasoning_delta === 'function' + ? createToolCallStreamParser({ allowedToolNames }) + : null + const controlStreamParser = typeof options.on_content_delta === 'function' + ? createAgentControlStreamParser() + : null + const controlToolStreamParser = controlStreamParser + ? createToolCallStreamParser({ allowedToolNames }) + : null + let streamedVisibleText = '' + let streamedControlKind = null + let controlToolParserFlushed = false + + const emitReasoningDelta = async (text) => { + if (typeof options.on_reasoning_delta !== 'function') return + await options.on_reasoning_delta(text || '', { + attemptNumber: Math.max(1, Number(options.attempt_number) || 1) + }) + } + + const emitContentDelta = async (text, kind) => { + if (!text || typeof options.on_content_delta !== 'function') return + streamedVisibleText += text + streamedControlKind = kind || streamedControlKind + await options.on_content_delta(text, { + attemptNumber: Math.max(1, Number(options.attempt_number) || 1), + kind: streamedControlKind + }) + } + + const consumeControlStreamResult = async (result) => { + if (!result || !controlToolStreamParser) return + if (result.textDelta) { + const parsed = controlToolStreamParser.push(result.textDelta) + await emitContentDelta(parsed.textDelta, result.kind) + } + if (result.closed && !controlToolParserFlushed) { + controlToolParserFlushed = true + const parsed = controlToolStreamParser.flush() + await emitContentDelta(parsed.textDelta, result.kind) + } + } + + const appendAnswer = async (text) => { + if (!text) return + answer += text + if (controlStreamParser) { + await consumeControlStreamResult(controlStreamParser.push(text)) + } + } + + let reasoning = '' + let answer = '' + let answerStarted = false + let webSearchInfo = null + let upstreamFinishReason = null + let acceptedResponseId = null + const createdByResponseId = new Map() + let primaryCreated = null + let lastCreated = null + const emittedImages = new Set() + const pendingImages = [] + let totalTokens = { + prompt_tokens: 0, + completion_tokens: 0, + total_tokens: 0 + } + + const streamResult = await consumeSSEStream(upstreamResponse, async (frame) => { + if (!frame.data || frame.data.trim() === '[DONE]') return + const decoded = isJson(frame.data) ? JSON.parse(frame.data) : null + if (decoded === null) return + assertNoUpstreamFailure(decoded) + + const created = normalizeCreatedMetadata(decoded) + if (created) { + lastCreated = created + if (created.responseId) createdByResponseId.set(created.responseId, created) + const responseIndex = created.responseIndex === null || created.responseIndex === '' + ? Number.NaN + : Number(created.responseIndex) + if (created.responseId && Number.isFinite(responseIndex) && responseIndex === 0) { + primaryCreated = created + acceptedResponseId = created.responseId + } + } + + if (!acceptUpstreamFrame(decoded)) return + if (decoded.response_id) acceptedResponseId = decoded.response_id + + if (decoded.usage) { + totalTokens = { + prompt_tokens: decoded.usage.prompt_tokens || totalTokens.prompt_tokens, + completion_tokens: decoded.usage.completion_tokens || totalTokens.completion_tokens, + total_tokens: decoded.usage.total_tokens || totalTokens.total_tokens + } + } + if (!Array.isArray(decoded.choices) || decoded.choices.length === 0) return + + const choice = decoded.choices[0] + const reportedFinishReason = choice.finish_reason ?? choice.delta?.finish_reason + if (reportedFinishReason !== undefined && reportedFinishReason !== null) { + upstreamFinishReason = reportedFinishReason + } + + const delta = choice.delta || {} + if (nativeTools && Array.isArray(delta.tool_calls)) { + nativeTools.push(delta.tool_calls) + } else if (nativeTools && delta.function_call) { + nativeTools.push([{ + index: 0, + type: 'function', + function: delta.function_call + }]) + } + + if (delta.name === 'web_search') { + webSearchInfo = delta.extra?.web_search_info || webSearchInfo + } + + const normalized = normalizeDelta(delta) + const phase = normalized?.phase || null + const images = imageMarkdownFromDelta(delta) + .filter(item => !emittedImages.has(item) && !pendingImages.includes(item)) + if (images.length > 0) { + if (phase === 'think' && !answerStarted) { + pendingImages.push(...images) + } else { + const markdown = `${images.join('\n\n')}\n\n` + await appendAnswer(markdown) + images.forEach(item => emittedImages.add(item)) + } + } + + if (!normalized) return + if (normalized.phase === 'think') { + reasoning += normalized.content + if (reasoningStreamParser) { + const streamed = reasoningStreamParser.push(normalized.content) + await emitReasoningDelta(streamed.textDelta) + } + return + } + + answerStarted = true + if (pendingImages.length > 0) { + await appendAnswer(`${pendingImages.join('\n\n')}\n\n`) + pendingImages.forEach(item => emittedImages.add(item)) + pendingImages.length = 0 + } + await appendAnswer(normalized.content) + }) + + if (reasoningStreamParser) { + const streamed = reasoningStreamParser.flush() + await emitReasoningDelta(streamed.textDelta) + } + if (controlStreamParser) { + await consumeControlStreamResult(controlStreamParser.flush()) + } + + const textTools = hasTools + ? parseToolCallsFromText(answer, { allowedToolNames }) + : { cleanedText: answer, toolCalls: [], errors: [] } + const reasoningTools = hasTools && textTools.toolCalls.length === 0 && !textTools.cleanedText.trim() + ? parseToolCallsFromText(reasoning, { allowedToolNames }) + : { cleanedText: reasoning, toolCalls: [], errors: [] } + const nativeToolCalls = nativeTools?.hasAny() ? nativeTools.finalize() : [] + // 部分 thinking 模型会把“整个可执行工具块”放进 think phase 后直接 EOF。 + // 仅当 thinking 除独立工具块外没有任何文字时才接纳，避免把推理中的示例或 + // 尚未决定执行的调用当成真实动作。 + const standaloneReasoningCalls = reasoningTools.toolCalls.length > 0 && + !reasoningTools.cleanedText.trim() && + reasoningTools.errors.length === 0 + ? reasoningTools.toolCalls + : [] + const toolCalls = nativeToolCalls.length > 0 + ? nativeToolCalls + : (textTools.toolCalls.length > 0 ? textTools.toolCalls : standaloneReasoningCalls) + const toolErrors = [ + ...(textTools.errors || []), + ...(textTools.toolCalls.length === 0 && !textTools.cleanedText.trim() + ? (reasoningTools.errors || []) + : []), + ...(nativeTools?.getErrors?.() || []) + ] + const control = parseAgentControlText(textTools.cleanedText) + const metadata = (acceptedResponseId && createdByResponseId.get(acceptedResponseId)) || primaryCreated || lastCreated || { + chatId: null, + parentId: null, + responseId: acceptedResponseId + } + + return { + reasoning: standaloneReasoningCalls.length > 0 ? reasoningTools.cleanedText : reasoning, + rawAnswer: answer, + visibleText: control.text, + controlKind: control.kind, + streamedVisibleText, + streamedControlKind, + streamedControlState: controlStreamParser?.getState?.() || null, + toolCalls, + toolErrors, + webSearchInfo, + totalTokens, + upstreamFinishReason, + upstreamCompleted: streamResult.completed, + upstreamEventCount: streamResult.eventCount, + sawDone: streamResult.sawDone, + metadata: { + ...metadata, + responseId: metadata?.responseId || acceptedResponseId || null + } + } +} + +const requiresToolCall = (toolChoice) => { + if (toolChoice === 'required') return true + return !!(toolChoice && typeof toolChoice === 'object' && toolChoice.type === 'function' && toolChoice.function?.name) +} + +const evaluateOpenAIAgentAttempt = (attempt, options = {}) => { + const finishReason = attempt.upstreamFinishReason + if (NON_RETRYABLE_FINISH_REASONS.has(finishReason)) { + const normalized = finishReason === 'max_tokens' ? 'length' : finishReason + return { accepted: true, finishReason: normalized, retryReason: null } + } + if (attempt.toolErrors.length > 0) { + return { accepted: false, finishReason: null, retryReason: 'invalid_tool_call' } + } + if (attempt.toolCalls.length > 0) { + if (attempt.controlKind !== 'empty' || attempt.visibleText.trim()) { + return { accepted: false, finishReason: null, retryReason: 'invalid_tool_call' } + } + return { accepted: true, finishReason: 'tool_calls', retryReason: null } + } + if (requiresToolCall(options.tool_choice)) { + return { accepted: false, finishReason: null, retryReason: 'required_tool' } + } + if (attempt.controlKind === 'final' || attempt.controlKind === 'blocked') { + if (attempt.visibleText.trim()) { + return { accepted: true, finishReason: 'stop', retryReason: null } + } + return { accepted: false, finishReason: null, retryReason: 'empty' } + } + if (attempt.controlKind === 'empty') { + return { accepted: false, finishReason: null, retryReason: 'empty' } + } + if (attempt.controlKind === 'invalid_control') { + return { accepted: false, finishReason: null, retryReason: 'invalid_control' } + } + return { accepted: false, finishReason: null, retryReason: 'bare' } +} + +const appendRetryHint = (requestBody, hint) => { + const clone = requestBody && typeof requestBody === 'object' + ? JSON.parse(JSON.stringify(requestBody)) + : {} + const messages = Array.isArray(clone.messages) ? clone.messages : [] + if (messages.length === 0) { + messages.push({ role: 'user', content: hint }) + } else { + const last = messages[messages.length - 1] + if (typeof last.content === 'string') { + last.content = `${last.content}\n\n${hint}` + } else if (Array.isArray(last.content)) { + const textPart = last.content.find(part => part?.type === 'text') + if (textPart) textPart.text = `${textPart.text || ''}\n\n${hint}` + else last.content.unshift({ type: 'text', text: hint }) + } else { + last.content = hint + } + } + clone.messages = messages + return clone +} + +const exhaustedError = (attempt, retryReason) => { + if (retryReason === 'empty' && !String(attempt?.reasoning || '').trim()) { + return { + status: 503, + message: '上游连续返回空 Agent 回合，任务状态未被标记为完成', + code: 'upstream_unavailable' + } + } + const messages = { + empty: '上游连续只返回思考内容，没有给出可执行工具调用或最终答复', + bare: '上游连续返回未声明完成状态的文本，已阻止 Agent 将未完成任务误判为结束', + invalid_control: '上游连续返回无效的 Agent 完成标记', + invalid_tool_call: '上游连续返回残缺、非法或不存在的工具调用', + required_tool: '上游连续违反 tool_choice，未返回要求的工具调用' + } + return { + status: 429, + message: messages[retryReason] || '上游未能生成有效的 Agent 回合', + code: retryReason === 'invalid_tool_call' ? 'invalid_tool_call' : 'upstream_agent_turn_incomplete' + } +} + +/** + * 执行严格 Agent 回合：每个 attempt 完全隔离；只有有效工具调用、显式完成/阻塞， + * 或标准非重试终止原因才能提交给客户端。 + */ +const runOpenAIAgentTurn = async (initialResponse, options = {}) => { + const requestSender = options.sendChatRequest + const maxAttempts = Math.min( + 6, + Math.max(2, Number(options.agent_turn_max_attempts) || config.agentTurnMaxAttempts) + ) + let currentResponse = initialResponse + let lastAttempt = null + let lastEvaluation = null + let upstreamContext = { ...(options.upstream_context || {}) } + const retryBaseBody = options.upstream_request_body || options.requestBody + let attemptsMade = 0 + + const mergePresent = (base, extra) => { + const merged = { ...base } + for (const [key, value] of Object.entries(extra || {})) { + if (value !== null && value !== undefined && value !== '') merged[key] = value + } + return merged + } + + for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber++) { + attemptsMade = attemptNumber + const attempt = await collectOpenAIAgentAttempt(currentResponse, { + ...options, + attempt_number: attemptNumber + }) + const evaluation = evaluateOpenAIAgentAttempt(attempt, options) + lastAttempt = attempt + lastEvaluation = evaluation + upstreamContext = mergePresent(upstreamContext, attempt.metadata) + + if (evaluation.accepted) { + return { + ok: true, + attempt, + finishReason: evaluation.finishReason, + attempts: attemptNumber + } + } + + logger.warn( + `Agent attempt ${attemptNumber}/${maxAttempts} 被回合门禁拒绝 (${evaluation.retryReason})`, + 'AGENT' + ) + if (attempt.streamedVisibleText) { + return { + ok: false, + error: { + status: 422, + message: '上游在已开始流式输出正式回复后返回了无效的 Agent 结束结构', + code: 'upstream_agent_stream_invalidated' + }, + attempt, + attempts: attemptNumber + } + } + if (attemptNumber >= maxAttempts || typeof requestSender !== 'function') break + + const retryBody = appendRetryHint( + retryBaseBody, + buildAgentRetryHint(evaluation.retryReason) + ) + const retryResponse = await requestSender(retryBody, { + chatId: upstreamContext.chatId || null, + parentId: upstreamContext.responseId || null, + currentAccount: options.currentAccount || null, + agentRetry: true + }) + if (!retryResponse?.status || !retryResponse.response) { + return { + ok: false, + error: { + status: 502, + message: retryResponse?.message || 'Agent 回合纠正请求失败', + code: 'upstream_retry_failed' + }, + attempt, + attempts: attemptNumber + } + } + currentResponse = retryResponse.response + upstreamContext = mergePresent(upstreamContext, { + chatId: retryResponse.chatId, + currentAccount: retryResponse.currentAccount + }) + } + + return { + ok: false, + error: exhaustedError(lastAttempt, lastEvaluation?.retryReason), + attempt: lastAttempt, + attempts: attemptsMade + } +} + +module.exports = { + NON_RETRYABLE_FINISH_REASONS, + normalizeCreatedMetadata, + collectOpenAIAgentAttempt, + evaluateOpenAIAgentAttempt, + appendRetryHint, + runOpenAIAgentTurn +} diff --git a/src/utils/request.js b/src/utils/request.js index 0d6e401..92d37b1 100644 --- a/src/utils/request.js +++ b/src/utils/request.js @@ -41,6 +41,180 @@ const truncateUtf8 = (value, maxBytes, fromEnd = false) => { return slice.toString('utf8').replace(/^\uFFFD|\uFFFD$/g, '') } +const truncateUtf8HeadTail = ( + value, + maxBytes, + headRatio = 0.55, + separator = '\n...[inline context compacted; complete copy is in the attachment]...\n' +) => { + const text = String(value || '') + const buffer = Buffer.from(text, 'utf8') + const limit = Math.max(0, Number(maxBytes) || 0) + if (buffer.length <= limit) return text + if (limit === 0) return '' + + const separatorBytes = byteLength(separator) + if (limit <= separatorBytes + 2) return truncateUtf8(text, limit) + + const contentBudget = limit - separatorBytes + const headBytes = Math.max(1, Math.floor(contentBudget * headRatio)) + const tailBytes = Math.max(1, contentBudget - headBytes) + return `${truncateUtf8(text, headBytes)}${separator}${truncateUtf8(text, tailBytes, true)}` +} + +const parseAgentEnvelope = (value) => { + const text = String(value || '') + const historyIndex = text.indexOf(HISTORY_MARKER) + const currentIndex = text.lastIndexOf(CURRENT_MESSAGE_MARKER) + if (historyIndex < 0) { + if (currentIndex >= 0) { + return { + prefix: text.slice(0, currentIndex).trim(), + history: '', + current: text.slice(currentIndex).trim(), + entries: [] + } + } + return { prefix: text, history: '', current: '', entries: [] } + } + + const historyStart = historyIndex + HISTORY_MARKER.length + const hasCurrent = currentIndex > historyStart + const history = text.slice(historyStart, hasCurrent ? currentIndex : text.length).trim() + const entries = [] + for (const line of history.split('\n')) { + const raw = line.trim() + if (!raw) continue + try { + const parsed = JSON.parse(raw) + if (parsed && typeof parsed === 'object') { + entries.push({ + raw, + role: String(parsed.role || '').toLowerCase(), + content: typeof parsed.content === 'string' + ? parsed.content + : JSON.stringify(parsed.content ?? '') + }) + } + } catch (_) { + // 历史中可能包含旧版非 JSONL 内容；recent history 仍会按原文保留。 + } + } + return { + prefix: text.slice(0, historyIndex).trim(), + history, + current: hasCurrent ? text.slice(currentIndex).trim() : '', + entries + } +} + +const buildEssentialAgentHistory = (entries) => { + if (!Array.isArray(entries) || entries.length === 0) return '' + const systemEntries = entries.filter(entry => ['system', 'developer'].includes(entry.role)) + const activeTask = [...entries].reverse().find(entry => + entry.role === 'user' && + !/^\s* 0) { + sections.push('## System/developer instructions', systemEntries.map(entry => entry.raw).join('\n')) + } + if (activeTask) { + sections.push('## Active user task', activeTask.raw) + } + return sections.join('\n') +} + +const buildRecentAgentHistory = ( + envelope, + maxBytes, + compactionSeparator = '\n...[inline context compacted; complete copy is in the attachment]...\n' +) => { + const limit = Math.max(0, Number(maxBytes) || 0) + if (limit === 0) return '' + const entries = envelope?.entries || [] + if (entries.length === 0) { + return truncateUtf8HeadTail(envelope?.history || '', limit, 0.35, compactionSeparator) + } + + const latest = entries[entries.length - 1].raw + const previous = entries.slice(Math.max(0, entries.length - 4), -1) + .map(entry => entry.raw) + .join('\n') + if (!previous) return truncateUtf8HeadTail(latest, limit, 0.4, compactionSeparator) + + const previousBudget = Math.floor(limit * 0.32) + const latestBudget = Math.max(0, limit - previousBudget - 1) + return [ + truncateUtf8HeadTail(previous, previousBudget, 0.45, compactionSeparator), + truncateUtf8HeadTail(latest, latestBudget, 0.4, compactionSeparator) + ].filter(Boolean).join('\n') +} + +const buildBudgetedAgentPrompt = ( + original, + maxBytes, + notice, + { attachmentAvailable = true } = {} +) => { + const envelope = parseAgentEnvelope(original) + const essential = buildEssentialAgentHistory(envelope.entries) + const sections = [ + { header: '', value: envelope.prefix, weight: 34, headRatio: 0.55, kind: 'text' }, + { + header: '# Essential Agent state retained inline', + value: essential, + weight: 24, + headRatio: 0.65, + kind: 'text' + }, + { + header: '# Recent Agent history retained inline', + value: envelope.history, + weight: 18, + headRatio: 0.4, + kind: 'recent' + }, + { header: '', value: envelope.current, weight: 24, headRatio: 0.45, kind: 'text' } + ].filter(section => String(section.value || '').trim()) + + const max = Math.max(1024, Number(maxBytes) || config.agentContextLivePromptBytes) + const joinSeparator = '\n\n' + const fixedBytes = byteLength(notice) + + (sections.length * byteLength(joinSeparator)) + + sections.reduce((sum, section) => ( + sum + (section.header ? byteLength(`${section.header}\n`) : 0) + ), 0) + if (fixedBytes >= max) return truncateUtf8(notice, max) + + let remainingBytes = max - fixedBytes + let remainingWeight = sections.reduce((sum, section) => sum + section.weight, 0) + const compactionSeparator = attachmentAvailable + ? '\n...[inline context compacted; complete copy is in the attachment]...\n' + : '\n...[older inline context compacted after attachment recovery failed]...\n' + const rendered = sections.map((section, index) => { + const isLast = index === sections.length - 1 + const budget = isLast + ? remainingBytes + : Math.floor(remainingBytes * section.weight / remainingWeight) + remainingBytes -= budget + remainingWeight -= section.weight + const content = section.kind === 'recent' + ? buildRecentAgentHistory(envelope, budget, compactionSeparator) + : truncateUtf8HeadTail( + section.value, + budget, + section.headRatio, + compactionSeparator + ) + return section.header ? `${section.header}\n${content}` : content + }) + + return [notice, ...rendered].join(joinSeparator) +} + const getMessageTextContent = (message) => { if (typeof message?.content === 'string') return message.content if (!Array.isArray(message?.content)) return null @@ -83,52 +257,24 @@ const buildAgentContextLivePrompt = ( maxBytes = config.agentContextLivePromptBytes, attachmentName = 'QWEN2API_AGENT_CONTEXT.txt' ) => { - const text = String(original || '') - const historyIndex = text.indexOf(HISTORY_MARKER) - const currentIndex = text.lastIndexOf(CURRENT_MESSAGE_MARKER) - const prefix = historyIndex >= 0 ? text.slice(0, historyIndex).trim() : '' - const current = currentIndex >= 0 ? text.slice(currentIndex).trim() : '' const notice = [ '# Agent context attachment', `The complete system instructions, tool schemas, conversation history and current task are attached as ${attachmentName}.`, - 'Read that attachment as authoritative context before acting. Continue from the latest state; do not restart the task or claim completion without verification.', + 'Read that attachment as authoritative context before acting. The essential task state and recent tool progress are also retained inline below so the Agent loop must not reset if attachment parsing is delayed.', + 'Continue from the latest state; do not restart the task, stop after one intermediate action, or claim completion without tool-result verification.', 'When an available tool is needed, emit the real `` block immediately. Do not replace it with prose such as “I will run...” or “done”.' ].join('\n') - - let live = [notice, prefix, current].filter(Boolean).join('\n\n') - if (byteLength(live) <= maxBytes) return live - - const reserved = byteLength(notice) + 4 - const remaining = Math.max(1024, maxBytes - reserved) - const currentBudget = Math.max(1024, Math.floor(remaining * 0.45)) - const prefixBudget = Math.max(1024, remaining - currentBudget) - live = [ - notice, - prefix ? truncateUtf8(prefix, prefixBudget) : '', - current ? truncateUtf8(current, currentBudget, true) : '' - ].filter(Boolean).join('\n\n') - return live + return buildBudgetedAgentPrompt(original, maxBytes, notice, { attachmentAvailable: true }) } const compactAgentContextFallback = (original, maxBytes = config.agentContextLivePromptBytes) => { - const text = String(original || '') const notice = [ '# Agent context recovery', 'The upstream document attachment failed, so older context was compacted to stay below the Qwen Web request limit.', - 'Continue the latest Agent task using the recent context below. Use a real tool call whenever more work is required; do not report completion before verification.' + 'The system/developer rules, active user task, recent tool progress, and current tool result retained below remain authoritative.', + 'Continue the latest Agent task using that state. Use a real tool call whenever more work is required; do not report completion before verification.' ].join('\n') - const limit = Math.max(1024, Number(maxBytes) || config.agentContextLivePromptBytes) - const historyIndex = text.indexOf(HISTORY_MARKER) - const prefix = historyIndex >= 0 ? text.slice(0, historyIndex).trim() : '' - const recent = historyIndex >= 0 ? text.slice(historyIndex).trim() : text - const remaining = Math.max(256, limit - byteLength(notice) - 4) - const prefixBudget = prefix ? Math.floor(remaining * 0.55) : 0 - const recentBudget = remaining - prefixBudget - return [ - notice, - prefix ? truncateUtf8(prefix, prefixBudget) : '', - truncateUtf8(recent, recentBudget, true) - ].filter(Boolean).join('\n\n') + return buildBudgetedAgentPrompt(original, maxBytes, notice, { attachmentAvailable: false }) } /** @@ -186,9 +332,11 @@ const externalizeOversizedAgentContext = async ( * @param {Object} body - 请求体 * @returns {Promise} 响应结果 */ -const sendChatRequest = async (body) => { +const sendChatRequest = async (body, options = {}) => { // 获取可用的账户（包含 proxy 等完整字段） - const currentAccount = accountManager.getAccount() + const currentAccount = options.currentAccount?.token + ? options.currentAccount + : accountManager.getAccount() const currentToken = currentAccount ? currentAccount.token : null if (!currentToken) { @@ -244,18 +392,34 @@ const sendChatRequest = async (body) => { } const chatType = body.chat_type || body.messages?.[0]?.chat_type || 't2t' - const chat_id = await generateChatID(currentToken, body.model, currentAccount, chatType) + const chat_id = options.chatId || await generateChatID(currentToken, body.model, currentAccount, chatType) + if (!chat_id) { + return { + status: false, + response: null, + message: '无法创建或续接 Qwen 会话' + } + } // 浏览器 referer 为 /c/（在 chat_id 生成后动态设置） requestConfig.headers.referer = `${chatBaseUrl}/c/${chat_id}` const url = `${chatBaseUrl}/api/v2/chat/completions?chat_id=` + chat_id // 对齐网页双写 chatId/parentId（FE 0.2.81） + const parentId = options.parentId ?? body.parentId ?? body.parent_id ?? null + const messages = Array.isArray(body.messages) + ? body.messages.map(message => ({ + ...message, + parent_id: parentId, + parentId + })) + : body.messages const rawPayload = { ...body, stream: true, chat_id, chatId: chat_id, - parent_id: body.parent_id ?? null, - parentId: body.parentId ?? body.parent_id ?? null + parent_id: parentId, + parentId, + messages } const contextResult = await externalizeOversizedAgentContext( rawPayload, @@ -287,6 +451,11 @@ const sendChatRequest = async (body) => { return { currentToken, currentAccount, + chatId: chat_id, + parentId, + // 返回真正提交给 Qwen 的请求体。严格 Agent 回合纠正可直接复用 + // 已外置的上下文附件，避免每次纠正都重新上传同一份长历史。 + requestBody: payload, status: true, response: response.data } diff --git a/src/utils/sse.js b/src/utils/sse.js index 4726c70..0573c2a 100644 --- a/src/utils/sse.js +++ b/src/utils/sse.js @@ -167,19 +167,48 @@ const consumeSSEStream = async (stream, onFrame) => { * 上游偶尔会对同一次请求开启多路候选回答：先连续下发多个 response.created * （chat_id/parent_id 相同，response_index 为 "0"/"1"，各自 response_id 不同）， * 随后两路增量帧交错到达。解析时不区分 response_id 就会把两路内容拼到一起， - * 回答被复读（"巴黎巴黎"）。这里锁定第一个带 response_id 的帧，丢弃其余各路。 - * 上游未带 response_id 时（旧协议）一律放行。 + * 回答被复读（"巴黎巴黎"）。新版协议优先锁定 response_index=0 的官方主回答； + * 只有旧协议缺少 response.created 元数据时，才退回锁定第一路正文。 + * 上游完全不带 response_id 时一律放行。 * @returns {(json: object) => boolean} true 表示该帧属于已锁定的那一路，应继续处理 */ const createUpstreamResponseFilter = () => { let acceptedId = null + let sawCreatedEvent = false + const responseIndexes = new Map() return (json) => { + const created = json?.['response.created'] || json?.response?.created + if (created && typeof created === 'object') { + sawCreatedEvent = true + const responseId = created.response_id || created.responseId || null + const responseIndex = Number(created.response_index ?? created.responseIndex) + if (responseId && Number.isFinite(responseIndex)) { + responseIndexes.set(responseId, responseIndex) + } + // 网页端定义 response_index=0 为主回答。不能按“谁先吐 delta”抢锁， + // 否则可能把备用候选当成主回答，并把错误 response_id 用作下一轮 parentId。 + if (responseId && responseIndex === 0) { + acceptedId = responseId + } else if (responseId && acceptedId === null && !Number.isFinite(responseIndex)) { + acceptedId = responseId + } + return !responseId || acceptedId === null || responseId === acceptedId + } + const responseId = json && json.response_id if (!responseId) return true - if (acceptedId === null) { + if (acceptedId !== null) return responseId === acceptedId + + const responseIndex = responseIndexes.get(responseId) + if (responseIndex === 0) { acceptedId = responseId + return true } - return responseId === acceptedId + if (Number.isFinite(responseIndex) || sawCreatedEvent) return false + + // 旧协议没有 response.created，只能锁定第一路带 response_id 的正文。 + acceptedId = responseId + return true } } diff --git a/src/utils/tool-prompt.js b/src/utils/tool-prompt.js index 7ec3f67..9238237 100644 --- a/src/utils/tool-prompt.js +++ b/src/utils/tool-prompt.js @@ -1,5 +1,11 @@ const { generateUUID } = require('./tools.js'); const { logger } = require('./logger'); +const { + AGENT_FINAL_OPEN, + AGENT_FINAL_CLOSE, + AGENT_BLOCKED_OPEN, + AGENT_BLOCKED_CLOSE +} = require('./agent-turn.js'); /** * 工具调用 XML 起始标签 @@ -164,7 +170,11 @@ const buildToolSystemPrompt = (tools, options = {}) => { '- You may emit multiple `` blocks back-to-back when more than one tool is needed.', '- After every tool result, evaluate the actual task state. If work remains, emit the next tool call. Only return a normal-language final answer after the requested task is genuinely complete or you are blocked on user input.', '- Never claim that a file was changed, a command succeeded, or a result was verified unless the corresponding tool result proves it.', - '- Do not call nonexistent tools, fabricate tool results, wrap `` in code fences, or mix extra commentary into a tool-call turn.' + '- Do not call nonexistent tools, fabricate tool results, wrap `` in code fences, or mix extra commentary into a tool-call turn.', + '- A non-tool response is valid only when it explicitly declares its state: use the completion or blocked wrapper below. Bare prose is invalid.', + `- Verified completion: ${AGENT_FINAL_OPEN}final report${AGENT_FINAL_CLOSE}`, + `- Requires user input/authority: ${AGENT_BLOCKED_OPEN}exact blocker${AGENT_BLOCKED_CLOSE}`, + '- Never emit the completion wrapper after merely finishing one intermediate tool action; continue with another tool call until every requested outcome is verified.' ]; const choice = options.tool_choice; @@ -194,9 +204,15 @@ const foldToolMessages = (messages) => { return messages.map((message) => { if (!message || typeof message !== 'object') return message; - if (message.role === 'assistant' && Array.isArray(message.tool_calls) && message.tool_calls.length > 0) { - const blocks = message.tool_calls.map((call) => { - let args = call?.function?.arguments; + const assistantCalls = message.role === 'assistant' + ? (Array.isArray(message.tool_calls) && message.tool_calls.length > 0 + ? message.tool_calls + : (message.function_call?.name ? [message.function_call] : [])) + : []; + if (assistantCalls.length > 0) { + const blocks = assistantCalls.map((call) => { + const fn = call?.function || call; + let args = fn?.arguments; if (typeof args === 'string') { try { args = JSON.parse(args); @@ -204,7 +220,7 @@ const foldToolMessages = (messages) => { // 保留原始字符串形式 } } - const name = call?.function?.name || 'unknown'; + const name = fn?.name || 'unknown'; const id = call?.id || `call_${generateUUID().replace(/-/g, '').slice(0, 24)}`; callIdToName.set(id, name); const payload = { id, name, arguments: args ?? {} }; @@ -217,9 +233,9 @@ const foldToolMessages = (messages) => { }; } - if (message.role === 'tool') { + if (message.role === 'tool' || message.role === 'function') { const callId = message.tool_call_id || ''; - const name = message.name || callIdToName.get(callId) || 'tool'; + const name = message.name || callIdToName.get(callId) || (message.role === 'function' ? 'function' : 'tool'); const content = typeof message.content === 'string' ? (message.content || 'null') : JSON.stringify(message.content ?? null); diff --git a/tests/agent-protocol.test.js b/tests/agent-protocol.test.js index d2d5140..b0da7ff 100644 --- a/tests/agent-protocol.test.js +++ b/tests/agent-protocol.test.js @@ -15,8 +15,20 @@ const { handleAnthropicStream, handleAnthropicNonStream } = require('../src/controllers/anthropic.js') -const { externalizeOversizedAgentContext } = require('../src/utils/request.js') +const { + externalizeOversizedAgentContext, + compactAgentContextFallback +} = require('../src/utils/request.js') const { assertNoUpstreamFailure } = require('../src/utils/upstream-error.js') +const { + shouldEnableToolRuntime, + ensureAgentCurrentEnvelope +} = require('../src/middlewares/chat-middleware.js') +const { + buildAgentTurnDirective, + parseAgentControlText, + createAgentControlStreamParser +} = require('../src/utils/agent-turn.js') test.after(() => { require('../src/utils/account.js').destroy() @@ -196,6 +208,434 @@ test('thinking-only Agent turns retry once and recover visible output', async () assert.match(anthropicRes.output, /event: message_stop/) }) +test('strict OpenAI Agent gate streams reasoning but keeps rejected answer attempts isolated', async () => { + const retryOptions = [] + let retries = 0 + const res = createMockResponse() + const retryFrames = [ + [ + 'data: {"response.created":{"chat_id":"chat_agent","parent_id":"p1","response_id":"resp_2","response_index":"0"}}\n\n', + 'data: {"choices":[{"delta":{"phase":"answer","content":"I will inspect the repository now."},"finish_reason":"stop"}],"response_id":"resp_2"}\n\n' + ], + [ + 'data: {"response.created":{"chat_id":"chat_agent","parent_id":"resp_2","response_id":"resp_3","response_index":"0"}}\n\n', + 'data: {"choices":[{"delta":{"phase":"answer","content":"{\\"name\\":\\"read_file\\",\\"arguments\\":{\\"path\\":\\"README.md\\"}}"},"finish_reason":"stop"}],"response_id":"resp_3"}\n\n' + ] + ] + + await handleStreamResponse( + res, + Readable.from([ + 'data: {"response.created":{"chat_id":"chat_agent","parent_id":"root","response_id":"resp_1","response_index":"0"}}\n\n', + 'data: {"choices":[{"delta":{"phase":"think","content":"first attempt planning"},"finish_reason":"stop"}],"response_id":"resp_1"}\n\n' + ]), + true, + false, + { messages: [{ role: 'user', content: 'finish the whole task' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'], + agent_turn_max_attempts: 3, + sendChatRequest: async (_body, options) => { + retryOptions.push(options) + const frames = retryFrames[retries] + retries += 1 + return { status: true, response: Readable.from(frames), chatId: 'chat_agent' } + } + } + ) + + assert.equal(retries, 2) + assert.equal(retryOptions[0].chatId, 'chat_agent') + assert.equal(retryOptions[0].parentId, 'resp_1') + assert.equal(retryOptions[1].parentId, 'resp_2') + assert.match(res.output, /"name":"read_file"/) + assert.match(res.output, /"finish_reason":"tool_calls"/) + assert.match(res.output, /"reasoning_content":"first attempt planning"/) + assert.doesNotMatch(res.output, /I will inspect/) +}) + +test('strict OpenAI Agent gate accepts only explicit verified completion and strips control tags', async () => { + let retries = 0 + const res = createMockResponse() + await handleStreamResponse( + res, + Readable.from([ + 'data: {"choices":[{"delta":{"phase":"answer","content":"All requested changes were implemented and tests passed."},"finish_reason":"stop"}]}\n\n' + ]), + false, + false, + { messages: [{ role: 'user', content: 'finish the task' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'], + sendChatRequest: async () => { retries += 1 } + } + ) + + assert.equal(retries, 0) + assert.match(res.output, /All requested changes were implemented and tests passed/) + assert.match(res.output, /"finish_reason":"stop"/) + assert.doesNotMatch(res.output, /agent_final/) +}) + +test('strict OpenAI Agent stream returns an in-band terminal error instead of a normal stop', async () => { + let retries = 0 + const res = createMockResponse() + await handleStreamResponse( + res, + Readable.from(['data: {"choices":[{"delta":{"phase":"answer","content":"Looks good."},"finish_reason":"stop"}]}\n\n']), + false, + false, + { messages: [{ role: 'user', content: 'complete and verify the task' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['run_tests'], + agent_turn_max_attempts: 3, + sendChatRequest: async () => { + retries += 1 + return { + status: true, + response: Readable.from(['data: {"choices":[{"delta":{"phase":"answer","content":"Still looks good."},"finish_reason":"stop"}]}\n\n']) + } + } + } + ) + + assert.equal(retries, 2) + assert.equal(res.statusCode, 200) + assert.equal(res.headers['Content-Type'], 'text/event-stream') + assert.match(res.output, /upstream_agent_turn_incomplete/) + assert.match(res.output, /\[DONE\]/) + assert.doesNotMatch(res.output, /"finish_reason":"stop"/) +}) + +test('strict OpenAI Agent stream exposes thinking incrementally before the gated turn completes', async () => { + let releaseUpstream + let firstFrameYielded + const firstFrameWasYielded = new Promise(resolve => { firstFrameYielded = resolve }) + const upstream = Readable.from((async function * () { + yield 'data: {"choices":[{"delta":{"phase":"think","content":"live thought"},"finish_reason":null}]}\n\n' + firstFrameYielded() + await new Promise(resolve => { releaseUpstream = resolve }) + yield 'data: {"choices":[{"delta":{"phase":"answer","content":"{\\"name\\":\\"read_file\\",\\"arguments\\":{\\"path\\":\\"README.md\\"}}"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n' + })()) + const res = createMockResponse() + + const responsePromise = handleStreamResponse( + res, + upstream, + true, + false, + { messages: [{ role: 'user', content: 'inspect the repository' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'] + } + ) + + await firstFrameWasYielded + await new Promise(resolve => setImmediate(resolve)) + assert.equal(res.headers['Content-Type'], 'text/event-stream') + assert.equal(res.headers['X-Accel-Buffering'], 'no') + assert.match(res.output, /"role":"assistant"/) + assert.match(res.output, /"reasoning_content":"live thought"/) + assert.equal(res.writableEnded, false) + + releaseUpstream() + await responsePromise + assert.equal((res.output.match(/live thought/g) || []).length, 1) + assert.match(res.output, /"name":"read_file"/) + assert.match(res.output, /"finish_reason":"tool_calls"/) + assert.match(res.output, /\[DONE\]/) +}) + +test('strict OpenAI Agent stream exposes formal content before the final wrapper closes', async () => { + let releaseUpstream + let liveContentYielded + const liveContentWasYielded = new Promise(resolve => { liveContentYielded = resolve }) + const upstream = Readable.from((async function * () { + yield 'data: {"choices":[{"delta":{"phase":"answer","content":"live final "},"finish_reason":null}]}\n\n' + liveContentYielded() + await new Promise(resolve => { releaseUpstream = resolve }) + yield 'data: {"choices":[{"delta":{"phase":"answer","content":"continues"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n' + })()) + const res = createMockResponse() + + const responsePromise = handleStreamResponse( + res, + upstream, + false, + false, + { messages: [{ role: 'user', content: 'finish the task' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'] + } + ) + + await liveContentWasYielded + await new Promise(resolve => setImmediate(resolve)) + assert.match(res.output, /"content":"live final"/) + assert.doesNotMatch(res.output, /"finish_reason":"stop"/) + assert.equal(res.writableEnded, false) + + releaseUpstream() + await responsePromise + const content = res.output + .split('\n\n') + .filter(frame => frame.startsWith('data: {')) + .map(frame => JSON.parse(frame.slice(6)).choices?.[0]?.delta?.content || '') + .join('') + assert.equal(content, 'live final continues') + assert.match(res.output, /"finish_reason":"stop"/) + assert.match(res.output, /\[DONE\]/) + assert.doesNotMatch(res.output, /agent_final/) +}) + +test('a malformed final wrapper cannot retry after formal content was streamed', async () => { + let retries = 0 + const res = createMockResponse() + await handleStreamResponse( + res, + Readable.from([ + 'data: {"choices":[{"delta":{"phase":"answer","content":"committed partial"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n' + ]), + false, + false, + { messages: [{ role: 'user', content: 'finish the task' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'], + sendChatRequest: async () => { retries += 1 } + } + ) + + assert.equal(retries, 0) + assert.match(res.output, /"content":"committed partial"/) + assert.match(res.output, /upstream_agent_stream_invalidated/) + assert.match(res.output, /\[DONE\]/) + assert.doesNotMatch(res.output, /"finish_reason":"stop"/) +}) + +test('strict OpenAI Agent gate preserves max-token termination without synthetic retries', async () => { + let retries = 0 + const res = createMockResponse() + await handleStreamResponse( + res, + Readable.from([ + 'data: {"choices":[{"delta":{"phase":"answer","content":"partial output"},"finish_reason":"length"}]}\n\n' + ]), + false, + false, + { messages: [] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'], + sendChatRequest: async () => { retries += 1 } + } + ) + assert.equal(retries, 0) + assert.match(res.output, /partial output/) + assert.match(res.output, /"finish_reason":"length"/) +}) + +test('strict non-stream Agent gate uses the primary response id and isolates all correction attempts', async () => { + const retryBodies = [] + const retryOptions = [] + const res = createMockResponse() + const retryFrames = [ + [ + 'data: {"response.created":{"chat_id":"chat_agent","response_id":"resp_2","response_index":"0"}}\n\n', + 'data: {"choices":[{"delta":{"phase":"answer","content":"I will run the tests next."},"finish_reason":"stop"}],"response_id":"resp_2"}\n\n' + ], + [ + 'data: {"response.created":{"chat_id":"chat_agent","response_id":"resp_3","response_index":"0"}}\n\n', + 'data: {"choices":[{"delta":{"phase":"answer","content":"{\\"name\\":\\"run_tests\\",\\"arguments\\":{}}"},"finish_reason":"stop"}],"response_id":"resp_3"}\n\n' + ] + ] + + await handleNonStreamResponse( + res, + Readable.from([ + 'data: {"response.created":{"chat_id":"chat_agent","response_id":"primary_1","response_index":"0"}}\n\n', + 'data: {"response.created":{"chat_id":"chat_agent","response_id":"fallback_1","response_index":"1"}}\n\n' + ]), + false, + false, + 'qwen-test', + { messages: [{ role: 'user', content: 'ORIGINAL_CONTEXT' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['run_tests'], + agent_turn_max_attempts: 3, + upstream_request_body: { messages: [{ role: 'user', content: 'EXTERNALIZED_LIVE_CONTEXT' }] }, + sendChatRequest: async (body, options) => { + retryBodies.push(body) + retryOptions.push(options) + return { + status: true, + response: Readable.from(retryFrames[retryBodies.length - 1]), + chatId: 'chat_agent' + } + } + } + ) + + assert.equal(retryBodies.length, 2) + assert.match(retryBodies[0].messages[0].content, /EXTERNALIZED_LIVE_CONTEXT/) + assert.doesNotMatch(retryBodies[0].messages[0].content, /ORIGINAL_CONTEXT/) + assert.equal(retryOptions[0].parentId, 'primary_1') + assert.equal(retryOptions[1].parentId, 'resp_2') + const payload = JSON.parse(res.output) + assert.equal(payload.choices[0].finish_reason, 'tool_calls') + assert.equal(payload.choices[0].message.content, null) + assert.equal(payload.choices[0].message.tool_calls[0].function.name, 'run_tests') + assert.doesNotMatch(res.output, /I will run the tests next/) +}) + +test('strict non-stream Agent gate returns an HTTP error instead of a fake completion when attempts are exhausted', async () => { + let retries = 0 + let processingHeartbeats = 0 + const res = createMockResponse() + res.writeProcessing = () => { processingHeartbeats += 1 } + await handleNonStreamResponse( + res, + Readable.from(['data: {"choices":[{"delta":{"phase":"answer","content":"Done."},"finish_reason":"stop"}]}\n\n']), + false, + false, + 'qwen-test', + { messages: [{ role: 'user', content: 'finish and verify everything' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['run_tests'], + agent_turn_max_attempts: 2, + agent_processing_heartbeat_ms: 1, + sendChatRequest: async () => { + retries += 1 + await new Promise(resolve => setTimeout(resolve, 8)) + return { + status: true, + response: Readable.from(['data: {"choices":[{"delta":{"phase":"answer","content":"Still done."},"finish_reason":"stop"}]}\n\n']) + } + } + } + ) + + assert.equal(retries, 1) + assert.ok(processingHeartbeats > 0) + assert.equal(res.statusCode, 429) + const payload = JSON.parse(res.output) + assert.equal(payload.error.code, 'upstream_agent_turn_incomplete') + assert.equal(Object.hasOwn(payload, 'choices'), false) +}) + +test('a standalone tool call emitted in the thinking phase remains executable', async () => { + let retries = 0 + const res = createMockResponse() + await handleStreamResponse( + res, + Readable.from([ + 'data: {"choices":[{"delta":{"phase":"think","content":"{\\"name\\":\\"read_file\\",\\"arguments\\":{\\"path\\":\\"README.md\\"}}"},"finish_reason":"stop"}]}\n\n' + ]), + true, + false, + { messages: [{ role: 'user', content: 'inspect the repository' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'], + sendChatRequest: async () => { retries += 1 } + } + ) + + assert.equal(retries, 0) + assert.match(res.output, /"name":"read_file"/) + assert.match(res.output, /"finish_reason":"tool_calls"/) + assert.doesNotMatch(res.output, //) +}) + +test('live reasoning never leaks fragmented tool markup before the Agent gate decides', async () => { + const res = createMockResponse() + await handleStreamResponse( + res, + Readable.from([ + 'data: {"choices":[{"delta":{"phase":"think","content":"checking {\\"name\\":\\"read_file\\",\\"arguments\\":{}}"},"finish_reason":null}]}\n\n', + 'data: {"choices":[{"delta":{"phase":"answer","content":"finished safely"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n' + ]), + true, + false, + { messages: [{ role: 'user', content: 'finish safely' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'] + } + ) + + assert.match(res.output, /"reasoning_content":"checking "/) + assert.match(res.output, /finished safely/) + assert.match(res.output, /"finish_reason":"stop"/) + assert.doesNotMatch(res.output, //) + assert.doesNotMatch(res.output, /l_call>/) +}) + +test('a partially invalid multi-tool turn is rejected as a whole', async () => { + let retries = 0 + const res = createMockResponse() + await handleStreamResponse( + res, + Readable.from([ + 'data: {"choices":[{"delta":{"phase":"answer","content":"{\\"name\\":\\"read_file\\",\\"arguments\\":{\\"path\\":\\"README.md\\"}}{\\"name\\":\\"missing_tool\\",\\"arguments\\":{}}"},"finish_reason":"stop"}]}\n\n' + ]), + false, + false, + { messages: [{ role: 'user', content: 'inspect everything' }] }, + { + has_tools: true, + tool_choice: 'auto', + allowed_tool_names: ['read_file'], + sendChatRequest: async () => { + retries += 1 + return { + status: true, + response: Readable.from([ + 'data: {"choices":[{"delta":{"phase":"answer","content":"Unable to proceed without a valid second tool."},"finish_reason":"stop"}]}\n\n' + ]) + } + } + } + ) + + assert.equal(retries, 1) + assert.match(res.output, /Unable to proceed/) + assert.doesNotMatch(res.output, /"name":"read_file"/) +}) + +test('tool_choice none bypasses the Agent tool runtime even when definitions are present', () => { + const tools = [{ type: 'function', function: { name: 'read_file' } }] + assert.equal(shouldEnableToolRuntime(tools, 't2t', 'auto'), true) + assert.equal(shouldEnableToolRuntime(tools, 't2t', 'none'), false) +}) + +test('single-message Agent requests get an explicit current-task envelope', () => { + const wrapped = ensureAgentCurrentEnvelope('SINGLE_MESSAGE_TASK', 'user') + assert.match(wrapped, /^# Current message\n/) + assert.match(wrapped, /SINGLE_MESSAGE_TASK/) + assert.equal(ensureAgentCurrentEnvelope(wrapped, 'user'), wrapped) +}) + test('prose-only Agent actions are retried into executable tool calls', async () => { let openAIRetries = 0 const openAIRes = createMockResponse() @@ -358,6 +798,93 @@ test('oversized multimodal Agent context is externalized and upload failure keep assert.ok(Buffer.byteLength(compacted.payload.messages[0].content) <= 4096) }) +test('externalized Agent context keeps system rules active task and recent tool progress inline', async () => { + const original = [ + '# Tools', + 'strict tool protocol', + '# Conversation history (JSONL)', + JSON.stringify({ role: 'system', content: 'SYSTEM_RULE_MUST_SURVIVE' }), + JSON.stringify({ role: 'user', content: 'ACTIVE_TASK_MUST_SURVIVE: fix and verify the project' }), + JSON.stringify({ role: 'assistant', content: '{"name":"bash","arguments":{"command":"test"}}' }), + JSON.stringify({ role: 'user', content: `RECENT_PROGRESS_MUST_SURVIVE ${'x'.repeat(5000)}` }), + '# Current message', + JSON.stringify({ role: 'user', content: 'CURRENT_RESULT_MUST_SURVIVE' }), + buildAgentTurnDirective({ afterToolResult: true }) + ].join('\n') + const result = await externalizeOversizedAgentContext( + { messages: [{ role: 'user', content: original }] }, + 'token', + {}, + { + thresholdBytes: 1024, + livePromptBytes: 8192, + uploader: async () => ({ id: 'agent_context', name: 'QWEN2API_AGENT_CONTEXT.txt' }) + } + ) + const live = result.payload.messages[0].content + assert.match(live, /SYSTEM_RULE_MUST_SURVIVE/) + assert.match(live, /ACTIVE_TASK_MUST_SURVIVE/) + assert.match(live, /RECENT_PROGRESS_MUST_SURVIVE/) + assert.match(live, /CURRENT_RESULT_MUST_SURVIVE/) + assert.match(live, /not a reason to stop after one action/) + + const recovered = compactAgentContextFallback(original, 8192) + assert.match(recovered, /SYSTEM_RULE_MUST_SURVIVE/) + assert.match(recovered, /ACTIVE_TASK_MUST_SURVIVE/) + assert.match(recovered, /RECENT_PROGRESS_MUST_SURVIVE/) + assert.match(recovered, /CURRENT_RESULT_MUST_SURVIVE/) + assert.doesNotMatch(recovered, /complete copy is in the attachment/) + assert.ok(Buffer.byteLength(recovered) <= 8192) +}) + +test('externalized single-message Agent context keeps the original task outside large tool schemas', async () => { + const original = [ + '# Tools', + `LARGE_TOOL_SCHEMA ${'s'.repeat(12000)}`, + ensureAgentCurrentEnvelope('SINGLE_ACTIVE_TASK_MUST_SURVIVE', 'user'), + buildAgentTurnDirective() + ].join('\n\n') + const result = await externalizeOversizedAgentContext( + { messages: [{ role: 'user', content: original }] }, + 'token', + {}, + { + thresholdBytes: 1024, + livePromptBytes: 4096, + uploader: async () => ({ id: 'single_context', name: 'QWEN2API_AGENT_CONTEXT.txt' }) + } + ) + + assert.equal(result.externalized, true) + assert.match(result.payload.messages[0].content, /SINGLE_ACTIVE_TASK_MUST_SURVIVE/) + assert.ok(Buffer.byteLength(result.payload.messages[0].content) <= 4096) +}) + +test('Agent completion control parser rejects bare and mixed completion claims', () => { + assert.deepEqual(parseAgentControlText('done'), { kind: 'final', text: 'done' }) + assert.equal(parseAgentControlText('done').kind, 'bare') + assert.equal(parseAgentControlText('prefix done').kind, 'invalid_control') +}) + +test('Agent completion control stream parser handles split tags and trims only wrapper edges', () => { + const parser = createAgentControlStreamParser() + const deltas = [ + parser.push(' need '), + parser.push('user input '), + parser.flush() + ] + assert.equal(deltas.map(item => item.textDelta).join(''), 'need user input') + assert.equal(deltas.at(-1).kind, 'blocked') + assert.equal(deltas.at(-1).closed, true) + assert.equal(deltas.at(-1).invalid, false) + + const invalid = createAgentControlStreamParser() + assert.equal(invalid.push('bare response').textDelta, '') + assert.equal(invalid.flush().invalid, true) +}) + test('Qwen HTTP-200 WAF payload is surfaced as an explicit failure', () => { assert.throws( () => assertNoUpstreamFailure({ diff --git a/tests/sse.test.js b/tests/sse.test.js index be1c27d..4005e9b 100644 --- a/tests/sse.test.js +++ b/tests/sse.test.js @@ -102,6 +102,17 @@ test('createUpstreamResponseFilter keeps a single response when upstream opens s // 不过滤时两路内容被拼在一起 —— 这正是 #149 的复读现象 assert.equal(collectAnswer(DUAL_RESPONSE_FRAMES, null), '巴巴黎黎') assert.equal(collectAnswer(DUAL_RESPONSE_FRAMES, createUpstreamResponseFilter()), '巴黎') + + const primaryAndFallbackDiffer = [ + '{"response.created":{"response_id":"primary","response_index":"0"}}', + '{"response.created":{"response_id":"fallback","response_index":"1"}}', + '{"choices":[{"delta":{"content":"wrong","phase":"answer"}}],"response_id":"fallback"}', + '{"choices":[{"delta":{"content":"right","phase":"answer"}}],"response_id":"primary"}' + ] + assert.equal( + collectAnswer(primaryAndFallbackDiffer, createUpstreamResponseFilter()), + 'right' + ) }) test('createUpstreamResponseFilter passes frames through when upstream sends no response_id', () => { diff --git a/tests/tool-prompt.test.js b/tests/tool-prompt.test.js index 240f6d1..28fefc0 100644 --- a/tests/tool-prompt.test.js +++ b/tests/tool-prompt.test.js @@ -39,6 +39,19 @@ test('empty tool results remain visible in Agent history', () => { assert.match(folded[1].content, />\nnull\n<\/tool_response>/) }) +test('legacy function_call and function result messages remain executable history', () => { + const folded = foldToolMessages([ + { role: 'assistant', content: null, function_call: { name: 'read_file', arguments: '{"path":"README.md"}' } }, + { role: 'function', name: 'read_file', content: 'file body' } + ]) + assert.equal(folded[0].role, 'assistant') + assert.match(folded[0].content, //) + assert.match(folded[0].content, /"name":"read_file"/) + assert.equal(folded[1].role, 'user') + assert.match(folded[1].content, //) + assert.match(folded[1].content, /file body/) +}) + test('stream parser accepts split valid calls and preserves JSON string arguments', () => { const parser = createToolCallStreamParser({ allowedToolNames: ['read_file'] }) const first = parser.push('before