diff --git a/packages/cli/src/__tests__/acp-agent.test.ts b/packages/cli/src/__tests__/acp-agent.test.ts index dd9dfbb186..709f0f3ca4 100644 --- a/packages/cli/src/__tests__/acp-agent.test.ts +++ b/packages/cli/src/__tests__/acp-agent.test.ts @@ -70,6 +70,43 @@ describe('Maka ACP agent', () => { assert.deepEqual(lists, [{ cwd: '/workspace' }]); }); + test('routes official SDK set-config requests through the Session registry', async () => { + const configurationRequests: unknown[] = []; + await client({ name: 'test-client' }).connectWith( + createMakaAcpAgent({ + version: '0.2.0', + sessionRegistry: fakeSessionRegistry({ configurationRequests }), + }), + async (agent) => { + assert.deepEqual( + await agent.request(methods.agent.session.setConfigOption, { + sessionId: 'session-1', + configId: 'collaboration_mode', + value: 'plan', + }), + { + configOptions: [ + { + type: 'select', + id: 'collaboration_mode', + name: 'Collaboration mode', + category: 'mode', + currentValue: 'plan', + options: [ + { value: 'agent', name: 'Agent' }, + { value: 'plan', name: 'Plan' }, + ], + }, + ], + }, + ); + }, + ); + assert.deepEqual(configurationRequests, [ + { sessionId: 'session-1', configId: 'collaboration_mode', value: 'plan' }, + ]); + }); + test('does not implement or advertise session/close', async () => { await client({ name: 'test-client' }).connectWith( createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), @@ -87,6 +124,26 @@ describe('Maka ACP agent', () => { ); }); + test('does not implement session/set_mode', async () => { + await client({ name: 'test-client' }).connectWith( + createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), + async (agent) => { + await assert.rejects( + agent.request(methods.agent.session.setMode, { + sessionId: 'session-1', + modeId: 'plan', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32601); + assert.deepEqual(error.data, { method: 'session/set_mode' }); + return true; + }, + ); + }, + ); + }); + 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( @@ -100,7 +157,9 @@ describe('Maka ACP agent', () => { }); }); -function fakeSessionRegistry(observations: { creates?: unknown[]; lists?: unknown[] } = {}) { +function fakeSessionRegistry( + observations: { creates?: unknown[]; lists?: unknown[]; configurationRequests?: unknown[] } = {}, +) { return { create: async (params: unknown) => { observations.creates?.push(params); @@ -119,5 +178,23 @@ function fakeSessionRegistry(observations: { creates?: unknown[]; lists?: unknow ], }; }, + setConfigOption: async (params: unknown) => { + observations.configurationRequests?.push(params); + return { + configOptions: [ + { + type: 'select' as const, + id: 'collaboration_mode', + name: 'Collaboration mode', + category: 'mode', + currentValue: 'plan', + options: [ + { value: 'agent', name: 'Agent' }, + { value: 'plan', name: 'Plan' }, + ], + }, + ], + }; + }, }; } diff --git a/packages/cli/src/__tests__/acp-child-process-harness.ts b/packages/cli/src/__tests__/acp-child-process-harness.ts index 7704917345..b350205b4a 100644 --- a/packages/cli/src/__tests__/acp-child-process-harness.ts +++ b/packages/cli/src/__tests__/acp-child-process-harness.ts @@ -30,10 +30,13 @@ import { type ClientConnection, type ClientContext, } from '@agentclientprotocol/sdk'; +import type { ThinkingLevel } from '@maka/core/model-thinking'; import { startExecutionRuntimeHostService, type RuntimeHostKernel, } from '@maka/runtime-host/server'; +import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores'; +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; import { STORAGE_ROOT_MARKER_FILE } from '@maka/storage/root-authority'; import { deriveMakaDataRoots, resolveMakaClientDataRoot } from '../workspace-root.js'; @@ -42,6 +45,10 @@ const DEFAULT_TIMEOUT_MS = 15_000; export interface AcpChildProcessHarnessOptions { readonly timeoutMs?: number; readonly startRuntimeHost?: boolean; + readonly model?: { + readonly id: string; + readonly thinkingLevels: readonly ThinkingLevel[]; + }; } export interface AcpChildProcessExit { @@ -289,6 +296,7 @@ export async function startAcpChildProcessHarness( let rootCleanupFollowsHostStartup = false; try { await mkdir(workspaceRoot, { recursive: true }); + if (options.model) await seedModelConnection(workspaceRoot, options.model); if (options.startRuntimeHost) { hostStartup = startExecutionRuntimeHostService({ rootPath: workspaceRoot }); try { @@ -341,6 +349,52 @@ export async function startAcpChildProcessHarness( } } +async function seedModelConnection( + rootPath: string, + model: NonNullable, +): Promise { + const capability = await resolveStorageRoot({ path: rootPath, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + if (!owner) throw new Error('Unable to acquire ACP model fixture root'); + try { + const policy = await openInteractiveRuntimePolicyStoresForWrite(owner.lease); + const created = await policy.connectionCatalog.create({ + expectedCatalogRevision: 0, + connection: { + slug: 'acp-fixture-model', + name: 'ACP fixture model', + providerType: 'openai-compatible', + baseUrl: 'https://acp-model.invalid/v1', + enabled: true, + enabledModelIds: [model.id], + relayModelProfiles: { + [model.id]: { thinkingLevels: model.thinkingLevels }, + }, + }, + }); + if (created.kind !== 'committed') throw new Error('ACP model fixture did not commit'); + const connection = created.snapshot.connections[0]; + if (!connection) throw new Error('ACP model fixture connection was not persisted'); + const credential = await policy.credentialVault.set({ + locator: { + scope: 'connection', + connectionId: connection.connectionId, + kind: 'api_key', + }, + expected: null, + secret: 'acp-fixture-key', + }); + if (credential.kind !== 'committed') throw new Error('ACP model fixture key did not commit'); + const defaulted = await policy.connectionCatalog.setDefaultTarget({ + expectedCatalogRevision: created.snapshot.revision, + target: { connectionId: connection.connectionId, modelId: model.id }, + }); + if (defaulted.kind !== 'committed') throw new Error('ACP model fixture was not selected'); + } finally { + await owner.close(); + } +} + export async function withAcpChildProcessHarness( operation: (harness: AcpChildProcessHarness) => Promise | T, options: AcpChildProcessHarnessOptions = {}, diff --git a/packages/cli/src/__tests__/acp-child-process.test.ts b/packages/cli/src/__tests__/acp-child-process.test.ts index 6789fd611b..ee88aef7d4 100644 --- a/packages/cli/src/__tests__/acp-child-process.test.ts +++ b/packages/cli/src/__tests__/acp-child-process.test.ts @@ -154,6 +154,27 @@ describe('Maka ACP child process', () => { cwd: harness.workspaceRoot, mcpServers: [], }); + assert.deepEqual( + (first.configOptions ?? []).map((option) => [option.id, option.currentValue]), + [ + ['permission_mode', 'ask'], + ['collaboration_mode', 'agent'], + ['orchestration_mode', 'default'], + ], + ); + const configured = await context.request(methods.agent.session.setConfigOption, { + sessionId: first.sessionId, + configId: 'collaboration_mode', + value: 'plan', + }); + assert.deepEqual( + (configured.configOptions ?? []).map((option) => [option.id, option.currentValue]), + [ + ['permission_mode', 'ask'], + ['collaboration_mode', 'plan'], + ['orchestration_mode', 'default'], + ], + ); const second = await context.request(methods.agent.session.new, { cwd: harness.workspaceRoot, mcpServers: [], @@ -198,6 +219,76 @@ describe('Maka ACP child process', () => { ); }); + test('configures every advertised option for a reasoning model 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 created = await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }); + assert.deepEqual( + (created.configOptions ?? []).map(({ id, currentValue }) => [id, currentValue]), + [ + ['permission_mode', 'ask'], + ['thinking_level', 'default'], + ['collaboration_mode', 'agent'], + ['orchestration_mode', 'default'], + ], + ); + const permission = created.configOptions?.find(({ id }) => id === 'permission_mode'); + assert.ok(permission?.type === 'select'); + assert.deepEqual( + permission.options.flatMap((option) => ('value' in option ? [option.value] : [])), + ['ask', 'bypass'], + ); + const thinking = created.configOptions?.find(({ id }) => id === 'thinking_level'); + assert.ok(thinking?.type === 'select'); + assert.deepEqual( + thinking.options.flatMap((option) => ('value' in option ? [option.value] : [])), + ['default', 'low', 'high'], + ); + + let configuredOptions = created.configOptions; + for (const [configId, value] of [ + ['permission_mode', 'bypass'], + ['thinking_level', 'high'], + ['collaboration_mode', 'plan'], + ['orchestration_mode', 'swarm'], + ] as const) { + configuredOptions = ( + await context.request(methods.agent.session.setConfigOption, { + sessionId: created.sessionId, + configId, + value, + }) + ).configOptions; + } + assert.deepEqual( + (configuredOptions ?? []).map(({ id, currentValue }) => [id, currentValue]), + [ + ['permission_mode', 'bypass'], + ['thinking_level', 'high'], + ['collaboration_mode', 'plan'], + ['orchestration_mode', 'swarm'], + ], + ); + }); + + await harness.closeStdin(); + assert.deepEqual(await harness.waitForExit(), { code: 0, signal: null }); + assert.equal(harness.stderr, ''); + }, + { + startRuntimeHost: true, + model: { id: 'relay-reasoner', thinkingLevels: ['low', 'high'] }, + }, + ); + }); + test('creates more than sixteen Sessions without attaching on a real Runtime Host', { timeout: 30_000, }, async () => { diff --git a/packages/cli/src/__tests__/acp-session-configuration.test.ts b/packages/cli/src/__tests__/acp-session-configuration.test.ts new file mode 100644 index 0000000000..eb1ba7d37d --- /dev/null +++ b/packages/cli/src/__tests__/acp-session-configuration.test.ts @@ -0,0 +1,212 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { SessionConfigOption, SetSessionConfigOptionRequest } from '@agentclientprotocol/sdk'; +import type { SessionCatalogProjection } from '@maka/runtime-host/protocol'; +import { + AcpSessionConfigInputError, + createAcpSessionConfigPatch, + projectAcpSessionConfigOptions, + validateAcpSessionConfigOptionRequest, +} from '../acp/session-configuration.js'; + +function selectOption( + id: string, + name: string, + category: string, + currentValue: string, + options: readonly [string, string][], +): SessionConfigOption { + return { + type: 'select', + id, + name, + category, + currentValue, + options: options.map(([value, optionName]) => ({ value, name: optionName })), + }; +} + +function sessionProjection( + overrides: Partial = {}, +): SessionCatalogProjection { + return { + id: 'session-1', + revision: 1, + workspace: { target: { kind: 'host_path', path: '/tmp' }, hostCwd: '/tmp' }, + createdAt: 1, + activityAt: 2, + name: 'Session', + isFlagged: false, + isArchived: false, + labels: [], + labelsTruncated: false, + hasUnread: false, + status: 'active', + backend: 'ai-sdk', + llmConnectionId: 'connection-1', + llmConnectionSlug: 'openai-main', + connectionLocked: true, + model: 'gpt-5', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + ...overrides, + } as SessionCatalogProjection; +} + +test('projects the ordered ACP configuration options', () => { + assert.deepEqual(projectAcpSessionConfigOptions(sessionProjection(), ['off', 'low', 'high']), [ + selectOption('permission_mode', 'Permission mode', '_maka/permission_mode', 'ask', [ + ['ask', 'Ask'], + ['bypass', 'Bypass'], + ]), + selectOption('thinking_level', 'Thinking level', 'thought_level', 'default', [ + ['default', 'Default'], + ['off', 'Off'], + ['low', 'Low'], + ['high', 'High'], + ]), + selectOption('collaboration_mode', 'Collaboration mode', 'mode', 'agent', [ + ['agent', 'Agent'], + ['plan', 'Plan'], + ]), + selectOption( + 'orchestration_mode', + 'Orchestration mode', + '_maka/orchestration_mode', + 'default', + [ + ['default', 'Default'], + ['swarm', 'Swarm'], + ['graph', 'Graph'], + ], + ), + ]); +}); + +test('projects canonical values and keeps reserved explore as a current value only', () => { + for (const [configId, values, field] of [ + ['permission_mode', ['explore', 'ask', 'bypass'], 'permissionMode'], + ['collaboration_mode', ['agent', 'plan'], 'collaborationMode'], + ['orchestration_mode', ['default', 'swarm', 'graph'], 'orchestrationMode'], + ] as const) { + for (const value of values) { + const overrides: Partial = + field === 'permissionMode' + ? { permissionMode: value as SessionCatalogProjection['permissionMode'] } + : field === 'collaborationMode' + ? { collaborationMode: value as SessionCatalogProjection['collaborationMode'] } + : { orchestrationMode: value as SessionCatalogProjection['orchestrationMode'] }; + const option = projectAcpSessionConfigOptions(sessionProjection(overrides), ['low'])[ + configId === 'permission_mode' ? 0 : configId === 'collaboration_mode' ? 2 : 3 + ]; + assert.equal(option.currentValue, value); + } + } + const permission = projectAcpSessionConfigOptions( + sessionProjection({ permissionMode: 'explore' }), + ['low'], + )[0]; + assert.equal(permission.currentValue, 'explore'); + assert.deepEqual(permission.options, [ + { value: 'ask', name: 'Ask' }, + { value: 'bypass', name: 'Bypass' }, + ]); +}); + +test('projects only the current model thinking levels and omits the switch when empty', () => { + const levels = ['minimal', 'high', 'max'] as const; + const option = projectAcpSessionConfigOptions( + sessionProjection({ thinkingLevel: 'high' }), + levels, + )[1]; + assert.deepEqual( + option, + selectOption('thinking_level', 'Thinking level', 'thought_level', 'high', [ + ['default', 'Default'], + ['minimal', 'Minimal'], + ['high', 'High'], + ['max', 'Max'], + ]), + ); + assert.deepEqual( + projectAcpSessionConfigOptions(sessionProjection(), []).map(({ id }) => id), + ['permission_mode', 'collaboration_mode', 'orchestration_mode'], + ); +}); + +test('validates requests and creates exact one-field patches', () => { + const cases = [ + ['permission_mode', 'ask', { permissionMode: 'ask' }], + ['permission_mode', 'bypass', { permissionMode: 'bypass' }], + ['thinking_level', 'default', { thinkingLevel: null }], + ['thinking_level', 'off', { thinkingLevel: 'off' }], + ['thinking_level', 'minimal', { thinkingLevel: 'minimal' }], + ['thinking_level', 'low', { thinkingLevel: 'low' }], + ['thinking_level', 'medium', { thinkingLevel: 'medium' }], + ['thinking_level', 'high', { thinkingLevel: 'high' }], + ['thinking_level', 'xhigh', { thinkingLevel: 'xhigh' }], + ['thinking_level', 'max', { thinkingLevel: 'max' }], + ['collaboration_mode', 'agent', { collaborationMode: 'agent' }], + ['collaboration_mode', 'plan', { collaborationMode: 'plan' }], + ['orchestration_mode', 'default', { orchestrationMode: 'default' }], + ['orchestration_mode', 'swarm', { orchestrationMode: 'swarm' }], + ['orchestration_mode', 'graph', { orchestrationMode: 'graph' }], + ] as const; + for (const [configId, value, patch] of cases) { + const request = { sessionId: 'session-1', configId, value } as SetSessionConfigOptionRequest; + validateAcpSessionConfigOptionRequest(request); + assert.deepEqual(createAcpSessionConfigPatch(request), patch); + } +}); + +test('rejects unsupported config ids, boolean values, and unsupported strings', () => { + for (const [request, field, reason] of [ + [{ sessionId: 'session-1', configId: 'unknown', value: 'ask' }, 'configId', 'unsupported'], + [ + { sessionId: 'session-1', configId: 'permission_mode', type: 'boolean', value: true }, + 'value', + 'invalid_type', + ], + [ + { sessionId: 'session-1', configId: 'permission_mode', value: 'invalid' }, + 'value', + 'unsupported', + ], + [ + { sessionId: 'session-1', configId: 'permission_mode', value: 'explore' }, + 'value', + 'unsupported', + ], + ] as const) { + assert.throws( + () => validateAcpSessionConfigOptionRequest(request as SetSessionConfigOptionRequest), + (error: unknown) => { + assert.ok(error instanceof AcpSessionConfigInputError); + if (!(error instanceof AcpSessionConfigInputError)) return false; + assert.equal(error.field, field); + assert.equal(error.reason, reason); + return true; + }, + ); + } +}); diff --git a/packages/cli/src/__tests__/acp-session-registry.test.ts b/packages/cli/src/__tests__/acp-session-registry.test.ts index 25a5a81419..bb37240505 100644 --- a/packages/cli/src/__tests__/acp-session-registry.test.ts +++ b/packages/cli/src/__tests__/acp-session-registry.test.ts @@ -22,14 +22,80 @@ 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 { + RequestError, + type NewSessionRequest, + type SessionConfigOption, + type SetSessionConfigOptionRequest, +} from '@agentclientprotocol/sdk'; +import { THINKING_LEVELS, type ThinkingLevel } from '@maka/core/model-thinking'; +import { + RuntimeHostOperationError, + RuntimeHostRequestInterruptedError, +} from '@maka/runtime-host/client'; +import { + SESSION_CATALOG_CWD_MAX_BYTES, + type SessionCatalogProjection, +} 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; +const DEFAULT_CONFIG_OPTIONS: Array> = [ + { + type: 'select', + id: 'permission_mode', + name: 'Permission mode', + category: '_maka/permission_mode', + currentValue: 'ask', + options: [ + { value: 'ask', name: 'Ask' }, + { value: 'bypass', name: 'Bypass' }, + ], + }, + { + type: 'select', + id: 'thinking_level', + name: 'Thinking level', + category: 'thought_level', + currentValue: 'default', + options: [ + { value: 'default', name: 'Default' }, + { value: 'off', name: 'Off' }, + { value: 'minimal', name: 'Minimal' }, + { value: 'low', name: 'Low' }, + { value: 'medium', name: 'Medium' }, + { value: 'high', name: 'High' }, + { value: 'xhigh', name: 'Extra high' }, + { value: 'max', name: 'Max' }, + ], + }, + { + type: 'select', + id: 'collaboration_mode', + name: 'Collaboration mode', + category: 'mode', + currentValue: 'agent', + options: [ + { value: 'agent', name: 'Agent' }, + { value: 'plan', name: 'Plan' }, + ], + }, + { + type: 'select', + id: 'orchestration_mode', + name: 'Orchestration mode', + category: '_maka/orchestration_mode', + currentValue: 'default', + options: [ + { value: 'default', name: 'Default' }, + { value: 'swarm', name: 'Swarm' }, + { value: 'graph', name: 'Graph' }, + ], + }, +]; + describe('ACP Session registry', () => { test('does not connect when disposed before a Session method is used', async () => { let connectCalls = 0; @@ -55,6 +121,15 @@ describe('ACP Session registry', () => { for (const [operation, request] of [ ['session.create', () => registry.create({ cwd: '/workspace', mcpServers: [] })], ['session.catalog.query', () => registry.list({})], + [ + 'session.configuration.update', + () => + registry.setConfigOption({ + sessionId: 'session-closed', + configId: 'permission_mode', + value: 'bypass', + }), + ], ] as const) { await assert.rejects(request(), (error: unknown) => { assert.ok(error instanceof RequestError); @@ -142,11 +217,14 @@ describe('ACP Session registry', () => { sessions: [], nextCursor: null, } - : {}, + : catalogSession('session-concurrent'), }), ); - assert.deepEqual(await create, { sessionId: 'session-concurrent' }); + assert.deepEqual(await create, { + sessionId: 'session-concurrent', + configOptions: DEFAULT_CONFIG_OPTIONS, + }); assert.deepEqual(await list, { sessions: [] }); assert.equal(connectCalls, 1); await registry.dispose(); @@ -226,8 +304,9 @@ describe('ACP Session registry', () => { const connection = fakeConnection({ request: async (operation, input) => { assert.equal(operation, 'session.create'); - createdSessionIds.push((input as { sessionId: string }).sessionId); - return {}; + const sessionId = (input as { sessionId: string }).sessionId; + createdSessionIds.push(sessionId); + return catalogSession(sessionId); }, }); return { @@ -253,6 +332,756 @@ describe('ACP Session registry', () => { await registry.dispose(); }); + test('returns projected configuration and owns only a representable successful create', async () => { + const requests: Array<{ operation: string; input: unknown }> = []; + let subscriptionOpens = 0; + const created = catalogSession('session-configured', '/workspace', { + thinkingLevel: 'high', + permissionMode: 'explore', + collaborationMode: 'plan', + orchestrationMode: 'swarm', + }); + const registry = new AcpSessionRegistry({ + connect: async () => { + const connection = fakeConnection({ + thinkingLevels: ['low', 'high'], + request: async (operation, input) => { + requests.push({ operation, input }); + return created; + }, + }); + return { + ...connection, + openSessionSubscriptionOnce: async () => { + subscriptionOpens += 1; + throw new Error('PR 2 must not open a subscription'); + }, + } as AcpSessionRegistryConnection; + }, + newSessionId: () => 'session-configured', + }); + + const response = await registry.create({ cwd: '/workspace', mcpServers: [] }); + + assert.deepEqual(response, { + sessionId: 'session-configured', + configOptions: configOptions( + { + permission_mode: 'explore', + thinking_level: 'high', + collaboration_mode: 'plan', + orchestration_mode: 'swarm', + }, + ['low', 'high'], + ), + }); + assert.deepEqual(requests, [ + { + operation: 'session.create', + input: { + sessionId: 'session-configured', + workspace: { kind: 'host_path', path: '/workspace' }, + modelTarget: { kind: 'default' }, + }, + }, + ]); + assert.equal(subscriptionOpens, 0); + await registry.dispose(); + }); + + test('omits thinking configuration when the selected model declares no levels', async () => { + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + thinkingLevels: [], + request: async (operation) => { + assert.equal(operation, 'session.create'); + return catalogSession('session-no-thinking'); + }, + }), + newSessionId: () => 'session-no-thinking', + }); + + const response = await registry.create({ cwd: '/workspace', mcpServers: [] }); + + assert.deepEqual( + response.configOptions?.map(({ id }) => id), + ['permission_mode', 'collaboration_mode', 'orchestration_mode'], + ); + await registry.dispose(); + }); + + test('does not grant ownership by listing a Session', async () => { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [catalogSession('listed-session')], + nextCursor: null, + }; + }, + }), + }); + await registry.list({}); + + await assertInvalidParams( + registry.setConfigOption({ + sessionId: 'listed-session', + configId: 'permission_mode', + value: 'bypass', + }), + { reason: 'unknown_session' }, + ); + assert.equal(requests, 1); + await registry.dispose(); + }); + + test('does not grant ownership after failed or legacy creates', async () => { + for (const [name, createOutcome] of [ + [ + 'failed', + new RuntimeHostOperationError('session.create', 'operation_conflict', 'create failed'), + ], + [ + 'legacy', + { + kind: 'unsupported_legacy_record', + id: 'session-legacy', + revision: 1, + reason: 'not_wire_representable', + }, + ], + ] as const) { + let requests = 0; + const sessionId = `session-${name}`; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + if (createOutcome instanceof Error) throw createOutcome; + return createOutcome; + }, + }), + newSessionId: () => sessionId, + }); + + await assert.rejects(registry.create({ cwd: '/workspace', mcpServers: [] })); + await assertInvalidParams( + registry.setConfigOption({ + sessionId, + configId: 'permission_mode', + value: 'bypass', + }), + { reason: 'unknown_session' }, + ); + assert.equal(requests, 1); + await registry.dispose(); + } + }); + + test('rejects non-owned and invalid configuration requests before Host I/O', async () => { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async () => { + requests += 1; + return catalogSession('session-owned'); + }, + }), + newSessionId: () => 'session-owned', + }); + + await assertInvalidParams( + registry.setConfigOption({ + sessionId: 'session-unowned', + configId: 'permission_mode', + value: 'bypass', + }), + { reason: 'unknown_session' }, + ); + assert.equal(requests, 0); + + await registry.create({ cwd: '/workspace', mcpServers: [] }); + assert.equal(requests, 1); + for (const [request, data] of [ + [ + { sessionId: 'session-owned', configId: 'unknown', value: 'bypass' }, + { field: 'configId', reason: 'unsupported' }, + ], + [ + { + sessionId: 'session-owned', + configId: 'permission_mode', + value: true, + type: 'boolean', + }, + { field: 'value', reason: 'invalid_type' }, + ], + [ + { sessionId: 'session-owned', configId: 'permission_mode', value: 'maybe' }, + { field: 'value', reason: 'unsupported' }, + ], + ] as const) { + await assertInvalidParams( + registry.setConfigOption(request as SetSessionConfigOptionRequest), + data, + ); + assert.equal(requests, 1); + } + await registry.dispose(); + }); + + test('updates one configuration field with the latest revision and returns committed options', async () => { + const current = catalogSession('session-cas', '/workspace', { + revision: 7, + thinkingLevel: 'minimal', + }); + const committed = catalogSession('session-cas', '/workspace', { + revision: 8, + permissionMode: 'bypass', + thinkingLevel: 'high', + }); + const requests: Array<{ operation: string; input: unknown }> = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + requests.push({ operation, input }); + if (operation === 'session.create') return catalogSession('session-cas'); + if (operation === 'session.catalog.query') { + return { kind: 'session', session: current }; + } + return { kind: 'committed', session: committed }; + }, + }), + newSessionId: () => 'session-cas', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + const response = await registry.setConfigOption({ + sessionId: 'session-cas', + configId: 'permission_mode', + value: 'bypass', + }); + + assert.deepEqual(requests.slice(1), [ + { + operation: 'session.catalog.query', + input: { kind: 'get', sessionId: 'session-cas' }, + }, + { + operation: 'session.configuration.update', + input: { + sessionId: 'session-cas', + expectedRevision: 7, + patch: { permissionMode: 'bypass' }, + }, + }, + ]); + assert.deepEqual(response, { + configOptions: configOptions({ permission_mode: 'bypass', thinking_level: 'high' }), + }); + await registry.dispose(); + }); + + test('rereads the Session after one revision conflict before retrying', async () => { + const requests: Array<{ operation: string; input: unknown }> = []; + let reads = 0; + let updates = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + requests.push({ operation, input }); + if (operation === 'session.create') return catalogSession('session-retry'); + if (operation === 'session.catalog.query') { + reads += 1; + return { + kind: 'session', + session: catalogSession('session-retry', '/workspace', { + revision: reads, + collaborationMode: reads === 1 ? 'agent' : 'plan', + }), + }; + } + updates += 1; + return updates === 1 + ? { kind: 'revision_conflict', expectedRevision: 1, actualRevision: 2 } + : { + kind: 'committed', + session: catalogSession('session-retry', '/workspace', { + revision: 3, + permissionMode: 'bypass', + collaborationMode: 'plan', + }), + }; + }, + }), + newSessionId: () => 'session-retry', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await registry.setConfigOption({ + sessionId: 'session-retry', + configId: 'permission_mode', + value: 'bypass', + }); + + assert.deepEqual( + requests.slice(1).map(({ operation }) => operation), + [ + 'session.catalog.query', + 'session.configuration.update', + 'session.catalog.query', + 'session.configuration.update', + ], + ); + assert.deepEqual(requests[4]?.input, { + sessionId: 'session-retry', + expectedRevision: 2, + patch: { permissionMode: 'bypass' }, + }); + await registry.dispose(); + }); + + test('concurrent different-field changes converge through one-field CAS patches', async () => { + const firstReads = deferred<{ kind: 'session'; session: SessionCatalogProjection }>(); + const requests: Array<{ operation: string; input: unknown }> = []; + let reads = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + requests.push({ operation, input }); + if (operation === 'session.create') return catalogSession('session-converge'); + if (operation === 'session.catalog.query') { + reads += 1; + if (reads <= 2) { + if (reads === 2) { + firstReads.resolve({ + kind: 'session', + session: catalogSession('session-converge'), + }); + } + return firstReads.promise; + } + return { + kind: 'session', + session: catalogSession('session-converge', '/workspace', { + revision: 2, + permissionMode: 'bypass', + }), + }; + } + const patch = (input as { patch: Record }).patch; + if ('permissionMode' in patch) { + return { + kind: 'committed', + session: catalogSession('session-converge', '/workspace', { + revision: 2, + permissionMode: 'bypass', + }), + }; + } + if ((input as { expectedRevision: number }).expectedRevision === 1) { + return { kind: 'revision_conflict', expectedRevision: 1, actualRevision: 2 }; + } + return { + kind: 'committed', + session: catalogSession('session-converge', '/workspace', { + revision: 3, + permissionMode: 'bypass', + collaborationMode: 'plan', + }), + }; + }, + }), + newSessionId: () => 'session-converge', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + const [permission, collaboration] = await Promise.all([ + registry.setConfigOption({ + sessionId: 'session-converge', + configId: 'permission_mode', + value: 'bypass', + }), + registry.setConfigOption({ + sessionId: 'session-converge', + configId: 'collaboration_mode', + value: 'plan', + }), + ]); + + const updates = requests.filter( + ({ operation }) => operation === 'session.configuration.update', + ); + assert.deepEqual( + updates.map(({ input }) => (input as { patch: unknown }).patch), + [{ permissionMode: 'bypass' }, { collaborationMode: 'plan' }, { collaborationMode: 'plan' }], + ); + assert.deepEqual(permission, { + configOptions: configOptions({ permission_mode: 'bypass' }), + }); + assert.deepEqual(collaboration, { + configOptions: configOptions({ + permission_mode: 'bypass', + collaboration_mode: 'plan', + }), + }); + await registry.dispose(); + }); + + test('stops after three revision conflicts without a fourth Host operation', async () => { + const requests: Array<{ operation: string; input: unknown }> = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + requests.push({ operation, input }); + if (operation === 'session.create') return catalogSession('session-conflicts'); + if (operation === 'session.catalog.query') { + return { kind: 'session', session: catalogSession('session-conflicts') }; + } + return { kind: 'revision_conflict', expectedRevision: 1, actualRevision: 2 }; + }, + }), + newSessionId: () => 'session-conflicts', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await assert.rejects( + registry.setConfigOption({ + sessionId: 'session-conflicts', + configId: 'thinking_level', + value: 'off', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'revision_conflict', + attempts: 3, + }); + return true; + }, + ); + assert.deepEqual( + requests.slice(1).map(({ operation }) => operation), + [ + 'session.catalog.query', + 'session.configuration.update', + 'session.catalog.query', + 'session.configuration.update', + 'session.catalog.query', + 'session.configuration.update', + ], + ); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(requests.length, 7); + await registry.dispose(); + }); + + test('rejects invalid, missing, and legacy catalog lookup results with stable errors', async () => { + for (const [name, result, acpCode, data] of [ + [ + 'invalid', + { + kind: 'page', + revision: SESSION_REVISION, + sessions: [], + nextCursor: null, + }, + -32603, + { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'catalog_read_failure', + reason: 'invalid_projection', + }, + ], + [ + 'missing', + { kind: 'session', session: null }, + -32602, + { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'not_found', + }, + ], + [ + 'legacy', + { + kind: 'session', + session: { + kind: 'unsupported_legacy_record', + id: 'session-legacy', + revision: 1, + reason: 'not_wire_representable', + }, + }, + -32603, + { + source: 'runtime_host', + operation: 'session.catalog.query', + code: 'unsupported_session_projection', + }, + ], + ] as const) { + const sessionId = `session-${name}`; + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + requests += 1; + return operation === 'session.create' ? catalogSession(sessionId) : result; + }, + }), + newSessionId: () => sessionId, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await assert.rejects( + registry.setConfigOption({ + sessionId, + configId: 'permission_mode', + value: 'bypass', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, acpCode); + assert.deepEqual(error.data, data); + return true; + }, + ); + assert.equal(requests, 2); + await registry.dispose(); + } + }); + + test('maps configuration Host failures without retrying them', async () => { + for (const [hostError, acpCode, data] of [ + [ + new RuntimeHostOperationError( + 'session.configuration.update', + 'invalid_request', + 'invalid update', + ), + -32602, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'invalid_request', + }, + ], + [ + new RuntimeHostOperationError( + 'session.configuration.update', + 'not_found', + 'missing Session', + ), + -32602, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'not_found', + }, + ], + ...(['session_busy', 'operation_conflict', 'commit_outcome_unknown'] as const).map( + (code) => + [ + new RuntimeHostOperationError('session.configuration.update', code, 'update failed'), + -32603, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code, + }, + ] as const, + ), + [ + new RuntimeHostRequestInterruptedError( + 'session.configuration.update', + 'command', + 'dispatched', + 'connection_lost', + ), + -32603, + { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'request_interrupted', + reason: 'connection_lost', + dispatch: 'dispatched', + }, + ], + ] as const) { + let requests = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + requests += 1; + if (operation === 'session.create') return catalogSession('session-errors'); + if (operation === 'session.catalog.query') { + return { kind: 'session', session: catalogSession('session-errors') }; + } + throw hostError; + }, + }), + newSessionId: () => 'session-errors', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await assert.rejects( + registry.setConfigOption({ + sessionId: 'session-errors', + configId: 'permission_mode', + value: 'bypass', + }), + (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, acpCode); + assert.deepEqual(error.data, data); + return true; + }, + ); + assert.equal(requests, 3); + await registry.dispose(); + } + }); + + test('does not start an update after disposal begins during its catalog read', async () => { + const catalogRead = deferred<{ kind: 'session'; session: SessionCatalogProjection }>(); + let catalogReads = 0; + let updates = 0; + let closeCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession('session-closing'); + if (operation === 'session.catalog.query') { + catalogReads += 1; + return catalogRead.promise; + } + updates += 1; + return { + kind: 'committed', + session: catalogSession('session-closing', '/workspace', { revision: 2 }), + }; + }, + close: async () => { + closeCalls += 1; + }, + }), + newSessionId: () => 'session-closing', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const update = registry.setConfigOption({ + sessionId: 'session-closing', + configId: 'permission_mode', + value: 'bypass', + }); + await waitFor(() => catalogReads === 1); + + const dispose = registry.dispose(); + catalogRead.resolve({ + kind: 'session', + session: catalogSession('session-closing'), + }); + + await assert.rejects(update, (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'registry_closed', + }); + return true; + }); + await dispose; + assert.equal(updates, 0); + assert.equal(closeCalls, 1); + }); + + test('does not reread after a held update conflicts during disposal', async () => { + const heldUpdate = deferred<{ + kind: 'revision_conflict'; + expectedRevision: number; + actualRevision: number; + }>(); + let catalogReads = 0; + let updates = 0; + let closeCalls = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession('session-conflict-closing'); + if (operation === 'session.catalog.query') { + catalogReads += 1; + if (catalogReads === 1) { + return { + kind: 'session', + session: catalogSession('session-conflict-closing'), + }; + } + throw new RuntimeHostRequestInterruptedError( + 'session.catalog.query', + 'query', + 'dispatched', + 'connection_lost', + ); + } + updates += 1; + return heldUpdate.promise; + }, + close: async () => { + closeCalls += 1; + }, + }), + newSessionId: () => 'session-conflict-closing', + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const update = registry.setConfigOption({ + sessionId: 'session-conflict-closing', + configId: 'permission_mode', + value: 'bypass', + }); + await waitFor(() => updates === 1); + + const dispose = registry.dispose(); + heldUpdate.resolve({ kind: 'revision_conflict', expectedRevision: 1, actualRevision: 2 }); + + await assert.rejects(update, (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32603); + assert.deepEqual(error.data, { + source: 'runtime_host', + operation: 'session.configuration.update', + code: 'registry_closed', + }); + return true; + }); + await dispose; + assert.equal(catalogReads, 1); + assert.equal(updates, 1); + assert.equal(closeCalls, 1); + }); + test('rejects unsupported creation inputs before touching Runtime Host', async () => { let requests = 0; const registry = new AcpSessionRegistry({ @@ -356,7 +1185,10 @@ describe('ACP Session registry', () => { kind: 'page', revision: SESSION_REVISION, sessions: [ - catalogSession('other', join(root, 'other'), 'Other', 1_000), + catalogSession('other', join(root, 'other'), { + name: 'Other', + activityAt: 1_000, + }), { kind: 'unsupported_legacy_record', id: 'legacy', @@ -371,13 +1203,14 @@ describe('ACP Session registry', () => { kind: 'page', revision: SESSION_REVISION, sessions: [ - catalogSession('matching', canonicalWorkspace, 'Matching session', 2_000), - catalogSession( - 'undated', - canonicalWorkspace, - 'Out-of-range activity', - Number.MAX_SAFE_INTEGER, - ), + catalogSession('matching', canonicalWorkspace, { + name: 'Matching session', + activityAt: 2_000, + }), + catalogSession('undated', canonicalWorkspace, { + name: 'Out-of-range activity', + activityAt: Number.MAX_SAFE_INTEGER, + }), ], nextCursor: null, }; @@ -567,22 +1400,73 @@ function fakeConnection( overrides: { request?: (operation: string, input: unknown) => Promise; close?: () => Promise; + thinkingLevels?: readonly ThinkingLevel[]; } = {}, ): AcpSessionRegistryConnection { return { - request: overrides.request ?? (async () => ({})), + request: async (operation, input) => + operation === 'connection.catalog.query' + ? connectionCatalogPage(overrides.thinkingLevels ?? THINKING_LEVELS) + : (overrides.request?.(operation, input) ?? {}), close: overrides.close ?? (async () => undefined), } as AcpSessionRegistryConnection; } -function catalogSession(id: string, cwd: string, name: string, activityAt: number) { +function connectionCatalogPage(thinkingLevels: readonly ThinkingLevel[]) { + return { + kind: 'page' as const, + revision: 1, + defaultTarget: { connectionId: 'connection-1', model: 'default' }, + connectionCount: 1, + items: [ + { + kind: 'connection' as const, + connectionIndex: 0, + connectionId: 'connection-1', + revision: 1, + slug: 'default', + name: 'Default', + providerType: 'openai' as const, + enabled: true, + enabledModelIdCount: 1, + modelCount: 0, + catalogEntryCount: 1, + }, + { + kind: 'enabled_model_id' as const, + connectionIndex: 0, + itemIndex: 0, + modelId: 'default', + }, + { + kind: 'catalog_entry' as const, + connectionIndex: 0, + itemIndex: 0, + entry: { + id: 'default', + canUseAsChatDefault: true, + isDefault: true, + supportsVision: false, + thinkingLevels, + }, + }, + ], + nextCursor: null, + }; +} + +function catalogSession( + id: string, + cwd = '/workspace', + overrides: Partial = {}, +): SessionCatalogProjection { return { id, revision: 1, workspace: { target: { kind: 'host_path', path: cwd }, hostCwd: cwd }, createdAt: 1, - activityAt, - name, + activityAt: 1, + name: id, isFlagged: false, isArchived: false, labels: [], @@ -590,15 +1474,55 @@ function catalogSession(id: string, cwd: string, name: string, activityAt: numbe hasUnread: false, status: 'active', backend: 'ai-sdk', + llmConnectionId: 'connection-1', llmConnectionSlug: 'default', connectionLocked: false, model: 'default', - permissionMode: 'default', - collaborationMode: 'default', + permissionMode: 'ask', + collaborationMode: 'agent', orchestrationMode: 'default', + ...overrides, }; } +function configOptions( + values: Partial< + Record< + 'permission_mode' | 'thinking_level' | 'collaboration_mode' | 'orchestration_mode', + string + > + >, + thinkingLevels: readonly ThinkingLevel[] = THINKING_LEVELS, +): SessionConfigOption[] { + const options: SessionConfigOption[] = structuredClone(DEFAULT_CONFIG_OPTIONS); + const thinking = options.find(({ id }) => id === 'thinking_level'); + if (thinking?.type === 'select') { + thinking.options = thinking.options.flatMap((option) => + 'value' in option && + (option.value === 'default' || thinkingLevels.includes(option.value as ThinkingLevel)) + ? [option] + : [], + ); + } + for (const option of options) { + if (option.type !== 'select') continue; + option.currentValue = values[option.id as keyof typeof values] ?? option.currentValue; + } + return options; +} + +async function assertInvalidParams( + promise: Promise, + data: Record, +): Promise { + await assert.rejects(promise, (error: unknown) => { + assert.ok(error instanceof RequestError); + assert.equal(error.code, -32602); + assert.deepEqual(error.data, data); + return true; + }); +} + async function waitFor(predicate: () => boolean): Promise { for (let attempt = 0; attempt < 100; attempt += 1) { if (predicate()) return; diff --git a/packages/cli/src/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index 4e703dd30e..0114d9234e 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -18,9 +18,10 @@ */ import assert from 'node:assert/strict'; -import { Readable, Writable } from 'node:stream'; +import { PassThrough, Readable, Writable } from 'node:stream'; import { describe, test } from 'node:test'; import type { RuntimeHostConnection } from '@maka/runtime-host/client'; +import type { SessionCatalogProjection } from '@maka/runtime-host/protocol'; import { runMakaAcpStdioServer } from '../acp/stdio-server.js'; describe('Maka ACP stdio server', () => { @@ -78,43 +79,121 @@ 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 () => { + test('serializes Session creation and configuration through the Runtime Host catalog', async () => { const lifecycle: string[] = []; + let created: SessionCatalogProjection | undefined; const connection = { - request: async (operation: string) => { + request: async (operation: string, input: unknown) => { lifecycle.push(operation); - return {}; + if (operation === 'session.create') { + const { sessionId } = input as { sessionId: string }; + created = sessionProjection({ id: sessionId }); + return created; + } + if (operation === 'connection.catalog.query') return connectionCatalogPage(); + if (operation === 'session.catalog.query') { + assert.ok(created); + return { kind: 'session', session: created }; + } + if (operation === 'session.configuration.update') { + assert.ok(created); + return { + kind: 'committed', + session: sessionProjection({ + id: created.id, + revision: created.revision + 1, + collaborationMode: 'plan', + }), + }; + } + assert.fail(`Unexpected Runtime Host operation: ${operation}`); }, 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 }, + const stdin = new PassThrough(); + const harness = createHarness([], { stdin, connection }); + const run = harness.run(); + stdin.write( + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1 }, + })}\n`, + ); + stdin.write( + `${JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'session/new', + params: { cwd: '/workspace', mcpServers: [] }, + })}\n`, + ); + await waitFor(() => created !== undefined); + stdin.end( + `${JSON.stringify({ + jsonrpc: '2.0', + id: 3, + method: 'session/set_config_option', + params: { + sessionId: created!.id, + configId: 'collaboration_mode', + value: 'plan', + }, + })}\n`, ); - assert.equal(await harness.run(), 0); - const response = harness - .stdoutMessages() - .find((message) => (message as { id?: unknown }).id === 2) as { - result?: { sessionId?: unknown }; + assert.equal(await run, 0); + const responses = new Map( + harness + .stdoutMessages() + .map((message) => [(message as { id?: unknown }).id, message] as const), + ); + const createdResponse = responses.get(2) as { + result?: { sessionId?: unknown; configOptions?: unknown[] }; + }; + assert.equal(createdResponse.result?.sessionId, created?.id); + assert.deepEqual( + createdResponse.result?.configOptions?.map((option) => (option as { id?: unknown }).id), + ['permission_mode', 'thinking_level', 'collaboration_mode', 'orchestration_mode'], + ); + const configuredResponse = responses.get(3) as { + result?: { + configOptions?: Array<{ id?: unknown; currentValue?: unknown }>; + }; }; - assert.equal(typeof response.result?.sessionId, 'string'); - assert.deepEqual(lifecycle, ['session.create', 'connection.close']); + assert.deepEqual( + configuredResponse.result?.configOptions?.find(({ id }) => id === 'collaboration_mode'), + { + type: 'select', + id: 'collaboration_mode', + name: 'Collaboration mode', + category: 'mode', + currentValue: 'plan', + options: [ + { value: 'agent', name: 'Agent' }, + { value: 'plan', name: 'Plan' }, + ], + }, + ); + assert.deepEqual(lifecycle, [ + 'session.create', + 'connection.catalog.query', + 'session.catalog.query', + 'session.configuration.update', + 'connection.catalog.query', + 'connection.close', + ]); + assert.equal('subscribe' in connection, false); + assert.ok(lifecycle.every((operation) => operation !== 'session.catalog.subscribe')); + assert.ok( + harness.stdoutMessages().every((message) => { + const record = message as { jsonrpc?: unknown }; + return record.jsonrpc === '2.0'; + }), + ); }); test('returns a Host connection failure from the Session request and keeps serving ACP', async () => { @@ -247,3 +326,82 @@ function createHarness( .map((line) => JSON.parse(line) as unknown), }; } + +function connectionCatalogPage() { + return { + kind: 'page' as const, + revision: 1, + defaultTarget: { connectionId: 'connection-1', model: 'default' }, + connectionCount: 1, + items: [ + { + kind: 'connection' as const, + connectionIndex: 0, + connectionId: 'connection-1', + revision: 1, + slug: 'default', + name: 'Default', + providerType: 'openai' as const, + enabled: true, + enabledModelIdCount: 1, + modelCount: 0, + catalogEntryCount: 1, + }, + { + kind: 'enabled_model_id' as const, + connectionIndex: 0, + itemIndex: 0, + modelId: 'default', + }, + { + kind: 'catalog_entry' as const, + connectionIndex: 0, + itemIndex: 0, + entry: { + id: 'default', + canUseAsChatDefault: true, + isDefault: true, + supportsVision: false, + thinkingLevels: ['low', 'high'] as const, + }, + }, + ], + nextCursor: null, + }; +} + +function sessionProjection( + overrides: Partial = {}, +): SessionCatalogProjection { + return { + id: 'session-1', + revision: 1, + workspace: { target: { kind: 'host_path', path: '/workspace' }, hostCwd: '/workspace' }, + createdAt: 1, + activityAt: 1, + name: 'Session', + isFlagged: false, + isArchived: false, + labels: [], + labelsTruncated: false, + hasUnread: false, + status: 'active', + backend: 'ai-sdk', + llmConnectionId: 'connection-1', + llmConnectionSlug: 'default', + connectionLocked: false, + model: 'default', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + ...overrides, + }; +} + +async function waitFor(predicate: () => boolean): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) return; + await new Promise((resolve) => setImmediate(resolve)); + } + assert.fail('condition was not reached'); +} diff --git a/packages/cli/src/__tests__/runtime-host-session-update.test.ts b/packages/cli/src/__tests__/runtime-host-session-update.test.ts new file mode 100644 index 0000000000..d184952858 --- /dev/null +++ b/packages/cli/src/__tests__/runtime-host-session-update.test.ts @@ -0,0 +1,197 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { RuntimeHostConnection } from '@maka/runtime-host/client'; +import type { SessionCatalogProjection, SessionUpdateResult } from '@maka/runtime-host/protocol'; +import { + RuntimeHostSessionUpdateError, + getRuntimeHostSession, + updateRuntimeHostSession, +} from '../runtime-host-session-update.js'; + +test('reads the latest Session and returns the committed update', async () => { + const requests: unknown[] = []; + const connection = fakeConnection(async (_operation, input) => { + requests.push(input); + return { kind: 'session', session: sessionProjection({ revision: 7 }) }; + }); + + const committed = await updateRuntimeHostSession( + connection, + 'session-1', + async (current) => { + assert.equal(current.revision, 7); + return { + kind: 'committed', + session: sessionProjection({ revision: 8, permissionMode: 'bypass' }), + }; + }, + { operation: 'session.configuration.update' }, + ); + + assert.equal(committed.revision, 8); + assert.equal(committed.permissionMode, 'bypass'); + assert.deepEqual(requests, [{ kind: 'get', sessionId: 'session-1' }]); +}); + +test('rereads after a revision conflict and stops after three attempts', async () => { + let reads = 0; + let updates = 0; + const connection = fakeConnection(async () => { + reads += 1; + return { kind: 'session', session: sessionProjection({ revision: reads }) }; + }); + + await assert.rejects( + updateRuntimeHostSession( + connection, + 'session-1', + async () => { + updates += 1; + return { + kind: 'revision_conflict', + expectedRevision: updates, + actualRevision: updates + 1, + }; + }, + { operation: 'session.configuration.update' }, + ), + (error: unknown) => { + assert.ok(error instanceof RuntimeHostSessionUpdateError); + assert.equal(error.operation, 'session.configuration.update'); + assert.equal(error.reason, 'revision_conflict'); + assert.equal(error.attempts, 3); + return true; + }, + ); + assert.equal(reads, 3); + assert.equal(updates, 3); +}); + +test('reports missing and unsupported Session projections without caller-specific errors', async () => { + const missing = fakeConnection(async () => ({ kind: 'session', session: null })); + await assert.rejects( + updateRuntimeHostSession( + missing, + 'session-1', + async (): Promise => { + assert.fail('missing Session must not reach the update'); + }, + { operation: 'session.configuration.update' }, + ), + (error: unknown) => { + assert.ok(error instanceof RuntimeHostSessionUpdateError); + assert.equal(error.operation, 'session.catalog.query'); + assert.equal(error.reason, 'not_found'); + return true; + }, + ); + + for (const [name, result, reason] of [ + [ + 'legacy', + { + kind: 'session', + session: { + kind: 'unsupported_legacy_record', + id: 'session-1', + revision: 1, + reason: 'not_wire_representable', + }, + }, + 'unsupported_session_projection', + ], + ['invalid', { kind: 'page', sessions: [], nextCursor: null }, 'invalid_projection'], + ] as const) { + const connection = fakeConnection(async () => result); + await assert.rejects(getRuntimeHostSession(connection, 'session-1'), (error: unknown) => { + assert.ok(error instanceof RuntimeHostSessionUpdateError, name); + assert.equal(error.operation, 'session.catalog.query'); + assert.equal(error.reason, reason); + return true; + }); + } +}); + +test('checks caller lifecycle before every read and update', async () => { + let allowed = true; + let checks = 0; + let reads = 0; + const connection = fakeConnection(async () => { + reads += 1; + return { kind: 'session', session: sessionProjection() }; + }); + + await assert.rejects( + updateRuntimeHostSession( + connection, + 'session-1', + async (): Promise => { + assert.fail('update must not start after the lifecycle closes'); + }, + { + operation: 'session.configuration.update', + assertRequestAllowed: () => { + checks += 1; + if (!allowed) throw new Error('closed'); + allowed = false; + }, + }, + ), + /closed/, + ); + assert.equal(checks, 2); + assert.equal(reads, 1); +}); + +function fakeConnection( + request: (operation: string, input: unknown) => Promise, +): Pick { + return { request } as Pick; +} + +function sessionProjection( + overrides: Partial = {}, +): SessionCatalogProjection { + return { + id: 'session-1', + revision: 1, + workspace: { target: { kind: 'host_path', path: '/workspace' }, hostCwd: '/workspace' }, + createdAt: 1, + activityAt: 1, + name: 'Session', + isFlagged: false, + isArchived: false, + labels: [], + labelsTruncated: false, + hasUnread: false, + status: 'active', + backend: 'ai-sdk', + llmConnectionId: 'connection-1', + llmConnectionSlug: 'default', + connectionLocked: false, + model: 'default', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + ...overrides, + }; +} diff --git a/packages/cli/src/acp/maka-acp-agent.ts b/packages/cli/src/acp/maka-acp-agent.ts index 18293dbec2..11926dc954 100644 --- a/packages/cli/src/acp/maka-acp-agent.ts +++ b/packages/cli/src/acp/maka-acp-agent.ts @@ -22,7 +22,7 @@ import type { AcpSessionRegistry } from './session-registry.js'; export interface MakaAcpAgentOptions { readonly version: string; - readonly sessionRegistry: Pick; + readonly sessionRegistry: Pick; } export function createMakaAcpAgent(options: MakaAcpAgentOptions): AgentApp { @@ -34,5 +34,8 @@ export function createMakaAcpAgent(options: MakaAcpAgentOptions): AgentApp { agentInfo: { name: 'maka', title: 'Maka', version: options.version }, })) .onRequest(methods.agent.session.new, ({ params }) => options.sessionRegistry.create(params)) - .onRequest(methods.agent.session.list, ({ params }) => options.sessionRegistry.list(params)); + .onRequest(methods.agent.session.list, ({ params }) => options.sessionRegistry.list(params)) + .onRequest(methods.agent.session.setConfigOption, ({ params }) => + options.sessionRegistry.setConfigOption(params), + ); } diff --git a/packages/cli/src/acp/session-configuration.ts b/packages/cli/src/acp/session-configuration.ts new file mode 100644 index 0000000000..a35f7b5977 --- /dev/null +++ b/packages/cli/src/acp/session-configuration.ts @@ -0,0 +1,184 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { SessionConfigOption, SetSessionConfigOptionRequest } from '@agentclientprotocol/sdk'; +import { COLLABORATION_MODES, type CollaborationMode } from '@maka/core/collaboration'; +import { THINKING_LEVELS, type ThinkingLevel } from '@maka/core/model-thinking'; +import { ORCHESTRATION_MODES, type OrchestrationMode } from '@maka/core/orchestration'; +import type { PermissionMode } from '@maka/core/permission'; +import { CHAT_DEFAULT_PERMISSION_MODES } from '@maka/core/settings'; +import type { + SessionCatalogProjection, + SessionConfigurationPatch, +} from '@maka/runtime-host/protocol'; + +interface AcpSessionConfigSpec { + readonly id: string; + readonly name: string; + readonly category: string; + readonly options: readonly (readonly [string, string])[]; +} + +const PERMISSION_NAMES: Readonly> = { + explore: 'Explore', + ask: 'Ask', + bypass: 'Bypass', +}; + +const THINKING_NAMES: Readonly> = { + default: 'Default', + off: 'Off', + minimal: 'Minimal', + low: 'Low', + medium: 'Medium', + high: 'High', + xhigh: 'Extra high', + max: 'Max', +}; + +const COLLABORATION_NAMES: Readonly> = { + agent: 'Agent', + plan: 'Plan', +}; + +const ORCHESTRATION_NAMES: Readonly> = { + default: 'Default', + swarm: 'Swarm', + graph: 'Graph', +}; + +const PERMISSION_SPEC = { + id: 'permission_mode', + name: 'Permission mode', + category: '_maka/permission_mode', + options: namedOptions(CHAT_DEFAULT_PERMISSION_MODES, PERMISSION_NAMES), +} as const satisfies AcpSessionConfigSpec; +const THINKING_SPEC = { + id: 'thinking_level', + name: 'Thinking level', + category: 'thought_level', + options: namedOptions(['default', ...THINKING_LEVELS], THINKING_NAMES), +} as const satisfies AcpSessionConfigSpec; +const COLLABORATION_SPEC = { + id: 'collaboration_mode', + name: 'Collaboration mode', + category: 'mode', + options: namedOptions(COLLABORATION_MODES, COLLABORATION_NAMES), +} as const satisfies AcpSessionConfigSpec; +const ORCHESTRATION_SPEC = { + id: 'orchestration_mode', + name: 'Orchestration mode', + category: '_maka/orchestration_mode', + options: namedOptions(ORCHESTRATION_MODES, ORCHESTRATION_NAMES), +} as const satisfies AcpSessionConfigSpec; + +const CONFIG_SPECS = [ + PERMISSION_SPEC, + THINKING_SPEC, + COLLABORATION_SPEC, + ORCHESTRATION_SPEC, +] as const; + +export function projectAcpSessionConfigOptions( + session: SessionCatalogProjection, + thinkingLevels: readonly ThinkingLevel[], +): SessionConfigOption[] { + return [ + configOption(PERMISSION_SPEC, session.permissionMode), + ...(thinkingLevels.length === 0 + ? [] + : [ + configOption( + { + ...THINKING_SPEC, + options: namedOptions(['default', ...thinkingLevels], THINKING_NAMES), + }, + session.thinkingLevel ?? 'default', + ), + ]), + configOption(COLLABORATION_SPEC, session.collaborationMode), + configOption(ORCHESTRATION_SPEC, session.orchestrationMode), + ]; +} + +function namedOptions( + values: readonly Value[], + names: Readonly>, +): readonly (readonly [Value, string])[] { + return values.map((value) => [value, names[value]] as const); +} + +function configOption(spec: AcpSessionConfigSpec, currentValue: string): SessionConfigOption { + return { + type: 'select', + id: spec.id, + name: spec.name, + category: spec.category, + currentValue, + options: spec.options.map(([value, name]) => ({ value, name })), + }; +} + +export class AcpSessionConfigInputError extends Error { + readonly name = 'AcpSessionConfigInputError'; + constructor( + readonly field: 'configId' | 'value', + readonly reason: 'unsupported' | 'invalid_type', + ) { + super(`Invalid ACP Session configuration ${field}`); + } +} + +export function validateAcpSessionConfigOptionRequest( + request: SetSessionConfigOptionRequest, +): asserts request is SetSessionConfigOptionRequest & { readonly value: string } { + const spec = CONFIG_SPECS.find(({ id }) => id === request.configId); + if (!spec) throw new AcpSessionConfigInputError('configId', 'unsupported'); + if (typeof request.value !== 'string') + throw new AcpSessionConfigInputError('value', 'invalid_type'); + if (!spec.options.some(([value]) => value === request.value)) { + throw new AcpSessionConfigInputError('value', 'unsupported'); + } +} + +export function createAcpSessionConfigPatch( + request: SetSessionConfigOptionRequest, +): SessionConfigurationPatch { + validateAcpSessionConfigOptionRequest(request); + switch (request.configId) { + case 'permission_mode': + return { permissionMode: request.value as SessionConfigurationPatch['permissionMode'] }; + case 'thinking_level': + return { + thinkingLevel: + request.value === 'default' + ? null + : (request.value as Exclude< + SessionConfigurationPatch['thinkingLevel'], + null | undefined + >), + }; + case 'collaboration_mode': + return { collaborationMode: request.value as SessionConfigurationPatch['collaborationMode'] }; + case 'orchestration_mode': + return { orchestrationMode: request.value as SessionConfigurationPatch['orchestrationMode'] }; + default: + throw new AcpSessionConfigInputError('configId', 'unsupported'); + } +} diff --git a/packages/cli/src/acp/session-registry.ts b/packages/cli/src/acp/session-registry.ts index f926e727f3..7d8f28e109 100644 --- a/packages/cli/src/acp/session-registry.ts +++ b/packages/cli/src/acp/session-registry.ts @@ -26,8 +26,12 @@ import { type ListSessionsResponse, type NewSessionRequest, type NewSessionResponse, + type SessionConfigOption, + type SetSessionConfigOptionRequest, + type SetSessionConfigOptionResponse, } from '@agentclientprotocol/sdk'; import { + readRuntimeHostConnectionCatalog, readRuntimeHostSessionCatalogPage, RuntimeHostCatalogReadError, RuntimeHostOperationError, @@ -39,11 +43,27 @@ import { import { SESSION_CATALOG_CURSOR_MAX_BYTES, SESSION_CATALOG_CWD_MAX_BYTES, + type SessionCatalogProjection, } from '@maka/runtime-host/protocol'; +import { + RuntimeHostSessionUpdateError, + requireRuntimeHostSessionProjection, + updateRuntimeHostSession, +} from '../runtime-host-session-update.js'; +import { + AcpSessionConfigInputError, + createAcpSessionConfigPatch, + projectAcpSessionConfigOptions, + validateAcpSessionConfigOptionRequest, +} from './session-configuration.js'; const ACP_SESSION_CURSOR_MAX_BYTES = 8 * 1024; -type AcpSessionRegistryOperation = 'session.create' | 'session.catalog.query'; +type AcpSessionRegistryOperation = + | 'connection.catalog.query' + | 'session.create' + | 'session.catalog.query' + | 'session.configuration.update'; type AcpSessionRegistryLifecycleOperation = 'connect' | AcpSessionRegistryOperation; export interface AcpSessionRegistryConnection { @@ -61,6 +81,7 @@ export class AcpSessionRegistry { readonly #connect: (signal: AbortSignal) => Promise; readonly #newSessionId: () => string; readonly #inFlightOperations = new Set>(); + readonly #ownedSessionIds = new Set(); #connection: AcpSessionRegistryConnection | undefined; #connectTask: Promise | undefined; #connectAbortController: AbortController | undefined; @@ -84,6 +105,24 @@ export class AcpSessionRegistry { return this.#track(this.#list(params)); } + async setConfigOption( + params: SetSessionConfigOptionRequest, + ): Promise { + this.#assertOpen('session.configuration.update'); + if (!this.#ownedSessionIds.has(params.sessionId)) { + throw RequestError.invalidParams( + { reason: 'unknown_session' }, + 'Session is not owned by this ACP connection', + ); + } + try { + validateAcpSessionConfigOptionRequest(params); + } catch (error) { + throw requestErrorFromConfigInput(error); + } + return this.#track(this.#setConfigOption(params)); + } + dispose(): Promise { this.#closing = true; this.#connectAbortController?.abort(); @@ -94,8 +133,9 @@ export class AcpSessionRegistry { async #create(params: NewSessionRequest): Promise { const connection = await this.#getConnection('session.create'); const sessionId = this.#newSessionId(); + let result; try { - await connection.request('session.create', { + result = await connection.request('session.create', { sessionId, workspace: { kind: 'host_path', path: params.cwd }, modelTarget: { kind: 'default' }, @@ -103,7 +143,58 @@ export class AcpSessionRegistry { } catch (error) { throw requestErrorFromRuntimeHost(error, 'session.create', { sessionId }); } - return { sessionId }; + let created: SessionCatalogProjection; + try { + created = requireRuntimeHostSessionProjection(result, 'session.create'); + } catch (error) { + throw requestErrorFromSessionUpdate(error, 'session.create', { sessionId }); + } + const configOptions = await this.#projectConfigOptions(connection, created); + this.#ownedSessionIds.add(sessionId); + return { sessionId, configOptions }; + } + + async #setConfigOption( + params: SetSessionConfigOptionRequest & { readonly value: string }, + ): Promise { + const connection = await this.#getConnection('session.configuration.update'); + let committed: SessionCatalogProjection; + try { + committed = await updateRuntimeHostSession( + connection, + params.sessionId, + (current) => + connection.request('session.configuration.update', { + sessionId: params.sessionId, + expectedRevision: current.revision, + patch: createAcpSessionConfigPatch(params), + }), + { + operation: 'session.configuration.update', + assertRequestAllowed: () => this.#assertOpen('session.configuration.update'), + }, + ); + } catch (error) { + throw requestErrorFromSessionUpdate(error, 'session.configuration.update'); + } + return { configOptions: await this.#projectConfigOptions(connection, committed) }; + } + + async #projectConfigOptions( + connection: AcpSessionRegistryConnection, + session: SessionCatalogProjection, + ): Promise { + let catalog; + try { + catalog = await readRuntimeHostConnectionCatalog(connection); + } catch (error) { + throw requestErrorFromRuntimeHost(error, 'connection.catalog.query'); + } + const selectedConnection = catalog.connections.find( + ({ connectionId }) => connectionId === session.llmConnectionId, + ); + const selectedModel = selectedConnection?.catalogEntries.find(({ id }) => id === session.model); + return projectAcpSessionConfigOptions(session, selectedModel?.thinkingLevels ?? []); } async #list(params: ListSessionsRequest): Promise { @@ -157,6 +248,7 @@ export class AcpSessionRegistry { const connectionClose = this.#closeOwnedConnection(); await Promise.allSettled([connectionClose]); await Promise.allSettled([...this.#inFlightOperations]); + this.#ownedSessionIds.clear(); } #closeOwnedConnection(): Promise { @@ -255,13 +347,67 @@ function validateNewSessionParams(params: NewSessionRequest): void { } } +function requestErrorFromConfigInput(error: unknown): RequestError { + if (error instanceof AcpSessionConfigInputError) { + return RequestError.invalidParams( + { field: error.field, reason: error.reason }, + 'Invalid Session configuration option', + ); + } + return RequestError.internalError( + { + source: 'adapter', + operation: 'session.configuration.update', + code: 'validation_failed', + }, + 'Session configuration validation failed', + ); +} + +function requestErrorFromSessionUpdate( + error: unknown, + operation: AcpSessionRegistryOperation, + extra: Record = {}, +): RequestError { + if (error instanceof RequestError) return error; + if (!(error instanceof RuntimeHostSessionUpdateError)) { + return requestErrorFromRuntimeHost(error, operation, extra); + } + const common = { source: 'runtime_host', operation: error.operation, ...extra }; + switch (error.reason) { + case 'not_found': + return RequestError.invalidParams( + { ...common, code: 'not_found' }, + 'Runtime Host Session was not found', + ); + case 'invalid_projection': + return RequestError.internalError( + { ...common, code: 'catalog_read_failure', reason: 'invalid_projection' }, + 'Runtime Host returned an invalid Session lookup', + ); + case 'unsupported_session_projection': + return RequestError.internalError( + { ...common, code: 'unsupported_session_projection' }, + 'Runtime Host Session cannot be represented in ACP', + ); + case 'revision_conflict': + return RequestError.internalError( + { ...common, code: 'revision_conflict', attempts: error.attempts }, + 'Session configuration kept changing', + ); + } +} + function requestErrorFromRuntimeHost( error: unknown, operation: AcpSessionRegistryOperation, extra: Record = {}, ): RequestError { const data = { ...runtimeHostErrorData(error, operation), ...extra }; - if (error instanceof RuntimeHostOperationError && error.code === 'invalid_request') { + if ( + error instanceof RuntimeHostOperationError && + (error.code === 'invalid_request' || error.code === 'not_found') + ) { return RequestError.invalidParams(data, 'Runtime Host rejected the request'); } return RequestError.internalError(data, 'Runtime Host request failed'); diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts index 36ba9707c8..45f0395b1c 100644 --- a/packages/cli/src/runtime-host-session-driver.ts +++ b/packages/cli/src/runtime-host-session-driver.ts @@ -68,7 +68,6 @@ import { OperationOutput, SessionCatalogItem, SessionCatalogProjection, - SessionUpdateResult, SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES, WorkspaceTarget, type GoalControlAction, @@ -78,6 +77,11 @@ import { } from '@maka/runtime-host/protocol'; import { RuntimeHostSessionChannel } from './runtime-host-session-channel.js'; import type { RuntimeHostSessionChannelOpenResult } from './runtime-host-session-channel.js'; +import { + getRuntimeHostSession, + requireRuntimeHostSessionProjection as requireSession, + updateRuntimeHostSession, +} from './runtime-host-session-update.js'; import type { InspectCwdChanges, MakaAttachedSessionTurn, @@ -671,12 +675,16 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { async renameSession(name: string): Promise { const sessionId = this.#requireSession('rename'); - const session = await updateRuntimeHostSession(this.#connection, sessionId, (current) => - this.#request('session.metadata.update', { - sessionId, - expectedRevision: current.revision, - patch: { name }, - }), + const session = await updateRuntimeHostSession( + this.#connection, + sessionId, + (current) => + this.#request('session.metadata.update', { + sessionId, + expectedRevision: current.revision, + patch: { name }, + }), + { operation: 'session.metadata.update' }, ); return session.name; } @@ -779,12 +787,16 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { } #commitCwdRelocation(sessionId: string, cwd: string): Promise { - return updateRuntimeHostSession(this.#connection, sessionId, (current) => - this.#request('session.workspace.relocate', { - sessionId, - expectedRevision: current.revision, - workspace: { kind: 'host_path', path: cwd }, - }), + return updateRuntimeHostSession( + this.#connection, + sessionId, + (current) => + this.#request('session.workspace.relocate', { + sessionId, + expectedRevision: current.revision, + workspace: { kind: 'host_path', path: cwd }, + }), + { operation: 'session.workspace.relocate' }, ); } @@ -1306,12 +1318,16 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { orchestrationMode?: OrchestrationMode; }, ): Promise { - return updateRuntimeHostSession(this.#connection, sessionId, (current) => - this.#request('session.configuration.update', { - sessionId, - expectedRevision: current.revision, - patch, - }), + return updateRuntimeHostSession( + this.#connection, + sessionId, + (current) => + this.#request('session.configuration.update', { + sessionId, + expectedRevision: current.revision, + patch, + }), + { operation: 'session.configuration.update' }, ); } @@ -1668,15 +1684,6 @@ interface LoadedSessionConfiguration { boundaryDisplayMode: PermissionMode | undefined; } -async function getRuntimeHostSession( - connection: RuntimeHostSessionDriverConnection, - sessionId: string, -): Promise { - const result = await connection.request('session.catalog.query', { kind: 'get', sessionId }); - if (result.kind !== 'session') throw new Error('Runtime Host returned an invalid Session lookup'); - return result.session === null ? null : requireSession(result.session); -} - function representableSession(item: SessionCatalogItem): SessionCatalogProjection[] { return 'kind' in item ? [] : [item]; } @@ -1736,11 +1743,6 @@ function visibleTranscriptMessages( return boundary < 0 ? messages : messages.slice(boundary + 1); } -function requireSession(item: SessionCatalogItem): SessionCatalogProjection { - if (!('kind' in item)) return item; - throw new Error(`Runtime Host Session is not representable by this CLI: ${item.id}`); -} - function inspectRuntimeHostSessionResumeAvailability( summary: SessionSummary, location: NonNullable, @@ -1765,20 +1767,6 @@ async function assertSessionResumeAvailable( } } -async function updateRuntimeHostSession( - connection: RuntimeHostSessionDriverConnection, - sessionId: string, - update: (current: SessionCatalogProjection) => Promise, -): Promise { - for (let attempt = 0; attempt < MAX_CATALOG_ATTEMPTS; attempt += 1) { - const current = await getRuntimeHostSession(connection, sessionId); - if (!current) throw new Error(`Session not found: ${sessionId}`); - const result = await update(current); - if (result.kind === 'committed') return requireSession(result.session); - } - throw new Error(`Session kept changing while updating: ${sessionId}`); -} - async function loadCurrentMessages( connection: RuntimeHostSessionDriverConnection, sessionId: string, diff --git a/packages/cli/src/runtime-host-session-update.ts b/packages/cli/src/runtime-host-session-update.ts new file mode 100644 index 0000000000..331252e434 --- /dev/null +++ b/packages/cli/src/runtime-host-session-update.ts @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { DirectRequestOperationKey, RuntimeHostConnection } from '@maka/runtime-host/client'; +import type { + SessionCatalogItem, + SessionCatalogProjection, + SessionUpdateResult, +} from '@maka/runtime-host/protocol'; + +const MAX_UPDATE_ATTEMPTS = 3; + +type RuntimeHostSessionUpdateErrorReason = + | 'invalid_projection' + | 'not_found' + | 'unsupported_session_projection' + | 'revision_conflict'; + +export class RuntimeHostSessionUpdateError extends Error { + readonly name = 'RuntimeHostSessionUpdateError'; + + constructor( + readonly operation: DirectRequestOperationKey, + readonly reason: RuntimeHostSessionUpdateErrorReason, + readonly sessionId: string, + readonly attempts?: number, + ) { + super(`Runtime Host Session update failed: ${reason}`); + } +} + +type RuntimeHostSessionUpdateConnection = Pick; + +export async function getRuntimeHostSession( + connection: RuntimeHostSessionUpdateConnection, + sessionId: string, +): Promise { + const result = await connection.request('session.catalog.query', { kind: 'get', sessionId }); + if (result.kind !== 'session') { + throw new RuntimeHostSessionUpdateError( + 'session.catalog.query', + 'invalid_projection', + sessionId, + ); + } + return result.session === null + ? null + : requireRuntimeHostSessionProjection(result.session, 'session.catalog.query'); +} + +export async function updateRuntimeHostSession( + connection: RuntimeHostSessionUpdateConnection, + sessionId: string, + update: (current: SessionCatalogProjection) => Promise, + options: { + readonly operation: DirectRequestOperationKey; + readonly assertRequestAllowed?: () => void; + }, +): Promise { + for (let attempt = 0; attempt < MAX_UPDATE_ATTEMPTS; attempt += 1) { + options.assertRequestAllowed?.(); + const current = await getRuntimeHostSession(connection, sessionId); + if (!current) { + throw new RuntimeHostSessionUpdateError('session.catalog.query', 'not_found', sessionId); + } + options.assertRequestAllowed?.(); + const result = await update(current); + if (result.kind === 'committed') { + return requireRuntimeHostSessionProjection(result.session, options.operation); + } + } + throw new RuntimeHostSessionUpdateError( + options.operation, + 'revision_conflict', + sessionId, + MAX_UPDATE_ATTEMPTS, + ); +} + +export function requireRuntimeHostSessionProjection( + session: SessionCatalogItem, + operation: DirectRequestOperationKey = 'session.catalog.query', +): SessionCatalogProjection { + if (!('kind' in session)) return session; + throw new RuntimeHostSessionUpdateError(operation, 'unsupported_session_projection', session.id); +}