diff --git a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts index 48fffe5211..d99b90a262 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts @@ -124,7 +124,10 @@ test('removes obsolete experimental Guest profiles and pairing intents at startu const credentials = createClientRuntimeHostCredentialStore(root); const catalog = createClientRuntimeHostProfileCatalog(root, credentials); const guest = { ...PROFILE, id: 'shared-obsolete', access: 'session_guest' as const }; - await createRuntimeHostProfileCredentialStore(credentials).set(guest, 'guest-token'); + await createRuntimeHostProfileCredentialStore(credentials).set(guest, { + credential: 'guest-token', + profileIncarnationId: 'guest-incarnation', + }); await writeFile( join(root, 'runtime-host-profiles.json'), `${JSON.stringify({ schemaVersion: 3, profiles: [guest] })}\n`, @@ -1301,6 +1304,7 @@ test("restores an existing profile when replacement finalization fails", async ( const root = await clientRoot(); const catalog = createClientRuntimeHostProfileCatalog(root); await catalog.create(PROFILE, "old-token"); + const original = await catalog.resolve(PROFILE.id); await writeFile( join(root, "runtime-host-profile-selection.json"), `${JSON.stringify({ @@ -1338,13 +1342,7 @@ test("restores an existing profile when replacement finalization fails", async ( /finalization failed/u, ); - assert.deepEqual(await catalog.resolve(PROFILE.id), { - profile: { - ...PROFILE, - transport: { kind: "tls", url: "wss://runtime.example.com/" }, - }, - credential: "old-token", - }); + assert.deepEqual(await catalog.resolve(PROFILE.id), original); assert.equal(enabled.at(-1)?.credential, "old-token"); }); diff --git a/packages/cli/src/__tests__/pi-tui-mcp-status.test.ts b/packages/cli/src/__tests__/pi-tui-mcp-status.test.ts index 81770f71bb..d255feb83f 100644 --- a/packages/cli/src/__tests__/pi-tui-mcp-status.test.ts +++ b/packages/cli/src/__tests__/pi-tui-mcp-status.test.ts @@ -19,6 +19,7 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; +import type { TUI } from '@earendil-works/pi-tui'; import { McpManagementOverlay } from '../pi-tui-mcp-status.js'; import type { TuiMcpAction, TuiMcpManagement, TuiMcpSnapshot } from '../tui-mcp-control.js'; import { stripAnsi } from '../tui-ansi.js'; @@ -68,6 +69,65 @@ describe('MCP management overlay', () => { assert.doesNotMatch(text, /尚未配置/u); }); + test('masks remote provider credentials while typing and clears them after submission', async () => { + const actions: TuiMcpAction[] = []; + const mcp = surface({ + initialization: 'ready', + configuration: 'ready', + publication: 'credential_rejected', + canManagePublicationCredential: true, + toolCount: 0, + servers: [], + }); + mcp.execute = async (action) => { + actions.push(action); + return { status: 'applied', effect: 'published' }; + }; + const overlay = new McpManagementOverlay({ + locale: 'en', + tui: {} as TUI, + surface: mcp, + viewportRows: () => 8, + onClose: () => undefined, + onChange: () => undefined, + }); + + let text = overlay.render(160).map(stripAnsi).join('\n'); + assert.match(text, /provider credential rejected/u); + assert.match(text, /p Set provider credential/u); + overlay.handleInput('p'); + overlay.handleInput('maka_rh_secret-marker'); + text = overlay.render(160).map(stripAnsi).join('\n'); + assert.match(text, /•••••••••••••••••••••/u); + assert.doesNotMatch(text, /maka_rh_secret-marker/u); + + overlay.handleInput('\n'); + await new Promise((resolve) => setImmediate(resolve)); + + assert.deepEqual(actions, [ + { kind: 'set_publication_credential', credential: 'maka_rh_secret-marker' }, + ]); + assert.doesNotMatch(overlay.render(160).map(stripAnsi).join('\n'), /maka_rh_secret-marker/u); + }); + + test('renders a competing remote provider as a visible conflict', () => { + const overlay = new McpManagementOverlay({ + locale: 'en', + surface: surface({ + initialization: 'ready', + configuration: 'ready', + publication: 'provider_conflict', + toolCount: 0, + servers: [], + }), + viewportRows: () => 6, + onClose: () => undefined, + onChange: () => undefined, + }); + + assert.match(overlay.render(100).map(stripAnsi).join('\n'), /provider active in another TUI/u); + }); + test('localizes manager states without changing their source values', () => { const overlay = new McpManagementOverlay({ locale: 'zh', diff --git a/packages/cli/src/__tests__/runtime-host-cli-context.test.ts b/packages/cli/src/__tests__/runtime-host-cli-context.test.ts index 9c64a96b38..e64df0fc6f 100644 --- a/packages/cli/src/__tests__/runtime-host-cli-context.test.ts +++ b/packages/cli/src/__tests__/runtime-host-cli-context.test.ts @@ -94,6 +94,7 @@ test('CLI Runtime Host bootstrap launches the execution composition', async () = assert.ok(candidateEntrypoint instanceof URL); assert.equal(basename(fileURLToPath(candidateEntrypoint)), 'execution-candidate-main.js'); assert.ok(clientInstanceId); + assert.equal(context.clientInstanceId, clientInstanceId); await context.close(); assert.equal(closes, 1); }); @@ -283,6 +284,7 @@ test('remote CLI profiles pin root identity and resolve credential outside the p rootId, }, credential: 'opaque-token', + profileIncarnationId: 'incarnation-a', }), create: async () => { throw new Error('unexpected write'); @@ -299,6 +301,12 @@ test('remote CLI profiles pin root identity and resolve credential outside the p rebindIfCurrent: async () => { throw new Error('unexpected write'); }, + mutateRemoteProfileIfCurrent: async () => { + throw new Error('unexpected write'); + }, + readRemoteProfileIfCurrent: async () => { + throw new Error('unexpected read'); + }, }, loadClientInstanceId: async () => '11111111-1111-4111-8111-111111111111', readConnectionCatalog: async () => ({ revision: 1, defaultTarget: null, connections: [] }), @@ -309,6 +317,8 @@ test('remote CLI profiles pin root identity and resolve credential outside the p assert.equal(remoteInput?.profile.rootId, rootId); assert.equal(remoteInput?.credential, 'opaque-token'); assert.equal(remoteInput?.clientInstanceId, '11111111-1111-4111-8111-111111111111'); + assert.equal(context.clientInstanceId, '11111111-1111-4111-8111-111111111111'); + assert.equal(context.profileIncarnationId, 'incarnation-a'); assert.equal(Object.hasOwn(context.profile, 'credential'), false); await context.close(); }); @@ -548,12 +558,18 @@ function singleRemoteProfileCatalog(profile: RemoteRuntimeHostProfile): RuntimeH read: async () => ({ schemaVersion: 3, profiles: [profile] }), resolve: async (profileId) => { assert.equal(profileId, profile.id); - return { profile, credential: 'opaque-token' }; + return { + profile, + credential: 'opaque-token', + profileIncarnationId: 'incarnation-a', + }; }, create: async () => assert.fail('unexpected write'), save: async () => assert.fail('unexpected write'), remove: async () => assert.fail('unexpected write'), removeIfCurrent: async () => assert.fail('unexpected write'), rebindIfCurrent: async () => assert.fail('unexpected write'), + mutateRemoteProfileIfCurrent: async () => assert.fail('unexpected write'), + readRemoteProfileIfCurrent: async () => assert.fail('unexpected read'), }; } diff --git a/packages/cli/src/__tests__/runtime-host-profile-command.test.ts b/packages/cli/src/__tests__/runtime-host-profile-command.test.ts index 4f89d2ce27..bf972eec2d 100644 --- a/packages/cli/src/__tests__/runtime-host-profile-command.test.ts +++ b/packages/cli/src/__tests__/runtime-host-profile-command.test.ts @@ -236,6 +236,8 @@ function createProfileCatalogCapture(): { remove: async () => assert.fail('unexpected profile removal'), removeIfCurrent: async () => assert.fail('unexpected conditional profile removal'), rebindIfCurrent: async () => assert.fail('unexpected conditional profile rebind'), + mutateRemoteProfileIfCurrent: async () => assert.fail('unexpected conditional mutation'), + readRemoteProfileIfCurrent: async () => assert.fail('unexpected conditional read'), }; return { get document() { diff --git a/packages/cli/src/__tests__/tui-mcp-control.test.ts b/packages/cli/src/__tests__/tui-mcp-control.test.ts index f4f788d259..3d5716131f 100644 --- a/packages/cli/src/__tests__/tui-mcp-control.test.ts +++ b/packages/cli/src/__tests__/tui-mcp-control.test.ts @@ -29,7 +29,7 @@ import type { RuntimeHostConnectionAvailability, } from '@maka/runtime-host/client'; import { createMcpConfigStore } from '@maka/storage/mcp-config-store'; -import { createTuiMcpController } from '../tui-mcp-control.js'; +import { createTuiMcpController, type TuiMcpPublicationAvailability } from '../tui-mcp-control.js'; import { waitFor } from './tui-terminal-mock.js'; test('TUI MCP startup stays backgrounded and publishes the discovered snapshot', async () => { @@ -71,6 +71,77 @@ test('TUI MCP startup stays backgrounded and publishes the discovered snapshot', assert.equal(manager.closed, 1); }); +test('TUI MCP serializes remote provider credential changes through its publication lane', async () => { + let availability: TuiMcpPublicationAvailability = { + kind: 'unavailable', + reason: 'credential_required', + }; + let listener: ((value: TuiMcpPublicationAvailability) => void) | undefined; + const credentials: string[] = []; + let removed = 0; + let closed = 0; + const connection = { + replaceClientCapabilities: async () => ({ registrationId: 'registration', revision: 1 }), + unregisterClientCapabilities: async () => ({ registrationId: 'registration', revision: 1 }), + subscribeConnectionAvailability: (next: (value: TuiMcpPublicationAvailability) => void) => { + listener = next; + next(availability); + return () => { + if (listener === next) listener = undefined; + }; + }, + setCredential: async (credential: string) => { + credentials.push(credential); + availability = { kind: 'connected', hostEpoch: 'host-1', connectionId: 'provider-1' }; + listener?.(availability); + }, + removeCredential: async () => { + removed += 1; + availability = { kind: 'unavailable', reason: 'credential_required' }; + listener?.(availability); + }, + closePublication: async () => { + closed += 1; + }, + }; + const manager = managerHarness(0, []); + const controller = createTuiMcpController( + { workspaceRoot: '/unused', connection }, + { + configStore: configStoreHarness(async () => emptyConfig()), + manager: manager.manager, + createProvider: () => undefined, + }, + ); + await waitFor( + () => controller.snapshot().initialization === 'ready', + 'remote MCP controller initialization', + ); + assert.equal(controller.snapshot().publication, 'credential_required'); + assert.equal(controller.snapshot().canManagePublicationCredential, true); + + assert.deepEqual( + await controller.execute({ + kind: 'set_publication_credential', + credential: 'provider-secret', + }), + { status: 'applied', effect: 'published' }, + ); + assert.deepEqual(credentials, ['provider-secret']); + assert.deepEqual(await controller.execute({ kind: 'remove_publication_credential' }), { + status: 'applied', + effect: 'pending_host', + }); + assert.equal(removed, 1); + assert.equal(controller.snapshot().publication, 'credential_required'); + availability = { kind: 'unavailable', reason: 'provider_conflict' }; + listener?.(availability); + assert.equal(controller.snapshot().publication, 'provider_conflict'); + assert.equal(controller.snapshot().canManagePublicationCredential, false); + await controller.close(); + assert.equal(closed, 1); +}); + test('TUI MCP publication coalesces a discovery change behind the in-flight revision', async () => { const manager = managerHarness(1, [connectedStatus('local', 1)]); const connection = connectionHarness(); diff --git a/packages/cli/src/__tests__/tui-mcp-remote-integration.test.ts b/packages/cli/src/__tests__/tui-mcp-remote-integration.test.ts new file mode 100644 index 0000000000..b643e24b7b --- /dev/null +++ b/packages/cli/src/__tests__/tui-mcp-remote-integration.test.ts @@ -0,0 +1,711 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'; +import { createServer as createNetServer } from 'node:net'; +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import test from 'node:test'; +import { + connectRemoteRuntimeHost, + connectRuntimeHost, + consumeAccessCredentialDelivery, + createClientRuntimeHostCredentialStore, + createClientRuntimeHostProfileCatalog, + createRuntimeHostCapabilityProviderCredentialStore, + type RemoteRuntimeHostProfile, + type RuntimeHostConnection, +} from '@maka/runtime-host/client'; +import { + INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, + RUNTIME_HOST_PROTOCOL_VERSION, +} from '@maka/runtime-host/protocol'; +import { startExecutionRuntimeHostService } from '@maka/runtime-host/server'; +import { mcpProxyToolName } from '@maka/runtime/mcp-tools'; +import { createMcpConfigStore } from '@maka/storage/mcp-config-store'; +import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores'; +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import { createRemoteTuiMcpPublicationTarget } from '../tui-mcp-remote-publication.js'; +import { createTuiMcpController, type TuiMcpController } from '../tui-mcp-control.js'; + +const PROTOCOL = { + min: RUNTIME_HOST_PROTOCOL_VERSION, + max: RUNTIME_HOST_PROTOCOL_VERSION, +} as const; + +test('remote TUI publication keeps its owner association across reconnect and revocation', { + timeout: 120_000, +}, async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-tui-remote-mcp-')); + const hostRoot = join(base, 'host'); + const clientRoot = join(base, 'client'); + const eventLog = join(base, 'stdio-events.jsonl'); + const port = await reservePort(); + const model = await startModelProvider(); + await seedModelConnection(hostRoot, model.baseUrl); + let host = await startHost(hostRoot, port); + let local: RuntimeHostConnection | undefined; + let terminal: RuntimeHostConnection | undefined; + let otherTerminal: RuntimeHostConnection | undefined; + let otherProvider: RuntimeHostConnection | undefined; + let controller: TuiMcpController | undefined; + let competingController: TuiMcpController | undefined; + try { + const capability = await resolveStorageRoot({ path: hostRoot, kind: 'interactive' }); + local = await connectLocal(hostRoot, 'local-owner'); + const firstOwner = await provisionOwner( + local, + hostRoot, + host.websocketEndpoints[0]!, + capability.rootId, + 'terminal-a', + ); + terminal = firstOwner.connection; + const firstProvider = await provisionProvider( + local, + hostRoot, + firstOwner.credentialId, + 'terminal-a-mcp', + ); + const secondOwner = await provisionOwner( + local, + hostRoot, + host.websocketEndpoints[0]!, + capability.rootId, + 'terminal-b', + ); + otherTerminal = secondOwner.connection; + const secondProvider = await provisionProvider( + local, + hostRoot, + secondOwner.credentialId, + 'terminal-b-mcp', + ); + assert.deepEqual(firstProvider.capabilityOwner, { + principalId: 'terminal-a', + clientInstanceId: 'terminal-a', + }); + assert.deepEqual(secondProvider.capabilityOwner, { + principalId: 'terminal-b', + clientInstanceId: 'terminal-b', + }); + otherProvider = await connectRemote( + host.websocketEndpoints[0]!, + capability.rootId, + secondProvider.credential, + 'provider-b', + ); + await otherProvider.replaceClientCapabilities(dummyProvider('provider-b')); + + const fixturePath = fileURLToPath( + new URL(import.meta.resolve('@maka/mcp/test-only/stdio-server')), + ); + await createMcpConfigStore(clientRoot).upsert('fixture', { + command: process.execPath, + args: [fixturePath], + env: { MAKA_MCP_STDIO_EVENT_LOG: eventLog }, + protocol: 'legacy', + }); + const profile = remoteProfile(host.websocketEndpoints[0]!, capability.rootId); + const profiles = createClientRuntimeHostProfileCatalog(clientRoot); + await profiles.create(profile, firstOwner.credential); + const resolvedProfile = await profiles.resolve(profile.id); + assert.ok(resolvedProfile.profileIncarnationId); + const profileTarget = { + profile, + profileIncarnationId: resolvedProfile.profileIncarnationId, + }; + const credentials = createRuntimeHostCapabilityProviderCredentialStore( + createClientRuntimeHostCredentialStore(clientRoot), + ); + await credentials.set(profileTarget, 'terminal-a', firstProvider.credential); + const publication = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: clientRoot, + profile, + profileIncarnationId: profileTarget.profileIncarnationId, + ownerClientInstanceId: 'terminal-a', + }, + { + credentials, + loadClientInstanceId: async () => 'provider-a', + }, + ); + controller = createTuiMcpController({ workspaceRoot: clientRoot, connection: publication }); + await waitFor(() => controller?.snapshot().publication === 'published'); + assert.equal(host.connectionCount, 5); + + const competingWorkspace = join(base, 'competing-workspace'); + await createMcpConfigStore(competingWorkspace).upsert('other-fixture', { + command: process.execPath, + args: [fixturePath], + enabled: false, + protocol: 'legacy', + }); + const competingPublication = createRemoteTuiMcpPublicationTarget({ + clientDataRoot: clientRoot, + profile, + profileIncarnationId: profileTarget.profileIncarnationId, + ownerClientInstanceId: 'terminal-a', + }); + competingController = createTuiMcpController({ + workspaceRoot: competingWorkspace, + connection: competingPublication, + }); + await waitFor(() => competingController?.snapshot().publication === 'provider_conflict'); + assert.equal(competingController.snapshot().canManagePublicationCredential, false); + assert.equal(controller.snapshot().publication, 'published'); + assert.equal(host.connectionCount, 5); + await competingController.close(); + competingController = undefined; + + const sessionId = 'remote-tui-mcp-session'; + const turnId = 'remote-tui-mcp-turn'; + await terminal.request('session.create', { + sessionId, + workspace: { kind: 'host_path', path: hostRoot }, + modelTarget: { kind: 'default' }, + permissionMode: 'bypass', + }); + const started = await terminal.request('turn.start', { + sessionId, + turnId, + content: { text: 'Call the fixture MCP echo tool.' }, + }); + assert.equal(started.kind, 'started'); + let completedTurn: Awaited> | undefined; + await waitFor(async () => { + const turn = await terminal?.request('turn.query', { sessionId, turnId }); + if ( + turn?.status !== 'completed' && + turn?.status !== 'failed' && + turn?.status !== 'cancelled' + ) { + return false; + } + completedTurn = turn; + return true; + }); + assert.equal( + completedTurn?.status, + 'completed', + JSON.stringify({ completedTurn, modelRequests: model.requestSummary() }), + ); + assert.equal(model.fixtureCalls(), 1); + assert.equal(model.observedToolResult(), 'remote-session-sentinel'); + + const wrongProfile = remoteProfile(host.websocketEndpoints[0]!, 'f'.repeat(64), 'wrong-office'); + await profiles.create(wrongProfile, firstOwner.credential); + const resolvedWrongProfile = await profiles.resolve(wrongProfile.id); + assert.ok(resolvedWrongProfile.profileIncarnationId); + const wrongProfileTarget = { + profile: wrongProfile, + profileIncarnationId: resolvedWrongProfile.profileIncarnationId, + }; + await credentials.set(wrongProfileTarget, 'terminal-a', firstProvider.credential); + const wrongTarget = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: clientRoot, + profile: wrongProfile, + profileIncarnationId: wrongProfileTarget.profileIncarnationId, + ownerClientInstanceId: 'terminal-a', + }, + { + credentials, + loadClientInstanceId: async () => 'provider-wrong-root', + }, + ); + let wrongTargetState = 'host_unavailable'; + const disposeWrongTarget = wrongTarget.subscribeConnectionAvailability((availability) => { + wrongTargetState = + availability.kind === 'unavailable' + ? (availability.reason ?? 'host_unavailable') + : 'connected'; + }); + try { + await waitFor(() => wrongTargetState === 'target_mismatch'); + assert.equal(wrongTargetState, 'target_mismatch'); + } finally { + disposeWrongTarget(); + await wrongTarget.closePublication?.(); + } + + await Promise.all([ + otherProvider.close(), + otherTerminal.close(), + terminal.close(), + local.close(), + ]); + otherProvider = undefined; + otherTerminal = undefined; + terminal = undefined; + local = undefined; + await host.close(); + await waitFor(() => controller?.snapshot().publication === 'host_unavailable'); + host = await startHost(hostRoot, port); + await waitFor(() => controller?.snapshot().publication === 'published'); + + local = await connectLocal(hostRoot, 'local-owner-after-restart'); + await local.request('access.credential.revoke', { + credentialId: firstProvider.credentialId, + }); + await waitFor(() => controller?.snapshot().publication === 'credential_rejected'); + assert.equal(await credentials.get(profileTarget, 'terminal-a'), firstProvider.credential); + + await createClientRuntimeHostProfileCatalog(clientRoot).remove(profile.id); + await waitFor(() => controller?.snapshot().publication === 'target_mismatch'); + + await controller.close(); + controller = undefined; + await waitFor(async () => + (await fixtureEvents(eventLog)).some((event) => event.event === 'exit'), + ); + const events = await fixtureEvents(eventLog); + assert.equal(events.filter((event) => event.event === 'start').length, 1); + assert.equal(events.filter((event) => event.event === 'exit').length, 1); + } finally { + await competingController?.close().catch(() => undefined); + await controller?.close().catch(() => undefined); + await otherProvider?.close().catch(() => undefined); + await otherTerminal?.close().catch(() => undefined); + await terminal?.close().catch(() => undefined); + await local?.close().catch(() => undefined); + await host.close().catch(() => undefined); + await model.close().catch(() => undefined); + await rm(base, { recursive: true, force: true }); + } +}); + +async function startHost(rootPath: string, port: number) { + return startExecutionRuntimeHostService({ + rootPath, + websocket: { host: '127.0.0.1', port, allowInsecureRemote: true }, + }); +} + +async function provisionOwner( + local: RuntimeHostConnection, + rootPath: string, + url: string, + rootId: string, + clientInstanceId: string, +): Promise<{ + readonly credentialId: string; + readonly credential: string; + readonly connection: RuntimeHostConnection; +}> { + const candidate = await local.request('access.credential.prepare', { + principalKind: 'remote_owner', + principalId: clientInstanceId, + operationGrants: [ + 'access.credential.finalize', + 'session.catalog.query', + 'session.create', + 'turn.start', + 'turn.query', + ], + canPublishClientCapabilities: false, + canUseHostPaths: true, + bindClientInstance: true, + }); + const credential = await consumeAccessCredentialDelivery( + rootPath, + candidate.deliveryId, + candidate.credentialId, + ); + const pairing = await connectRemote(url, rootId, credential, clientInstanceId); + assert.deepEqual(await pairing.request('access.credential.finalize', {}), { + reconnectRequired: true, + }); + await pairing.close(); + return { + credentialId: candidate.credentialId, + credential, + connection: await connectRemote(url, rootId, credential, clientInstanceId), + }; +} + +async function provisionProvider( + local: RuntimeHostConnection, + rootPath: string, + ownerCredentialId: string, + principalId: string, +): Promise<{ + readonly credentialId: string; + readonly credential: string; + readonly capabilityOwner?: { readonly principalId: string; readonly clientInstanceId: string }; +}> { + const issued = await local.request('access.credential.issue', { + principalKind: 'capability_provider', + principalId, + operationGrants: ['host.status', 'client.capability.replace', 'client.capability.unregister'], + canPublishClientCapabilities: true, + canUseHostPaths: false, + capabilityOwnerCredentialId: ownerCredentialId, + }); + return { + credentialId: issued.credentialId, + capabilityOwner: issued.capabilityOwner, + credential: await consumeAccessCredentialDelivery( + rootPath, + issued.deliveryId, + issued.credentialId, + ), + }; +} + +async function connectLocal(rootPath: string, clientInstanceId: string) { + const result = await connectRuntimeHost({ rootPath, clientInstanceId, protocol: PROTOCOL }); + assert.equal(result.kind, 'connected'); + if (result.kind !== 'connected') throw new Error('Unable to connect to local Runtime Host'); + return result.connection; +} + +async function connectRemote( + url: string, + expectedRootId: string, + credential: string, + clientInstanceId: string, +): Promise { + const result = await connectRemoteRuntimeHost({ + url, + allowInsecureRemote: true, + credential, + expectedRootId, + compositionId: INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, + clientInstanceId, + protocol: PROTOCOL, + }); + assert.equal(result.kind, 'connected'); + if (result.kind !== 'connected') throw new Error('Unable to connect to remote Runtime Host'); + return result.connection; +} + +function remoteProfile(url: string, rootId: string, id = 'office'): RemoteRuntimeHostProfile { + return { + id, + name: 'Office', + kind: 'remote', + transport: { kind: 'plaintext', url, acknowledgement: 'plaintext-bearer-v1' }, + rootId, + }; +} + +function dummyProvider(id: string) { + return { + offers: () => [ + { + offerId: id, + version: '1', + affinity: 'session' as const, + hostPathAccess: 'none' as const, + label: id, + tools: [ + { + serverId: id, + name: 'echo', + inputSchema: { type: 'object' }, + }, + ], + }, + ], + call: async () => ({ content: [{ type: 'text' as const, text: id }] }), + }; +} + +async function fixtureEvents(path: string): Promise> { + try { + return (await readFile(path, 'utf8')) + .trim() + .split('\n') + .filter(Boolean) + .map((line) => JSON.parse(line) as { event: string }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw error; + } +} + +async function seedModelConnection(rootPath: string, baseUrl: string): Promise { + const capability = await resolveStorageRoot({ path: rootPath, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) throw new Error('Unable to acquire model fixture root'); + try { + const policy = await openInteractiveRuntimePolicyStoresForWrite(owner.lease); + const created = await policy.connectionCatalog.create({ + expectedCatalogRevision: 0, + connection: { + slug: 'remote-mcp-fixture-model', + name: 'Remote MCP fixture model', + providerType: 'moonshot', + baseUrl, + enabled: true, + enabledModelIds: ['hosted-real-model'], + }, + }); + assert.equal(created.kind, 'committed'); + if (created.kind !== 'committed') throw new Error('Model fixture connection did not commit'); + const connection = created.snapshot.connections[0]; + assert.ok(connection); + if (!connection) throw new Error('Model fixture connection was not persisted'); + assert.equal( + ( + await policy.credentialVault.set({ + locator: { + scope: 'connection', + connectionId: connection.connectionId, + kind: 'api_key', + }, + expected: null, + secret: 'fixture-model-key', + }) + ).kind, + 'committed', + ); + const prepared = await policy.operations.beginModelFetch(connection.connectionId); + assert.equal(prepared.kind, 'ready'); + if (prepared.kind !== 'ready') throw new Error('Model fixture discovery was not ready'); + const discovered = await policy.operations.completeModelFetch(prepared.ticket, { + models: [ + { + id: 'hosted-real-model', + capabilities: { chat: true, functionCalling: true }, + contextWindow: 8_192, + maxOutputTokens: 128, + }, + ], + source: 'fetched', + fetchedAt: Date.now(), + }); + assert.equal(discovered.kind, 'committed'); + if (discovered.kind !== 'committed') throw new Error('Model fixture discovery did not commit'); + assert.equal( + ( + await policy.connectionCatalog.setDefaultTarget({ + expectedCatalogRevision: discovered.snapshot.revision, + target: { + connectionId: connection.connectionId, + modelId: 'hosted-real-model', + }, + }) + ).kind, + 'committed', + ); + } finally { + await owner.close(); + } +} + +async function startModelProvider(): Promise<{ + readonly baseUrl: string; + fixtureCalls(): number; + observedToolResult(): string | undefined; + requestSummary(): readonly unknown[]; + close(): Promise; +}> { + const proxyToolName = mcpProxyToolName('fixture', 'echo'); + let streamRequests = 0; + let fixtureCalls = 0; + let observedToolResult: string | undefined; + const requestSummary: unknown[] = []; + const server = createServer((request, response) => { + void readRequestBody(request) + .then((body) => { + const input = JSON.parse(body) as Record; + requestSummary.push({ stream: input.stream, tools: modelToolNames(input) }); + if (input.stream !== true) { + respondModelSummary(response); + return; + } + streamRequests += 1; + if (streamRequests === 1) { + assert.ok(modelToolNames(input).includes('tool_search')); + respondModelToolCall(response, streamRequests, 'tool_search', { + query: proxyToolName, + }); + return; + } + if (streamRequests === 2) { + assert.ok(modelToolNames(input).includes(proxyToolName)); + fixtureCalls += 1; + respondModelToolCall(response, streamRequests, proxyToolName, { + value: 'remote-session-sentinel', + }); + return; + } + const serialized = JSON.stringify(input); + if (serialized.includes('remote-session-sentinel')) { + observedToolResult = 'remote-session-sentinel'; + } + respondModelText(response, 'Remote MCP fixture completed.'); + }) + .catch((error) => response.destroy(error as Error)); + }); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const address = server.address(); + assert.ok(address && typeof address !== 'string'); + return { + baseUrl: `http://127.0.0.1:${address.port}/v1`, + fixtureCalls: () => fixtureCalls, + observedToolResult: () => observedToolResult, + requestSummary: () => requestSummary, + close: () => + new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ), + }; +} + +function modelToolNames(body: Record): string[] { + return (Array.isArray(body.tools) ? body.tools : []).flatMap((tool) => { + if (!tool || typeof tool !== 'object') return []; + const fn = (tool as { function?: unknown }).function; + if (!fn || typeof fn !== 'object') return []; + const name = (fn as { name?: unknown }).name; + return typeof name === 'string' ? [name] : []; + }); +} + +function respondModelToolCall( + response: ServerResponse, + step: number, + toolName: string, + args: Record, +): void { + respondModelEvents(response, [ + { + id: `chatcmpl-remote-mcp-${step}`, + object: 'chat.completion.chunk', + created: step, + model: 'hosted-real-model', + choices: [ + { + index: 0, + delta: { + role: 'assistant', + tool_calls: [ + { + index: 0, + id: `remote-mcp-tool-call-${step}`, + type: 'function', + function: { name: toolName, arguments: JSON.stringify(args) }, + }, + ], + }, + finish_reason: null, + }, + ], + }, + { + id: `chatcmpl-remote-mcp-${step}`, + object: 'chat.completion.chunk', + created: step, + model: 'hosted-real-model', + choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }], + usage: { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, + }, + ]); +} + +function respondModelText(response: ServerResponse, text: string): void { + respondModelEvents(response, [ + { + id: 'chatcmpl-remote-mcp-complete', + object: 'chat.completion.chunk', + created: 3, + model: 'hosted-real-model', + choices: [{ index: 0, delta: { role: 'assistant', content: text }, finish_reason: null }], + }, + { + id: 'chatcmpl-remote-mcp-complete', + object: 'chat.completion.chunk', + created: 3, + model: 'hosted-real-model', + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + usage: { prompt_tokens: 11, completion_tokens: 5, total_tokens: 16 }, + }, + ]); +} + +function respondModelEvents(response: ServerResponse, events: readonly unknown[]): void { + response.writeHead(200, { 'content-type': 'text/event-stream' }); + for (const event of events) response.write(`data: ${JSON.stringify(event)}\n\n`); + response.end('data: [DONE]\n\n'); +} + +function respondModelSummary(response: ServerResponse): void { + response.writeHead(200, { 'content-type': 'application/json' }); + response.end( + JSON.stringify({ + id: 'chatcmpl-remote-mcp-summary', + object: 'chat.completion', + created: 1, + model: 'hosted-real-model', + choices: [ + { + index: 0, + message: { role: 'assistant', content: 'Remote MCP Session' }, + finish_reason: 'stop', + }, + ], + usage: { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 }, + }), + ); +} + +function readRequestBody(request: IncomingMessage): Promise { + return new Promise((resolve, reject) => { + let body = ''; + request.setEncoding('utf8'); + request.on('data', (chunk) => { + body += chunk; + }); + request.on('end', () => resolve(body)); + request.on('error', reject); + }); +} + +async function reservePort(): Promise { + const server = createNetServer(); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const address = server.address(); + assert.ok(address && typeof address !== 'string'); + const port = address.port; + await new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ); + return port; +} + +async function waitFor(condition: () => boolean | Promise): Promise { + for (let attempt = 0; attempt < 1_500 && !(await condition()); attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.ok(await condition()); +} diff --git a/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts b/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts new file mode 100644 index 0000000000..1e366974ad --- /dev/null +++ b/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts @@ -0,0 +1,902 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + RuntimeHostProfileConnectionError, + sameRemoteRuntimeHostProfileTarget, + type RemoteRuntimeHostProfile, + type RuntimeHostCapabilityProviderCredentialStore, + type RuntimeHostConnection, + type RuntimeHostPeerClient, + type RuntimeHostProfileCatalog, + type RuntimeHostRemoteProfileIncarnation, +} from '@maka/runtime-host/client'; +import { createRemoteTuiMcpPublicationTarget as createProductionRemoteTuiMcpPublicationTarget } from '../tui-mcp-remote-publication.js'; +import { waitFor } from './tui-terminal-mock.js'; + +const PROFILE: RemoteRuntimeHostProfile = { + id: 'office', + name: 'Office', + kind: 'remote', + transport: { kind: 'tls', url: 'wss://runtime.example.com/runtime-host' }, + rootId: 'a'.repeat(64), +}; + +const DIRECT_PROFILE: RemoteRuntimeHostProfile = { + ...PROFILE, + transport: { + kind: 'libp2p-direct', + peerId: 'peer-a', + routeHints: ['/ip4/127.0.0.1/tcp/4001'], + coordinationRelays: [], + }, +}; + +const PROFILE_INCARNATION_ID = 'incarnation-a'; + +test('remote TUI publication activates, rotates, and removes one profile-bound credential', async () => { + const credentials = credentialHarness(); + const connected: Array<{ credential?: string; clientInstanceId: string }> = []; + const connections: ConnectionHarness[] = []; + const identityPaths: string[] = []; + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + ...profileDeps(), + credentials: credentials.store, + loadClientInstanceId: async (path) => { + identityPaths.push(path); + return 'provider-client'; + }, + connectProfile: async (input) => { + connected.push({ + credential: input.credential, + clientInstanceId: input.clientInstanceId, + }); + const connection = connectionHarness(`connection-${connections.length + 1}`); + connections.push(connection); + return connection.connection; + }, + }, + ); + let latest = await availability(target); + assert.deepEqual(latest(), { kind: 'unavailable', reason: 'credential_required' }); + + await target.setCredential?.('provider-secret-a'); + await waitFor(() => latest().kind === 'connected', 'provider companion to connect'); + assert.deepEqual(connected, [ + { credential: 'provider-secret-a', clientInstanceId: 'provider-client' }, + ]); + assert.equal( + credentials.values.get('office\0incarnation-a\0terminal-client'), + 'provider-secret-a', + ); + assert.match(identityPaths[0] ?? '', /capability-provider-identities/u); + + await target.setCredential?.('provider-secret-b'); + await waitFor( + () => latest().kind === 'connected' && connections.length === 2, + 'rotated provider companion to connect', + ); + assert.equal(connections[0]?.unregisters, 0); + assert.equal(connections[0]?.closes, 1); + assert.equal( + credentials.values.get('office\0incarnation-a\0terminal-client'), + 'provider-secret-b', + ); + assert.equal(identityPaths[0], identityPaths[1]); + + await target.removeCredential?.(); + assert.deepEqual(latest(), { kind: 'unavailable', reason: 'credential_required' }); + assert.equal(connections[1]?.unregisters, 0); + assert.equal(connections[1]?.closes, 1); + assert.equal(credentials.values.has('office\0incarnation-a\0terminal-client'), false); + await target.closePublication?.(); +}); + +test('remote TUI publication fails closed while the same provider lifetime is active', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(); + const leases = publicationLeaseHarness(); + const firstConnection = connectionHarness('connection-1'); + const secondConnection = connectionHarness('connection-2'); + let secondAttempts = 0; + const input = { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + } as const; + const first = createRemoteTuiMcpPublicationTarget(input, { + credentials: credentials.store, + loadClientInstanceId: async () => 'shared-provider-client', + connectProfile: async () => firstConnection.connection, + acquirePublicationLease: leases.acquire, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }); + const firstLatest = await availability(first); + await waitFor(() => firstLatest().kind === 'connected', 'first provider companion to connect'); + + const second = createRemoteTuiMcpPublicationTarget(input, { + credentials: credentials.store, + loadClientInstanceId: async () => 'shared-provider-client', + connectProfile: async () => { + secondAttempts += 1; + return secondConnection.connection; + }, + acquirePublicationLease: leases.acquire, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }); + const secondLatest = await availability(second); + await waitFor(() => { + const current = secondLatest(); + return current.kind === 'unavailable' && current.reason === 'provider_conflict'; + }, 'competing provider companion to fail closed'); + assert.equal(secondAttempts, 0); + assert.equal(leases.active.size, 1); + + await first.replaceClientCapabilities({ offers: () => [] }); + assert.equal(firstConnection.replacements, 1); + await first.closePublication?.(); + assert.equal(leases.active.size, 0); + + const successor = createRemoteTuiMcpPublicationTarget(input, { + credentials: credentials.store, + loadClientInstanceId: async () => 'shared-provider-client', + connectProfile: async () => secondConnection.connection, + acquirePublicationLease: leases.acquire, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }); + const successorLatest = await availability(successor); + await waitFor( + () => successorLatest().kind === 'connected', + 'successor provider companion to connect', + ); + await successor.closePublication?.(); + await second.closePublication?.(); +}); + +test('remote TUI publication cannot bypass a failed lifetime lease', async () => { + const credentials = credentialHarness(); + let attempts = 0; + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + ...profileDeps(), + credentials: credentials.store, + acquirePublicationLease: async () => { + throw new Error('lease storage unavailable'); + }, + connectProfile: async () => { + attempts += 1; + return connectionHarness('unexpected').connection; + }, + }, + ); + const latest = await availability(target); + await waitFor(() => { + const current = latest(); + return current.kind === 'unavailable' && current.reason === 'host_unavailable'; + }, 'lease failure to retire provider companion'); + + const setCredential = target.setCredential?.('provider-secret'); + assert.ok(setCredential); + await assert.rejects(setCredential, /closed/u); + assert.equal(attempts, 0); + assert.equal(credentials.values.size, 0); + await target.closePublication?.(); +}); + +test('remote TUI publication surfaces rejected credentials without a retry authority', async () => { + const credentials = credentialHarness('revoked-secret'); + let attempts = 0; + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + ...profileDeps(), + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => { + attempts += 1; + throw new RuntimeHostProfileConnectionError( + 'credential_rejected', + 'Runtime Host rejected its access credential', + ); + }, + }, + ); + const latest = await availability(target); + await waitFor(() => { + const current = latest(); + return current.kind === 'unavailable' && current.reason === 'credential_rejected'; + }, 'rejected provider credential state'); + assert.equal(attempts, 1); + await target.closePublication?.(); +}); + +test('remote TUI publication aborts an in-flight connection before closing', async () => { + const credentials = credentialHarness('provider-secret'); + let observedSignal: AbortSignal | undefined; + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + ...profileDeps(), + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async (input) => { + observedSignal = input.signal; + return new Promise((_resolve, reject) => { + input.signal?.addEventListener('abort', () => reject(input.signal?.reason), { + once: true, + }); + }); + }, + }, + ); + await waitFor(() => observedSignal !== undefined, 'provider connection attempt to start'); + + await target.closePublication?.(); + + assert.equal(observedSignal?.aborted, true); +}); + +test('remote TUI publication revalidates its profile before publishing an initial connection', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(); + const connection = connectionHarness('connection-1'); + const wrapperStarted = deferred(); + const allowWrapper = deferred(); + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => connection.connection, + createReconnectingConnection: async () => { + wrapperStarted.resolve(); + await allowWrapper.promise; + return reconnectingConnection(connection.connection); + }, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await wrapperStarted.promise; + + profiles.remove(); + profiles.recreate('incarnation-b'); + allowWrapper.resolve(); + + await waitFor(() => { + const current = latest(); + return current.kind === 'unavailable' && current.reason === 'target_mismatch'; + }, 'recreated profile to reject the uninstalled connection'); + assert.equal(connection.closes, 1); + await target.closePublication?.(); +}); + +test('remote TUI publication revalidates every reconnect result', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(); + const initial = connectionHarness('connection-1'); + const replacement = connectionHarness('connection-2'); + const reconnectStarted = deferred(); + const allowReconnect = deferred(); + let attempts = 0; + let reconnect: ((signal: AbortSignal) => Promise) | undefined; + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => { + attempts += 1; + if (attempts === 1) return initial.connection; + reconnectStarted.resolve(); + await allowReconnect.promise; + return replacement.connection; + }, + createReconnectingConnection: async (input) => { + reconnect = input.connect; + return reconnectingConnection(initial.connection); + }, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await waitFor(() => latest().kind === 'connected', 'provider companion to connect'); + assert.ok(reconnect); + + const attempt = reconnect(new AbortController().signal); + await reconnectStarted.promise; + profiles.remove(); + profiles.recreate('incarnation-b'); + allowReconnect.resolve(); + + await assert.rejects( + attempt, + (error: unknown) => + error instanceof RuntimeHostProfileConnectionError && error.reason === 'target_mismatch', + ); + assert.equal(replacement.closes, 1); + await target.closePublication?.(); +}); + +test('remote TUI publication reconnects through current direct-peer routes', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(DIRECT_PROFILE); + const connectedProfiles: RemoteRuntimeHostProfile[] = []; + const initial = connectionHarness('connection-1'); + const replacement = connectionHarness('connection-2'); + let reconnect: ((signal: AbortSignal) => Promise) | undefined; + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: DIRECT_PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async (input) => { + assert.equal(input.profile.kind, 'remote'); + if (input.profile.kind !== 'remote') throw new Error('expected a remote profile'); + connectedProfiles.push(input.profile); + return connectedProfiles.length === 1 ? initial.connection : replacement.connection; + }, + createPeerClient: () => ({ close: async () => undefined }) as RuntimeHostPeerClient, + createReconnectingConnection: async (input) => { + reconnect = input.connect; + return reconnectingConnection(initial.connection); + }, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await waitFor(() => latest().kind === 'connected', 'direct provider companion to connect'); + assert.ok(reconnect); + + const moved: RemoteRuntimeHostProfile = { + ...DIRECT_PROFILE, + transport: { + kind: 'libp2p-direct', + peerId: 'peer-a', + routeHints: ['/ip6/2001:db8::10/udp/4001/quic-v1'], + coordinationRelays: ['/dns4/relay.example.com/tcp/443/wss/p2p/relay-a'], + }, + }; + profiles.update(moved); + await reconnect(new AbortController().signal); + + assert.deepEqual(connectedProfiles, [DIRECT_PROFILE, moved]); + await target.closePublication?.(); +}); + +test('remote TUI publication retires when another process removes its profile', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(); + const connection = connectionHarness('connection-1'); + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => connection.connection, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await waitFor(() => latest().kind === 'connected', 'provider companion to connect'); + + profiles.remove(); + + await waitFor(() => { + const current = latest(); + return current.kind === 'unavailable' && current.reason === 'target_mismatch'; + }, 'removed profile to retire provider companion'); + assert.equal(connection.closes, 1); + await assert.rejects(async () => { + await target.setCredential?.('replacement-secret'); + }); +}); + +test('remote TUI publication cannot write into a recreated profile incarnation', async () => { + const credentials = credentialHarness(); + const profiles = profileHarness(); + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + assert.deepEqual(latest(), { kind: 'unavailable', reason: 'credential_required' }); + + const heldMutation = profiles.holdNextMutation(); + const setCredential = target.setCredential?.('replacement-secret'); + assert.ok(setCredential); + await heldMutation.started; + profiles.remove(); + profiles.recreate('incarnation-b'); + heldMutation.release(); + + await assert.rejects(setCredential, /profile is no longer current/u); + assert.equal(credentials.values.has('office\0incarnation-a\0terminal-client'), false); + assert.equal(credentials.values.has('office\0incarnation-b\0terminal-client'), false); + assert.deepEqual(latest(), { kind: 'unavailable', reason: 'target_mismatch' }); + await target.closePublication?.(); +}); + +test('remote TUI publication cannot register capabilities after profile recreation', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(); + const connection = connectionHarness('connection-1'); + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => connection.connection, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await waitFor(() => latest().kind === 'connected', 'provider companion to connect'); + + const heldMutation = profiles.holdNextMutation(); + const replace = target.replaceClientCapabilities({ offers: () => [] }); + await heldMutation.started; + profiles.remove(); + profiles.recreate('incarnation-b'); + heldMutation.release(); + + await assert.rejects( + replace, + (error: unknown) => + error instanceof RuntimeHostProfileConnectionError && error.reason === 'target_mismatch', + ); + assert.equal(connection.replacements, 0); + assert.deepEqual(latest(), { kind: 'unavailable', reason: 'target_mismatch' }); + await target.closePublication?.(); +}); + +test('remote TUI publication retires across a coalesced same-target recreation', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(); + const connection = connectionHarness('connection-1'); + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => connection.connection, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await waitFor(() => latest().kind === 'connected', 'provider companion to connect'); + + const heldValidation = profiles.holdNextValidation(); + profiles.invalidate(); + await heldValidation.started; + profiles.remove(); + profiles.recreate('incarnation-b'); + heldValidation.release(); + + await waitFor(() => { + const current = latest(); + return current.kind === 'unavailable' && current.reason === 'target_mismatch'; + }, 'later profile invalidation to retire provider companion'); + assert.equal(connection.closes, 1); + await target.closePublication?.(); +}); + +test('remote TUI publication close waits for concurrent profile retirement cleanup', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(DIRECT_PROFILE); + const connectionCloseStarted = deferred(); + const allowConnectionClose = deferred(); + const peerCloseStarted = deferred(); + const allowPeerClose = deferred(); + let peerCloses = 0; + const connection = connectionHarness('connection-1', async () => { + connectionCloseStarted.resolve(); + await allowConnectionClose.promise; + }); + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: DIRECT_PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => connection.connection, + createPeerClient: () => + ({ + close: async () => { + peerCloses += 1; + peerCloseStarted.resolve(); + await allowPeerClose.promise; + }, + }) as RuntimeHostPeerClient, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await waitFor(() => latest().kind === 'connected', 'provider companion to connect'); + + profiles.remove(); + await connectionCloseStarted.promise; + let closeSettled = false; + const close = target.closePublication?.().then(() => { + closeSettled = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(closeSettled, false); + + allowConnectionClose.resolve(); + await peerCloseStarted.promise; + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(closeSettled, false); + + allowPeerClose.resolve(); + await close; + assert.equal(closeSettled, true); + assert.equal(connection.closes, 1); + assert.equal(peerCloses, 1); +}); + +test('remote TUI publication closes its direct peer after a permanent reconnect failure', async () => { + const credentials = credentialHarness('provider-secret'); + const profiles = profileHarness(DIRECT_PROFILE); + const initial = connectionHarness('connection-1'); + let peerCloses = 0; + let fatal: ((error: Error) => void) | undefined; + const target = createRemoteTuiMcpPublicationTarget( + { + clientDataRoot: '/client-data', + profile: DIRECT_PROFILE, + profileIncarnationId: PROFILE_INCARNATION_ID, + ownerClientInstanceId: 'terminal-client', + }, + { + credentials: credentials.store, + loadClientInstanceId: async () => 'provider-client', + connectProfile: async () => initial.connection, + createPeerClient: () => + ({ close: async () => void (peerCloses += 1) }) as RuntimeHostPeerClient, + createReconnectingConnection: async (input) => { + fatal = input.onFatalError; + return reconnectingConnection(initial.connection); + }, + profiles: profiles.catalog, + subscribeProfileChanges: profiles.subscribe, + }, + ); + const latest = await availability(target); + await waitFor(() => latest().kind === 'connected', 'direct provider companion to connect'); + + fatal?.(new RuntimeHostProfileConnectionError('credential_rejected', 'revoked')); + + await waitFor(() => peerCloses === 1, 'direct peer endpoint to close'); + assert.deepEqual(latest(), { kind: 'unavailable', reason: 'credential_rejected' }); + await target.closePublication?.(); + assert.equal(peerCloses, 1); +}); + +async function availability(target: ReturnType) { + let current: Parameters[0]>[0] = { + kind: 'unavailable', + }; + target.subscribeConnectionAvailability((next) => { + current = next; + }); + await new Promise((resolve) => setImmediate(resolve)); + return () => current; +} + +function createRemoteTuiMcpPublicationTarget( + input: Parameters[0], + overrides: NonNullable[1]> = {}, +) { + return createProductionRemoteTuiMcpPublicationTarget(input, { + acquirePublicationLease: async () => ({ close: async () => undefined }), + ...overrides, + }); +} + +function publicationLeaseHarness() { + const active = new Set(); + return { + active, + acquire: async (path: string) => { + if (active.has(path)) return undefined; + active.add(path); + let closed = false; + return { + close: async () => { + if (closed) return; + closed = true; + active.delete(path); + }, + }; + }, + }; +} + +function credentialHarness(initial?: string) { + const values = new Map(); + if (initial) values.set('office\0incarnation-a\0terminal-client', initial); + const key = (target: RuntimeHostRemoteProfileIncarnation, ownerClientInstanceId: string) => + `${target.profile.id}\0${target.profileIncarnationId}\0${ownerClientInstanceId}`; + const store: RuntimeHostCapabilityProviderCredentialStore = { + get: async (target, ownerClientInstanceId) => + values.get(key(target, ownerClientInstanceId)) ?? null, + set: async (target, ownerClientInstanceId, credential) => { + values.set(key(target, ownerClientInstanceId), credential); + }, + delete: async (target, ownerClientInstanceId) => { + values.delete(key(target, ownerClientInstanceId)); + }, + }; + return { store, values }; +} + +function profileDeps(profile: RemoteRuntimeHostProfile = PROFILE) { + const profiles = profileHarness(profile); + return { profiles: profiles.catalog, subscribeProfileChanges: profiles.subscribe }; +} + +function profileHarness(initial: RemoteRuntimeHostProfile = PROFILE) { + let current: + | { readonly profile: RemoteRuntimeHostProfile; readonly profileIncarnationId: string } + | undefined = { profile: initial, profileIncarnationId: PROFILE_INCARNATION_ID }; + const listeners = new Set<(error?: Error) => void>(); + let heldValidation: + | { + readonly started: ReturnType; + readonly release: ReturnType; + } + | undefined; + let heldMutation: + | { + readonly started: ReturnType; + readonly release: ReturnType; + } + | undefined; + const catalog: Pick< + RuntimeHostProfileCatalog, + 'readRemoteProfileIfCurrent' | 'mutateRemoteProfileIfCurrent' + > = { + readRemoteProfileIfCurrent: async (expected) => { + const snapshot = current; + const held = heldValidation; + heldValidation = undefined; + if (held) { + held.started.resolve(); + await held.release.promise; + } + return snapshot !== undefined && + snapshot.profileIncarnationId === expected.profileIncarnationId && + sameRemoteRuntimeHostProfileTarget(snapshot.profile, expected.profile) + ? snapshot.profile + : undefined; + }, + mutateRemoteProfileIfCurrent: async (expected, mutation) => { + const held = heldMutation; + heldMutation = undefined; + if (held) { + held.started.resolve(); + await held.release.promise; + } + if ( + !current || + current.profileIncarnationId !== expected.profileIncarnationId || + !sameRemoteRuntimeHostProfileTarget(current.profile, expected.profile) + ) { + return false; + } + await mutation(current.profile); + return true; + }, + }; + const invalidate = () => { + for (const listener of listeners) listener(); + }; + return { + catalog, + subscribe: (listener: (error?: Error) => void) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + invalidate, + remove: () => { + current = undefined; + invalidate(); + }, + recreate: (profileIncarnationId: string) => { + current = { profile: initial, profileIncarnationId }; + invalidate(); + }, + update: (profile: RemoteRuntimeHostProfile) => { + assert.ok(current); + current = { profile, profileIncarnationId: current.profileIncarnationId }; + invalidate(); + }, + holdNextValidation: () => { + const started = deferred(); + const release = deferred(); + heldValidation = { started, release }; + return { started: started.promise, release: release.resolve }; + }, + holdNextMutation: () => { + const started = deferred(); + const release = deferred(); + heldMutation = { started, release }; + return { started: started.promise, release: release.resolve }; + }, + }; +} + +interface ConnectionHarness { + connection: RuntimeHostConnection; + replacements: number; + unregisters: number; + closes: number; +} + +function connectionHarness( + connectionId: string, + beforeClose: () => Promise = async () => undefined, +): ConnectionHarness { + let resolveClosed!: () => void; + const closed = new Promise((resolve) => { + resolveClosed = resolve; + }); + const harness: ConnectionHarness = { + connection: undefined as unknown as RuntimeHostConnection, + replacements: 0, + unregisters: 0, + closes: 0, + }; + harness.connection = { + rootId: PROFILE.rootId, + hostEpoch: 'host-epoch', + connectionId, + selectedProtocol: 0, + compositionId: 'maka.interactive', + compositionRevision: 'composition-revision', + closed, + replaceClientCapabilities: async () => { + harness.replacements += 1; + return { registrationId: 'registration-a', revision: harness.replacements }; + }, + unregisterClientCapabilities: async () => { + harness.unregisters += 1; + return { registrationId: 'registration-a', revision: harness.unregisters }; + }, + subscribeConfigurationChanges: () => () => undefined, + subscribeProjectCatalogChanges: () => () => undefined, + subscribeSessionCatalogChanges: () => () => undefined, + subscribeScheduledTaskChanges: () => () => undefined, + close: async () => { + harness.closes += 1; + await beforeClose(); + resolveClosed(); + }, + } as unknown as RuntimeHostConnection; + return harness; +} + +function deferred() { + let resolve!: () => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + +function reconnectingConnection(connection: RuntimeHostConnection) { + return { + ...connection, + reconnecting: true as const, + subscribeConnectionAvailability: ( + listener: (availability: { + kind: 'connected'; + hostEpoch: string; + connectionId: string; + }) => void, + ) => { + listener({ + kind: 'connected', + hostEpoch: connection.hostEpoch, + connectionId: connection.connectionId, + }); + return () => undefined; + }, + }; +} diff --git a/packages/cli/src/pi-tui-mcp-status.ts b/packages/cli/src/pi-tui-mcp-status.ts index 8f9f03b9b1..5777570ad2 100644 --- a/packages/cli/src/pi-tui-mcp-status.ts +++ b/packages/cli/src/pi-tui-mcp-status.ts @@ -19,6 +19,7 @@ import { Editor, + Input, Key, matchesKey, truncateToWidth, @@ -52,6 +53,7 @@ interface TuiMcpStatusCopy { readonly back: string; readonly readOnly: string; readonly manage: string; + readonly managePublication: string; }; readonly unavailableTitle: string; readonly unavailableDetail: string; @@ -71,6 +73,14 @@ const MCP_STATUS_COPY = resolveUiMessageCatalog( defineUiMessageCatalog()(TUI_COPY_RESOURCES['mcp-status']), ); +interface OverlayTextInput extends Component { + focused: boolean; + onSubmit?: (value: string) => void; + onChange?: (value: string) => void; + setText(value: string): void; + handleInput(data: string): void; +} + type GuidedDraft = { serverId: string; transport?: 'stdio' | 'remote'; @@ -92,7 +102,8 @@ type InputKind = | 'env' | 'headers' | 'edit' - | 'import'; + | 'import' + | 'publication_credential'; type McpOverlayPhase = | { kind: 'list' } @@ -109,10 +120,11 @@ type McpOverlayPhase = | { kind: 'confirm_add'; draft: GuidedDraft } | { kind: 'confirm_import'; preview: TuiMcpImportPreview } | { kind: 'confirm_remove'; serverId: string } + | { kind: 'confirm_remove_publication_credential' } | { kind: 'busy'; label: string }; /** One in-frame state machine for status, editing, confirmation, and errors. - * Raw config values live only in the editor and are cleared on every exit; + * Raw config values live only in the input component and are cleared on every exit; * no management result is written into the conversation transcript. */ export class McpManagementOverlay implements Component { private top = 0; @@ -123,7 +135,7 @@ export class McpManagementOverlay implements Component { private phase: McpOverlayPhase = { kind: 'list' }; private notice: { level: 'info' | 'error'; text: string } | undefined; private readonly dispose: () => void; - private editor: Editor | undefined; + private editor: OverlayTextInput | undefined; private closed = false; private actionAttempt = 0; @@ -187,6 +199,11 @@ export class McpManagementOverlay implements Component { void this.runAction({ kind: 'commit_import', previewId: this.phase.preview.previewId }); } else if (this.phase.kind === 'confirm_remove' && matchesKey(data, 'y')) { void this.runAction({ kind: 'remove', serverId: this.phase.serverId }); + } else if ( + this.phase.kind === 'confirm_remove_publication_credential' && + matchesKey(data, 'y') + ) { + void this.runAction({ kind: 'remove_publication_credential' }); } } @@ -222,7 +239,8 @@ export class McpManagementOverlay implements Component { } private handleListInput(data: string): void { - const servers = this.input.surface?.snapshot().servers ?? []; + const snapshot = this.input.surface?.snapshot(); + const servers = snapshot?.servers ?? []; if (matchesKey(data, Key.up)) { this.selected = clamp(this.selected - 1, 0, servers.length - 1); } else if (matchesKey(data, Key.down)) { @@ -234,7 +252,19 @@ export class McpManagementOverlay implements Component { } else if (matchesKey(data, Key.home)) this.selected = 0; else if (matchesKey(data, Key.end)) this.selected = Math.max(0, servers.length - 1); else if (matchesKey(data, 'a') && this.management()) this.phase = { kind: 'add_choice' }; - else { + else if ( + matchesKey(data, 'p') && + this.management() && + snapshot?.canManagePublicationCredential + ) { + this.startInput('publication_credential'); + } else if ( + matchesKey(data, 'x') && + this.management() && + snapshot?.canManagePublicationCredential + ) { + this.phase = { kind: 'confirm_remove_publication_credential' }; + } else { const server = servers[this.selected]; if (!server || !this.management()) return; if (matchesKey(data, Key.enter)) this.startEdit(server.serverId); @@ -311,7 +341,10 @@ export class McpManagementOverlay implements Component { this.clearEditor(); this.notice = undefined; this.phase = { kind: 'input', input, draft, serverId, revision }; - this.editor = new Editor(this.input.tui, editorTheme(), { paddingX: 0 }); + this.editor = + input === 'publication_credential' + ? new MaskedTextInput() + : new Editor(this.input.tui, editorTheme(), { paddingX: 0 }); this.editor.onSubmit = (submitted) => this.submitInput(submitted); this.editor.setText(value); this.editor.focused = true; @@ -367,6 +400,10 @@ export class McpManagementOverlay implements Component { expectedRevision: phase.revision, config, }); + } else if (phase.input === 'publication_credential') { + if (!trimmed) throw new Error(); + this.clearEditor(); + void this.runAction({ kind: 'set_publication_credential', credential: trimmed }); } else { const preview = this.management()?.previewImport(value); if (!preview || preview.status !== 'ready') throw new Error(); @@ -454,6 +491,21 @@ export class McpManagementOverlay implements Component { confirmCopy(this.input.locale), ]; } + if (this.phase.kind === 'confirm_remove_publication_credential') { + return [ + heading( + this.input.locale, + 'Remove the remote provider credential?', + '删除远程 Provider 凭据？', + ), + '', + this.input.locale === 'zh' + ? '这会停止向所选 Runtime Host 发布 MCP 工具。' + : 'This stops publishing MCP tools to the selected Runtime Host.', + '', + confirmCopy(this.input.locale), + ]; + } if (this.phase.kind === 'busy') return [ansi.yellow(this.phase.label)]; const lines = [publicationLine(snapshot, this.input.locale)]; if (snapshot.configuration !== 'ready') { @@ -497,7 +549,9 @@ export class McpManagementOverlay implements Component { const copy = MCP_STATUS_COPY[this.input.locale].footer; if (this.phase.kind !== 'list') return copy.back; if (!this.management()) return copy.readOnly; - return copy.manage; + return this.input.surface?.snapshot().canManagePublicationCredential + ? copy.managePublication + : copy.manage; } private backToList(clearNotice = true): void { @@ -565,6 +619,47 @@ export class McpManagementOverlay implements Component { } } +class MaskedTextInput implements OverlayTextInput { + readonly #input = new Input(); + onSubmit?: (value: string) => void; + onChange?: (value: string) => void; + + constructor() { + this.#input.onSubmit = (value) => this.onSubmit?.(value); + } + + get focused(): boolean { + return this.#input.focused; + } + + set focused(value: boolean) { + this.#input.focused = value; + } + + setText(value: string): void { + this.#input.setValue(value); + } + + handleInput(data: string): void { + this.#input.handleInput(data); + this.onChange?.(this.#input.getValue()); + } + + invalidate(): void { + this.#input.invalidate(); + } + + render(width: number): string[] { + const value = this.#input.getValue(); + this.#input.setValue('•'.repeat(value.length)); + try { + return this.#input.render(width); + } finally { + this.#input.setValue(value); + } + } +} + function normalizeOneServer(serverId: string, source: string): McpServerConfig { const value: unknown = JSON.parse(source); return normalizeMcpConfig({ version: 3, mcpServers: { [serverId]: value } }).mcpServers[serverId]; @@ -669,6 +764,7 @@ function copy(locale: UiLocale, key: string): string { 'invalid-config': 'The server configuration is invalid.', 'credential-cleanup-failed': 'Stored credentials could not be removed; the configuration was not changed.', + 'publication-credential-failed': 'The provider credential could not be stored or applied.', 'persist-failed': 'The configuration could not be saved.', 'manager-failed': 'The MCP connection action failed.', turn_active: 'MCP cannot be changed while a turn or another control action is running.', @@ -691,6 +787,7 @@ function copy(locale: UiLocale, key: string): string { closed: 'MCP 控制器已关闭。', 'invalid-config': '服务器配置无效。', 'credential-cleanup-failed': '无法删除旧凭据，配置未修改。', + 'publication-credential-failed': '无法保存或应用 Provider 凭据。', 'persist-failed': '无法保存配置。', 'manager-failed': 'MCP 连接操作失败。', turn_active: 'Turn 或其他控制操作运行期间不能修改 MCP。', @@ -749,6 +846,7 @@ function inputLabel(kind: InputKind, locale: UiLocale): string { headers: ['Request headers', '请求头'], edit: ['Edit server JSON', '编辑服务器 JSON'], import: ['Paste MCP JSON', '粘贴 MCP JSON'], + publication_credential: ['Provider credential', 'Provider 凭据'], }; return labels[kind][locale === 'zh' ? 1 : 0]; } @@ -758,6 +856,11 @@ function inputHint(kind: InputKind, locale: UiLocale): string { if (kind === 'args') return `JSON string array, ${optional}`; if (kind === 'env' || kind === 'headers') return `JSON string map, ${optional}`; if (kind === 'cwd') return optional; + if (kind === 'publication_credential') { + return locale === 'zh' + ? '仅保存到本机凭据存储；不会写入 profile、参数或对话 · Enter 提交 · Esc 返回' + : 'Stored only in the local credential store; never written to profiles, arguments, or chat · Enter submit · Esc back'; + } return locale === 'zh' ? 'Enter 提交 · Esc 返回' : 'Enter submit · Esc back'; } diff --git a/packages/cli/src/runtime-host-cli-context.ts b/packages/cli/src/runtime-host-cli-context.ts index 41c64c5cb9..535752c3df 100644 --- a/packages/cli/src/runtime-host-cli-context.ts +++ b/packages/cli/src/runtime-host-cli-context.ts @@ -90,6 +90,12 @@ export interface RuntimeHostCliConnectionContext { close(): Promise; } +export interface RuntimeHostCliConnectionContextWithIdentity + extends RuntimeHostCliConnectionContext { + readonly clientInstanceId: string; + readonly profileIncarnationId?: string; +} + export interface RuntimeHostCliTarget { readonly connection: ConnectionCatalogEntry; readonly model: string; @@ -114,7 +120,7 @@ export async function connectRuntimeHostCli( readonly interactiveSsh?: boolean; }, overrides: Partial = {}, -): Promise { +): Promise { const deps: RuntimeHostCliContextDeps = { connectOrSpawn: connectOrSpawnRuntimeHost, connectProfile: connectRuntimeHostProfile, @@ -201,6 +207,10 @@ export async function connectRuntimeHostCli( connection: liveConnection, catalog: await deps.readConnectionCatalog(liveConnection), profile, + clientInstanceId, + ...(resolvedProfile.profileIncarnationId + ? { profileIncarnationId: resolvedProfile.profileIncarnationId } + : {}), close: async () => { try { await liveConnection.close(); diff --git a/packages/cli/src/runtime-host-tui-context.ts b/packages/cli/src/runtime-host-tui-context.ts index 1247ebc169..4aad47ee27 100644 --- a/packages/cli/src/runtime-host-tui-context.ts +++ b/packages/cli/src/runtime-host-tui-context.ts @@ -68,6 +68,7 @@ import { type TuiMcpController, type TuiMcpManagement, } from './tui-mcp-control.js'; +import { createRemoteTuiMcpPublicationTarget } from './tui-mcp-remote-publication.js'; export interface RuntimeHostTuiContext { readonly connection: RuntimeHostConnection; @@ -159,7 +160,7 @@ export async function createRuntimeHostTuiContext( }; const driver = createRuntimeHostMakaSessionDriver(driverInput); await driver.recoverSideConversations(); - if (!runtimeHostProfileUsesHostWorkspace(connected.profile.kind)) { + if (connected.profile.kind === 'local') { if (!isRuntimeHostReconnectingConnection(connection)) { throw new Error('Local Runtime Host TUI connection is not reconnectable'); } @@ -167,6 +168,19 @@ export async function createRuntimeHostTuiContext( workspaceRoot: input.rootPath, connection, }); + } else if (connected.profile.kind === 'remote') { + if (!connected.profileIncarnationId) { + throw new Error('Remote Runtime Host profile incarnation is unavailable'); + } + mcp = createTuiMcpController({ + workspaceRoot: input.rootPath, + connection: createRemoteTuiMcpPublicationTarget({ + clientDataRoot: input.clientDataRoot, + profile: connected.profile, + profileIncarnationId: connected.profileIncarnationId, + ownerClientInstanceId: connected.clientInstanceId, + }), + }); } const modelContextWindow = selectedTarget.connection?.models.find( (model) => model.id === selectedTarget.model, diff --git a/packages/cli/src/tui-copy-catalog.ts b/packages/cli/src/tui-copy-catalog.ts index 7576f96b38..6f1302b8a5 100644 --- a/packages/cli/src/tui-copy-catalog.ts +++ b/packages/cli/src/tui-copy-catalog.ts @@ -26,6 +26,8 @@ export const TUI_COPY_RESOURCES = { readOnly: '↑/↓ scroll · q/Esc close', manage: 'a Add · Enter Edit · Space Enable/disable · t Test · r Reconnect · d Remove · Esc Close', + managePublication: + 'a Add · Enter Edit · Space Enable/disable · t Test · r Reconnect · d Remove · p Set provider credential · x Remove credential · Esc Close', }, unavailableTitle: 'This TUI is not connected to a local MCP control plane.', unavailableDetail: @@ -37,6 +39,10 @@ export const TUI_COPY_RESOURCES = { publication: { waiting: 'waiting to publish', host_unavailable: 'Runtime Host reconnecting', + credential_required: 'provider credential required', + credential_rejected: 'provider credential rejected', + provider_conflict: 'provider active in another TUI', + target_mismatch: 'provider target mismatch', publishing: 'publishing', published: 'published', not_published: 'not published', @@ -60,6 +66,8 @@ export const TUI_COPY_RESOURCES = { back: 'Esc 返回', readOnly: '↑/↓ 滚动 · q/Esc 关闭', manage: 'a 添加 · Enter 编辑 · Space 启用/停用 · t 测试 · r 重连 · d 删除 · Esc 关闭', + managePublication: + 'a 添加 · Enter 编辑 · Space 启用/停用 · t 测试 · r 重连 · d 删除 · p 设置 Provider 凭据 · x 删除凭据 · Esc 关闭', }, unavailableTitle: '当前 TUI 未连接本地 MCP 控制面。', unavailableDetail: '远程 Runtime Host 的客户端 MCP 工具关联将在后续版本提供。', @@ -69,6 +77,10 @@ export const TUI_COPY_RESOURCES = { publication: { waiting: '等待发布', host_unavailable: 'Runtime Host 重连中', + credential_required: '需要 Provider 凭据', + credential_rejected: 'Provider 凭据已被拒绝', + provider_conflict: 'Provider 已在另一个 TUI 中运行', + target_mismatch: 'Provider 目标不匹配', publishing: '正在发布', published: '已发布', not_published: '未发布', diff --git a/packages/cli/src/tui-mcp-control.ts b/packages/cli/src/tui-mcp-control.ts index dfaab64064..e0df998355 100644 --- a/packages/cli/src/tui-mcp-control.ts +++ b/packages/cli/src/tui-mcp-control.ts @@ -51,6 +51,10 @@ const RUNTIME_HOST_CREDENTIAL_ENV = 'MAKA_RUNTIME_HOST_ACCESS_CREDENTIAL'; export type TuiMcpPublicationState = | 'waiting' | 'host_unavailable' + | 'credential_required' + | 'credential_rejected' + | 'provider_conflict' + | 'target_mismatch' | 'publishing' | 'published' | 'not_published' @@ -74,6 +78,7 @@ export interface TuiMcpSnapshot { readonly initialization: 'loading' | 'ready' | 'error'; readonly configuration: 'ready' | 'synchronizing' | 'out_of_sync'; readonly publication: TuiMcpPublicationState; + readonly canManagePublicationCredential?: boolean; readonly toolCount: number; readonly servers: readonly TuiMcpServerSnapshot[]; } @@ -119,7 +124,9 @@ export type TuiMcpAction = | { readonly kind: 'set_enabled'; readonly serverId: string; readonly enabled: boolean } | { readonly kind: 'remove'; readonly serverId: string } | { readonly kind: 'test'; readonly serverId: string } - | { readonly kind: 'reconnect'; readonly serverId: string }; + | { readonly kind: 'reconnect'; readonly serverId: string } + | { readonly kind: 'set_publication_credential'; readonly credential: string } + | { readonly kind: 'remove_publication_credential' }; export type TuiMcpActionEffect = | 'published' @@ -140,6 +147,7 @@ export type TuiMcpActionResult = | 'closed' | 'invalid-config' | 'credential-cleanup-failed' + | 'publication-credential-failed' | 'persist-failed' | 'manager-failed'; }; @@ -168,10 +176,32 @@ type TuiMcpManager = Pick< | 'close' >; -type TuiMcpConnection = Pick< - RuntimeHostReconnectingConnection, - 'replaceClientCapabilities' | 'unregisterClientCapabilities' | 'subscribeConnectionAvailability' ->; +export type TuiMcpPublicationUnavailableReason = + | 'host_unavailable' + | 'credential_required' + | 'credential_rejected' + | 'provider_conflict' + | 'target_mismatch'; + +export type TuiMcpPublicationAvailability = + | { + readonly kind: 'unavailable'; + readonly reason?: TuiMcpPublicationUnavailableReason; + } + | Extract; + +export interface TuiMcpPublicationTarget + extends Pick< + RuntimeHostReconnectingConnection, + 'replaceClientCapabilities' | 'unregisterClientCapabilities' + > { + subscribeConnectionAvailability( + listener: (availability: TuiMcpPublicationAvailability) => void, + ): () => void; + setCredential?(credential: string): Promise; + removeCredential?(): Promise; + closePublication?(): Promise; +} interface TuiMcpControllerDeps { readonly configStore: Pick; @@ -182,7 +212,7 @@ interface TuiMcpControllerDeps { export function createTuiMcpController( input: { readonly workspaceRoot: string; - readonly connection: TuiMcpConnection; + readonly connection: TuiMcpPublicationTarget; }, overrides: Partial = {}, ): TuiMcpController { @@ -201,13 +231,13 @@ export function createTuiMcpController( } class TuiMcpControllerImpl implements TuiMcpController { - readonly #connection: TuiMcpConnection; + readonly #connection: TuiMcpPublicationTarget; readonly #deps: TuiMcpControllerDeps; readonly #listeners = new Set<() => void>(); readonly #disposeManagerChange: () => void; readonly #disposeConnectionAvailability: () => void; readonly #initialization: Promise; - #availability: RuntimeHostConnectionAvailability = { kind: 'unavailable' }; + #availability: TuiMcpPublicationAvailability = { kind: 'unavailable' }; #closed = false; #config: McpConfigFile | undefined; #preparedImport: @@ -232,13 +262,20 @@ class TuiMcpControllerImpl implements TuiMcpController { initialization: 'loading', configuration: 'synchronizing', publication: 'waiting', + canManagePublicationCredential: false, toolCount: 0, servers: [], }); - constructor(connection: TuiMcpConnection, deps: TuiMcpControllerDeps) { + constructor(connection: TuiMcpPublicationTarget, deps: TuiMcpControllerDeps) { this.#connection = connection; this.#deps = deps; + this.#snapshot = freezeSnapshot({ + ...this.#snapshot, + canManagePublicationCredential: Boolean( + connection.setCredential && connection.removeCredential, + ), + }); this.#disposeManagerChange = deps.manager.onChange(() => { try { this.#refreshManagerSnapshot(); @@ -254,7 +291,12 @@ class TuiMcpControllerImpl implements TuiMcpController { this.#availability = availability; if (availability.kind === 'unavailable') { this.#published = undefined; - this.#updateSnapshot({ publication: 'host_unavailable' }); + this.#updateSnapshot({ + publication: availability.reason ?? 'host_unavailable', + ...(availability.reason === 'provider_conflict' + ? { canManagePublicationCredential: false } + : {}), + }); } else { this.#updateSnapshot({ publication: 'waiting' }); if (this.#snapshot.initialization === 'ready') this.#requestPublication(); @@ -338,6 +380,7 @@ class TuiMcpControllerImpl implements TuiMcpController { await this.#connection.unregisterClientCapabilities().catch(() => undefined); } this.#published = undefined; + await this.#connection.closePublication?.().catch(() => undefined); await managerClosing; await this.#initialization.catch(() => undefined); } @@ -368,6 +411,28 @@ class TuiMcpControllerImpl implements TuiMcpController { async #executeAction(action: TuiMcpAction): Promise { if (this.#closed) return { status: 'failed', reason: 'closed' }; + if (action.kind === 'set_publication_credential') { + if (!this.#connection.setCredential) { + return { status: 'failed', reason: 'publication-credential-failed' }; + } + try { + await this.#connection.setCredential(action.credential); + return { status: 'applied', effect: await this.#settlePublication() }; + } catch { + return { status: 'failed', reason: 'publication-credential-failed' }; + } + } + if (action.kind === 'remove_publication_credential') { + if (!this.#connection.removeCredential) { + return { status: 'failed', reason: 'publication-credential-failed' }; + } + try { + await this.#connection.removeCredential(); + return { status: 'applied', effect: 'pending_host' }; + } catch { + return { status: 'failed', reason: 'publication-credential-failed' }; + } + } if (action.kind === 'test') { try { const test = await this.#deps.manager.test(action.serverId); @@ -391,7 +456,11 @@ class TuiMcpControllerImpl implements TuiMcpController { } async #commitMutation( - action: Exclude, + action: Exclude< + TuiMcpAction, + | { kind: 'test' | 'reconnect' } + | { kind: 'set_publication_credential' | 'remove_publication_credential' } + >, ): Promise { let committed: McpConfigFile; try { @@ -452,7 +521,11 @@ class TuiMcpControllerImpl implements TuiMcpController { #prepareMutation( current: McpConfigFile, - action: Exclude, + action: Exclude< + TuiMcpAction, + | { kind: 'test' | 'reconnect' } + | { kind: 'set_publication_credential' | 'remove_publication_credential' } + >, ): | { readonly next: McpConfigFile } | Extract { @@ -502,8 +575,20 @@ class TuiMcpControllerImpl implements TuiMcpController { while (!this.#closed && (this.#publicationTask || this.#publicationRequested)) { await this.#publicationTask?.catch(() => undefined); } - if (this.#snapshot.publication === 'error') return 'publication_failed'; - if (this.#snapshot.publication === 'host_unavailable') return 'pending_host'; + if ( + this.#snapshot.publication === 'error' || + this.#snapshot.publication === 'credential_rejected' || + this.#snapshot.publication === 'provider_conflict' || + this.#snapshot.publication === 'target_mismatch' + ) { + return 'publication_failed'; + } + if ( + this.#snapshot.publication === 'host_unavailable' || + this.#snapshot.publication === 'credential_required' + ) { + return 'pending_host'; + } return 'published'; } @@ -521,6 +606,7 @@ class TuiMcpControllerImpl implements TuiMcpController { initialization, configuration, publication: this.#snapshot.publication, + canManagePublicationCredential: this.#snapshot.canManagePublicationCredential, toolCount: this.#deps.manager.toolSnapshot().tools.length, servers: [...serverIds] .sort((left, right) => left.localeCompare(right)) @@ -560,7 +646,7 @@ class TuiMcpControllerImpl implements TuiMcpController { async #publishCurrentSnapshot(): Promise { const availability = this.#availability; if (availability.kind !== 'connected') { - this.#updateSnapshot({ publication: 'host_unavailable' }); + this.#updateSnapshot({ publication: availability.reason ?? 'host_unavailable' }); return; } const identity = connectionIdentity(availability); diff --git a/packages/cli/src/tui-mcp-remote-publication.ts b/packages/cli/src/tui-mcp-remote-publication.ts new file mode 100644 index 0000000000..b0dec0b88e --- /dev/null +++ b/packages/cli/src/tui-mcp-remote-publication.ts @@ -0,0 +1,538 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { createHash } from 'node:crypto'; +import { join } from 'node:path'; +import { + connectRuntimeHostProfile, + createClientRuntimeHostCredentialStore, + createClientRuntimeHostProfileCatalog, + createRuntimeHostCapabilityProviderCredentialStore, + createRuntimeHostPeerClientFromEnvironment, + createRuntimeHostReconnectingConnection, + loadOrCreateRuntimeHostClientInstanceId, + RuntimeHostPermanentReconnectError, + RuntimeHostProfileConnectionError, + subscribeClientRuntimeHostProfileCatalogChanges, + RuntimeHostRemoteCompatibilityError, + runtimeHostProfileTargetFingerprint, + type RemoteRuntimeHostProfile, + type RuntimeHostCapabilityProviderCredentialStore, + type RuntimeHostConnection, + type RuntimeHostPeerClient, + type RuntimeHostProfileCatalog, + type RuntimeHostRemoteProfileIncarnation, + type RuntimeHostReconnectingConnection, +} from '@maka/runtime-host/client'; +import type { ClientCapabilityProvider } from '@maka/runtime-host/client'; +import { + tryAcquireFileLifetimeOwner, + type FileLifetimeOwner, +} from '@maka/storage/file-lifetime-owner'; +import type { + TuiMcpPublicationAvailability, + TuiMcpPublicationTarget, + TuiMcpPublicationUnavailableReason, +} from './tui-mcp-control.js'; + +interface RemoteTuiMcpPublicationDeps { + readonly credentials: RuntimeHostCapabilityProviderCredentialStore; + readonly loadClientInstanceId: typeof loadOrCreateRuntimeHostClientInstanceId; + readonly connectProfile: typeof connectRuntimeHostProfile; + readonly createPeerClient: typeof createRuntimeHostPeerClientFromEnvironment; + readonly createReconnectingConnection: typeof createRuntimeHostReconnectingConnection; + readonly acquirePublicationLease: typeof tryAcquireFileLifetimeOwner; + readonly profiles: Pick< + RuntimeHostProfileCatalog, + 'readRemoteProfileIfCurrent' | 'mutateRemoteProfileIfCurrent' + >; + readonly subscribeProfileChanges: (listener: (error?: Error) => void) => () => void; +} + +export function createRemoteTuiMcpPublicationTarget( + input: { + readonly clientDataRoot: string; + readonly profile: RemoteRuntimeHostProfile; + readonly profileIncarnationId: string; + readonly ownerClientInstanceId: string; + }, + overrides: Partial = {}, +): TuiMcpPublicationTarget { + const deps: RemoteTuiMcpPublicationDeps = { + credentials: createRuntimeHostCapabilityProviderCredentialStore( + createClientRuntimeHostCredentialStore(input.clientDataRoot), + ), + loadClientInstanceId: loadOrCreateRuntimeHostClientInstanceId, + connectProfile: connectRuntimeHostProfile, + createPeerClient: createRuntimeHostPeerClientFromEnvironment, + createReconnectingConnection: createRuntimeHostReconnectingConnection, + acquirePublicationLease: tryAcquireFileLifetimeOwner, + profiles: createClientRuntimeHostProfileCatalog(input.clientDataRoot), + subscribeProfileChanges: (listener) => + subscribeClientRuntimeHostProfileCatalogChanges(input.clientDataRoot, listener), + ...overrides, + }; + return new RemoteTuiMcpPublicationTarget(input, deps); +} + +class RemoteTuiMcpPublicationTarget implements TuiMcpPublicationTarget { + readonly #input: { + readonly clientDataRoot: string; + readonly profile: RemoteRuntimeHostProfile; + readonly profileIncarnationId: string; + readonly ownerClientInstanceId: string; + }; + readonly #deps: RemoteTuiMcpPublicationDeps; + readonly #listeners = new Set<(availability: TuiMcpPublicationAvailability) => void>(); + #availability: TuiMcpPublicationAvailability = { + kind: 'unavailable', + reason: 'host_unavailable', + }; + #connection: RuntimeHostReconnectingConnection | undefined; + #publicationLease: FileLifetimeOwner | undefined; + #disposeAvailability: (() => void) | undefined; + #peerClient: RuntimeHostPeerClient | undefined; + #peerCloseTask = Promise.resolve(); + #operation = Promise.resolve(); + #connectAbort: AbortController | undefined; + #generation = 0; + #closed = false; + #closeTask: Promise | undefined; + #disposeProfileChanges: (() => void) | undefined; + #profileValidationQueued = false; + #profileInvalidationGeneration = 0; + #profileValidationError: Error | undefined; + + constructor( + input: { + readonly clientDataRoot: string; + readonly profile: RemoteRuntimeHostProfile; + readonly profileIncarnationId: string; + readonly ownerClientInstanceId: string; + }, + deps: RemoteTuiMcpPublicationDeps, + ) { + this.#input = input; + this.#deps = deps; + try { + this.#disposeProfileChanges = deps.subscribeProfileChanges((error) => { + this.#scheduleProfileValidation(error); + }); + } catch (error) { + this.#scheduleProfileValidation(error instanceof Error ? error : new Error(String(error))); + } + void this.#serialize(async () => { + let lease: FileLifetimeOwner | undefined; + try { + lease = await deps.acquirePublicationLease(providerLeasePath(input)); + } catch { + await this.#retire('host_unavailable'); + return; + } + if (!lease) { + await this.#retire('provider_conflict'); + return; + } + if (this.#closed) { + await lease.close(); + return; + } + this.#publicationLease = lease; + const profileCurrent = await this.#profileStillCurrent().catch(() => undefined); + if (profileCurrent !== true) { + await this.#retire(profileCurrent === false ? 'target_mismatch' : 'host_unavailable'); + return; + } + const credential = await deps.credentials.get( + this.#profileTarget(), + input.ownerClientInstanceId, + ); + if (this.#closed) return; + if (!credential) { + this.#setUnavailable('credential_required'); + return; + } + await this.#connect(credential); + }).catch(() => { + if (!this.#closed) this.#setUnavailable('host_unavailable'); + }); + } + + replaceClientCapabilities(provider: ClientCapabilityProvider, timeoutMs?: number) { + return this.#serialize(async () => { + if (this.#closed) throw new Error('Remote MCP publication is closed'); + let result: + | Awaited> + | undefined; + const committed = await this.#deps.profiles.mutateRemoteProfileIfCurrent( + this.#profileTarget(), + async () => { + result = await this.#requireConnection().replaceClientCapabilities(provider, timeoutMs); + }, + ); + if (!committed) { + await this.#retire('target_mismatch'); + throw new RuntimeHostProfileConnectionError( + 'target_mismatch', + 'Remote MCP publication profile is no longer current', + ); + } + if (!result) throw new Error('Runtime Host did not confirm MCP capability registration'); + return result; + }); + } + + unregisterClientCapabilities(timeoutMs?: number) { + return this.#requireConnection().unregisterClientCapabilities(timeoutMs); + } + + subscribeConnectionAvailability( + listener: (availability: TuiMcpPublicationAvailability) => void, + ): () => void { + this.#listeners.add(listener); + try { + listener(this.#availability); + } catch { + // Presentation cannot invalidate the companion lifecycle. + } + return () => this.#listeners.delete(listener); + } + + setCredential(credential: string): Promise { + this.#cancelConnect(); + return this.#serialize(async () => { + if (this.#closed) throw new Error('Remote MCP publication is closed'); + const committed = await this.#deps.profiles.mutateRemoteProfileIfCurrent( + this.#profileTarget(), + (profile) => + this.#deps.credentials.set( + { profile, profileIncarnationId: this.#input.profileIncarnationId }, + this.#input.ownerClientInstanceId, + credential, + ), + ); + if (!committed) { + await this.#retire('target_mismatch'); + throw new RuntimeHostProfileConnectionError( + 'target_mismatch', + 'Remote MCP publication profile is no longer current', + ); + } + await this.#disconnect(); + await this.#connect(credential); + }); + } + + removeCredential(): Promise { + this.#cancelConnect(); + return this.#serialize(async () => { + if (this.#closed) throw new Error('Remote MCP publication is closed'); + await this.#disconnect(); + const committed = await this.#deps.profiles.mutateRemoteProfileIfCurrent( + this.#profileTarget(), + (profile) => + this.#deps.credentials.delete( + { profile, profileIncarnationId: this.#input.profileIncarnationId }, + this.#input.ownerClientInstanceId, + ), + ); + if (!committed) { + await this.#retire('target_mismatch'); + throw new RuntimeHostProfileConnectionError( + 'target_mismatch', + 'Remote MCP publication profile is no longer current', + ); + } + this.#setUnavailable('credential_required'); + }); + } + + closePublication(): Promise { + this.#cancelConnect(); + return this.#beginClose({ waitForOperations: true }); + } + + #serialize(work: () => Promise): Promise { + const pending = this.#operation.then(work, work); + this.#operation = pending.then( + () => undefined, + () => undefined, + ); + return pending; + } + + async #connect(credential: string): Promise { + const generation = ++this.#generation; + const abort = new AbortController(); + this.#connectAbort = abort; + this.#setUnavailable('host_unavailable'); + try { + const clientInstanceId = await this.#deps.loadClientInstanceId( + providerIdentityPath(this.#input), + ); + const peerClient = + this.#input.profile.transport.kind === 'libp2p-direct' + ? this.#deps.createPeerClient() + : undefined; + this.#peerClient = peerClient; + const connect = async (signal?: AbortSignal): Promise => { + const profile = await this.#requireCurrentProfile(); + const connection = await this.#deps.connectProfile({ + profile, + credential, + clientInstanceId, + sshInteraction: 'batch', + ...(peerClient ? { peerClient } : {}), + signal: signal ? AbortSignal.any([abort.signal, signal]) : abort.signal, + }); + return this.#keepIfProfileCurrent(connection); + }; + const initial = await connect(); + if (this.#closed || generation !== this.#generation) { + await initial.close().catch(() => undefined); + await this.#closePeer(peerClient); + return; + } + const connection = await this.#keepIfProfileCurrent( + await this.#deps.createReconnectingConnection({ + initialConnection: initial, + connect, + onFatalError: (error) => { + if (!this.#closed && generation === this.#generation) { + this.#setUnavailable(classifyUnavailable(error)); + if (this.#peerClient === peerClient) { + void this.#closePeer(peerClient); + } + } + }, + }), + ); + if (this.#closed || generation !== this.#generation) { + await connection.close().catch(() => undefined); + await this.#closePeer(peerClient); + return; + } + this.#connection = connection; + this.#disposeAvailability = connection.subscribeConnectionAvailability((availability) => { + if (this.#closed || generation !== this.#generation) return; + this.#setAvailability( + availability.kind === 'connected' + ? availability + : { kind: 'unavailable', reason: 'host_unavailable' }, + ); + }); + } catch (error) { + if (!this.#closed && generation === this.#generation) { + this.#setUnavailable(classifyUnavailable(error)); + } + await this.#closePeer(this.#peerClient); + } finally { + if (this.#connectAbort === abort) this.#connectAbort = undefined; + } + } + + async #disconnect(): Promise { + this.#cancelConnect(); + this.#generation += 1; + this.#disposeAvailability?.(); + this.#disposeAvailability = undefined; + const connection = this.#connection; + this.#connection = undefined; + await connection?.close().catch(() => undefined); + await this.#closePeer(this.#peerClient); + await this.#peerCloseTask; + if (!this.#closed) this.#setUnavailable('host_unavailable'); + } + + #scheduleProfileValidation(error?: Error): void { + if (this.#closed) return; + this.#profileInvalidationGeneration += 1; + this.#profileValidationError ??= error; + if (this.#profileValidationQueued) return; + this.#profileValidationQueued = true; + void this.#serialize(async () => { + try { + while (!this.#closed) { + const generation = this.#profileInvalidationGeneration; + const profileCurrent = await this.#profileStillCurrent().catch(() => undefined); + const validationError = this.#profileValidationError; + this.#profileValidationError = undefined; + if (this.#closed) return; + if (validationError || profileCurrent !== true) { + await this.#retire( + validationError || profileCurrent === undefined + ? 'host_unavailable' + : 'target_mismatch', + ); + return; + } + if (generation === this.#profileInvalidationGeneration) return; + } + } finally { + this.#profileValidationQueued = false; + } + }); + } + + async #currentProfile(): Promise { + return this.#deps.profiles.readRemoteProfileIfCurrent(this.#profileTarget()); + } + + async #profileStillCurrent(): Promise { + return (await this.#currentProfile()) !== undefined; + } + + async #requireCurrentProfile(): Promise { + const profile = await this.#currentProfile(); + if (profile) return profile; + throw new RuntimeHostProfileConnectionError( + 'target_mismatch', + 'Remote MCP publication profile is no longer current', + ); + } + + async #keepIfProfileCurrent(connection: T): Promise { + try { + await this.#requireCurrentProfile(); + return connection; + } catch (error) { + await connection.close().catch(() => undefined); + throw error; + } + } + + #profileTarget(): RuntimeHostRemoteProfileIncarnation { + return { + profile: this.#input.profile, + profileIncarnationId: this.#input.profileIncarnationId, + }; + } + + async #retire(reason: TuiMcpPublicationUnavailableReason): Promise { + await this.#beginClose({ reason, waitForOperations: false }); + } + + #beginClose(input: { + readonly reason?: TuiMcpPublicationUnavailableReason; + readonly waitForOperations: boolean; + }): Promise { + if (this.#closeTask) return this.#closeTask; + this.#closed = true; + this.#disposeProfileChanges?.(); + this.#disposeProfileChanges = undefined; + if (input.reason) this.#setUnavailable(input.reason); + const operations = input.waitForOperations + ? this.#operation.catch(() => undefined) + : Promise.resolve(); + this.#closeTask = operations.then(async () => { + await this.#disconnect(); + const lease = this.#publicationLease; + this.#publicationLease = undefined; + try { + await lease?.close(); + } finally { + this.#listeners.clear(); + } + }); + return this.#closeTask; + } + + #requireConnection(): RuntimeHostReconnectingConnection { + if (this.#connection && this.#availability.kind === 'connected') return this.#connection; + throw new RuntimeHostPermanentReconnectError('Remote MCP publication is unavailable'); + } + + #cancelConnect(): void { + this.#connectAbort?.abort(new Error('Remote MCP publication target changed')); + } + + #closePeer(peerClient: RuntimeHostPeerClient | undefined): Promise { + if (!peerClient) return this.#peerCloseTask; + if (this.#peerClient === peerClient) this.#peerClient = undefined; + const closing = this.#peerCloseTask.then(() => peerClient.close()).catch(() => undefined); + this.#peerCloseTask = closing; + return closing; + } + + #setUnavailable(reason: TuiMcpPublicationUnavailableReason): void { + this.#setAvailability({ kind: 'unavailable', reason }); + } + + #setAvailability(availability: TuiMcpPublicationAvailability): void { + this.#availability = availability; + for (const listener of this.#listeners) { + try { + listener(availability); + } catch { + // Presentation cannot invalidate the companion lifecycle. + } + } + } +} + +function providerIdentityPath(input: { + readonly clientDataRoot: string; + readonly profile: RemoteRuntimeHostProfile; + readonly profileIncarnationId: string; + readonly ownerClientInstanceId: string; +}): string { + return join( + input.clientDataRoot, + 'runtime-host-client', + 'capability-provider-identities', + `${providerIdentity(input)}.json`, + ); +} + +function providerLeasePath(input: { + readonly clientDataRoot: string; + readonly profile: RemoteRuntimeHostProfile; + readonly profileIncarnationId: string; + readonly ownerClientInstanceId: string; +}): string { + return join( + input.clientDataRoot, + 'runtime-host-client', + 'capability-provider-leases', + `${providerIdentity(input)}.lease`, + ); +} + +function providerIdentity(input: { + readonly profile: RemoteRuntimeHostProfile; + readonly profileIncarnationId: string; + readonly ownerClientInstanceId: string; +}): string { + return createHash('sha256') + .update('tui-mcp-capability-provider') + .update('\0') + .update(runtimeHostProfileTargetFingerprint(input.profile)) + .update('\0') + .update(input.profileIncarnationId) + .update('\0') + .update(input.ownerClientInstanceId) + .digest('hex') + .slice(0, 24); +} + +function classifyUnavailable(error: unknown): TuiMcpPublicationUnavailableReason { + if (error instanceof RuntimeHostProfileConnectionError) return error.reason; + if (error instanceof RuntimeHostRemoteCompatibilityError) return 'target_mismatch'; + return 'host_unavailable'; +} diff --git a/packages/runtime-host/src/__tests__/host-profile.test.ts b/packages/runtime-host/src/__tests__/host-profile.test.ts index 23a6197b46..fcf5fd00fa 100644 --- a/packages/runtime-host/src/__tests__/host-profile.test.ts +++ b/packages/runtime-host/src/__tests__/host-profile.test.ts @@ -30,11 +30,15 @@ import { import { connectRemoteRuntimeHostProfile, createFileRuntimeHostProfileCatalog, + createRuntimeHostCapabilityProviderCredentialStore, createRuntimeHostProfileCredentialStore, + decodeRemoteRuntimeHostProfile, decodeRuntimeHostProfileDocument, RUNTIME_HOST_PLAINTEXT_ACKNOWLEDGEMENT, + RuntimeHostProfileConnectionError, sameRemoteRuntimeHostProfileTarget, type RemoteRuntimeHostProfile, + type RuntimeHostProfileCredential, type RuntimeHostProfileCredentialStore, } from '../client/host-profile.js'; import { RuntimeHostPermanentReconnectError } from '../client/reconnect-lifecycle.js'; @@ -267,6 +271,7 @@ describe('Runtime Host profiles', () => { await desktop.create(profile, 'desktop-token'); const created = await desktop.resolve(profile.id); + assert.ok(created.profileIncarnationId); await assert.rejects(() => cli.create(profile, 'duplicate-token'), /new profile id/u); await cli.save({ ...profile, name: 'Rotated' }, 'rotated-token'); @@ -285,6 +290,7 @@ describe('Runtime Host profiles', () => { }); const rotated = await desktop.resolve(profile.id); assert.equal(rotated.credential, 'rotated-token'); + assert.equal(rotated.profileIncarnationId, created.profileIncarnationId); assert.equal((await desktop.removeIfCurrent(rotated)).removed, true); assert.deepEqual(await desktop.read(), { schemaVersion: 3, profiles: [] }); }); @@ -299,13 +305,16 @@ describe('Runtime Host profiles', () => { await desktop.create(original, 'old-token'); const expected = await desktop.resolve(original.id); + assert.ok(expected.profileIncarnationId); assert.equal((await desktop.rebindIfCurrent(expected, replacement, 'new-token')).rebound, true); - assert.deepEqual(await desktop.resolve(original.id), { + const rebound = await desktop.resolve(original.id); + assert.deepEqual(rebound, { profile: { ...replacement, transport: { kind: 'tls', url: 'wss://runtime.example.com/' }, }, credential: 'new-token', + profileIncarnationId: expected.profileIncarnationId, }); await external.save({ ...replacement, name: 'Externally updated' }, 'external-token'); @@ -411,11 +420,218 @@ describe('Runtime Host profiles', () => { const resolved = await first.resolve('office'); assert.equal(resolved.credential, 'token-a'); - await credentials.set(targetB, 'token-b'); - assert.equal(await credentials.get(targetA), 'token-a'); - assert.equal(await credentials.get(targetB), 'token-b'); + await credentials.set(targetB, { + credential: 'token-b', + profileIncarnationId: 'target-b-incarnation', + }); + assert.equal((await credentials.get(targetA))?.credential, 'token-a'); + assert.equal((await credentials.get(targetB))?.credential, 'token-b'); await credentials.delete(targetB); - assert.equal(await credentials.get(targetA), 'token-a'); + assert.equal((await credentials.get(targetA))?.credential, 'token-a'); + }); + + test('keeps legacy access credentials readable while assigning a stable incarnation', async () => { + const profile = remoteProfile('office', 'wss://a.example.com', ROOT_A); + let stored = 'legacy-token'; + const credentials = createRuntimeHostProfileCredentialStore({ + getSecret: async () => stored, + setSecret: async (_slot, _kind, value) => { + stored = value; + }, + deleteSecret: async () => { + stored = ''; + }, + }); + + const first = await credentials.get(profile); + const second = await credentials.get(profile); + assert.equal(first?.credential, 'legacy-token'); + assert.equal(first?.profileIncarnationId, second?.profileIncarnationId); + assert.ok(first?.profileIncarnationId); + + await credentials.set(profile, { + credential: 'rotated-token', + profileIncarnationId: first.profileIncarnationId, + }); + assert.deepEqual(await credentials.get(profile), { + credential: 'rotated-token', + profileIncarnationId: first.profileIncarnationId, + }); + }); + + test('isolates capability-provider credentials by target and owning Client', async () => { + const path = await profilePath(); + const credentials = createRuntimeHostCapabilityProviderCredentialStore( + createFileCredentialStore(join(dirname(path), 'credentials')), + ); + const targetA = remoteProfile('office', 'wss://a.example.com', ROOT_A); + const targetB = remoteProfile('office', 'wss://b.example.com', ROOT_B); + const incarnationA = { profile: targetA, profileIncarnationId: 'incarnation-a' }; + const recreatedIncarnationA = { + profile: targetA, + profileIncarnationId: 'incarnation-a-recreated', + }; + const incarnationB = { profile: targetB, profileIncarnationId: 'incarnation-b' }; + + await assert.rejects( + () => credentials.set(incarnationA, 'owner-a', 'not a token'), + /credential is invalid/, + ); + await credentials.set(incarnationA, 'owner-a', 'provider-a'); + assert.equal(await credentials.get(incarnationA, 'owner-b'), null); + await credentials.set(incarnationA, 'owner-b', 'provider-b'); + await credentials.set(incarnationB, 'owner-a', 'provider-other-target'); + + assert.equal(await credentials.get(incarnationA, 'owner-a'), null); + assert.equal(await credentials.get(incarnationA, 'owner-b'), 'provider-b'); + assert.equal(await credentials.get(recreatedIncarnationA, 'owner-b'), null); + assert.equal(await credentials.get(incarnationB, 'owner-a'), 'provider-other-target'); + await credentials.delete(incarnationA, 'owner-a'); + assert.equal(await credentials.get(incarnationA, 'owner-a'), null); + assert.equal(await credentials.get(incarnationA, 'owner-b'), 'provider-b'); + }); + + test('removing a profile retires its terminal and provider credentials together', async () => { + const path = await profilePath(); + const credentialStore = createFileCredentialStore(join(dirname(path), 'credentials')); + const catalog = createFileRuntimeHostProfileCatalog( + path, + createRuntimeHostProfileCredentialStore(credentialStore), + ); + const providers = createRuntimeHostCapabilityProviderCredentialStore(credentialStore); + const profile = remoteProfile('office', 'wss://a.example.com', ROOT_A); + await catalog.save(profile, 'terminal-token'); + const target = await catalog.resolve(profile.id); + assert.ok(target.profileIncarnationId); + const incarnation = { profile, profileIncarnationId: target.profileIncarnationId }; + await providers.set(incarnation, 'owner-a', 'provider-token'); + + await catalog.remove(profile.id); + + assert.equal(await providers.get(incarnation, 'owner-a'), null); + }); + + test('profile removal excludes a queued provider credential mutation', async () => { + const path = await profilePath(); + const credentialStore = createFileCredentialStore(join(dirname(path), 'credentials')); + const stored = createRuntimeHostProfileCredentialStore(credentialStore); + const removalStarted = deferred(); + const allowRemoval = deferred(); + const credentials: RuntimeHostProfileCredentialStore = { + ...stored, + delete: async (profile) => { + removalStarted.resolve(); + await allowRemoval.promise; + await stored.delete(profile); + }, + }; + const removingCatalog = createFileRuntimeHostProfileCatalog(path, credentials); + const mutatingCatalog = createFileRuntimeHostProfileCatalog(path, credentials); + const providers = createRuntimeHostCapabilityProviderCredentialStore(credentialStore); + const profile = remoteProfile('office', 'wss://a.example.com', ROOT_A); + await removingCatalog.save(profile, 'terminal-token'); + const resolved = await removingCatalog.resolve(profile.id); + assert.ok(resolved.profileIncarnationId); + const incarnation = { profile, profileIncarnationId: resolved.profileIncarnationId }; + + const removal = removingCatalog.remove(profile.id); + await removalStarted.promise; + let mutationRan = false; + const mutation = mutatingCatalog.mutateRemoteProfileIfCurrent(incarnation, async (current) => { + mutationRan = true; + await providers.set( + { profile: current, profileIncarnationId: incarnation.profileIncarnationId }, + 'owner-a', + 'provider-token', + ); + }); + allowRemoval.resolve(); + + await removal; + assert.equal(await mutation, false); + assert.equal(mutationRan, false); + assert.equal(await providers.get(incarnation, 'owner-a'), null); + }); + + test('profile incarnation validation waits for removal rollback', async () => { + const path = await profilePath(); + const credentialStore = createFileCredentialStore(join(dirname(path), 'credentials')); + const stored = createRuntimeHostProfileCredentialStore(credentialStore); + const removalStarted = deferred(); + const allowRemovalFailure = deferred(); + const credentials: RuntimeHostProfileCredentialStore = { + ...stored, + delete: async () => { + removalStarted.resolve(); + await allowRemovalFailure.promise; + throw new Error('credential store unavailable'); + }, + }; + const removingCatalog = createFileRuntimeHostProfileCatalog(path, credentials); + const validatingCatalog = createFileRuntimeHostProfileCatalog(path, credentials); + const profile = remoteProfile('office', 'wss://a.example.com', ROOT_A); + await removingCatalog.save(profile, 'terminal-token'); + const resolved = await removingCatalog.resolve(profile.id); + assert.ok(resolved.profileIncarnationId); + const incarnation = { profile, profileIncarnationId: resolved.profileIncarnationId }; + + const removal = removingCatalog.remove(profile.id); + await removalStarted.promise; + let validationSettled = false; + const validation = validatingCatalog.readRemoteProfileIfCurrent(incarnation).then((current) => { + validationSettled = true; + return current; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(validationSettled, false); + + allowRemovalFailure.resolve(); + await assert.rejects(removal, /credential store unavailable/u); + assert.deepEqual(await validation, decodeRemoteRuntimeHostProfile(profile)); + }); + + test('recreating the same profile id and target assigns a new incarnation', async () => { + const path = await profilePath(); + const credentialStore = createFileCredentialStore(join(dirname(path), 'credentials')); + const catalog = createFileRuntimeHostProfileCatalog( + path, + createRuntimeHostProfileCredentialStore(credentialStore), + ); + const providers = createRuntimeHostCapabilityProviderCredentialStore(credentialStore); + const profile = remoteProfile('office', 'wss://a.example.com', ROOT_A); + await catalog.create(profile, 'terminal-token'); + const first = await catalog.resolve(profile.id); + assert.ok(first.profileIncarnationId); + const firstIncarnation = { + profile, + profileIncarnationId: first.profileIncarnationId, + }; + await providers.set(firstIncarnation, 'owner-a', 'provider-token'); + + await catalog.remove(profile.id); + await catalog.create(profile, 'terminal-token'); + const second = await catalog.resolve(profile.id); + assert.ok(second.profileIncarnationId); + const secondIncarnation = { + profile, + profileIncarnationId: second.profileIncarnationId, + }; + + assert.notEqual(second.profileIncarnationId, first.profileIncarnationId); + assert.equal(await catalog.readRemoteProfileIfCurrent(firstIncarnation), undefined); + assert.deepEqual( + await catalog.readRemoteProfileIfCurrent(secondIncarnation), + decodeRemoteRuntimeHostProfile(profile), + ); + assert.equal(await providers.get(secondIncarnation, 'owner-a'), null); + let staleMutationRan = false; + assert.equal( + await catalog.mutateRemoteProfileIfCurrent(firstIncarnation, async () => { + staleMutationRan = true; + }), + false, + ); + assert.equal(staleMutationRan, false); }); test('pins a direct-peer profile to its PeerId while allowing route discovery to change', () => { @@ -433,7 +649,7 @@ describe('Runtime Host profiles', () => { test('keeps profile metadata when credential removal fails', async () => { const path = await profilePath(); - const values = new Map(); + const values = new Map(); const credentials: RuntimeHostProfileCredentialStore = { get: async (profile) => values.get(profile.id) ?? null, set: async (profile, credential) => { @@ -472,6 +688,8 @@ describe('Runtime Host profiles', () => { const targetA = remoteProfile('office', 'wss://a.example.com', ROOT_A); const targetB = { ...targetA, name: 'updated' }; await catalog.save(targetA, 'token-a'); + const original = await catalog.resolve('office'); + assert.ok(original.profileIncarnationId); rejectNextSet = true; await assert.rejects(() => catalog.save(targetB, 'token-b'), /credential store unavailable/); @@ -481,6 +699,7 @@ describe('Runtime Host profiles', () => { transport: { kind: 'tls', url: 'wss://a.example.com/' }, }, credential: 'token-a', + profileIncarnationId: original.profileIncarnationId, }); }); @@ -689,7 +908,11 @@ describe('Runtime Host profiles', () => { connect: async () => ({ kind: 'unavailable', reason: 'root_mismatch' }), }, ), - RuntimeHostPermanentReconnectError, + (error: unknown) => { + assert.ok(error instanceof RuntimeHostProfileConnectionError); + assert.equal(error.reason, 'target_mismatch'); + return true; + }, ); }); @@ -873,6 +1096,8 @@ describe('Runtime Host profiles', () => { ), (error: unknown) => { assert.ok(error instanceof RuntimeHostPermanentReconnectError); + assert.ok(error instanceof RuntimeHostProfileConnectionError); + assert.equal(error.reason, 'credential_rejected'); assert.match(error.message, /rejected its access credential/u); return true; }, @@ -932,7 +1157,7 @@ function directPeerProfile( } function memoryCredentials(): RuntimeHostProfileCredentialStore { - const values = new Map(); + const values = new Map(); const key = (profile: RemoteRuntimeHostProfile) => `${profile.id}\0${JSON.stringify(profile.transport)}\0${profile.rootId}`; return { @@ -946,6 +1171,14 @@ function memoryCredentials(): RuntimeHostProfileCredentialStore { }; } +function deferred() { + let resolve!: () => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + function incompatibleHandshake(overrides: Partial = {}): HostIncompatible { return { kind: 'incompatible', diff --git a/packages/runtime-host/src/client/host-profile.ts b/packages/runtime-host/src/client/host-profile.ts index f3230ff452..e9a22ad544 100644 --- a/packages/runtime-host/src/client/host-profile.ts +++ b/packages/runtime-host/src/client/host-profile.ts @@ -18,6 +18,7 @@ */ import { createHash, randomUUID } from 'node:crypto'; +import { watch } from 'node:fs'; import { chmod, mkdir, open, readFile, rename, rm } from 'node:fs/promises'; import { dirname, join, posix } from 'node:path'; import { createFileCredentialStore, type CredentialStore } from '@maka/storage/credential-store'; @@ -26,6 +27,7 @@ import { INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, isCanonicalRuntimeHostWebSocketPath, RUNTIME_HOST_PROTOCOL_VERSION, + requireClientInstanceId, requireHostRootId, } from '../protocol/index.js'; import type { RuntimeHostProfileOfKind } from '../profile-kind.js'; @@ -61,6 +63,7 @@ import { } from './wsl-environment.js'; const PROFILE_SCHEMA_VERSION = 3; +const CLIENT_PROFILE_DOCUMENT_NAME = 'runtime-host-profiles.json'; const PROFILE_DOCUMENT_MAX_BYTES = 64 * 1024; const PROFILE_COUNT_MAX = 32; const PROFILE_NAME_MAX_BYTES = 128; @@ -68,6 +71,8 @@ const PROFILE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/; const PEER_ID_MAX_BYTES = 160; const PEER_ADDRESS_MAX_BYTES = 2 * 1024; const PEER_ROUTE_MAX = 16; +const PROFILE_CREDENTIAL_RECORD_PREFIX = 'maka-runtime-host-profile-credential-v1:'; +const PROFILE_INCARNATION_ID_MAX_BYTES = 128; export const RUNTIME_HOST_ACCESS_CREDENTIAL_MAX_BYTES = 8 * 1024; export const RUNTIME_HOST_PLAINTEXT_ACKNOWLEDGEMENT = 'plaintext-bearer-v1' as const; @@ -154,6 +159,12 @@ export interface RuntimeHostProfileDocument { export interface ResolvedRuntimeHostProfile { readonly profile: RuntimeHostProfile; readonly credential?: string; + readonly profileIncarnationId?: string; +} + +export interface RuntimeHostRemoteProfileIncarnation { + readonly profile: RemoteRuntimeHostProfile; + readonly profileIncarnationId: string; } export type RuntimeHostConnectionPhase = @@ -212,14 +223,57 @@ export interface RuntimeHostProfileCatalog { readonly rebound: boolean; readonly document: RuntimeHostProfileDocument; }>; + /** Serialize one sidecar mutation with catalog updates while this profile lifetime remains current. */ + mutateRemoteProfileIfCurrent( + target: RuntimeHostRemoteProfileIncarnation, + mutation: (profile: RemoteRuntimeHostProfile) => Promise, + ): Promise; + /** Return the canonical profile while this exact profile lifetime remains current. */ + readRemoteProfileIfCurrent( + target: RuntimeHostRemoteProfileIncarnation, + ): Promise; +} + +export interface RuntimeHostProfileCredential { + readonly credential: string; + /** Stable for updates, replaced when removal and recreation start a new profile lifetime. */ + readonly profileIncarnationId: string; } export interface RuntimeHostProfileCredentialStore { - get(profile: RemoteRuntimeHostProfile): Promise; - set(profile: RemoteRuntimeHostProfile, credential: string): Promise; + get(profile: RemoteRuntimeHostProfile): Promise; + set(profile: RemoteRuntimeHostProfile, credential: RuntimeHostProfileCredential): Promise; delete(profile: RemoteRuntimeHostProfile): Promise; } +export interface RuntimeHostCapabilityProviderCredentialStore { + get( + target: RuntimeHostRemoteProfileIncarnation, + ownerClientInstanceId: string, + ): Promise; + set( + target: RuntimeHostRemoteProfileIncarnation, + ownerClientInstanceId: string, + credential: string, + ): Promise; + delete(target: RuntimeHostRemoteProfileIncarnation, ownerClientInstanceId: string): Promise; +} + +export type RuntimeHostProfileConnectionFailureReason = + | 'credential_required' + | 'credential_rejected' + | 'target_mismatch'; + +export class RuntimeHostProfileConnectionError extends RuntimeHostPermanentReconnectError { + constructor( + readonly reason: RuntimeHostProfileConnectionFailureReason, + message: string, + ) { + super(message); + this.name = 'RuntimeHostProfileConnectionError'; + } +} + export function createFileRuntimeHostProfileCatalog( path: string, credentials: RuntimeHostProfileCredentialStore, @@ -232,11 +286,22 @@ export function createClientRuntimeHostProfileCatalog( credentialStore: CredentialStore = createClientRuntimeHostCredentialStore(clientDataRoot), ): RuntimeHostProfileCatalog { return createFileRuntimeHostProfileCatalog( - join(clientDataRoot, 'runtime-host-profiles.json'), + join(clientDataRoot, CLIENT_PROFILE_DOCUMENT_NAME), createRuntimeHostProfileCredentialStore(credentialStore), ); } +export function subscribeClientRuntimeHostProfileCatalogChanges( + clientDataRoot: string, + listener: (error?: Error) => void, +): () => void { + const watcher = watch(clientDataRoot, (_eventType, filename) => { + if (filename === null || filename.toString() === CLIENT_PROFILE_DOCUMENT_NAME) listener(); + }); + watcher.on('error', (error) => listener(error)); + return () => watcher.close(); +} + export function createClientRuntimeHostCredentialStore(clientDataRoot: string): CredentialStore { return createFileCredentialStore(join(clientDataRoot, 'runtime-host-client')); } @@ -246,27 +311,65 @@ export function createRuntimeHostProfileCredentialStore( ): RuntimeHostProfileCredentialStore { return { get: async (profile) => { - return credentials.getSecret(profileCredentialSlot(profile), 'runtime_host_access'); + const stored = await credentials.getSecret( + profileCredentialSlot(profile), + 'runtime_host_access', + ); + return stored === null ? null : decodeProfileCredential(profile, stored); }, set: (profile, credential) => { - if ( - !credential || - /\s/u.test(credential) || - Buffer.byteLength(credential, 'utf8') > RUNTIME_HOST_ACCESS_CREDENTIAL_MAX_BYTES - ) { - return Promise.reject(new Error('Runtime Host access credential is invalid')); + try { + const encoded = encodeProfileCredential(credential); + return credentials.setSecret( + profileCredentialSlot(profile), + 'runtime_host_access', + encoded, + ); + } catch (error) { + return Promise.reject(error); } - return credentials.setSecret( - profileCredentialSlot(profile), - 'runtime_host_access', - credential, + }, + delete: (profile) => credentials.deleteSecret(profileCredentialSlot(profile)), + }; +} + +export function createRuntimeHostCapabilityProviderCredentialStore( + credentials: Pick, +): RuntimeHostCapabilityProviderCredentialStore { + return { + get: async (target, ownerClientInstanceId) => { + const stored = await credentials.getSecret( + profileCredentialSlot(target.profile), + 'runtime_host_capability_provider', + ); + if (stored === null) return null; + const decoded = decodeCapabilityProviderCredential(stored); + return decoded.ownerClientInstanceId === requireClientInstanceId(ownerClientInstanceId) && + decoded.profileIncarnationId === requireProfileIncarnationId(target.profileIncarnationId) + ? decoded.credential + : null; + }, + set: async (target, ownerClientInstanceId, credential) => { + await credentials.setSecret( + profileCredentialSlot(target.profile), + 'runtime_host_capability_provider', + JSON.stringify({ + schemaVersion: 1, + profileIncarnationId: requireProfileIncarnationId(target.profileIncarnationId), + ownerClientInstanceId: requireClientInstanceId(ownerClientInstanceId), + credential: requireRuntimeHostAccessCredential(credential), + }), ); }, - delete: (profile) => - credentials.deleteSecret(profileCredentialSlot(profile), 'runtime_host_access'), + delete: (target, ownerClientInstanceId) => + deleteCapabilityProviderCredential(credentials, target, ownerClientInstanceId), }; } +export function runtimeHostProfileTargetFingerprint(profile: RemoteRuntimeHostProfile): string { + return profileCredentialBinding(profile); +} + export async function connectRuntimeHostProfile( input: { readonly profile: PersistedRuntimeHostProfile; @@ -312,7 +415,8 @@ export async function connectRuntimeHostProfile( ); } if (!input.credential) { - throw new RuntimeHostPermanentReconnectError( + throw new RuntimeHostProfileConnectionError( + 'credential_required', `Runtime Host profile ${input.profile.id} has no access credential`, ); } @@ -423,6 +527,20 @@ export async function connectRemoteRuntimeHostProfile( if (connected.kind === 'draining') { throw new Error(`Runtime Host profile ${input.profile.id} is draining`); } + if (connected.reason === 'authentication_failed') { + throw new RuntimeHostProfileConnectionError( + 'credential_rejected', + `Runtime Host profile ${input.profile.id} rejected its access credential`, + ); + } + if (connected.reason === 'root_mismatch' || connected.reason === 'composition_mismatch') { + throw new RuntimeHostProfileConnectionError( + 'target_mismatch', + connected.reason === 'root_mismatch' + ? `Runtime Host profile ${input.profile.id} connected to an unexpected State Root` + : `Runtime Host profile ${input.profile.id} has an incompatible Host composition`, + ); + } throw remoteRuntimeHostUnavailableError( `Runtime Host profile ${input.profile.id}`, connected.reason, @@ -493,7 +611,8 @@ export async function connectPeerRuntimeHost(input: { input.handshakeTimeoutMs, ); if (!authentication.accepted) { - throw new RuntimeHostPermanentReconnectError( + throw new RuntimeHostProfileConnectionError( + 'credential_rejected', `Runtime Host profile ${input.profileId} rejected its access credential`, ); } @@ -519,6 +638,14 @@ export async function connectPeerRuntimeHost(input: { } if (result.kind === 'draining') throw new Error('Runtime Host direct peer is draining'); if (result.kind === 'unavailable') { + if (result.reason === 'root_mismatch' || result.reason === 'composition_mismatch') { + throw new RuntimeHostProfileConnectionError( + 'target_mismatch', + result.reason === 'root_mismatch' + ? `Runtime Host profile ${input.profileId} connected to an unexpected State Root` + : `Runtime Host profile ${input.profileId} has an incompatible Host composition`, + ); + } throw remoteRuntimeHostUnavailableError('Runtime Host direct peer', result.reason); } transferred = true; @@ -670,13 +797,17 @@ class FileRuntimeHostProfileCatalog implements RuntimeHostProfileCatalog { ); } if (profile.kind === 'environment') return { profile }; - const credential = await this.credentials.get(profile); - if (!credential) { + const storedCredential = await this.credentials.get(profile); + if (!storedCredential) { throw new RuntimeHostPermanentReconnectError( `Runtime Host profile ${profile.id} has no access credential`, ); } - return { profile, credential }; + return { + profile, + credential: storedCredential.credential, + profileIncarnationId: storedCredential.profileIncarnationId, + }; } save( @@ -732,7 +863,10 @@ class FileRuntimeHostProfileCatalog implements RuntimeHostProfileCatalog { : [...current.profiles, profile], }); if (profile.kind === 'remote' && suppliedCredential !== undefined) { - await this.credentials.set(profile, suppliedCredential); + await this.credentials.set(profile, { + credential: suppliedCredential, + profileIncarnationId: previousCredential?.profileIncarnationId ?? randomUUID(), + }); } try { await writeProfileDocument(this.path, next); @@ -786,7 +920,7 @@ class FileRuntimeHostProfileCatalog implements RuntimeHostProfileCatalog { !samePersistedRuntimeHostProfile(profile, expectedProfile) || (profile.kind === 'remote' && (target.credential === undefined || - (await this.credentials.get(profile)) !== target.credential)) + !sameProfileCredential(await this.credentials.get(profile), target))) ) { return { removed: false, document: current }; } @@ -819,11 +953,14 @@ class FileRuntimeHostProfileCatalog implements RuntimeHostProfileCatalog { return this.#exclusive(async () => { const current = await this.#readSnapshot(); const stored = current.profiles.find((candidate) => candidate.id === expectedProfile.id); + const storedCredential = + stored?.kind === 'remote' ? await this.credentials.get(stored) : null; if ( !stored || stored.kind !== 'remote' || !sameRemoteRuntimeHostProfile(stored, expectedProfile) || - (await this.credentials.get(stored)) !== target.credential + !storedCredential || + !sameProfileCredential(storedCredential, target) ) { return { rebound: false, document: current }; } @@ -833,11 +970,14 @@ class FileRuntimeHostProfileCatalog implements RuntimeHostProfileCatalog { candidate.id === profile.id ? profile : candidate, ), }); - await this.credentials.set(profile, credential); + await this.credentials.set(profile, { + credential, + profileIncarnationId: storedCredential.profileIncarnationId, + }); try { await writeProfileDocument(this.path, next); } catch (error) { - await restoreCredential(this.credentials, profile, target.credential).catch( + await restoreCredential(this.credentials, profile, storedCredential).catch( (rollbackError) => { throw new AggregateError( [error, rollbackError], @@ -851,6 +991,46 @@ class FileRuntimeHostProfileCatalog implements RuntimeHostProfileCatalog { }); } + mutateRemoteProfileIfCurrent( + target: RuntimeHostRemoteProfileIncarnation, + mutation: (profile: RemoteRuntimeHostProfile) => Promise, + ): Promise { + const expectedProfile = decodeRemoteRuntimeHostProfile(target.profile); + const expectedIncarnationId = requireProfileIncarnationId(target.profileIncarnationId); + return this.#exclusive(async () => { + const current = await this.#readSnapshot(); + const profile = current.profiles.find( + (candidate): candidate is RemoteRuntimeHostProfile => + candidate.id === expectedProfile.id && candidate.kind === 'remote', + ); + if (!profile || !sameRemoteRuntimeHostProfileTarget(profile, expectedProfile)) return false; + const credential = await this.credentials.get(profile); + if (credential?.profileIncarnationId !== expectedIncarnationId) return false; + await mutation(profile); + return true; + }); + } + + async readRemoteProfileIfCurrent( + target: RuntimeHostRemoteProfileIncarnation, + ): Promise { + const expectedProfile = decodeRemoteRuntimeHostProfile(target.profile); + const expectedIncarnationId = requireProfileIncarnationId(target.profileIncarnationId); + return this.#exclusive(async () => { + const current = await this.#readSnapshot(); + const profile = current.profiles.find( + (candidate): candidate is RemoteRuntimeHostProfile => + candidate.id === expectedProfile.id && candidate.kind === 'remote', + ); + if (!profile || !sameRemoteRuntimeHostProfileTarget(profile, expectedProfile)) { + return undefined; + } + return (await this.credentials.get(profile))?.profileIncarnationId === expectedIncarnationId + ? profile + : undefined; + }); + } + async #removeProfile( current: RuntimeHostProfileDocument, profile: PersistedRuntimeHostProfile, @@ -1074,6 +1254,121 @@ function profileCredentialSlot(profile: RemoteRuntimeHostProfile): string { return `runtime-host-profile:${requireProfileId(profile.id)}:${profileCredentialBinding(profile)}`; } +async function deleteCapabilityProviderCredential( + credentials: Pick, + target: RuntimeHostRemoteProfileIncarnation, + ownerClientInstanceId: string, +): Promise { + const slot = profileCredentialSlot(target.profile); + const stored = await credentials.getSecret(slot, 'runtime_host_capability_provider'); + if (stored === null) return; + const decoded = decodeCapabilityProviderCredential(stored); + if ( + decoded.ownerClientInstanceId !== requireClientInstanceId(ownerClientInstanceId) || + decoded.profileIncarnationId !== requireProfileIncarnationId(target.profileIncarnationId) + ) { + return; + } + await credentials.deleteSecret(slot, 'runtime_host_capability_provider'); +} + +function decodeCapabilityProviderCredential(value: string): { + readonly ownerClientInstanceId: string; + readonly credential: string; + readonly profileIncarnationId: string; +} { + try { + const parsed: unknown = JSON.parse(value); + const record = requireExactRecord(parsed, 'Runtime Host capability-provider credential', [ + 'schemaVersion', + 'profileIncarnationId', + 'ownerClientInstanceId', + 'credential', + ]); + if (record.schemaVersion !== 1) { + throw new Error('Runtime Host capability-provider credential schema is unsupported'); + } + return { + ownerClientInstanceId: requireClientInstanceId(record.ownerClientInstanceId), + credential: requireRuntimeHostAccessCredential(record.credential as string), + profileIncarnationId: requireProfileIncarnationId(record.profileIncarnationId), + }; + } catch (error) { + throw new Error('Runtime Host capability-provider credential is invalid', { cause: error }); + } +} + +function encodeProfileCredential(credential: RuntimeHostProfileCredential): string { + return `${PROFILE_CREDENTIAL_RECORD_PREFIX}${JSON.stringify({ + schemaVersion: 1, + profileIncarnationId: requireProfileIncarnationId(credential.profileIncarnationId), + credential: requireRuntimeHostAccessCredential(credential.credential), + })}`; +} + +function decodeProfileCredential( + profile: RemoteRuntimeHostProfile, + value: string, +): RuntimeHostProfileCredential { + if (!value.startsWith(PROFILE_CREDENTIAL_RECORD_PREFIX)) { + const credential = requireRuntimeHostAccessCredential(value); + return { + credential, + profileIncarnationId: legacyProfileIncarnationId(profile), + }; + } + try { + const parsed: unknown = JSON.parse(value.slice(PROFILE_CREDENTIAL_RECORD_PREFIX.length)); + const record = requireExactRecord(parsed, 'Runtime Host profile credential', [ + 'schemaVersion', + 'profileIncarnationId', + 'credential', + ]); + if (record.schemaVersion !== 1) { + throw new Error('Runtime Host profile credential schema is unsupported'); + } + return { + credential: requireRuntimeHostAccessCredential(record.credential as string), + profileIncarnationId: requireProfileIncarnationId(record.profileIncarnationId), + }; + } catch (error) { + throw new Error('Runtime Host profile credential is invalid', { cause: error }); + } +} + +function legacyProfileIncarnationId(profile: RemoteRuntimeHostProfile): string { + // Existing plaintext records predate incarnations. Their target-bound value + // remains stable until the next catalog write migrates the credential record. + return createHash('sha256') + .update('legacy-runtime-host-profile-incarnation') + .update('\0') + .update(profileCredentialSlot(profile)) + .digest('hex'); +} + +function requireProfileIncarnationId(value: unknown): string { + if ( + typeof value !== 'string' || + value.length === 0 || + Buffer.byteLength(value, 'utf8') > PROFILE_INCARNATION_ID_MAX_BYTES || + !/^[A-Za-z0-9._-]+$/u.test(value) + ) { + throw new Error('Runtime Host profile incarnation is invalid'); + } + return value; +} + +function requireRuntimeHostAccessCredential(credential: string): string { + if ( + !credential || + /\s/u.test(credential) || + Buffer.byteLength(credential, 'utf8') > RUNTIME_HOST_ACCESS_CREDENTIAL_MAX_BYTES + ) { + throw new Error('Runtime Host access credential is invalid'); + } + return credential; +} + function profileTargetBinding(profile: PersistedRuntimeHostProfile): string { if (profile.kind === 'remote') return `remote\0${profileCredentialBinding(profile)}`; const normalized = decodeEnvironmentRuntimeHostProfile(profile); @@ -1172,13 +1467,25 @@ function samePersistedRuntimeHostProfile( function restoreCredential( credentials: RuntimeHostProfileCredentialStore, profile: RemoteRuntimeHostProfile, - previousCredential: string | null, + previousCredential: RuntimeHostProfileCredential | null, ): Promise { return previousCredential === null ? credentials.delete(profile) : credentials.set(profile, previousCredential); } +function sameProfileCredential( + stored: RuntimeHostProfileCredential | null, + expected: ResolvedRuntimeHostProfile, +): boolean { + return ( + stored !== null && + stored.credential === expected.credential && + (expected.profileIncarnationId === undefined || + stored.profileIncarnationId === expected.profileIncarnationId) + ); +} + function requireProfileName(value: unknown): string { const name = requireString(value, 'Runtime Host profile name').trim(); if ( diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts index e898eb0e87..72ec2478a8 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -44,6 +44,7 @@ export { createClientRuntimeHostCredentialStore, createClientRuntimeHostProfileCatalog, createFileRuntimeHostProfileCatalog, + createRuntimeHostCapabilityProviderCredentialStore, createRuntimeHostProfileCredentialStore, connectRuntimeHostProfile, connectRemoteRuntimeHostProfile, @@ -52,8 +53,10 @@ export { decodeRemoteRuntimeHostProfile, remoteRuntimeHostUnavailableError, runtimeHostProfileAccess, + runtimeHostProfileTargetFingerprint, sameRemoteRuntimeHostProfileTarget, sameResolvedRuntimeHostProfileTarget, + subscribeClientRuntimeHostProfileCatalogChanges, type EnvironmentRuntimeHostProfile, type PersistedRuntimeHostProfile, type RemoteRuntimeHostProfile, @@ -63,6 +66,10 @@ export { type RuntimeHostProfileAccess, type RuntimeHostProfileCatalog, type RuntimeHostConnectionPhase, + type RuntimeHostRemoteProfileIncarnation, + type RuntimeHostCapabilityProviderCredentialStore, + RuntimeHostProfileConnectionError, + type RuntimeHostProfileConnectionFailureReason, type RuntimeHostProfileDocument, } from './host-profile.js'; export { diff --git a/packages/storage/src/__tests__/file-lifetime-owner.test.ts b/packages/storage/src/__tests__/file-lifetime-owner.test.ts new file mode 100644 index 0000000000..1384f09cc2 --- /dev/null +++ b/packages/storage/src/__tests__/file-lifetime-owner.test.ts @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { fork } from 'node:child_process'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { tryAcquireFileLifetimeOwner } from '../file-lifetime-owner.js'; + +test('a file lifetime owner fails closed and recovers after owner death', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-file-lifetime-owner-')); + const path = join(root, 'nested', 'publication.lease'); + const holder = fork( + new URL('./fixtures/file-lifetime-owner-holder.js', import.meta.url), + [path], + { + stdio: ['ignore', 'ignore', 'inherit', 'ipc'], + }, + ); + t.after(async () => { + if (holder.exitCode === null && holder.signalCode === null) holder.kill('SIGKILL'); + await rm(root, { recursive: true, force: true }); + }); + await new Promise((resolve, reject) => { + holder.once('message', (message) => { + if (message === 'owned') resolve(); + else reject(new Error(`Unexpected file owner message: ${String(message)}`)); + }); + holder.once('error', reject); + holder.once('exit', (code, signal) => { + reject(new Error(`File owner exited before acquisition (${String(code)}, ${signal})`)); + }); + }); + + assert.equal(await tryAcquireFileLifetimeOwner(path), undefined); + + holder.kill('SIGKILL'); + await new Promise((resolve) => holder.once('exit', () => resolve())); + const successor = await tryAcquireFileLifetimeOwner(path); + assert.ok(successor); + await Promise.all([successor.close(), successor.close()]); +}); diff --git a/packages/storage/src/__tests__/fixtures/file-lifetime-owner-holder.ts b/packages/storage/src/__tests__/fixtures/file-lifetime-owner-holder.ts new file mode 100644 index 0000000000..39e2d55c70 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/file-lifetime-owner-holder.ts @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { tryAcquireFileLifetimeOwner } from '../../file-lifetime-owner.js'; + +const path = process.argv[2]; +if (!path) throw new Error('Missing file lifetime owner path'); +const owner = await tryAcquireFileLifetimeOwner(path); +if (!owner) throw new Error('File lifetime owner is already active'); +process.send?.('owned'); +setInterval(() => undefined, 1_000).unref(); +await new Promise((resolve) => process.once('disconnect', resolve)); +await owner.close(); diff --git a/packages/storage/src/credential-store.ts b/packages/storage/src/credential-store.ts index eaaf81cd95..325227e8d5 100644 --- a/packages/storage/src/credential-store.ts +++ b/packages/storage/src/credential-store.ts @@ -52,7 +52,8 @@ type StoredCredentialKind = | 'botAppSecret' | 'proxyPassword' | 'tavilyApiKey' - | 'runtimeHostAccess'; + | 'runtimeHostAccess' + | 'runtimeHostCapabilityProvider'; export type CredentialKind = | 'api_key' | 'oauth_token' @@ -61,7 +62,8 @@ export type CredentialKind = | 'app_secret' | 'proxy_password' | 'tavily_api_key' - | 'runtime_host_access'; + | 'runtime_host_access' + | 'runtime_host_capability_provider'; /** Current on-disk schema version. Unknown versions fail closed on read. */ export const CREDENTIAL_SCHEMA_VERSION = 1; @@ -349,6 +351,7 @@ const STORED_CREDENTIAL_KINDS = [ 'proxyPassword', 'tavilyApiKey', 'runtimeHostAccess', + 'runtimeHostCapabilityProvider', ] as const satisfies readonly StoredCredentialKind[]; function toStoredKind(kind: CredentialKind): StoredCredentialKind { @@ -369,5 +372,7 @@ function toStoredKind(kind: CredentialKind): StoredCredentialKind { return 'tavilyApiKey'; case 'runtime_host_access': return 'runtimeHostAccess'; + case 'runtime_host_capability_provider': + return 'runtimeHostCapabilityProvider'; } } diff --git a/packages/storage/src/file-lifetime-owner.ts b/packages/storage/src/file-lifetime-owner.ts index 843ee62af8..c2792e84e2 100644 --- a/packages/storage/src/file-lifetime-owner.ts +++ b/packages/storage/src/file-lifetime-owner.ts @@ -17,7 +17,8 @@ * under the License. */ -import type { FileHandle } from 'node:fs/promises'; +import { chmod, lstat, mkdir, type FileHandle } from 'node:fs/promises'; +import { dirname } from 'node:path'; import { openStableNativeLockFile, releaseNativeFileLock, @@ -29,10 +30,40 @@ export interface FileLifetimeOwner { } export async function acquireFileLifetimeOwner(path: string): Promise { + const owner = await tryAcquireOpenedFileLifetimeOwner(path); + if (!owner) throw new Error(`Another process owns ${path}`); + return owner; +} + +/** Try once to own one named file for the lifetime of this process handle. */ +export async function tryAcquireFileLifetimeOwner( + path: string, +): Promise { + const directory = dirname(path); + await mkdir(directory, { recursive: true, mode: 0o700 }); + const directoryStat = await lstat(directory); + if (!directoryStat.isDirectory() || directoryStat.isSymbolicLink()) { + throw new Error(`File lifetime owner root is not a directory: ${directory}`); + } + if (process.platform !== 'win32') await chmod(directory, 0o700); + + return tryAcquireOpenedFileLifetimeOwner(path); +} + +async function tryAcquireOpenedFileLifetimeOwner( + path: string, +): Promise { const handle = await openStableNativeLockFile(path); - if (!tryAcquireNativeFileLock(handle)) { + let acquired: boolean; + try { + acquired = tryAcquireNativeFileLock(handle); + } catch (error) { + await handle.close().catch(() => undefined); + throw error; + } + if (!acquired) { await handle.close(); - throw new Error(`Another process owns ${path}`); + return undefined; } return new FileLifetimeOwnerImpl(handle); }