diff --git a/packages/cli/src/__tests__/acp-agent.test.ts b/packages/cli/src/__tests__/acp-agent.test.ts index fad4260895..2d1662c2b1 100644 --- a/packages/cli/src/__tests__/acp-agent.test.ts +++ b/packages/cli/src/__tests__/acp-agent.test.ts @@ -23,13 +23,13 @@ import { client, methods, RequestError } from '@agentclientprotocol/sdk'; import { createMakaAcpAgent } from '../acp/maka-acp-agent.js'; describe('Maka ACP agent', () => { - test('returns the Maka identity with no advertised capabilities or authentication', async () => { + test('returns the Maka identity and advertises only Session listing', async () => { await client({ name: 'test-client' }).connectWith( - createMakaAcpAgent({ version: '0.2.0' }), + createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), async (agent) => { assert.deepEqual(await agent.request(methods.agent.initialize, { protocolVersion: 1 }), { protocolVersion: 1, - agentCapabilities: {}, + agentCapabilities: { sessionCapabilities: { list: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }); @@ -37,16 +37,85 @@ describe('Maka ACP agent', () => { ); }); - test('rejects unimplemented session requests with method details', async () => { + test('routes official SDK new, list, and configuration requests through the Session registry', async () => { + const creates: unknown[] = []; + const lists: unknown[] = []; + const configurations: unknown[] = []; await client({ name: 'test-client' }).connectWith( - createMakaAcpAgent({ version: '0.2.0' }), + createMakaAcpAgent({ + version: '0.2.0', + sessionRegistry: fakeSessionRegistry({ creates, lists, configurations }), + }), + async (agent) => { + assert.deepEqual( + await agent.request(methods.agent.session.new, { + cwd: '/workspace', + mcpServers: [], + _meta: { ignored: true }, + }), + { sessionId: 'session-1', configOptions: CONFIG_OPTIONS }, + ); + assert.deepEqual(await agent.request(methods.agent.session.list, { cwd: '/workspace' }), { + sessions: [ + { + sessionId: 'session-1', + cwd: '/workspace', + title: 'Session', + updatedAt: '2026-08-24T00:00:00.000Z', + }, + ], + }); + assert.deepEqual( + await agent.request(methods.agent.session.setConfigOption, { + sessionId: 'session-1', + configId: 'collaboration_mode', + value: 'plan', + }), + { configOptions: CONFIG_OPTIONS }, + ); + }, + ); + assert.deepEqual(creates, [{ cwd: '/workspace', mcpServers: [], _meta: { ignored: true } }]); + assert.deepEqual(lists, [{ cwd: '/workspace' }]); + assert.deepEqual(configurations, [ + { + sessionId: 'session-1', + configId: 'collaboration_mode', + value: 'plan', + }, + ]); + }); + + test('does not implement or advertise session/close', async () => { + await client({ name: 'test-client' }).connectWith( + createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), + async (agent) => { + await assert.rejects( + agent.request(methods.agent.session.close, { sessionId: 'session-1' }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32601); + assert.deepEqual(error.data, { method: 'session/close' }); + return true; + }, + ); + }, + ); + }); + + test('keeps session/set_mode unsupported', async () => { + await client({ name: 'test-client' }).connectWith( + createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), async (agent) => { await assert.rejects( - agent.request('session/new', { cwd: '/workspace' }), + agent.request(methods.agent.session.setMode, { + sessionId: 'session-1', + modeId: 'plan', + }), (error: unknown) => { assert.ok(error instanceof RequestError); assert.equal(error.code, -32601); - assert.deepEqual(error.data, { method: 'session/new' }); + assert.deepEqual(error.data, { method: 'session/set_mode' }); return true; }, ); @@ -57,7 +126,7 @@ describe('Maka ACP agent', () => { test('selects v1 when the client requests an unsupported lower or higher version', async () => { for (const protocolVersion of [0, 2]) { await client({ name: 'test-client' }).connectWith( - createMakaAcpAgent({ version: '0.2.0' }), + createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), async (agent) => { const response = await agent.request(methods.agent.initialize, { protocolVersion }); assert.equal(response.protocolVersion, 1); @@ -66,3 +135,86 @@ describe('Maka ACP agent', () => { } }); }); + +const CONFIG_OPTIONS = [ + { + type: 'select' as const, + id: 'permission_mode', + name: 'Permission mode', + category: '_maka/permission_mode', + currentValue: 'ask', + options: [ + { value: 'explore', name: 'Explore' }, + { value: 'ask', name: 'Ask' }, + { value: 'bypass', name: 'Bypass' }, + ], + }, + { + type: 'select' as const, + id: 'thinking_level', + name: 'Thinking level', + category: 'thought_level', + currentValue: 'default', + options: [ + { value: 'default', name: 'Default' }, + { value: 'off', name: 'Off' }, + { value: 'minimal', name: 'Minimal' }, + { value: 'low', name: 'Low' }, + { value: 'medium', name: 'Medium' }, + { value: 'high', name: 'High' }, + { value: 'xhigh', name: 'Extra high' }, + { value: 'max', name: 'Max' }, + ], + }, + { + type: 'select' as const, + id: 'collaboration_mode', + name: 'Collaboration mode', + category: 'mode', + currentValue: 'plan', + options: [ + { value: 'agent', name: 'Agent' }, + { value: 'plan', name: 'Plan' }, + ], + }, + { + type: 'select' as const, + id: 'orchestration_mode', + name: 'Orchestration mode', + category: '_maka/orchestration_mode', + currentValue: 'default', + options: [ + { value: 'default', name: 'Default' }, + { value: 'swarm', name: 'Swarm' }, + { value: 'graph', name: 'Graph' }, + ], + }, +]; + +function fakeSessionRegistry( + observations: { creates?: unknown[]; lists?: unknown[]; configurations?: unknown[] } = {}, +) { + return { + create: async (params: unknown) => { + observations.creates?.push(params); + return { sessionId: 'session-1', configOptions: CONFIG_OPTIONS }; + }, + list: async (params: unknown) => { + observations.lists?.push(params); + return { + sessions: [ + { + sessionId: 'session-1', + cwd: '/workspace', + title: 'Session', + updatedAt: '2026-08-24T00:00:00.000Z', + }, + ], + }; + }, + setConfigOption: async (params: unknown) => { + observations.configurations?.push(params); + return { configOptions: CONFIG_OPTIONS }; + }, + }; +} diff --git a/packages/cli/src/__tests__/acp-child-process-harness.ts b/packages/cli/src/__tests__/acp-child-process-harness.ts index 1f1f5d69e9..7704917345 100644 --- a/packages/cli/src/__tests__/acp-child-process-harness.ts +++ b/packages/cli/src/__tests__/acp-child-process-harness.ts @@ -18,7 +18,7 @@ */ import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'; -import { mkdtemp, mkdir, rm } from 'node:fs/promises'; +import { lstat, mkdtemp, mkdir, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { PassThrough, Readable, Writable } from 'node:stream'; @@ -34,6 +34,7 @@ import { startExecutionRuntimeHostService, type RuntimeHostKernel, } from '@maka/runtime-host/server'; +import { STORAGE_ROOT_MARKER_FILE } from '@maka/storage/root-authority'; import { deriveMakaDataRoots, resolveMakaClientDataRoot } from '../workspace-root.js'; const DEFAULT_TIMEOUT_MS = 15_000; @@ -117,6 +118,16 @@ export class AcpChildProcessHarness { return Buffer.concat(this.#stderr).toString('utf8'); } + async hasRuntimeHostRootMarker(): Promise { + try { + await lstat(join(this.#workspaceRoot, STORAGE_ROOT_MARKER_FILE)); + return true; + } catch (error) { + if (isErrorWithCode(error, 'ENOENT')) return false; + throw error; + } + } + async withClient( operation: (client: AcpChildProcessClient) => Promise | T, configureClient: ConfigureAcpClient = (app) => app, @@ -515,3 +526,7 @@ class StartupTimeoutError extends Error {} function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } + +function isErrorWithCode(error: unknown, code: string): boolean { + return error instanceof Error && 'code' in error && error.code === code; +} diff --git a/packages/cli/src/__tests__/acp-child-process.test.ts b/packages/cli/src/__tests__/acp-child-process.test.ts index b7b7aebe50..213dcb3d28 100644 --- a/packages/cli/src/__tests__/acp-child-process.test.ts +++ b/packages/cli/src/__tests__/acp-child-process.test.ts @@ -19,6 +19,7 @@ import assert from 'node:assert/strict'; import { once } from 'node:events'; +import { realpath } from 'node:fs/promises'; import { PassThrough } from 'node:stream'; import { describe, test } from 'node:test'; import { RequestError, methods } from '@agentclientprotocol/sdk'; @@ -118,19 +119,14 @@ describe('Maka ACP child process', () => { await harness.withClient(async ({ context }) => { assert.deepEqual(await context.request(methods.agent.initialize, { protocolVersion: 1 }), { protocolVersion: 1, - agentCapabilities: {}, + agentCapabilities: { sessionCapabilities: { list: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }); - - await assert.rejects( - context.request('session/new', { cwd: harness.workspaceRoot }), - (error: unknown) => { - assert.ok(error instanceof RequestError); - assert.equal(error.code, -32601); - assert.deepEqual(error.data, { method: 'session/new' }); - return true; - }, + assert.equal( + await harness.hasRuntimeHostRootMarker(), + false, + 'initialize must not begin Runtime Host discovery or candidate startup', ); }); @@ -139,13 +135,94 @@ describe('Maka ACP child process', () => { assert.equal(harness.stderr, ''); const lines = harness.stdout.split(/\r?\n/u).filter((line) => line.trim().length > 0); - assert.ok(lines.length >= 2, 'expected initialize and method-not-found responses'); + assert.ok(lines.length >= 1, 'expected initialize response'); for (const line of lines) { const message: unknown = JSON.parse(line); assertJsonRpcMessage(message); } }); }); + + test('serves multiple ACP Sessions through a real Runtime Host', { + timeout: 30_000, + }, async () => { + await withAcpChildProcessHarness( + async (harness) => { + await harness.withClient(async ({ context }) => { + await context.request(methods.agent.initialize, { protocolVersion: 1 }); + const first = await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }); + assert.deepEqual( + first.configOptions?.map(({ id, currentValue }) => [id, currentValue]), + [ + ['permission_mode', 'ask'], + ['thinking_level', 'default'], + ['collaboration_mode', 'agent'], + ['orchestration_mode', 'default'], + ], + ); + const second = await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }); + assert.notEqual(first.sessionId, second.sessionId); + const configured = await context.request(methods.agent.session.setConfigOption, { + sessionId: first.sessionId, + configId: 'collaboration_mode', + value: 'plan', + }); + assert.deepEqual( + configured.configOptions.map(({ id, currentValue }) => [id, currentValue]), + [ + ['permission_mode', 'ask'], + ['thinking_level', 'default'], + ['collaboration_mode', 'plan'], + ['orchestration_mode', 'default'], + ], + ); + const listed = await context.request(methods.agent.session.list, { + cwd: harness.workspaceRoot, + }); + assert.deepEqual( + new Set(listed.sessions.map((session) => session.sessionId)), + new Set([first.sessionId, second.sessionId]), + ); + const hostCwd = await realpath(harness.workspaceRoot); + assert.equal( + listed.sessions.every((session) => session.cwd === hostCwd), + true, + ); + + await assert.rejects( + context.request(methods.agent.session.close, { sessionId: first.sessionId }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32601); + assert.deepEqual(error.data, { method: 'session/close' }); + return true; + }, + ); + }); + + await harness.closeStdin(); + assert.deepEqual(await harness.waitForExit(), { code: 0, signal: null }); + assert.equal(harness.stderr, ''); + + const lines = harness.stdout.split(/\r?\n/u).filter((line) => line.trim().length > 0); + assert.ok( + lines.length >= 6, + 'expected initialize, new, configuration, list, and method responses', + ); + for (const line of lines) { + const message: unknown = JSON.parse(line); + assertJsonRpcMessage(message); + } + }, + { startRuntimeHost: true }, + ); + }); }); function assertJsonRpcMessage(message: unknown): void { diff --git a/packages/cli/src/__tests__/acp-session-configuration.test.ts b/packages/cli/src/__tests__/acp-session-configuration.test.ts new file mode 100644 index 0000000000..7b4f56d1f8 --- /dev/null +++ b/packages/cli/src/__tests__/acp-session-configuration.test.ts @@ -0,0 +1,249 @@ +/* + * 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 { describe, test } from 'node:test'; +import type { SessionConfigOption, SetSessionConfigOptionRequest } from '@agentclientprotocol/sdk'; +import type { SessionCatalogProjection } from '@maka/runtime-host/protocol'; +import { + AcpSessionConfigInputError, + applyAcpSessionConfigOption, + projectAcpSessionConfigOptions, + validateAcpSessionConfigOptionRequest, +} from '../acp/session-configuration.js'; + +describe('ACP Session configuration', () => { + test('projects the exact ordered select option contract', () => { + assert.deepEqual(projectAcpSessionConfigOptions(catalogSession()), [ + selectOption('permission_mode', 'Permission mode', '_maka/permission_mode', 'ask', [ + ['explore', 'Explore'], + ['ask', 'Ask'], + ['bypass', 'Bypass'], + ]), + selectOption('thinking_level', 'Thinking level', 'thought_level', 'default', [ + ['default', 'Default'], + ['off', 'Off'], + ['minimal', 'Minimal'], + ['low', 'Low'], + ['medium', 'Medium'], + ['high', 'High'], + ['xhigh', 'Extra high'], + ['max', 'Max'], + ]), + selectOption('collaboration_mode', 'Collaboration mode', 'mode', 'agent', [ + ['agent', 'Agent'], + ['plan', 'Plan'], + ]), + selectOption( + 'orchestration_mode', + 'Orchestration mode', + '_maka/orchestration_mode', + 'default', + [ + ['default', 'Default'], + ['swarm', 'Swarm'], + ['graph', 'Graph'], + ], + ), + ]); + }); + + test('keeps default and explicit thinking levels distinct', () => { + for (const [thinkingLevel, expected] of [ + [undefined, 'default'], + ['off', 'off'], + ['minimal', 'minimal'], + ['low', 'low'], + ['medium', 'medium'], + ['high', 'high'], + ['xhigh', 'xhigh'], + ['max', 'max'], + ] as const) { + const option = projectAcpSessionConfigOptions(catalogSession({ thinkingLevel })).find( + ({ id }) => id === 'thinking_level', + ); + assert.equal(option?.currentValue, expected); + } + }); + + test('projects every mutable Runtime Host configuration value', () => { + for (const [field, values] of [ + ['permissionMode', ['explore', 'ask', 'bypass']], + ['collaborationMode', ['agent', 'plan']], + ['orchestrationMode', ['default', 'swarm', 'graph']], + ] as const) { + const optionId = field.replace(/[A-Z]/g, (letter) => `_${letter.toLowerCase()}`); + for (const value of values) { + const option = projectAcpSessionConfigOptions(catalogSession({ [field]: value })).find( + ({ id }) => id === optionId, + ); + assert.equal(option?.currentValue, value); + } + } + }); + + test('maps every legal request value and preserves unrelated configuration', () => { + const base = catalogSession({ + thinkingLevel: 'medium', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'swarm', + }); + const cases: Array< + readonly [SetSessionConfigOptionRequest, Partial>] + > = [ + [request('permission_mode', 'explore'), { permissionMode: 'explore' }], + [request('permission_mode', 'ask'), { permissionMode: 'ask' }], + [request('permission_mode', 'bypass'), { permissionMode: 'bypass' }], + [request('thinking_level', 'default'), { thinkingLevel: null }], + [request('thinking_level', 'off'), { thinkingLevel: 'off' }], + [request('thinking_level', 'minimal'), { thinkingLevel: 'minimal' }], + [request('thinking_level', 'low'), { thinkingLevel: 'low' }], + [request('thinking_level', 'medium'), { thinkingLevel: 'medium' }], + [request('thinking_level', 'high'), { thinkingLevel: 'high' }], + [request('thinking_level', 'xhigh'), { thinkingLevel: 'xhigh' }], + [request('thinking_level', 'max'), { thinkingLevel: 'max' }], + [request('collaboration_mode', 'agent'), { collaborationMode: 'agent' }], + [request('collaboration_mode', 'plan'), { collaborationMode: 'plan' }], + [request('orchestration_mode', 'default'), { orchestrationMode: 'default' }], + [request('orchestration_mode', 'swarm'), { orchestrationMode: 'swarm' }], + [request('orchestration_mode', 'graph'), { orchestrationMode: 'graph' }], + ]; + + for (const [input, patch] of cases) { + assert.deepEqual(applyAcpSessionConfigOption(base, input), { + ...expectedConfiguration(base), + ...patch, + }); + } + }); + + test('preserves an explicit model target', () => { + const session = catalogSession({ + connectionLocked: true, + llmConnectionSlug: 'openai-work', + model: 'gpt-5.6', + }); + + assert.deepEqual( + applyAcpSessionConfigOption(session, request('thinking_level', 'high')).modelTarget, + { kind: 'explicit', connectionSlug: 'openai-work', model: 'gpt-5.6' }, + ); + }); + + test('rejects unknown ids, boolean payloads, and unsupported values', () => { + const cases: Array< + readonly [SetSessionConfigOptionRequest, 'configId' | 'value', 'unsupported' | 'invalid_type'] + > = [ + [request('unknown', 'ask'), 'configId', 'unsupported'], + [ + { + sessionId: 'session-1', + configId: 'permission_mode', + type: 'boolean', + value: true, + }, + 'value', + 'invalid_type', + ], + [request('permission_mode', 'future'), 'value', 'unsupported'], + ]; + + for (const [input, field, reason] of cases) { + for (const invoke of [ + () => validateAcpSessionConfigOptionRequest(input), + () => applyAcpSessionConfigOption(catalogSession(), input), + ]) { + assert.throws(invoke, (error: unknown) => { + assert.ok(error instanceof AcpSessionConfigInputError); + assert.equal(error.field, field); + assert.equal(error.reason, reason); + return true; + }); + } + } + }); +}); + +function selectOption( + id: string, + name: string, + category: string, + currentValue: string, + options: ReadonlyArray, +): SessionConfigOption { + return { + type: 'select', + id, + name, + category, + currentValue, + options: options.map(([value, optionName]) => ({ value, name: optionName })), + }; +} + +function request(configId: string, value: string): SetSessionConfigOptionRequest { + return { sessionId: 'session-1', configId, value }; +} + +function expectedConfiguration(session: SessionCatalogProjection) { + return { + modelTarget: session.connectionLocked + ? { + kind: 'explicit' as const, + connectionSlug: session.llmConnectionSlug, + model: session.model, + } + : { kind: 'default' as const }, + thinkingLevel: session.thinkingLevel ?? null, + permissionMode: session.permissionMode, + collaborationMode: session.collaborationMode, + orchestrationMode: session.orchestrationMode, + }; +} + +function catalogSession( + overrides: Partial = {}, +): SessionCatalogProjection { + return { + id: 'session-1', + revision: 1, + workspace: { + target: { kind: 'host_path', path: '/workspace' }, + hostCwd: '/workspace', + }, + createdAt: 1, + activityAt: 1, + name: 'Session', + isFlagged: false, + isArchived: false, + labels: [], + labelsTruncated: false, + hasUnread: false, + status: 'active', + backend: 'ai-sdk', + llmConnectionSlug: 'default', + connectionLocked: false, + model: 'default', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + ...overrides, + }; +} diff --git a/packages/cli/src/__tests__/acp-session-registry.test.ts b/packages/cli/src/__tests__/acp-session-registry.test.ts new file mode 100644 index 0000000000..7662a982eb --- /dev/null +++ b/packages/cli/src/__tests__/acp-session-registry.test.ts @@ -0,0 +1,1551 @@ +/* + * 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 { mkdtemp, mkdir, realpath, rm, symlink } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, test } from 'node:test'; +import { RequestError, type NewSessionRequest } from '@agentclientprotocol/sdk'; +import { + RuntimeHostOperationError, + RuntimeHostRequestInterruptedError, + RuntimeHostSubscriptionError, +} from '@maka/runtime-host/client'; +import { + SESSION_CATALOG_CWD_MAX_BYTES, + SESSION_CONTINUITY_SCHEMA_VERSION, + type SessionCatalogProjection, + type SessionContinuitySnapshot, + type SubscriptionFrame, +} from '@maka/runtime-host/protocol'; +import { projectAcpSessionConfigOptions } from '../acp/session-configuration.js'; +import { AcpSessionRegistry, type AcpSessionRegistryConnection } from '../acp/session-registry.js'; + +const SESSION_REVISION = `sha256:${'a'.repeat(64)}` as const; +const NEW_SESSION_REVISION = `sha256:${'b'.repeat(64)}` as const; +type TestableSubscription = Awaited< + ReturnType +>; + +describe('ACP Session registry', () => { + test('does not connect when disposed before a Session method is used', async () => { + let connectCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + return fakeConnection(); + }, + }); + + await registry.dispose(); + await registry.dispose(); + + assert.equal(connectCalls, 0); + }); + + test('does not start a queued connection after disposal begins', async () => { + let connectCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + return fakeConnection(); + }, + }); + + const list = registry.list({}); + const dispose = registry.dispose(); + + await assert.rejects( + list, + (error: unknown) => + error instanceof RequestError && + error.code === -32603 && + (error.data as { code?: string }).code === 'registry_closed', + ); + await dispose; + assert.equal(connectCalls, 0); + }); + + test('aborts an in-flight connection before disposal waits for it', async () => { + let connectSignal: AbortSignal | undefined; + const registry = new AcpSessionRegistry({ + connect: async (signal) => { + connectSignal = signal; + return new Promise>((_, reject) => { + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + }); + }, + }); + + const list = registry.list({}); + await waitFor(() => connectSignal !== undefined); + const dispose = registry.dispose(); + + await assert.rejects( + list, + (error: unknown) => + error instanceof RequestError && + error.code === -32603 && + (error.data as { code?: string }).code === 'registry_closed', + ); + await dispose; + assert.equal(connectSignal?.aborted, true); + }); + + test('shares one in-flight connection across concurrent Session methods', async () => { + const connecting = deferred>(); + let connectCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + return connecting.promise; + }, + newSessionId: () => 'session-concurrent', + }); + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + const list = registry.list({}); + await waitFor(() => connectCalls === 1); + + connecting.resolve( + fakeConnection({ + request: async (operation) => + operation === 'session.catalog.query' + ? { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: null, + } + : catalogSession('session-concurrent', '/workspace'), + }), + ); + + const concurrentSession = catalogSession('session-concurrent', '/workspace'); + assert.deepEqual(await create, { + sessionId: 'session-concurrent', + configOptions: projectAcpSessionConfigOptions(concurrentSession), + }); + assert.deepEqual(await list, { sessions: [] }); + assert.equal(connectCalls, 1); + await registry.dispose(); + }); + + test('reports a stable connection error and retries on a later Session request', async () => { + let connectCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + if (connectCalls === 1) throw new Error('Host unavailable'); + return fakeConnection({ + request: async () => ({ + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: null, + }), + }); + }, + }); + + await assert.rejects(registry.list({}), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'connect', + code: 'connection_failed', + }); + return true; + }); + assert.deepEqual(await registry.list({}), { sessions: [] }); + assert.equal(connectCalls, 2); + await registry.dispose(); + }); + + test('closes a connection that resolves after disposal starts', async () => { + const connecting = deferred>(); + let connectCalls = 0; + let closeCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + return connecting.promise; + }, + }); + const list = registry.list({}); + await waitFor(() => connectCalls === 1); + const dispose = registry.dispose(); + + connecting.resolve( + fakeConnection({ + close: async () => { + closeCalls += 1; + }, + }), + ); + + await assert.rejects(list, (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.equal((error.data as { code?: string }).code, 'registry_closed'); + return true; + }); + await dispose; + assert.equal(closeCalls, 1); + }); + + test('creates exact Host sessions and continuously tracks isolated subscription snapshots', async () => { + const subscriptions = new Map(); + const requests: Array<{ operation: string; input: unknown }> = []; + let nextId = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + requests.push({ operation, input }); + const create = input as { + sessionId: string; + workspace: { kind: 'host_path'; path: string }; + }; + return catalogSession(create.sessionId, create.workspace.path); + }, + open: async ({ sessionId }) => { + const subscription = new TestSubscription(sessionId); + subscriptions.set(sessionId, subscription); + return subscription; + }, + }), + newSessionId: () => `session-${++nextId}`, + }); + + assert.deepEqual( + await registry.create({ + cwd: '/workspace/one', + mcpServers: [], + additionalDirectories: [], + _meta: { projectId: 'must-be-ignored' }, + }), + { + sessionId: 'session-1', + configOptions: projectAcpSessionConfigOptions( + catalogSession('session-1', '/workspace/one'), + ), + }, + ); + assert.deepEqual(await registry.create({ cwd: '/workspace/two', mcpServers: [] }), { + sessionId: 'session-2', + configOptions: projectAcpSessionConfigOptions(catalogSession('session-2', '/workspace/two')), + }); + assert.deepEqual(requests, [ + { + operation: 'session.create', + input: { + sessionId: 'session-1', + workspace: { kind: 'host_path', path: '/workspace/one' }, + modelTarget: { kind: 'default' }, + }, + }, + { + operation: 'session.create', + input: { + sessionId: 'session-2', + workspace: { kind: 'host_path', path: '/workspace/two' }, + modelTarget: { kind: 'default' }, + }, + }, + ]); + + const first = subscriptions.get('session-1')!; + const second = subscriptions.get('session-2')!; + for (let revision = 2; revision <= 40; revision += 1) { + first.push(projectionFrame(first, snapshot('session-1', revision), revision)); + } + second.push(projectionFrame(second, snapshot('session-2', 7), 2)); + await waitFor(() => registry.inspect('session-1')?.snapshot.projectionRevision === 40); + await waitFor(() => registry.inspect('session-2')?.snapshot.projectionRevision === 7); + + const inspection = registry.inspect('session-1')!; + assert.equal(inspection.failure, undefined); + inspection.snapshot.session.status = 'running'; + assert.equal(registry.inspect('session-1')?.snapshot.session.status, 'active'); + assert.equal(first.nextCalls >= 40, true, 'the consumer keeps an iterator read pending'); + + await registry.dispose(); + await registry.dispose(); + assert.equal(first.closeCalls, 1); + assert.equal(second.closeCalls, 1); + }); + + test('rejects non-owned Sessions and invalid configuration before Host I/O', async () => { + let connects = 0; + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connects += 1; + return fakeConnection({ + request: async () => { + requests += 1; + return {}; + }, + }); + }, + }); + + await assert.rejects( + registry.setConfigOption({ + sessionId: 'not-owned', + configId: 'permission_mode', + value: 'ask', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32602); + assert.deepEqual(error.data, { reason: 'unknown_session' }); + return true; + }, + ); + assert.equal(connects, 0); + assert.equal(requests, 0); + await registry.dispose(); + }); + + test('updates one option with an exact full configuration', async () => { + const created = catalogSession('session-configured', '/workspace'); + const current = catalogSession('session-configured', '/workspace', 'Configured', 2, { + revision: 4, + orchestrationMode: 'swarm', + }); + const committed = catalogSession('session-configured', '/workspace', 'Configured', 3, { + revision: 5, + permissionMode: 'bypass', + orchestrationMode: 'swarm', + }); + const updates: unknown[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return created; + if (operation === 'session.catalog.query') { + assert.deepEqual(input, { kind: 'get', sessionId: 'session-configured' }); + return { kind: 'session', session: current }; + } + assert.equal(operation, 'session.configuration.update'); + updates.push(input); + return { kind: 'committed', session: committed }; + }, + }), + newSessionId: () => 'session-configured', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + assert.deepEqual( + await registry.setConfigOption({ + sessionId: 'session-configured', + configId: 'permission_mode', + value: 'bypass', + }), + { configOptions: projectAcpSessionConfigOptions(committed) }, + ); + assert.deepEqual(updates, [ + { + sessionId: 'session-configured', + expectedRevision: 4, + configuration: { + modelTarget: { kind: 'default' }, + thinkingLevel: null, + permissionMode: 'bypass', + collaborationMode: 'agent', + orchestrationMode: 'swarm', + }, + }, + ]); + await registry.dispose(); + }); + + test('rereads after a revision conflict and preserves a concurrent change', async () => { + const created = catalogSession('session-conflict', '/workspace'); + const reads = [ + catalogSession('session-conflict', '/workspace', 'Conflict', 2, { revision: 4 }), + catalogSession('session-conflict', '/workspace', 'Conflict', 3, { + revision: 5, + thinkingLevel: 'high', + }), + ]; + const updates: Array<{ expectedRevision: number; configuration: unknown }> = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return created; + if (operation === 'session.catalog.query') { + return { kind: 'session', session: reads.shift()! }; + } + const update = input as { expectedRevision: number; configuration: unknown }; + updates.push(update); + if (updates.length === 1) { + return { kind: 'revision_conflict', expectedRevision: 4, actualRevision: 5 }; + } + return { + kind: 'committed', + session: catalogSession('session-conflict', '/workspace', 'Conflict', 4, { + revision: 6, + thinkingLevel: 'high', + permissionMode: 'bypass', + }), + }; + }, + }), + newSessionId: () => 'session-conflict', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await registry.setConfigOption({ + sessionId: 'session-conflict', + configId: 'permission_mode', + value: 'bypass', + }); + assert.deepEqual( + updates.map(({ expectedRevision, configuration }) => ({ + expectedRevision, + thinkingLevel: (configuration as { thinkingLevel: unknown }).thinkingLevel, + })), + [ + { expectedRevision: 4, thinkingLevel: null }, + { expectedRevision: 5, thinkingLevel: 'high' }, + ], + ); + await registry.dispose(); + }); + + test('concurrent different-field updates converge without overwriting either change', async () => { + const created = catalogSession('session-concurrent-config', '/workspace'); + let current = created; + let reads = 0; + const firstReads = deferred(); + const expectedRevisions: number[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return created; + if (operation === 'session.catalog.query') { + reads += 1; + if (reads === 2) firstReads.resolve(); + if (reads <= 2) { + await firstReads.promise; + return { kind: 'session', session: created }; + } + return { kind: 'session', session: current }; + } + const update = input as { + expectedRevision: number; + configuration: { + thinkingLevel: SessionCatalogProjection['thinkingLevel'] | null; + permissionMode: SessionCatalogProjection['permissionMode']; + collaborationMode: SessionCatalogProjection['collaborationMode']; + orchestrationMode: SessionCatalogProjection['orchestrationMode']; + }; + }; + expectedRevisions.push(update.expectedRevision); + if (update.expectedRevision !== current.revision) { + return { + kind: 'revision_conflict', + expectedRevision: update.expectedRevision, + actualRevision: current.revision, + }; + } + current = { + ...current, + revision: current.revision + 1, + thinkingLevel: update.configuration.thinkingLevel ?? undefined, + permissionMode: update.configuration.permissionMode, + collaborationMode: update.configuration.collaborationMode, + orchestrationMode: update.configuration.orchestrationMode, + }; + return { kind: 'committed', session: current }; + }, + }), + newSessionId: () => 'session-concurrent-config', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await Promise.all([ + registry.setConfigOption({ + sessionId: 'session-concurrent-config', + configId: 'collaboration_mode', + value: 'plan', + }), + registry.setConfigOption({ + sessionId: 'session-concurrent-config', + configId: 'orchestration_mode', + value: 'swarm', + }), + ]); + + assert.equal(current.collaborationMode, 'plan'); + assert.equal(current.orchestrationMode, 'swarm'); + assert.deepEqual(expectedRevisions, [1, 1, 2]); + await registry.dispose(); + }); + + test('stops after three consecutive configuration revision conflicts', async () => { + const created = catalogSession('session-hot', '/workspace'); + let reads = 0; + let updates = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return created; + if (operation === 'session.catalog.query') { + reads += 1; + return { + kind: 'session', + session: catalogSession('session-hot', '/workspace', 'Hot', reads, { + revision: reads, + }), + }; + } + updates += 1; + return { + kind: 'revision_conflict', + expectedRevision: updates, + actualRevision: updates + 1, + }; + }, + }), + newSessionId: () => 'session-hot', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await assert.rejects( + registry.setConfigOption({ + sessionId: 'session-hot', + configId: 'collaboration_mode', + value: 'plan', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'revision_conflict', + attempts: 3, + }); + return true; + }, + ); + assert.equal(reads, 3); + assert.equal(updates, 3); + await registry.dispose(); + }); + + test('rejects invalid configuration for an owned Session before catalog I/O', async () => { + const created = catalogSession('session-validation', '/workspace'); + let requestsAfterCreate = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return created; + requestsAfterCreate += 1; + return {}; + }, + }), + newSessionId: () => 'session-validation', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + for (const [input, field, reason] of [ + [ + { sessionId: 'session-validation', configId: 'unknown', value: 'ask' }, + 'configId', + 'unsupported', + ], + [ + { + sessionId: 'session-validation', + configId: 'permission_mode', + type: 'boolean', + value: true, + }, + 'value', + 'invalid_type', + ], + [ + { sessionId: 'session-validation', configId: 'permission_mode', value: 'future' }, + 'value', + 'unsupported', + ], + ] as const) { + await assert.rejects(registry.setConfigOption(input), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32602); + assert.deepEqual(error.data, { field, reason }); + return true; + }); + } + assert.equal(requestsAfterCreate, 0); + await registry.dispose(); + }); + + test('maps Runtime Host configuration update failures to stable ACP errors', async () => { + for (const [hostError, acpCode, expectedData] of [ + [ + new RuntimeHostOperationError( + 'session.configuration.update', + 'invalid_request', + 'invalid configuration', + ), + -32602, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'invalid_request', + }, + ], + [ + new RuntimeHostOperationError( + 'session.configuration.update', + 'not_found', + 'missing Session', + ), + -32602, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'not_found', + }, + ], + [ + new RuntimeHostOperationError( + 'session.configuration.update', + 'session_busy', + 'active Turn', + ), + -32603, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'session_busy', + }, + ], + [ + new RuntimeHostOperationError( + 'session.configuration.update', + 'commit_outcome_unknown', + 'unknown outcome', + ), + -32603, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'commit_outcome_unknown', + }, + ], + [ + new RuntimeHostRequestInterruptedError( + 'session.configuration.update', + 'control', + 'not_dispatched', + 'connection_lost', + ), + -32603, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'request_interrupted', + reason: 'connection_lost', + dispatch: 'not_dispatched', + }, + ], + [ + RequestError.invalidParams({ source: 'untrusted_dependency' }, 'spoofed error'), + -32603, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'internal_failure', + }, + ], + ] as const) { + const created = catalogSession('session-error', '/workspace'); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return created; + if (operation === 'session.catalog.query') { + return { kind: 'session', session: created }; + } + throw hostError; + }, + }), + newSessionId: () => 'session-error', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await assert.rejects( + registry.setConfigOption({ + sessionId: 'session-error', + configId: 'permission_mode', + value: 'bypass', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, acpCode); + assert.deepEqual(error.data, expectedData); + return true; + }, + ); + await registry.dispose(); + } + }); + + test('rejects missing and legacy Session catalog lookups deterministically', async () => { + for (const [session, acpCode, expectedCode] of [ + [null, -32602, 'not_found'], + [ + { + kind: 'unsupported_legacy_record', + id: 'session-catalog-error', + revision: 1, + reason: 'not_wire_representable', + }, + -32603, + 'unsupported_session_projection', + ], + ] as const) { + const created = catalogSession('session-catalog-error', '/workspace'); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return created; + return { kind: 'session', session }; + }, + }), + newSessionId: () => 'session-catalog-error', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await assert.rejects( + registry.setConfigOption({ + sessionId: 'session-catalog-error', + configId: 'permission_mode', + value: 'bypass', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, acpCode); + assert.equal((error.data as { code?: string }).code, expectedCode); + return true; + }, + ); + await registry.dispose(); + } + }); + + test('disposal waits for an in-flight configuration update and closes once', async () => { + const created = catalogSession('session-dispose-update', '/workspace'); + const updating = deferred(); + let updateCalls = 0; + let connectionCloses = 0; + const subscription = new TestSubscription('session-dispose-update'); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return created; + if (operation === 'session.catalog.query') { + return { kind: 'session', session: created }; + } + updateCalls += 1; + return updating.promise; + }, + open: async () => subscription, + close: async () => { + connectionCloses += 1; + }, + }), + newSessionId: () => 'session-dispose-update', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const update = registry.setConfigOption({ + sessionId: 'session-dispose-update', + configId: 'collaboration_mode', + value: 'plan', + }); + await waitFor(() => updateCalls === 1); + + let disposed = false; + const dispose = registry.dispose().then(() => { + disposed = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(disposed, false); + assert.equal(connectionCloses, 1); + updating.resolve({ + kind: 'committed', + session: catalogSession('session-dispose-update', '/workspace', 'Disposed', 2, { + revision: 2, + collaborationMode: 'plan', + }), + }); + + await update; + await dispose; + assert.equal(subscription.closeCalls, 1); + assert.equal(connectionCloses, 1); + }); + + test('disposal during a catalog read prevents a later configuration update', async () => { + const created = catalogSession('session-dispose-read', '/workspace'); + const reading = deferred(); + let readCalls = 0; + let updateCalls = 0; + let connectionCloses = 0; + const subscription = new TestSubscription('session-dispose-read'); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return created; + if (operation === 'session.catalog.query') { + readCalls += 1; + return reading.promise; + } + updateCalls += 1; + return { kind: 'committed', session: created }; + }, + open: async () => subscription, + close: async () => { + connectionCloses += 1; + }, + }), + newSessionId: () => 'session-dispose-read', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const update = registry.setConfigOption({ + sessionId: 'session-dispose-read', + configId: 'collaboration_mode', + value: 'plan', + }); + await waitFor(() => readCalls === 1); + + let disposed = false; + const dispose = registry.dispose().then(() => { + disposed = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(disposed, false); + assert.equal(connectionCloses, 1); + reading.resolve({ kind: 'session', session: created }); + + await assert.rejects(update, (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'registry_closed', + }); + return true; + }); + await dispose; + assert.equal(updateCalls, 0); + assert.equal(subscription.closeCalls, 1); + assert.equal(connectionCloses, 1); + }); + + test('records subscription failures without producing an unhandled rejection', async () => { + const subscription = new TestSubscription('session-1'); + const registry = new AcpSessionRegistry({ + connect: async () => fakeConnection({ open: async () => subscription }), + newSessionId: () => 'session-1', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + subscription.fail( + new RuntimeHostSubscriptionError('slow_consumer', 'consumer exceeded its queue'), + ); + await waitFor(() => registry.inspect('session-1')?.failure !== undefined); + assert.deepEqual(registry.inspect('session-1')?.failure, { + source: 'runtime_host', + operation: 'subscription.consume', + code: 'subscription_failure', + reason: 'slow_consumer', + }); + + await registry.dispose(); + }); + + test('records an unexpected clean subscription end as a failure', async () => { + const subscription = new TestSubscription('session-1'); + const registry = new AcpSessionRegistry({ + connect: async () => fakeConnection({ open: async () => subscription }), + newSessionId: () => 'session-1', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + subscription.end(); + await waitFor(() => registry.inspect('session-1')?.failure !== undefined); + assert.deepEqual(registry.inspect('session-1')?.failure, { + source: 'runtime_host', + operation: 'subscription.consume', + code: 'subscription_closed', + }); + await registry.dispose(); + }); + + test('rejects unsupported creation inputs before touching Runtime Host', async () => { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return {}; + }, + }), + }); + + const cases: Array = [ + [ + 'mcpServers', + { + cwd: '/workspace', + mcpServers: [{ name: 'server', command: 'server', args: [], env: [] }], + }, + ], + [ + 'additionalDirectories', + { + cwd: '/workspace', + mcpServers: [], + additionalDirectories: ['/other'], + }, + ], + ['cwd', { cwd: 'relative', mcpServers: [] }], + [ + 'cwd', + { + cwd: `/${'x'.repeat(SESSION_CATALOG_CWD_MAX_BYTES)}`, + mcpServers: [], + }, + ], + ]; + for (const [field, input] of cases) { + await assert.rejects( + registry.create(input), + (error: unknown) => + error instanceof RequestError && + error.code === -32602 && + (error.data as { field?: string }).field === field, + ); + } + assert.equal(requests, 0); + await registry.dispose(); + }); + + test('reports a durable session identity when subscription opening fails without rollback', async () => { + const operations: string[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + operations.push(operation); + return catalogSession('session-durable', '/workspace'); + }, + open: async () => { + throw new RuntimeHostOperationError( + 'subscription.open', + 'operation_unavailable', + 'subscription unavailable', + ); + }, + }), + newSessionId: () => 'session-durable', + }); + + await assert.rejects( + registry.create({ cwd: '/workspace', mcpServers: [] }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'subscription.open', + code: 'operation_unavailable', + sessionId: 'session-durable', + durableSessionCreated: true, + }); + return true; + }, + ); + assert.equal(registry.inspect('session-durable'), undefined); + assert.deepEqual(operations, ['session.create']); + await registry.dispose(); + }); + + test('keeps failed and outcome-unknown creates distinct from known durable sessions', async () => { + for (const [hostCode, acpCode] of [ + ['invalid_request', -32602], + ['commit_outcome_unknown', -32603], + ] as const) { + let opens = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + throw new RuntimeHostOperationError('session.create', hostCode, 'create failed'); + }, + open: async ({ sessionId }) => { + opens += 1; + return new TestSubscription(sessionId); + }, + }), + newSessionId: () => `session-${hostCode}`, + }); + + await assert.rejects( + registry.create({ cwd: '/workspace', mcpServers: [] }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, acpCode); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.create', + code: hostCode, + sessionId: `session-${hostCode}`, + }); + return true; + }, + ); + assert.equal(opens, 0); + await registry.dispose(); + } + }); + + test('closes a subscription that opens after disposal starts and never registers it', async () => { + const opening = deferred(); + const subscription = new TestSubscription('session-race'); + let opens = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + open: async () => { + opens += 1; + return opening.promise; + }, + }), + newSessionId: () => 'session-race', + }); + + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + await waitFor(() => opens === 1); + const dispose = registry.dispose(); + opening.resolve(subscription); + + await assert.rejects( + create, + (error: unknown) => + error instanceof RequestError && + error.code === -32603 && + (error.data as { code?: string }).code === 'registry_closed', + ); + await dispose; + assert.equal(subscription.closeCalls, 1); + assert.equal(registry.inspect('session-race'), undefined); + }); + + test('connection cleanup interrupts an in-flight open before disposal waits for it', async () => { + const opening = deferred(); + let opens = 0; + let connectionCloses = 0; + const interruption = new RuntimeHostRequestInterruptedError( + 'subscription.open', + 'control', + 'not_dispatched', + 'connection_lost', + ); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + open: async () => { + opens += 1; + return opening.promise; + }, + close: async () => { + connectionCloses += 1; + opening.reject(interruption); + }, + }), + newSessionId: () => 'session-race', + }); + + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + const createRejected = assert.rejects(create, RequestError); + await waitFor(() => opens === 1); + const dispose = registry.dispose(); + try { + await waitFor(() => connectionCloses === 1); + } finally { + opening.reject(interruption); + await Promise.allSettled([createRejected, dispose]); + } + await createRejected; + await dispose; + assert.equal(connectionCloses, 1); + }); + + test('connection cleanup terminates a raced subscription that cannot close', async () => { + const opening = deferred(); + const subscription = new TestSubscription('session-race'); + subscription.closeError = new Error('subscription close failed'); + let opens = 0; + let connectionCloses = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + open: async () => { + opens += 1; + return opening.promise; + }, + close: async () => { + connectionCloses += 1; + subscription.end(); + }, + }), + newSessionId: () => 'session-race', + }); + + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + await waitFor(() => opens === 1); + const dispose = registry.dispose(); + opening.resolve(subscription); + + await assert.rejects(create, RequestError); + await dispose; + assert.equal(subscription.closeCalls, 1); + assert.equal(connectionCloses, 1); + }); + + test('connection cleanup terminates a registered subscription that cannot close', async () => { + const subscription = new TestSubscription('session-1'); + subscription.closeError = new Error('subscription close failed'); + let connectionCloses = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + open: async () => subscription, + close: async () => { + connectionCloses += 1; + subscription.end(); + }, + }), + newSessionId: () => 'session-1', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await registry.dispose(); + assert.equal(subscription.closeCalls, 1); + assert.equal(connectionCloses, 1); + assert.equal(registry.inspect('session-1'), undefined); + }); + + test('maps one filtered Host catalog page per ACP page and carries cwd across pages', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-acp-list-')); + t.after(() => rm(root, { recursive: true, force: true })); + const workspace = join(root, 'workspace'); + const alias = join(root, 'workspace-alias'); + await mkdir(workspace); + await symlink(workspace, alias); + const canonicalWorkspace = await realpath(workspace); + const inputs: unknown[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + assert.equal(operation, 'session.catalog.query'); + inputs.push(input); + if ((input as { kind: string }).kind === 'list_start') { + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [ + catalogSession('other', join(root, 'other'), 'Other', 1_000), + { + kind: 'unsupported_legacy_record', + id: 'legacy', + revision: 1, + reason: 'not_wire_representable', + }, + ], + nextCursor: 'page-2', + }; + } + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [ + catalogSession('matching', canonicalWorkspace, 'Matching session', 2_000), + catalogSession( + 'undated', + canonicalWorkspace, + 'Out-of-range activity', + Number.MAX_SAFE_INTEGER, + ), + ], + nextCursor: null, + }; + }, + }), + }); + + const first = await registry.list({ cwd: alias }); + assert.deepEqual(first.sessions, []); + assert.equal(typeof first.nextCursor, 'string'); + const second = await registry.list({ cursor: first.nextCursor }); + assert.deepEqual(second, { + sessions: [ + { + sessionId: 'matching', + cwd: canonicalWorkspace, + title: 'Matching session', + updatedAt: '1970-01-01T00:00:02.000Z', + }, + { + sessionId: 'undated', + cwd: canonicalWorkspace, + title: 'Out-of-range activity', + }, + ], + }); + assert.deepEqual(inputs, [ + { kind: 'list_start' }, + { kind: 'list_continue', revision: SESSION_REVISION, cursor: 'page-2' }, + ]); + await registry.dispose(); + }); + + test('rejects a cursor reused with a different normalized cwd before Host I/O', async () => { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: 'page-2', + }; + }, + }), + }); + const first = await registry.list({ cwd: '/workspace/one/../one' }); + + await assert.rejects( + registry.list({ cwd: '/workspace/two', cursor: first.nextCursor }), + (error: unknown) => + error instanceof RequestError && + error.code === -32602 && + (error.data as { reason?: string }).reason === 'cursor_cwd_mismatch', + ); + assert.equal(requests, 1); + await registry.dispose(); + }); + + test('rejects malformed and oversized ACP cursors as invalid params', async () => { + const registry = new AcpSessionRegistry({ + connect: async () => fakeConnection(), + }); + const invalidRevisionCursor = Buffer.from( + JSON.stringify({ + v: 1, + revision: 'sha256:bad', + cursor: 'page-2', + cwd: null, + }), + 'utf8', + ).toString('base64url'); + for (const cursor of ['not-a-cursor', 'x'.repeat(8 * 1024 + 1), invalidRevisionCursor]) { + await assert.rejects( + registry.list({ cursor }), + (error: unknown) => + error instanceof RequestError && + error.code === -32602 && + (error.data as { reason?: string }).reason === 'invalid_cursor', + ); + } + await registry.dispose(); + }); + + test('translates stale and repeated Host cursors into stable ACP errors', async () => { + for (const [nextResult, expectedCode, expectedReason] of [ + [ + { + kind: 'revision_changed', + expectedRevision: SESSION_REVISION, + actualRevision: NEW_SESSION_REVISION, + }, + -32602, + 'stale_cursor', + ], + [ + { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: 'page-2', + }, + -32603, + 'repeated_cursor', + ], + ] as const) { + let first = true; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + if (!first) return nextResult; + first = false; + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: 'page-2', + }; + }, + }), + }); + const page = await registry.list({}); + await assert.rejects(registry.list({ cursor: page.nextCursor }), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, expectedCode); + assert.equal((error.data as { reason?: string; code?: string }).reason, expectedReason); + return true; + }); + await registry.dispose(); + } + }); + + test('maps Runtime Host invalid_request from session/list to invalid params', async () => { + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + throw new RuntimeHostOperationError( + 'session.catalog.query', + 'invalid_request', + 'invalid query', + ); + }, + }), + }); + + await assert.rejects(registry.list({}), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32602); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'invalid_request', + }); + return true; + }); + await registry.dispose(); + }); +}); + +class TestSubscription implements TestableSubscription { + readonly hostEpoch = 'host-1'; + readonly subscriptionId: string; + readonly snapshot: SessionContinuitySnapshot; + readonly #frames: SubscriptionFrame[] = []; + #waiting: ReturnType>> | undefined; + #failure: Error | undefined; + #done = false; + closeError: Error | undefined; + closeCalls = 0; + nextCalls = 0; + + constructor(sessionId: string) { + this.subscriptionId = `subscription-${sessionId}`; + this.snapshot = snapshot(sessionId, 1); + } + + [Symbol.asyncIterator](): AsyncIterator { + return { next: () => this.next() }; + } + + next(): Promise> { + this.nextCalls += 1; + const frame = this.#frames.shift(); + if (frame) return Promise.resolve({ done: false, value: frame }); + if (this.#failure) return Promise.reject(this.#failure); + if (this.#done) return Promise.resolve({ done: true, value: undefined }); + assert.equal(this.#waiting, undefined, 'only one iterator read may be pending'); + this.#waiting = deferred>(); + return this.#waiting.promise; + } + + push(frame: SubscriptionFrame): void { + if (this.#waiting) { + const waiting = this.#waiting; + this.#waiting = undefined; + waiting.resolve({ done: false, value: frame }); + return; + } + this.#frames.push(frame); + } + + fail(error: Error): void { + this.#failure = error; + this.#waiting?.reject(error); + this.#waiting = undefined; + } + + end(): void { + this.#done = true; + this.#waiting?.resolve({ done: true, value: undefined }); + this.#waiting = undefined; + } + + async close(): Promise { + this.closeCalls += 1; + if (this.closeError) throw this.closeError; + this.end(); + } +} + +function fakeConnection( + overrides: { + request?: (operation: string, input: unknown) => Promise; + open?: (input: { sessionId: string }) => Promise; + close?: () => Promise; + } = {}, +): AcpSessionRegistryConnection { + return { + request: + overrides.request ?? + (async (operation, input) => { + if (operation !== 'session.create') return {}; + const create = input as { + sessionId: string; + workspace: { kind: 'host_path'; path: string }; + }; + return catalogSession(create.sessionId, create.workspace.path); + }), + openSessionSubscriptionOnce: + overrides.open ?? (async ({ sessionId }) => new TestSubscription(sessionId)), + close: overrides.close ?? (async () => undefined), + } as AcpSessionRegistryConnection; +} + +function snapshot(sessionId: string, projectionRevision: number): SessionContinuitySnapshot { + return { + schemaVersion: SESSION_CONTINUITY_SCHEMA_VERSION, + session: { + sessionId, + metadataRevision: 1, + status: 'active', + createdAt: 1, + isArchived: false, + }, + projectionRevision, + rootTurn: null, + goal: null, + queue: { + hostEpoch: 'host-1', + queueRevision: 0, + steering: [], + followup: [], + }, + interactions: { pending: [] }, + }; +} + +function projectionFrame( + subscription: TestSubscription, + next: SessionContinuitySnapshot, + sequence: number, +): SubscriptionFrame { + return { + kind: 'subscription.session_projection', + hostEpoch: subscription.hostEpoch, + subscriptionId: subscription.subscriptionId, + sequence, + snapshot: next, + }; +} + +function catalogSession( + id: string, + cwd: string, + name = id, + activityAt = 1, + overrides: Partial = {}, +): SessionCatalogProjection { + return { + id, + revision: 1, + workspace: { target: { kind: 'host_path', path: cwd }, hostCwd: cwd }, + createdAt: 1, + activityAt, + name, + isFlagged: false, + isArchived: false, + labels: [], + labelsTruncated: false, + hasUnread: false, + status: 'active', + backend: 'ai-sdk', + llmConnectionSlug: 'default', + connectionLocked: false, + model: 'default', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + ...overrides, + }; +} + +async function waitFor(predicate: () => boolean): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) return; + await new Promise((resolve) => setImmediate(resolve)); + } + assert.fail('condition was not reached'); +} + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} diff --git a/packages/cli/src/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index a1feb5bcfd..a4ba52311c 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -20,10 +20,19 @@ import assert from 'node:assert/strict'; import { Readable, Writable } from 'node:stream'; import { describe, test } from 'node:test'; +import type { + RuntimeHostConnection, + RuntimeHostSessionSubscription, +} from '@maka/runtime-host/client'; +import { + SESSION_CONTINUITY_SCHEMA_VERSION, + type SessionCatalogProjection, +} from '@maka/runtime-host/protocol'; +import { projectAcpSessionConfigOptions } from '../acp/session-configuration.js'; import { runMakaAcpStdioServer } from '../acp/stdio-server.js'; describe('Maka ACP stdio server', () => { - test('answers initialize without Runtime Host input or dependencies', async () => { + test('answers initialize without connecting a Runtime Host', async () => { const harness = createHarness([ `${JSON.stringify({ jsonrpc: '2.0', @@ -40,18 +49,20 @@ describe('Maka ACP stdio server', () => { id: 1, result: { protocolVersion: 1, - agentCapabilities: {}, + agentCapabilities: { sessionCapabilities: { list: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }, }, ]); + assert.equal(harness.connectCalls(), 0); }); - test('returns zero after normal EOF', async () => { + test('returns zero after normal EOF without connecting a Runtime Host', async () => { const harness = createHarness([]); assert.equal(await harness.run(), 0); + assert.equal(harness.connectCalls(), 0); }); test('returns a JSON-RPC parse error and then zero after EOF', async () => { @@ -74,10 +85,160 @@ describe('Maka ACP stdio server', () => { await assert.rejects(harness.run(), (error: unknown) => error === transportError); }); + + test('disposes ACP subscriptions before closing the lazily acquired Host connection', async () => { + const lifecycle: string[] = []; + const connection = { + request: async (operation: string, input: unknown) => { + assert.equal(operation, 'session.create'); + const create = input as { + sessionId: string; + workspace: { kind: 'host_path'; path: string }; + }; + return catalogSession(create.sessionId, create.workspace.path); + }, + openSessionSubscriptionOnce: async ({ sessionId }: { sessionId: string }) => + closingSubscription(sessionId, lifecycle), + close: async () => { + lifecycle.push('connection.close'); + }, + } as unknown as RuntimeHostConnection; + const harness = createHarness( + [ + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1 }, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'session/new', + params: { cwd: '/workspace', mcpServers: [] }, + })}\n`, + ], + { + connection, + }, + ); + + assert.equal(await harness.run(), 0); + const response = harness + .stdoutMessages() + .find((message) => (message as { id?: unknown }).id === 2) as { + jsonrpc?: unknown; + id?: unknown; + result?: { sessionId?: unknown; configOptions?: unknown[] }; + }; + assert.equal(response.jsonrpc, '2.0'); + assert.equal(response.id, 2); + assert.equal(typeof response.result?.sessionId, 'string'); + assert.deepEqual( + response.result?.configOptions, + projectAcpSessionConfigOptions( + catalogSession(response.result?.sessionId as string, '/workspace'), + ), + ); + assert.equal(harness.connectCalls(), 1); + assert.deepEqual(lifecycle, ['subscription.close', 'connection.close']); + }); + + test('returns a Host connection failure from the Session request and keeps serving ACP', async () => { + const harness = createHarness( + [ + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1 }, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'session/list', + params: {}, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 3, + method: 'session/close', + params: { sessionId: 'missing' }, + })}\n`, + ], + { connectError: new Error('Host unavailable') }, + ); + + assert.equal(await harness.run(), 0); + const responses = new Map( + harness + .stdoutMessages() + .map((message) => [(message as { id?: unknown }).id, message] as const), + ); + const connectionFailure = responses.get(2) as { + error?: { code?: unknown; data?: unknown }; + }; + assert.equal(connectionFailure.error?.code, -32603); + assert.deepEqual(connectionFailure.error?.data, { + source: 'runtime_host', + operation: 'connect', + code: 'connection_failed', + }); + const methodFailure = responses.get(3) as { + error?: { code?: unknown; data?: unknown }; + }; + assert.equal(methodFailure.error?.code, -32601); + assert.deepEqual(methodFailure.error?.data, { method: 'session/close' }); + assert.equal(harness.connectCalls(), 1); + }); + + test('keeps an unimplemented Session method Host-independent', async () => { + const harness = createHarness([ + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1 }, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'session/close', + params: { sessionId: 'missing' }, + })}\n`, + ]); + + assert.equal(await harness.run(), 0); + const response = harness + .stdoutMessages() + .find((message) => (message as { id?: unknown }).id === 2) as { + error?: { code?: unknown; data?: unknown }; + }; + assert.equal(response.error?.code, -32601); + assert.deepEqual(response.error?.data, { method: 'session/close' }); + assert.equal(harness.connectCalls(), 0); + }); }); -function createHarness(chunks: string[], options: { readonly stdin?: Readable } = {}) { +function createHarness( + chunks: string[], + options: { + readonly stdin?: Readable; + readonly connection?: RuntimeHostConnection; + readonly connectError?: Error; + } = {}, +) { const stdin = options.stdin ?? Readable.from(chunks.map((chunk) => Buffer.from(chunk))); + let connects = 0; + const connection = + options.connection ?? + ({ + request: async () => ({ kind: 'unsupported_legacy_record' }), + openSessionSubscriptionOnce: async () => { + throw new Error('subscription is not available in this fixture'); + }, + close: async () => undefined, + } as unknown as RuntimeHostConnection); const stdoutChunks: Buffer[] = []; const stdout = new Writable({ write(chunk, _encoding, callback) { @@ -86,7 +247,25 @@ function createHarness(chunks: string[], options: { readonly stdin?: Readable } }, }); return { - run: () => runMakaAcpStdioServer({ version: '0.2.0' }, { stdin, stdout }), + run: () => + runMakaAcpStdioServer( + { workspaceRoot: '/workspace', clientDataRoot: '/client-data', version: '0.2.0' }, + { + stdin, + stdout, + connectRuntimeHostCli: async () => { + connects += 1; + if (options.connectError) throw options.connectError; + return { + connection, + close: () => connection.close(), + } as Awaited< + ReturnType + >; + }, + }, + ), + connectCalls: () => connects, stdoutMessages: () => Buffer.concat(stdoutChunks) .toString('utf8') @@ -96,3 +275,70 @@ function createHarness(chunks: string[], options: { readonly stdin?: Readable } .map((line) => JSON.parse(line) as unknown), }; } + +function catalogSession(id: string, cwd: string): SessionCatalogProjection { + return { + id, + revision: 1, + workspace: { target: { kind: 'host_path', path: cwd }, hostCwd: cwd }, + createdAt: 1, + activityAt: 1, + name: id, + isFlagged: false, + isArchived: false, + labels: [], + labelsTruncated: false, + hasUnread: false, + status: 'active', + backend: 'ai-sdk', + llmConnectionSlug: 'default', + connectionLocked: false, + model: 'default', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }; +} + +function closingSubscription( + sessionId: string, + lifecycle: string[], +): RuntimeHostSessionSubscription { + let finish!: () => void; + const closed = new Promise((resolve) => { + finish = resolve; + }); + return { + hostEpoch: 'host-1', + subscriptionId: `subscription-${sessionId}`, + snapshot: { + schemaVersion: SESSION_CONTINUITY_SCHEMA_VERSION, + session: { + sessionId, + metadataRevision: 1, + status: 'active', + createdAt: 1, + isArchived: false, + }, + projectionRevision: 1, + rootTurn: null, + goal: null, + queue: { hostEpoch: 'host-1', queueRevision: 0, steering: [], followup: [] }, + interactions: { pending: [] }, + }, + activeAssistantStreams: [], + transcriptBootstrap: null, + [Symbol.asyncIterator]() { + return { + next: async () => { + await closed; + return { done: true, value: undefined }; + }, + }; + }, + close: async () => { + lifecycle.push('subscription.close'); + finish(); + }, + } as unknown as RuntimeHostSessionSubscription; +} 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 8593f557fa..3ed96565a2 100644 --- a/packages/cli/src/__tests__/runtime-host-cli-context.test.ts +++ b/packages/cli/src/__tests__/runtime-host-cli-context.test.ts @@ -96,6 +96,82 @@ test('CLI Runtime Host bootstrap launches the execution composition', async () = assert.equal(closes, 1); }); +test('CLI Runtime Host bootstrap aborts a stalled catalog read and closes its connection', async () => { + const controller = new AbortController(); + const catalogStarted = deferred(); + const abortReason = new Error('ACP connection closed'); + let closes = 0; + let connectSignal: AbortSignal | undefined; + const connection = { + rootId: 'root-id', + hostEpoch: 'host-epoch', + connectionId: 'connection-id', + selectedProtocol: 0, + closed: new Promise(() => {}), + status: async () => ({ state: 'ready' }), + subscribeConfigurationChanges: () => () => {}, + subscribeProjectCatalogChanges: () => () => {}, + subscribeSessionCatalogChanges: () => () => {}, + subscribeScheduledTaskChanges: () => () => {}, + close: async () => { + closes += 1; + }, + } as unknown as RuntimeHostConnection; + + const connecting = connectRuntimeHostCli( + { rootPath: '/runtime-host-root', signal: controller.signal }, + { + connectOrSpawn: async (input) => { + connectSignal = input.signal; + return { + kind: 'connected', + connection, + registration: hostRegistration(), + }; + }, + readConnectionCatalog: async () => { + catalogStarted.resolve(); + return new Promise(() => {}); + }, + }, + ); + await catalogStarted.promise; + controller.abort(abortReason); + + await assert.rejects(connecting, (error: unknown) => error === abortReason); + assert.equal(connectSignal, controller.signal); + assert.equal(closes, 1); +}); + +test('CLI Runtime Host bootstrap closes an initial connection acquired after abort', async () => { + const controller = new AbortController(); + const connectStarted = deferred(); + const acquired = deferred>(); + const abortReason = new Error('ACP connection closed'); + let closes = 0; + const connection = { + close: async () => { + closes += 1; + }, + } as unknown as RuntimeHostConnection; + + const connecting = connectRuntimeHostCli( + { rootPath: '/runtime-host-root', signal: controller.signal }, + { + connectOrSpawn: async () => { + connectStarted.resolve(); + return acquired.promise; + }, + }, + ); + await connectStarted.promise; + controller.abort(abortReason); + + await assert.rejects(connecting, (error: unknown) => error === abortReason); + acquired.resolve(connectedHostResult(connection)); + await waitFor(() => closes === 1); +}); + test('non-interactive CLI reports how to retire an incompatible Runtime Host', async () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > V0_1_11_HOST_COMPATIBILITY_EPOCH); await assert.rejects( @@ -501,6 +577,14 @@ function hostRegistration( }; } +function connectedHostResult(connection: RuntimeHostConnection) { + return { + kind: 'connected' as const, + connection, + registration: hostRegistration(), + }; +} + function incompatibleRemoteHandshake(overrides: Partial = {}): HostIncompatible { return { kind: 'incompatible', @@ -530,3 +614,21 @@ function singleRemoteProfileCatalog(profile: RemoteRuntimeHostProfile): RuntimeH rebindIfCurrent: async () => assert.fail('unexpected write'), }; } + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +async function waitFor(predicate: () => boolean): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) return; + await new Promise((resolve) => setImmediate(resolve)); + } + assert.fail('condition was not reached'); +} diff --git a/packages/cli/src/acp/maka-acp-agent.ts b/packages/cli/src/acp/maka-acp-agent.ts index 2508c16c3c..11926dc954 100644 --- a/packages/cli/src/acp/maka-acp-agent.ts +++ b/packages/cli/src/acp/maka-acp-agent.ts @@ -18,16 +18,24 @@ */ import { agent, methods, type AgentApp } from '@agentclientprotocol/sdk'; +import type { AcpSessionRegistry } from './session-registry.js'; export interface MakaAcpAgentOptions { readonly version: string; + readonly sessionRegistry: Pick; } export function createMakaAcpAgent(options: MakaAcpAgentOptions): AgentApp { - return agent({ name: 'maka' }).onRequest(methods.agent.initialize, () => ({ - protocolVersion: 1, - agentCapabilities: {}, - authMethods: [], - agentInfo: { name: 'maka', title: 'Maka', version: options.version }, - })); + return agent({ name: 'maka' }) + .onRequest(methods.agent.initialize, () => ({ + protocolVersion: 1, + agentCapabilities: { sessionCapabilities: { list: {} } }, + authMethods: [], + agentInfo: { name: 'maka', title: 'Maka', version: options.version }, + })) + .onRequest(methods.agent.session.new, ({ params }) => options.sessionRegistry.create(params)) + .onRequest(methods.agent.session.list, ({ params }) => options.sessionRegistry.list(params)) + .onRequest(methods.agent.session.setConfigOption, ({ params }) => + options.sessionRegistry.setConfigOption(params), + ); } diff --git a/packages/cli/src/acp/session-configuration.ts b/packages/cli/src/acp/session-configuration.ts new file mode 100644 index 0000000000..eee6a60212 --- /dev/null +++ b/packages/cli/src/acp/session-configuration.ts @@ -0,0 +1,181 @@ +/* + * 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 type { SessionConfigOption, SetSessionConfigOptionRequest } from '@agentclientprotocol/sdk'; +import type { SessionCatalogProjection, SessionConfiguration } from '@maka/runtime-host/protocol'; + +interface AcpSessionConfigSpec { + readonly id: string; + readonly name: string; + readonly category: string; + readonly options: ReadonlyArray; +} + +const PERMISSION_SPEC = { + id: 'permission_mode', + name: 'Permission mode', + category: '_maka/permission_mode', + options: [ + ['explore', 'Explore'], + ['ask', 'Ask'], + ['bypass', 'Bypass'], + ], +} as const satisfies AcpSessionConfigSpec; + +const THINKING_SPEC = { + id: 'thinking_level', + name: 'Thinking level', + category: 'thought_level', + options: [ + ['default', 'Default'], + ['off', 'Off'], + ['minimal', 'Minimal'], + ['low', 'Low'], + ['medium', 'Medium'], + ['high', 'High'], + ['xhigh', 'Extra high'], + ['max', 'Max'], + ], +} as const satisfies AcpSessionConfigSpec; + +const COLLABORATION_SPEC = { + id: 'collaboration_mode', + name: 'Collaboration mode', + category: 'mode', + options: [ + ['agent', 'Agent'], + ['plan', 'Plan'], + ], +} as const satisfies AcpSessionConfigSpec; + +const ORCHESTRATION_SPEC = { + id: 'orchestration_mode', + name: 'Orchestration mode', + category: '_maka/orchestration_mode', + options: [ + ['default', 'Default'], + ['swarm', 'Swarm'], + ['graph', 'Graph'], + ], +} as const satisfies AcpSessionConfigSpec; + +const CONFIG_SPECS = [ + PERMISSION_SPEC, + THINKING_SPEC, + COLLABORATION_SPEC, + ORCHESTRATION_SPEC, +] as const; + +export class AcpSessionConfigInputError extends Error { + readonly name = 'AcpSessionConfigInputError'; + + constructor( + readonly field: 'configId' | 'value', + readonly reason: 'unsupported' | 'invalid_type', + ) { + super(`Invalid ACP Session configuration ${field}`); + } +} + +export function projectAcpSessionConfigOptions( + session: SessionCatalogProjection, +): SessionConfigOption[] { + return [ + configOption(PERMISSION_SPEC, session.permissionMode), + configOption(THINKING_SPEC, session.thinkingLevel ?? 'default'), + configOption(COLLABORATION_SPEC, session.collaborationMode), + configOption(ORCHESTRATION_SPEC, session.orchestrationMode), + ]; +} + +export function validateAcpSessionConfigOptionRequest( + request: SetSessionConfigOptionRequest, +): asserts request is SetSessionConfigOptionRequest & { readonly value: string } { + const spec = configSpec(request.configId); + if (!spec) throw new AcpSessionConfigInputError('configId', 'unsupported'); + if (typeof request.value !== 'string') { + throw new AcpSessionConfigInputError('value', 'invalid_type'); + } + if (!spec.options.some(([value]) => value === request.value)) { + throw new AcpSessionConfigInputError('value', 'unsupported'); + } +} + +export function applyAcpSessionConfigOption( + session: SessionCatalogProjection, + request: SetSessionConfigOptionRequest, +): SessionConfiguration { + validateAcpSessionConfigOptionRequest(request); + const current: SessionConfiguration = { + modelTarget: session.connectionLocked + ? { + kind: 'explicit', + connectionSlug: session.llmConnectionSlug, + model: session.model, + } + : { kind: 'default' }, + thinkingLevel: session.thinkingLevel ?? null, + permissionMode: session.permissionMode, + collaborationMode: session.collaborationMode, + orchestrationMode: session.orchestrationMode, + }; + + switch (request.configId) { + case 'permission_mode': + return { + ...current, + permissionMode: request.value as SessionConfiguration['permissionMode'], + }; + case 'thinking_level': + return { + ...current, + thinkingLevel: + request.value === 'default' + ? null + : (request.value as Exclude), + }; + case 'collaboration_mode': + return { + ...current, + collaborationMode: request.value as SessionConfiguration['collaborationMode'], + }; + case 'orchestration_mode': + return { + ...current, + orchestrationMode: request.value as SessionConfiguration['orchestrationMode'], + }; + default: + throw new AcpSessionConfigInputError('configId', 'unsupported'); + } +} + +function configOption(spec: AcpSessionConfigSpec, currentValue: string): SessionConfigOption { + return { + type: 'select', + id: spec.id, + name: spec.name, + category: spec.category, + currentValue, + options: spec.options.map(([value, name]) => ({ value, name })), + }; +} + +function configSpec(configId: string): AcpSessionConfigSpec | undefined { + return CONFIG_SPECS.find(({ id }) => id === configId); +} diff --git a/packages/cli/src/acp/session-registry.ts b/packages/cli/src/acp/session-registry.ts new file mode 100644 index 0000000000..2d9da2c848 --- /dev/null +++ b/packages/cli/src/acp/session-registry.ts @@ -0,0 +1,682 @@ +/* + * 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 { randomUUID } from 'node:crypto'; +import { realpath } from 'node:fs/promises'; +import { isAbsolute, normalize } from 'node:path'; +import { + RequestError, + type ListSessionsRequest, + type ListSessionsResponse, + type NewSessionRequest, + type NewSessionResponse, + type SetSessionConfigOptionRequest, + type SetSessionConfigOptionResponse, +} from '@agentclientprotocol/sdk'; +import { + readRuntimeHostSessionCatalogPage, + RuntimeHostCatalogReadError, + RuntimeHostOperationError, + RuntimeHostRequestInterruptedError, + RuntimeHostSessionCatalogRevisionChangedError, + RuntimeHostSubscriptionError, + type RuntimeHostConnection, + type RuntimeHostSessionCatalogPageCursor, +} from '@maka/runtime-host/client'; +import { + SESSION_CATALOG_CURSOR_MAX_BYTES, + SESSION_CATALOG_CWD_MAX_BYTES, + type SessionCatalogItem, + type SessionCatalogProjection, + type SessionContinuitySnapshot, + type SubscriptionFrame, + type SubscriptionOpenInput, +} from '@maka/runtime-host/protocol'; +import { + AcpSessionConfigInputError, + applyAcpSessionConfigOption, + projectAcpSessionConfigOptions, + validateAcpSessionConfigOptionRequest, +} from './session-configuration.js'; + +const ACP_SESSION_CURSOR_VERSION = 1 as const; +const ACP_SESSION_CURSOR_MAX_BYTES = 8 * 1024; +const ACP_SESSION_CONFIGURATION_MAX_ATTEMPTS = 3; + +interface AcpSessionSubscription extends AsyncIterable { + readonly snapshot: SessionContinuitySnapshot; + close(): Promise; +} + +export interface AcpSessionRegistryConnection { + readonly request: RuntimeHostConnection['request']; + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise; + close(): Promise; +} + +export interface AcpSessionRegistryOptions { + readonly connect: (signal: AbortSignal) => Promise; + readonly newSessionId?: () => string; +} + +export interface AcpSessionRegistryFailure { + readonly source: 'runtime_host'; + readonly operation: 'subscription.consume'; + readonly code: string; + readonly reason?: string; +} + +export interface AcpSessionRegistryInspection { + readonly snapshot: SessionContinuitySnapshot; + readonly failure?: AcpSessionRegistryFailure; +} + +interface AcpSessionRecord { + readonly sessionId: string; + readonly subscription: AcpSessionSubscription; + snapshot: SessionContinuitySnapshot; + failure?: AcpSessionRegistryFailure; + consumerTask: Promise; + closing: boolean; +} + +/** Owns all Runtime Host resources associated with one ACP connection. */ +export class AcpSessionRegistry { + readonly #connect: (signal: AbortSignal) => Promise; + readonly #newSessionId: () => string; + readonly #records = new Map(); + readonly #inFlightOperations = new Set>(); + #connection: AcpSessionRegistryConnection | undefined; + #connectTask: Promise | undefined; + #connectAbortController: AbortController | undefined; + #closing = false; + #connectionCloseTask: Promise | undefined; + #disposeTask: Promise | undefined; + + constructor(options: AcpSessionRegistryOptions) { + this.#connect = options.connect; + this.#newSessionId = options.newSessionId ?? randomUUID; + } + + async create(params: NewSessionRequest): Promise { + this.#assertOpen(); + validateNewSessionParams(params); + return this.#track(this.#create(params)); + } + + async list(params: ListSessionsRequest): Promise { + this.#assertOpen(); + return this.#track(this.#list(params)); + } + + async setConfigOption( + params: SetSessionConfigOptionRequest, + ): Promise { + this.#assertOpen(); + this.#assertOwnedSession(params.sessionId); + try { + validateAcpSessionConfigOptionRequest(params); + } catch (error) { + throw requestErrorFromConfigInput(error); + } + return this.#track(this.#setConfigOption(params)); + } + + inspect(sessionId: string): AcpSessionRegistryInspection | undefined { + const record = this.#records.get(sessionId); + if (!record) return undefined; + return { + snapshot: structuredClone(record.snapshot), + ...(record.failure ? { failure: { ...record.failure } } : {}), + }; + } + + dispose(): Promise { + this.#closing = true; + this.#connectAbortController?.abort(); + this.#disposeTask ??= this.#dispose(); + return this.#disposeTask; + } + + async #create(params: NewSessionRequest): Promise { + const connection = await this.#getConnection(); + const sessionId = this.#newSessionId(); + let result; + try { + result = await connection.request('session.create', { + sessionId, + workspace: { kind: 'host_path', path: params.cwd }, + modelTarget: { kind: 'default' }, + }); + } catch (error) { + throw requestErrorFromRuntimeHost(error, 'session.create', { sessionId }); + } + const created = requireConfigurableSession(result, 'session.create', { sessionId }); + + let subscription: AcpSessionSubscription; + try { + subscription = await connection.openSessionSubscriptionOnce({ + sessionId, + transcript: { kind: 'none' }, + }); + } catch (error) { + throw RequestError.internalError( + { + ...runtimeHostErrorData(error, 'subscription.open'), + sessionId, + durableSessionCreated: true, + }, + 'Runtime Host subscription could not be opened for the durable session', + ); + } + + if (this.#closing) { + try { + await subscription.close(); + } catch { + await this.#closeOwnedConnection(); + } + throw RequestError.internalError( + { + source: 'runtime_host', + operation: 'subscription.open', + code: 'registry_closed', + sessionId, + durableSessionCreated: true, + }, + 'ACP connection closed while the durable session was being attached', + ); + } + + const record: AcpSessionRecord = { + sessionId, + subscription, + snapshot: structuredClone(subscription.snapshot), + consumerTask: Promise.resolve(), + closing: false, + }; + this.#records.set(sessionId, record); + record.consumerTask = this.#consume(record); + return { sessionId, configOptions: projectAcpSessionConfigOptions(created) }; + } + + async #consume(record: AcpSessionRecord): Promise { + try { + for await (const frame of record.subscription) { + if (frame.kind === 'subscription.session_projection') { + record.snapshot = structuredClone(frame.snapshot); + } + } + if (!record.closing) { + record.failure = { + source: 'runtime_host', + operation: 'subscription.consume', + code: 'subscription_closed', + }; + } + } catch (error) { + if (!record.closing) { + record.failure = runtimeHostSubscriptionFailure(error); + } + } + } + + async #list(params: ListSessionsRequest): Promise { + const cursor = params.cursor == null ? undefined : decodeAcpSessionCursor(params.cursor); + const requestedCwd = params.cwd == null ? undefined : await normalizeCwd(params.cwd); + if (cursor && requestedCwd !== undefined && cursor.cwd !== requestedCwd) { + throw RequestError.invalidParams( + { reason: 'cursor_cwd_mismatch' }, + 'cursor was created for a different cwd filter', + ); + } + const cwd = requestedCwd ?? cursor?.cwd ?? null; + const connection = await this.#getConnection(); + let page; + try { + page = await readRuntimeHostSessionCatalogPage( + connection, + cursor ? { revision: cursor.revision, cursor: cursor.cursor } : undefined, + ); + } catch (error) { + if (error instanceof RuntimeHostSessionCatalogRevisionChangedError) { + throw RequestError.invalidParams( + { reason: 'stale_cursor' }, + 'session catalog changed; restart listing from the first page', + ); + } + throw requestErrorFromRuntimeHost(error, 'session.catalog.query'); + } + + const sessions = page.sessions.flatMap((session) => { + if ('kind' in session || (cwd !== null && session.workspace.hostCwd !== cwd)) return []; + const updatedAt = isoTimestamp(session.activityAt); + return [ + { + sessionId: session.id, + cwd: session.workspace.hostCwd, + title: session.name, + ...(updatedAt ? { updatedAt } : {}), + }, + ]; + }); + return { + sessions, + ...(page.nextCursor + ? { nextCursor: encodeAcpSessionCursor({ ...page.nextCursor, cwd }) } + : {}), + }; + } + + async #setConfigOption( + params: SetSessionConfigOptionRequest, + ): Promise { + const connection = await this.#getConnection(); + for (let attempt = 0; attempt < ACP_SESSION_CONFIGURATION_MAX_ATTEMPTS; attempt += 1) { + const current = await this.#getConfigurableSession(connection, params.sessionId); + if (this.#closing) throw registryClosedError('session.configuration.update'); + let result; + try { + result = await connection.request('session.configuration.update', { + sessionId: params.sessionId, + expectedRevision: current.revision, + configuration: applyAcpSessionConfigOption(current, params), + }); + } catch (error) { + throw requestErrorFromRuntimeHost(error, 'session.configuration.update'); + } + if (result.kind === 'committed') { + const committed = requireConfigurableSession( + result.session, + 'session.configuration.update', + ); + return { configOptions: projectAcpSessionConfigOptions(committed) }; + } + } + throw RequestError.internalError( + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'revision_conflict', + attempts: ACP_SESSION_CONFIGURATION_MAX_ATTEMPTS, + }, + 'Session configuration kept changing', + ); + } + + async #getConfigurableSession( + connection: AcpSessionRegistryConnection, + sessionId: string, + ): Promise { + let result; + try { + result = await connection.request('session.catalog.query', { kind: 'get', sessionId }); + } catch (error) { + throw requestErrorFromRuntimeHost(error, 'session.catalog.query'); + } + if (result.kind !== 'session') { + throw RequestError.internalError( + { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'invalid_catalog_result', + }, + 'Runtime Host returned an invalid Session lookup', + ); + } + if (result.session === null) { + throw RequestError.invalidParams( + { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'not_found', + }, + 'Session does not exist', + ); + } + return requireConfigurableSession(result.session, 'session.catalog.query'); + } + + async #dispose(): Promise { + const records = [...this.#records.values()]; + for (const record of records) record.closing = true; + const subscriptionCloses = records.map((record) => + Promise.resolve().then(() => record.subscription.close()), + ); + const connectionClose = this.#closeOwnedConnection(); + await Promise.allSettled([...subscriptionCloses, connectionClose]); + await Promise.allSettled([...this.#inFlightOperations]); + await Promise.allSettled(records.map((record) => record.consumerTask)); + this.#records.clear(); + } + + #closeOwnedConnection(): Promise { + const connection = this.#connection; + const connectTask = this.#connectTask; + if (!connection && !connectTask) return Promise.resolve(); + this.#connectionCloseTask ??= connection + ? Promise.resolve().then(() => connection.close()) + : connectTask!.then( + (connected) => connected.close(), + () => undefined, + ); + return this.#connectionCloseTask; + } + + async #getConnection(): Promise { + this.#assertOpen(); + if (this.#connection) return this.#connection; + let connectController = this.#connectAbortController; + if (!this.#connectTask) { + connectController = new AbortController(); + this.#connectAbortController = connectController; + this.#connectTask = Promise.resolve().then(() => { + if (this.#closing) throw registryClosedError('connect'); + connectController!.signal.throwIfAborted(); + return this.#connect(connectController!.signal); + }); + } + const connectTask = this.#connectTask; + let connection: AcpSessionRegistryConnection; + try { + connection = await connectTask; + } catch { + if (this.#connectTask === connectTask) this.#connectTask = undefined; + if (this.#connectAbortController === connectController) { + this.#connectAbortController = undefined; + } + if (this.#closing) throw registryClosedError('connect'); + throw RequestError.internalError( + { + source: 'runtime_host', + operation: 'connect', + code: 'connection_failed', + }, + 'Runtime Host connection failed', + ); + } + if (this.#connectAbortController === connectController) { + this.#connectAbortController = undefined; + } + if (this.#closing) { + await this.#closeOwnedConnection().catch(() => undefined); + throw registryClosedError('connect'); + } + this.#connection ??= connection; + return this.#connection; + } + + async #track(operation: Promise): Promise { + this.#inFlightOperations.add(operation); + try { + return await operation; + } finally { + this.#inFlightOperations.delete(operation); + } + } + + #assertOpen(): void { + if (!this.#closing) return; + throw registryClosedError('subscription.open'); + } + + #assertOwnedSession(sessionId: string): void { + if (this.#records.has(sessionId)) return; + throw RequestError.invalidParams( + { reason: 'unknown_session' }, + 'Session is not attached to this ACP connection', + ); + } +} + +function registryClosedError( + operation: 'connect' | 'subscription.open' | 'session.configuration.update', +): RequestError { + return RequestError.internalError( + { source: 'runtime_host', operation, code: 'registry_closed' }, + 'ACP session registry is closed', + ); +} + +function requireConfigurableSession( + session: SessionCatalogItem, + operation: 'session.create' | 'session.catalog.query' | 'session.configuration.update', + extra: Record = {}, +): SessionCatalogProjection { + if ('kind' in session) { + throw RequestError.internalError( + { + source: 'runtime_host', + operation, + code: 'unsupported_session_projection', + ...extra, + }, + 'Runtime Host returned an unsupported Session projection', + ); + } + return session; +} + +function requestErrorFromConfigInput(error: unknown): RequestError { + if (error instanceof AcpSessionConfigInputError) { + return RequestError.invalidParams( + { field: error.field, reason: error.reason }, + `Session configuration ${error.field} is ${error.reason}`, + ); + } + return RequestError.internalError( + { source: 'acp_adapter', operation: 'session/set_config_option', code: 'internal_failure' }, + 'Session configuration validation failed', + ); +} + +function validateNewSessionParams(params: NewSessionRequest): void { + assertBoundedAbsoluteCwd(params.cwd); + if (params.mcpServers.length > 0) { + throw RequestError.invalidParams( + { field: 'mcpServers', reason: 'unsupported' }, + 'MCP servers are not supported by this ACP adapter yet', + ); + } + if ((params.additionalDirectories?.length ?? 0) > 0) { + throw RequestError.invalidParams( + { field: 'additionalDirectories', reason: 'unsupported' }, + 'Additional directories are not supported by this ACP adapter yet', + ); + } +} + +function requestErrorFromRuntimeHost( + error: unknown, + operation: 'session.create' | 'session.catalog.query' | 'session.configuration.update', + extra: Record = {}, +): RequestError { + const data = { ...runtimeHostErrorData(error, operation), ...extra }; + if ( + error instanceof RuntimeHostOperationError && + (error.code === 'invalid_request' || error.code === 'not_found') + ) { + return RequestError.invalidParams(data, 'Runtime Host rejected the request'); + } + return RequestError.internalError(data, 'Runtime Host request failed'); +} + +function runtimeHostErrorData(error: unknown, operation: string): Record { + if (error instanceof RuntimeHostOperationError) { + return { + source: 'runtime_host', + operation: error.operation, + code: error.code, + }; + } + if (error instanceof RuntimeHostRequestInterruptedError) { + return { + source: 'runtime_host', + operation: error.operation, + code: 'request_interrupted', + reason: error.reason, + dispatch: error.dispatch, + }; + } + if (error instanceof RuntimeHostSubscriptionError) { + return { + source: 'runtime_host', + operation, + code: 'subscription_failure', + reason: error.reason, + }; + } + if (error instanceof RuntimeHostCatalogReadError) { + return { + source: 'runtime_host', + operation, + code: 'catalog_read_failure', + reason: error.reason, + }; + } + return { source: 'runtime_host', operation, code: 'internal_failure' }; +} + +function runtimeHostSubscriptionFailure(error: unknown): AcpSessionRegistryFailure { + if (error instanceof RuntimeHostSubscriptionError) { + return { + source: 'runtime_host', + operation: 'subscription.consume', + code: 'subscription_failure', + reason: error.reason, + }; + } + return { + source: 'runtime_host', + operation: 'subscription.consume', + code: 'internal_failure', + }; +} + +interface AcpSessionCursor extends RuntimeHostSessionCatalogPageCursor { + readonly v: typeof ACP_SESSION_CURSOR_VERSION; + readonly cwd: string | null; +} + +function encodeAcpSessionCursor( + cursor: RuntimeHostSessionCatalogPageCursor & { readonly cwd: string | null }, +): string { + const encoded = Buffer.from( + JSON.stringify({ v: ACP_SESSION_CURSOR_VERSION, ...cursor }), + 'utf8', + ).toString('base64url'); + if (Buffer.byteLength(encoded, 'utf8') > ACP_SESSION_CURSOR_MAX_BYTES) { + throw RequestError.internalError( + { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'cursor_too_large', + }, + 'Runtime Host cursor cannot be represented safely in ACP', + ); + } + return encoded; +} + +function decodeAcpSessionCursor(encoded: string): AcpSessionCursor { + try { + if (encoded.length === 0 || Buffer.byteLength(encoded, 'utf8') > ACP_SESSION_CURSOR_MAX_BYTES) { + throw new Error('cursor size is invalid'); + } + const decoded = Buffer.from(encoded, 'base64url'); + if (decoded.toString('base64url') !== encoded) throw new Error('cursor encoding is invalid'); + const value: unknown = JSON.parse(decoded.toString('utf8')); + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('cursor body is invalid'); + } + const record = value as Record; + if ( + Object.keys(record).length !== 4 || + record.v !== ACP_SESSION_CURSOR_VERSION || + typeof record.revision !== 'string' || + !/^sha256:[0-9a-f]{64}$/.test(record.revision) || + typeof record.cursor !== 'string' || + record.cursor.length === 0 || + Buffer.byteLength(record.cursor, 'utf8') > SESSION_CATALOG_CURSOR_MAX_BYTES || + !validCursorCwd(record.cwd) + ) { + throw new Error('cursor fields are invalid'); + } + return { + v: ACP_SESSION_CURSOR_VERSION, + revision: record.revision as RuntimeHostSessionCatalogPageCursor['revision'], + cursor: record.cursor, + cwd: record.cwd, + }; + } catch { + throw RequestError.invalidParams({ reason: 'invalid_cursor' }, 'cursor is invalid'); + } +} + +function validCursorCwd(value: unknown): value is string | null { + return ( + value === null || + (typeof value === 'string' && + isAbsolute(value) && + normalize(value) === value && + Buffer.byteLength(value, 'utf8') <= SESSION_CATALOG_CWD_MAX_BYTES) + ); +} + +async function normalizeCwd(cwd: string): Promise { + assertBoundedAbsoluteCwd(cwd); + const lexical = normalize(cwd); + try { + return await realpath(lexical); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === 'ENOENT' || code === 'ENOTDIR') return lexical; + throw RequestError.internalError( + { + source: 'filesystem', + operation: 'cwd.realpath', + code: code ?? 'internal_failure', + }, + 'cwd could not be canonicalized', + ); + } +} + +function assertBoundedAbsoluteCwd(cwd: string): void { + if (!isAbsolute(cwd)) { + throw RequestError.invalidParams( + { field: 'cwd', reason: 'must_be_absolute' }, + 'cwd must be an absolute path', + ); + } + if (Buffer.byteLength(cwd, 'utf8') > SESSION_CATALOG_CWD_MAX_BYTES) { + throw RequestError.invalidParams( + { field: 'cwd', reason: 'too_large' }, + 'cwd exceeds the Runtime Host path limit', + ); + } +} + +function isoTimestamp(timestamp: number): string | undefined { + const date = new Date(timestamp); + return Number.isNaN(date.getTime()) ? undefined : date.toISOString(); +} diff --git a/packages/cli/src/acp/stdio-server.ts b/packages/cli/src/acp/stdio-server.ts index e5c5df732b..6b022702c7 100644 --- a/packages/cli/src/acp/stdio-server.ts +++ b/packages/cli/src/acp/stdio-server.ts @@ -19,13 +19,19 @@ import { Readable, Writable } from 'node:stream'; import { ndJsonStream } from '@agentclientprotocol/sdk'; +import type { RuntimeHostConnection } from '@maka/runtime-host/client'; import { createMakaAcpAgent } from './maka-acp-agent.js'; +import { AcpSessionRegistry } from './session-registry.js'; +import { connectRuntimeHostCli } from '../runtime-host-cli-context.js'; export interface MakaAcpStdioServerInput { + readonly workspaceRoot: string; + readonly clientDataRoot: string; readonly version: string; } export interface MakaAcpStdioServerDependencies { + readonly connectRuntimeHostCli?: typeof connectRuntimeHostCli; readonly stdin?: Readable; readonly stdout?: Writable; } @@ -34,6 +40,24 @@ export async function runMakaAcpStdioServer( input: MakaAcpStdioServerInput, dependencies: MakaAcpStdioServerDependencies = {}, ): Promise { + const sessionRegistry = new AcpSessionRegistry({ + connect: async (signal) => { + const context = await (dependencies.connectRuntimeHostCli ?? connectRuntimeHostCli)({ + rootPath: input.workspaceRoot, + clientDataRoot: input.clientDataRoot, + signal, + }); + return { + request: context.connection.request.bind( + context.connection, + ) as RuntimeHostConnection['request'], + openSessionSubscriptionOnce: context.connection.openSessionSubscriptionOnce.bind( + context.connection, + ), + close: () => context.close(), + }; + }, + }); const stdin = dependencies.stdin ?? process.stdin; const stdout = dependencies.stdout ?? process.stdout; let stdioError: Error | undefined; @@ -47,7 +71,10 @@ export async function runMakaAcpStdioServer( Writable.toWeb(stdout) as WritableStream, Readable.toWeb(stdin) as ReadableStream, ); - const connection = createMakaAcpAgent({ version: input.version }).connect(stream); + const connection = createMakaAcpAgent({ + version: input.version, + sessionRegistry, + }).connect(stream); await connection.closed; if (stdioError) { throw stdioError; @@ -56,5 +83,6 @@ export async function runMakaAcpStdioServer( } finally { stdin.off('error', recordStdioError); stdout.off('error', recordStdioError); + await sessionRegistry.dispose(); } } diff --git a/packages/cli/src/cli-core.ts b/packages/cli/src/cli-core.ts index 1c5759b050..95ec108cba 100644 --- a/packages/cli/src/cli-core.ts +++ b/packages/cli/src/cli-core.ts @@ -241,7 +241,11 @@ export async function runMakaCli( } case 'acp': { const { runMakaAcpStdioServer } = await import('./acp/stdio-server.js'); - return runMakaAcpStdioServer({ version }); + return runMakaAcpStdioServer({ + workspaceRoot: dataRoots.workspaceRoot, + clientDataRoot: dataRoots.clientDataRoot, + version, + }); } case 'runtime-host-serve': { const { runRuntimeHostServiceCli } = await import('./runtime-host-service-command.js'); diff --git a/packages/cli/src/runtime-host-cli-context.ts b/packages/cli/src/runtime-host-cli-context.ts index dd4cc4bb38..7934e067f0 100644 --- a/packages/cli/src/runtime-host-cli-context.ts +++ b/packages/cli/src/runtime-host-cli-context.ts @@ -107,6 +107,7 @@ export async function connectRuntimeHostCli( readonly profileId?: string; readonly clientDataRoot?: string; readonly interactiveSsh?: boolean; + readonly signal?: AbortSignal; }, overrides: Partial = {}, ): Promise { @@ -165,27 +166,96 @@ export async function connectRuntimeHostCli( } return connected.connection; }; - const initialConnection = await connect( - undefined, - input.interactiveSsh && process.stdin.isTTY && process.stdout.isTTY ? 'inherit' : 'batch', - ); - const connection = await createRuntimeHostReconnectingConnection({ - initialConnection, - connect: (signal) => connect(signal, 'batch'), - }); + let initialConnection: RuntimeHostConnection | undefined; + let connection: RuntimeHostConnection | undefined; try { + initialConnection = await acquireAbortably( + () => + connect( + input.signal, + input.interactiveSsh && process.stdin.isTTY && process.stdout.isTTY ? 'inherit' : 'batch', + ), + input.signal, + ); + connection = await createRuntimeHostReconnectingConnection({ + initialConnection, + connect: (signal) => connect(signal, 'batch'), + }); + initialConnection = undefined; + const catalog = await runAbortably(() => deps.readConnectionCatalog(connection!), input.signal); + const connected = connection; return { - connection, - catalog: await deps.readConnectionCatalog(connection), + connection: connected, + catalog, profile, - close: () => connection.close(), + close: () => connected.close(), }; } catch (error) { - await connection.close().catch(() => undefined); + await (connection ?? initialConnection)?.close().catch(() => undefined); throw error; } } +function acquireAbortably }>( + operation: () => Promise, + signal?: AbortSignal, +): Promise { + if (!signal) return operation(); + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + let settled = false; + const settle = (callback: () => void) => { + if (settled) return false; + settled = true; + signal.removeEventListener('abort', onAbort); + callback(); + return true; + }; + const onAbort = () => settle(() => reject(signal.reason)); + signal.addEventListener('abort', onAbort, { once: true }); + let running: Promise; + try { + running = operation(); + } catch (error) { + settle(() => reject(error)); + return; + } + void running.then( + (value) => { + if (!settle(() => resolve(value))) void value.close().catch(() => undefined); + }, + (error: unknown) => settle(() => reject(error)), + ); + }); +} + +function runAbortably(operation: () => Promise, signal?: AbortSignal): Promise { + if (!signal) return operation(); + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + let settled = false; + const settle = (callback: () => void) => { + if (settled) return; + settled = true; + signal.removeEventListener('abort', onAbort); + callback(); + }; + const onAbort = () => settle(() => reject(signal.reason)); + signal.addEventListener('abort', onAbort, { once: true }); + let running: Promise; + try { + running = operation(); + } catch (error) { + settle(() => reject(error)); + return; + } + void running.then( + (value) => settle(() => resolve(value)), + (error: unknown) => settle(() => reject(error)), + ); + }); +} + async function resolveHostProfile( input: { readonly profileId?: string; readonly clientDataRoot?: string }, deps: RuntimeHostCliContextDeps, diff --git a/packages/runtime-host/src/__tests__/catalog-reader.test.ts b/packages/runtime-host/src/__tests__/catalog-reader.test.ts index f35516651b..4d69b33af8 100644 --- a/packages/runtime-host/src/__tests__/catalog-reader.test.ts +++ b/packages/runtime-host/src/__tests__/catalog-reader.test.ts @@ -22,13 +22,108 @@ import test from 'node:test'; import type { RuntimeHostConnection } from '../client/connection.js'; import { RuntimeHostCatalogReadError, + RuntimeHostSessionCatalogRevisionChangedError, readRuntimeHostConnectionCatalog, readRuntimeHostProjectDetails, readRuntimeHostProjects, + readRuntimeHostSessionCatalogPage, readRuntimeHostSessions, readRuntimeHostSkillCatalog, } from '../client/catalog-reader.js'; +test('reads one Session catalog page and carries its revision into the continuation cursor', async () => { + const inputs: Record[] = []; + const connection = fakeConnection(async (operation, input) => { + assert.equal(operation, 'session.catalog.query'); + inputs.push(input); + const continuation = input.kind === 'list_continue'; + return { + kind: 'page', + revision: 'sha256:sessions', + sessions: [ + { + kind: 'unsupported_legacy_record', + id: continuation ? 'legacy-2' : 'legacy-1', + revision: 1, + reason: 'not_wire_representable', + }, + ], + nextCursor: continuation ? null : 'page-2', + }; + }); + + const first = await readRuntimeHostSessionCatalogPage(connection); + assert.deepEqual(first, { + revision: 'sha256:sessions', + sessions: [ + { + kind: 'unsupported_legacy_record', + id: 'legacy-1', + revision: 1, + reason: 'not_wire_representable', + }, + ], + nextCursor: { revision: 'sha256:sessions', cursor: 'page-2' }, + }); + assert.deepEqual(await readRuntimeHostSessionCatalogPage(connection, first.nextCursor!), { + revision: 'sha256:sessions', + sessions: [ + { + kind: 'unsupported_legacy_record', + id: 'legacy-2', + revision: 1, + reason: 'not_wire_representable', + }, + ], + nextCursor: null, + }); + assert.deepEqual(inputs, [ + { kind: 'list_start' }, + { kind: 'list_continue', revision: 'sha256:sessions', cursor: 'page-2' }, + ]); +}); + +test('reports a changed Session catalog revision through a typed page-reader error', async () => { + const connection = fakeConnection(async () => ({ + kind: 'revision_changed', + expectedRevision: 'sha256:old', + actualRevision: 'sha256:new', + })); + + await assert.rejects( + () => + readRuntimeHostSessionCatalogPage(connection, { + revision: 'sha256:old', + cursor: 'page-2', + }), + (error) => + error instanceof RuntimeHostSessionCatalogRevisionChangedError && + error.expectedRevision === 'sha256:old' && + error.actualRevision === 'sha256:new', + ); +}); + +test('rejects a Session page reader cursor that does not advance', async () => { + const connection = fakeConnection(async () => ({ + kind: 'page', + revision: 'sha256:sessions', + sessions: [], + nextCursor: 'page-2', + })); + + await assert.rejects( + () => + readRuntimeHostSessionCatalogPage(connection, { + revision: 'sha256:sessions', + cursor: 'page-2', + }), + (error) => + error instanceof RuntimeHostCatalogReadError && + error.catalog === 'session' && + error.reason === 'repeated_cursor', + ); +}); + test('waits out a burst of Session catalog revisions', async () => { let starts = 0; const connection = fakeConnection(async (operation, input) => { diff --git a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts index a13b13a39a..9260b5da00 100644 --- a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts +++ b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts @@ -262,6 +262,85 @@ test('a Session observation reopens safely after its first connection starts dra await connection.close(); }); +test('a one-shot Session observation preserves the first open failure without reconnecting', async () => { + const first = connectionHarness( + 'first', + () => undefined, + async () => { + throw new RuntimeHostOperationError( + 'subscription.open', + 'host_draining', + 'Runtime Host is draining', + ); + }, + ); + const replacement = connectionHarness( + 'replacement', + () => undefined, + async () => ({ subscriptionId: 'replacement-subscription' }), + ); + let reconnectAttempts = 0; + const connection = await createRuntimeHostReconnectingConnection({ + initialConnection: first.connection, + connect: async () => { + reconnectAttempts += 1; + return replacement.connection; + }, + }); + + await assert.rejects( + connection.openSessionSubscriptionOnce({ + sessionId: 'session-1', + transcript: { kind: 'none' }, + }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'host_draining', + ); + assert.equal(first.openedSubscriptions, 1); + assert.equal(replacement.openedSubscriptions, 0); + assert.equal(reconnectAttempts, 0); + await connection.close(); +}); + +test('a one-shot Session observation composes through a decorated connection', async () => { + const expected = new RuntimeHostOperationError( + 'subscription.open', + 'host_draining', + 'Runtime Host is draining', + ); + const base = connectionHarness('base', () => undefined); + let retryCapableCalls = 0; + let oneShotCalls = 0; + const decorated = { + ...base.connection, + openSessionSubscription: async () => { + retryCapableCalls += 1; + throw new Error('retry-capable open must not be selected'); + }, + openSessionSubscriptionOnce: async () => { + oneShotCalls += 1; + throw expected; + }, + } as RuntimeHostConnection; + const connection = await createRuntimeHostReconnectingConnection({ + initialConnection: decorated, + connect: async () => { + throw new Error('reconnect must not be attempted'); + }, + }); + + await assert.rejects( + connection.openSessionSubscriptionOnce({ + sessionId: 'session-1', + transcript: { kind: 'none' }, + }), + (error: unknown) => error === expected, + ); + assert.equal(retryCapableCalls, 0); + assert.equal(oneShotCalls, 1); + await connection.close(); +}); + test('a reconnecting Client rejects a different Host composition permanently', async () => { const first = connectionHarness('first', () => undefined); const replacement = connectionHarness('replacement', () => undefined, undefined, { @@ -585,6 +664,10 @@ function connectionHarness( openedSubscriptions += 1; return openSubscription(); }, + openSessionSubscriptionOnce: async () => { + openedSubscriptions += 1; + return openSubscription(); + }, subscribeConfigurationChanges: () => () => {}, subscribeProjectCatalogChanges: () => () => {}, subscribeSessionCatalogChanges: () => () => {}, diff --git a/packages/runtime-host/src/client/catalog-reader.ts b/packages/runtime-host/src/client/catalog-reader.ts index e176dc64b3..bc7bb98842 100644 --- a/packages/runtime-host/src/client/catalog-reader.ts +++ b/packages/runtime-host/src/client/catalog-reader.ts @@ -26,6 +26,7 @@ import { type RelayModelProfile, type RelayModelProfiles, type SessionCatalogItem, + type SessionCatalogRevision, type SkillCatalogWorkspaceContext, type SkillCatalogInvocableItem, type SkillCatalogInvocableTarget, @@ -79,6 +80,27 @@ export class RuntimeHostCatalogReadError extends Error { } } +export interface RuntimeHostSessionCatalogPageCursor { + readonly revision: SessionCatalogRevision; + readonly cursor: string; +} + +export interface RuntimeHostSessionCatalogPage { + readonly revision: SessionCatalogRevision; + readonly sessions: readonly SessionCatalogItem[]; + readonly nextCursor: RuntimeHostSessionCatalogPageCursor | null; +} + +export class RuntimeHostSessionCatalogRevisionChangedError extends Error { + constructor( + readonly expectedRevision: SessionCatalogRevision, + readonly actualRevision: SessionCatalogRevision, + ) { + super('Runtime Host Session catalog revision changed'); + this.name = 'RuntimeHostSessionCatalogRevisionChangedError'; + } +} + export async function readRuntimeHostConnectionCatalog( connection: RuntimeHostCatalogConnection, ): Promise { @@ -186,24 +208,49 @@ export async function readRuntimeHostSessions( ): Promise { const { pages } = await collectStablePages( 'session', - async () => { - const result = await connection.request('session.catalog.query', { - kind: 'list_start', - }); - return result.kind === 'page' ? result : null; - }, - async (revision, cursor) => { - const result = await connection.request('session.catalog.query', { - kind: 'list_continue', - revision, - cursor, - }); - return result.kind === 'page' ? result : null; + () => readRuntimeHostSessionCatalogPage(connection), + async (_revision, cursor) => { + try { + return await readRuntimeHostSessionCatalogPage(connection, cursor); + } catch (error) { + if (error instanceof RuntimeHostSessionCatalogRevisionChangedError) return null; + throw error; + } }, ); return pages.flatMap((page) => page.sessions); } +export async function readRuntimeHostSessionCatalogPage( + connection: RuntimeHostCatalogConnection, + cursor?: RuntimeHostSessionCatalogPageCursor, +): Promise { + const result = await connection.request( + 'session.catalog.query', + cursor + ? { kind: 'list_continue', revision: cursor.revision, cursor: cursor.cursor } + : { kind: 'list_start' }, + ); + if (result.kind === 'revision_changed') { + throw new RuntimeHostSessionCatalogRevisionChangedError( + result.expectedRevision, + result.actualRevision, + ); + } + if (result.kind !== 'page' || (cursor && result.revision !== cursor.revision)) { + throw new RuntimeHostCatalogReadError('session', 'invalid_projection'); + } + if (cursor && result.nextCursor === cursor.cursor) { + throw new RuntimeHostCatalogReadError('session', 'repeated_cursor'); + } + return { + revision: result.revision, + sessions: result.sessions, + nextCursor: + result.nextCursor === null ? null : { revision: result.revision, cursor: result.nextCursor }, + }; +} + export async function readRuntimeHostProjects( connection: RuntimeHostCatalogConnection, ): Promise { @@ -283,7 +330,11 @@ export async function readRuntimeHostResources( interface StableCatalogPage { readonly revision: string | number; - readonly nextCursor: string | ConnectionCatalogCursor | null; + readonly nextCursor: + | string + | ConnectionCatalogCursor + | RuntimeHostSessionCatalogPageCursor + | null; } async function collectStablePages( diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index 465c3cc567..ca78517d60 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -241,6 +241,11 @@ export interface RuntimeHostConnection { input: SubscriptionOpenInput, timeoutMs?: number, ): Promise; + /** Opens on this concrete connection without retrying on a replacement Host. */ + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise; close(): Promise; replaceClientCapabilities( provider: ClientCapabilityProvider, @@ -594,6 +599,13 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection { ); } + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise { + return this.openSessionSubscription(input, timeoutMs); + } + async close(): Promise { this.#clientCapabilities.close(new Error('Runtime Host connection closed by Client')); this.#transport.abort(); diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts index 7f73e67acf..361809bbd0 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -87,13 +87,17 @@ export { } from './startup-error.js'; export { RuntimeHostCatalogReadError, + RuntimeHostSessionCatalogRevisionChangedError, readRuntimeHostConnectionCatalog, readRuntimeHostInvocableSkills, readRuntimeHostProjectDetails, readRuntimeHostResources, readRuntimeHostProjects, + readRuntimeHostSessionCatalogPage, readRuntimeHostSessions, readRuntimeHostSkillCatalog, + type RuntimeHostSessionCatalogPage, + type RuntimeHostSessionCatalogPageCursor, } from './catalog-reader.js'; export { connectOrSpawnRuntimeHost, diff --git a/packages/runtime-host/src/client/reconnecting-connection.ts b/packages/runtime-host/src/client/reconnecting-connection.ts index 4287cd4c8e..773e1d6e9d 100644 --- a/packages/runtime-host/src/client/reconnecting-connection.ts +++ b/packages/runtime-host/src/client/reconnecting-connection.ts @@ -49,6 +49,11 @@ export interface RuntimeHostReconnectingConnection extends RuntimeHostConnection subscribeConnectionAvailability( listener: (availability: RuntimeHostConnectionAvailability) => void, ): () => void; + /** Opens only on the currently connected Host and never retries on a replacement. */ + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise; } export type RuntimeHostConnectionAvailability = @@ -211,6 +216,13 @@ class RuntimeHostReconnectingConnectionImpl implements RuntimeHostReconnectingCo } } + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise { + return this.#requireCurrent('subscription.open').openSessionSubscriptionOnce(input, timeoutMs); + } + async replaceClientCapabilities( provider: ClientCapabilityProvider, timeoutMs?: number,