diff --git a/apps/desktop-client/src/main/adapters/kimi.test.ts b/apps/desktop-client/src/main/adapters/kimi.test.ts index 7181975..4ad19f2 100644 --- a/apps/desktop-client/src/main/adapters/kimi.test.ts +++ b/apps/desktop-client/src/main/adapters/kimi.test.ts @@ -1,17 +1,40 @@ import { describe, expect, it } from 'vitest' -import { __kimiTestUtils } from './kimi' +import type { AdapterContext } from './base' +import { __kimiTestUtils, kimiAdapter } from './kimi' const { buildSourceItem, classifyKimiSources, + decodeKimiConnectFrames, extractQuestionText, isKimiAnswerComplete, isKimiPromotionalAnswerNoise, mergeKimiSourceLinksIntoContentSnapshot, normalizeUrl, + shouldReturnStableKimiAnswer, + summarizeKimiStreamEvents, } = __kimiTestUtils +function encodeConnectFrames(frames: Array<{ flags?: number; value: unknown }>): Uint8Array { + const encoder = new TextEncoder() + const encoded = frames.map(({ flags = 0, value }) => { + const payload = encoder.encode(JSON.stringify(value)) + const frame = new Uint8Array(payload.length + 5) + frame[0] = flags + new DataView(frame.buffer).setUint32(1, payload.length, false) + frame.set(payload, 5) + return frame + }) + const result = new Uint8Array(encoded.reduce((total, frame) => total + frame.length, 0)) + let offset = 0 + for (const frame of encoded) { + result.set(frame, offset) + offset += frame.length + } + return result +} + function buildSnapshot( overrides: Partial[0]>, ): Parameters[0] { @@ -42,6 +65,261 @@ function buildSnapshot( } describe('kimi adapter helpers', () => { + it('decodes Kimi Connect JSON frames including the terminal envelope', () => { + const decoded = decodeKimiConnectFrames( + encodeConnectFrames([ + { value: { heartbeat: {} } }, + { value: { op: 'set', block: { id: 'answer', text: { content: '流式答案' } } } }, + { flags: 2, value: {} }, + ]), + ) + + expect(decoded.pendingBytes).toBe(0) + expect(decoded.frames).toEqual([ + { flags: 0, value: { heartbeat: {} } }, + { + flags: 0, + value: { op: 'set', block: { id: 'answer', text: { content: '流式答案' } } }, + }, + { flags: 2, value: {} }, + ]) + }) + + it('summarizes Kimi stream set/append events into answer, reasoning, and sources', () => { + const summary = summarizeKimiStreamEvents([ + { op: 'set', chat: { id: 'chat-api' } }, + { + op: 'set', + mask: 'message', + message: { + id: 'assistant-api', + role: 'assistant', + status: 'MESSAGE_STATUS_GENERATING', + }, + }, + { + op: 'set', + mask: 'block.reasoning.content', + block: { + id: 'reasoning', + messageId: 'assistant-api', + reasoning: { content: '先分析' }, + }, + }, + { + op: 'append', + mask: 'block.reasoning.content', + block: { + id: 'reasoning', + messageId: 'assistant-api', + reasoning: { content: '，再判断。' }, + }, + }, + { + op: 'set', + mask: 'block.text.content', + block: { + id: 'answer', + messageId: 'assistant-api', + text: { content: '官方流式' }, + }, + }, + { + op: 'append', + mask: 'block.text.content', + block: { + id: 'answer', + messageId: 'assistant-api', + text: { content: '答案。' }, + }, + }, + { + op: 'set', + block: { + id: 'search', + messageId: 'assistant-api', + search: { + results: [ + { + url: 'https://example.com/source', + title: '示例来源', + site_name: 'Example', + }, + ], + }, + }, + }, + { + op: 'set', + mask: 'message.status', + message: { id: 'assistant-api', status: 'MESSAGE_STATUS_COMPLETED' }, + }, + ]) + + expect(summary).toMatchObject({ + answer: '官方流式答案。', + reasoning: '先分析，再判断。', + conversationID: 'chat-api', + providerRequestID: 'assistant-api', + completed: true, + }) + expect(summary.searchResults).toEqual([ + expect.objectContaining({ + url: 'https://example.com/source', + title: '示例来源', + site_name: 'Example', + }), + ]) + }) + + it('turns the K3 overload stream into a submit-worthy capacity notice', () => { + const summary = summarizeKimiStreamEvents([ + { + op: 'set', + mask: 'message', + message: { id: 'assistant-overloaded', role: 'assistant' }, + }, + { + op: 'set', + mask: 'block.exception', + block: { + id: 'exception', + messageId: 'assistant-overloaded', + exception: { error: { reason: 'REASON_COMPLETION_OVERLOADED' } }, + }, + }, + { + op: 'set', + block: { + id: 'tip', + messageId: 'assistant-overloaded', + text: { tips: 'Kimi 遇到了一些问题，暂停了这个任务。' }, + }, + }, + ]) + + expect(summary.answer).toBeNull() + expect(summary.capacityNotice).toBe('Kimi 遇到了一些问题，暂停了这个任务。') + expect(summary.errorReason).toBe('REASON_COMPLETION_OVERLOADED') + }) + + it('does not treat a completed system frame as a completed assistant answer', () => { + const summary = summarizeKimiStreamEvents([ + { + op: 'set', + message: { id: 'system', role: 'system', status: 'MESSAGE_STATUS_COMPLETED' }, + }, + { + op: 'set', + message: { + id: 'assistant-partial', + role: 'assistant', + status: 'MESSAGE_STATUS_GENERATING', + }, + }, + { + op: 'append', + block: { + id: 'answer', + messageId: 'assistant-partial', + text: { content: '尚未完成的回答' }, + }, + }, + ]) + + expect(summary.answer).toBe('尚未完成的回答') + expect(summary.completed).toBe(false) + }) + + it('treats the new K3 page capacity notice as a stable RPA result', () => { + expect( + shouldReturnStableKimiAnswer( + buildSnapshot({ + answerText: null, + answerBlocks: [], + capacityNotice: 'Kimi 遇到了一些问题，暂停了这个任务。', + }), + null, + Date.now(), + ), + ).toBe(true) + }) + + it('uses the K3 stream API result without touching the page RPA controls', async () => { + let evaluateCalls = 0 + let locatorCalls = 0 + const progress: string[] = [] + const page = { + url: () => 'https://www.kimi.com/', + waitForLoadState: async () => undefined, + evaluate: async () => { + evaluateCalls += 1 + if (evaluateCalls === 1) { + return buildSnapshot({ + currentModelLabel: 'K3', + answerText: null, + answerBlocks: [], + questionMatched: false, + candidateAnswerCount: 0, + signature: null, + }) + } + return { + ok: true, + status: 200, + contentType: 'application/connect+json', + frameCount: 5, + pendingBytes: 0, + elapsedMs: 120, + events: [ + { op: 'set', chat: { id: 'chat-api' } }, + { + op: 'set', + mask: 'message', + message: { id: 'assistant-api', role: 'assistant' }, + }, + { + op: 'set', + mask: 'block.text.content', + block: { + id: 'answer', + messageId: 'assistant-api', + text: { content: 'K3 API 完整答案' }, + }, + }, + { + op: 'set', + mask: 'message.status', + message: { id: 'assistant-api', status: 'MESSAGE_STATUS_COMPLETED' }, + }, + ], + } + }, + locator: () => { + locatorCalls += 1 + throw new Error('page RPA should not run') + }, + } + const context = { + playwright: { page, browser: {} }, + signal: new AbortController().signal, + reportProgress: (stage: string) => progress.push(stage), + } as unknown as AdapterContext + + const result = await kimiAdapter.query(context, { question_text: '测试问题' }) + + expect(result.status).toBe('succeeded') + expect(result.payload?.answer).toBe('K3 API 完整答案') + expect(result.payload?.provider_model).toBe('K3') + expect(result.payload?.response_source).toBe('api_stream') + expect((result.payload?.raw_response_json as Record).response_source).toBe( + 'api_stream', + ) + expect(progress).toContain('kimi.api_query') + expect(progress).not.toContain('kimi.mode_switch') + expect(locatorCalls).toBe(0) + }) + it('does not use task title as monitoring question text', () => { expect(extractQuestionText({ question_text: '幽灵门轨道长度定制' })).toBe('幽灵门轨道长度定制') expect(() => extractQuestionText({ title: '监控任务 · Kimi' })).toThrow( diff --git a/apps/desktop-client/src/main/adapters/kimi.ts b/apps/desktop-client/src/main/adapters/kimi.ts index fbae454..e2321f1 100644 --- a/apps/desktop-client/src/main/adapters/kimi.ts +++ b/apps/desktop-client/src/main/adapters/kimi.ts @@ -5,14 +5,15 @@ import type { MonitorAdapter } from './base' import { normalizeText } from './common' const KIMI_BOOTSTRAP_URL = 'https://www.kimi.com/' +const KIMI_STREAM_ENDPOINT = 'https://www.kimi.com/apiv2/kimi.gateway.chat.v1.ChatService/Chat' const KIMI_PAGE_READY_TIMEOUT_MS = 20_000 const KIMI_MODE_SWITCH_TIMEOUT_MS = 8_000 -// kimi 算力紧张时回答生成慢（思考模式尤甚），查询超时保留 180s 兜底； -// 现在默认切快速模式，正常情况下答案稳定后轮询会提前返回。 const KIMI_QUERY_TIMEOUT_MS = 180_000 const KIMI_QUERY_POLL_INTERVAL_MS = 1_200 const KIMI_STABLE_POLLS_REQUIRED = 5 const KIMI_INCOMPLETE_ANSWER_GRACE_MS = 18_000 +const KIMI_STREAM_MAX_FRAME_LENGTH = 4 * 1024 * 1024 +const KIMI_STREAM_MAX_FRAME_COUNT = 4_096 const KIMI_EDITOR_SELECTOR = 'div[role="textbox"].chat-input-editor, .chat-input-editor[role="textbox"], [role="textbox"][class*="chat-input-editor"]' const KIMI_SEND_BUTTON_SELECTOR = '.send-button-container' @@ -22,11 +23,10 @@ type KimiModeTarget = { keyword: string } -// Kimi 网页端模型切换器的两档模式。思考档切换逻辑保留备用； -// 采集现在默认切到快速档，要回退思考档时把 KIMI_TARGET_MODE 指回 KIMI_THINKING_MODE 即可。 +const KIMI_K3_MODE: KimiModeTarget = { label: 'K3', keyword: 'K3' } const KIMI_THINKING_MODE: KimiModeTarget = { label: 'K2.6 思考', keyword: '思考' } const KIMI_FAST_MODE: KimiModeTarget = { label: 'K2.6 快速', keyword: '快速' } -const KIMI_TARGET_MODE: KimiModeTarget = KIMI_FAST_MODE +const KIMI_TARGET_MODE: KimiModeTarget = KIMI_K3_MODE const KIMI_REDIRECT_QUERY_KEYS = [ 'url', 'u', @@ -132,6 +132,42 @@ type KimiQueryWaitResult = elapsedMs: number } +type KimiConnectFrame = { + flags: number + value: unknown +} + +type KimiConnectDecodeResult = { + frames: KimiConnectFrame[] + pendingBytes: number + error: string | null +} + +type KimiStreamTransportResult = { + ok: boolean + status: number | null + contentType: string | null + frameCount: number + pendingBytes: number + elapsedMs: number + events: unknown[] + error: string | null + detail: string | null +} + +type KimiStreamSummary = { + answer: string | null + answerBlocks: string[] + reasoning: string | null + reasoningBlocks: string[] + capacityNotice: string | null + conversationID: string | null + providerRequestID: string | null + completed: boolean + errorReason: string | null + searchResults: MonitoringSourceItem[] +} + function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value) } @@ -145,6 +181,48 @@ function normalizeOptionalString(value: unknown): string | null { return trimmed ? trimmed : null } +function decodeKimiConnectFrames(input: Uint8Array): KimiConnectDecodeResult { + const frames: KimiConnectFrame[] = [] + const decoder = new TextDecoder('utf-8') + let offset = 0 + + while (offset + 5 <= input.length) { + const flags = input[offset] ?? 0 + const length = new DataView(input.buffer, input.byteOffset + offset + 1, 4).getUint32(0, false) + if (length > KIMI_STREAM_MAX_FRAME_LENGTH) { + return { + frames, + pendingBytes: input.length - offset, + error: 'kimi_stream_frame_too_large', + } + } + if (offset + 5 + length > input.length) { + break + } + + const payload = input.subarray(offset + 5, offset + 5 + length) + try { + frames.push({ + flags, + value: JSON.parse(decoder.decode(payload)) as unknown, + }) + } catch { + return { + frames, + pendingBytes: input.length - offset, + error: 'kimi_stream_frame_invalid_json', + } + } + offset += 5 + length + } + + return { + frames, + pendingBytes: input.length - offset, + error: null, + } +} + function decodeUrlCandidate(value: string): string { let decoded = value.trim() for (let index = 0; index < 3; index += 1) { @@ -276,6 +354,293 @@ function dedupeSourceItems(items: MonitoringSourceItem[]): MonitoringSourceItem[ return Array.from(keyed.values()) } +function mergeKimiStreamText( + current: string | undefined, + next: unknown, + operation: string | null, +): string | undefined { + if (typeof next !== 'string' || !next) { + return current + } + if (!current) { + return next + } + if (operation === 'append') { + return `${current}${next}` + } + if (next.startsWith(current)) { + return next + } + if (current.startsWith(next)) { + return current + } + return next +} + +function collectKimiStreamSources( + input: unknown, + bucket: MonitoringSourceItem[], + depth = 0, + sourceContext = false, +): void { + if (input == null || depth > 12) { + return + } + if (Array.isArray(input)) { + for (const item of input) { + collectKimiStreamSources(item, bucket, depth + 1, sourceContext) + } + return + } + if (!isRecord(input)) { + return + } + + if (sourceContext) { + const url = normalizeText( + input.url ?? + input.href ?? + input.link ?? + input.source_url ?? + input.sourceUrl ?? + input.target_url ?? + input.targetUrl, + ) + if (url) { + const source = buildSourceItem({ + url, + title: normalizeText(input.title ?? input.name ?? input.page_title ?? input.pageTitle), + text: normalizeText(input.snippet ?? input.summary ?? input.description), + siteName: normalizeText( + input.site_name ?? input.siteName ?? input.domain ?? input.host ?? input.source_name, + ), + }) + if (source) { + bucket.push(source) + } + } + } + + for (const [key, value] of Object.entries(input)) { + const nextSourceContext = + sourceContext || + /(?:^|_)(?:search|source|citation|reference|result|web_page|webpage)(?:s|_data|_list)?$/i.test( + key, + ) + if (Array.isArray(value) || isRecord(value)) { + collectKimiStreamSources(value, bucket, depth + 1, nextSourceContext) + } + } +} + +function collectKimiStreamErrorStrings(input: unknown, bucket: string[], depth = 0): void { + if (input == null || depth > 8) { + return + } + if (Array.isArray(input)) { + for (const item of input) { + collectKimiStreamErrorStrings(item, bucket, depth + 1) + } + return + } + if (!isRecord(input)) { + return + } + for (const [key, value] of Object.entries(input)) { + if (/^(?:code|reason|message)$/i.test(key)) { + const normalized = normalizeText(value) + if (normalized) { + bucket.push(normalized) + } + } + if (Array.isArray(value) || isRecord(value)) { + collectKimiStreamErrorStrings(value, bucket, depth + 1) + } + } +} + +function summarizeKimiStreamEvents(events: unknown[]): KimiStreamSummary { + const answerByBlock = new Map() + const reasoningByBlock = new Map() + const sourceItems: MonitoringSourceItem[] = [] + const errorStrings: string[] = [] + const tips: string[] = [] + let conversationID: string | null = null + let providerRequestID: string | null = null + let completed = false + let blockSequence = 0 + + const processBlock = ( + block: Record, + operation: string | null, + mask: string | null, + ) => { + const messageID = normalizeText(block.messageId ?? block.message_id) + if (providerRequestID && messageID && messageID !== providerRequestID) { + return + } + + const blockID = normalizeText(block.id) ?? `block-${blockSequence++}` + const reasoningContainer = isRecord(block.reasoning) + ? block.reasoning + : isRecord(block.thinking) + ? block.thinking + : isRecord(block.thought) + ? block.thought + : null + const textContainer = isRecord(block.text) ? block.text : null + const reasoningValue = reasoningContainer + ? (reasoningContainer.content ?? reasoningContainer.delta ?? reasoningContainer.text) + : null + const maskLooksLikeReasoning = Boolean(mask && /reason|think|thought|analysis/i.test(mask)) + + if (reasoningValue != null) { + const merged = mergeKimiStreamText(reasoningByBlock.get(blockID), reasoningValue, operation) + if (merged) { + reasoningByBlock.set(blockID, merged) + } + } + + const textValue = textContainer + ? (textContainer.content ?? textContainer.delta ?? textContainer.text) + : typeof block.text === 'string' + ? block.text + : null + if (textValue != null) { + const target = maskLooksLikeReasoning ? reasoningByBlock : answerByBlock + const merged = mergeKimiStreamText(target.get(blockID), textValue, operation) + if (merged) { + target.set(blockID, merged) + } + } + + const tip = normalizeText(textContainer?.tips ?? block.tips) + if (tip) { + tips.push(tip) + } + if (isRecord(block.exception)) { + collectKimiStreamErrorStrings(block.exception, errorStrings) + } + collectKimiStreamSources(block, sourceItems) + } + + for (const event of events) { + if (!isRecord(event)) { + continue + } + const operation = normalizeText(event.op)?.toLowerCase() ?? null + const mask = normalizeText(event.mask) + const chat = isRecord(event.chat) ? event.chat : null + conversationID = normalizeText(chat?.id) ?? conversationID + + const message = isRecord(event.message) ? event.message : null + const role = normalizeText(message?.role)?.toLowerCase() + const messageID = normalizeText(message?.id) + if (role === 'assistant' && messageID) { + providerRequestID = messageID + } + const messageStatus = normalizeText(message?.status) + const isAssistantMessage = + role === 'assistant' || Boolean(providerRequestID && messageID === providerRequestID) + if (isAssistantMessage && messageStatus?.includes('COMPLETED')) { + completed = true + } + if (role === 'assistant' && Array.isArray(message?.blocks)) { + for (const block of message.blocks) { + if (isRecord(block)) { + processBlock(block, operation, mask) + } + } + } + + if (isRecord(event.block)) { + processBlock(event.block, operation, mask) + } + if (isRecord(event.error)) { + collectKimiStreamErrorStrings(event.error, errorStrings) + } + collectKimiStreamSources(event, sourceItems) + } + + const answerBlocks = Array.from(answerByBlock.values()) + .map((value) => normalizeText(value)) + .filter((value): value is string => value !== null) + const reasoningBlocks = Array.from(reasoningByBlock.values()) + .map((value) => normalizeText(value)) + .filter((value): value is string => value !== null) + const errorReason = errorStrings.find(Boolean) ?? null + const capacityLimited = errorStrings.some((value) => + /overload|capacity|too[_\s-]*many|service[_\s-]*(?:busy|unavailable)|算力|服务繁忙/i.test( + value, + ), + ) + const capacityTip = tips.find((value) => + /遇到了一些问题|暂停了这个任务|算力|服务繁忙|请稍后再试|请求过多/i.test(value), + ) + + return { + answer: normalizeText(answerBlocks.join('\n\n')), + answerBlocks, + reasoning: normalizeText(reasoningBlocks.join('\n\n')), + reasoningBlocks, + capacityNotice: capacityLimited ? (capacityTip ?? 'Kimi 服务繁忙，请稍后再试。') : null, + conversationID, + providerRequestID, + completed, + errorReason, + searchResults: dedupeSourceItems(sourceItems), + } +} + +function isUsableKimiStreamSummary(summary: KimiStreamSummary): boolean { + return Boolean( + (summary.completed && normalizeText(summary.answer)) || normalizeText(summary.capacityNotice), + ) +} + +function buildKimiStreamSnapshot( + initialSnapshot: KimiPageSnapshot, + summary: KimiStreamSummary, +): KimiPageSnapshot { + const searchResultLinks: KimiDomLink[] = summary.searchResults.map((item) => ({ + url: item.url, + title: item.title ?? null, + text: null, + siteName: item.site_name ?? item.host ?? null, + })) + const answer = normalizeText(summary.answer) + const reasoning = normalizeText(summary.reasoning) + const capacityNotice = normalizeText(summary.capacityNotice) + const url = summary.conversationID + ? `${KIMI_BOOTSTRAP_URL}chat/${encodeURIComponent(summary.conversationID)}` + : initialSnapshot.url + + return { + ...initialSnapshot, + url, + currentModelLabel: KIMI_K3_MODE.label, + busy: false, + busySignals: [], + sendDisabled: null, + answerText: answer, + answerBlocks: summary.answerBlocks, + reasoningText: reasoning, + reasoningBlocks: summary.reasoningBlocks, + capacityNotice, + questionMatched: true, + candidateAnswerCount: summary.answerBlocks.length, + candidateReasoningCount: summary.reasoningBlocks.length, + citationLinks: searchResultLinks, + panelCitationLinks: [], + inlineCitationCandidateLinks: [], + searchResultLinks, + signature: + [answer, reasoning, capacityNotice, ...searchResultLinks.map((item) => item.url)] + .filter(Boolean) + .join('\n') || null, + } +} + function dedupeKimiDomLinks(items: KimiDomLink[]): KimiDomLink[] { const keyed = new Map() for (const item of items) { @@ -486,6 +851,230 @@ async function sleep(ms: number, signal?: AbortSignal): Promise { }) } +async function queryKimiStreamAPI( + page: PlaywrightPage, + questionText: string, + signal: AbortSignal, +): Promise { + if (signal.aborted) { + throw new Error('adapter_aborted') + } + + try { + const result = await page.evaluate( + async ({ endpoint, question, timeoutMs, maxFrameLength, maxFrameCount }) => { + const startedAt = Date.now() + const fail = ( + error: string, + detail: string | null, + extras: { + status?: number | null + contentType?: string | null + frameCount?: number + pendingBytes?: number + events?: unknown[] + } = {}, + ) => ({ + ok: false, + status: extras.status ?? null, + contentType: extras.contentType ?? null, + frameCount: extras.frameCount ?? 0, + pendingBytes: extras.pendingBytes ?? 0, + elapsedMs: Date.now() - startedAt, + events: extras.events ?? [], + error, + detail, + }) + + const accessToken = window.localStorage.getItem('access_token')?.trim() + if (!accessToken) { + return fail('kimi_api_access_token_missing', 'Kimi access_token is unavailable') + } + + const requestPayload = { + scenario: 'SCENARIO_OK_COMPUTER', + tools: [{ type: 'TOOL_TYPE_SEARCH', search: {} }, { type: 'TOOL_TYPE_ASK_USER' }], + message: { + role: 'user', + blocks: [{ message_id: '', text: { content: question } }], + scenario: 'SCENARIO_OK_COMPUTER', + is_goal: false, + }, + options: { + thinking: true, + enable_plugin: true, + reasoning_effort: 'REASONING_EFFORT_LOW', + context_length: 'CONTEXT_LENGTH_L', + }, + project_id: '', + } + const json = new TextEncoder().encode(JSON.stringify(requestPayload)) + const requestBody = new Uint8Array(json.length + 5) + requestBody[0] = 0 + new DataView(requestBody.buffer).setUint32(1, json.length, false) + requestBody.set(json, 5) + + const controller = new AbortController() + const timer = window.setTimeout(() => controller.abort(), timeoutMs) + try { + const response = await window.fetch(endpoint, { + method: 'POST', + headers: { + authorization: `Bearer ${accessToken}`, + 'content-type': 'application/connect+json', + 'connect-protocol-version': '1', + 'x-msh-platform': 'web', + 'x-msh-version': '2.0.0', + 'x-language': 'zh-CN', + 'r-timezone': Intl.DateTimeFormat().resolvedOptions().timeZone, + }, + body: requestBody, + signal: controller.signal, + }) + const contentType = response.headers.get('content-type') + if (!response.ok) { + return fail('kimi_api_http_error', `HTTP ${response.status}`, { + status: response.status, + contentType, + }) + } + if (!contentType || !/application\/connect\+json/i.test(contentType)) { + return fail('kimi_api_content_type_invalid', contentType, { + status: response.status, + contentType, + }) + } + if (!response.body) { + return fail('kimi_api_response_body_missing', 'response.body is unavailable', { + status: response.status, + contentType, + }) + } + + const reader = response.body.getReader() + const decoder = new TextDecoder('utf-8') + const events: unknown[] = [] + let pending = new Uint8Array(0) + + while (true) { + const { done, value } = await reader.read() + if (done) { + break + } + if (!value?.length) { + continue + } + + const merged = new Uint8Array(pending.length + value.length) + merged.set(pending) + merged.set(value, pending.length) + pending = merged + + while (pending.length >= 5) { + const length = new DataView(pending.buffer, pending.byteOffset + 1, 4).getUint32( + 0, + false, + ) + if (length > maxFrameLength) { + return fail('kimi_api_frame_too_large', String(length), { + status: response.status, + contentType, + frameCount: events.length, + pendingBytes: pending.length, + events, + }) + } + if (pending.length < 5 + length) { + break + } + + const payload = pending.slice(5, 5 + length) + try { + events.push(JSON.parse(decoder.decode(payload)) as unknown) + } catch { + return fail('kimi_api_frame_invalid_json', null, { + status: response.status, + contentType, + frameCount: events.length, + pendingBytes: pending.length, + events, + }) + } + pending = pending.slice(5 + length) + if (events.length > maxFrameCount) { + return fail('kimi_api_frame_limit_exceeded', String(events.length), { + status: response.status, + contentType, + frameCount: events.length, + pendingBytes: pending.length, + events, + }) + } + } + } + + if (pending.length > 0) { + return fail('kimi_api_incomplete_frame', String(pending.length), { + status: response.status, + contentType, + frameCount: events.length, + pendingBytes: pending.length, + events, + }) + } + + return { + ok: true, + status: response.status, + contentType, + frameCount: events.length, + pendingBytes: 0, + elapsedMs: Date.now() - startedAt, + events, + error: null, + detail: null, + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + return fail( + controller.signal.aborted ? 'kimi_api_timeout' : 'kimi_api_fetch_failed', + message, + ) + } finally { + window.clearTimeout(timer) + } + }, + { + endpoint: KIMI_STREAM_ENDPOINT, + question: questionText, + timeoutMs: KIMI_QUERY_TIMEOUT_MS, + maxFrameLength: KIMI_STREAM_MAX_FRAME_LENGTH, + maxFrameCount: KIMI_STREAM_MAX_FRAME_COUNT, + }, + ) + + if (signal.aborted) { + throw new Error('adapter_aborted') + } + return result + } catch (error) { + if (signal.aborted) { + throw new Error('adapter_aborted', { cause: error }) + } + return { + ok: false, + status: null, + contentType: null, + frameCount: 0, + pendingBytes: 0, + elapsedMs: 0, + events: [], + error: 'kimi_api_evaluate_failed', + detail: error instanceof Error ? error.message : String(error), + } + } +} + async function ensurePageOnKimiChat(page: PlaywrightPage, signal: AbortSignal): Promise { if (signal.aborted) { throw new Error('adapter_aborted') @@ -1013,6 +1602,9 @@ function shouldReturnStableKimiAnswer( firstChangedContentAt: number | null, now: number, ): boolean { + if (normalizeOptionalString(snapshot.capacityNotice)) { + return true + } if (isKimiAnswerComplete(snapshot)) { return true } @@ -1058,6 +1650,7 @@ function hasKimiObservedContent(snapshot: KimiPageSnapshot | null): boolean { (reasoning && !isKimiPromotionalAnswerNoise(reasoning)) || answerBlocks.length > 0 || reasoningBlocks.length > 0 || + normalizeOptionalString(snapshot.capacityNotice) || snapshot.citationLinks.length > 0 || snapshot.searchResultLinks.length > 0, ) @@ -1077,6 +1670,9 @@ function answerQualityScore(snapshot: KimiPageSnapshot | null): number { if (reasoning) { score += Math.min(reasoning.length, 2_000) / 2 } + if (normalizeOptionalString(snapshot.capacityNotice)) { + score += 160 + } score += snapshot.answerBlocks.length * 240 score += snapshot.reasoningBlocks.length * 80 score += snapshot.citationLinks.length * 180 @@ -1486,41 +2082,44 @@ async function ensureKimiMode( } } - result = await page.evaluate(({ modeKeyword }) => { - const normalize = (value: unknown): string | null => { - if (typeof value !== 'string') { - return null + result = await page.evaluate( + ({ modeKeyword }) => { + const normalize = (value: unknown): string | null => { + if (typeof value !== 'string') { + return null + } + const trimmed = value + .replace(/\u00a0/g, ' ') + .split(/\n+/) + .map((line) => line.trim()) + .filter(Boolean) + .join(' ') + return trimmed ? trimmed : null } - const trimmed = value - .replace(/\u00a0/g, ' ') - .split(/\n+/) - .map((line) => line.trim()) - .filter(Boolean) - .join(' ') - return trimmed ? trimmed : null - } - const currentModelLabel = normalize( - document.querySelector('.current-model .name')?.textContent ?? - document.querySelector('.current-model .model-name')?.textContent ?? - document.querySelector('.current-model')?.textContent, - ) + const currentModelLabel = normalize( + document.querySelector('.current-model .name')?.textContent ?? + document.querySelector('.current-model .model-name')?.textContent ?? + document.querySelector('.current-model')?.textContent, + ) + + if (currentModelLabel?.includes(modeKeyword)) { + return { + ok: true as const, + currentModelLabel, + } + } - if (currentModelLabel?.includes(modeKeyword)) { return { - ok: true as const, + ok: false as const, + error: 'kimi_mode_not_selected', + detail: normalize(document.body?.innerText)?.slice(0, 200) ?? null, currentModelLabel, + url: window.location.href || null, } - } - - return { - ok: false as const, - error: 'kimi_mode_not_selected', - detail: normalize(document.body?.innerText)?.slice(0, 200) ?? null, - currentModelLabel, - url: window.location.href || null, - } - }, { modeKeyword: target.keyword }) + }, + { modeKeyword: target.keyword }, + ) return result } @@ -2897,7 +3496,7 @@ async function readKimiPageSnapshot( // 请稍后再试" etc). The pattern matches phrases only — no class names — // so it stays stable across UI redesigns. const capacityLimitPattern = - /(算力不够|算力不足|服务繁忙|当前用户量较大|请稍后再试|请稍候再试|请求过多|系统繁忙|too\s*many\s*requests|service\s*(?:busy|unavailable)|capacity\s*(?:limit|exhaust))/i + /(算力不够|算力不足|服务繁忙|当前用户量较大|请稍后再试|请稍候再试|请求过多|系统繁忙|Kimi\s*遇到了一些问题|暂停了这个任务|权益.*退还|too\s*many\s*requests|service\s*(?:busy|unavailable)|capacity\s*(?:limit|exhaust))/i const capacityNotice = (() => { const assistantNodes = Array.from( document.querySelectorAll('.chat-content-item-assistant, .segment-assistant'), @@ -2937,6 +3536,7 @@ async function readKimiPageSnapshot( currentModelLabel, answerText, reasoningText, + capacityNotice, answerBlocks.join('\n'), reasoningBlocks.join('\n'), ...citationLinks.map((item) => `${item.siteName ?? item.text ?? item.url}:${item.url}`), @@ -2996,7 +3596,8 @@ async function waitForKimiAnswer( const hasAnswer = Boolean(normalizeOptionalString(snapshot.answerText)) const hasLinks = snapshot.citationLinks.length > 0 || snapshot.searchResultLinks.length > 0 const hasReasoning = snapshot.reasoningBlocks.length > 0 - const hasContent = hasAnswer || hasLinks || hasReasoning + const hasCapacityNotice = Boolean(normalizeOptionalString(snapshot.capacityNotice)) + const hasContent = hasAnswer || hasLinks || hasReasoning || hasCapacityNotice if (snapshot.signature && hasContent) { if (snapshot.signature !== initialSignature) { sawChange = true @@ -3014,7 +3615,7 @@ async function waitForKimiAnswer( const now = Date.now() if ( - sawChange && + (sawChange || hasCapacityNotice) && hasContent && !snapshot.busy && stablePollCount >= KIMI_STABLE_POLLS_REQUIRED && @@ -3117,48 +3718,69 @@ export const kimiAdapter: MonitorAdapter = { } } - context.reportProgress('kimi.mode_switch') - const modeSwitchResult = await ensureKimiMode(page, KIMI_TARGET_MODE) - if (!modeSwitchResult.ok) { - const isAuthFailure = modeSwitchResult.error === 'kimi_login_required' - return { - status: 'failed', - summary: isAuthFailure - ? 'Kimi 账号未登录或登录态已失效，请先在 desktop client 中重新授权。' - : `Kimi 未能切换到 ${KIMI_TARGET_MODE.label}，已中止本次监控任务。`, - error: buildAdapterError( - modeSwitchResult.error, - modeSwitchResult.detail ?? modeSwitchResult.error, - { - page_url: modeSwitchResult.url ?? safePageURL(page), - current_model_label: modeSwitchResult.currentModelLabel, - }, - ), + context.reportProgress('kimi.api_query') + const apiTransport = await queryKimiStreamAPI(page, questionText, context.signal) + const apiSummary = summarizeKimiStreamEvents(apiTransport.events) + const apiUsable = apiTransport.ok && isUsableKimiStreamSummary(apiSummary) + const apiFallbackReason = apiUsable + ? null + : (apiTransport.error ?? apiSummary.errorReason ?? 'kimi_api_empty_response') + let responseSource: 'api_stream' | 'page_rpa' + let modeSwitchResult: KimiModeSwitchResult | null = null + let queryResult: KimiQueryWaitResult + + if (apiUsable) { + responseSource = 'api_stream' + queryResult = { + ok: true, + finalSnapshot: buildKimiStreamSnapshot(initialSnapshot, apiSummary), + stablePollCount: 0, + elapsedMs: apiTransport.elapsedMs, } + } else { + responseSource = 'page_rpa' + context.reportProgress('kimi.page_rpa_fallback') + context.reportProgress('kimi.mode_switch') + modeSwitchResult = await ensureKimiMode(page, KIMI_TARGET_MODE) + if (!modeSwitchResult.ok && modeSwitchResult.error === 'kimi_login_required') { + return { + status: 'failed', + summary: 'Kimi 账号未登录或登录态已失效，请先在 desktop client 中重新授权。', + error: buildAdapterError( + 'kimi_login_required', + modeSwitchResult.detail ?? modeSwitchResult.error, + { + page_url: modeSwitchResult.url ?? safePageURL(page), + current_model_label: modeSwitchResult.currentModelLabel, + }, + ), + } + } + + context.reportProgress('kimi.query') + await submitKimiQuestion(page, questionText, context.signal) + await sleep(400, context.signal).catch(() => undefined) + await pinKimiConversationToBottom(page) + + const postSubmitSnapshot = await readKimiPageSnapshot(page, questionText) + if (postSubmitSnapshot.loginRequired) { + return { + status: 'failed', + summary: 'Kimi 账号未登录或登录态已失效，请先在 desktop client 中重新授权。', + error: buildAdapterError( + 'kimi_login_expired', + postSubmitSnapshot.loginReason ?? 'login_required', + { + page_url: postSubmitSnapshot.url, + }, + ), + } + } + + context.reportProgress('kimi.wait_answer') + queryResult = await waitForKimiAnswer(page, questionText, context.signal) } - context.reportProgress('kimi.query') - await submitKimiQuestion(page, questionText, context.signal) - await sleep(400, context.signal).catch(() => undefined) - await pinKimiConversationToBottom(page) - - const postSubmitSnapshot = await readKimiPageSnapshot(page, questionText) - if (postSubmitSnapshot.loginRequired) { - return { - status: 'failed', - summary: 'Kimi 账号未登录或登录态已失效，请先在 desktop client 中重新授权。', - error: buildAdapterError( - 'kimi_login_expired', - postSubmitSnapshot.loginReason ?? 'login_required', - { - page_url: postSubmitSnapshot.url, - }, - ), - } - } - - context.reportProgress('kimi.wait_answer') - const queryResult = await waitForKimiAnswer(page, questionText, context.signal) const finalSnapshot = queryResult.finalSnapshot const classifiedSources = classifyKimiSources(finalSnapshot) const citations = classifiedSources.citations @@ -3180,7 +3802,10 @@ export const kimiAdapter: MonitorAdapter = { const answerComplete = isKimiAnswerComplete(finalSnapshot) const keywordBlobDetected = answer ? isKimiKeywordBlob(answer) : false const providerModel = normalizeText(finalSnapshot.currentModelLabel) ?? KIMI_TARGET_MODE.label - const conversationID = extractConversationID(finalSnapshot.url) + const providerRequestID = responseSource === 'api_stream' ? apiSummary.providerRequestID : null + const conversationID = + (responseSource === 'api_stream' ? apiSummary.conversationID : null) ?? + extractConversationID(finalSnapshot.url) if (!queryResult.ok && queryResult.error === 'kimi_login_expired') { return { @@ -3203,7 +3828,7 @@ export const kimiAdapter: MonitorAdapter = { status: 'unknown', summary: 'Kimi 未拿到完整答案，已回写 unknown 等待后续补采。', error: buildAdapterError( - timedOut ? queryResult.error : 'kimi_incomplete_response', + timedOut ? 'kimi_query_timeout' : 'kimi_incomplete_response', incompleteDetail, { page_url: finalSnapshot.url, @@ -3248,10 +3873,11 @@ export const kimiAdapter: MonitorAdapter = { payload: { platform: 'kimi', provider_model: providerModel, - provider_request_id: null, + provider_request_id: providerRequestID, request_id: conversationID, conversation_id: conversationID, answer, + response_source: responseSource, reasoning, citation_count: citations.length, search_result_count: searchResults.length, @@ -3260,10 +3886,30 @@ export const kimiAdapter: MonitorAdapter = { raw_response_json: { platform: 'kimi', status: queryResult.ok ? 'succeeded' : queryResult.error, + response_source: responseSource, page_url: finalSnapshot.url, page_title: finalSnapshot.title, model_label: finalSnapshot.currentModelLabel, mode_requested: KIMI_TARGET_MODE.label, + mode_switch: modeSwitchResult + ? { + ok: modeSwitchResult.ok, + current_model_label: modeSwitchResult.currentModelLabel, + error: modeSwitchResult.ok ? null : modeSwitchResult.error, + } + : null, + api_stream: { + ok: apiTransport.ok, + status: apiTransport.status, + content_type: apiTransport.contentType, + frame_count: apiTransport.frameCount, + pending_bytes: apiTransport.pendingBytes, + elapsed_ms: apiTransport.elapsedMs, + completed: apiSummary.completed, + error: apiTransport.error, + error_reason: apiSummary.errorReason, + fallback_reason: apiFallbackReason, + }, login_required: finalSnapshot.loginRequired, busy: finalSnapshot.busy, busy_signals: finalSnapshot.busySignals, @@ -3312,14 +3958,18 @@ export const kimiAdapter: MonitorAdapter = { export const __kimiTestUtils = { KIMI_FAST_MODE, + KIMI_K3_MODE, KIMI_TARGET_MODE, KIMI_THINKING_MODE, buildSourceItem, classifyKimiSources, + decodeKimiConnectFrames, dedupeSourceItems, extractQuestionText, isKimiAnswerComplete, isKimiPromotionalAnswerNoise, mergeKimiSourceLinksIntoContentSnapshot, normalizeUrl, + shouldReturnStableKimiAnswer, + summarizeKimiStreamEvents, }