perf: reduce streaming render cost and refresh activity page on demand

- Segment streaming Markdown so completed blocks are reused and only the growing tail is re-parsed; fall back to whole-document parsing for document-wide constructs
- Memoize conversation sidebar rows and cache per-conversation derived activity, ordering and pending approvals
- Activity page loads on entry, refreshes every 15 s while visible, and has a single header refresh button
- Include pending settings, model download source, device sharing and knowledge gateway changes in the working tree
- Add Electron streaming input benchmark and regression tests
This commit is contained in:
mesalogo
2026-10-01 21:49:09 +08:00
parent 016f987026
commit c8f327fc4e
75 changed files with 3042 additions and 905 deletions
+67 -1
View File
@@ -9,11 +9,13 @@ import {
} from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'
import type { KnowledgeService } from '../knowledge/knowledge-service'
import {
ContinueHostRunError,
type ContinueHostAdapterOptions
} from './continue-host-adapter'
import type { KnowledgeMcpGateway } from './knowledge-mcp-gateway'
import { KnowledgeMcpGateway } from './knowledge-mcp-gateway'
const mocks = vi.hoisted(() => ({
detectRuntimeBinary: vi.fn(),
@@ -464,6 +466,70 @@ describe('ContinueAgentRuntime', () => {
expect(gateway.grantCustomMcp).not.toHaveBeenCalled()
})
it('completes the Continue run when an assigned custom MCP server returns HTTP 503', async () => {
const requests = vi.fn((_request: IncomingMessage, response: ServerResponse) => {
response.writeHead(503, { Connection: 'close' })
response.end('Service unavailable')
})
const upstream = createServer(requests)
const gateway = new KnowledgeMcpGateway({} as KnowledgeService)
let runtime: ContinueAgentRuntime | undefined
try {
await new Promise<void>((resolve, reject) => {
upstream.once('error', reject)
upstream.listen(0, '127.0.0.1', resolve)
})
const address = upstream.address()
if (!address || typeof address === 'string') {
throw new Error('Custom MCP test server did not bind a TCP port')
}
await gateway.start()
runtime = new ContinueAgentRuntime({
binaryPath: '',
configPath: 'C:\\safe config\\continue.yaml',
defaultWorkspace: process.cwd(),
hostCacheRoot: 'C:\\safe\\continue-host',
knowledgeGateway: gateway,
mcpServers: [{
id: randomUUID(),
name: 'Unavailable MCP',
description: '',
enabled: true,
allowDynamicTools: false,
assignments: ['continue'],
secretConfigured: false,
transport: 'http',
url: `http://127.0.0.1:${address.port}/mcp`
}],
createHostAdapter: () => ({
getPreparedHost: mocks.prepareHost,
run: mocks.runHost,
dispose: mocks.disposeHost
})
})
const events = await collectEvents(runtime, 'execute')
expect(requests).toHaveBeenCalled()
expect(mocks.runHost).toHaveBeenCalledOnce()
expect(events).toContainEqual({
requestId: '3f496642-f47d-4e0a-8944-a32c77b0d6ef',
type: 'text',
delta: 'Continue response'
})
expect(events.at(-1)).toMatchObject({ type: 'done' })
} finally {
await Promise.all([
runtime?.dispose(),
gateway.dispose(),
new Promise<void>((resolve) => {
upstream.close(() => resolve())
upstream.closeAllConnections()
})
])
}
})
it.each([undefined, 'existing-capability'])('cleans up image capabilities after custom MCP preparation fails (existing=%s)', async existingToken => {
const gateway = {
getEndpoint: () => 'http://127.0.0.1:4567/mcp',
+61 -10
View File
@@ -1333,7 +1333,7 @@ describe('KnowledgeMcpGateway', () => {
expect(outputValidator).toHaveBeenCalledWith({ value: 'done' })
})
it('rejects cyclic cursors and custom MCP tool counts over 100', async () => {
it('omits servers with cyclic cursors or custom MCP tool counts over 100', async () => {
const cyclicUpstream = await startToolUpstream(
(cursor: string | undefined) => ({
tools: [testTool(cursor ? 'second' : 'first')],
@@ -1366,18 +1366,63 @@ describe('KnowledgeMcpGateway', () => {
await expect(
gateway.prepareCustomMcpTools(cycleToken)
).rejects.toMatchObject({
cause: expect.objectContaining({
message: expect.stringContaining('分页游标发生循环')
})
})
).resolves.toEqual([])
await expect(
gateway.prepareCustomMcpTools(excessiveToken)
).rejects.toMatchObject({
cause: expect.objectContaining({
message: expect.stringContaining('工具数量超过安全限制')
})
).resolves.toEqual([])
})
it('keeps healthy tools available when another upstream fails discovery', async () => {
const healthyList = vi.fn(() => ({ tools: [testTool('healthy')] }))
const healthy = await startToolUpstream(healthyList)
const failedList = vi.fn(() => { throw new Error('private-upstream-failure') })
const failed = await startToolUpstream(failedList)
const observeFailure = vi.fn()
const gateway = new KnowledgeMcpGateway(createService().service, { observeFailure })
gateways.push(gateway)
await gateway.start()
const controller = new AbortController()
const token = gateway.grantCustomMcp('partial-discovery', [
customMcpServer(healthy.url),
customMcpServer(failed.url, '00000000-0000-4000-8000-000000000093')
], controller.signal)!
await expect(gateway.prepareCustomMcpTools(token)).resolves.toEqual([
expect.objectContaining({ name: expect.stringMatching(/_healthy$/u) })
])
const client = new Client({ name: 'partial-discovery', version: '1.0.0' })
try {
await client.connect(new StreamableHTTPClientTransport(new URL(gateway.getEndpoint()!), {
requestInit: { headers: { authorization: `Bearer ${token}` } }
}))
const listed = await client.listTools()
expect(listed.tools.map(tool => tool.name)).toEqual([
expect.stringMatching(/_healthy$/u)
])
expect(healthyList).toHaveBeenCalledTimes(1)
expect(failedList).toHaveBeenCalledTimes(1)
expect(observeFailure).toHaveBeenCalledWith(expect.objectContaining({
mcp: expect.objectContaining({ phase: 'tool-discovery' })
}))
expect(JSON.stringify(observeFailure.mock.calls)).not.toContain('private-upstream-failure')
} finally {
await client.close()
}
})
it.each(['abort', 'revoke'] as const)('does not swallow %s during discovery', async action => {
const controller = new AbortController()
const reason = new Error('request cancelled')
const gateway = new KnowledgeMcpGateway(createService().service)
gateways.push(gateway)
const upstream = await startToolUpstream(() => {
if (action === 'abort') controller.abort(reason)
else gateway.revoke(token)
return { tools: [testTool('late')] }
})
const token = gateway.grantCustomMcp('cancel-discovery', [customMcpServer(upstream.url)], controller.signal)!
const preparation = gateway.prepareCustomMcpTools(token)
if (action === 'abort') await expect(preparation).rejects.toBe(reason)
else await expect(preparation).rejects.toThrow('MCP capability was revoked')
})
it('releases rejected initialize attempts before enforcing the session limit', async () => {
@@ -1542,6 +1587,12 @@ describe('KnowledgeMcpGateway', () => {
})
await new Promise((resolve) => setTimeout(resolve, 25))
expect(listChanged).not.toHaveBeenCalled()
await expect(client.listTools()).resolves.toMatchObject({ tools: [] })
failRefresh = false
const recovered = await client.listTools()
expect(recovered.tools.map(tool => tool.name)).toEqual([
expect.stringMatching(/_stable$/u)
])
} finally {
await client.close()
}
+30 -14
View File
@@ -960,6 +960,7 @@ export class KnowledgeMcpGateway {
capability.brokerController.signal
])
const changeVersion = connection.dynamicToolsChangeVersion
const startedAt = performance.now()
let refreshSucceeded = false
const refresh = (async () => {
try {
@@ -979,10 +980,15 @@ export class KnowledgeMcpGateway {
refreshSucceeded = true
await this.publishCustomMcpToolListChanged(capability)
} catch (error) {
connection.bindings = []
connection.dynamicToolsChanged = true
if (effectiveSignal.aborted) {
throw effectiveSignal.reason
}
this.observeMcpFailure({
phase: 'tool-discovery', category: 'handler-failure', kind: 'custom',
elapsedMs: Math.round(performance.now() - startedAt)
})
throw new Error(
`无法刷新 MCP Server「${connection.server.name}」的工具`,
{ cause: error }
@@ -1008,6 +1014,7 @@ export class KnowledgeMcpGateway {
capability: Capability,
server: ResolvedMcpServer
): Promise<CustomMcpConnection> {
const startedAt = performance.now()
let connection: CustomMcpConnection | undefined
let dynamicToolsChangeVersion = 0
const client = new Client(
@@ -1079,6 +1086,11 @@ export class KnowledgeMcpGateway {
return connection
} catch (error) {
await client.close().catch(() => undefined)
signal.throwIfAborted()
this.observeMcpFailure({
phase: 'tool-discovery', category: 'handler-failure', kind: 'custom',
elapsedMs: Math.round(performance.now() - startedAt)
})
throw new Error(
`无法加载 MCP Server「${server.name}」的工具`,
{ cause: error }
@@ -1092,6 +1104,12 @@ export class KnowledgeMcpGateway {
refreshDynamic = true
): Promise<Map<string, CustomMcpBinding>> {
const capability = this.getCapability(token)
const effectiveSignal = AbortSignal.any([
capability.signal,
capability.brokerController.signal,
...(signal ? [signal] : [])
])
effectiveSignal.throwIfAborted()
if (capability.customMcpServers.length === 0) {
return new Map()
}
@@ -1105,15 +1123,6 @@ export class KnowledgeMcpGateway {
const connections = results.flatMap((result) =>
result.status === 'fulfilled' ? [result.value] : []
)
const failure = results.find(
(result) => result.status === 'rejected'
)
if (failure?.status === 'rejected') {
await Promise.allSettled(
connections.map((connection) => connection.client.close())
)
throw failure.reason
}
return connections
})()
}
@@ -1124,6 +1133,7 @@ export class KnowledgeMcpGateway {
capability.customMcpConnections = undefined
throw error
}
effectiveSignal.throwIfAborted()
if (refreshDynamic) {
for (const connection of connections) {
if (
@@ -1133,13 +1143,19 @@ export class KnowledgeMcpGateway {
) {
continue
}
await this.refreshDynamicTools(
capability,
connection,
signal
)
try {
await this.refreshDynamicTools(
capability,
connection,
signal
)
} catch {
// Failed catalogs are cleared by refresh; healthy servers remain usable.
effectiveSignal.throwIfAborted()
}
}
}
effectiveSignal.throwIfAborted()
const bindings = new Map<string, CustomMcpBinding>()
for (const connection of connections) {
for (const binding of connection.bindings) {
@@ -16,6 +16,95 @@ const binaryPath = join(
process.platform === 'win32' ? 'opencode.exe' : 'opencode'
)
it.skipIf(!existsSync(binaryPath))(
'completes Execute with the real OpenCode binary when an assigned custom MCP returns HTTP 503',
async () => {
const workspace = await mkdtemp(join(tmpdir(), 'goodbuddy-opencode-mcp-unavailable-'))
let modelCalls = 0
const model = createServer((_request, response) => {
modelCalls++
response.writeHead(200, { 'content-type': 'text/event-stream' })
response.end(`data: ${JSON.stringify({
id: 'unavailable', object: 'chat.completion.chunk', created: 1, model: 'unavailable',
choices: [{ index: 0, delta: { role: 'assistant', content: 'MCP_UNAVAILABLE_OK' }, finish_reason: 'stop' }]
})}\n\ndata: [DONE]\n\n`)
})
let mcpRequests = 0
const failureBody = 'private-mcp-failure-detail'
const upstream = createServer((_request, response) => {
mcpRequests++
response.writeHead(503, { Connection: 'close' })
response.end(failureBody)
})
const records: DesktopDiagnosticFailure[] = []
const gateway = new KnowledgeMcpGateway({} as never, {
observeFailure: failure => records.push(failure)
})
const controller = new AbortController()
const deadline = setTimeout(() => controller.abort(), 30_000)
let runtime: OpenCodeRuntime | undefined
try {
await new Promise<void>(resolve => model.listen(0, '127.0.0.1', resolve))
await new Promise<void>(resolve => upstream.listen(0, '127.0.0.1', resolve))
const modelAddress = model.address()
const mcpAddress = upstream.address()
if (!modelAddress || typeof modelAddress === 'string') throw new Error('No model port')
if (!mcpAddress || typeof mcpAddress === 'string') throw new Error('No MCP port')
await gateway.start()
const secret = 'private-mcp-fixture-secret'
const url = `http://127.0.0.1:${mcpAddress.port}/mcp`
runtime = new OpenCodeRuntime({
embedded: true, binaryPath: '', bundledBinaryPath: binaryPath,
configPath: '', defaultWorkspace: workspace, knowledgeGateway: gateway,
mcpServers: [{
id: crypto.randomUUID(), name: 'Unavailable MCP', description: '',
enabled: true, allowDynamicTools: false, assignments: ['opencode'],
secretConfigured: true, secret, transport: 'http', url
}],
modelProfile: {
id: crypto.randomUUID(), name: 'Unavailable MCP fixture', modelName: 'unavailable',
baseUrl: `http://127.0.0.1:${modelAddress.port}/v1`,
protocol: 'openai-chat-completions', authentication: 'none'
}
})
let text = ''
let lastEventType: string | undefined
for await (const event of runtime.run({
requestId: crypto.randomUUID(), conversationId: crypto.randomUUID(), workMode: 'execute',
prompt: 'Reply MCP_UNAVAILABLE_OK without tools.'
}, controller.signal)) {
if (event.type === 'text') text += event.delta
lastEventType = event.type
}
expect(mcpRequests).toBeGreaterThan(0)
expect(text).toBe('MCP_UNAVAILABLE_OK')
expect(lastEventType).toBe('done')
expect(modelCalls).toBe(1)
expect(records).toHaveLength(1)
expect(records[0]).toMatchObject({
code: 'runtime.mcp.failed',
mcp: { phase: 'tool-discovery', category: 'handler-failure' }
})
const diagnostics = JSON.stringify(records, (_key, value) => value instanceof Error
? { message: value.message, stack: value.stack, cause: value.cause }
: value)
expect(diagnostics).not.toContain(secret)
expect(diagnostics).not.toContain(url)
expect(diagnostics).not.toContain(failureBody)
} finally {
clearTimeout(deadline)
await runtime?.dispose()
await gateway.dispose()
upstream.closeAllConnections()
model.closeAllConnections()
await new Promise<void>(resolve => upstream.close(() => resolve()))
await new Promise<void>(resolve => model.close(() => resolve()))
await rm(workspace, { recursive: true, force: true })
}
},
60_000
)
it.skipIf(!existsSync(binaryPath))(
'recovers a lost MCP add response with the real OpenCode binary and gateway',
async () => {