@@ -492,7 +550,23 @@ function ConversationCard(props) {
size={16}
text={t('Clear Conversation')}
onConfirm={async () => {
- await postMessage({ stop: true })
+ ++requestGenerationIdRef.current
+ const stopGenerationId = ++retryGenerationIdRef.current
+ try {
+ await postMessage({ stop: true, stopGenerationId })
+ } catch (error) {
+ console.warn(
+ '[ConversationCard] Failed to stop generation before clearing conversation:',
+ error,
+ )
+ }
+ if (!useForegroundFetch) {
+ replacedPortRef.current = port
+ port.disconnect()
+ setPort(Browser.runtime.connect())
+ }
+ partialAnswerRef.current = ''
+ retryRecordRef.current = null
Browser.runtime.sendMessage({
type: 'DELETE_CONVERSATION',
data: {
@@ -507,6 +581,7 @@ function ConversationCard(props) {
})
newSession.sessionId = session.sessionId
setSession(newSession)
+ setIsReady(true)
}}
/>
{!props.pageMode && (
@@ -616,6 +691,8 @@ function ConversationCard(props) {
'answer',
`
${t('Waiting for response...')}
`,
)
+ partialAnswerRef.current = ''
+ retryRecordRef.current = null
setConversationItemData([...conversationItemData, newQuestion, newAnswer])
setIsReady(false)
diff --git a/src/components/ConversationCard/session.mjs b/src/components/ConversationCard/session.mjs
new file mode 100644
index 000000000..40c60f166
--- /dev/null
+++ b/src/components/ConversationCard/session.mjs
@@ -0,0 +1,79 @@
+import { pushRecord } from '../../services/apis/shared.mjs'
+
+export function finalizeInterruptedSession(session, answer, retryRecord = null) {
+ if (!answer) {
+ if (!session.isRetry && !retryRecord) return session
+ const lastRecord = session.conversationRecords.at(-1)
+ const shouldRestoreRetryRecord =
+ retryRecord &&
+ (lastRecord?.question !== retryRecord.question || lastRecord?.answer !== retryRecord.answer)
+ return {
+ ...session,
+ conversationRecords: shouldRestoreRetryRecord
+ ? [...session.conversationRecords, { ...retryRecord }]
+ : session.conversationRecords,
+ isRetry: false,
+ }
+ }
+ const updatedSession = {
+ ...session,
+ conversationRecords: session.conversationRecords.map((record) => ({ ...record })),
+ }
+ pushRecord(updatedSession, session.question, answer)
+ updatedSession.isRetry = false
+ return updatedSession
+}
+
+export function isSupersededGenerationMessage(message, latestSupersededGenerationId) {
+ return (
+ message.stoppedGenerationId !== undefined &&
+ message.stoppedGenerationId <= latestSupersededGenerationId
+ )
+}
+
+export function isSupersededRequestMessage(message, currentRequestGenerationId) {
+ return (
+ message.requestGenerationId !== undefined &&
+ message.requestGenerationId !== currentRequestGenerationId
+ )
+}
+
+export function createConversationPortMessage({
+ session,
+ stop,
+ stopGenerationId,
+ requestGenerationId,
+}) {
+ return {
+ session,
+ stop,
+ ...(stopGenerationId === undefined ? {} : { stopGenerationId }),
+ ...(requestGenerationId === undefined ? {} : { requestGenerationId }),
+ }
+}
+
+export function createRetrySession(session, conversationRecords, retryRecord) {
+ return {
+ ...session,
+ conversationRecords,
+ isRetry: retryRecord === null,
+ }
+}
+
+export function getCompletedAnswerUpdate(restoredRetryAnswer) {
+ return {
+ value: restoredRetryAnswer ?? '',
+ appended: restoredRetryAnswer === null,
+ }
+}
+
+export function getInterruptedCompletionState(message, partialAnswer, retryRecord) {
+ const shouldFinalize = Boolean(
+ message.proxyDisconnected || (!message.session && (partialAnswer || retryRecord)),
+ )
+ return {
+ shouldFinalize,
+ restoredRetryAnswer:
+ shouldFinalize && !partialAnswer && retryRecord ? retryRecord.answer : null,
+ }
+}
diff --git a/src/content-script/index.jsx b/src/content-script/index.jsx
index ea70dfbd2..07320a240 100644
--- a/src/content-script/index.jsx
+++ b/src/content-script/index.jsx
@@ -872,7 +872,7 @@ function ensureChatGptPortListenerRegistered() {
try {
console.log('[content] Attempting to register port listener for chatgpt.com.')
- registerPortListener(async (session, port) => {
+ registerPortListener(async (session, port, _config, isLatestSessionRequest) => {
console.debug(
`[content] Port listener callback triggered. Session model: ${session?.modelName}, Port: ${port.name}`,
)
@@ -883,6 +883,10 @@ function ensureChatGptPortListenerRegistered() {
session.question,
)
const accessToken = await getChatGptAccessToken()
+ if (!isLatestSessionRequest()) {
+ console.debug('[content] Skipping a superseded ChatGPT Web session request.')
+ return
+ }
if (!accessToken) {
console.warn('[content] No ChatGPT access token available for web API call.')
port.postMessage({ error: 'Missing ChatGPT access token.' })
@@ -896,6 +900,10 @@ function ensureChatGptPortListenerRegistered() {
)
}
} catch (e) {
+ if (!isLatestSessionRequest()) {
+ console.debug('[content] Ignoring an error from a superseded session request.')
+ return
+ }
console.error('[content] Error in port listener callback:', e, 'Session:', session)
try {
port.postMessage({
diff --git a/src/services/apis/azure-openai-api.mjs b/src/services/apis/azure-openai-api.mjs
index 40585348d..a8e8eaf80 100644
--- a/src/services/apis/azure-openai-api.mjs
+++ b/src/services/apis/azure-openai-api.mjs
@@ -68,10 +68,15 @@ export async function generateAnswersWithAzureOpenaiApi(port, question, session)
}
},
async onStart() {},
- async onEnd() {
- port.postMessage({ done: true })
- port.onMessage.removeListener(messageListener)
- port.onDisconnect.removeListener(disconnectListener)
+ async onEnd(aborted) {
+ try {
+ if (!aborted) {
+ port.postMessage({ done: true })
+ }
+ } finally {
+ port.onMessage.removeListener(messageListener)
+ port.onDisconnect.removeListener(disconnectListener)
+ }
},
async onError(resp) {
port.onMessage.removeListener(messageListener)
diff --git a/src/services/apis/bard-web.mjs b/src/services/apis/bard-web.mjs
index 95445ffce..8e2dd7069 100644
--- a/src/services/apis/bard-web.mjs
+++ b/src/services/apis/bard-web.mjs
@@ -1,19 +1,27 @@
import { pushRecord } from './shared.mjs'
-import Bard from '../clients/bard'
+import Bard from '../clients/bard/index.mjs'
/**
* @param {Runtime.Port} port
* @param {string} question
* @param {Session} session
* @param {string} cookies
+ * @param {() => boolean} isLatestSessionRequest
*/
-export async function generateAnswersWithBardWebApi(port, question, session, cookies) {
+export async function generateAnswersWithBardWebApi(
+ port,
+ question,
+ session,
+ cookies,
+ isLatestSessionRequest = () => true,
+) {
// const { controller, messageListener, disconnectListener } = setAbortController(port)
const bot = new Bard(cookies)
// eslint-disable-next-line
try {
const { answer, conversationObj } = await bot.ask(question, session.bard_conversationObj || {})
+ if (!isLatestSessionRequest()) return
session.bard_conversationObj = conversationObj
pushRecord(session, question, answer)
console.debug('conversation history', { content: session.conversationRecords })
diff --git a/src/services/apis/chatgpt-web.mjs b/src/services/apis/chatgpt-web.mjs
index 8684e9cdc..37f63d518 100644
--- a/src/services/apis/chatgpt-web.mjs
+++ b/src/services/apis/chatgpt-web.mjs
@@ -212,16 +212,34 @@ export async function registerWebsocket(accessToken) {
* @param {string} accessToken
*/
export async function generateAnswersWithChatgptWebApi(port, question, session, accessToken) {
- const { controller, cleanController } = setAbortController(
+ let wsCallback
+ let cleanController = () => {}
+ const removeWebsocketCallback = () => {
+ if (!wsCallback) return
+ wsCallbacks = wsCallbacks.filter((callback) => callback !== wsCallback)
+ wsCallback = null
+ }
+ const stopWebsocketRequest = (conversationId, wsRequestId) => {
+ if (!wsRequestId) return
+ stopWebsocketConversation(accessToken, conversationId, wsRequestId).catch((error) => {
+ console.warn('[chatgpt-web] Failed to stop WebSocket conversation:', error)
+ })
+ }
+ const abortControllerState = setAbortController(
port,
() => {
- if (session.wsRequestId)
- stopWebsocketConversation(accessToken, session.conversationId, session.wsRequestId)
+ stopWebsocketRequest(session.conversationId, session.wsRequestId)
+ removeWebsocketCallback()
+ cleanController()
},
() => {
+ removeWebsocketCallback()
+ cleanController()
if (session.autoClean) deleteConversation(accessToken, session.conversationId)
},
)
+ const { controller } = abortControllerState
+ cleanController = abortControllerState.cleanController
const config = await getUserConfig()
let arkoseError
@@ -320,8 +338,16 @@ export async function generateAnswersWithChatgptWebApi(port, question, session,
let generatedImageUrl = ''
if (useWebsocket) {
+ if (controller.signal.aborted) {
+ cleanController()
+ return
+ }
await registerWebsocket(accessToken)
- const wsCallback = async (event) => {
+ if (controller.signal.aborted) {
+ cleanController()
+ return
+ }
+ wsCallback = async (event) => {
let wsData
try {
wsData = JSON.parse(event.data)
@@ -343,7 +369,6 @@ export async function generateAnswersWithChatgptWebApi(port, question, session,
console.debug('ws message', '[DONE]')
if (wsData.conversation_id === session.conversationId) {
finishMessage()
- wsCallbacks = wsCallbacks.filter((cb) => cb !== wsCallback)
}
} else {
console.debug('json error', error)
@@ -352,50 +377,73 @@ export async function generateAnswersWithChatgptWebApi(port, question, session,
}
}
wsCallbacks.push(wsCallback)
- const { conversationId, wsRequestId } = await sendWebsocketConversation(accessToken, options)
- session.conversationId = conversationId
- session.wsRequestId = wsRequestId
- port.postMessage({ session: session })
- } else {
- await fetchSSE(url, {
- ...options,
- onMessage(message) {
- console.debug('sse message', message)
- if (message.trim() === '[DONE]') {
- finishMessage()
- return
- }
- let data
- try {
- data = JSON.parse(message)
- } catch (error) {
- console.debug('json error', error)
- return
- }
- handleMessage(data)
- },
- async onStart() {
- // sendModerations(accessToken, question, session.conversationId, session.messageId)
- },
- async onEnd() {
- port.postMessage({ done: true })
- cleanController()
- },
- async onError(resp) {
+ try {
+ const { conversationId, wsRequestId } = await sendWebsocketConversation(accessToken, options)
+ if (controller.signal.aborted) {
+ removeWebsocketCallback()
cleanController()
- if (resp instanceof Error) throw resp
- if (resp.status === 403) {
- throw new Error('CLOUDFLARE')
- }
- const error = await resp.json().catch(() => ({}))
- throw new Error(
- !isEmpty(error) ? JSON.stringify(error) : `${resp.status} ${resp.statusText}`,
- )
- },
- })
+ stopWebsocketRequest(conversationId, wsRequestId)
+ return
+ }
+ session.conversationId = conversationId
+ session.wsRequestId = wsRequestId
+ port.postMessage({ session: session })
+ } catch (error) {
+ removeWebsocketCallback()
+ cleanController()
+ if (controller.signal.aborted) return
+ throw error
+ }
+ } else {
+ try {
+ await fetchSSE(url, {
+ ...options,
+ onMessage(message) {
+ console.debug('sse message', message)
+ if (message.trim() === '[DONE]') {
+ finishMessage()
+ return
+ }
+ let data
+ try {
+ data = JSON.parse(message)
+ } catch (error) {
+ console.debug('json error', error)
+ return
+ }
+ handleMessage(data)
+ },
+ async onStart() {
+ // sendModerations(accessToken, question, session.conversationId, session.messageId)
+ },
+ async onEnd(aborted) {
+ try {
+ if (!aborted) {
+ port.postMessage({ done: true })
+ }
+ } finally {
+ cleanController()
+ }
+ },
+ async onError(resp) {
+ cleanController()
+ if (resp instanceof Error) throw resp
+ if (resp.status === 403) {
+ throw new Error('CLOUDFLARE')
+ }
+ const error = await resp.json().catch(() => ({}))
+ throw new Error(
+ !isEmpty(error) ? JSON.stringify(error) : `${resp.status} ${resp.statusText}`,
+ )
+ },
+ })
+ } finally {
+ cleanController()
+ }
}
function handleMessage(data) {
+ if (controller.signal.aborted) return
if (data.error) {
throw new Error(JSON.stringify(data.error))
}
@@ -442,6 +490,9 @@ export async function generateAnswersWithChatgptWebApi(port, question, session,
}
function finishMessage() {
+ removeWebsocketCallback()
+ cleanController()
+ if (controller.signal.aborted) return
pushRecord(session, question, answer)
console.debug('conversation history', { content: session.conversationRecords })
port.postMessage({ answer: answer, done: true, session: session })
diff --git a/src/services/apis/claude-api.mjs b/src/services/apis/claude-api.mjs
index 648720c01..341a4dd9d 100644
--- a/src/services/apis/claude-api.mjs
+++ b/src/services/apis/claude-api.mjs
@@ -78,10 +78,15 @@ export async function generateAnswersWithClaudeApi(port, question, session) {
}
},
async onStart() {},
- async onEnd() {
- port.postMessage({ done: true })
- port.onMessage.removeListener(messageListener)
- port.onDisconnect.removeListener(disconnectListener)
+ async onEnd(aborted) {
+ try {
+ if (!aborted) {
+ port.postMessage({ done: true })
+ }
+ } finally {
+ port.onMessage.removeListener(messageListener)
+ port.onDisconnect.removeListener(disconnectListener)
+ }
},
async onError(resp) {
port.onMessage.removeListener(messageListener)
diff --git a/src/services/apis/claude-web.mjs b/src/services/apis/claude-web.mjs
index 2c5307f0f..0573b9c7b 100644
--- a/src/services/apis/claude-web.mjs
+++ b/src/services/apis/claude-web.mjs
@@ -9,9 +9,19 @@ import { getModelValue } from '../../utils/model-name-convert.mjs'
* @param {string} sessionKey
*/
export async function generateAnswersWithClaudeWebApi(port, question, session, sessionKey) {
- const bot = new Claude({ sessionKey })
- await bot.init()
const { controller, cleanController } = setAbortController(port)
+ let bot
+ try {
+ bot = new Claude({ sessionKey })
+ await bot.init()
+ } catch (error) {
+ cleanController()
+ throw error
+ }
+ if (controller.signal.aborted) {
+ cleanController()
+ return
+ }
const model = getModelValue(session)
let answer = ''
diff --git a/src/services/apis/moonshot-web.mjs b/src/services/apis/moonshot-web.mjs
index 2f6334386..2f2bfdd6d 100644
--- a/src/services/apis/moonshot-web.mjs
+++ b/src/services/apis/moonshot-web.mjs
@@ -1,6 +1,6 @@
import { pushRecord, setAbortController } from './shared.mjs'
import { setUserConfig } from '../../config/index.mjs'
-import { fetchSSE } from '../../utils/fetch-sse'
+import { fetchSSE } from '../../utils/fetch-sse.mjs'
import { isEmpty } from 'lodash-es'
import { getModelValue } from '../../utils/model-name-convert.mjs'
@@ -134,7 +134,7 @@ export class MoonshotWeb {
* @async
* @returns {Promise
} Void
*/
- async init() {
+ async init(signal) {
const response = this.request('/api/user', {
headers: {
accept: '*/*',
@@ -142,6 +142,7 @@ export class MoonshotWeb {
Origin: 'https://www.kimi.com',
},
method: 'GET',
+ signal,
})
if ((await response).status === 200) {
this.ready = true
@@ -153,6 +154,7 @@ export class MoonshotWeb {
Origin: 'https://www.kimi.com',
},
method: 'GET',
+ signal,
})
.then((r) => r.json())
.catch(errorHandle('get kimi.moonshoot.cn access_token'))
@@ -426,7 +428,11 @@ export class Conversation {
}
},
async onStart() {},
- async onEnd() {
+ async onEnd(aborted = false) {
+ if (aborted) {
+ reject(new DOMException('Aborted', 'AbortError'))
+ return
+ }
resolve({
completion: fullResponse,
})
@@ -579,9 +585,20 @@ export class Message {
* @param {UserConfig} config
*/
export async function generateAnswersWithMoonshotWebApi(port, question, session, config) {
- const bot = new MoonshotWeb({ config })
- await bot.init()
const { controller, cleanController } = setAbortController(port)
+ let bot
+ try {
+ bot = new MoonshotWeb({ config })
+ await bot.init(controller.signal)
+ } catch (error) {
+ cleanController()
+ if (controller.signal.aborted) return
+ throw error
+ }
+ if (controller.signal.aborted) {
+ cleanController()
+ return
+ }
const model = getModelValue(session)
let answer = ''
diff --git a/src/services/apis/openai-compatible-core.mjs b/src/services/apis/openai-compatible-core.mjs
index 86d08500d..f65142667 100644
--- a/src/services/apis/openai-compatible-core.mjs
+++ b/src/services/apis/openai-compatible-core.mjs
@@ -60,7 +60,13 @@ export async function generateAnswersWithOpenAICompatible({
extraHeaders = {},
allowLegacyResponseField = false,
}) {
- const { controller, messageListener, disconnectListener } = setAbortController(port)
+ const {
+ controller,
+ messageListener,
+ disconnectListener,
+ getStopGenerationId,
+ isCurrentSessionRequest,
+ } = setAbortController(port)
let requestBody
const conversationRecords = Array.isArray(session.conversationRecords)
@@ -142,12 +148,34 @@ export async function generateAnswersWithOpenAICompatible({
}
},
async onStart() {},
- async onEnd() {
- if (!finished) {
- finish()
+ async onEnd(aborted = false) {
+ try {
+ if (!finished) {
+ if (aborted) {
+ const shouldPostSession = Boolean(answer) || session.isRetry
+ if (shouldPostSession && isCurrentSessionRequest()) {
+ if (answer) {
+ pushRecord(session, question, answer)
+ }
+ session.isRetry = false
+ try {
+ const stoppedGenerationId = getStopGenerationId()
+ port.postMessage({
+ session,
+ ...(stoppedGenerationId === undefined ? {} : { stoppedGenerationId }),
+ })
+ } catch (e) {
+ console.warn('[openai-compatible-core] Failed to post session on abort:', e)
+ }
+ }
+ } else {
+ finish()
+ }
+ }
+ } finally {
+ port.onMessage.removeListener(messageListener)
+ port.onDisconnect.removeListener(disconnectListener)
}
- port.onMessage.removeListener(messageListener)
- port.onDisconnect.removeListener(disconnectListener)
},
async onError(resp) {
port.onMessage.removeListener(messageListener)
diff --git a/src/services/apis/shared.mjs b/src/services/apis/shared.mjs
index 77fcab618..a425305f1 100644
--- a/src/services/apis/shared.mjs
+++ b/src/services/apis/shared.mjs
@@ -15,13 +15,33 @@ export const getCustomApiPromptBase = async () => {
return `I am a helpful, creative, clever, and very friendly assistant. I am familiar with various languages in the world.`
}
+export function acknowledgePortStop(port, message) {
+ if (message.stopAcknowledged || port._stopAcknowledged) return false
+ try {
+ port.postMessage({
+ done: true,
+ ...(message.stopGenerationId === undefined
+ ? {}
+ : { stoppedGenerationId: message.stopGenerationId }),
+ })
+ } catch (e) {
+ return false
+ }
+ port._stopAcknowledged = true
+ message.stopAcknowledged = true
+ return true
+}
+
export function setAbortController(port, onStop, onDisconnect) {
const controller = new AbortController()
+ const sessionRequestGeneration = port._sessionRequestGeneration
+ let stopGenerationId
const messageListener = (msg) => {
if (msg.stop) {
+ stopGenerationId = msg.stopGenerationId
port.onMessage.removeListener(messageListener)
console.debug('stop generating')
- port.postMessage({ done: true })
+ acknowledgePortStop(port, msg)
controller.abort()
if (onStop) onStop()
}
@@ -45,7 +65,14 @@ export function setAbortController(port, onStop, onDisconnect) {
}
}
- return { controller, cleanController, messageListener, disconnectListener }
+ return {
+ controller,
+ cleanController,
+ messageListener,
+ disconnectListener,
+ getStopGenerationId: () => stopGenerationId,
+ isCurrentSessionRequest: () => port._sessionRequestGeneration === sessionRequestGeneration,
+ }
}
export function pushRecord(session, question, answer) {
diff --git a/src/services/apis/waylaidwanderer-api.mjs b/src/services/apis/waylaidwanderer-api.mjs
index d30603488..a07cdf1bf 100644
--- a/src/services/apis/waylaidwanderer-api.mjs
+++ b/src/services/apis/waylaidwanderer-api.mjs
@@ -63,10 +63,15 @@ export async function generateAnswersWithWaylaidwandererApi(port, question, sess
}
},
async onStart() {},
- async onEnd() {
- port.postMessage({ done: true })
- port.onMessage.removeListener(messageListener)
- port.onDisconnect.removeListener(disconnectListener)
+ async onEnd(aborted) {
+ try {
+ if (!aborted) {
+ port.postMessage({ done: true })
+ }
+ } finally {
+ port.onMessage.removeListener(messageListener)
+ port.onDisconnect.removeListener(disconnectListener)
+ }
},
async onError(resp) {
port.onMessage.removeListener(messageListener)
diff --git a/src/services/clients/claude/index.mjs b/src/services/clients/claude/index.mjs
index f21652fad..02b9b1117 100644
--- a/src/services/clients/claude/index.mjs
+++ b/src/services/clients/claude/index.mjs
@@ -627,7 +627,11 @@ export class Conversation {
}
},
async onStart() {},
- async onEnd() {
+ async onEnd(aborted = false) {
+ if (aborted) {
+ reject(new DOMException('Aborted', 'AbortError'))
+ return
+ }
resolve({
completion: fullResponse,
})
diff --git a/src/services/wrappers.mjs b/src/services/wrappers.mjs
index 71dce351a..163a73138 100644
--- a/src/services/wrappers.mjs
+++ b/src/services/wrappers.mjs
@@ -12,6 +12,7 @@ import {
modelNameToDesc,
normalizeApiMode,
} from '../utils/model-name-convert.mjs'
+import { acknowledgePortStop } from './apis/shared.mjs'
export async function getChatGptAccessToken() {
await clearOldAccessToken()
@@ -66,9 +67,31 @@ function isAbortError(err) {
return name === 'AbortError' || message.includes('aborted') || message.includes('aborterror')
}
+function isDisconnectedPortError(err) {
+ if (!err || typeof err !== 'object') return false
+ const message =
+ typeof err.message === 'string' ? err.message.trim().toLowerCase().replace(/\.$/, '') : ''
+ return (
+ message === 'attempting to use a disconnected port object' ||
+ message === 'attempt to postmessage on disconnected port' ||
+ message === 'extension context invalidated'
+ )
+}
+
export function handlePortError(session, port, err) {
if (isAbortError(err)) return
+ if (isDisconnectedPortError(err)) {
+ console.warn('[handlePortError] Ignoring disconnected port error:', err.message)
+ return
+ }
console.error(err)
+ const postError = (error) => {
+ try {
+ port.postMessage({ error })
+ } catch (postErr) {
+ console.warn('[handlePortError] Failed to post error:', postErr)
+ }
+ }
const message = typeof err?.message === 'string' ? err.message : ''
if (message) {
if (
@@ -76,46 +99,87 @@ export function handlePortError(session, port, err) {
message.includes(m),
)
)
- port.postMessage({ error: t('Exceeded maximum context length') + '\n\n' + message })
+ postError(t('Exceeded maximum context length') + '\n\n' + message)
else if (['CaptchaChallenge', 'CAPTCHA'].some((m) => message.includes(m)))
- port.postMessage({ error: t('Bing CaptchaChallenge') + '\n\n' + message })
+ postError(t('Bing CaptchaChallenge') + '\n\n' + message)
else if (['exceeded your current quota'].some((m) => message.includes(m)))
- port.postMessage({ error: t('Exceeded quota') + '\n\n' + message })
+ postError(t('Exceeded quota') + '\n\n' + message)
else if (['Rate limit reached'].some((m) => message.includes(m)))
- port.postMessage({ error: t('Rate limit') + '\n\n' + message })
+ postError(t('Rate limit') + '\n\n' + message)
else if (['authentication token has expired'].some((m) => message.includes(m)))
- port.postMessage({ error: 'UNAUTHORIZED' })
+ postError('UNAUTHORIZED')
else if (
isUsingClaudeWebModel(session) &&
['Invalid authorization', 'Session key required'].some((m) => message.includes(m))
)
- port.postMessage({
- error: t('Please login at https://claude.ai first, and then click the retry button'),
- })
+ postError(t('Please login at https://claude.ai first, and then click the retry button'))
else if (
isUsingBingWebModel(session) &&
['/turing/conversation/create: failed to parse response body.'].some((m) =>
message.includes(m),
)
)
- port.postMessage({ error: t('Please login at https://bing.com first') })
- else port.postMessage({ error: message })
+ postError(t('Please login at https://bing.com first'))
+ else postError(message)
} else {
const errMsg = JSON.stringify(err) ?? 'unknown error'
if (isUsingBingWebModel(session) && errMsg.includes('isTrusted'))
- port.postMessage({ error: t('Please login at https://bing.com first') })
- else port.postMessage({ error: errMsg })
+ postError(t('Please login at https://bing.com first'))
+ else postError(errMsg)
}
}
+export function claimLatestPortSessionRequest(port) {
+ const requestId = (port._latestSessionRequestId ?? 0) + 1
+ port._latestSessionRequestId = requestId
+ port._sessionRequestGeneration = (port._sessionRequestGeneration ?? 0) + 1
+ port._stopAcknowledged = false
+ return () => port._latestSessionRequestId === requestId
+}
+
+export function invalidateLatestPortSessionRequest(port) {
+ port._latestSessionRequestId = (port._latestSessionRequestId ?? 0) + 1
+}
+
+function createSessionRequestPort(port, proxyGenerationId, requestGenerationId) {
+ const sessionRequestGeneration = port._sessionRequestGeneration
+ return new Proxy(port, {
+ get(target, property, receiver) {
+ if (property === 'postMessage') {
+ return (message) => {
+ if (target._sessionRequestGeneration !== sessionRequestGeneration) return
+ target.postMessage({
+ ...message,
+ ...(proxyGenerationId === undefined ? {} : { proxyGenerationId }),
+ ...(requestGenerationId === undefined ? {} : { requestGenerationId }),
+ })
+ }
+ }
+ return Reflect.get(target, property, receiver)
+ },
+ })
+}
+
export function registerPortListener(executor) {
Browser.runtime.onConnect.addListener((port) => {
console.debug('connected')
const onMessage = async (msg) => {
console.debug('received msg', msg)
+ if (msg.stop) {
+ invalidateLatestPortSessionRequest(port)
+ acknowledgePortStop(port, msg)
+ return
+ }
const session = msg.session
if (!session) return
+ const isLatestSessionRequest = claimLatestPortSessionRequest(port)
+ const requestPort = createSessionRequestPort(
+ port,
+ msg.proxyGenerationId,
+ msg.requestGenerationId,
+ )
const config = await getUserConfig()
+ if (!isLatestSessionRequest()) return
if (!session.modelName) session.modelName = config.modelName
if (!session.apiMode && session.modelName !== 'customModel') session.apiMode = config.apiMode
if (session.apiMode) session.apiMode = normalizeApiMode(session.apiMode)
@@ -125,11 +189,18 @@ export function registerPortListener(executor) {
t,
config.customModelName,
)
- port.postMessage({ session })
+ requestPort.postMessage({ session })
try {
- await executor(session, port, config)
+ await executor(
+ session,
+ requestPort,
+ config,
+ isLatestSessionRequest,
+ msg.requestGenerationId,
+ port,
+ )
} catch (err) {
- handlePortError(session, port, err)
+ if (isLatestSessionRequest()) handlePortError(session, requestPort, err)
}
}
diff --git a/src/utils/fetch-sse.mjs b/src/utils/fetch-sse.mjs
index 059cc0430..06027cee3 100644
--- a/src/utils/fetch-sse.mjs
+++ b/src/utils/fetch-sse.mjs
@@ -1,11 +1,28 @@
import { createParser } from './eventsource-parser.mjs'
+function isAbortError(err) {
+ if (!err || typeof err !== 'object') return false
+ const name = typeof err.name === 'string' ? err.name : ''
+ return name === 'AbortError'
+}
+
export async function fetchSSE(resource, options) {
const { onMessage, onStart, onEnd, onError, ...fetchOptions } = options
- const resp = await fetch(resource, fetchOptions).catch(async (err) => {
+ let resp
+ try {
+ resp = await fetch(resource, fetchOptions)
+ } catch (err) {
+ if (isAbortError(err)) {
+ try {
+ await onEnd(true)
+ } catch (e) {
+ console.warn('[fetch-sse] onEnd threw during abort:', e)
+ }
+ return
+ }
await onError(err)
- })
- if (!resp) return
+ return
+ }
if (!resp.ok) {
await onError(resp)
return
@@ -15,15 +32,42 @@ export async function fetchSSE(resource, options) {
onMessage(event.data)
}
})
+ const handleCallbackError = async (err) => {
+ await onError(err)
+ throw err
+ }
let hasStarted = false
const reader = resp.body.getReader()
let result
- while (!(result = await reader.read()).done) {
+ let done = false
+ while (!done) {
+ try {
+ result = await reader.read()
+ } catch (err) {
+ if (isAbortError(err)) {
+ try {
+ await onEnd(true)
+ } catch (e) {
+ console.warn('[fetch-sse] onEnd threw during abort:', e)
+ }
+ return
+ }
+ await onError(err)
+ return
+ }
+
+ done = result.done
+ if (done) break
+
const chunk = result.value
if (!hasStarted) {
const str = new TextDecoder().decode(chunk)
hasStarted = true
- await onStart(str)
+ try {
+ await onStart(str)
+ } catch (err) {
+ await handleCallbackError(err)
+ }
let fakeSseData
try {
@@ -33,11 +77,19 @@ export async function fetchSSE(resource, options) {
console.debug('not common response', error)
}
if (fakeSseData) {
- parser.feed(new TextEncoder().encode(fakeSseData))
+ try {
+ parser.feed(new TextEncoder().encode(fakeSseData))
+ } catch (err) {
+ await handleCallbackError(err)
+ }
break
}
}
- parser.feed(chunk)
+ try {
+ parser.feed(chunk)
+ } catch (err) {
+ await handleCallbackError(err)
+ }
}
await onEnd()
}
diff --git a/tests/unit/background/proxy-generation-state.test.mjs b/tests/unit/background/proxy-generation-state.test.mjs
new file mode 100644
index 000000000..b68aa3586
--- /dev/null
+++ b/tests/unit/background/proxy-generation-state.test.mjs
@@ -0,0 +1,187 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import {
+ clearProxyReconnectErrorSuppression,
+ consumeProxyReconnectErrorSuppression,
+ forwardProxyMessage,
+ interruptProxyGeneration,
+ isCurrentProxyGenerationMessage,
+ markProxyGenerationFinished,
+ markProxyGenerationFinishedFromMessage,
+ markProxyGenerationStarted,
+ shouldSkipProxyReconnect,
+ tagProxyRequestGeneration,
+} from '../../../src/background/proxy-generation-state.mjs'
+
+test('only the current proxy generation can update generation state', () => {
+ const port = {
+ proxy: {
+ postMessage() {},
+ },
+ }
+ markProxyGenerationStarted(port, 2, 5)
+
+ assert.equal(isCurrentProxyGenerationMessage(port, { done: true, proxyGenerationId: 2 }), true)
+ assert.deepEqual(tagProxyRequestGeneration(port, { done: true }), {
+ done: true,
+ requestGenerationId: 5,
+ })
+ assert.equal(isCurrentProxyGenerationMessage(port, { done: true, proxyGenerationId: 1 }), false)
+ assert.equal(
+ markProxyGenerationFinishedFromMessage(port, { done: true, proxyGenerationId: 1 }),
+ false,
+ )
+ assert.equal(port._generating, true)
+ assert.equal(isCurrentProxyGenerationMessage(port, { done: true }), false)
+ assert.equal(isCurrentProxyGenerationMessage(port, { done: true, stoppedGenerationId: 1 }), true)
+
+ forwardProxyMessage(port, { stop: true })
+
+ assert.equal(isCurrentProxyGenerationMessage(port, { done: true, proxyGenerationId: 2 }), false)
+})
+
+test('interrupted generation suppresses only its next reconnect error', () => {
+ const port = {}
+
+ markProxyGenerationStarted(port)
+
+ assert.equal(interruptProxyGeneration(port), true)
+ assert.equal(port._generating, false)
+ assert.equal(interruptProxyGeneration(port), false)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), true)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), false)
+})
+
+test('retry stop messages do not finish the replacement generation', () => {
+ const port = {}
+
+ for (const message of [
+ { done: true, stoppedGenerationId: 3 },
+ { error: 'stopped', stoppedGenerationId: 3 },
+ ]) {
+ markProxyGenerationStarted(port)
+
+ assert.equal(markProxyGenerationFinishedFromMessage(port, message), false)
+ assert.equal(port._generating, true)
+ }
+})
+
+test('retry stop finishes the old generation before its acknowledgement arrives', () => {
+ const postedMessages = []
+ const port = {
+ proxy: {
+ postMessage(message) {
+ postedMessages.push(message)
+ },
+ },
+ }
+
+ markProxyGenerationStarted(port)
+ forwardProxyMessage(port, { stop: true, stopGenerationId: 3 })
+
+ assert.deepEqual(postedMessages, [{ stop: true, stopGenerationId: 3 }])
+ assert.equal(port._generating, false)
+ assert.equal(
+ markProxyGenerationFinishedFromMessage(port, {
+ done: true,
+ stoppedGenerationId: 3,
+ }),
+ false,
+ )
+ assert.equal(port._generating, false)
+})
+
+test('failed retry stop forwarding still interrupts the old generation', () => {
+ const port = {
+ proxy: {
+ postMessage() {
+ throw new Error('disconnected')
+ },
+ },
+ }
+
+ markProxyGenerationStarted(port)
+
+ assert.throws(
+ () => forwardProxyMessage(port, { stop: true, stopGenerationId: 3 }),
+ /disconnected/,
+ )
+ assert.equal(port._generating, false)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), true)
+})
+
+test('forwardProxyMessage preserves an upstream stop acknowledgement across the proxy', () => {
+ const postedMessages = []
+ const port = {
+ _stopAcknowledged: true,
+ proxy: {
+ postMessage(message) {
+ postedMessages.push(message)
+ },
+ },
+ }
+
+ markProxyGenerationStarted(port)
+ forwardProxyMessage(port, { stop: true })
+
+ assert.deepEqual(postedMessages, [{ stop: true, stopAcknowledged: true }])
+ assert.equal(port._generating, false)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), true)
+})
+
+test('forwarded stop preserves reconnect error suppression from an interrupted generation', () => {
+ const port = {
+ proxy: {
+ postMessage() {},
+ },
+ }
+
+ markProxyGenerationStarted(port)
+ interruptProxyGeneration(port)
+ forwardProxyMessage(port, { stop: true })
+
+ assert.equal(port._generating, false)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), true)
+})
+
+test('normal completion and errors finish the active proxy generation', () => {
+ for (const message of [{ done: true }, { error: 'failed' }]) {
+ const port = {}
+ markProxyGenerationStarted(port)
+
+ assert.equal(markProxyGenerationFinishedFromMessage(port, message), true)
+ assert.equal(port._generating, false)
+ }
+})
+
+test('new generation clears reconnect error suppression from an interrupted generation', () => {
+ const port = {}
+
+ markProxyGenerationStarted(port)
+ interruptProxyGeneration(port)
+ markProxyGenerationStarted(port)
+
+ assert.equal(port._generating, true)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), false)
+})
+
+test('stable reconnect and normal completion clear reconnect error suppression', () => {
+ const port = {}
+
+ markProxyGenerationStarted(port)
+ interruptProxyGeneration(port)
+ clearProxyReconnectErrorSuppression(port)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), false)
+
+ markProxyGenerationStarted(port)
+ interruptProxyGeneration(port)
+ markProxyGenerationFinished(port)
+ assert.equal(port._generating, false)
+ assert.equal(consumeProxyReconnectErrorSuppression(port), false)
+})
+
+test('active or closed ports skip stale reconnect callbacks', () => {
+ assert.equal(shouldSkipProxyReconnect({}), false)
+ assert.equal(shouldSkipProxyReconnect({ proxy: {} }), true)
+ assert.equal(shouldSkipProxyReconnect({ _isClosed: true }), true)
+})
diff --git a/tests/unit/components/conversation-card-session.test.mjs b/tests/unit/components/conversation-card-session.test.mjs
new file mode 100644
index 000000000..fdf543c72
--- /dev/null
+++ b/tests/unit/components/conversation-card-session.test.mjs
@@ -0,0 +1,209 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import {
+ createConversationPortMessage,
+ createRetrySession,
+ finalizeInterruptedSession,
+ getCompletedAnswerUpdate,
+ getInterruptedCompletionState,
+ isSupersededGenerationMessage,
+ isSupersededRequestMessage,
+} from '../../../src/components/ConversationCard/session.mjs'
+
+test('finalizeInterruptedSession appends a partial answer without mutating the source session', () => {
+ const session = {
+ question: 'Q1',
+ isRetry: false,
+ conversationRecords: [],
+ }
+
+ const updatedSession = finalizeInterruptedSession(session, 'Partial answer')
+
+ assert.notEqual(updatedSession, session)
+ assert.deepEqual(session.conversationRecords, [])
+ assert.deepEqual(updatedSession.conversationRecords, [
+ { question: 'Q1', answer: 'Partial answer' },
+ ])
+})
+
+test('finalizeInterruptedSession replaces a retry answer and clears retry state', () => {
+ const session = {
+ question: 'Q1',
+ isRetry: true,
+ conversationRecords: [{ question: 'Q1', answer: 'Old answer' }],
+ }
+
+ const updatedSession = finalizeInterruptedSession(session, 'Partial retry answer')
+
+ assert.equal(session.isRetry, true)
+ assert.deepEqual(session.conversationRecords, [{ question: 'Q1', answer: 'Old answer' }])
+ assert.equal(updatedSession.isRetry, false)
+ assert.deepEqual(updatedSession.conversationRecords, [
+ { question: 'Q1', answer: 'Partial retry answer' },
+ ])
+})
+
+test('finalizeInterruptedSession appends after existing records when not retrying', () => {
+ const session = {
+ question: 'Q2',
+ isRetry: false,
+ conversationRecords: [{ question: 'Q1', answer: 'A1' }],
+ }
+
+ const updatedSession = finalizeInterruptedSession(session, 'Partial answer')
+
+ assert.deepEqual(session.conversationRecords, [{ question: 'Q1', answer: 'A1' }])
+ assert.deepEqual(updatedSession.conversationRecords, [
+ { question: 'Q1', answer: 'A1' },
+ { question: 'Q2', answer: 'Partial answer' },
+ ])
+})
+
+test('finalizeInterruptedSession restores a retry answer when interrupted before tokens arrive', () => {
+ const session = {
+ question: 'Q1',
+ isRetry: true,
+ conversationRecords: [],
+ }
+ const retryRecord = { question: 'Q1', answer: 'Old answer' }
+
+ const updatedSession = finalizeInterruptedSession(session, '', retryRecord)
+
+ assert.equal(session.isRetry, true)
+ assert.deepEqual(session.conversationRecords, [])
+ assert.equal(updatedSession.isRetry, false)
+ assert.deepEqual(updatedSession.conversationRecords, [{ question: 'Q1', answer: 'Old answer' }])
+})
+
+test('finalizeInterruptedSession restores a retry record even before retry state commits', () => {
+ const retryRecord = { question: 'Q1', answer: 'Old answer' }
+ const session = {
+ question: 'Q1',
+ isRetry: false,
+ conversationRecords: [],
+ }
+
+ const updatedSession = finalizeInterruptedSession(session, '', retryRecord)
+
+ assert.equal(updatedSession.isRetry, false)
+ assert.deepEqual(updatedSession.conversationRecords, [retryRecord])
+})
+
+test('finalizeInterruptedSession clears retry state when no record was stashed', () => {
+ const session = {
+ question: 'Q1',
+ isRetry: true,
+ conversationRecords: [],
+ }
+
+ const updatedSession = finalizeInterruptedSession(session, '')
+
+ assert.equal(updatedSession.isRetry, false)
+ assert.deepEqual(updatedSession.conversationRecords, [])
+})
+
+test('finalizeInterruptedSession does not duplicate an already restored retry record', () => {
+ const retryRecord = { question: 'Q1', answer: 'Old answer' }
+ const session = {
+ question: 'Q1',
+ isRetry: false,
+ conversationRecords: [retryRecord],
+ }
+
+ const updatedSession = finalizeInterruptedSession(session, '', retryRecord)
+
+ assert.deepEqual(updatedSession.conversationRecords, [retryRecord])
+})
+
+test('isSupersededGenerationMessage ignores only stopped generations', () => {
+ assert.equal(isSupersededGenerationMessage({ done: true }, 3), false)
+ assert.equal(isSupersededGenerationMessage({ done: true, stoppedGenerationId: 4 }, 3), false)
+ assert.equal(isSupersededGenerationMessage({ done: true, stoppedGenerationId: 3 }, 3), true)
+ assert.equal(isSupersededGenerationMessage({ session: {}, stoppedGenerationId: 2 }, 3), true)
+})
+
+test('isSupersededRequestMessage ignores responses from older requests', () => {
+ assert.equal(isSupersededRequestMessage({ done: true, requestGenerationId: 2 }, 3), true)
+ assert.equal(isSupersededRequestMessage({ done: true, requestGenerationId: 3 }, 3), false)
+ assert.equal(isSupersededRequestMessage({ done: true }, 3), false)
+})
+
+test('createConversationPortMessage preserves retry stop acknowledgement identity', () => {
+ assert.deepEqual(createConversationPortMessage({ stop: true, stopGenerationId: 3 }), {
+ session: undefined,
+ stop: true,
+ stopGenerationId: 3,
+ })
+})
+
+test('createConversationPortMessage keeps the existing payload for normal messages', () => {
+ const session = { question: 'Q1' }
+
+ assert.deepEqual(createConversationPortMessage({ session, requestGenerationId: 4 }), {
+ session,
+ stop: undefined,
+ requestGenerationId: 4,
+ })
+})
+
+test('createRetrySession appends after a removed retry target', () => {
+ const conversationRecords = [{ question: 'Q1', answer: 'A1' }]
+
+ const retrySession = createRetrySession(
+ { question: 'Q1', isRetry: true, conversationRecords: [] },
+ conversationRecords,
+ { question: 'Q1', answer: 'Old answer' },
+ )
+
+ assert.equal(retrySession.isRetry, false)
+ assert.equal(retrySession.conversationRecords, conversationRecords)
+})
+
+test('createRetrySession keeps provider retry mode when no target was removed', () => {
+ const retrySession = createRetrySession(
+ { question: 'Q1', isRetry: false, conversationRecords: [] },
+ [],
+ null,
+ )
+
+ assert.equal(retrySession.isRetry, true)
+})
+
+test('getCompletedAnswerUpdate replaces loading content when restoring a retry answer', () => {
+ assert.deepEqual(getCompletedAnswerUpdate('Old answer'), {
+ value: 'Old answer',
+ appended: false,
+ })
+})
+
+test('getCompletedAnswerUpdate preserves streamed content on normal completion', () => {
+ assert.deepEqual(getCompletedAnswerUpdate(null), {
+ value: '',
+ appended: true,
+ })
+})
+
+test('getInterruptedCompletionState restores retry context for a sessionless stop', () => {
+ const retryRecord = { question: 'Q1', answer: 'Old answer' }
+
+ assert.deepEqual(getInterruptedCompletionState({ done: true }, '', retryRecord), {
+ shouldFinalize: true,
+ restoredRetryAnswer: 'Old answer',
+ })
+})
+
+test('getInterruptedCompletionState finalizes on proxy disconnect with a session', () => {
+ const retryRecord = { question: 'Q1', answer: 'Old answer' }
+
+ assert.deepEqual(
+ getInterruptedCompletionState(
+ { done: true, proxyDisconnected: true, session: {} },
+ '',
+ retryRecord,
+ ),
+ {
+ shouldFinalize: true,
+ restoredRetryAnswer: 'Old answer',
+ },
+ )
+})
diff --git a/tests/unit/services/apis/bard-web.test.mjs b/tests/unit/services/apis/bard-web.test.mjs
new file mode 100644
index 000000000..ef7a4366b
--- /dev/null
+++ b/tests/unit/services/apis/bard-web.test.mjs
@@ -0,0 +1,31 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { generateAnswersWithBardWebApi } from '../../../../src/services/apis/bard-web.mjs'
+import Bard from '../../../../src/services/clients/bard/index.mjs'
+import { createFakePort } from '../../helpers/port.mjs'
+
+test('generateAnswersWithBardWebApi ignores a superseded response', async (t) => {
+ let resolveResponse
+ const response = new Promise((resolve) => {
+ resolveResponse = resolve
+ })
+ t.mock.method(Bard.prototype, 'ask', () => response)
+
+ const session = { conversationRecords: [] }
+ const port = createFakePort()
+ let isLatest = true
+ const generation = generateAnswersWithBardWebApi(
+ port,
+ 'CurrentQ',
+ session,
+ 'cookie',
+ () => isLatest,
+ )
+
+ isLatest = false
+ resolveResponse({ answer: 'Stale answer', conversationObj: { id: 'stale' } })
+ await generation
+
+ assert.deepEqual(session, { conversationRecords: [] })
+ assert.deepEqual(port.postedMessages, [])
+})
diff --git a/tests/unit/services/apis/moonshot-web.test.mjs b/tests/unit/services/apis/moonshot-web.test.mjs
new file mode 100644
index 000000000..4e9b47008
--- /dev/null
+++ b/tests/unit/services/apis/moonshot-web.test.mjs
@@ -0,0 +1,36 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { MoonshotWeb } from '../../../../src/services/apis/moonshot-web.mjs'
+
+test('MoonshotWeb.init forwards its signal to user and refresh requests', async () => {
+ const controller = new AbortController()
+ const requests = []
+ const config = {
+ kimiMoonShotAccessToken: 'old-access-token',
+ kimiMoonShotRefreshToken: 'old-refresh-token',
+ }
+ const bot = new MoonshotWeb({
+ config,
+ fetch: async (endpoint, options) => {
+ requests.push({ endpoint, options })
+ if (endpoint.endsWith('/api/user')) return { status: 401 }
+ return {
+ json: async () => ({
+ access_token: 'new-access-token',
+ refresh_token: 'new-refresh-token',
+ }),
+ }
+ },
+ })
+
+ await bot.init(controller.signal)
+
+ assert.deepEqual(
+ requests.map(({ endpoint }) => endpoint),
+ ['https://www.kimi.com/api/user', 'https://www.kimi.com/api/auth/token/refresh'],
+ )
+ assert.equal(
+ requests.every(({ options }) => options.signal === controller.signal),
+ true,
+ )
+})
diff --git a/tests/unit/services/apis/openai-api-compat.test.mjs b/tests/unit/services/apis/openai-api-compat.test.mjs
index d4c6494f9..f4a7f96ca 100644
--- a/tests/unit/services/apis/openai-api-compat.test.mjs
+++ b/tests/unit/services/apis/openai-api-compat.test.mjs
@@ -6,6 +6,7 @@ import {
generateAnswersWithGptCompletionApi,
generateAnswersWithOpenAICompatibleApi,
} from '../../../../src/services/apis/openai-api.mjs'
+import { claimLatestPortSessionRequest } from '../../../../src/services/wrappers.mjs'
import { createFakePort } from '../../helpers/port.mjs'
import { createMockSseResponse } from '../../helpers/sse-response.mjs'
@@ -28,6 +29,36 @@ const setStorage = (values) => {
globalThis.__TEST_BROWSER_SHIM__.replaceStorage(values)
}
+const createStoppedSseResponse = (port, afterStop) => {
+ const encoder = new TextEncoder()
+ let readCount = 0
+ return {
+ ok: true,
+ status: 200,
+ statusText: 'OK',
+ body: {
+ getReader() {
+ return {
+ async read() {
+ readCount += 1
+ if (readCount === 1) {
+ return {
+ done: false,
+ value: encoder.encode('data: {"choices":[{"delta":{"content":"Partial"}}]}\n\n'),
+ }
+ }
+ port.emitMessage({ stop: true })
+ afterStop?.()
+ throw Object.assign(new Error('The operation was aborted'), {
+ name: 'AbortError',
+ })
+ },
+ }
+ },
+ },
+ }
+}
+
beforeEach(() => {
globalThis.__TEST_BROWSER_SHIM__.clearStorage()
})
@@ -138,6 +169,188 @@ test('generateAnswersWithOpenAiApiCompat emits fallback done message when stream
})
})
+test('generateAnswersWithOpenAiApiCompat preserves partial retry answer when aborted port is closed', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ t.mock.method(console, 'warn', () => {})
+ setStorage({
+ maxConversationContextLength: 3,
+ maxResponseTokenLength: 256,
+ temperature: 0.25,
+ })
+
+ const session = {
+ modelName: 'chatgptApi4oMini',
+ conversationRecords: [{ question: 'CurrentQ', answer: 'Old answer' }],
+ isRetry: true,
+ }
+ const port = createFakePort()
+ const originalPostMessage = port.postMessage.bind(port)
+ port.postMessage = (message) => {
+ if (message?.session && !message.done) {
+ throw new Error('Port closed')
+ }
+ originalPostMessage(message)
+ }
+
+ t.mock.method(globalThis, 'fetch', async () => {
+ const encoder = new TextEncoder()
+ let readCount = 0
+ return {
+ ok: true,
+ status: 200,
+ statusText: 'OK',
+ body: {
+ getReader() {
+ return {
+ async read() {
+ readCount += 1
+ if (readCount === 1) {
+ return {
+ done: false,
+ value: encoder.encode('data: {"choices":[{"delta":{"content":"Partial"}}]}\n\n'),
+ }
+ }
+ throw Object.assign(new Error('The operation was aborted'), {
+ name: 'AbortError',
+ })
+ },
+ }
+ },
+ },
+ }
+ })
+
+ await generateAnswersWithOpenAiApiCompat(
+ 'https://api.example.com/v1',
+ port,
+ 'CurrentQ',
+ session,
+ 'sk-test',
+ )
+
+ assert.equal(
+ port.postedMessages.some((message) => message.done === false && message.answer === 'Partial'),
+ true,
+ )
+ assert.deepEqual(session.conversationRecords, [{ question: 'CurrentQ', answer: 'Partial' }])
+ assert.equal(session.isRetry, false)
+ assert.deepEqual(port.listenerCounts(), { onMessage: 0, onDisconnect: 0 })
+})
+
+test('generateAnswersWithOpenAiApiCompat clears retry state when aborted before first chunk', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ setStorage({
+ maxConversationContextLength: 3,
+ maxResponseTokenLength: 256,
+ temperature: 0.25,
+ })
+
+ const session = {
+ modelName: 'chatgptApi4oMini',
+ conversationRecords: [{ question: 'CurrentQ', answer: 'Old answer' }],
+ isRetry: true,
+ }
+ const port = createFakePort()
+
+ t.mock.method(globalThis, 'fetch', async () => ({
+ ok: true,
+ status: 200,
+ statusText: 'OK',
+ body: {
+ getReader() {
+ return {
+ async read() {
+ port.emitMessage({ stop: true, stopGenerationId: 7 })
+ throw Object.assign(new Error('The operation was aborted'), {
+ name: 'AbortError',
+ })
+ },
+ }
+ },
+ },
+ }))
+
+ await generateAnswersWithOpenAiApiCompat(
+ 'https://api.example.com/v1',
+ port,
+ 'CurrentQ',
+ session,
+ 'sk-test',
+ )
+
+ assert.deepEqual(session.conversationRecords, [{ question: 'CurrentQ', answer: 'Old answer' }])
+ assert.equal(session.isRetry, false)
+ assert.deepEqual(port.postedMessages.at(-1), { session, stoppedGenerationId: 7 })
+ assert.deepEqual(port.listenerCounts(), { onMessage: 0, onDisconnect: 0 })
+})
+
+test('generateAnswersWithOpenAiApiCompat ignores an aborted session after a newer request starts', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ setStorage({
+ maxConversationContextLength: 3,
+ maxResponseTokenLength: 256,
+ temperature: 0.25,
+ })
+
+ const session = {
+ modelName: 'chatgptApi4oMini',
+ conversationRecords: [],
+ isRetry: false,
+ }
+ const port = createFakePort()
+ claimLatestPortSessionRequest(port)
+
+ t.mock.method(globalThis, 'fetch', async () =>
+ createStoppedSseResponse(port, () => claimLatestPortSessionRequest(port)),
+ )
+
+ await generateAnswersWithOpenAiApiCompat(
+ 'https://api.example.com/v1',
+ port,
+ 'CurrentQ',
+ session,
+ 'sk-test',
+ )
+
+ assert.deepEqual(session.conversationRecords, [])
+ assert.equal(
+ port.postedMessages.some((message) => message.session),
+ false,
+ )
+ assert.deepEqual(port.listenerCounts(), { onMessage: 0, onDisconnect: 0 })
+})
+
+test('generateAnswersWithOpenAiApiCompat preserves an aborted session without a newer request', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ setStorage({
+ maxConversationContextLength: 3,
+ maxResponseTokenLength: 256,
+ temperature: 0.25,
+ })
+
+ const session = {
+ modelName: 'chatgptApi4oMini',
+ conversationRecords: [],
+ isRetry: false,
+ }
+ const port = createFakePort()
+ claimLatestPortSessionRequest(port)
+
+ t.mock.method(globalThis, 'fetch', async () => createStoppedSseResponse(port))
+
+ await generateAnswersWithOpenAiApiCompat(
+ 'https://api.example.com/v1',
+ port,
+ 'CurrentQ',
+ session,
+ 'sk-test',
+ )
+
+ assert.deepEqual(session.conversationRecords, [{ question: 'CurrentQ', answer: 'Partial' }])
+ assert.deepEqual(port.postedMessages.at(-1), { session })
+ assert.deepEqual(port.listenerCounts(), { onMessage: 0, onDisconnect: 0 })
+})
+
test('generateAnswersWithOpenAiApiCompat records an empty answer when stream ends before first chunk', async (t) => {
t.mock.method(console, 'debug', () => {})
setStorage({
diff --git a/tests/unit/services/apis/shared.test.mjs b/tests/unit/services/apis/shared.test.mjs
index f06fa8ec1..981ec61e4 100644
--- a/tests/unit/services/apis/shared.test.mjs
+++ b/tests/unit/services/apis/shared.test.mjs
@@ -1,6 +1,10 @@
import assert from 'node:assert/strict'
import { test } from 'node:test'
-import { pushRecord, setAbortController } from '../../../../src/services/apis/shared.mjs'
+import {
+ acknowledgePortStop,
+ pushRecord,
+ setAbortController,
+} from '../../../../src/services/apis/shared.mjs'
import { createFakePort } from '../../helpers/port.mjs'
test('pushRecord appends a new record in normal mode', () => {
@@ -58,6 +62,34 @@ test('setAbortController aborts and cleans listeners on stop message', (t) => {
assert.deepEqual(port.listenerCounts(), { onMessage: 0, onDisconnect: 1 })
})
+test('setAbortController echoes a retry stop generation id', (t) => {
+ t.mock.method(console, 'debug', () => {})
+ const port = createFakePort()
+ const { getStopGenerationId } = setAbortController(port)
+
+ port.emitMessage({ stop: true, stopGenerationId: 7 })
+
+ assert.deepEqual(port.postedMessages, [{ done: true, stoppedGenerationId: 7 }])
+ assert.equal(getStopGenerationId(), 7)
+})
+
+test('acknowledgePortStop posts only once for the current session request', () => {
+ const port = createFakePort()
+ const message = { stop: true, stopGenerationId: 7 }
+
+ assert.equal(acknowledgePortStop(port, message), true)
+ assert.equal(acknowledgePortStop(port, message), false)
+ assert.equal(message.stopAcknowledged, true)
+ assert.deepEqual(port.postedMessages, [{ done: true, stoppedGenerationId: 7 }])
+})
+
+test('acknowledgePortStop respects an acknowledgement from an upstream port', () => {
+ const port = createFakePort()
+
+ assert.equal(acknowledgePortStop(port, { stop: true, stopAcknowledged: true }), false)
+ assert.deepEqual(port.postedMessages, [])
+})
+
test('setAbortController aborts on disconnect and removes disconnect listener', (t) => {
t.mock.method(console, 'debug', () => {})
const port = createFakePort()
diff --git a/tests/unit/services/handle-port-error.test.mjs b/tests/unit/services/handle-port-error.test.mjs
index 0962e0026..ec3b63226 100644
--- a/tests/unit/services/handle-port-error.test.mjs
+++ b/tests/unit/services/handle-port-error.test.mjs
@@ -1,9 +1,33 @@
import assert from 'node:assert/strict'
import { test } from 'node:test'
import { t as translate } from 'i18next'
-import { handlePortError } from '../../../src/services/wrappers.mjs'
+import {
+ claimLatestPortSessionRequest,
+ handlePortError,
+ invalidateLatestPortSessionRequest,
+} from '../../../src/services/wrappers.mjs'
import { createFakePort } from '../helpers/port.mjs'
+test('claimLatestPortSessionRequest supersedes pending requests on the same port', () => {
+ const port = {}
+ const isFirstRequestLatest = claimLatestPortSessionRequest(port)
+ const isSecondRequestLatest = claimLatestPortSessionRequest(port)
+
+ assert.equal(isFirstRequestLatest(), false)
+ assert.equal(isSecondRequestLatest(), true)
+ assert.equal(port._sessionRequestGeneration, 2)
+})
+
+test('invalidateLatestPortSessionRequest cancels a pending request', () => {
+ const port = {}
+ const isRequestLatest = claimLatestPortSessionRequest(port)
+
+ invalidateLatestPortSessionRequest(port)
+
+ assert.equal(isRequestLatest(), false)
+ assert.equal(port._sessionRequestGeneration, 1)
+})
+
test('handlePortError reports exceeded maximum context length', (t) => {
t.mock.method(console, 'error', () => {})
const port = createFakePort()
@@ -112,6 +136,37 @@ test('handlePortError ignores AbortError by name even when message text differs'
assert.equal(consoleError.mock.callCount(), 0)
})
+test('handlePortError ignores disconnected port errors', (t) => {
+ const consoleError = t.mock.method(console, 'error', () => {})
+ const consoleWarn = t.mock.method(console, 'warn', () => {})
+
+ for (const message of [
+ 'Attempting to use a disconnected port object',
+ 'Attempt to postMessage on disconnected port',
+ ]) {
+ const port = createFakePort()
+
+ handlePortError({ modelName: 'chatgptApi4oMini' }, port, { message })
+
+ assert.deepEqual(port.postedMessages, [])
+ }
+ assert.equal(consoleError.mock.callCount(), 0)
+ assert.equal(consoleWarn.mock.callCount(), 2)
+})
+
+test('handlePortError reports upstream errors that mention a closed port', (t) => {
+ t.mock.method(console, 'error', () => {})
+ const port = createFakePort()
+
+ handlePortError({ modelName: 'chatgptApi4oMini' }, port, {
+ message: 'Upstream reset the request because its port closed',
+ })
+
+ assert.deepEqual(port.postedMessages, [
+ { error: 'Upstream reset the request because its port closed' },
+ ])
+})
+
test('handlePortError reports Claude web authorization hint', (t) => {
t.mock.method(console, 'error', () => {})
const port = createFakePort()
@@ -197,3 +252,20 @@ test('handlePortError handles undefined thrown values without throwing again', (
assert.equal(port.postedMessages.length, 1)
assert.equal(port.postedMessages[0].error, 'unknown error')
})
+
+test('handlePortError does not throw when the error port is closed', (t) => {
+ t.mock.method(console, 'error', () => {})
+ const consoleWarn = t.mock.method(console, 'warn', () => {})
+ const port = {
+ postMessage() {
+ throw new Error('Port closed')
+ },
+ }
+
+ assert.doesNotThrow(() => {
+ handlePortError({ modelName: 'chatgptApi4oMini' }, port, {
+ message: 'done failed',
+ })
+ })
+ assert.equal(consoleWarn.mock.callCount(), 1)
+})
diff --git a/tests/unit/services/wrappers-register.test.mjs b/tests/unit/services/wrappers-register.test.mjs
index 2b5086722..e04c24aca 100644
--- a/tests/unit/services/wrappers-register.test.mjs
+++ b/tests/unit/services/wrappers-register.test.mjs
@@ -222,6 +222,116 @@ test('registerPortListener ignores messages without session', async (t) => {
assert.deepEqual(port.postedMessages, [])
})
+test('registerPortListener tags responses with proxy and request generation ids', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ setStorage({ modelName: 'chatgptApi4oMini' })
+
+ let resolveExec
+ const execDone = new Promise((resolve) => {
+ resolveExec = resolve
+ })
+ const executor = t.mock.fn(async (_session, requestPort) => {
+ requestPort.postMessage({ done: true })
+ resolveExec()
+ })
+
+ registerPortListener(executor)
+ const port = createFakePort()
+ triggerConnect(port)
+
+ port.emitMessage({
+ session: { conversationRecords: [] },
+ proxyGenerationId: 7,
+ requestGenerationId: 11,
+ })
+ await execDone
+
+ assert.equal(port.postedMessages.length, 2)
+ assert.equal(
+ port.postedMessages.every(
+ (message) => message.proxyGenerationId === 7 && message.requestGenerationId === 11,
+ ),
+ true,
+ )
+})
+
+test('registerPortListener drops responses from a superseded session request', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ setStorage({ modelName: 'chatgptApi4oMini' })
+
+ const requestPorts = []
+ const connectionPorts = []
+ let resolveRequest
+ const requestReady = () =>
+ new Promise((resolve) => {
+ resolveRequest = resolve
+ })
+ let ready = requestReady()
+ const executor = t.mock.fn(
+ async (
+ _session,
+ requestPort,
+ _config,
+ _isLatestSessionRequest,
+ _requestGenerationId,
+ connectionPort,
+ ) => {
+ requestPorts.push(requestPort)
+ connectionPorts.push(connectionPort)
+ resolveRequest()
+ },
+ )
+
+ registerPortListener(executor)
+ const port = createFakePort()
+ triggerConnect(port)
+
+ port.emitMessage({ session: { conversationRecords: [] } })
+ await ready
+ ready = requestReady()
+ port.emitMessage({ session: { conversationRecords: [] } })
+ await ready
+
+ assert.notEqual(requestPorts[0], requestPorts[1])
+ assert.deepEqual(connectionPorts, [port, port])
+
+ requestPorts[0].postMessage({ done: true })
+ assert.equal(port.postedMessages.length, 2)
+ requestPorts[1].postMessage({ done: true })
+
+ assert.equal(port.postedMessages.length, 3)
+ assert.deepEqual(port.postedMessages.at(-1), { done: true })
+})
+
+test('registerPortListener allows a stopped request to post before its replacement starts', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ setStorage({ modelName: 'chatgptApi4oMini' })
+
+ let requestPort
+ let resolveExec
+ const execReady = new Promise((resolve) => {
+ resolveExec = resolve
+ })
+ const executor = t.mock.fn(async (_session, currentRequestPort) => {
+ requestPort = currentRequestPort
+ resolveExec()
+ })
+
+ registerPortListener(executor)
+ const port = createFakePort()
+ triggerConnect(port)
+
+ port.emitMessage({ session: { conversationRecords: [] } })
+ await execReady
+ port.emitMessage({ stop: true })
+ requestPort.postMessage({ session: { conversationRecords: [] } })
+
+ assert.deepEqual(port.postedMessages.slice(-2), [
+ { done: true },
+ { session: { conversationRecords: [] } },
+ ])
+})
+
test('registerPortListener catches executor errors and calls handlePortError', async (t) => {
t.mock.method(console, 'debug', () => {})
t.mock.method(console, 'error', () => {})
diff --git a/tests/unit/utils/fetch-sse.test.mjs b/tests/unit/utils/fetch-sse.test.mjs
index dc6254c4f..3f5425086 100644
--- a/tests/unit/utils/fetch-sse.test.mjs
+++ b/tests/unit/utils/fetch-sse.test.mjs
@@ -112,3 +112,93 @@ test('fetchSSE forwards fetch rejection errors to onError', async (t) => {
assert.equal(errors.length, 1)
assert.equal(errors[0].message, 'network down')
})
+
+test('fetchSSE treats an AbortError before streaming as cancelled completion', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ const consoleWarn = t.mock.method(console, 'warn', () => {})
+ const errors = []
+ let aborted = false
+
+ t.mock.method(globalThis, 'fetch', async () => {
+ throw Object.assign(new Error('The operation was aborted'), {
+ name: 'AbortError',
+ })
+ })
+
+ await fetchSSE('https://example.com/abort', {
+ onStart: async () => {},
+ onMessage: () => {},
+ onEnd: async (wasAborted) => {
+ aborted = wasAborted
+ throw new Error('cleanup failed')
+ },
+ onError: async (error) => {
+ errors.push(error)
+ },
+ })
+
+ assert.equal(aborted, true)
+ assert.deepEqual(errors, [])
+ assert.equal(consoleWarn.mock.callCount(), 1)
+})
+
+test('fetchSSE propagates onEnd errors on normal completion', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ t.mock.method(globalThis, 'fetch', async () => createMockSseResponse(['data: {"delta":"A"}\n\n']))
+
+ await assert.rejects(
+ fetchSSE('https://example.com/sse', {
+ onStart: async () => {},
+ onMessage: () => {},
+ onEnd: async () => {
+ throw new Error('done failed')
+ },
+ onError: async () => {},
+ }),
+ /done failed/,
+ )
+})
+
+test('fetchSSE propagates onStart errors', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ t.mock.method(globalThis, 'fetch', async () => createMockSseResponse(['data: {"delta":"A"}\n\n']))
+ const errors = []
+
+ await assert.rejects(
+ fetchSSE('https://example.com/sse', {
+ onStart: async () => {
+ throw new Error('start failed')
+ },
+ onMessage: () => {},
+ onEnd: async () => {},
+ onError: async (error) => {
+ errors.push(error)
+ },
+ }),
+ /start failed/,
+ )
+ assert.equal(errors.length, 1)
+ assert.equal(errors[0].message, 'start failed')
+})
+
+test('fetchSSE propagates onMessage errors', async (t) => {
+ t.mock.method(console, 'debug', () => {})
+ t.mock.method(globalThis, 'fetch', async () => createMockSseResponse(['data: {"delta":"A"}\n\n']))
+ const errors = []
+
+ await assert.rejects(
+ fetchSSE('https://example.com/sse', {
+ onStart: async () => {},
+ onMessage: () => {
+ throw new Error('message failed')
+ },
+ onEnd: async () => {},
+ onError: async (error) => {
+ errors.push(error)
+ },
+ }),
+ /message failed/,
+ )
+ assert.equal(errors.length, 1)
+ assert.equal(errors[0].message, 'message failed')
+})