feat(kimi-adapter): query Kimi Connect stream API with page RPA fallback

- Call the kimi.com apiv2 ChatService/Chat endpoint directly using
  Connect+JSON framing: encode the request as a length-prefixed frame,
  decode streamed frames, and summarize set/append events into answer,
  reasoning, sources, and capacity notices
- Fall back to the existing page RPA flow (mode switch + DOM polling)
  when the API transport fails or yields no usable summary
- Switch the target model from K2.6 fast to K3
- Treat capacity notices as terminal answers: include them in snapshot
  signature/content checks and broaden the capacity-limit pattern
This commit is contained in:
2026-07-18 21:15:57 +08:00
parent 6adcec4849
commit 457d8c1890
2 changed files with 1008 additions and 80 deletions
@@ -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<Parameters<typeof classifyKimiSources>[0]>,
): Parameters<typeof classifyKimiSources>[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<string, unknown>).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(
+671 -21
View File
@@ -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<string, unknown> {
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<string, string>()
const reasoningByBlock = new Map<string, string>()
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<string, unknown>,
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<string, KimiDomLink>()
for (const item of items) {
@@ -486,6 +851,230 @@ async function sleep(ms: number, signal?: AbortSignal): Promise<void> {
})
}
async function queryKimiStreamAPI(
page: PlaywrightPage,
questionText: string,
signal: AbortSignal,
): Promise<KimiStreamTransportResult> {
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<void> {
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,7 +2082,8 @@ async function ensureKimiMode(
}
}
result = await page.evaluate(({ modeKeyword }) => {
result = await page.evaluate(
({ modeKeyword }) => {
const normalize = (value: unknown): string | null => {
if (typeof value !== 'string') {
return null
@@ -1520,7 +2117,9 @@ async function ensureKimiMode(
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,17 +3718,36 @@ export const kimiAdapter: MonitorAdapter = {
}
}
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')
const modeSwitchResult = await ensureKimiMode(page, KIMI_TARGET_MODE)
if (!modeSwitchResult.ok) {
const isAuthFailure = modeSwitchResult.error === 'kimi_login_required'
modeSwitchResult = await ensureKimiMode(page, KIMI_TARGET_MODE)
if (!modeSwitchResult.ok && modeSwitchResult.error === 'kimi_login_required') {
return {
status: 'failed',
summary: isAuthFailure
? 'Kimi 账号未登录或登录态已失效,请先在 desktop client 中重新授权。'
: `Kimi 未能切换到 ${KIMI_TARGET_MODE.label},已中止本次监控任务。`,
summary: 'Kimi 账号未登录或登录态已失效,请先在 desktop client 中重新授权。',
error: buildAdapterError(
modeSwitchResult.error,
'kimi_login_required',
modeSwitchResult.detail ?? modeSwitchResult.error,
{
page_url: modeSwitchResult.url ?? safePageURL(page),
@@ -3158,7 +3778,9 @@ export const kimiAdapter: MonitorAdapter = {
}
context.reportProgress('kimi.wait_answer')
const queryResult = await waitForKimiAnswer(page, questionText, context.signal)
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,
}