fix(connectors): synchronize authorization and chat runtime readiness

Clear stale MCP authentication and discovery failures after successful connections.
Verify plugin skills and MCP tools in active chats across all connector types,
and remove stale tools using normalized server names.

Cover all 54 catalog entries with offline runtime and lifecycle regressions.
This commit is contained in:
程序员阿江(Relakkes)
2026-09-14 14:37:59 +08:00
parent f0b27a1198
commit 0e7cecd81f
16 changed files with 870 additions and 22 deletions
+80
View File
@@ -0,0 +1,80 @@
import { afterEach, beforeEach, expect, mock, spyOn, test } from 'bun:test'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import type { Tool } from '../Tool.js'
import * as mcpClient from '../services/mcp/client.js'
import type { ConnectedMCPServer } from '../services/mcp/types.js'
import { getDefaultAppState } from '../state/AppStateStore.js'
import { reconcileMcpServers, type DynamicMcpState } from './print.js'
const name = 'plugin:office-supabase:service'
const config = { type: 'http' as const, url: 'http://127.0.0.1:1/original', scope: 'dynamic' as const }
const oldTools = [
{ name: 'mcp__plugin_office-supabase_service__authenticate' },
{ name: 'mcp__plugin_office-supabase_service__list_tables' },
] as Tool[]
const unrelated = { name: 'mcp__plugin_other_service__lookup' } as Tool
const fresh = { name: 'mcp__plugin_office-supabase_service__list_projects' } as Tool
let root: string
let previousConfigDir: string | undefined
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'mcp-plugin-reconcile-'))
previousConfigDir = process.env.CLAUDE_CONFIG_DIR
process.env.CLAUDE_CONFIG_DIR = root
spyOn(mcpClient, 'clearServerCache').mockResolvedValue(undefined)
spyOn(mcpClient, 'connectToServer').mockImplementation(async (serverName, serverConfig) => ({
name: serverName, config: serverConfig, type: 'connected', capabilities: {},
client: {} as ConnectedMCPServer['client'], cleanup: async () => {},
}))
spyOn(mcpClient, 'fetchToolsForClient').mockResolvedValue([fresh])
})
afterEach(async () => {
mock.restore()
if (previousConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR
else process.env.CLAUDE_CONFIG_DIR = previousConfigDir
await rm(root, { recursive: true, force: true })
})
function fixture() {
const dynamic: DynamicMcpState = {
clients: [{ name, config, type: 'needs-auth' }],
configs: { [name]: config },
tools: [...oldTools],
}
let state = getDefaultAppState()
state = { ...state, mcp: { ...state.mcp, clients: [...dynamic.clients], tools: [...oldTools, unrelated] } }
return {
dynamic,
state: () => state,
setState: (update: (previous: typeof state) => typeof state) => { state = update(state) },
}
}
test('removing a plugin MCP server clears business and authentication tools from both pools', async () => {
const f = fixture()
const result = await reconcileMcpServers({}, f.dynamic, f.setState)
expect(result.newState.tools).toEqual([])
expect(f.state().mcp.tools).toEqual([unrelated])
expect(result.newState.clients).toEqual([])
expect(result.response.removed).toEqual([name])
expect(mcpClient.connectToServer).not.toHaveBeenCalled()
})
test('replacing a plugin MCP connection publishes only fresh tools while preserving other servers', async () => {
const f = fixture()
const result = await reconcileMcpServers({ [name]: { ...config, url: 'http://127.0.0.1:1/replacement' } }, f.dynamic, f.setState)
expect(result.newState.tools).toEqual([fresh])
expect(f.state().mcp.tools).toEqual([unrelated, fresh])
expect(result.newState.clients.map(client => client.type)).toEqual(['connected'])
expect(mcpClient.clearServerCache).toHaveBeenCalledWith(name, config)
})
test('adding a plugin server replaces its stale startup authentication tool in app state', async () => {
const f = fixture()
const result = await reconcileMcpServers({ [name]: config }, { clients: [], tools: [], configs: {} }, f.setState)
expect(result.newState.tools).toEqual([fresh])
expect(f.state().mcp.tools).toEqual([unrelated, fresh])
})
+3 -3
View File
@@ -5690,8 +5690,8 @@ export async function reconcileMcpServers(
await clearServerCache(name, config) await clearServerCache(name, config)
} }
// Remove tools from this server // Plugin identities contain colons; tools use the normalized MCP prefix.
const prefix = `mcp__${name}__` const prefix = getMcpPrefix(name)
newTools = newTools.filter(t => !t.name.startsWith(prefix)) newTools = newTools.filter(t => !t.name.startsWith(prefix))
// Remove from clients list // Remove from clients list
@@ -5761,7 +5761,7 @@ export async function reconcileMcpServers(
// Remove old dynamic tools // Remove old dynamic tools
const nonDynamicTools = prev.mcp.tools.filter(t => { const nonDynamicTools = prev.mcp.tools.filter(t => {
for (const serverName of allDynamicServerNames) { for (const serverName of allDynamicServerNames) {
if (t.name.startsWith(`mcp__${serverName}__`)) { if (t.name.startsWith(getMcpPrefix(serverName))) {
return false return false
} }
} }
@@ -0,0 +1,71 @@
import { expect, test } from 'bun:test'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { ALL_CONNECTORS } from '../../services/connectors/catalog.js'
import { ConnectorService } from './connectorService.js'
// Exercise every catalog definition through the production state machine.
// Adapters and chat transport are fixtures: no downloaded binaries or accounts.
test.each(ALL_CONNECTORS)('$id gates readiness on chat capabilities through connect, recheck, disable and removal', async definition => {
const root = mkdtempSync(join(tmpdir(), 'connector-catalog-state-'))
let enabled = false
let available = true
let authenticated = definition.transport === 'skills'
const required: string[] = []
const service = new ConnectorService({
root, definitions: [definition], platform: definition.platforms[0],
createAdapter: () => ({
prepare: async () => ({ directory: root, command: '', args: [], env: {} }),
authenticate: async () => { authenticated = true },
check: async () => ({ authenticated, verification: definition.transport === 'mcp' ? 'remote' : 'local' }),
deactivate: async () => {}, remove: async () => {},
}),
bridge: {
installConnectorPlugin: async () => {}, removeConnectorPlugin: async () => {},
setConnectorPluginEnabled: async (_def, value) => { enabled = value },
isConnectorPluginReady: async () => true,
reloadConnectorSessions: async (_session, target) => {
if (!target) return
expect(enabled).toBe(true)
expect(target).toEqual(definition)
required.push(target.id)
if (!available) throw new Error('Fixture chat did not load its connector')
},
},
})
async function settle() {
for (let i = 0; i < 200 && service.get(definition.id).operation; i++) await Bun.sleep(1)
expect(service.get(definition.id).operation).toBeUndefined()
}
try {
service.action(definition.id, 'prepare')
await settle()
if (definition.transport !== 'skills') {
expect(required).toEqual([])
service.action(definition.id, 'authenticate', { acknowledgeSharedCredentials: true })
await settle()
}
expect(required).toEqual([definition.id])
expect(service.get(definition.id)).toMatchObject({ enabled: true, runtime: 'ready' })
available = false
service.action(definition.id, 'check')
await settle()
expect(service.get(definition.id)).toMatchObject({ enabled: false, runtime: 'error', status: 'error' })
expect(enabled).toBe(false)
available = true
service.action(definition.id, 'check')
await settle()
expect(service.get(definition.id)).toMatchObject({ enabled: true, runtime: 'ready' })
expect(required).toEqual([definition.id, definition.id, definition.id])
service.action(definition.id, 'deactivate')
await settle()
expect(service.get(definition.id)).toMatchObject({ enabled: false, status: 'disabled' })
service.action(definition.id, 'remove')
await settle()
expect(service.get(definition.id)).toMatchObject({ enabled: false, installed: false })
expect(required).toHaveLength(3)
} finally { rmSync(root, { recursive: true, force: true }) }
})
@@ -24,6 +24,30 @@ function fixture(overrides: Partial<ConnectorAdapter> = {}) {
removeConnectorPlugin: async () => { calls.push('remove-plugin') }, isConnectorPluginReady: async () => true, reloadConnectorSessions: async () => { calls.push('reload') } } } removeConnectorPlugin: async () => { calls.push('remove-plugin') }, isConnectorPluginReady: async () => true, reloadConnectorSessions: async () => { calls.push('reload') } } }
return { service: new ConnectorService(deps), deps, calls, cleanup: () => rmSync(root, { recursive: true, force: true }) } return { service: new ConnectorService(deps), deps, calls, cleanup: () => rmSync(root, { recursive: true, force: true }) }
} }
test('remote authorization verifies chat tools before publishing ready and fails closed on refresh failure', async () => {
const f = fixture()
try {
f.deps.definitions = [{ ...definition, transport: 'mcp' }]
f.service.action('feishu', 'prepare')
await settled(f.service)
const refreshes: Array<[string | undefined, unknown]> = []
let ready = true
f.deps.bridge.reloadConnectorSessions = async (sessionId, requiredServer) => {
refreshes.push([sessionId, requiredServer])
if (requiredServer && !ready) throw new Error('Chat still needs auth')
}
f.service.action('feishu', 'authenticate', { sessionId: 'active-chat', acknowledgeSharedCredentials: true })
await settled(f.service)
expect(refreshes).toEqual([['active-chat', undefined], ['active-chat', { ...definition, transport: 'mcp' }]])
expect(f.service.get('feishu')).toMatchObject({ status: 'ready', runtime: 'ready' })
ready = false
f.service.action('feishu', 'check', { sessionId: 'active-chat' })
await settled(f.service)
expect(f.service.get('feishu')).toMatchObject({ status: 'error', runtime: 'error', enabled: false, failedPhase: 'refreshing-sessions' })
expect(refreshes.at(-1)).toEqual(['active-chat', undefined])
} finally { f.cleanup() }
})
test('installation, shared authorization and runtime readiness remain independent; restart invalidates readiness', async () => { test('installation, shared authorization and runtime readiness remain independent; restart invalidates readiness', async () => {
const f = fixture() const f = fixture()
try { try {
+4 -4
View File
@@ -15,7 +15,7 @@ type Bridge = {
setConnectorPluginEnabled(def: ConnectorDefinition, enabled: boolean): Promise<void> setConnectorPluginEnabled(def: ConnectorDefinition, enabled: boolean): Promise<void>
removeConnectorPlugin(def: ConnectorDefinition): Promise<void> removeConnectorPlugin(def: ConnectorDefinition): Promise<void>
isConnectorPluginReady(def: ConnectorDefinition): Promise<boolean> isConnectorPluginReady(def: ConnectorDefinition): Promise<boolean>
reloadConnectorSessions(sessionId?: string): Promise<void> reloadConnectorSessions(sessionId?: string, requiredConnector?: ConnectorDefinition): Promise<void>
} }
export type ConnectorServiceDependencies = { export type ConnectorServiceDependencies = {
definitions: readonly ConnectorDefinition[] definitions: readonly ConnectorDefinition[]
@@ -160,7 +160,7 @@ export class ConnectorService {
progress('enabling-plugin') progress('enabling-plugin')
await bridge.setConnectorPluginEnabled(def, true) await bridge.setConnectorPluginEnabled(def, true)
assertActive() assertActive()
await bridge.reloadConnectorSessions(options.sessionId) await bridge.reloadConnectorSessions(options.sessionId, def)
assertActive() assertActive()
if (!await bridge.isConnectorPluginReady(def)) throw new Error('Skill package could not load') if (!await bridge.isConnectorPluginReady(def)) throw new Error('Skill package could not load')
assertActive() assertActive()
@@ -195,7 +195,7 @@ export class ConnectorService {
assertActive() assertActive()
progress('refreshing-sessions') progress('refreshing-sessions')
assertActive() assertActive()
await bridge.reloadConnectorSessions(options.sessionId) await bridge.reloadConnectorSessions(options.sessionId, def)
assertActive() assertActive()
progress('verifying-runtime') progress('verifying-runtime')
assertActive() assertActive()
@@ -230,7 +230,7 @@ export class ConnectorService {
const oldDefinition = this.installedDefinition(def, previous) const oldDefinition = this.installedDefinition(def, previous)
await bridge.installConnectorPlugin(oldDefinition, previous.installation) await bridge.installConnectorPlugin(oldDefinition, previous.installation)
await bridge.setConnectorPluginEnabled(oldDefinition, previous.enabled ?? false) await bridge.setConnectorPluginEnabled(oldDefinition, previous.enabled ?? false)
await bridge.reloadConnectorSessions(options.sessionId) await bridge.reloadConnectorSessions(options.sessionId, previous.enabled ? oldDefinition : undefined)
restored = true restored = true
} catch { /* failed rollback must remain disabled */ } } catch { /* failed rollback must remain disabled */ }
} }
@@ -0,0 +1,82 @@
import { afterAll, afterEach, expect, spyOn, test } from 'bun:test'
import { conversationService } from './conversationService.js'
import { reloadSessionComponents } from './sessionComponentReloadService.js'
import { __resetWebSocketHandlerStateForTests } from '../ws/handler.js'
const serverName = 'plugin:office-supabase:service'
const ready = { name: serverName, status: 'connected', tools: [{ name: 'list_tables' }] }
const hasSession = spyOn(conversationService, 'hasSession').mockReturnValue(true)
const requestControl = spyOn(conversationService, 'requestControl')
test.each(['feishu', 'dingtalk', 'wecom', 'remotion', 'supabase'])('requires the %s connector plugin and skill in the chat, not only on disk', async id => {
const requirement = { pluginId: `office-${id}@haha-connectors`, skillName: `office-${id}:office-${id}` }
for (const response of [
{ plugins: [], commands: [{ name: requirement.skillName }] },
{ plugins: [{ source: requirement.pluginId }], commands: [] },
{ plugins: [{ source: 'other@market' }], commands: [{ name: requirement.skillName }] },
]) {
requestControl.mockResolvedValue({ ...response, error_count: 0 })
expect(await reloadSessionComponents('fixture-session', undefined, requirement)).toMatchObject({ applied: false, reason: 'failed' })
}
requestControl.mockReset().mockResolvedValue({
plugins: [{ source: requirement.pluginId }], commands: [{ name: requirement.skillName }], error_count: 0,
})
expect(await reloadSessionComponents('fixture-session', undefined, requirement)).toMatchObject({ applied: true, commands: 1, plugins: 1 })
expect(requestControl).toHaveBeenCalledTimes(1)
})
afterAll(() => {
requestControl.mockRestore()
hasSession.mockRestore()
})
afterEach(() => {
requestControl.mockReset()
hasSession.mockReturnValue(true)
__resetWebSocketHandlerStateForTests()
})
test('connector reload replaces stale authentication state with tools from the chat process', async () => {
requestControl.mockResolvedValueOnce({ commands: [], error_count: 0, mcpServers: [{ name: serverName, status: 'needs-auth' }] })
.mockResolvedValueOnce({})
.mockResolvedValueOnce({ mcpServers: [ready] })
expect(await reloadSessionComponents('fixture-session', serverName)).toMatchObject({ applied: true, errors: 0 })
expect(requestControl.mock.calls.map(call => call[1])).toEqual([
{ subtype: 'reload_plugins' },
{ subtype: 'mcp_reconnect', serverName },
{ subtype: 'mcp_status' },
])
})
test.each([
[],
[{ name: serverName, status: 'needs-auth', tools: [{ name: 'authenticate' }] }],
[{ name: serverName, status: 'failed' }],
[{ name: serverName, status: 'disabled' }],
[{ name: serverName, status: 'connected', tools: [] }],
[{ name: serverName, status: 'connected', tools: [{ name: 'authenticate' }] }],
[{ name: 'another-server', status: 'connected', tools: [{ name: 'list_tables' }] }],
].map(mcpServers => ({ mcpServers })))('a successful plugin reload cannot hide an unusable connector: %j', async ({ mcpServers }) => {
requestControl.mockResolvedValue({ commands: [], error_count: 0, mcpServers })
expect(await reloadSessionComponents('fixture-session', serverName)).toMatchObject({ applied: false, reason: 'failed' })
})
test('healthy tools avoid reconnecting and ordinary skill reload does not require MCP', async () => {
requestControl.mockResolvedValue({ commands: [], error_count: 0, mcpServers: [ready] })
expect((await reloadSessionComponents('fixture-session', serverName)).applied).toBe(true)
expect(requestControl).toHaveBeenCalledTimes(1)
requestControl.mockReset().mockResolvedValue({ commands: [], error_count: 0 })
expect((await reloadSessionComponents('fixture-session')).applied).toBe(true)
expect(requestControl).toHaveBeenCalledTimes(1)
})
test('a stopped session needs no reconnect and a failed reconnect is not ready', async () => {
hasSession.mockReturnValue(false)
expect(await reloadSessionComponents('fixture-session', serverName)).toMatchObject({ applied: false, reason: 'not_running' })
expect(requestControl).not.toHaveBeenCalled()
hasSession.mockReturnValue(true)
requestControl.mockResolvedValueOnce({ commands: [], error_count: 0, mcpServers: [] })
.mockRejectedValueOnce(new Error('Server status: needs-auth'))
expect(await reloadSessionComponents('fixture-session', serverName)).toMatchObject({ applied: false, reason: 'failed' })
})
@@ -19,6 +19,8 @@ export type SessionComponentReloadSummary = {
*/ */
export async function reloadSessionComponents( export async function reloadSessionComponents(
sessionId: string, sessionId: string,
requiredMcpServer?: string,
requiredPlugin?: { pluginId: string, skillName: string },
): Promise<SessionComponentReloadSummary> { ): Promise<SessionComponentReloadSummary> {
if (!conversationService.hasSession(sessionId)) { if (!conversationService.hasSession(sessionId)) {
return emptySummary('not_running') return emptySummary('not_running')
@@ -32,13 +34,37 @@ export async function reloadSessionComponents(
) )
const commands = Array.isArray(response.commands) ? response.commands : [] const commands = Array.isArray(response.commands) ? response.commands : []
const normalizedCommands = updateSessionSlashCommands(sessionId, commands) const normalizedCommands = updateSessionSlashCommands(sessionId, commands)
let mcpServers = response.mcpServers
// All connectors publish a skill entry point, including native CLIs and
// skill bundles. A plugin on disk does not prove that this chat loaded it.
if (requiredPlugin && (
!Array.isArray(response.plugins) || !response.plugins.some(plugin => plugin?.source === requiredPlugin.pluginId) ||
!commands.some(command => command?.name === requiredPlugin.skillName)
)) {
throw new Error('Connector plugin and skill are not available in the active task')
}
if (requiredMcpServer && !hasServerTools(mcpServers, requiredMcpServer)) {
// OAuth can finish in the desktop server while this CLI still holds a
// needs-auth connection and cached credentials. Reconnect in that process
// so it reads fresh credentials and replaces its actual query tools.
await conversationService.requestControl(sessionId, {
subtype: 'mcp_reconnect', serverName: requiredMcpServer,
}, 120_000)
const status = await conversationService.requestControl(sessionId, { subtype: 'mcp_status' }, 30_000)
mcpServers = status.mcpServers
if (!hasServerTools(mcpServers, requiredMcpServer)) {
throw new Error('Connector tools are not available in the active task')
}
}
return { return {
applied: true, applied: true,
commands: normalizedCommands.length, commands: normalizedCommands.length,
agents: Array.isArray(response.agents) ? response.agents.length : 0, agents: Array.isArray(response.agents) ? response.agents.length : 0,
plugins: Array.isArray(response.plugins) ? response.plugins.length : 0, plugins: Array.isArray(response.plugins) ? response.plugins.length : 0,
mcpServers: Array.isArray(response.mcpServers) ? response.mcpServers.length : 0, mcpServers: Array.isArray(mcpServers) ? mcpServers.length : 0,
errors: typeof response.error_count === 'number' ? response.error_count : 0, errors: typeof response.error_count === 'number' ? response.error_count : 0,
} }
} catch (error) { } catch (error) {
@@ -49,6 +75,16 @@ export async function reloadSessionComponents(
} }
} }
function hasServerTools(servers: unknown, name: string): boolean {
return Array.isArray(servers) && servers.some(server =>
server?.name === name && server.status === 'connected' &&
// The local OAuth helper is not evidence of a discovered business tool.
Array.isArray(server.tools) && server.tools.some((tool: { name?: unknown } | null) =>
typeof tool?.name === 'string' && tool.name.length > 0 && tool.name !== 'authenticate',
),
)
}
function emptySummary( function emptySummary(
reason: 'not_running' | 'failed', reason: 'not_running' | 'failed',
): SessionComponentReloadSummary { ): SessionComponentReloadSummary {
@@ -140,3 +140,46 @@ test('every check and authentication verifies the installed bytes before executi
expect(executions).toBe(1) expect(executions).toBe(1)
} finally { await rm(root, { recursive: true, force: true }) } } finally { await rm(root, { recursive: true, force: true }) }
}) })
for (const definition of CONNECTORS) {
test(`${definition.id} checks and authorizes the same managed executable and account environment`, async () => {
const root = await mkdtemp(join(tmpdir(), 'connector account fixture '))
const invocations: Array<{ command: string, args: string[], env?: Record<string, string> }> = []
const dependencies: RuntimeDependencies = {
platform: 'darwin', arch: 'arm64',
binaryIntegrity: () => 'sha256-' + createHash('sha256').update('binary').digest('hex'),
readBinary: async () => Buffer.from('binary'),
download: async () => { throw new Error('Unexpected download') },
extract: async () => { throw new Error('Unexpected extraction') },
async run(command, args, options) {
invocations.push({ command, args, env: options.env })
return result(definition.id === 'wecom' ? 'authorized' : definition.id === 'feishu'
? { ok: true, identity: 'user', verified: true }
: { success: true, authenticated: true })
},
}
try {
const installation = managedInstallation(definition, root, dependencies)
const adapter = createConnectorAdapter(definition, root, dependencies)
const signal = new AbortController().signal
expect((await adapter.check(installation, signal)).authenticated).toBe(true)
await adapter.authenticate(installation, signal, () => {})
expect((await adapter.check(installation, signal)).authenticated).toBe(true)
expect(invocations.length).toBeGreaterThanOrEqual(3)
for (const invocation of invocations) {
expect(invocation.command).toBe(installation.command)
expect(invocation.env).toEqual(installation.env)
}
if (definition.id === 'dingtalk') {
expect(installation.env).toEqual({
DWS_CONFIG_DIR: join(root, 'accounts', 'dingtalk', 'dws'),
DWS_KEYCHAIN_DIR: join(root, 'accounts', 'dingtalk', 'keychain'),
DWS_DISABLE_KEYCHAIN: '1',
})
} else {
// Feishu and WeCom deliberately share the desktop user's account.
expect(installation.env).toEqual({})
}
} finally { await rm(root, { recursive: true, force: true }) }
})
}
@@ -0,0 +1,61 @@
import { expect, test } from 'bun:test'
import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { dirname, join } from 'node:path'
import { getCommandName } from '../../types/command.js'
import type { LoadedPlugin } from '../../types/plugin.js'
import { loadPluginSkillsFromEnabledPlugins } from '../../utils/plugins/loadPluginCommands.js'
import { CONNECTORS } from './catalog.js'
import { defaultRuntimeDependencies, managedInstallation } from './managedRuntime.js'
import { renderConnectorSkill } from './pluginBridge.js'
import { getSkillRecipe, SKILL_CONNECTORS } from './skillCatalog.js'
// Use the real catalog paths and generated connector entry points, but synthetic
// upstream content: this exercises the chat's loader without downloads or login.
for (const definition of [...CONNECTORS, ...SKILL_CONNECTORS]) {
test(`${definition.id} exposes its connector entry and declared skills to fresh and refreshed chat loaders`, async () => {
const root = await mkdtemp(join(tmpdir(), 'connector-loader-'))
try {
const name = `office-${definition.id}`
const skillsPath = join(root, 'skills')
const wrapper = join(skillsPath, name, 'SKILL.md')
const installation = definition.transport === 'skills'
? { directory: root, command: '', args: [], env: {} }
: managedInstallation(definition, root, { ...defaultRuntimeDependencies, platform: 'darwin', arch: 'arm64' })
await mkdir(dirname(wrapper), { recursive: true })
await writeFile(wrapper, renderConnectorSkill(definition, installation))
const upstream = getSkillRecipe(definition.id)?.files
.filter(file => /^skills\/[^/]+\/SKILL.md$/.test(file.target)) ?? []
for (const file of upstream) {
const path = join(root, file.target)
await mkdir(dirname(path), { recursive: true })
await writeFile(path, '---\ndescription: Synthetic upstream workflow\n---\nUse this fixture.\n')
}
const plugin: LoadedPlugin = {
name, manifest: { name, version: definition.version, skills: './skills' },
path: root, source: definition.pluginId, repository: definition.pluginId,
skillsPath,
}
const fresh = await loadPluginSkillsFromEnabledPlugins([plugin])
const entry = fresh.find(skill => skill.name === `${name}:${name}`)
expect(entry).toBeDefined()
expect(getCommandName(entry!)).toBe(`${name}:${name}`)
expect(entry?.type === 'prompt' && entry.disableModelInvocation).toBe(false)
expect(fresh.map(skill => skill.name).sort()).toEqual([
`${name}:${name}`,
...upstream.map(file => `${name}:${file.target.split('/')[1]}`),
].sort())
// A session reload receives the newly enabled plugin set. No old skill
// commands may survive disablement; enabling again reloads current files.
expect(await loadPluginSkillsFromEnabledPlugins([])).toEqual([])
await writeFile(wrapper, renderConnectorSkill(definition, installation)
.replace(/^description:.*$/m, 'description: Refreshed connector fixture'))
const refreshed = await loadPluginSkillsFromEnabledPlugins([plugin])
expect(refreshed.find(skill => skill.name === `${name}:${name}`)?.description)
.toBe('Refreshed connector fixture')
} finally {
await rm(root, { recursive: true, force: true })
}
})
}
@@ -0,0 +1,50 @@
import { afterEach, expect, mock, spyOn, test } from 'bun:test'
import { reloadConnectorSessions } from './pluginBridge.js'
import { ALL_CONNECTORS } from './catalog.js'
const remote = ALL_CONNECTORS.find(def => def.id === 'supabase')!
const remoteSkill = { pluginId: remote.pluginId, skillName: 'office-supabase:office-supabase' }
import { conversationService } from '../../server/services/conversationService.js'
import * as sessionReload from '../../server/services/sessionComponentReloadService.js'
afterEach(() => { mock.restore() })
test.each(ALL_CONNECTORS)('$id requires its own plugin and skill; only MCP requires remote tools', async definition => {
spyOn(conversationService, 'getActiveSessions').mockReturnValue(['catalog-chat'])
const reload = spyOn(sessionReload, 'reloadSessionComponents').mockResolvedValue({
applied: true, commands: 1, agents: 0, plugins: 1, mcpServers: 0, errors: 0,
})
await reloadConnectorSessions(undefined, definition)
expect(reload.mock.calls).toEqual([[
'catalog-chat',
definition.transport === 'mcp' ? `plugin:office-${definition.id}:service` : undefined,
{ pluginId: definition.pluginId, skillName: `office-${definition.id}:office-${definition.id}` },
]])
})
test('connector readiness verifies its server in every active chat and deduplicates the requested chat', async () => {
spyOn(conversationService, 'getActiveSessions').mockReturnValue(['chat-a', 'chat-b'])
spyOn(conversationService, 'hasSession').mockReturnValue(true)
const reload = spyOn(sessionReload, 'reloadSessionComponents').mockResolvedValue({
applied: true, commands: 1, agents: 0, plugins: 1, mcpServers: 1, errors: 0,
})
await reloadConnectorSessions('chat-a', remote)
expect(reload.mock.calls).toEqual([
['chat-a', 'plugin:office-supabase:service', remoteSkill],
['chat-b', 'plugin:office-supabase:service', remoteSkill],
])
reload.mockClear()
await reloadConnectorSessions('chat-a')
expect(reload.mock.calls).toEqual([['chat-a', undefined, undefined], ['chat-b', undefined, undefined]])
})
test('one chat failing to obtain tools prevents connector readiness; stopped chats need no tools', async () => {
spyOn(conversationService, 'getActiveSessions').mockReturnValue(['chat-a'])
spyOn(conversationService, 'hasSession').mockReturnValue(false)
const reload = spyOn(sessionReload, 'reloadSessionComponents').mockResolvedValue({
applied: false, reason: 'failed', commands: 0, agents: 0, plugins: 0, mcpServers: 0, errors: 0,
})
await expect(reloadConnectorSessions(undefined, remote)).rejects.toThrow('active task')
reload.mockResolvedValue({ applied: false, reason: 'not_running', commands: 0, agents: 0, plugins: 0, mcpServers: 0, errors: 0 })
await expect(reloadConnectorSessions(undefined, remote)).resolves.toBeUndefined()
})
+6 -3
View File
@@ -255,13 +255,16 @@ export function removeConnectorPlugin(definition: ConnectorDefinition): Promise<
}) })
} }
export async function reloadConnectorSessions(sessionId?: string): Promise<void> { export async function reloadConnectorSessions(sessionId?: string, requiredConnector?: ConnectorDefinition): Promise<void> {
const name = requiredConnector ? pluginName(requiredConnector) : undefined
const requiredPlugin = requiredConnector ? { pluginId: requiredConnector.pluginId, skillName: `${name}:${name}` } : undefined
const requiredMcpServer = requiredConnector?.transport === 'mcp' ? `plugin:${name}:service` : undefined
const { conversationService } = await import('../../server/services/conversationService.js') const { conversationService } = await import('../../server/services/conversationService.js')
const { reloadSessionComponents } = await import('../../server/services/sessionComponentReloadService.js') const { reloadSessionComponents } = await import('../../server/services/sessionComponentReloadService.js')
const sessions = new Set(conversationService.getActiveSessions()) const sessions = new Set(conversationService.getActiveSessions())
if (sessionId && conversationService.hasSession(sessionId)) sessions.add(sessionId) if (sessionId && conversationService.hasSession(sessionId)) sessions.add(sessionId)
const results = await Promise.all([...sessions].map(id => reloadSessionComponents(id))) const results = await Promise.all([...sessions].map(id => reloadSessionComponents(id, requiredMcpServer, requiredPlugin)))
if (results.some(result => result.reason === 'failed' || result.errors > 0)) { if (results.some(result => result.reason === 'failed' || result.errors > 0)) {
throw new Error('Connector changed on disk, but an active task could not refresh its skills. Retry the connection check before use.') throw new Error('Connector changed on disk, but an active task could not refresh its tools and skills. Retry the connection check before use.')
} }
} }
@@ -0,0 +1,174 @@
import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from 'bun:test'
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import type { LoadedPlugin, PluginError } from '../../types/plugin.js'
import * as optionStorage from '../../utils/plugins/pluginOptionsStorage.js'
import { extractMcpServersFromPlugins, getPluginMcpServers } from '../../utils/plugins/mcpPluginIntegration.js'
import { REMOTE_RECIPES } from './remoteCatalog.js'
import { buildRemotePlugin, createRemoteConnectorBridge, saveRemoteApiKey, type RemoteConnectorDependencies, type RemoteConnectorRecipe } from './remoteConnector.js'
type Call = { kind: string, name: string, config?: unknown }
const signal = () => new AbortController().signal
let root: string
let previousHome: string | undefined
let previousConfig: string | undefined
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'remote-catalog-runtime-'))
previousHome = process.env.HOME
previousConfig = process.env.CLAUDE_CONFIG_DIR
process.env.HOME = root
process.env.CLAUDE_CONFIG_DIR = root
})
afterEach(async () => {
mock.restore()
if (previousHome === undefined) delete process.env.HOME
else process.env.HOME = previousHome
if (previousConfig === undefined) delete process.env.CLAUDE_CONFIG_DIR
else process.env.CLAUDE_CONFIG_DIR = previousConfig
await rm(root, { recursive: true, force: true })
})
function fixture() {
const otherId = 'office-unrelated@haha-connectors'
const options: Record<string, optionStorage.PluginOptionValues> = { [otherId]: { apiKey: 'unrelated-fixture-key' } }
const calls: Call[] = []
const dependencies: RemoteConnectorDependencies = {
async probe(name, config, abortSignal) {
abortSignal.throwIfAborted()
calls.push({ kind: 'probe', name, config })
return { status: 'connected', toolCount: 2 }
},
async oauth(name, config, onUrl, abortSignal) {
abortSignal.throwIfAborted()
calls.push({ kind: 'oauth', name, config })
onUrl('https://accounts.example/authorize?state=offline-fixture')
},
async disconnect(name, config) { calls.push({ kind: 'disconnect', name, config }) },
loadOptions: id => options[id] ?? {},
saveOptions(id, values, schema) {
expect(schema.apiKey?.sensitive).toBe(true)
options[id] = values
},
deleteOptions(id) { delete options[id] },
clearOAuth(name, config) { calls.push({ kind: 'clear-oauth', name, config }) },
}
// Exercise the real file loader and user_config resolution without accessing
// saved settings, credentials, keychain, or any provider endpoint.
spyOn(optionStorage, 'loadPluginOptions').mockImplementation(dependencies.loadOptions)
return { dependencies, options, calls, otherId }
}
async function loadedPlugin(recipe: RemoteConnectorRecipe): Promise<LoadedPlugin> {
const plugin = buildRemotePlugin(recipe)
await writeFile(join(root, '.mcp.json'), JSON.stringify(plugin.mcpConfig))
return {
name: plugin.manifest.name, manifest: plugin.manifest,
path: root, source: recipe.pluginId, repository: recipe.pluginId, enabled: true,
}
}
// Catalog-driven: newly added services automatically inherit the contract.
// No live OAuth, network discovery, or business-tool execution is performed.
function runtimeContract(recipe: RemoteConnectorRecipe) {
test(`${recipe.id}: settings probe and both chat loaders use the same isolated server configuration`, async () => {
const f = fixture()
const bridge = createRemoteConnectorBridge(recipe, f.dependencies)
const plugin = await loadedPlugin(recipe)
const errors: PluginError[] = []
const rawKey = `fixture-${recipe.id}&scope=read#fragment`
if (recipe.auth.type === 'api-key') {
expect(bridge.isConfigured()).toBe(false)
await expect(bridge.check(signal())).rejects.toThrow('API key')
expect(await getPluginMcpServers(plugin, errors)).toEqual({})
expect(errors).toHaveLength(1)
expect(f.calls).toHaveLength(0)
errors.length = 0
saveRemoteApiKey(recipe, rawKey, f.dependencies)
expect(f.options[recipe.pluginId]?.apiKey).toBe(recipe.auth.in === 'query' ? encodeURIComponent(rawKey) : rawKey)
expect(JSON.stringify(buildRemotePlugin(recipe))).not.toContain(rawKey)
}
const config = (await getPluginMcpServers(plugin, errors))?.[bridge.serverName]
expect(config).toBeDefined()
expect(config).toMatchObject({ type: recipe.transport, scope: 'dynamic', pluginSource: recipe.pluginId })
expect(await extractMcpServersFromPlugins([plugin], errors)).toEqual({ [bridge.serverName]: config })
expect(errors).toEqual([])
const phases: string[] = []
await bridge.authenticate(signal(), phase => phases.push(phase))
expect(f.calls.filter(call => call.kind === 'probe')).toHaveLength(0)
expect(phases).toEqual(recipe.auth.type === 'oauth' ? ['authorizing', 'awaiting-authorization'] : [])
expect(f.calls.filter(call => call.kind === 'oauth')).toHaveLength(recipe.auth.type === 'oauth' ? 1 : 0)
expect(await bridge.check(signal())).toEqual({ authenticated: true, verification: 'remote', toolCount: 2 })
expect(f.calls.slice(-2).map(call => call.kind)).toEqual(['disconnect', 'probe'])
for (const call of f.calls) {
expect(call.name).toBe(bridge.serverName)
expect(call.config).toEqual(config)
}
if (recipe.auth.type === 'api-key') {
const replacement = `replacement-${recipe.id}&scope=write#fragment`
await bridge.deactivate()
expect(f.calls.at(-1)?.config).toEqual(config)
saveRemoteApiKey(recipe, replacement, f.dependencies)
// extractMcpServersFromPlugins caches unresolved templates. Both the
// cached loader and the bridge must resolve the newly saved credential.
const next = (await getPluginMcpServers(plugin, errors))?.[bridge.serverName]
expect(next).not.toEqual(config)
await bridge.check(signal())
expect(f.calls.at(-1)?.config).toEqual(next)
expect(JSON.stringify(plugin.mcpServers)).not.toContain(replacement)
if (next?.type === 'http' || next?.type === 'sse') {
if (recipe.auth.in === 'query') {
expect(new URL(next.url).searchParams.get(recipe.auth.name)).toBe(replacement)
expect(new URL(next.url).hash).toBe('')
} else expect(next.headers?.[recipe.auth.name]).toBe(`${recipe.auth.prefix ?? ''}${replacement}`)
}
}
await bridge.removeCredentials()
expect(f.options[recipe.pluginId]).toBeUndefined()
expect(f.options[f.otherId]).toEqual({ apiKey: 'unrelated-fixture-key' })
expect(f.calls.filter(call => call.kind === 'clear-oauth')).toHaveLength(recipe.auth.type === 'oauth' ? 1 : 0)
expect(await getPluginMcpServers({ ...plugin, enabled: false }, errors)).toBeUndefined()
})
test(`${recipe.id}: authorization alone, empty discovery, failed discovery and cancellation cannot pass readiness`, async () => {
const f = fixture()
if (recipe.auth.type === 'api-key') saveRemoteApiKey(recipe, 'offline-fixture-key', f.dependencies)
const bridge = createRemoteConnectorBridge(recipe, f.dependencies)
await bridge.authenticate(signal(), () => {})
f.dependencies.probe = async () => ({ status: 'needs-auth', toolCount: 0 })
expect(await bridge.check(signal())).toEqual({ authenticated: false, verification: 'remote', toolCount: 0 })
for (const status of ['connected', 'failed', 'disabled'] as const) {
f.dependencies.probe = async () => ({ status, toolCount: 0 })
await expect(bridge.check(signal())).rejects.toThrow('Unable to connect')
}
f.dependencies.probe = async () => { throw new Error('transport failed: fake-private-key') }
await expect(bridge.check(signal())).rejects.toThrow('Remote connector connection or tool discovery failed')
const count = f.calls.length
const controller = new AbortController()
controller.abort()
await expect(bridge.check(controller.signal)).rejects.toThrow()
expect(f.calls).toHaveLength(count)
})
}
describe('every remote catalog recipe obeys the shared runtime contract', () => {
for (const recipe of REMOTE_RECIPES) runtimeContract(recipe)
})
// Currently all catalog entries use HTTP. Retain explicit SSE coverage for
// every supported authentication shape without implying any live SSE service.
describe('SSE transport authentication matrix (synthetic fixtures)', () => {
for (const auth of [
{ type: 'oauth', clientId: 'offline-public-client' },
{ type: 'api-key', in: 'header', name: 'Authorization', prefix: 'Bearer ' },
{ type: 'api-key', in: 'query', name: 'key' },
{ type: 'none' },
] satisfies RemoteConnectorRecipe['auth'][]) {
const id = `sse-${auth.type}${auth.type === 'api-key' ? `-${auth.in}` : ''}`
runtimeContract({ id, pluginId: `office-${id}@haha-connectors`, version: '1.0.0', endpoint: 'https://fixture.example/sse?format=0', transport: 'sse', auth })
}
})
+28 -4
View File
@@ -342,9 +342,9 @@ export function getServerKey(
/** /**
* True when we have probed this server before (OAuth discovery state is * True when we have probed this server before (OAuth discovery state is
* stored) but hold no credentials to try. A connection attempt in this * stored) but hold no credentials to try. Keep automatic batches from
* state is guaranteed to 401 — the only way out is the user running * repeating unfinished OAuth discovery. An explicit successful connection
* /mcp to authenticate. * clears this evidence when the server accepts another authentication method.
*/ */
export function hasMcpDiscoveryButNoToken( export function hasMcpDiscoveryButNoToken(
serverName: string, serverName: string,
@@ -359,7 +359,31 @@ export function hasMcpDiscoveryButNoToken(
} }
const serverKey = getServerKey(serverName, serverConfig) const serverKey = getServerKey(serverName, serverConfig)
const entry = getSecureStorage().read()?.mcpOAuth?.[serverKey] const entry = getSecureStorage().read()?.mcpOAuth?.[serverKey]
return entry !== undefined && !entry.accessToken && !entry.refreshToken return entry?.discoveryState !== undefined && !entry.accessToken && !entry.refreshToken
}
/**
* A successful connection without OAuth tokens supersedes failed discovery.
* Remove only that regenerable evidence, preserving client registration,
* secrets, unknown fields, and credentials for every other configuration.
*/
export function clearMcpDiscoveryWithoutTokens(
serverName: string,
serverConfig: McpSSEServerConfig | McpHTTPServerConfig,
): void {
const storage = getSecureStorage()
const data = storage.read()
const serverKey = getServerKey(serverName, serverConfig)
const entry = data?.mcpOAuth?.[serverKey]
if (!entry?.discoveryState || entry.accessToken || entry.refreshToken) return
storage.update({
...data,
mcpOAuth: {
...data.mcpOAuth,
[serverKey]: { ...entry, discoveryState: undefined },
},
})
} }
/** /**
+157
View File
@@ -0,0 +1,157 @@
import '../../../preload.ts'
import { afterEach, beforeEach, expect, mock, spyOn, test } from 'bun:test'
import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Client } from '@modelcontextprotocol/sdk/client/index.js'
import { UnauthorizedError } from '@modelcontextprotocol/sdk/client/auth.js'
import * as secureStorage from '../../utils/secureStorage/index.js'
import { getServerKey, hasMcpDiscoveryButNoToken } from './auth.js'
import type { ScopedMcpServerConfig } from './types.js'
import { clearMcpAuthCache, clearServerCache, connectToServer, getMcpToolsCommandsAndResources } from './client.js'
const name = 'plugin:fixture-connector:service'
const otherName = 'plugin:other:service'
const config = { type: 'http' as const, url: 'http://127.0.0.1:1/mcp', scope: 'dynamic' as const }
let credentials: { mcpOAuth: Record<string, Record<string, unknown>> }
const connections: ScopedMcpServerConfig[] = []
let root: string
let previousConfigDir: string | undefined
const cachePath = () => join(root, 'mcp-needs-auth-cache.json')
const recentFailures = () => ({ [name]: { timestamp: Date.now() }, [otherName]: { timestamp: Date.now() } })
async function discover(serverConfig: ScopedMcpServerConfig = config) {
connections.push(serverConfig)
const results: Parameters<Parameters<typeof getMcpToolsCommandsAndResources>[0]>[0][] = []
await getMcpToolsCommandsAndResources(result => results.push(result), { [name]: serverConfig })
return results[0]!
}
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'mcp-auth-cache-'))
previousConfigDir = process.env.CLAUDE_CONFIG_DIR
process.env.CLAUDE_CONFIG_DIR = root
await clearMcpAuthCache()
// Model the token another process saved after OAuth without touching keychain.
credentials = { mcpOAuth: { [getServerKey(name, config)]: { accessToken: 'fixture-token', expiresAt: Date.now() + 60_000 } } }
spyOn(secureStorage, 'getSecureStorage').mockReturnValue({ name: 'fixture', read: () => credentials, readAsync: async () => credentials, update: next => { credentials = next; return { success: true } }, delete: () => true } as never)
spyOn(Client.prototype, 'connect').mockResolvedValue(undefined)
spyOn(Client.prototype, 'getServerCapabilities').mockReturnValue({ tools: {} })
spyOn(Client.prototype, 'request').mockResolvedValue({ tools: [{ name: 'list_projects', inputSchema: { type: 'object' } }] })
})
afterEach(async () => {
for (const serverConfig of [config, ...connections.splice(0)]) await clearServerCache(name, serverConfig)
await clearMcpAuthCache()
mock.restore()
if (previousConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR
else process.env.CLAUDE_CONFIG_DIR = previousConfigDir
await rm(root, { recursive: true, force: true })
})
const transports = ['http', 'sse'] as const
const authenticationModes = ['oauth', 'authorization-header', 'api-key-header', 'api-key-query', 'anonymous'] as const
for (const transport of transports) {
for (const authentication of authenticationModes) {
test(`${transport}/${authentication}: successful explicit connection replaces old failure evidence for subsequent chats`, async () => {
const serverConfig = {
...config,
type: transport,
...(authentication === 'authorization-header' && { headers: { Authorization: 'Bearer fixture-key' } }),
...(authentication === 'api-key-header' && { headers: { 'X-Api-Key': 'fixture-key' } }),
...(authentication === 'api-key-query' && { url: `${config.url}?api_key=fixture-key` }),
}
const serverKey = getServerKey(name, serverConfig)
// OAuth discovery can remain from a previous 401 even when the service
// now accepts an API key or anonymous requests without OAuth tokens.
credentials.mcpOAuth[serverKey] = {
accessToken: authentication === 'oauth' ? 'fixture-token' : '',
expiresAt: authentication === 'oauth' ? Date.now() + 60_000 : 0,
discoveryState: { authorizationServerUrl: 'https://auth.fixture.invalid' },
clientId: 'preserve-client-id',
clientSecret: 'preserve-client-secret',
futureField: 'preserve-unknown-data',
}
const otherKey = getServerKey(otherName, config)
credentials.mcpOAuth[otherKey] = { accessToken: '', discoveryState: { authorizationServerUrl: 'https://other.fixture.invalid' } }
const otherEntry = structuredClone(credentials.mcpOAuth[otherKey])
await writeFile(cachePath(), JSON.stringify(recentFailures()))
expect((await discover(serverConfig)).tools.map(tool => tool.name)).toEqual([expect.stringContaining('authenticate')])
expect((await connectToServer(name, serverConfig)).type).toBe('connected')
await clearServerCache(name, serverConfig)
const nextChat = await discover(serverConfig)
expect(nextChat.client.type).toBe('connected')
expect(nextChat.tools.map(tool => tool.mcpInfo?.toolName)).toEqual(['list_projects'])
const remaining = JSON.parse(await readFile(cachePath(), 'utf8'))
expect(remaining[name]).toBeUndefined()
expect(remaining[otherName]).toBeDefined()
expect(credentials.mcpOAuth[serverKey]).toMatchObject({ clientId: 'preserve-client-id', clientSecret: 'preserve-client-secret', futureField: 'preserve-unknown-data' })
expect(credentials.mcpOAuth[otherKey]).toEqual(otherEntry)
if (authentication === 'oauth') expect(credentials.mcpOAuth[serverKey]!.discoveryState).toBeDefined()
})
}
}
test('client registration without discovery is not evidence of an authentication failure', async () => {
credentials.mcpOAuth[getServerKey(name, config)] = { accessToken: '', clientId: 'fixture-client' }
expect(hasMcpDiscoveryButNoToken(name, config)).toBe(false)
expect((await discover()).client.type).toBe('connected')
})
test('successful connections preserve refresh-token-only OAuth credentials and discovery', async () => {
credentials.mcpOAuth[getServerKey(name, config)] = {
accessToken: '',
refreshToken: 'fixture-refresh-token',
discoveryState: { authorizationServerUrl: 'https://auth.fixture.invalid' },
}
const previousCredentials = structuredClone(credentials)
expect((await discover()).client.type).toBe('connected')
expect(credentials).toEqual(previousCredentials)
})
test('unresolved OAuth discovery still avoids repeated connection attempts after the short failure cache expires', async () => {
credentials.mcpOAuth[getServerKey(name, config)] = { accessToken: '', discoveryState: { authorizationServerUrl: 'https://auth.fixture.invalid' } }
await writeFile(cachePath(), JSON.stringify({ [name]: { timestamp: Date.now() - 16 * 60_000 } }))
expect((await discover()).client.type).toBe('needs-auth')
expect((await discover()).client.type).toBe('needs-auth')
expect(Client.prototype.connect).not.toHaveBeenCalled()
})
for (const placement of ['header', 'query'] as const) {
test(`changing an API key in the ${placement} isolates OAuth discovery from the old credentials`, async () => {
const withKey = (key: string) => ({ ...config, ...(placement === 'header' ? { headers: { 'X-Api-Key': key } } : { url: `${config.url}?api_key=${key}` }) })
const oldConfig = withKey('old-fixture-key')
const newConfig = withKey('new-fixture-key')
expect(getServerKey(name, newConfig)).not.toBe(getServerKey(name, oldConfig))
credentials.mcpOAuth = { [getServerKey(name, oldConfig)]: { accessToken: '', discoveryState: { authorizationServerUrl: 'https://auth.fixture.invalid' } } }
const previousCredentials = structuredClone(credentials)
await writeFile(cachePath(), JSON.stringify(recentFailures()))
expect((await connectToServer(name, newConfig)).type).toBe('connected')
expect((await discover(newConfig)).client.type).toBe('connected')
expect(credentials).toEqual(previousCredentials)
})
}
test('a new discovery batch observes another process clearing the cached OAuth failure', async () => {
await writeFile(cachePath(), JSON.stringify(recentFailures()))
expect((await discover()).client.type).toBe('needs-auth')
await writeFile(cachePath(), JSON.stringify({ [otherName]: { timestamp: Date.now() } }))
expect((await discover()).tools.map(tool => tool.mcpInfo?.toolName)).toEqual(['list_projects'])
})
test('targeted clears complete before returning and preserve unrelated failures', async () => {
await writeFile(cachePath(), JSON.stringify(recentFailures()))
await clearMcpAuthCache(name)
expect(JSON.parse(await readFile(cachePath(), 'utf8'))).toEqual({ [otherName]: expect.any(Object) })
})
test('authorization cleanup runs after an already queued 401 write', async () => {
await writeFile(cachePath(), JSON.stringify({ [otherName]: { timestamp: Date.now() } }))
spyOn(Client.prototype, 'connect').mockRejectedValueOnce(new UnauthorizedError())
expect((await connectToServer(name, config)).type).toBe('needs-auth')
await clearMcpAuthCache(name)
const remaining = JSON.parse(await readFile(cachePath(), 'utf8'))
expect(remaining[name]).toBeUndefined()
expect(remaining[otherName]).toBeDefined()
})
+15 -1
View File
@@ -1,5 +1,8 @@
import '../../../preload.ts' import '../../../preload.ts'
import { afterEach, describe, expect, mock, spyOn, test } from 'bun:test' import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from 'bun:test'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Client } from '@modelcontextprotocol/sdk/client/index.js' import { Client } from '@modelcontextprotocol/sdk/client/index.js'
import { import {
clearServerCache, clearServerCache,
@@ -12,6 +15,14 @@ import {
const config = { type: 'sse' as const, url: 'http://127.0.0.1:1/mcp' } const config = { type: 'sse' as const, url: 'http://127.0.0.1:1/mcp' }
const name = 'lifecycle-test' const name = 'lifecycle-test'
let root: string
let previousConfigDir: string | undefined
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'mcp-lifecycle-'))
previousConfigDir = process.env.CLAUDE_CONFIG_DIR
process.env.CLAUDE_CONFIG_DIR = root
})
function deferred() { function deferred() {
let resolve!: () => void let resolve!: () => void
@@ -29,6 +40,9 @@ afterEach(async () => {
setMcpConnectionClosedHandler(undefined) setMcpConnectionClosedHandler(undefined)
await clearServerCache(name, config) await clearServerCache(name, config)
mock.restore() mock.restore()
if (previousConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR
else process.env.CLAUDE_CONFIG_DIR = previousConfigDir
await rm(root, { recursive: true, force: true })
}) })
describe('MCP connection ownership', () => { describe('MCP connection ownership', () => {
+35 -6
View File
@@ -128,6 +128,7 @@ import { clearKeychainCache } from '../../utils/secureStorage/macOsKeychainHelpe
import { sleep } from '../../utils/sleep.js' import { sleep } from '../../utils/sleep.js'
import { import {
ClaudeAuthProvider, ClaudeAuthProvider,
clearMcpDiscoveryWithoutTokens,
hasMcpDiscoveryButNoToken, hasMcpDiscoveryButNoToken,
wrapFetchWithStepUpDetection, wrapFetchWithStepUpDetection,
} from './auth.js' } from './auth.js'
@@ -289,11 +290,15 @@ async function isMcpAuthCached(serverId: string): Promise<boolean> {
let writeChain = Promise.resolve() let writeChain = Promise.resolve()
function setMcpAuthCacheEntry(serverId: string): void { function setMcpAuthCacheEntry(serverId: string): void {
const cachePath = getMcpAuthCachePath()
writeChain = writeChain writeChain = writeChain
.then(async () => { .then(async () => {
const cache = await getMcpAuthCache() // Re-read before a write: another process may have completed OAuth and
// removed an entry since this process's last discovery batch.
const cache = await readFile(cachePath, 'utf-8')
.then(data => jsonParse(data) as McpAuthCacheData)
.catch(() => ({} as McpAuthCacheData))
cache[serverId] = { timestamp: Date.now() } cache[serverId] = { timestamp: Date.now() }
const cachePath = getMcpAuthCachePath()
await mkdir(dirname(cachePath), { recursive: true }) await mkdir(dirname(cachePath), { recursive: true })
await writeFile(cachePath, jsonStringify(cache)) await writeFile(cachePath, jsonStringify(cache))
// Invalidate the read cache so subsequent reads see the new entry. // Invalidate the read cache so subsequent reads see the new entry.
@@ -306,11 +311,27 @@ function setMcpAuthCacheEntry(serverId: string): void {
}) })
} }
export function clearMcpAuthCache(): void { export function clearMcpAuthCache(serverId?: string): Promise<void> {
const cachePath = getMcpAuthCachePath()
authCachePromise = null authCachePromise = null
void unlink(getMcpAuthCachePath()).catch(() => { // Join pending 401 writes so an older failure cannot recreate an entry
// Cache file may not exist // after successful authorization. Keep other servers' retry backoff intact.
}) writeChain = writeChain.then(async () => {
if (serverId === undefined) {
await unlink(cachePath).catch(() => {})
} else {
const cache = await readFile(cachePath, 'utf-8')
.then(data => jsonParse(data) as McpAuthCacheData)
.catch(() => ({} as McpAuthCacheData))
if (Object.hasOwn(cache, serverId)) {
delete cache[serverId]
await writeFile(cachePath, jsonStringify(cache))
}
}
}).catch(() => {
// This is a best-effort optimization, not credential storage.
}).finally(() => { authCachePromise = null })
return writeChain
} }
/** /**
@@ -1636,6 +1657,10 @@ const connectToServerMemoized = memoize(
} }
attempt.cleanup = wrappedCleanup attempt.cleanup = wrappedCleanup
if (serverRef.type === 'http' || serverRef.type === 'sse') {
clearMcpDiscoveryWithoutTokens(name, serverRef)
await clearMcpAuthCache(name)
}
const connectionDurationMs = Date.now() - connectStartTime const connectionDurationMs = Date.now() - connectStartTime
logEvent('tengu_mcp_server_connection_succeeded', { logEvent('tengu_mcp_server_connection_succeeded', {
connectionDurationMs, connectionDurationMs,
@@ -2331,6 +2356,10 @@ export async function getMcpToolsCommandsAndResources(
mcpConfigs?: Record<string, ScopedMcpServerConfig>, mcpConfigs?: Record<string, ScopedMcpServerConfig>,
): Promise<void> { ): Promise<void> {
let resourceToolsAdded = false let resourceToolsAdded = false
// OAuth may complete in the desktop server while this CLI stays alive.
// Share reads within a batch, but never freeze another process's old 401.
await writeChain
authCachePromise = null
const allConfigEntries = Object.entries( const allConfigEntries = Object.entries(
mcpConfigs ?? (await getAllMcpConfigs()).servers, mcpConfigs ?? (await getAllMcpConfigs()).servers,