From 27b2963acf6dd091b411628211c113e4c187a08d Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Mon, 24 Aug 2026 11:41:51 +0800 Subject: [PATCH 1/6] feat(runtime-host): expose paged sessions and one-shot opens Generated-by: Codex --- .../src/__tests__/catalog-reader.test.ts | 95 +++++++++++++++++++ .../__tests__/reconnecting-connection.test.ts | 83 ++++++++++++++++ .../runtime-host/src/client/catalog-reader.ts | 79 ++++++++++++--- .../runtime-host/src/client/connection.ts | 12 +++ packages/runtime-host/src/client/index.ts | 4 + .../src/client/reconnecting-connection.ts | 12 +++ 6 files changed, 271 insertions(+), 14 deletions(-) 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 59ac6eca8e..acdfedcc61 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, { @@ -614,6 +693,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 966cc2e789..a280b8da56 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -113,13 +113,17 @@ export { } from './wsl-environment.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, From f4d8724d627a9b9411a5690cef7bb8fb1e6512cb Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Mon, 24 Aug 2026 11:41:55 +0800 Subject: [PATCH 2/6] feat(cli): add ACP session registry Generated-by: Codex --- .../__tests__/acp-session-registry.test.ts | 716 ++++++++++++++++++ packages/cli/src/acp/session-registry.ts | 469 ++++++++++++ 2 files changed, 1185 insertions(+) create mode 100644 packages/cli/src/__tests__/acp-session-registry.test.ts create mode 100644 packages/cli/src/acp/session-registry.ts 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..b854a37da7 --- /dev/null +++ b/packages/cli/src/__tests__/acp-session-registry.test.ts @@ -0,0 +1,716 @@ +/* + * 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 SessionContinuitySnapshot, + type SubscriptionFrame, +} from '@maka/runtime-host/protocol'; +import { AcpSessionRegistry, type AcpSessionRegistryOptions } 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('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({ + connection: fakeConnection({ + request: async (operation, input) => { + requests.push({ operation, input }); + return { kind: 'unsupported_legacy_record' }; + }, + 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' }, + ); + assert.deepEqual(await registry.create({ cwd: '/workspace/two', mcpServers: [] }), { + sessionId: 'session-2', + }); + 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('records subscription failures without producing an unhandled rejection', async () => { + const subscription = new TestSubscription('session-1'); + const registry = new AcpSessionRegistry({ + connection: 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({ + connection: 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({ + connection: 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 registry = new AcpSessionRegistry({ + connection: fakeConnection({ + 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); + 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({ + connection: 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'); + const registry = new AcpSessionRegistry({ + connection: fakeConnection({ open: async () => opening.promise }), + newSessionId: () => 'session-race', + }); + + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + 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 connectionCloses = 0; + const interruption = new RuntimeHostRequestInterruptedError( + 'subscription.open', + 'control', + 'not_dispatched', + 'connection_lost', + ); + const registry = new AcpSessionRegistry({ + connection: fakeConnection({ + open: async () => 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); + 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 connectionCloses = 0; + const registry = new AcpSessionRegistry({ + connection: fakeConnection({ + open: async () => opening.promise, + close: async () => { + connectionCloses += 1; + subscription.end(); + }, + }), + newSessionId: () => 'session-race', + }); + + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + 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({ + connection: 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({ + connection: 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({ + connection: 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({ connection: 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({ + connection: 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({ + connection: 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; + } = {}, +): AcpSessionRegistryOptions['connection'] { + return { + request: overrides.request ?? (async () => ({})), + openSessionSubscriptionOnce: + overrides.open ?? (async ({ sessionId }) => new TestSubscription(sessionId)), + close: overrides.close ?? (async () => undefined), + } as AcpSessionRegistryOptions['connection']; +} + +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: string, activityAt: number) { + 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: 'default', + collaborationMode: 'default', + orchestrationMode: 'default', + }; +} + +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/acp/session-registry.ts b/packages/cli/src/acp/session-registry.ts new file mode 100644 index 0000000000..716f685928 --- /dev/null +++ b/packages/cli/src/acp/session-registry.ts @@ -0,0 +1,469 @@ +/* + * 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, +} 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 SessionContinuitySnapshot, + type SubscriptionFrame, + type SubscriptionOpenInput, +} from '@maka/runtime-host/protocol'; + +const ACP_SESSION_CURSOR_VERSION = 1 as const; +const ACP_SESSION_CURSOR_MAX_BYTES = 8 * 1024; + +interface AcpSessionSubscription extends AsyncIterable { + readonly snapshot: SessionContinuitySnapshot; + close(): Promise; +} + +interface AcpSessionRegistryConnection { + readonly request: RuntimeHostConnection['request']; + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise; + close(): Promise; +} + +export interface AcpSessionRegistryOptions { + readonly connection: AcpSessionRegistryConnection; + 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 #connection: AcpSessionRegistryConnection; + readonly #newSessionId: () => string; + readonly #records = new Map(); + readonly #inFlightOperations = new Set>(); + #closing = false; + #connectionCloseTask: Promise | undefined; + #disposeTask: Promise | undefined; + + constructor(options: AcpSessionRegistryOptions) { + this.#connection = options.connection; + 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)); + } + + 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.#disposeTask ??= this.#dispose(); + return this.#disposeTask; + } + + async #create(params: NewSessionRequest): Promise { + const sessionId = this.#newSessionId(); + try { + await this.#connection.request('session.create', { + sessionId, + workspace: { kind: 'host_path', path: params.cwd }, + modelTarget: { kind: 'default' }, + }); + } catch (error) { + throw requestErrorFromRuntimeHost(error, 'session.create', { sessionId }); + } + + let subscription: AcpSessionSubscription; + try { + subscription = await this.#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 }; + } + + 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; + let page; + try { + page = await readRuntimeHostSessionCatalogPage( + this.#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 #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 { + this.#connectionCloseTask ??= Promise.resolve().then(() => this.#connection.close()); + return this.#connectionCloseTask; + } + + 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 RequestError.internalError( + { source: 'runtime_host', operation: 'subscription.open', code: 'registry_closed' }, + 'ACP session registry is closed', + ); + } +} + +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', + extra: Record = {}, +): RequestError { + const data = { ...runtimeHostErrorData(error, operation), ...extra }; + if (error instanceof RuntimeHostOperationError && error.code === 'invalid_request') { + 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(); +} From 6fd60f57fb9eb09d1b9a6bdcbe64aea9fb187b0f Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Mon, 24 Aug 2026 11:41:59 +0800 Subject: [PATCH 3/6] feat(cli): map ACP session creation and listing Generated-by: Codex --- packages/cli/src/__tests__/acp-agent.test.ts | 71 ++++++++-- .../src/__tests__/acp-child-process.test.ts | 53 ++++++- .../src/__tests__/acp-stdio-server.test.ts | 132 +++++++++++++++++- packages/cli/src/acp/maka-acp-agent.ts | 17 ++- packages/cli/src/acp/stdio-server.ts | 63 ++++++--- packages/cli/src/cli-core.ts | 6 +- 6 files changed, 300 insertions(+), 42 deletions(-) diff --git a/packages/cli/src/__tests__/acp-agent.test.ts b/packages/cli/src/__tests__/acp-agent.test.ts index fad4260895..dd9dfbb186 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,49 @@ describe('Maka ACP agent', () => { ); }); - test('rejects unimplemented session requests with method details', async () => { + test('routes official SDK new and list requests through the Session registry', async () => { + const creates: unknown[] = []; + const lists: unknown[] = []; await client({ name: 'test-client' }).connectWith( - createMakaAcpAgent({ version: '0.2.0' }), + createMakaAcpAgent({ + version: '0.2.0', + sessionRegistry: fakeSessionRegistry({ creates, lists }), + }), + async (agent) => { + assert.deepEqual( + await agent.request(methods.agent.session.new, { + cwd: '/workspace', + mcpServers: [], + _meta: { ignored: true }, + }), + { sessionId: 'session-1' }, + ); + 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(creates, [{ cwd: '/workspace', mcpServers: [], _meta: { ignored: true } }]); + assert.deepEqual(lists, [{ cwd: '/workspace' }]); + }); + + 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('session/new', { cwd: '/workspace' }), + 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/new' }); + assert.deepEqual(error.data, { method: 'session/close' }); return true; }, ); @@ -57,7 +90,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 +99,25 @@ describe('Maka ACP agent', () => { } }); }); + +function fakeSessionRegistry(observations: { creates?: unknown[]; lists?: unknown[] } = {}) { + return { + create: async (params: unknown) => { + observations.creates?.push(params); + return { sessionId: 'session-1' }; + }, + list: async (params: unknown) => { + observations.lists?.push(params); + return { + sessions: [ + { + sessionId: 'session-1', + cwd: '/workspace', + title: 'Session', + updatedAt: '2026-08-24T00:00:00.000Z', + }, + ], + }; + }, + }; +} diff --git a/packages/cli/src/__tests__/acp-child-process.test.ts b/packages/cli/src/__tests__/acp-child-process.test.ts index b7b7aebe50..58509548a8 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,17 +119,59 @@ 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 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 >= 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: [], + }); + const second = await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }); + assert.notEqual(first.sessionId, second.sessionId); + 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('session/new', { cwd: harness.workspaceRoot }), + 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/new' }); + assert.deepEqual(error.data, { method: 'session/close' }); return true; }, ); @@ -139,12 +182,12 @@ 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 >= 5, 'expected initialize, new, list, and method responses'); for (const line of lines) { const message: unknown = JSON.parse(line); assertJsonRpcMessage(message); } - }); + }, { startRuntimeHost: true }); }); }); diff --git a/packages/cli/src/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index a1feb5bcfd..104469b643 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -20,6 +20,11 @@ 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 } from '@maka/runtime-host/protocol'; import { runMakaAcpStdioServer } from '../acp/stdio-server.js'; describe('Maka ACP stdio server', () => { @@ -40,7 +45,7 @@ describe('Maka ACP stdio server', () => { id: 1, result: { protocolVersion: 1, - agentCapabilities: {}, + agentCapabilities: { sessionCapabilities: { list: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }, @@ -74,10 +79,70 @@ describe('Maka ACP stdio server', () => { await assert.rejects(harness.run(), (error: unknown) => error === transportError); }); + + test('disposes ACP subscriptions before closing the Runtime Host context', async () => { + const lifecycle: string[] = []; + const connection = { + request: async () => ({ kind: 'unsupported_legacy_record' }), + openSessionSubscriptionOnce: async ({ sessionId }: { sessionId: string }) => + closingSubscription(sessionId, lifecycle), + close: async () => undefined, + } 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, + onClose: () => lifecycle.push('context.close'), + }, + ); + + 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 }; + }; + assert.equal(response.jsonrpc, '2.0'); + assert.equal(response.id, 2); + assert.equal(typeof response.result?.sessionId, 'string'); + assert.deepEqual(lifecycle, ['subscription.close', 'context.close']); + }); }); -function createHarness(chunks: string[], options: { readonly stdin?: Readable } = {}) { +function createHarness( + chunks: string[], + options: { + readonly stdin?: Readable; + readonly connection?: RuntimeHostConnection; + readonly onClose?: () => void; + } = {}, +) { const stdin = options.stdin ?? Readable.from(chunks.map((chunk) => Buffer.from(chunk))); + let closes = 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 +151,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 () => + ({ + connection, + close: async () => { + closes += 1; + options.onClose?.(); + }, + }) as Awaited< + ReturnType + >, + }, + ), + closeCalls: () => closes, stdoutMessages: () => Buffer.concat(stdoutChunks) .toString('utf8') @@ -96,3 +179,46 @@ function createHarness(chunks: string[], options: { readonly stdin?: Readable } .map((line) => JSON.parse(line) as unknown), }; } + +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/acp/maka-acp-agent.ts b/packages/cli/src/acp/maka-acp-agent.ts index 2508c16c3c..18293dbec2 100644 --- a/packages/cli/src/acp/maka-acp-agent.ts +++ b/packages/cli/src/acp/maka-acp-agent.ts @@ -18,16 +18,21 @@ */ 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)); } diff --git a/packages/cli/src/acp/stdio-server.ts b/packages/cli/src/acp/stdio-server.ts index e5c5df732b..deb41afa93 100644 --- a/packages/cli/src/acp/stdio-server.ts +++ b/packages/cli/src/acp/stdio-server.ts @@ -20,12 +20,17 @@ import { Readable, Writable } from 'node:stream'; import { ndJsonStream } from '@agentclientprotocol/sdk'; 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,27 +39,47 @@ export async function runMakaAcpStdioServer( input: MakaAcpStdioServerInput, dependencies: MakaAcpStdioServerDependencies = {}, ): Promise { - const stdin = dependencies.stdin ?? process.stdin; - const stdout = dependencies.stdout ?? process.stdout; - let stdioError: Error | undefined; - const recordStdioError = (error: Error) => { - stdioError ??= error; - }; - stdin.once('error', recordStdioError); - stdout.once('error', recordStdioError); + const context = await (dependencies.connectRuntimeHostCli ?? connectRuntimeHostCli)({ + rootPath: input.workspaceRoot, + clientDataRoot: input.clientDataRoot, + }); + let closeContextTask: Promise | undefined; + const closeContext = () => (closeContextTask ??= context.close()); + const sessionRegistry = new AcpSessionRegistry({ + connection: context.connection, + }); try { - const stream = ndJsonStream( - Writable.toWeb(stdout) as WritableStream, - Readable.toWeb(stdin) as ReadableStream, - ); - const connection = createMakaAcpAgent({ version: input.version }).connect(stream); - await connection.closed; - if (stdioError) { - throw stdioError; + const stdin = dependencies.stdin ?? process.stdin; + const stdout = dependencies.stdout ?? process.stdout; + let stdioError: Error | undefined; + const recordStdioError = (error: Error) => { + stdioError ??= error; + }; + stdin.once('error', recordStdioError); + stdout.once('error', recordStdioError); + try { + const stream = ndJsonStream( + Writable.toWeb(stdout) as WritableStream, + Readable.toWeb(stdin) as ReadableStream, + ); + const connection = createMakaAcpAgent({ + version: input.version, + sessionRegistry, + }).connect(stream); + await connection.closed; + if (stdioError) { + throw stdioError; + } + return 0; + } finally { + stdin.off('error', recordStdioError); + stdout.off('error', recordStdioError); } - return 0; } finally { - stdin.off('error', recordStdioError); - stdout.off('error', recordStdioError); + try { + await sessionRegistry.dispose(); + } finally { + await closeContext(); + } } } diff --git a/packages/cli/src/cli-core.ts b/packages/cli/src/cli-core.ts index a54e1c0f2a..c75bb4c452 100644 --- a/packages/cli/src/cli-core.ts +++ b/packages/cli/src/cli-core.ts @@ -284,7 +284,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'); From 50d4decebe52b6cc6234d21792f70f21dd8cb6fe Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Tue, 25 Aug 2026 09:51:06 +0800 Subject: [PATCH 4/6] fix(cli): lazily connect ACP Session registry Generated-by: Codex --- .../__tests__/acp-session-registry.test.ts | 420 ++++++++++++------ packages/cli/src/acp/session-registry.ts | 85 +++- packages/cli/src/acp/stdio-server.ts | 2 +- 3 files changed, 361 insertions(+), 146 deletions(-) diff --git a/packages/cli/src/__tests__/acp-session-registry.test.ts b/packages/cli/src/__tests__/acp-session-registry.test.ts index b854a37da7..c757a7d691 100644 --- a/packages/cli/src/__tests__/acp-session-registry.test.ts +++ b/packages/cli/src/__tests__/acp-session-registry.test.ts @@ -34,31 +34,141 @@ import { type SessionContinuitySnapshot, type SubscriptionFrame, } from '@maka/runtime-host/protocol'; -import { AcpSessionRegistry, type AcpSessionRegistryOptions } from '../acp/session-registry.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 + 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('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; + }, + }); + const first = registry.list({}); + const second = registry.list({}); + await waitFor(() => connectCalls === 1); + + connecting.resolve( + fakeConnection({ + request: async () => ({ + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: null, + }), + }), + ); + + assert.deepEqual(await first, { sessions: [] }); + assert.deepEqual(await second, { 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({ - connection: fakeConnection({ - request: async (operation, input) => { - requests.push({ operation, input }); - return { kind: 'unsupported_legacy_record' }; - }, - open: async ({ sessionId }) => { - const subscription = new TestSubscription(sessionId); - subscriptions.set(sessionId, subscription); - return subscription; - }, - }), + connect: async () => + fakeConnection({ + request: async (operation, input) => { + requests.push({ operation, input }); + return { kind: 'unsupported_legacy_record' }; + }, + open: async ({ sessionId }) => { + const subscription = new TestSubscription(sessionId); + subscriptions.set(sessionId, subscription); + return subscription; + }, + }), newSessionId: () => `session-${++nextId}`, }); @@ -117,7 +227,7 @@ describe('ACP Session registry', () => { test('records subscription failures without producing an unhandled rejection', async () => { const subscription = new TestSubscription('session-1'); const registry = new AcpSessionRegistry({ - connection: fakeConnection({ open: async () => subscription }), + connect: async () => fakeConnection({ open: async () => subscription }), newSessionId: () => 'session-1', }); await registry.create({ cwd: '/workspace', mcpServers: [] }); @@ -139,7 +249,7 @@ describe('ACP Session registry', () => { test('records an unexpected clean subscription end as a failure', async () => { const subscription = new TestSubscription('session-1'); const registry = new AcpSessionRegistry({ - connection: fakeConnection({ open: async () => subscription }), + connect: async () => fakeConnection({ open: async () => subscription }), newSessionId: () => 'session-1', }); await registry.create({ cwd: '/workspace', mcpServers: [] }); @@ -157,12 +267,13 @@ describe('ACP Session registry', () => { test('rejects unsupported creation inputs before touching Runtime Host', async () => { let requests = 0; const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - request: async () => { - requests += 1; - return {}; - }, - }), + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return {}; + }, + }), }); const cases: Array = [ @@ -175,10 +286,20 @@ describe('ACP Session registry', () => { ], [ 'additionalDirectories', - { cwd: '/workspace', mcpServers: [], additionalDirectories: ['/other'] }, + { + cwd: '/workspace', + mcpServers: [], + additionalDirectories: ['/other'], + }, ], ['cwd', { cwd: 'relative', mcpServers: [] }], - ['cwd', { cwd: `/${'x'.repeat(SESSION_CATALOG_CWD_MAX_BYTES)}`, mcpServers: [] }], + [ + 'cwd', + { + cwd: `/${'x'.repeat(SESSION_CATALOG_CWD_MAX_BYTES)}`, + mcpServers: [], + }, + ], ]; for (const [field, input] of cases) { await assert.rejects( @@ -195,15 +316,16 @@ describe('ACP Session registry', () => { test('reports a durable session identity when subscription opening fails without rollback', async () => { const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - open: async () => { - throw new RuntimeHostOperationError( - 'subscription.open', - 'operation_unavailable', - 'subscription unavailable', - ); - }, - }), + connect: async () => + fakeConnection({ + open: async () => { + throw new RuntimeHostOperationError( + 'subscription.open', + 'operation_unavailable', + 'subscription unavailable', + ); + }, + }), newSessionId: () => 'session-durable', }); @@ -233,15 +355,16 @@ describe('ACP Session registry', () => { ] as const) { let opens = 0; const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - request: async () => { - throw new RuntimeHostOperationError('session.create', hostCode, 'create failed'); - }, - open: async ({ sessionId }) => { - opens += 1; - return new TestSubscription(sessionId); - }, - }), + 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}`, }); @@ -267,12 +390,20 @@ describe('ACP Session registry', () => { 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({ - connection: fakeConnection({ open: async () => opening.promise }), + 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); @@ -290,6 +421,7 @@ describe('ACP Session registry', () => { 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', @@ -298,18 +430,23 @@ describe('ACP Session registry', () => { 'connection_lost', ); const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - open: async () => opening.promise, - close: async () => { - connectionCloses += 1; - opening.reject(interruption); - }, - }), + 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); @@ -326,19 +463,25 @@ describe('ACP Session registry', () => { 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({ - connection: fakeConnection({ - open: async () => opening.promise, - close: async () => { - connectionCloses += 1; - subscription.end(); - }, - }), + 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); @@ -353,13 +496,14 @@ describe('ACP Session registry', () => { subscription.closeError = new Error('subscription close failed'); let connectionCloses = 0; const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - open: async () => subscription, - close: async () => { - connectionCloses += 1; - subscription.end(); - }, - }), + connect: async () => + fakeConnection({ + open: async () => subscription, + close: async () => { + connectionCloses += 1; + subscription.end(); + }, + }), newSessionId: () => 'session-1', }); await registry.create({ cwd: '/workspace', mcpServers: [] }); @@ -380,42 +524,43 @@ describe('ACP Session registry', () => { const canonicalWorkspace = await realpath(workspace); const inputs: unknown[] = []; const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - request: async (operation, input) => { - assert.equal(operation, 'session.catalog.query'); - inputs.push(input); - if ((input as { kind: string }).kind === 'list_start') { + 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('other', join(root, 'other'), 'Other', 1_000), - { - kind: 'unsupported_legacy_record', - id: 'legacy', - revision: 1, - reason: 'not_wire_representable', - }, + catalogSession('matching', canonicalWorkspace, 'Matching session', 2_000), + catalogSession( + 'undated', + canonicalWorkspace, + 'Out-of-range activity', + Number.MAX_SAFE_INTEGER, + ), ], - nextCursor: 'page-2', + nextCursor: null, }; - } - 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 }); @@ -447,17 +592,18 @@ describe('ACP Session registry', () => { test('rejects a cursor reused with a different normalized cwd before Host I/O', async () => { let requests = 0; const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - request: async () => { - requests += 1; - return { - kind: 'page', - revision: SESSION_REVISION, - sessions: [], - nextCursor: 'page-2', - }; - }, - }), + 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' }); @@ -473,9 +619,16 @@ describe('ACP Session registry', () => { }); test('rejects malformed and oversized ACP cursors as invalid params', async () => { - const registry = new AcpSessionRegistry({ connection: fakeConnection() }); + const registry = new AcpSessionRegistry({ + connect: async () => fakeConnection(), + }); const invalidRevisionCursor = Buffer.from( - JSON.stringify({ v: 1, revision: 'sha256:bad', cursor: 'page-2', cwd: null }), + 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]) { @@ -514,18 +667,19 @@ describe('ACP Session registry', () => { ] as const) { let first = true; const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - request: async () => { - if (!first) return nextResult; - first = false; - return { - kind: 'page', - revision: SESSION_REVISION, - sessions: [], - nextCursor: 'page-2', - }; - }, - }), + 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) => { @@ -540,15 +694,16 @@ describe('ACP Session registry', () => { test('maps Runtime Host invalid_request from session/list to invalid params', async () => { const registry = new AcpSessionRegistry({ - connection: fakeConnection({ - request: async () => { - throw new RuntimeHostOperationError( - 'session.catalog.query', - 'invalid_request', - 'invalid query', - ); - }, - }), + connect: async () => + fakeConnection({ + request: async () => { + throw new RuntimeHostOperationError( + 'session.catalog.query', + 'invalid_request', + 'invalid query', + ); + }, + }), }); await assert.rejects(registry.list({}), (error: unknown) => { @@ -632,13 +787,13 @@ function fakeConnection( open?: (input: { sessionId: string }) => Promise; close?: () => Promise; } = {}, -): AcpSessionRegistryOptions['connection'] { +): AcpSessionRegistryConnection { return { request: overrides.request ?? (async () => ({})), openSessionSubscriptionOnce: overrides.open ?? (async ({ sessionId }) => new TestSubscription(sessionId)), close: overrides.close ?? (async () => undefined), - } as AcpSessionRegistryOptions['connection']; + } as AcpSessionRegistryConnection; } function snapshot(sessionId: string, projectionRevision: number): SessionContinuitySnapshot { @@ -654,7 +809,12 @@ function snapshot(sessionId: string, projectionRevision: number): SessionContinu projectionRevision, rootTurn: null, goal: null, - queue: { hostEpoch: 'host-1', queueRevision: 0, steering: [], followup: [] }, + queue: { + hostEpoch: 'host-1', + queueRevision: 0, + steering: [], + followup: [], + }, interactions: { pending: [] }, }; } diff --git a/packages/cli/src/acp/session-registry.ts b/packages/cli/src/acp/session-registry.ts index 716f685928..47623caa86 100644 --- a/packages/cli/src/acp/session-registry.ts +++ b/packages/cli/src/acp/session-registry.ts @@ -53,7 +53,7 @@ interface AcpSessionSubscription extends AsyncIterable { close(): Promise; } -interface AcpSessionRegistryConnection { +export interface AcpSessionRegistryConnection { readonly request: RuntimeHostConnection['request']; openSessionSubscriptionOnce( input: SubscriptionOpenInput, @@ -63,7 +63,7 @@ interface AcpSessionRegistryConnection { } export interface AcpSessionRegistryOptions { - readonly connection: AcpSessionRegistryConnection; + readonly connect: () => Promise; readonly newSessionId?: () => string; } @@ -90,16 +90,18 @@ interface AcpSessionRecord { /** Owns all Runtime Host resources associated with one ACP connection. */ export class AcpSessionRegistry { - readonly #connection: AcpSessionRegistryConnection; + readonly #connect: () => Promise; readonly #newSessionId: () => string; readonly #records = new Map(); readonly #inFlightOperations = new Set>(); + #connection: AcpSessionRegistryConnection | undefined; + #connectTask: Promise | undefined; #closing = false; #connectionCloseTask: Promise | undefined; #disposeTask: Promise | undefined; constructor(options: AcpSessionRegistryOptions) { - this.#connection = options.connection; + this.#connect = options.connect; this.#newSessionId = options.newSessionId ?? randomUUID; } @@ -130,9 +132,10 @@ export class AcpSessionRegistry { } async #create(params: NewSessionRequest): Promise { + const connection = await this.#getConnection(); const sessionId = this.#newSessionId(); try { - await this.#connection.request('session.create', { + await connection.request('session.create', { sessionId, workspace: { kind: 'host_path', path: params.cwd }, modelTarget: { kind: 'default' }, @@ -143,7 +146,7 @@ export class AcpSessionRegistry { let subscription: AcpSessionSubscription; try { - subscription = await this.#connection.openSessionSubscriptionOnce({ + subscription = await connection.openSessionSubscriptionOnce({ sessionId, transcript: { kind: 'none' }, }); @@ -219,10 +222,11 @@ export class AcpSessionRegistry { ); } const cwd = requestedCwd ?? cursor?.cwd ?? null; + const connection = await this.#getConnection(); let page; try { page = await readRuntimeHostSessionCatalogPage( - this.#connection, + connection, cursor ? { revision: cursor.revision, cursor: cursor.cursor } : undefined, ); } catch (error) { @@ -269,10 +273,45 @@ export class AcpSessionRegistry { } #closeOwnedConnection(): Promise { - this.#connectionCloseTask ??= Promise.resolve().then(() => this.#connection.close()); + 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; + const connectTask = (this.#connectTask ??= Promise.resolve().then(() => this.#connect())); + let connection: AcpSessionRegistryConnection; + try { + connection = await connectTask; + } catch { + if (this.#connectTask === connectTask) this.#connectTask = undefined; + if (this.#closing) throw registryClosedError('connect'); + throw RequestError.internalError( + { + source: 'runtime_host', + operation: 'connect', + code: 'connection_failed', + }, + 'Runtime Host connection failed', + ); + } + 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 { @@ -284,13 +323,17 @@ export class AcpSessionRegistry { #assertOpen(): void { if (!this.#closing) return; - throw RequestError.internalError( - { source: 'runtime_host', operation: 'subscription.open', code: 'registry_closed' }, - 'ACP session registry is closed', - ); + throw registryClosedError('subscription.open'); } } +function registryClosedError(operation: 'connect' | 'subscription.open'): RequestError { + return RequestError.internalError( + { source: 'runtime_host', operation, code: 'registry_closed' }, + 'ACP session registry is closed', + ); +} + function validateNewSessionParams(params: NewSessionRequest): void { assertBoundedAbsoluteCwd(params.cwd); if (params.mcpServers.length > 0) { @@ -321,7 +364,11 @@ function requestErrorFromRuntimeHost( function runtimeHostErrorData(error: unknown, operation: string): Record { if (error instanceof RuntimeHostOperationError) { - return { source: 'runtime_host', operation: error.operation, code: error.code }; + return { + source: 'runtime_host', + operation: error.operation, + code: error.code, + }; } if (error instanceof RuntimeHostRequestInterruptedError) { return { @@ -381,7 +428,11 @@ function encodeAcpSessionCursor( ).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' }, + { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'cursor_too_large', + }, 'Runtime Host cursor cannot be represented safely in ACP', ); } @@ -442,7 +493,11 @@ async function normalizeCwd(cwd: string): Promise { 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' }, + { + source: 'filesystem', + operation: 'cwd.realpath', + code: code ?? 'internal_failure', + }, 'cwd could not be canonicalized', ); } diff --git a/packages/cli/src/acp/stdio-server.ts b/packages/cli/src/acp/stdio-server.ts index deb41afa93..e8a5b80397 100644 --- a/packages/cli/src/acp/stdio-server.ts +++ b/packages/cli/src/acp/stdio-server.ts @@ -46,7 +46,7 @@ export async function runMakaAcpStdioServer( let closeContextTask: Promise | undefined; const closeContext = () => (closeContextTask ??= context.close()); const sessionRegistry = new AcpSessionRegistry({ - connection: context.connection, + connect: async () => context.connection, }); try { const stdin = dependencies.stdin ?? process.stdin; From f0f766d6427044ad295c078e8b2977aae60052c3 Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Tue, 25 Aug 2026 09:53:18 +0800 Subject: [PATCH 5/6] fix(cli): defer ACP Host connection to Session methods Generated-by: Codex --- .../src/__tests__/acp-child-process.test.ts | 93 ++++++++++--------- .../src/__tests__/acp-stdio-server.test.ts | 86 +++++++++++++---- packages/cli/src/acp/stdio-server.ts | 67 ++++++------- 3 files changed, 147 insertions(+), 99 deletions(-) diff --git a/packages/cli/src/__tests__/acp-child-process.test.ts b/packages/cli/src/__tests__/acp-child-process.test.ts index 58509548a8..0c8b7019cc 100644 --- a/packages/cli/src/__tests__/acp-child-process.test.ts +++ b/packages/cli/src/__tests__/acp-child-process.test.ts @@ -141,53 +141,56 @@ describe('Maka ACP child process', () => { 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: [], - }); - const second = await context.request(methods.agent.session.new, { - cwd: harness.workspaceRoot, - mcpServers: [], - }); - assert.notEqual(first.sessionId, second.sessionId); - const listed = await context.request(methods.agent.session.list, { - cwd: harness.workspaceRoot, + 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: [], + }); + const second = await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }); + assert.notEqual(first.sessionId, second.sessionId); + 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; + }, + ); }); - 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 >= 5, 'expected initialize, new, list, and method responses'); - for (const line of lines) { - const message: unknown = JSON.parse(line); - assertJsonRpcMessage(message); - } - }, { startRuntimeHost: 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 >= 5, 'expected initialize, new, list, and method responses'); + for (const line of lines) { + const message: unknown = JSON.parse(line); + assertJsonRpcMessage(message); + } + }, + { startRuntimeHost: true }, + ); }); }); diff --git a/packages/cli/src/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index 104469b643..5a8463b68d 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -28,7 +28,7 @@ import { SESSION_CONTINUITY_SCHEMA_VERSION } from '@maka/runtime-host/protocol'; 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', @@ -51,12 +51,14 @@ describe('Maka ACP stdio server', () => { }, }, ]); + 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 () => { @@ -80,13 +82,15 @@ describe('Maka ACP stdio server', () => { await assert.rejects(harness.run(), (error: unknown) => error === transportError); }); - test('disposes ACP subscriptions before closing the Runtime Host context', async () => { + test('disposes ACP subscriptions before closing the lazily acquired Host connection', async () => { const lifecycle: string[] = []; const connection = { request: async () => ({ kind: 'unsupported_legacy_record' }), openSessionSubscriptionOnce: async ({ sessionId }: { sessionId: string }) => closingSubscription(sessionId, lifecycle), - close: async () => undefined, + close: async () => { + lifecycle.push('connection.close'); + }, } as unknown as RuntimeHostConnection; const harness = createHarness( [ @@ -105,7 +109,6 @@ describe('Maka ACP stdio server', () => { ], { connection, - onClose: () => lifecycle.push('context.close'), }, ); @@ -120,7 +123,56 @@ describe('Maka ACP stdio server', () => { assert.equal(response.jsonrpc, '2.0'); assert.equal(response.id, 2); assert.equal(typeof response.result?.sessionId, 'string'); - assert.deepEqual(lifecycle, ['subscription.close', 'context.close']); + 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); }); }); @@ -129,11 +181,11 @@ function createHarness( options: { readonly stdin?: Readable; readonly connection?: RuntimeHostConnection; - readonly onClose?: () => void; + readonly connectError?: Error; } = {}, ) { const stdin = options.stdin ?? Readable.from(chunks.map((chunk) => Buffer.from(chunk))); - let closes = 0; + let connects = 0; const connection = options.connection ?? ({ @@ -157,19 +209,19 @@ function createHarness( { stdin, stdout, - connectRuntimeHostCli: async () => - ({ + connectRuntimeHostCli: async () => { + connects += 1; + if (options.connectError) throw options.connectError; + return { connection, - close: async () => { - closes += 1; - options.onClose?.(); - }, - }) as Awaited< + close: async () => undefined, + } as Awaited< ReturnType - >, + >; + }, }, ), - closeCalls: () => closes, + connectCalls: () => connects, stdoutMessages: () => Buffer.concat(stdoutChunks) .toString('utf8') diff --git a/packages/cli/src/acp/stdio-server.ts b/packages/cli/src/acp/stdio-server.ts index e8a5b80397..40a25d06f8 100644 --- a/packages/cli/src/acp/stdio-server.ts +++ b/packages/cli/src/acp/stdio-server.ts @@ -39,47 +39,40 @@ export async function runMakaAcpStdioServer( input: MakaAcpStdioServerInput, dependencies: MakaAcpStdioServerDependencies = {}, ): Promise { - const context = await (dependencies.connectRuntimeHostCli ?? connectRuntimeHostCli)({ - rootPath: input.workspaceRoot, - clientDataRoot: input.clientDataRoot, - }); - let closeContextTask: Promise | undefined; - const closeContext = () => (closeContextTask ??= context.close()); const sessionRegistry = new AcpSessionRegistry({ - connect: async () => context.connection, + connect: async () => + ( + await (dependencies.connectRuntimeHostCli ?? connectRuntimeHostCli)({ + rootPath: input.workspaceRoot, + clientDataRoot: input.clientDataRoot, + }) + ).connection, }); + const stdin = dependencies.stdin ?? process.stdin; + const stdout = dependencies.stdout ?? process.stdout; + let stdioError: Error | undefined; + const recordStdioError = (error: Error) => { + stdioError ??= error; + }; + stdin.once('error', recordStdioError); + stdout.once('error', recordStdioError); try { - const stdin = dependencies.stdin ?? process.stdin; - const stdout = dependencies.stdout ?? process.stdout; - let stdioError: Error | undefined; - const recordStdioError = (error: Error) => { - stdioError ??= error; - }; - stdin.once('error', recordStdioError); - stdout.once('error', recordStdioError); - try { - const stream = ndJsonStream( - Writable.toWeb(stdout) as WritableStream, - Readable.toWeb(stdin) as ReadableStream, - ); - const connection = createMakaAcpAgent({ - version: input.version, - sessionRegistry, - }).connect(stream); - await connection.closed; - if (stdioError) { - throw stdioError; - } - return 0; - } finally { - stdin.off('error', recordStdioError); - stdout.off('error', recordStdioError); + const stream = ndJsonStream( + Writable.toWeb(stdout) as WritableStream, + Readable.toWeb(stdin) as ReadableStream, + ); + const connection = createMakaAcpAgent({ + version: input.version, + sessionRegistry, + }).connect(stream); + await connection.closed; + if (stdioError) { + throw stdioError; } + return 0; } finally { - try { - await sessionRegistry.dispose(); - } finally { - await closeContext(); - } + stdin.off('error', recordStdioError); + stdout.off('error', recordStdioError); + await sessionRegistry.dispose(); } } From 951bd81810a23b315244d2fe648a61083e39262a Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Tue, 25 Aug 2026 10:15:30 +0800 Subject: [PATCH 6/6] fix(cli): make lazy ACP Host bootstrap abortable Generated-by: Codex --- .../__tests__/acp-child-process-harness.ts | 17 ++- .../src/__tests__/acp-child-process.test.ts | 5 + .../__tests__/acp-session-registry.test.ts | 79 ++++++++++++-- .../src/__tests__/acp-stdio-server.test.ts | 29 ++++- .../runtime-host-cli-context.test.ts | 102 ++++++++++++++++++ packages/cli/src/acp/session-registry.ts | 24 ++++- packages/cli/src/acp/stdio-server.ts | 24 +++-- packages/cli/src/runtime-host-cli-context.ts | 81 +++++++++++++- 8 files changed, 334 insertions(+), 27 deletions(-) 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 0c8b7019cc..64a4a15d0b 100644 --- a/packages/cli/src/__tests__/acp-child-process.test.ts +++ b/packages/cli/src/__tests__/acp-child-process.test.ts @@ -123,6 +123,11 @@ describe('Maka ACP child process', () => { authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }); + assert.equal( + await harness.hasRuntimeHostRootMarker(), + false, + 'initialize must not begin Runtime Host discovery or candidate startup', + ); }); await harness.closeStdin(); diff --git a/packages/cli/src/__tests__/acp-session-registry.test.ts b/packages/cli/src/__tests__/acp-session-registry.test.ts index c757a7d691..c3b795c0a9 100644 --- a/packages/cli/src/__tests__/acp-session-registry.test.ts +++ b/packages/cli/src/__tests__/acp-session-registry.test.ts @@ -58,6 +58,55 @@ describe('ACP Session registry', () => { 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; @@ -66,24 +115,28 @@ describe('ACP Session registry', () => { connectCalls += 1; return connecting.promise; }, + newSessionId: () => 'session-concurrent', }); - const first = registry.list({}); - const second = registry.list({}); + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + const list = registry.list({}); await waitFor(() => connectCalls === 1); connecting.resolve( fakeConnection({ - request: async () => ({ - kind: 'page', - revision: SESSION_REVISION, - sessions: [], - nextCursor: null, - }), + request: async (operation) => + operation === 'session.catalog.query' + ? { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: null, + } + : {}, }), ); - assert.deepEqual(await first, { sessions: [] }); - assert.deepEqual(await second, { sessions: [] }); + assert.deepEqual(await create, { sessionId: 'session-concurrent' }); + assert.deepEqual(await list, { sessions: [] }); assert.equal(connectCalls, 1); await registry.dispose(); }); @@ -315,9 +368,14 @@ describe('ACP Session registry', () => { }); 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 {}; + }, open: async () => { throw new RuntimeHostOperationError( 'subscription.open', @@ -345,6 +403,7 @@ describe('ACP Session registry', () => { }, ); assert.equal(registry.inspect('session-durable'), undefined); + assert.deepEqual(operations, ['session.create']); await registry.dispose(); }); diff --git a/packages/cli/src/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index 5a8463b68d..d86d3a6e67 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -174,6 +174,33 @@ describe('Maka ACP stdio server', () => { 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( @@ -214,7 +241,7 @@ function createHarness( if (options.connectError) throw options.connectError; return { connection, - close: async () => undefined, + close: () => connection.close(), } as Awaited< ReturnType >; 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 ce00c11a8a..7e1aa5724a 100644 --- a/packages/cli/src/__tests__/runtime-host-cli-context.test.ts +++ b/packages/cli/src/__tests__/runtime-host-cli-context.test.ts @@ -123,6 +123,82 @@ test('CLI refuses a staged Host whose durable installation claim is missing', as 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( @@ -528,6 +604,14 @@ function hostRegistration(overrides: Partial = {}): HostRegist }; } +function connectedHostResult(connection: RuntimeHostConnection) { + return { + kind: 'connected' as const, + connection, + registration: hostRegistration(), + }; +} + function incompatibleRemoteHandshake(overrides: Partial = {}): HostIncompatible { return { kind: 'incompatible', @@ -557,3 +641,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/session-registry.ts b/packages/cli/src/acp/session-registry.ts index 47623caa86..4f9bcd405d 100644 --- a/packages/cli/src/acp/session-registry.ts +++ b/packages/cli/src/acp/session-registry.ts @@ -63,7 +63,7 @@ export interface AcpSessionRegistryConnection { } export interface AcpSessionRegistryOptions { - readonly connect: () => Promise; + readonly connect: (signal: AbortSignal) => Promise; readonly newSessionId?: () => string; } @@ -90,12 +90,13 @@ interface AcpSessionRecord { /** Owns all Runtime Host resources associated with one ACP connection. */ export class AcpSessionRegistry { - readonly #connect: () => Promise; + 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; @@ -127,6 +128,7 @@ export class AcpSessionRegistry { dispose(): Promise { this.#closing = true; + this.#connectAbortController?.abort(); this.#disposeTask ??= this.#dispose(); return this.#disposeTask; } @@ -288,12 +290,25 @@ export class AcpSessionRegistry { async #getConnection(): Promise { this.#assertOpen(); if (this.#connection) return this.#connection; - const connectTask = (this.#connectTask ??= Promise.resolve().then(() => this.#connect())); + 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( { @@ -304,6 +319,9 @@ export class AcpSessionRegistry { 'Runtime Host connection failed', ); } + if (this.#connectAbortController === connectController) { + this.#connectAbortController = undefined; + } if (this.#closing) { await this.#closeOwnedConnection().catch(() => undefined); throw registryClosedError('connect'); diff --git a/packages/cli/src/acp/stdio-server.ts b/packages/cli/src/acp/stdio-server.ts index 40a25d06f8..6b022702c7 100644 --- a/packages/cli/src/acp/stdio-server.ts +++ b/packages/cli/src/acp/stdio-server.ts @@ -19,6 +19,7 @@ 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'; @@ -40,13 +41,22 @@ export async function runMakaAcpStdioServer( dependencies: MakaAcpStdioServerDependencies = {}, ): Promise { const sessionRegistry = new AcpSessionRegistry({ - connect: async () => - ( - await (dependencies.connectRuntimeHostCli ?? connectRuntimeHostCli)({ - rootPath: input.workspaceRoot, - clientDataRoot: input.clientDataRoot, - }) - ).connection, + 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; diff --git a/packages/cli/src/runtime-host-cli-context.ts b/packages/cli/src/runtime-host-cli-context.ts index 41c64c5cb9..3260c3d347 100644 --- a/packages/cli/src/runtime-host-cli-context.ts +++ b/packages/cli/src/runtime-host-cli-context.ts @@ -112,6 +112,7 @@ export async function connectRuntimeHostCli( readonly profileId?: string; readonly clientDataRoot?: string; readonly interactiveSsh?: boolean; + readonly signal?: AbortSignal; }, overrides: Partial = {}, ): Promise { @@ -186,20 +187,30 @@ export async function connectRuntimeHostCli( } return connected.connection; }; + let initialConnection: RuntimeHostConnection | undefined; let connection: Awaited> | undefined; try { - const initialConnection = await connect( - undefined, - input.interactiveSsh && process.stdin.isTTY && process.stdout.isTTY ? 'inherit' : 'batch', + 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 liveConnection = connection; + const catalog = await runAbortably( + () => deps.readConnectionCatalog(liveConnection), + input.signal, + ); return { connection: liveConnection, - catalog: await deps.readConnectionCatalog(liveConnection), + catalog, profile, close: async () => { try { @@ -210,12 +221,72 @@ export async function connectRuntimeHostCli( }, }; } catch (error) { - await connection?.close().catch(() => undefined); + await (connection ?? initialConnection)?.close().catch(() => undefined); await peerClient?.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,