From 6b6f9ed01c1c17128381fa7341d820977c104b5b Mon Sep 17 00:00:00 2001 From: john Date: Sat, 25 Jul 2026 21:55:00 +0800 Subject: [PATCH] fix: unblock local news page generation --- agent-run-gateway.mjs | 43 +- agent-run-gateway.test.mjs | 21 + chat-agent-run-gate.mjs | 5 +- chat-agent-run-gate.test.mjs | 8 + chat-intent-router.mjs | 14 +- chat-intent-router.test.mjs | 36 ++ deepseek-no-think-proxy.mjs | 403 +++++++++++++++++++ deepseek-no-think-proxy.test.mjs | 170 ++++++++ llm-providers.mjs | 91 ++++- llm-providers.test.mjs | 131 ++++++ scripts/dev-core.mjs | 53 ++- server/portal-gateway-services-bootstrap.mjs | 21 + session-reply-wait.mjs | 19 +- session-reply-wait.test.mjs | 32 ++ src/hooks/useTKMindChat.ts | 20 +- tkmind-proxy.mjs | 55 ++- 16 files changed, 1075 insertions(+), 47 deletions(-) create mode 100644 deepseek-no-think-proxy.mjs create mode 100644 deepseek-no-think-proxy.test.mjs diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 02abf3a..14f2334 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -516,6 +516,7 @@ export function createAgentRunGateway({ observeWorkflowValidation = null, isSessionExternallyBusy = null, validateRunDeliverables = null, + cancelSessionOnRetry = null, quiesceSessionOnTerminal = null, retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS, autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true), @@ -613,6 +614,35 @@ export function createAgentRunGateway({ } } + async function cancelRetrySession(runId, { + userId, + sessionId, + requestId = null, + }) { + if (!sessionId || typeof cancelSessionOnRetry !== 'function') return; + try { + const result = await cancelSessionOnRetry({ + runId, + userId, + sessionId, + requestId, + }); + await appendEvent(runId, 'session_request_cancelled_for_retry', { + sessionId, + cancellation: result ?? null, + }); + } catch (err) { + await appendEvent(runId, 'session_retry_cancellation_failed', { + sessionId, + error: err instanceof Error ? err.message : String(err), + }).catch(() => {}); + console.warn( + `[AgentRun] failed to cancel session ${sessionId} before retry:`, + err instanceof Error ? err.message : err, + ); + } + } + async function appendExecutionPlanEvent(input, source) { const executionPlan = source?.executionPlan; if (!executionPlan?.dryRun) return; @@ -1563,7 +1593,10 @@ export function createAgentRunGateway({ } catch (err) { if ( !replacedPoisonedSession - && err?.code === 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED' + && ( + err?.code === 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED' + || err?.code === 'SESSION_REASONING_CONTENT_POISONED' + ) ) { const previousSessionId = sessionId; const repairedConversation = Array.isArray(err?.repairedConversation) @@ -1585,6 +1618,7 @@ export function createAgentRunGateway({ await appendEvent(runId, 'poisoned_session_replaced', { previousSessionId, sessionId, + reason: err?.code ?? null, repairedMessageCount: repairedConversation.length, persistedMessageCount: persisted.saved, }); @@ -1844,6 +1878,13 @@ export function createAgentRunGateway({ await appendEvent(runId, 'timeout', { timeoutMs: runTimeoutMs }); } const retryable = !timedOut && err?.retryable !== false && nextAttempt < retryDelaysMs.length; + if (retryable) { + await cancelRetrySession(runId, { + userId: row.user_id, + sessionId: recoverySessionId, + requestId: row.request_id ?? null, + }); + } await markRun(runId, retryable ? 'retryable' : 'failed', { error_message: message, completed_at: retryable ? null : nowMs(), diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index f351c31..6ff752f 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -2652,6 +2652,7 @@ test('agent run fails non-retryably when tool gateway artifact validation fails' test('agent run retries transient failures and then becomes terminal', async () => { const pool = createFakePool(); + const retryCancellations = []; const gateway = createAgentRunGateway({ pool, userAuth: {}, @@ -2663,6 +2664,10 @@ test('agent run retries transient failures and then becomes terminal', async () throw new Error('upstream unavailable'); }, }, + cancelSessionOnRetry: async (input) => { + retryCancellations.push(input); + return { cancelled: true, skipped: false }; + }, retryDelaysMs: [0, 0], }); @@ -2674,6 +2679,22 @@ test('agent run retries transient failures and then becomes terminal', async () await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); assert.equal(pool.runs.get(run.id).attempts, 2); assert.match(pool.runs.get(run.id).error_message, /upstream unavailable/); + assert.deepEqual(retryCancellations, [ + { + runId: run.id, + userId: 'user-1', + sessionId: 'session-1', + requestId: 'req-1', + }, + ]); + assert.equal( + pool.events.some( + (event) => + event.runId === run.id + && event.eventType === 'session_request_cancelled_for_retry', + ), + true, + ); }); test('agent run queue limits concurrent execution', async () => { diff --git a/chat-agent-run-gate.mjs b/chat-agent-run-gate.mjs index 825f80d..1db20b2 100644 --- a/chat-agent-run-gate.mjs +++ b/chat-agent-run-gate.mjs @@ -46,7 +46,10 @@ export function shouldKeepStreamingAfterRunError(status, message = '', code = '' // the session SSE is still the source of truth, so recover from the session // instead of showing a terminal error in the chat composer. const text = `${String(code ?? '')} ${String(message ?? '')}`.toLowerCase(); - if (String(code ?? '').trim() === 'SESSION_RUN_CONFLICT') return false; + const normalizedCode = String(code ?? '').trim(); + if (normalizedCode === 'SESSION_RUN_CONFLICT' || normalizedCode === 'AGENT_RUN_FAILED') { + return false; + } return text.includes('session already has an active request') || text.includes('active request. cancel it first'); } diff --git a/chat-agent-run-gate.test.mjs b/chat-agent-run-gate.test.mjs index a24ad05..282908a 100644 --- a/chat-agent-run-gate.test.mjs +++ b/chat-agent-run-gate.test.mjs @@ -68,6 +68,14 @@ test('only ambiguous transport failures keep the chat streaming', () => { ), true, ); + assert.equal( + shouldKeepStreamingAfterRunError( + undefined, + 'Session already has an active request. Cancel it first.', + 'AGENT_RUN_FAILED', + ), + false, + ); assert.equal(shouldKeepStreamingAfterRunError(undefined, '后台任务失败'), false); }); diff --git a/chat-intent-router.mjs b/chat-intent-router.mjs index 003fbd1..61a71ee 100644 --- a/chat-intent-router.mjs +++ b/chat-intent-router.mjs @@ -261,16 +261,24 @@ export function coercePageGenerationSkill(classification, text, { grantedSkills if (grantedSkills.length > 0 && !grantedSkills.includes('static-page-publish')) return classification; const shouldUsePublishSkill = - !classification.suggestedSkill || classification.suggestedSkill === 'product-campaign-page'; + !classification.suggestedSkill + || classification.suggestedSkill === 'product-campaign-page' + || classification.suggestedSkill === 'web'; if (!shouldUsePublishSkill) return classification; const reasonSuffix = classification.suggestedSkill === 'product-campaign-page' ? '（内容页意图，改用 static-page-publish）' - : '（内容页生成意图）'; + : classification.suggestedSkill === 'web' + ? '（实时资料整理后交付 static-page-publish 页面）' + : '（内容页生成意图）'; + const realtimeBrief = classification.suggestedSkill === 'web' + ? `${String(classification.agentBrief ?? '').trim()}\n` + : ''; return { ...classification, suggestedSkill: 'static-page-publish', - agentBrief: '生成或更新 MindSpace 公开内容页面，并返回可访问链接；信息不足时使用合理默认，不要停在追问。', + agentBrief: + `${realtimeBrief}生成或更新 MindSpace 公开内容页面，并返回可访问链接；信息不足时使用合理默认，不要停在追问。`, reason: `${classification.reason}${reasonSuffix}`, }; } diff --git a/chat-intent-router.test.mjs b/chat-intent-router.test.mjs index 345cc2f..58bbb6f 100644 --- a/chat-intent-router.test.mjs +++ b/chat-intent-router.test.mjs @@ -232,6 +232,42 @@ test('coercePageGenerationSkill replaces product-campaign-page for content pages assert.match(coerced.reason, /static-page-publish/); }); +test('coercePageGenerationSkill preserves realtime research while making the page the deliverable', () => { + const coerced = coercePageGenerationSkill( + { + route: CHAT_INTENT_ROUTE.AGENT, + suggestedSkill: 'web', + reason: '需要搜索实时资料（实时查询强制 web）', + agentBrief: '先搜索并核对今天的新闻来源。', + confidence: 1, + source: 'rule', + }, + '帮我整理今天的新闻做成页面', + { grantedSkills: ['web', 'static-page-publish'] }, + ); + assert.equal(coerced.suggestedSkill, 'static-page-publish'); + assert.match(coerced.agentBrief, /先搜索并核对/); + assert.match(coerced.agentBrief, /公开内容页面/); + assert.match(coerced.reason, /实时资料整理后交付/); +}); + +test('createChatIntentRouter keeps static-page-publish for forced realtime page tasks', async () => { + const router = createChatIntentRouter(); + const result = await router.classify({ + userMessage: { + role: 'user', + content: [{ type: 'text', text: '帮我整理今天的新闻做成页面' }], + metadata: { displayText: '帮我整理今天的新闻做成页面' }, + }, + forceDeepReasoning: true, + grantedSkills: ['web', 'static-page-publish'], + }); + assert.equal(result.route, CHAT_INTENT_ROUTE.AGENT); + assert.equal(result.suggestedSkill, 'static-page-publish'); + assert.match(result.agentBrief, /tkmind_search/); + assert.match(result.agentBrief, /公开内容页面/); +}); + test('classifyWithRules routes themed article requests to agent orchestration', () => { const result = classifyWithRules({ text: '帮我写个主题文章', diff --git a/deepseek-no-think-proxy.mjs b/deepseek-no-think-proxy.mjs new file mode 100644 index 0000000..5e92e42 --- /dev/null +++ b/deepseek-no-think-proxy.mjs @@ -0,0 +1,403 @@ +import http from 'node:http'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { Agent, fetch as undiciFetch } from 'undici'; + +export const DEFAULT_DEEPSEEK_NO_THINK_PORT = 18036; +export const DEFAULT_DEEPSEEK_UPSTREAM = 'https://api.deepseek.com'; +export const DEFAULT_MOONSHOT_UPSTREAM = 'https://api.moonshot.cn'; + +const insecureDispatcher = new Agent({ + connect: { rejectUnauthorized: false }, +}); + +function envFlag(value, defaultValue = false) { + if (value == null || String(value).trim() === '') return defaultValue; + return ['1', 'true', 'yes', 'on'].includes(String(value).trim().toLowerCase()); +} + +export function deepseekDisableThinkingEnabled(env = process.env) { + const raw = env.MEMIND_DEEPSEEK_DISABLE_THINKING; + if (raw != null && String(raw).trim() !== '') { + return envFlag(raw, false); + } + const runtimeProfile = String(env.MEMIND_RUNTIME_PROFILE ?? '').trim().toLowerCase(); + return runtimeProfile === 'local' || runtimeProfile === 'split-service'; +} + +export function moonshotToolSchemaCompatEnabled(env = process.env) { + const raw = env.MEMIND_MOONSHOT_TOOL_SCHEMA_COMPAT; + if (raw != null && String(raw).trim() !== '') { + return envFlag(raw, false); + } + const runtimeProfile = String(env.MEMIND_RUNTIME_PROFILE ?? '').trim().toLowerCase(); + return runtimeProfile === 'local' || runtimeProfile === 'split-service'; +} + +export function resolveDeepseekNoThinkListenPort(env = process.env) { + const port = Number(env.MEMIND_DEEPSEEK_NO_THINK_PORT ?? DEFAULT_DEEPSEEK_NO_THINK_PORT); + return Number.isFinite(port) && port > 0 ? port : DEFAULT_DEEPSEEK_NO_THINK_PORT; +} + +export function resolveDeepseekNoThinkProxyBaseUrl(env = process.env) { + const explicit = String(env.MEMIND_DEEPSEEK_NO_THINK_BASE_URL ?? '').trim(); + if (explicit) return explicit.replace(/\/$/, ''); + const host = String( + env.MEMIND_DEEPSEEK_NO_THINK_HOST + ?? env.MEMIND_GOOSED_HOST_GATEWAY + ?? 'host.lima.internal', + ).trim() || 'host.lima.internal'; + const port = resolveDeepseekNoThinkListenPort(env); + return `http://${host}:${port}/v1`; +} + +export function resolveMoonshotCompatProxyBaseUrl(env = process.env) { + const explicit = String(env.MEMIND_MOONSHOT_COMPAT_BASE_URL ?? '').trim(); + if (explicit) return explicit.replace(/\/$/, ''); + const host = String( + env.MEMIND_DEEPSEEK_NO_THINK_HOST + ?? env.MEMIND_GOOSED_HOST_GATEWAY + ?? 'host.lima.internal', + ).trim() || 'host.lima.internal'; + const port = resolveDeepseekNoThinkListenPort(env); + return `http://${host}:${port}/moonshot/v1`; +} + +export function resolveDeepseekUpstreamBase(env = process.env) { + const raw = String( + env.MEMIND_DEEPSEEK_UPSTREAM + ?? env.DEEPSEEK_API_BASE_URL + ?? DEFAULT_DEEPSEEK_UPSTREAM, + ).trim().replace(/\/$/, ''); + return raw.replace(/\/v1$/i, '') || DEFAULT_DEEPSEEK_UPSTREAM; +} + +export function resolveMoonshotUpstreamBase(env = process.env) { + const raw = String( + env.MEMIND_MOONSHOT_UPSTREAM + ?? DEFAULT_MOONSHOT_UPSTREAM, + ).trim().replace(/\/$/, ''); + return raw.replace(/\/v1$/i, '') || DEFAULT_MOONSHOT_UPSTREAM; +} + +export function isMoonshotApiUrl(value) { + try { + const hostname = new URL(String(value ?? '')).hostname.toLowerCase(); + return hostname === 'api.moonshot.cn' || hostname === 'api.moonshot.ai'; + } catch { + return false; + } +} + +/** + * DeepSeek V4 defaults to thinking mode. Tool rounds then require reasoning_content + * to be replayed; goosed can drop it and fail with HTTP 400. Force-disable thinking + * unless the caller already set an explicit thinking object. + */ +export function injectDeepseekThinkingDisabled(body) { + if (!body || typeof body !== 'object' || Array.isArray(body)) { + return { body, injected: false }; + } + if (body.thinking && typeof body.thinking === 'object') { + return { body, injected: false }; + } + return { + body: { + ...body, + thinking: { type: 'disabled' }, + }, + injected: true, + }; +} + +function decodeJsonPointerToken(value) { + return String(value ?? '').replace(/~1/g, '/').replace(/~0/g, '~'); +} + +function resolveLocalSchemaRef(root, ref) { + if (!String(ref ?? '').startsWith('#/')) return null; + const tokens = String(ref) + .slice(2) + .split('/') + .map(decodeJsonPointerToken); + let current = root; + for (const token of tokens) { + if (!current || typeof current !== 'object' || !(token in current)) return null; + current = current[token]; + } + return current; +} + +/** + * Moonshot rejects otherwise valid local JSON Schema references with + * "detected infinite recursion". Inline local references before forwarding + * tools so Kimi receives an equivalent, self-contained parameter schema. + */ +export function flattenLocalJsonSchemaRefs(schema) { + if (!schema || typeof schema !== 'object' || Array.isArray(schema)) return schema; + const root = schema; + + function visit(value, activeRefs = new Set()) { + if (Array.isArray(value)) return value.map((item) => visit(item, activeRefs)); + if (!value || typeof value !== 'object') return value; + + const ref = typeof value.$ref === 'string' ? value.$ref : ''; + if (ref.startsWith('#/')) { + const target = resolveLocalSchemaRef(root, ref); + const siblings = Object.fromEntries( + Object.entries(value).filter(([key]) => key !== '$ref'), + ); + if (!target || activeRefs.has(ref)) { + return visit({ + type: 'object', + description: String( + siblings.description ?? 'Recursive schema value', + ), + }, activeRefs); + } + const nextRefs = new Set(activeRefs); + nextRefs.add(ref); + const resolved = visit(target, nextRefs); + return { + ...(resolved && typeof resolved === 'object' && !Array.isArray(resolved) + ? resolved + : {}), + ...visit(siblings, activeRefs), + }; + } + + const output = {}; + for (const [key, child] of Object.entries(value)) { + if (key === '$schema' || key === '$defs' || key === 'definitions') continue; + output[key] = visit(child, activeRefs); + } + return output; + } + + return visit(schema); +} + +export function sanitizeMoonshotToolSchemas(body) { + if (!body || typeof body !== 'object' || Array.isArray(body) || !Array.isArray(body.tools)) { + return { body, sanitized: 0 }; + } + let sanitized = 0; + const tools = body.tools.map((tool) => { + const parameters = tool?.function?.parameters; + if (!parameters || typeof parameters !== 'object') return tool; + const serialized = JSON.stringify(parameters); + if (!serialized.includes('"$ref"') && !serialized.includes('"$defs"')) return tool; + sanitized += 1; + return { + ...tool, + function: { + ...tool.function, + parameters: flattenLocalJsonSchemaRefs(parameters), + }, + }; + }); + return sanitized > 0 + ? { body: { ...body, tools }, sanitized } + : { body, sanitized: 0 }; +} + +function readRequestBody(req) { + return new Promise((resolve, reject) => { + const chunks = []; + req.on('data', (chunk) => chunks.push(chunk)); + req.on('end', () => resolve(Buffer.concat(chunks))); + req.on('error', reject); + }); +} + +function buildUpstreamUrl(upstreamBase, reqUrl) { + const incoming = new URL(reqUrl || '/', 'http://127.0.0.1'); + let pathname = incoming.pathname || '/'; + // Accept both /v1/chat/completions and /chat/completions from OpenAI-compatible clients. + if (!pathname.startsWith('/v1/') && pathname !== '/v1') { + pathname = pathname === '/' ? '/v1' : `/v1${pathname}`; + } + return `${upstreamBase}${pathname}${incoming.search}`; +} + +export function decodedUpstreamResponseHeaders(headers) { + const responseHeaders = {}; + headers?.forEach?.((value, key) => { + const normalized = String(key).toLowerCase(); + // undici transparently decodes gzip/br but keeps the upstream encoding + // headers. Forwarding them makes the downstream client decode plaintext + // again and silently lose the SSE stream. + if ( + normalized === 'transfer-encoding' + || normalized === 'content-encoding' + || normalized === 'content-length' + || normalized === 'connection' + ) { + return; + } + responseHeaders[key] = value; + }); + return responseHeaders; +} + +export function createDeepseekNoThinkProxy({ + upstreamBase = resolveDeepseekUpstreamBase(), + moonshotUpstreamBase = resolveMoonshotUpstreamBase(), + fetchImpl = undiciFetch, + logger = console, +} = {}) { + const normalizedUpstream = String(upstreamBase || DEFAULT_DEEPSEEK_UPSTREAM).replace(/\/$/, ''); + const normalizedMoonshotUpstream = String( + moonshotUpstreamBase || DEFAULT_MOONSHOT_UPSTREAM, + ).replace(/\/$/, ''); + + async function handle(req, res) { + if (req.method === 'GET' && (req.url === '/health' || req.url === '/healthz')) { + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ + ok: true, + upstream: normalizedUpstream, + moonshotUpstream: normalizedMoonshotUpstream, + })); + return; + } + + if (req.method === 'OPTIONS') { + res.writeHead(204, { + 'Access-Control-Allow-Origin': '*', + 'Access-Control-Allow-Headers': '*', + 'Access-Control-Allow-Methods': 'GET,POST,OPTIONS', + }); + res.end(); + return; + } + + const incoming = new URL(req.url || '/', 'http://127.0.0.1'); + const isMoonshot = incoming.pathname === '/moonshot' + || incoming.pathname.startsWith('/moonshot/'); + const routedRequestUrl = isMoonshot + ? `${incoming.pathname.slice('/moonshot'.length) || '/'}${incoming.search}` + : req.url; + const upstreamUrl = buildUpstreamUrl( + isMoonshot ? normalizedMoonshotUpstream : normalizedUpstream, + routedRequestUrl, + ); + const headers = { ...req.headers }; + delete headers.host; + delete headers['content-length']; + + let body; + if (req.method !== 'GET' && req.method !== 'HEAD') { + const raw = await readRequestBody(req); + if (raw.length > 0 && /chat\/completions/i.test(upstreamUrl)) { + try { + const parsed = JSON.parse(raw.toString('utf8')); + if (isMoonshot) { + const schemaResult = sanitizeMoonshotToolSchemas(parsed); + const thinkingResult = injectDeepseekThinkingDisabled(schemaResult.body); + body = Buffer.from(JSON.stringify(thinkingResult.body), 'utf8'); + if (schemaResult.sanitized > 0) { + logger.info?.( + `[moonshot-compat] inlined local refs in ${schemaResult.sanitized} tool schema(s) model=${String(parsed?.model ?? '')}`, + ); + } + if (thinkingResult.injected) { + logger.info?.( + `[moonshot-compat] injected thinking.disabled model=${String(parsed?.model ?? '')}`, + ); + } + } else { + const { body: next, injected } = injectDeepseekThinkingDisabled(parsed); + body = Buffer.from(JSON.stringify(next), 'utf8'); + if (injected) { + logger.info?.( + `[deepseek-no-think] injected thinking.disabled model=${String(parsed?.model ?? '')}`, + ); + } + } + } catch { + body = raw; + } + } else { + body = raw; + } + } + + const upstream = await fetchImpl(upstreamUrl, { + method: req.method, + headers, + body, + dispatcher: upstreamUrl.startsWith('https://') ? insecureDispatcher : undefined, + }); + + const responseHeaders = decodedUpstreamResponseHeaders(upstream.headers); + res.writeHead(upstream.status, responseHeaders); + if (!upstream.body) { + res.end(); + return; + } + const reader = upstream.body.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + res.write(Buffer.from(value)); + } + res.end(); + } catch (err) { + logger.warn?.('[deepseek-no-think] stream error', err instanceof Error ? err.message : err); + res.end(); + } + } + + return { handle, upstreamBase: normalizedUpstream }; +} + +export function startDeepseekNoThinkProxy({ + port = resolveDeepseekNoThinkListenPort(), + host = '0.0.0.0', + upstreamBase = resolveDeepseekUpstreamBase(), + moonshotUpstreamBase = resolveMoonshotUpstreamBase(), + fetchImpl = undiciFetch, + logger = console, +} = {}) { + const proxy = createDeepseekNoThinkProxy({ + upstreamBase, + moonshotUpstreamBase, + fetchImpl, + logger, + }); + const server = http.createServer((req, res) => { + proxy.handle(req, res).catch((err) => { + logger.error?.('[deepseek-no-think] request failed', err instanceof Error ? err.message : err); + if (!res.headersSent) { + res.writeHead(502, { 'Content-Type': 'application/json' }); + } + res.end(JSON.stringify({ + error: { + message: err instanceof Error ? err.message : 'deepseek no-think proxy failed', + }, + })); + }); + }); + return new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(port, host, () => { + logger.info?.( + `[llm-compat-proxy] listening on http://${host}:${port} -> ${proxy.upstreamBase}`, + ); + resolve(server); + }); + }); +} + +const isMain = Boolean( + process.argv[1] + && fileURLToPath(import.meta.url) === path.resolve(process.argv[1]), +); +if (isMain) { + startDeepseekNoThinkProxy().catch((err) => { + console.error(err); + process.exit(1); + }); +} diff --git a/deepseek-no-think-proxy.test.mjs b/deepseek-no-think-proxy.test.mjs new file mode 100644 index 0000000..efd3912 --- /dev/null +++ b/deepseek-no-think-proxy.test.mjs @@ -0,0 +1,170 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + decodedUpstreamResponseHeaders, + deepseekDisableThinkingEnabled, + flattenLocalJsonSchemaRefs, + injectDeepseekThinkingDisabled, + isMoonshotApiUrl, + moonshotToolSchemaCompatEnabled, + resolveDeepseekNoThinkProxyBaseUrl, + resolveMoonshotCompatProxyBaseUrl, + sanitizeMoonshotToolSchemas, +} from './deepseek-no-think-proxy.mjs'; + +test('injectDeepseekThinkingDisabled adds thinking.disabled when absent', () => { + const { body, injected } = injectDeepseekThinkingDisabled({ + model: 'deepseek-v4-flash', + messages: [{ role: 'user', content: 'hi' }], + }); + assert.equal(injected, true); + assert.deepEqual(body.thinking, { type: 'disabled' }); + assert.equal(body.model, 'deepseek-v4-flash'); +}); + +test('injectDeepseekThinkingDisabled preserves explicit thinking config', () => { + const { body, injected } = injectDeepseekThinkingDisabled({ + model: 'deepseek-v4-pro', + thinking: { type: 'enabled' }, + }); + assert.equal(injected, false); + assert.deepEqual(body.thinking, { type: 'enabled' }); +}); + +test('deepseekDisableThinkingEnabled defaults on for local runtime profile', () => { + assert.equal(deepseekDisableThinkingEnabled({ MEMIND_RUNTIME_PROFILE: 'local' }), true); + assert.equal(deepseekDisableThinkingEnabled({ MEMIND_RUNTIME_PROFILE: 'split-service' }), true); + assert.equal(deepseekDisableThinkingEnabled({ MEMIND_RUNTIME_PROFILE: 'production' }), false); + assert.equal( + deepseekDisableThinkingEnabled({ + MEMIND_RUNTIME_PROFILE: 'local', + MEMIND_DEEPSEEK_DISABLE_THINKING: '0', + }), + false, + ); + assert.equal( + deepseekDisableThinkingEnabled({ + MEMIND_RUNTIME_PROFILE: 'production', + MEMIND_DEEPSEEK_DISABLE_THINKING: '1', + }), + true, + ); +}); + +test('moonshotToolSchemaCompatEnabled defaults on for local runtime profiles', () => { + assert.equal(moonshotToolSchemaCompatEnabled({ MEMIND_RUNTIME_PROFILE: 'local' }), true); + assert.equal(moonshotToolSchemaCompatEnabled({ MEMIND_RUNTIME_PROFILE: 'split-service' }), true); + assert.equal(moonshotToolSchemaCompatEnabled({ MEMIND_RUNTIME_PROFILE: 'production' }), false); + assert.equal( + moonshotToolSchemaCompatEnabled({ + MEMIND_RUNTIME_PROFILE: 'local', + MEMIND_MOONSHOT_TOOL_SCHEMA_COMPAT: '0', + }), + false, + ); +}); + +test('resolveDeepseekNoThinkProxyBaseUrl uses host gateway defaults', () => { + assert.equal( + resolveDeepseekNoThinkProxyBaseUrl({}), + 'http://host.lima.internal:18036/v1', + ); + assert.equal( + resolveDeepseekNoThinkProxyBaseUrl({ + MEMIND_DEEPSEEK_NO_THINK_BASE_URL: 'http://10.0.0.2:9/v1/', + }), + 'http://10.0.0.2:9/v1', + ); +}); + +test('Moonshot compat routing recognizes official endpoints and uses a distinct path', () => { + assert.equal(isMoonshotApiUrl('https://api.moonshot.cn/v1'), true); + assert.equal(isMoonshotApiUrl('https://api.moonshot.ai/v1/'), true); + assert.equal(isMoonshotApiUrl('https://api.deepseek.com/v1'), false); + assert.equal( + resolveMoonshotCompatProxyBaseUrl({}), + 'http://host.lima.internal:18036/moonshot/v1', + ); +}); + +test('flattenLocalJsonSchemaRefs inlines local definitions without mutating the source', () => { + const source = { + $schema: 'https://json-schema.org/draft/2020-12/schema', + type: 'object', + properties: { + command: { + description: 'command override', + $ref: '#/$defs/CacheCommand', + }, + }, + required: ['command'], + $defs: { + CacheCommand: { + type: 'string', + oneOf: [{ const: 'list' }, { const: 'clear' }], + }, + }, + }; + const flattened = flattenLocalJsonSchemaRefs(source); + assert.equal(JSON.stringify(flattened).includes('$ref'), false); + assert.equal(JSON.stringify(flattened).includes('$defs'), false); + assert.equal(flattened.properties.command.type, 'string'); + assert.equal(flattened.properties.command.description, 'command override'); + assert.equal(source.properties.command.$ref, '#/$defs/CacheCommand'); +}); + +test('sanitizeMoonshotToolSchemas only rewrites function schemas with local refs', () => { + const plain = { + type: 'function', + function: { name: 'plain', parameters: { type: 'object' } }, + }; + const body = { + model: 'kimi-k2.6', + tools: [ + plain, + { + type: 'function', + function: { + name: 'cache', + parameters: { + type: 'object', + properties: { command: { $ref: '#/$defs/Command' } }, + $defs: { Command: { type: 'string', enum: ['list', 'clear'] } }, + }, + }, + }, + ], + }; + const result = sanitizeMoonshotToolSchemas(body); + assert.equal(result.sanitized, 1); + assert.equal(result.body.tools[0], plain); + assert.deepEqual( + result.body.tools[1].function.parameters.properties.command, + { type: 'string', enum: ['list', 'clear'] }, + ); +}); + +test('Moonshot requests can reuse thinking.disabled injection', () => { + const schemaResult = sanitizeMoonshotToolSchemas({ + model: 'kimi-k2.6', + tools: [], + }); + const thinkingResult = injectDeepseekThinkingDisabled(schemaResult.body); + assert.equal(thinkingResult.injected, true); + assert.deepEqual(thinkingResult.body.thinking, { type: 'disabled' }); +}); + +test('decodedUpstreamResponseHeaders drops stale compression metadata', () => { + const headers = new Headers({ + 'content-type': 'text/event-stream', + 'content-encoding': 'br', + 'content-length': '123', + 'transfer-encoding': 'chunked', + connection: 'keep-alive', + 'cache-control': 'no-cache', + }); + assert.deepEqual(decodedUpstreamResponseHeaders(headers), { + 'cache-control': 'no-cache', + 'content-type': 'text/event-stream', + }); +}); diff --git a/llm-providers.mjs b/llm-providers.mjs index 78c2139..8e46cb5 100644 --- a/llm-providers.mjs +++ b/llm-providers.mjs @@ -4,6 +4,15 @@ import os from 'node:os'; import path from 'node:path'; import { spawn, spawnSync } from 'node:child_process'; import { Agent, fetch as undiciFetch } from 'undici'; +import { + deepseekDisableThinkingEnabled, + isMoonshotApiUrl, + moonshotToolSchemaCompatEnabled, + resolveDeepseekNoThinkProxyBaseUrl, + resolveMoonshotCompatProxyBaseUrl, +} from './deepseek-no-think-proxy.mjs'; + +export const MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID = 'custom_memind_deepseek_no_think'; export const CUSTOM_PROVIDER_ID = '__custom__'; @@ -765,6 +774,9 @@ async function upsertCustomProviderOnGoosed(apiTarget, apiSecret, profile, fetch requires_auth: true, ...(profile.basePath ? { base_path: profile.basePath } : {}), ...(headers ? { headers } : {}), + ...(typeof profile.preservesThinking === 'boolean' + ? { preserves_thinking: profile.preservesThinking } + : {}), }; if (profile.goosedProviderId) { @@ -775,11 +787,12 @@ async function upsertCustomProviderOnGoosed(apiTarget, apiSecret, profile, fetch { method: 'PUT', body: JSON.stringify(body) }, fetchImpl, ); - if (!upstream.ok) { + if (upstream.ok) return profile.goosedProviderId; + // First-time ensure: fixed ids may 404 until created. + if (upstream.status !== 404) { const text = await upstream.text().catch(() => ''); throw new Error(`更新 TKMind Agent 自定义 provider 失败: ${text || upstream.status}`); } - return profile.goosedProviderId; } const upstream = await goosedApiFetch( @@ -794,7 +807,51 @@ async function upsertCustomProviderOnGoosed(apiTarget, apiSecret, profile, fetch throw new Error(`创建 TKMind Agent 自定义 provider 失败: ${text || upstream.status}`); } const data = await upstream.json(); - return data.provider_name; + return data.provider_name ?? profile.goosedProviderId; +} + +async function syncDeepseekNoThinkProfileToGoosed(apiTarget, apiSecret, profile, fetchImpl) { + const catalogItem = catalogById.custom_deepseek; + const models = [ + ...new Set([ + ...(Array.isArray(catalogItem?.models) ? catalogItem.models : []), + profile.defaultModel, + ...(Array.isArray(profile.models) ? profile.models : []), + ].filter(Boolean).map((item) => String(item).trim()).filter(Boolean)), + ]; + const goosedProviderId = await upsertCustomProviderOnGoosed( + apiTarget, + apiSecret, + { + name: 'memind_deepseek_no_think', + goosedProviderId: MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID, + apiUrl: resolveDeepseekNoThinkProxyBaseUrl(), + apiKey: profile.apiKey, + models, + defaultModel: profile.defaultModel, + engine: 'openai', + preservesThinking: false, + }, + fetchImpl, + ); + // Keep builtin secret for direct DeepSeek callers / admin tests. + if (catalogItem?.apiKeyEnv) { + await writeGoosedConfig( + apiTarget, + apiSecret, + catalogItem.apiKeyEnv, + profile.apiKey, + true, + fetchImpl, + ); + } + await writeGoosedConfig(apiTarget, apiSecret, 'GOOSE_PROVIDER', goosedProviderId, false, fetchImpl); + await writeGoosedConfig(apiTarget, apiSecret, 'GOOSE_MODEL', profile.defaultModel, false, fetchImpl); + await writeGoosedConfig(apiTarget, apiSecret, 'TKMIND_PROVIDER', goosedProviderId, false, fetchImpl); + await writeGoosedConfig(apiTarget, apiSecret, 'TKMIND_MODEL', profile.defaultModel, false, fetchImpl); + // Avoid DeepSeek thinking mode when the proxy is unavailable for a turn. + await writeGoosedConfig(apiTarget, apiSecret, 'GOOSE_THINKING_EFFORT', 'off', false, fetchImpl); + return goosedProviderId; } async function removeCustomProviderOnGoosed(apiTarget, apiSecret, goosedProviderId, fetchImpl) { @@ -813,6 +870,12 @@ async function removeCustomProviderOnGoosed(apiTarget, apiSecret, goosedProvider } async function syncBuiltinProfileToGoosed(apiTarget, apiSecret, profile, fetchImpl) { + if ( + profile.providerId === 'custom_deepseek' + && deepseekDisableThinkingEnabled() + ) { + return syncDeepseekNoThinkProfileToGoosed(apiTarget, apiSecret, profile, fetchImpl); + } const catalogItem = catalogById[profile.providerId]; if (!catalogItem?.apiKeyEnv) { throw new Error(`未知内置 provider: ${profile.providerId}`); @@ -858,13 +921,22 @@ async function syncBuiltinProfileToGoosed(apiTarget, apiSecret, profile, fetchIm fetchImpl, ); } - export async function syncProfileToGoosed(apiTarget, apiSecret, profile, fetchImpl = undiciFetch) { if (profile.providerKind === 'custom') { + const goosedProfile = ( + moonshotToolSchemaCompatEnabled() + && isMoonshotApiUrl(profile.apiUrl) + ) + ? { + ...profile, + apiUrl: resolveMoonshotCompatProxyBaseUrl(), + preservesThinking: false, + } + : profile; const goosedProviderId = await upsertCustomProviderOnGoosed( apiTarget, apiSecret, - profile, + goosedProfile, fetchImpl, ); await writeGoosedConfig( @@ -902,8 +974,13 @@ export async function syncProfileToGoosed(apiTarget, apiSecret, profile, fetchIm return goosedProviderId; } - await syncBuiltinProfileToGoosed(apiTarget, apiSecret, profile, fetchImpl); - return profile.providerId; + const syncedProviderId = await syncBuiltinProfileToGoosed( + apiTarget, + apiSecret, + profile, + fetchImpl, + ); + return syncedProviderId || profile.providerId; } function isCustomPayload(payload) { diff --git a/llm-providers.test.mjs b/llm-providers.test.mjs index 656f41f..3646545 100644 --- a/llm-providers.test.mjs +++ b/llm-providers.test.mjs @@ -3,6 +3,7 @@ import test from 'node:test'; import { createLlmProviderService, CUSTOM_PROVIDER_ID, + MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID, buildExecutorLaunchPlan, decryptSecret, encryptSecret, @@ -753,6 +754,136 @@ test('syncProfileToGoosed writes provider, model and secret keys for builtin', a ); }); +test('syncProfileToGoosed routes local DeepSeek through the no-thinking proxy', async () => { + const previousProfile = process.env.MEMIND_RUNTIME_PROFILE; + const previousDisableThinking = process.env.MEMIND_DEEPSEEK_DISABLE_THINKING; + const previousProxyBase = process.env.MEMIND_DEEPSEEK_NO_THINK_BASE_URL; + const previousGateway = process.env.MEMIND_GOOSED_HOST_GATEWAY; + process.env.MEMIND_RUNTIME_PROFILE = 'local'; + delete process.env.MEMIND_DEEPSEEK_DISABLE_THINKING; + delete process.env.MEMIND_DEEPSEEK_NO_THINK_BASE_URL; + process.env.MEMIND_GOOSED_HOST_GATEWAY = 'host.lima.internal'; + + try { + const calls = []; + const mockFetch = async (url, init = {}) => { + const href = String(url); + const body = init.body ? JSON.parse(init.body) : null; + calls.push({ href, method: init.method ?? 'GET', body }); + if ( + init.method === 'PUT' + && href.includes(`/config/custom-providers/${MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID}`) + ) { + return { ok: false, status: 404, text: async () => '' }; + } + if (init.method === 'POST' && href.endsWith('/config/custom-providers')) { + return { + ok: true, + json: async () => ({ provider_name: MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID }), + text: async () => '', + }; + } + return { ok: true, status: 200, text: async () => '' }; + }; + + const providerId = await syncProfileToGoosed( + 'https://127.0.0.1:18006', + 'secret', + { + providerKind: 'builtin', + providerId: 'custom_deepseek', + defaultModel: 'deepseek-v4-pro', + apiKey: 'sk-test', + }, + mockFetch, + ); + + assert.equal(providerId, MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID); + const createCall = calls.find( + (call) => call.method === 'POST' && call.href.endsWith('/config/custom-providers'), + ); + assert.equal(createCall?.body?.api_url, 'http://host.lima.internal:18036/v1'); + assert.equal(createCall?.body?.preserves_thinking, false); + const providerWrite = calls.find( + (call) => call.href.includes('/config/upsert') && call.body?.key === 'GOOSE_PROVIDER', + ); + assert.equal(providerWrite?.body?.value, MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID); + } finally { + if (previousProfile == null) delete process.env.MEMIND_RUNTIME_PROFILE; + else process.env.MEMIND_RUNTIME_PROFILE = previousProfile; + if (previousDisableThinking == null) delete process.env.MEMIND_DEEPSEEK_DISABLE_THINKING; + else process.env.MEMIND_DEEPSEEK_DISABLE_THINKING = previousDisableThinking; + if (previousProxyBase == null) delete process.env.MEMIND_DEEPSEEK_NO_THINK_BASE_URL; + else process.env.MEMIND_DEEPSEEK_NO_THINK_BASE_URL = previousProxyBase; + if (previousGateway == null) delete process.env.MEMIND_GOOSED_HOST_GATEWAY; + else process.env.MEMIND_GOOSED_HOST_GATEWAY = previousGateway; + } +}); + +test('syncProfileToGoosed routes local Moonshot profiles through schema compat proxy', async () => { + const previousProfile = process.env.MEMIND_RUNTIME_PROFILE; + const previousCompat = process.env.MEMIND_MOONSHOT_TOOL_SCHEMA_COMPAT; + const previousBase = process.env.MEMIND_MOONSHOT_COMPAT_BASE_URL; + const previousGateway = process.env.MEMIND_GOOSED_HOST_GATEWAY; + process.env.MEMIND_RUNTIME_PROFILE = 'split-service'; + delete process.env.MEMIND_MOONSHOT_TOOL_SCHEMA_COMPAT; + delete process.env.MEMIND_MOONSHOT_COMPAT_BASE_URL; + process.env.MEMIND_GOOSED_HOST_GATEWAY = 'host.lima.internal'; + + try { + const calls = []; + const mockFetch = async (url, init = {}) => { + const href = String(url); + const body = init.body ? JSON.parse(init.body) : null; + calls.push({ href, method: init.method ?? 'GET', body }); + if ( + init.method === 'PUT' + && href.includes('/config/custom-providers/custom_kimi') + ) { + return { ok: true, status: 200, text: async () => '' }; + } + return { ok: true, status: 200, text: async () => '' }; + }; + + const providerId = await syncProfileToGoosed( + 'https://127.0.0.1:18006', + 'secret', + { + providerKind: 'custom', + providerId: 'custom_kimi', + goosedProviderId: 'custom_kimi', + name: 'kimi', + apiUrl: 'https://api.moonshot.cn/v1', + apiKey: 'sk-test', + models: ['kimi-k2.6', 'kimi-k3'], + defaultModel: 'kimi-k2.6', + engine: 'openai', + }, + mockFetch, + ); + + assert.equal(providerId, 'custom_kimi'); + const updateCall = calls.find( + (call) => call.method === 'PUT' + && call.href.includes('/config/custom-providers/custom_kimi'), + ); + assert.equal( + updateCall?.body?.api_url, + 'http://host.lima.internal:18036/moonshot/v1', + ); + assert.equal(updateCall?.body?.preserves_thinking, false); + } finally { + if (previousProfile == null) delete process.env.MEMIND_RUNTIME_PROFILE; + else process.env.MEMIND_RUNTIME_PROFILE = previousProfile; + if (previousCompat == null) delete process.env.MEMIND_MOONSHOT_TOOL_SCHEMA_COMPAT; + else process.env.MEMIND_MOONSHOT_TOOL_SCHEMA_COMPAT = previousCompat; + if (previousBase == null) delete process.env.MEMIND_MOONSHOT_COMPAT_BASE_URL; + else process.env.MEMIND_MOONSHOT_COMPAT_BASE_URL = previousBase; + if (previousGateway == null) delete process.env.MEMIND_GOOSED_HOST_GATEWAY; + else process.env.MEMIND_GOOSED_HOST_GATEWAY = previousGateway; + } +}); + test('syncSelectedToGoosed writes selected provider to every configured target', async () => { const encrypted = encryptSecret('sk-test', 'unit-test-secret'); const row = { diff --git a/scripts/dev-core.mjs b/scripts/dev-core.mjs index e00455d..4f6d195 100644 --- a/scripts/dev-core.mjs +++ b/scripts/dev-core.mjs @@ -8,14 +8,28 @@ import { describeMemindRuntimeProfile, loadMemindEnvFiles, } from './memind-runtime-profile.mjs'; +import { + deepseekDisableThinkingEnabled, + resolveDeepseekNoThinkListenPort, +} from '../deepseek-no-think-proxy.mjs'; const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..'); const opsDir = path.join(root, 'ops'); + +// Load .env before reading ports — otherwise ADMIN_PORT defaults to 8082 and +// split-service MindSpace (also on 8082) gets killed by freePort(). +loadMemindEnvFiles(root); +applyMemindRuntimeProfile({ rootDir: root }); +console.log(`==> Runtime profile: ${describeMemindRuntimeProfile()}`); + const portalPort = Number(process.env.H5_PORT ?? 8081); const vitePort = Number(process.env.VITE_PORT ?? 5173); const plazaPort = Number(process.env.PLAZA_PORT ?? 3001); const opsPort = Number(process.env.OPS_PORT ?? 3002); const adminPort = Number(process.env.ADMIN_PORT ?? 8082); +const deepseekNoThinkEnabled = deepseekDisableThinkingEnabled(); +const deepseekNoThinkPort = resolveDeepseekNoThinkListenPort(); +const deepseekNoThinkUrl = `http://127.0.0.1:${deepseekNoThinkPort}`; const portalUrl = `http://127.0.0.1:${portalPort}`; const adminUrl = `http://127.0.0.1:${adminPort}`; const localMindSpacePublicBase = `http://127.0.0.1:${vitePort}`; @@ -23,10 +37,6 @@ const plazaPublicBase = ( process.env.PLAZA_PUBLIC_BASE ?? `http://127.0.0.1:${plazaPort}` ).replace(/\/$/, ''); -loadMemindEnvFiles(root); -applyMemindRuntimeProfile({ rootDir: root }); -console.log(`==> Runtime profile: ${describeMemindRuntimeProfile()}`); - function freePort(port) { try { execSync(`lsof -ti TCP:${port} -sTCP:LISTEN | xargs kill -9`, { stdio: 'ignore' }); @@ -55,6 +65,7 @@ let server; let admin; let ops; let vite; +let deepseekNoThinkProxy; let stopping = false; function shutdown(code = 0) { @@ -64,6 +75,7 @@ function shutdown(code = 0) { admin?.kill('SIGTERM'); ops?.kill('SIGTERM'); vite?.kill('SIGTERM'); + deepseekNoThinkProxy?.kill('SIGTERM'); setTimeout(() => process.exit(code), 300); } @@ -105,17 +117,39 @@ const opsEnv = { VITE_MINDSPACE_BASE: mindSpacePublicBase, }; -console.log(`==> 清理端口 ${portalPort} / ${adminPort} / ${vitePort} / ${opsPort}...`); +const managedPorts = [ + portalPort, + adminPort, + vitePort, + opsPort, + ...(deepseekNoThinkEnabled ? [deepseekNoThinkPort] : []), +]; +console.log(`==> 清理端口 ${managedPorts.join(' / ')}...`); freePort(portalPort); freePort(adminPort); freePort(vitePort); freePort(opsPort); +if (deepseekNoThinkEnabled) freePort(deepseekNoThinkPort); await new Promise((resolve) => setTimeout(resolve, 500)); -console.log('==> 启动 portal (server.mjs)...'); -server = spawnChild('node', [path.join(root, 'server.mjs')], 'portal'); - try { + if (deepseekNoThinkEnabled) { + console.log('==> 启动 DeepSeek no-thinking proxy...'); + deepseekNoThinkProxy = spawnChild( + 'node', + [path.join(root, 'deepseek-no-think-proxy.mjs')], + 'deepseek-no-think', + ); + await waitFor( + deepseekNoThinkUrl, + async (url) => (await fetch(`${url}/health`)).ok, + 'DeepSeek no-thinking proxy', + ); + console.log(`==> DeepSeek no-thinking proxy 就绪: ${deepseekNoThinkUrl}`); + } + + console.log('==> 启动 portal (server.mjs)...'); + server = spawnChild('node', [path.join(root, 'server.mjs')], 'portal'); await waitFor(portalUrl, async (url) => (await fetch(`${url}/auth/status`)).ok, 'Portal'); console.log(`==> Portal 就绪: ${portalUrl}`); @@ -156,6 +190,9 @@ try { console.log(` Ops 审核后台 http://127.0.0.1:${opsPort}/ops/`); console.log(` API / Portal ${portalUrl}`); console.log(` memind_adm ${adminUrl}`); + if (deepseekNoThinkEnabled) { + console.log(` DeepSeek Proxy ${deepseekNoThinkUrl}`); + } console.log(''); console.log('Plaza 已独立出去：请另开终端运行 pnpm dev:plaza 或 pnpm start:plaza'); } catch (err) { diff --git a/server/portal-gateway-services-bootstrap.mjs b/server/portal-gateway-services-bootstrap.mjs index f975d97..4a68abe 100644 --- a/server/portal-gateway-services-bootstrap.mjs +++ b/server/portal-gateway-services-bootstrap.mjs @@ -360,6 +360,27 @@ export function bootstrapPortalGatewayServices({ isSessionExternallyBusy: ({ sessionId }) => isSessionPageDeliveryActive(sessionId), validateRunDeliverables, + cancelSessionOnRetry: async ({ + sessionId, + requestId, + }) => { + if (isDirectChatSessionId(sessionId)) { + return { cancelled: false, skipped: true }; + } + const target = + await tkmindProxy.resolveTarget(sessionId); + const sessionApiFetch = (pathname, init) => + tkmindProxy.apiFetchTo( + target, + pathname, + init, + ); + return cancelSessionActiveRequest( + sessionApiFetch, + sessionId, + requestId, + ); + }, quiesceSessionOnTerminal: async ({ sessionId, requestId, diff --git a/session-reply-wait.mjs b/session-reply-wait.mjs index 58cee4c..aa06c2c 100644 --- a/session-reply-wait.mjs +++ b/session-reply-wait.mjs @@ -33,6 +33,19 @@ function messageContentItems(event) { return candidates.find((content) => Array.isArray(content)) ?? []; } +export function classifySessionProviderErrorMessage(message) { + const normalized = String(message ?? '').trim(); + if (!normalized) return 'SESSION_REPLY_ERROR'; + if (/tool_calls|tool_call_id|insufficient tool messages/i.test(normalized)) { + return 'SESSION_TOOL_HISTORY_POISONED'; + } + // DeepSeek/Kimi thinking mode: assistant tool-call turns must replay reasoning_content. + if (/reasoning_content/i.test(normalized)) { + return 'SESSION_REASONING_CONTENT_POISONED'; + } + return 'SESSION_REPLY_ERROR'; +} + function providerReplyError(event) { if (event?.type !== 'Message') return null; const text = messageContentItems(event) @@ -46,9 +59,7 @@ function providerReplyError(event) { .replace(/\n\nPlease retry if you think this is a transient or recoverable error\.?\s*$/i, '') .trim(); const error = new Error(normalized || 'session reply failed'); - error.code = /tool_calls|tool_call_id|insufficient tool messages/i.test(normalized) - ? 'SESSION_TOOL_HISTORY_POISONED' - : 'SESSION_REPLY_ERROR'; + error.code = classifySessionProviderErrorMessage(normalized); return error; } @@ -162,7 +173,7 @@ export async function consumeSessionEventsUntilFinish( onEvent?.(event); if (event.type === 'Error') { const err = new Error(String(event.error ?? 'session reply failed')); - err.code = 'SESSION_REPLY_ERROR'; + err.code = classifySessionProviderErrorMessage(err.message); err.retryable = false; throw err; } diff --git a/session-reply-wait.test.mjs b/session-reply-wait.test.mjs index 2c9d57a..857af54 100644 --- a/session-reply-wait.test.mjs +++ b/session-reply-wait.test.mjs @@ -76,6 +76,38 @@ test('consumeSessionEventsUntilFinish records a successful raster generate_image assert.equal(result.toolEvidence.generateImage.jobId, 'job-1'); }); +test('consumeSessionEventsUntilFinish surfaces reasoning_content protocol errors before Finish', async () => { + const frames = [ + `data: ${JSON.stringify({ + type: 'Message', + request_id: 'req-1', + message: { + role: 'assistant', + content: [{ + type: 'text', + text: 'Ran into this error: Bad request (400): The reasoning_content in the thinking mode must be passed back to the API.\n\nPlease retry if you think this is a transient or recoverable error.', + }], + }, + })}\n\n`, + 'data: {"type":"Finish","request_id":"req-1"}\n\n', + ]; + const stream = new ReadableStream({ + start(controller) { + for (const frame of frames) controller.enqueue(new TextEncoder().encode(frame)); + controller.close(); + }, + }); + + await assert.rejects( + consumeSessionEventsUntilFinish(stream, { requestId: 'req-1', timeoutMs: 5000 }), + (error) => { + assert.equal(error.code, 'SESSION_REASONING_CONTENT_POISONED'); + assert.match(error.message, /reasoning_content/); + return true; + }, + ); +}); + test('consumeSessionEventsUntilFinish surfaces provider tool history errors before Finish', async () => { const frames = [ `data: ${JSON.stringify({ diff --git a/src/hooks/useTKMindChat.ts b/src/hooks/useTKMindChat.ts index 926a1f1..02a87f6 100644 --- a/src/hooks/useTKMindChat.ts +++ b/src/hooks/useTKMindChat.ts @@ -88,6 +88,20 @@ const INSUFFICIENT_BALANCE_NOTICE = '余额不足，请充值后继续使用'; const REPLY_RECOVERY_SYNC_DELAYS_MS = [1500, 5000, 12000]; const FINISH_SYNC_RETRY_DELAYS_MS = [500, 1500, 3000]; const ACTIVE_REQUEST_MISSING_GRACE_MS = 2500; + +function createAgentRunFailedError(message: string) { + const error = new Error(message) as Error & { code: string }; + error.code = 'AGENT_RUN_FAILED'; + return error; +} + +function errorCode(error: unknown) { + if (error instanceof ApiError) return error.code; + if (error && typeof error === 'object' && 'code' in error) { + return String((error as { code?: unknown }).code ?? ''); + } + return ''; +} const DIRECT_CHAT_SESSION_PREFIX = 'h5direct_'; export { INSUFFICIENT_BALANCE_NOTICE }; @@ -198,7 +212,7 @@ async function waitForAgentRunWithDirectChatPreview( return; } if (run.status === 'failed') { - settle(() => reject(new Error(run.error || '后台任务失败，请稍后重试'))); + settle(() => reject(createAgentRunFailedError(run.error || '后台任务失败，请稍后重试'))); } }; @@ -235,7 +249,7 @@ async function waitForAgentRunWithDirectChatPreview( return; } if (run.status === 'failed') { - settle(() => reject(new Error(run.error || '后台任务失败，请稍后重试'))); + settle(() => reject(createAgentRunFailedError(run.error || '后台任务失败，请稍后重试'))); return; } settle(() => @@ -1671,7 +1685,7 @@ export function useTKMindChat( shouldKeepStreamingAfterRunError( undefined, err instanceof Error ? err.message : String(err), - err instanceof ApiError ? err.code : '', + errorCode(err), ) ) { // Goose may report its session concurrency guard as a failed run diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 34d13e5..b503981 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -1896,28 +1896,43 @@ export function createTkmindProxy({ requestId, timeoutMs, }); - const replyResponse = await apiFetch( - target, - apiSecret, - `/sessions/${encodeURIComponent(sessionId)}/reply`, - { - method: 'POST', - body: JSON.stringify(body), - }, + // Prevent unhandled rejection from killing the Portal process when reply + // setup fails before we await Finish (e.g. provider Error frames arrive later). + const trackedFinish = finishPromise.then( + (value) => ({ ok: true, value }), + (error) => ({ ok: false, error }), ); - if (!replyResponse.ok) { - const text = await replyResponse.text().catch(() => ''); - throw new Error(text || `发送失败 (${replyResponse.status})`); + try { + const replyResponse = await apiFetch( + target, + apiSecret, + `/sessions/${encodeURIComponent(sessionId)}/reply`, + { + method: 'POST', + body: JSON.stringify(body), + }, + ); + if (!replyResponse.ok) { + const text = await replyResponse.text().catch(() => ''); + eventsResponse.body?.cancel?.().catch?.(() => {}); + throw new Error(text || `发送失败 (${replyResponse.status})`); + } + replyResponse.body?.cancel?.().catch?.(() => {}); + const finishResult = await trackedFinish; + if (!finishResult.ok) throw finishResult.error; + const finish = finishResult.value; + const tokenState = await resolveSessionBillingTokenState(sessionId, finish.tokenState); + if (tokenState && userAuth.billSessionUsage) { + await userAuth + .billSessionUsage(userId, sessionId, tokenState, requestId) + .catch(() => {}); + } + return { ok: true, ...finish, tokenState }; + } catch (err) { + eventsResponse.body?.cancel?.().catch?.(() => {}); + await trackedFinish.catch?.(() => {}); + throw err; } - replyResponse.body?.cancel?.().catch?.(() => {}); - const finish = await finishPromise; - const tokenState = await resolveSessionBillingTokenState(sessionId, finish.tokenState); - if (tokenState && userAuth.billSessionUsage) { - await userAuth - .billSessionUsage(userId, sessionId, tokenState, requestId) - .catch(() => {}); - } - return { ok: true, ...finish, tokenState }; } const requireUser = async (req, res, next) => {