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-harness.ts b/packages/cli/src/__tests__/acp-child-process-harness.ts index 1f1f5d69e9..7704917345 100644 --- a/packages/cli/src/__tests__/acp-child-process-harness.ts +++ b/packages/cli/src/__tests__/acp-child-process-harness.ts @@ -18,7 +18,7 @@ */ import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'; -import { mkdtemp, mkdir, rm } from 'node:fs/promises'; +import { lstat, mkdtemp, mkdir, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { PassThrough, Readable, Writable } from 'node:stream'; @@ -34,6 +34,7 @@ import { startExecutionRuntimeHostService, type RuntimeHostKernel, } from '@maka/runtime-host/server'; +import { STORAGE_ROOT_MARKER_FILE } from '@maka/storage/root-authority'; import { deriveMakaDataRoots, resolveMakaClientDataRoot } from '../workspace-root.js'; const DEFAULT_TIMEOUT_MS = 15_000; @@ -117,6 +118,16 @@ export class AcpChildProcessHarness { return Buffer.concat(this.#stderr).toString('utf8'); } + async hasRuntimeHostRootMarker(): Promise { + try { + await lstat(join(this.#workspaceRoot, STORAGE_ROOT_MARKER_FILE)); + return true; + } catch (error) { + if (isErrorWithCode(error, 'ENOENT')) return false; + throw error; + } + } + async withClient( operation: (client: AcpChildProcessClient) => Promise | T, configureClient: ConfigureAcpClient = (app) => app, @@ -515,3 +526,7 @@ class StartupTimeoutError extends Error {} function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } + +function isErrorWithCode(error: unknown, code: string): boolean { + return error instanceof Error && 'code' in error && error.code === code; +} diff --git a/packages/cli/src/__tests__/acp-child-process.test.ts b/packages/cli/src/__tests__/acp-child-process.test.ts index b7b7aebe50..6789fd611b 100644 --- a/packages/cli/src/__tests__/acp-child-process.test.ts +++ b/packages/cli/src/__tests__/acp-child-process.test.ts @@ -19,6 +19,7 @@ import assert from 'node:assert/strict'; import { once } from 'node:events'; +import { realpath } from 'node:fs/promises'; import { PassThrough } from 'node:stream'; import { describe, test } from 'node:test'; import { RequestError, methods } from '@agentclientprotocol/sdk'; @@ -118,19 +119,14 @@ describe('Maka ACP child process', () => { await harness.withClient(async ({ context }) => { assert.deepEqual(await context.request(methods.agent.initialize, { protocolVersion: 1 }), { protocolVersion: 1, - agentCapabilities: {}, + agentCapabilities: { sessionCapabilities: { list: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }); - - await assert.rejects( - context.request('session/new', { cwd: harness.workspaceRoot }), - (error: unknown) => { - assert.ok(error instanceof RequestError); - assert.equal(error.code, -32601); - assert.deepEqual(error.data, { method: 'session/new' }); - return true; - }, + assert.equal( + await harness.hasRuntimeHostRootMarker(), + false, + 'initialize must not begin Runtime Host discovery or candidate startup', ); }); @@ -139,13 +135,101 @@ describe('Maka ACP child process', () => { assert.equal(harness.stderr, ''); const lines = harness.stdout.split(/\r?\n/u).filter((line) => line.trim().length > 0); - assert.ok(lines.length >= 2, 'expected initialize and method-not-found responses'); + assert.ok(lines.length >= 1, 'expected initialize response'); for (const line of lines) { const message: unknown = JSON.parse(line); assertJsonRpcMessage(message); } }); }); + + test('serves multiple ACP Sessions through a real Runtime Host', { + timeout: 30_000, + }, async () => { + await withAcpChildProcessHarness( + async (harness) => { + await harness.withClient(async ({ context }) => { + await context.request(methods.agent.initialize, { protocolVersion: 1 }); + const first = await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }); + 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; + }, + ); + }); + + 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 }, + ); + }); + + test('creates more than sixteen Sessions without attaching on 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 createdSessionIds: string[] = []; + for (let index = 0; index < 17; index += 1) { + const created = await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }); + createdSessionIds.push(created.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(createdSessionIds), + ); + }); + + await harness.closeStdin(); + assert.deepEqual(await harness.waitForExit(), { code: 0, signal: null }); + assert.equal(harness.stderr, ''); + }, + { startRuntimeHost: true }, + ); + }); }); function assertJsonRpcMessage(message: unknown): void { 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..25a5a81419 --- /dev/null +++ b/packages/cli/src/__tests__/acp-session-registry.test.ts @@ -0,0 +1,618 @@ +/* + * 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 } from '@maka/runtime-host/client'; +import { SESSION_CATALOG_CWD_MAX_BYTES } from '@maka/runtime-host/protocol'; +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; + +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('reports the requested Session operation after disposal', async () => { + const registry = new AcpSessionRegistry({ + connect: async () => fakeConnection(), + }); + await registry.dispose(); + + for (const [operation, request] of [ + ['session.create', () => registry.create({ cwd: '/workspace', mcpServers: [] })], + ['session.catalog.query', () => registry.list({})], + ] as const) { + await assert.rejects(request(), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation, + code: 'registry_closed', + }); + return true; + }); + } + }); + + test('does not start a queued connection after disposal begins', async () => { + let connectCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + return fakeConnection(); + }, + }); + + const list = registry.list({}); + const dispose = registry.dispose(); + + await assert.rejects( + list, + (error: unknown) => + error instanceof RequestError && + error.code === -32603 && + (error.data as { code?: string }).code === 'registry_closed', + ); + await dispose; + assert.equal(connectCalls, 0); + }); + + test('aborts an in-flight connection before disposal waits for it', async () => { + let connectSignal: AbortSignal | undefined; + const registry = new AcpSessionRegistry({ + connect: async (signal) => { + connectSignal = signal; + return new Promise>((_, reject) => { + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + }); + }, + }); + + const list = registry.list({}); + await waitFor(() => connectSignal !== undefined); + const dispose = registry.dispose(); + + await assert.rejects( + list, + (error: unknown) => + error instanceof RequestError && + error.code === -32603 && + (error.data as { code?: string }).code === 'registry_closed', + ); + await dispose; + assert.equal(connectSignal?.aborted, true); + }); + + test('shares one in-flight connection across concurrent Session methods', async () => { + const connecting = deferred>(); + let connectCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + return connecting.promise; + }, + newSessionId: () => 'session-concurrent', + }); + const create = registry.create({ cwd: '/workspace', mcpServers: [] }); + const list = registry.list({}); + await waitFor(() => connectCalls === 1); + + connecting.resolve( + fakeConnection({ + request: async (operation) => + operation === 'session.catalog.query' + ? { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: null, + } + : {}, + }), + ); + + assert.deepEqual(await create, { sessionId: 'session-concurrent' }); + assert.deepEqual(await list, { sessions: [] }); + assert.equal(connectCalls, 1); + await registry.dispose(); + }); + + test('reports a stable connection error and retries on a later Session request', async () => { + let connectCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + if (connectCalls === 1) throw new Error('Host unavailable'); + return fakeConnection({ + request: async () => ({ + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: null, + }), + }); + }, + }); + + await assert.rejects(registry.list({}), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'connect', + code: 'connection_failed', + }); + return true; + }); + assert.deepEqual(await registry.list({}), { sessions: [] }); + assert.equal(connectCalls, 2); + await registry.dispose(); + }); + + test('closes a connection that resolves after disposal starts', async () => { + const connecting = deferred>(); + let connectCalls = 0; + let closeCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + connectCalls += 1; + return connecting.promise; + }, + }); + const list = registry.list({}); + await waitFor(() => connectCalls === 1); + const dispose = registry.dispose(); + + connecting.resolve( + fakeConnection({ + close: async () => { + closeCalls += 1; + }, + }), + ); + + await assert.rejects(list, (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.equal((error.data as { code?: string }).code, 'registry_closed'); + return true; + }); + await dispose; + assert.equal(closeCalls, 1); + }); + + test('creates more than the Host subscription limit without opening a subscription', async () => { + const sessionCount = 17; + const createdSessionIds: string[] = []; + let subscriptionOpens = 0; + let nextId = 0; + const registry = new AcpSessionRegistry({ + connect: async () => { + const connection = fakeConnection({ + request: async (operation, input) => { + assert.equal(operation, 'session.create'); + createdSessionIds.push((input as { sessionId: string }).sessionId); + return {}; + }, + }); + return { + ...connection, + openSessionSubscriptionOnce: async () => { + subscriptionOpens += 1; + throw new Error('PR 2 must not open a subscription'); + }, + } as AcpSessionRegistryConnection; + }, + newSessionId: () => `session-unattached-${++nextId}`, + }); + + const creates = await Promise.all( + Array.from({ length: sessionCount }, () => + registry.create({ cwd: '/workspace', mcpServers: [] }), + ), + ); + + assert.equal(creates.length, sessionCount); + assert.equal(createdSessionIds.length, sessionCount); + assert.equal(subscriptionOpens, 0); + await registry.dispose(); + }); + + test('rejects unsupported creation inputs before touching Runtime Host', async () => { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return {}; + }, + }), + }); + + const cases: Array = [ + [ + 'mcpServers', + { + cwd: '/workspace', + mcpServers: [{ name: 'server', command: 'server', args: [], env: [] }], + }, + ], + [ + 'additionalDirectories', + { + cwd: '/workspace', + mcpServers: [], + additionalDirectories: ['/other'], + }, + ], + ['cwd', { cwd: 'relative', mcpServers: [] }], + [ + 'cwd', + { + cwd: `/${'x'.repeat(SESSION_CATALOG_CWD_MAX_BYTES)}`, + mcpServers: [], + }, + ], + ]; + for (const [field, input] of cases) { + await assert.rejects( + registry.create(input), + (error: unknown) => + error instanceof RequestError && + error.code === -32602 && + (error.data as { field?: string }).field === field, + ); + } + assert.equal(requests, 0); + await registry.dispose(); + }); + + test('keeps failed and outcome-unknown creates distinct', async () => { + for (const [hostCode, acpCode] of [ + ['invalid_request', -32602], + ['commit_outcome_unknown', -32603], + ] as const) { + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + throw new RuntimeHostOperationError('session.create', hostCode, 'create failed'); + }, + }), + 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; + }, + ); + await registry.dispose(); + } + }); + + test('maps one filtered Host catalog page per ACP page and carries cwd across pages', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-acp-list-')); + t.after(() => rm(root, { recursive: true, force: true })); + const workspace = join(root, 'workspace'); + const alias = join(root, 'workspace-alias'); + await mkdir(workspace); + await symlink(workspace, alias); + const canonicalWorkspace = await realpath(workspace); + const inputs: unknown[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + assert.equal(operation, 'session.catalog.query'); + inputs.push(input); + if ((input as { kind: string }).kind === 'list_start') { + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [ + catalogSession('other', join(root, 'other'), 'Other', 1_000), + { + kind: 'unsupported_legacy_record', + id: 'legacy', + revision: 1, + reason: 'not_wire_representable', + }, + ], + nextCursor: 'page-2', + }; + } + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [ + catalogSession('matching', canonicalWorkspace, 'Matching session', 2_000), + catalogSession( + 'undated', + canonicalWorkspace, + 'Out-of-range activity', + Number.MAX_SAFE_INTEGER, + ), + ], + nextCursor: null, + }; + }, + }), + }); + + const first = await registry.list({ cwd: alias }); + assert.deepEqual(first.sessions, []); + assert.equal(typeof first.nextCursor, 'string'); + const second = await registry.list({ cursor: first.nextCursor }); + assert.deepEqual(second, { + sessions: [ + { + sessionId: 'matching', + cwd: canonicalWorkspace, + title: 'Matching session', + updatedAt: '1970-01-01T00:00:02.000Z', + }, + { + sessionId: 'undated', + cwd: canonicalWorkspace, + title: 'Out-of-range activity', + }, + ], + }); + assert.deepEqual(inputs, [ + { kind: 'list_start' }, + { kind: 'list_continue', revision: SESSION_REVISION, cursor: 'page-2' }, + ]); + await registry.dispose(); + }); + + test('rejects a cursor reused with a different normalized cwd before Host I/O', async () => { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: 'page-2', + }; + }, + }), + }); + const first = await registry.list({ cwd: '/workspace/one/../one' }); + + await assert.rejects( + registry.list({ cwd: '/workspace/two', cursor: first.nextCursor }), + (error: unknown) => + error instanceof RequestError && + error.code === -32602 && + (error.data as { reason?: string }).reason === 'cursor_cwd_mismatch', + ); + assert.equal(requests, 1); + await registry.dispose(); + }); + + test('rejects malformed and oversized ACP cursors as invalid params', async () => { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return {}; + }, + }), + }); + const invalidRevisionCursor = Buffer.from( + JSON.stringify({ + revision: 'sha256:bad', + cursor: 'page-2', + cwd: null, + }), + 'utf8', + ).toString('base64url'); + const versionedCursor = Buffer.from( + JSON.stringify({ + v: 1, + revision: SESSION_REVISION, + cursor: 'page-2', + cwd: null, + }), + 'utf8', + ).toString('base64url'); + for (const cursor of [ + 'not-a-cursor', + 'x'.repeat(8 * 1024 + 1), + invalidRevisionCursor, + versionedCursor, + ]) { + await assert.rejects( + registry.list({ cursor }), + (error: unknown) => + error instanceof RequestError && + error.code === -32602 && + (error.data as { reason?: string }).reason === 'invalid_cursor', + ); + } + assert.equal(requests, 0); + await registry.dispose(); + }); + + test('translates stale and repeated Host cursors into stable ACP errors', async () => { + for (const [nextResult, expectedCode, expectedReason] of [ + [ + { + kind: 'revision_changed', + expectedRevision: SESSION_REVISION, + actualRevision: NEW_SESSION_REVISION, + }, + -32602, + 'stale_cursor', + ], + [ + { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: 'page-2', + }, + -32603, + 'repeated_cursor', + ], + ] as const) { + let first = true; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + if (!first) return nextResult; + first = false; + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: 'page-2', + }; + }, + }), + }); + const page = await registry.list({}); + await assert.rejects(registry.list({ cursor: page.nextCursor }), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, expectedCode); + assert.equal((error.data as { reason?: string; code?: string }).reason, expectedReason); + return true; + }); + await registry.dispose(); + } + }); + + test('maps Runtime Host invalid_request from session/list to invalid params', async () => { + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + throw new RuntimeHostOperationError( + 'session.catalog.query', + 'invalid_request', + 'invalid query', + ); + }, + }), + }); + + await assert.rejects(registry.list({}), (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32602); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'invalid_request', + }); + return true; + }); + await registry.dispose(); + }); +}); + +function fakeConnection( + overrides: { + request?: (operation: string, input: unknown) => Promise; + close?: () => Promise; + } = {}, +): AcpSessionRegistryConnection { + return { + request: overrides.request ?? (async () => ({})), + close: overrides.close ?? (async () => undefined), + } as AcpSessionRegistryConnection; +} + +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/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index a1feb5bcfd..4e703dd30e 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -20,10 +20,11 @@ import assert from 'node:assert/strict'; import { Readable, Writable } from 'node:stream'; import { describe, test } from 'node:test'; +import type { RuntimeHostConnection } from '@maka/runtime-host/client'; import { runMakaAcpStdioServer } from '../acp/stdio-server.js'; describe('Maka ACP stdio server', () => { - test('answers initialize without Runtime Host input or dependencies', async () => { + test('answers initialize without connecting a Runtime Host', async () => { const harness = createHarness([ `${JSON.stringify({ jsonrpc: '2.0', @@ -40,18 +41,20 @@ describe('Maka ACP stdio server', () => { id: 1, result: { protocolVersion: 1, - agentCapabilities: {}, + agentCapabilities: { sessionCapabilities: { list: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }, }, ]); + assert.equal(harness.connectCalls(), 0); }); - test('returns zero after normal EOF', async () => { + test('returns zero after normal EOF without connecting a Runtime Host', async () => { const harness = createHarness([]); assert.equal(await harness.run(), 0); + assert.equal(harness.connectCalls(), 0); }); test('returns a JSON-RPC parse error and then zero after EOF', async () => { @@ -74,10 +77,138 @@ describe('Maka ACP stdio server', () => { await assert.rejects(harness.run(), (error: unknown) => error === transportError); }); + + test('creates a Session without requiring a subscription-capable Host connection', async () => { + const lifecycle: string[] = []; + const connection = { + request: async (operation: string) => { + lifecycle.push(operation); + return {}; + }, + close: async () => { + lifecycle.push('connection.close'); + }, + } as unknown as RuntimeHostConnection; + const harness = createHarness( + [ + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1 }, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'session/new', + params: { cwd: '/workspace', mcpServers: [] }, + })}\n`, + ], + { connection }, + ); + + assert.equal(await harness.run(), 0); + const response = harness + .stdoutMessages() + .find((message) => (message as { id?: unknown }).id === 2) as { + result?: { sessionId?: unknown }; + }; + assert.equal(typeof response.result?.sessionId, 'string'); + assert.deepEqual(lifecycle, ['session.create', 'connection.close']); + }); + + test('returns a Host connection failure from the Session request and keeps serving ACP', async () => { + const harness = createHarness( + [ + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1 }, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'session/list', + params: {}, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 3, + method: 'session/close', + params: { sessionId: 'missing' }, + })}\n`, + ], + { connectError: new Error('Host unavailable') }, + ); + + assert.equal(await harness.run(), 0); + const responses = new Map( + harness + .stdoutMessages() + .map((message) => [(message as { id?: unknown }).id, message] as const), + ); + const connectionFailure = responses.get(2) as { + error?: { code?: unknown; data?: unknown }; + }; + assert.equal(connectionFailure.error?.code, -32603); + assert.deepEqual(connectionFailure.error?.data, { + source: 'runtime_host', + operation: 'connect', + code: 'connection_failed', + }); + const methodFailure = responses.get(3) as { + error?: { code?: unknown; data?: unknown }; + }; + assert.equal(methodFailure.error?.code, -32601); + assert.deepEqual(methodFailure.error?.data, { method: 'session/close' }); + assert.equal(harness.connectCalls(), 1); + }); + + test('keeps an unimplemented Session method Host-independent', async () => { + const harness = createHarness([ + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1 }, + })}\n`, + `${JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'session/close', + params: { sessionId: 'missing' }, + })}\n`, + ]); + + assert.equal(await harness.run(), 0); + const response = harness + .stdoutMessages() + .find((message) => (message as { id?: unknown }).id === 2) as { + error?: { code?: unknown; data?: unknown }; + }; + assert.equal(response.error?.code, -32601); + assert.deepEqual(response.error?.data, { method: 'session/close' }); + assert.equal(harness.connectCalls(), 0); + }); }); -function createHarness(chunks: string[], options: { readonly stdin?: Readable } = {}) { +function createHarness( + chunks: string[], + options: { + readonly stdin?: Readable; + readonly connection?: RuntimeHostConnection; + readonly connectError?: Error; + } = {}, +) { const stdin = options.stdin ?? Readable.from(chunks.map((chunk) => Buffer.from(chunk))); + let connects = 0; + const connection = + options.connection ?? + ({ + request: async () => ({ kind: 'unsupported_legacy_record' }), + close: async () => undefined, + } as unknown as RuntimeHostConnection); const stdoutChunks: Buffer[] = []; const stdout = new Writable({ write(chunk, _encoding, callback) { @@ -86,7 +217,27 @@ 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, + connectRuntimeHostCliConnection: async () => { + connects += 1; + if (options.connectError) throw options.connectError; + return { + connection, + close: () => connection.close(), + } as Awaited< + ReturnType< + typeof import('../runtime-host-cli-context.js').connectRuntimeHostCliConnection + > + >; + }, + }, + ), + connectCalls: () => connects, stdoutMessages: () => Buffer.concat(stdoutChunks) .toString('utf8') diff --git a/packages/cli/src/__tests__/cli.test.ts b/packages/cli/src/__tests__/cli.test.ts index 3fd6ba00b2..a4975f9409 100644 --- a/packages/cli/src/__tests__/cli.test.ts +++ b/packages/cli/src/__tests__/cli.test.ts @@ -54,7 +54,7 @@ describe('Maka CLI args', () => { assert.match(help.text, /^ maka update --target /m); assert.match( help.text, - /^ maka --acp Serve ACP v1 over stdio \(initialize only; session support in progress\)$/m, + /^ maka --acp Serve ACP v1 over stdio \(initialize, session\/new, session\/list\)$/m, ); assert.match(help.text, /^ maka runtime-host serve /m); assert.doesNotMatch(help.text, /cli:dev/); 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 83fe1dad99..8856ee4798 100644 --- a/packages/cli/src/__tests__/runtime-host-cli-context.test.ts +++ b/packages/cli/src/__tests__/runtime-host-cli-context.test.ts @@ -42,6 +42,7 @@ import { } from '@maka/runtime-host/protocol'; import { connectRuntimeHostCli, + connectRuntimeHostCliConnection, resolveRuntimeHostCliConflictDecision, RuntimeHostCliConflictError, shouldRetryRuntimeHostConflict, @@ -100,6 +101,35 @@ test('CLI Runtime Host bootstrap launches the execution composition', async () = assert.equal(closes, 1); }); +test('connection-only CLI bootstrap does not read the model connection catalog', async () => { + const connection = { + rootId: 'root-id', + hostEpoch: 'host-epoch', + connectionId: 'connection-id', + selectedProtocol: 0, + closed: new Promise(() => {}), + status: async () => ({ state: 'ready' }), + subscribeConfigurationChanges: () => () => {}, + subscribeConnectionCatalogChanges: () => () => {}, + subscribeProjectCatalogChanges: () => () => {}, + subscribeSessionCatalogChanges: () => () => {}, + subscribeScheduledTaskChanges: () => () => {}, + close: async () => {}, + } as unknown as RuntimeHostConnection; + const context = await connectRuntimeHostCliConnection( + { rootPath: '/runtime-host-root' }, + { + connectOrSpawn: async () => connectedHostResult(connection), + readConnectionCatalog: async () => { + throw new Error('model connection catalog unavailable'); + }, + }, + ); + + assert.equal(context.connection.connectionId, connection.connectionId); + await context.close(); +}); + test('CLI refuses a staged Host whose durable installation claim is missing', async () => { let closes = 0; await assert.rejects( @@ -125,6 +155,83 @@ 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: () => () => {}, + subscribeConnectionCatalogChanges: () => () => {}, + 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( @@ -542,6 +649,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', @@ -577,3 +692,21 @@ function singleRemoteProfileCatalog(profile: RemoteRuntimeHostProfile): RuntimeH readRemoteProfileIfCurrent: async () => assert.fail('unexpected read'), }; } + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +async function waitFor(predicate: () => boolean): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) return; + await new Promise((resolve) => setImmediate(resolve)); + } + assert.fail('condition was not reached'); +} diff --git a/packages/cli/src/acp/maka-acp-agent.ts b/packages/cli/src/acp/maka-acp-agent.ts index 2508c16c3c..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/session-registry.ts b/packages/cli/src/acp/session-registry.ts new file mode 100644 index 0000000000..f926e727f3 --- /dev/null +++ b/packages/cli/src/acp/session-registry.ts @@ -0,0 +1,399 @@ +/* + * 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, + type RuntimeHostConnection, + type RuntimeHostSessionCatalogPageCursor, +} from '@maka/runtime-host/client'; +import { + SESSION_CATALOG_CURSOR_MAX_BYTES, + SESSION_CATALOG_CWD_MAX_BYTES, +} from '@maka/runtime-host/protocol'; + +const ACP_SESSION_CURSOR_MAX_BYTES = 8 * 1024; + +type AcpSessionRegistryOperation = 'session.create' | 'session.catalog.query'; +type AcpSessionRegistryLifecycleOperation = 'connect' | AcpSessionRegistryOperation; + +export interface AcpSessionRegistryConnection { + readonly request: RuntimeHostConnection['request']; + close(): Promise; +} + +export interface AcpSessionRegistryOptions { + readonly connect: (signal: AbortSignal) => Promise; + readonly newSessionId?: () => string; +} + +/** Owns all Runtime Host resources associated with one ACP connection. */ +export class AcpSessionRegistry { + readonly #connect: (signal: AbortSignal) => Promise; + readonly #newSessionId: () => string; + readonly #inFlightOperations = new Set>(); + #connection: AcpSessionRegistryConnection | undefined; + #connectTask: Promise | undefined; + #connectAbortController: AbortController | undefined; + #closing = false; + #connectionCloseTask: Promise | undefined; + #disposeTask: Promise | undefined; + + constructor(options: AcpSessionRegistryOptions) { + this.#connect = options.connect; + this.#newSessionId = options.newSessionId ?? randomUUID; + } + + async create(params: NewSessionRequest): Promise { + this.#assertOpen('session.create'); + validateNewSessionParams(params); + return this.#track(this.#create(params)); + } + + async list(params: ListSessionsRequest): Promise { + this.#assertOpen('session.catalog.query'); + return this.#track(this.#list(params)); + } + + dispose(): Promise { + this.#closing = true; + this.#connectAbortController?.abort(); + this.#disposeTask ??= this.#dispose(); + return this.#disposeTask; + } + + async #create(params: NewSessionRequest): Promise { + const connection = await this.#getConnection('session.create'); + const sessionId = this.#newSessionId(); + try { + await connection.request('session.create', { + sessionId, + workspace: { kind: 'host_path', path: params.cwd }, + modelTarget: { kind: 'default' }, + }); + } catch (error) { + throw requestErrorFromRuntimeHost(error, 'session.create', { sessionId }); + } + return { sessionId }; + } + + async #list(params: ListSessionsRequest): Promise { + const cursor = params.cursor == null ? undefined : decodeAcpSessionCursor(params.cursor); + const requestedCwd = params.cwd == null ? undefined : await normalizeCwd(params.cwd); + if (cursor && requestedCwd !== undefined && cursor.cwd !== requestedCwd) { + throw RequestError.invalidParams( + { reason: 'cursor_cwd_mismatch' }, + 'cursor was created for a different cwd filter', + ); + } + const cwd = requestedCwd ?? cursor?.cwd ?? null; + const connection = await this.#getConnection('session.catalog.query'); + let page; + try { + page = await readRuntimeHostSessionCatalogPage( + connection, + cursor ? { revision: cursor.revision, cursor: cursor.cursor } : undefined, + ); + } catch (error) { + if (error instanceof RuntimeHostSessionCatalogRevisionChangedError) { + throw RequestError.invalidParams( + { reason: 'stale_cursor' }, + 'session catalog changed; restart listing from the first page', + ); + } + throw requestErrorFromRuntimeHost(error, 'session.catalog.query'); + } + + const sessions = page.sessions.flatMap((session) => { + if ('kind' in session || (cwd !== null && session.workspace.hostCwd !== cwd)) return []; + const updatedAt = isoTimestamp(session.activityAt); + return [ + { + sessionId: session.id, + cwd: session.workspace.hostCwd, + title: session.name, + ...(updatedAt ? { updatedAt } : {}), + }, + ]; + }); + return { + sessions, + ...(page.nextCursor + ? { nextCursor: encodeAcpSessionCursor({ ...page.nextCursor, cwd }) } + : {}), + }; + } + + async #dispose(): Promise { + const connectionClose = this.#closeOwnedConnection(); + await Promise.allSettled([connectionClose]); + await Promise.allSettled([...this.#inFlightOperations]); + } + + #closeOwnedConnection(): Promise { + const connection = this.#connection; + const connectTask = this.#connectTask; + if (!connection && !connectTask) return Promise.resolve(); + this.#connectionCloseTask ??= connection + ? Promise.resolve().then(() => connection.close()) + : connectTask!.then( + (connected) => connected.close(), + () => undefined, + ); + return this.#connectionCloseTask; + } + + async #getConnection( + operation: AcpSessionRegistryOperation, + ): Promise { + this.#assertOpen(operation); + if (this.#connection) return this.#connection; + let connectController = this.#connectAbortController; + if (!this.#connectTask) { + connectController = new AbortController(); + this.#connectAbortController = connectController; + this.#connectTask = Promise.resolve().then(() => { + if (this.#closing) throw registryClosedError('connect'); + connectController!.signal.throwIfAborted(); + return this.#connect(connectController!.signal); + }); + } + const connectTask = this.#connectTask; + let connection: AcpSessionRegistryConnection; + try { + connection = await connectTask; + } catch { + if (this.#connectTask === connectTask) this.#connectTask = undefined; + if (this.#connectAbortController === connectController) { + this.#connectAbortController = undefined; + } + if (this.#closing) throw registryClosedError('connect'); + throw RequestError.internalError( + { + source: 'runtime_host', + operation: 'connect', + code: 'connection_failed', + }, + 'Runtime Host connection failed', + ); + } + if (this.#connectAbortController === connectController) { + this.#connectAbortController = undefined; + } + if (this.#closing) { + await this.#closeOwnedConnection().catch(() => undefined); + throw registryClosedError('connect'); + } + this.#connection ??= connection; + return this.#connection; + } + + async #track(operation: Promise): Promise { + this.#inFlightOperations.add(operation); + try { + return await operation; + } finally { + this.#inFlightOperations.delete(operation); + } + } + + #assertOpen(operation: AcpSessionRegistryOperation): void { + if (!this.#closing) return; + throw registryClosedError(operation); + } +} + +function registryClosedError(operation: AcpSessionRegistryLifecycleOperation): 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) { + 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: AcpSessionRegistryOperation, + 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 RuntimeHostCatalogReadError) { + return { + source: 'runtime_host', + operation, + code: 'catalog_read_failure', + reason: error.reason, + }; + } + return { source: 'runtime_host', operation, code: 'internal_failure' }; +} + +interface AcpSessionCursor extends RuntimeHostSessionCatalogPageCursor { + readonly cwd: string | null; +} + +function encodeAcpSessionCursor( + cursor: RuntimeHostSessionCatalogPageCursor & { readonly cwd: string | null }, +): string { + const encoded = Buffer.from(JSON.stringify(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 !== 3 || + 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 { + revision: record.revision as RuntimeHostSessionCatalogPageCursor['revision'], + cursor: record.cursor, + cwd: record.cwd, + }; + } catch { + throw RequestError.invalidParams({ reason: 'invalid_cursor' }, 'cursor is invalid'); + } +} + +function validCursorCwd(value: unknown): value is string | null { + return ( + value === null || + (typeof value === 'string' && + isAbsolute(value) && + normalize(value) === value && + Buffer.byteLength(value, 'utf8') <= SESSION_CATALOG_CWD_MAX_BYTES) + ); +} + +async function normalizeCwd(cwd: string): Promise { + assertBoundedAbsoluteCwd(cwd); + const lexical = normalize(cwd); + try { + return await realpath(lexical); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === 'ENOENT' || code === 'ENOTDIR') return lexical; + throw RequestError.internalError( + { + source: 'filesystem', + operation: 'cwd.realpath', + code: code ?? 'internal_failure', + }, + 'cwd could not be canonicalized', + ); + } +} + +function assertBoundedAbsoluteCwd(cwd: string): void { + if (!isAbsolute(cwd)) { + throw RequestError.invalidParams( + { field: 'cwd', reason: 'must_be_absolute' }, + 'cwd must be an absolute path', + ); + } + if (Buffer.byteLength(cwd, 'utf8') > SESSION_CATALOG_CWD_MAX_BYTES) { + throw RequestError.invalidParams( + { field: 'cwd', reason: 'too_large' }, + 'cwd exceeds the Runtime Host path limit', + ); + } +} + +function isoTimestamp(timestamp: number): string | undefined { + const date = new Date(timestamp); + return Number.isNaN(date.getTime()) ? undefined : date.toISOString(); +} diff --git a/packages/cli/src/acp/stdio-server.ts b/packages/cli/src/acp/stdio-server.ts index e5c5df732b..27e2a1d985 100644 --- a/packages/cli/src/acp/stdio-server.ts +++ b/packages/cli/src/acp/stdio-server.ts @@ -19,13 +19,19 @@ import { Readable, Writable } from 'node:stream'; import { ndJsonStream } from '@agentclientprotocol/sdk'; +import type { RuntimeHostConnection } from '@maka/runtime-host/client'; import { createMakaAcpAgent } from './maka-acp-agent.js'; +import { AcpSessionRegistry } from './session-registry.js'; +import { connectRuntimeHostCliConnection } from '../runtime-host-cli-context.js'; export interface MakaAcpStdioServerInput { + readonly workspaceRoot: string; + readonly clientDataRoot: string; readonly version: string; } export interface MakaAcpStdioServerDependencies { + readonly connectRuntimeHostCliConnection?: typeof connectRuntimeHostCliConnection; readonly stdin?: Readable; readonly stdout?: Writable; } @@ -34,6 +40,23 @@ export async function runMakaAcpStdioServer( input: MakaAcpStdioServerInput, dependencies: MakaAcpStdioServerDependencies = {}, ): Promise { + const sessionRegistry = new AcpSessionRegistry({ + connect: async (signal) => { + const context = await ( + dependencies.connectRuntimeHostCliConnection ?? connectRuntimeHostCliConnection + )({ + rootPath: input.workspaceRoot, + clientDataRoot: input.clientDataRoot, + signal, + }); + return { + request: context.connection.request.bind( + context.connection, + ) as RuntimeHostConnection['request'], + close: () => context.close(), + }; + }, + }); const stdin = dependencies.stdin ?? process.stdin; const stdout = dependencies.stdout ?? process.stdout; let stdioError: Error | undefined; @@ -47,7 +70,10 @@ export async function runMakaAcpStdioServer( Writable.toWeb(stdout) as WritableStream, Readable.toWeb(stdin) as ReadableStream, ); - const connection = createMakaAcpAgent({ version: input.version }).connect(stream); + const connection = createMakaAcpAgent({ + version: input.version, + sessionRegistry, + }).connect(stream); await connection.closed; if (stdioError) { throw stdioError; @@ -56,5 +82,6 @@ export async function runMakaAcpStdioServer( } finally { stdin.off('error', recordStdioError); stdout.off('error', recordStdioError); + await sessionRegistry.dispose(); } } diff --git a/packages/cli/src/cli-core.ts b/packages/cli/src/cli-core.ts index b74adcbd71..f3755a8839 100644 --- a/packages/cli/src/cli-core.ts +++ b/packages/cli/src/cli-core.ts @@ -133,7 +133,7 @@ function helpText(cliCommand: string): string { '', 'Commands:', ` ${cliCommand} Start the TUI`, - ` ${cliCommand} --acp Serve ACP v1 over stdio (initialize only; session support in progress)`, + ` ${cliCommand} --acp Serve ACP v1 over stdio (initialize, session/new, session/list)`, ` ${cliCommand} run ... Run one non-interactive model turn`, ` ${cliCommand} activate ... Run one Cloud Session activation and emit JSONL`, ` ${cliCommand} -p ... Alias for ${cliCommand} run`, @@ -301,7 +301,11 @@ export async function runMakaCli( } case 'acp': { const { runMakaAcpStdioServer } = await import('./acp/stdio-server.js'); - return runMakaAcpStdioServer({ version }); + return runMakaAcpStdioServer({ + workspaceRoot: dataRoots.workspaceRoot, + clientDataRoot: dataRoots.clientDataRoot, + version, + }); } case 'runtime-host-serve': { const { runRuntimeHostServiceCli } = await import('./runtime-host-service-command.js'); diff --git a/packages/cli/src/runtime-host-cli-context.ts b/packages/cli/src/runtime-host-cli-context.ts index 58db9c08d3..b148320530 100644 --- a/packages/cli/src/runtime-host-cli-context.ts +++ b/packages/cli/src/runtime-host-cli-context.ts @@ -86,19 +86,34 @@ export class RuntimeHostCliConflictError extends RuntimeHostPermanentReconnectEr } } -export interface RuntimeHostCliConnectionContext { +export interface RuntimeHostCliConnectionOnlyContext { readonly connection: RuntimeHostConnection; - readonly catalog: ConnectionCatalogSnapshot; readonly profile: RuntimeHostProfile; close(): Promise; } -export interface RuntimeHostCliConnectionContextWithIdentity - extends RuntimeHostCliConnectionContext { +export interface RuntimeHostCliConnectionOnlyContextWithIdentity + extends RuntimeHostCliConnectionOnlyContext { readonly clientInstanceId: string; readonly profileIncarnationId?: string; } +export interface RuntimeHostCliConnectionContext extends RuntimeHostCliConnectionOnlyContext { + readonly catalog: ConnectionCatalogSnapshot; +} + +export interface RuntimeHostCliConnectionContextWithIdentity + extends RuntimeHostCliConnectionContext, + RuntimeHostCliConnectionOnlyContextWithIdentity {} + +export interface RuntimeHostCliConnectionInput { + readonly rootPath: string; + readonly profileId?: string; + readonly clientDataRoot?: string; + readonly interactiveSsh?: boolean; + readonly signal?: AbortSignal; +} + export interface RuntimeHostCliTarget { readonly connection: ConnectionCatalogEntry; readonly model: string; @@ -116,14 +131,27 @@ interface RuntimeHostCliContextDeps { } export async function connectRuntimeHostCli( - input: { - readonly rootPath: string; - readonly profileId?: string; - readonly clientDataRoot?: string; - readonly interactiveSsh?: boolean; - }, + input: RuntimeHostCliConnectionInput, overrides: Partial = {}, ): Promise { + const context = await connectRuntimeHostCliConnection(input, overrides); + try { + const catalog = await runAbortably( + () => + (overrides.readConnectionCatalog ?? readRuntimeHostConnectionCatalog)(context.connection), + input.signal, + ); + return { ...context, catalog }; + } catch (error) { + await context.close().catch(() => undefined); + throw error; + } +} + +export async function connectRuntimeHostCliConnection( + input: RuntimeHostCliConnectionInput, + overrides: Partial = {}, +): Promise { const deps: RuntimeHostCliContextDeps = { connectOrSpawn: connectOrSpawnRuntimeHost, connectProfile: connectRuntimeHostProfile, @@ -195,20 +223,25 @@ 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; return { connection: liveConnection, - catalog: await deps.readConnectionCatalog(liveConnection), profile, clientInstanceId, ...(resolvedProfile.profileIncarnationId @@ -223,12 +256,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, diff --git a/packages/runtime-host/src/__tests__/catalog-reader.test.ts b/packages/runtime-host/src/__tests__/catalog-reader.test.ts index da60b1fe68..8528074f10 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) => { @@ -62,6 +157,31 @@ test('waits out a burst of Session catalog revisions', async () => { assert.equal(starts, 4); }); +test('retries when the first Session catalog page reports a revision change', async () => { + let starts = 0; + const connection = fakeConnection(async (operation, input) => { + assert.equal(operation, 'session.catalog.query'); + assert.equal(input.kind, 'list_start'); + starts += 1; + if (starts === 1) { + return { + kind: 'revision_changed', + expectedRevision: 'sha256:old', + actualRevision: 'sha256:new', + }; + } + return { + kind: 'page', + revision: 'sha256:new', + sessions: [], + nextCursor: null, + }; + }); + + assert.deepEqual(await readRuntimeHostSessions(connection), []); + assert.equal(starts, 2); +}); + test('rejects a repeated Skill catalog cursor instead of looping forever', async () => { const connection = fakeConnection(async (operation, input) => { assert.equal(operation, 'skill.catalog.query'); diff --git a/packages/runtime-host/src/client/catalog-reader.ts b/packages/runtime-host/src/client/catalog-reader.ts index b5eb5bec64..d13272fe9c 100644 --- a/packages/runtime-host/src/client/catalog-reader.ts +++ b/packages/runtime-host/src/client/catalog-reader.ts @@ -27,6 +27,7 @@ import { type RelayModelProfile, type RelayModelProfiles, type SessionCatalogItem, + type SessionCatalogRevision, type SkillCatalogWorkspaceContext, type SkillCatalogInvocableItem, type SkillCatalogInvocableTarget, @@ -82,6 +83,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 { @@ -187,26 +209,54 @@ export async function readRuntimeHostInvocableSkills( export async function readRuntimeHostSessions( connection: RuntimeHostCatalogConnection, ): Promise { + const readPageOrRestart = async ( + cursor?: RuntimeHostSessionCatalogPageCursor, + ): Promise => { + try { + return await readRuntimeHostSessionCatalogPage(connection, cursor); + } catch (error) { + if (error instanceof RuntimeHostSessionCatalogRevisionChangedError) return null; + throw error; + } + }; 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; - }, + () => readPageOrRestart(), + (_revision, cursor) => readPageOrRestart(cursor), ); 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 { @@ -286,7 +336,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/index.ts b/packages/runtime-host/src/client/index.ts index c63200dc45..d1f007e159 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -129,13 +129,17 @@ export { } from './wsl-environment.js'; export { RuntimeHostCatalogReadError, + RuntimeHostSessionCatalogRevisionChangedError, readRuntimeHostConnectionCatalog, readRuntimeHostInvocableSkills, readRuntimeHostProjectDetails, readRuntimeHostResources, readRuntimeHostProjects, + readRuntimeHostSessionCatalogPage, readRuntimeHostSessions, readRuntimeHostSkillCatalog, + type RuntimeHostSessionCatalogPage, + type RuntimeHostSessionCatalogPageCursor, type RuntimeHostConnectionCatalogEntry, type RuntimeHostConnectionCatalogSnapshot, } from './catalog-reader.js';