diff --git a/docs/windows-test-inventory.md b/docs/windows-test-inventory.md index 79974a1a54..27d877724b 100644 --- a/docs/windows-test-inventory.md +++ b/docs/windows-test-inventory.md @@ -17,9 +17,9 @@ Locations intentionally omit line numbers so unrelated edits do not invalidate t |---|---:| | windows-backend-gap | 27 | | portable-candidate | 31 | -| platform-contract | 31 | +| platform-contract | 32 | -Total Windows-excluded declarations: **89** +Total Windows-excluded declarations: **90** ## Inventory @@ -34,6 +34,7 @@ Total Windows-excluded declarations: **89** | platform-contract | `apps/desktop/src/main/__tests__/shell-env.test.ts` keeps the inherited PATH and does not log shell stderr when capture fails | `process.platform === 'win32'` | | platform-contract | `apps/desktop/src/main/__tests__/shell-env.test.ts` kills login-shell descendants when capture times out | `process.platform === 'win32'` | | platform-contract | `apps/desktop/src/main/__tests__/shell-env.test.ts` bounds shell output instead of buffering until the global timeout | `process.platform === 'win32'` | +| platform-contract | `packages/cli/src/__tests__/acp-prompt-content.test.ts` rejects a FIFO without blocking the process | `process.platform === 'win32'` | | portable-candidate | `packages/cli/src/__tests__/pi-transcript.test.ts` shortens POSIX paths under the home directory | `process.platform === 'win32'` | | portable-candidate | `packages/cli/src/__tests__/pi-transcript.test.ts` keeps POSIX paths outside the home directory absolute | `process.platform === 'win32'` | | portable-candidate | `packages/cli/src/__tests__/runtime-host-local-managed-activation.test.ts` local CLI cold-starts through the installed ${legacy ? 'legacy' : 'Node'} operator | `process.platform === 'win32'` | diff --git a/packages/cli/src/__tests__/acp-agent.test.ts b/packages/cli/src/__tests__/acp-agent.test.ts index 709f0f3ca4..ae5c53f293 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 and advertises only Session listing', async () => { + test('returns the Maka identity and advertises Session listing and close', async () => { await client({ name: 'test-client' }).connectWith( createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), async (agent) => { assert.deepEqual(await agent.request(methods.agent.initialize, { protocolVersion: 1 }), { protocolVersion: 1, - agentCapabilities: { sessionCapabilities: { list: {} } }, + agentCapabilities: { sessionCapabilities: { list: {}, close: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }); @@ -107,21 +107,50 @@ describe('Maka ACP agent', () => { ]); }); - test('does not implement or advertise session/close', async () => { - await client({ name: 'test-client' }).connectWith( - createMakaAcpAgent({ version: '0.2.0', sessionRegistry: fakeSessionRegistry() }), + test('routes prompt, cancel, and close through the Session registry', async () => { + const prompts: unknown[] = []; + const cancellations: unknown[] = []; + const closes: unknown[] = []; + const updates: unknown[] = []; + const testClient = client({ name: 'test-client' }).onNotification( + methods.client.session.update, + ({ params }) => void updates.push(params), + ); + await testClient.connectWith( + createMakaAcpAgent({ + version: '0.2.0', + sessionRegistry: fakeSessionRegistry({ prompts, cancellations, closes }), + }), async (agent) => { - await assert.rejects( - agent.request(methods.agent.session.close, { sessionId: 'session-1' }), - (error: unknown) => { - assert.ok(error instanceof RequestError); - assert.equal(error.code, -32601); - assert.deepEqual(error.data, { method: 'session/close' }); - return true; - }, + assert.deepEqual( + await agent.request(methods.agent.session.prompt, { + sessionId: 'session-1', + prompt: [{ type: 'text', text: 'hello' }], + }), + { stopReason: 'end_turn' }, + ); + await agent.notify(methods.agent.session.cancel, { sessionId: 'session-1' }); + assert.deepEqual( + await agent.request(methods.agent.session.close, { sessionId: 'session-1' }), + {}, ); }, ); + assert.deepEqual(prompts, [ + { sessionId: 'session-1', prompt: [{ type: 'text', text: 'hello' }] }, + ]); + assert.deepEqual(cancellations, [{ sessionId: 'session-1' }]); + assert.deepEqual(closes, [{ sessionId: 'session-1' }]); + assert.deepEqual(updates, [ + { + sessionId: 'session-1', + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'hello' }, + messageId: 'message-1', + }, + }, + ]); }); test('does not implement session/set_mode', async () => { @@ -158,7 +187,14 @@ describe('Maka ACP agent', () => { }); function fakeSessionRegistry( - observations: { creates?: unknown[]; lists?: unknown[]; configurationRequests?: unknown[] } = {}, + observations: { + creates?: unknown[]; + lists?: unknown[]; + configurationRequests?: unknown[]; + prompts?: unknown[]; + cancellations?: unknown[]; + closes?: unknown[]; + } = {}, ) { return { create: async (params: unknown) => { @@ -196,5 +232,22 @@ function fakeSessionRegistry( ], }; }, + prompt: async (params: unknown, context: { notify(notification: unknown): Promise }) => { + observations.prompts?.push(params); + await context.notify({ + sessionId: 'session-1', + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'hello' }, + messageId: 'message-1', + }, + }); + return { stopReason: 'end_turn' as const }; + }, + cancel: async (params: unknown) => void observations.cancellations?.push(params), + close: async (params: unknown) => { + observations.closes?.push(params); + return {}; + }, }; } diff --git a/packages/cli/src/__tests__/acp-child-process-harness.ts b/packages/cli/src/__tests__/acp-child-process-harness.ts index f1b1c3fd0c..045c1c7805 100644 --- a/packages/cli/src/__tests__/acp-child-process-harness.ts +++ b/packages/cli/src/__tests__/acp-child-process-harness.ts @@ -48,6 +48,7 @@ export interface AcpChildProcessHarnessOptions { readonly model?: { readonly id: string; readonly thinkingLevels: readonly ThinkingLevel[]; + readonly baseUrl?: string; }; } @@ -364,7 +365,7 @@ async function seedModelConnection( slug: 'acp-fixture-model', name: 'ACP fixture model', providerType: 'openai-compatible', - baseUrl: 'https://acp-model.invalid/v1', + baseUrl: model.baseUrl ?? 'https://acp-model.invalid/v1', enabled: true, enabledModelIds: [model.id], ...(model.thinkingLevels.length === 0 diff --git a/packages/cli/src/__tests__/acp-child-process.test.ts b/packages/cli/src/__tests__/acp-child-process.test.ts index f0b2b129bc..5fcd8b59f8 100644 --- a/packages/cli/src/__tests__/acp-child-process.test.ts +++ b/packages/cli/src/__tests__/acp-child-process.test.ts @@ -20,9 +20,14 @@ import assert from 'node:assert/strict'; import { once } from 'node:events'; import { realpath } from 'node:fs/promises'; +import { createServer, type ServerResponse } from 'node:http'; import { PassThrough } from 'node:stream'; import { describe, test } from 'node:test'; -import { RequestError, methods } from '@agentclientprotocol/sdk'; +import { methods, type SessionNotification } from '@agentclientprotocol/sdk'; +import { waitFor } from '@maka/core/test-only/async-primitives'; +import { connectRuntimeHost } from '@maka/runtime-host/client'; +import { RUNTIME_HOST_PROTOCOL_VERSION } from '@maka/runtime-host/protocol'; +import { getRuntimeHostSession } from '../runtime-host-session-update.js'; import { pipeCapturedStdout, StdoutCaptureBridge, @@ -119,7 +124,7 @@ describe('Maka ACP child process', () => { await harness.withClient(async ({ context }) => { assert.deepEqual(await context.request(methods.agent.initialize, { protocolVersion: 1 }), { protocolVersion: 1, - agentCapabilities: { sessionCapabilities: { list: {} } }, + agentCapabilities: { sessionCapabilities: { list: {}, close: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }); @@ -194,14 +199,16 @@ describe('Maka ACP child process', () => { true, ); - await assert.rejects( - context.request(methods.agent.session.close, { sessionId: first.sessionId }), - (error: unknown) => { - assert.ok(error instanceof RequestError); - assert.equal(error.code, -32601); - assert.deepEqual(error.data, { method: 'session/close' }); - return true; - }, + assert.deepEqual( + await context.request(methods.agent.session.close, { sessionId: first.sessionId }), + {}, + ); + const listedAfterClose = await context.request(methods.agent.session.list, { + cwd: harness.workspaceRoot, + }); + assert.equal( + listedAfterClose.sessions.some((session) => session.sessionId === first.sessionId), + true, ); }); @@ -325,8 +332,278 @@ describe('Maka ACP child process', () => { { startRuntimeHost: true }, ); }); + + test('Host admission rejects an extra attachment before starting a Turn and close releases capacity', { + timeout: 60_000, + }, async () => { + const model = await startAcpModelFixture(); + try { + await withAcpChildProcessHarness( + async (harness) => { + await harness.withClient(async ({ context }) => { + await context.request(methods.agent.initialize, { protocolVersion: 1 }); + const ids: string[] = []; + for (let index = 0; index < 17; index += 1) { + ids.push( + ( + await context.request(methods.agent.session.new, { + cwd: harness.workspaceRoot, + mcpServers: [], + }) + ).sessionId, + ); + } + const prompt = (sessionId: string) => + context.request(methods.agent.session.prompt, { + sessionId, + prompt: [{ type: 'text', text: 'COMPLETE_ME' }], + }); + // Independent Sessions can finish concurrently. All attachments must + // remain retained after completion before we test the next admission. + await Promise.all( + ids.slice(0, 16).map(async (id) => { + assert.deepEqual(await prompt(id), { stopReason: 'end_turn' }); + }), + ); + await assert.rejects(prompt(ids[16]!), (error: unknown) => { + assert.equal( + (error as { data?: { operation?: string } }).data?.operation, + 'subscription.open', + ); + return true; + }); + const connected = await connectRuntimeHost({ + rootPath: harness.workspaceRoot, + protocol: { min: RUNTIME_HOST_PROTOCOL_VERSION, max: RUNTIME_HOST_PROTOCOL_VERSION }, + }); + assert.equal(connected.kind, 'connected'); + if (connected.kind !== 'connected') assert.fail(); + try { + const untouched = await connected.connection.openSessionSubscription({ + sessionId: ids[16]!, + transcript: { kind: 'none' }, + }); + assert.equal( + untouched.snapshot.rootTurn, + null, + 'capacity rejection must not create a Turn', + ); + await untouched.close(); + await context.request(methods.agent.session.close, { sessionId: ids[0]! }); + assert.ok(await getRuntimeHostSession(connected.connection, ids[0]!)); + assert.deepEqual(await prompt(ids[16]!), { stopReason: 'end_turn' }); + } finally { + await connected.connection.close(); + } + }); + }, + { + startRuntimeHost: true, + // This operation covers 17 creates and 17 complete Turns, not one RPC. + timeoutMs: 45_000, + model: { id: 'capacity-fixture', thinkingLevels: [], baseUrl: model.baseUrl }, + }, + ); + } finally { + await model.close(); + } + }); + + test('streams, cancels, and closes through the real ACP and Runtime Host process boundary', { + timeout: 30_000, + }, async () => { + const model = await startAcpModelFixture(); + try { + await withAcpChildProcessHarness( + async (harness) => { + const updates: SessionNotification[] = []; + 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( + await context.request(methods.agent.session.prompt, { + sessionId: created.sessionId, + prompt: [{ type: 'text', text: 'COMPLETE_ME' }], + }), + { stopReason: 'end_turn' }, + ); + assert.equal( + updates.some( + ({ update }) => + update.sessionUpdate === 'agent_message_chunk' && + update.content.type === 'text' && + update.content.text.includes('ACP fixture completed'), + ), + true, + ); + + // A second client mutates the Host while ACP retains its attachment. + const connected = await connectRuntimeHost({ + rootPath: harness.workspaceRoot, + protocol: { + min: RUNTIME_HOST_PROTOCOL_VERSION, + max: RUNTIME_HOST_PROTOCOL_VERSION, + }, + }); + assert.equal(connected.kind, 'connected'); + if (connected.kind !== 'connected') assert.fail('Host connection unavailable'); + try { + const current = await getRuntimeHostSession( + connected.connection, + created.sessionId, + ); + assert.ok(current); + const changed = await connected.connection.request('session.configuration.update', { + sessionId: created.sessionId, + expectedRevision: current.revision, + patch: { permissionMode: 'bypass', thinkingLevel: 'low' }, + }); + assert.equal(changed.kind, 'committed'); + await waitFor( + () => + updates.some( + ({ update }) => + update.sessionUpdate === 'config_option_update' && + update.configOptions.some( + (option) => + option.id === 'permission_mode' && option.currentValue === 'bypass', + ) && + update.configOptions.some( + (option) => + option.id === 'thinking_level' && option.currentValue === 'low', + ), + ), + { timeoutMs: 5000, pollMs: 10, message: 'external configuration notification' }, + ); + } finally { + await connected.connection.close(); + } + + const cancelled = context.request(methods.agent.session.prompt, { + sessionId: created.sessionId, + prompt: [{ type: 'text', text: 'CANCEL_ME' }], + }); + await model.cancelStarted; + await context.notify(methods.agent.session.cancel, { sessionId: created.sessionId }); + assert.deepEqual(await cancelled, { stopReason: 'cancelled' }); + assert.deepEqual( + await context.request(methods.agent.session.close, { + sessionId: created.sessionId, + }), + {}, + ); + }, + (app) => + app.onNotification(methods.client.session.update, ({ params }) => { + updates.push(params); + }), + ); + + await harness.closeStdin(); + assert.deepEqual(await harness.waitForExit(), { code: 0, signal: null }); + assert.equal(harness.stderr, ''); + }, + { + startRuntimeHost: true, + model: { + id: 'acp-stream-fixture', + thinkingLevels: ['low'], + baseUrl: model.baseUrl, + }, + }, + ); + } finally { + await model.close(); + } + }); }); +async function startAcpModelFixture(): Promise<{ + readonly baseUrl: string; + readonly cancelStarted: Promise; + close(): Promise; +}> { + let markCancelStarted!: () => void; + const cancelStarted = new Promise((resolve) => { + markCancelStarted = resolve; + }); + const server = createServer((request, response) => { + void readBody(request) + .then((body) => { + if (body.includes('CANCEL_ME')) { + response.writeHead(200, { 'content-type': 'text/event-stream' }); + response.write(`data: ${JSON.stringify(modelChunk('partial', null))}\n\n`); + markCancelStarted(); + request.once('close', () => response.end()); + return; + } + if (body.includes('COMPLETE_ME')) { + respondModelText(response, 'ACP fixture completed.'); + return; + } + respondModelText(response, 'ACP fixture session'); + }) + .catch((error) => response.destroy(error as Error)); + }); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); + }); + const address = server.address(); + assert.ok(address && typeof address !== 'string'); + return { + baseUrl: `http://127.0.0.1:${address.port}/v1`, + cancelStarted, + close: () => + new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ), + }; +} + +function respondModelText(response: ServerResponse, text: string): void { + response.writeHead(200, { 'content-type': 'text/event-stream' }); + response.write(`data: ${JSON.stringify(modelChunk(text, null))}\n\n`); + response.write(`data: ${JSON.stringify(modelChunk('', 'stop'))}\n\n`); + response.end('data: [DONE]\n\n'); +} + +function modelChunk(text: string, finishReason: 'stop' | null) { + return { + id: 'chatcmpl-acp-fixture', + object: 'chat.completion.chunk', + created: 1, + model: 'acp-stream-fixture', + choices: [ + { + index: 0, + delta: finishReason === null ? { role: 'assistant', content: text } : {}, + finish_reason: finishReason, + }, + ], + ...(finishReason === 'stop' + ? { usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 } } + : {}), + }; +} + +function readBody(request: import('node:http').IncomingMessage): Promise { + return new Promise((resolve, reject) => { + let body = ''; + request.setEncoding('utf8'); + request.on('data', (chunk) => { + body += chunk; + }); + request.on('end', () => resolve(body)); + request.on('error', reject); + }); +} + function assertJsonRpcMessage(message: unknown): void { assert.ok(message && typeof message === 'object' && !Array.isArray(message)); const record = message as Record; diff --git a/packages/cli/src/__tests__/acp-prompt-content.test.ts b/packages/cli/src/__tests__/acp-prompt-content.test.ts new file mode 100644 index 0000000000..6a8e1c5ed7 --- /dev/null +++ b/packages/cli/src/__tests__/acp-prompt-content.test.ts @@ -0,0 +1,174 @@ +/* + * 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 { execFileSync, spawn } from 'node:child_process'; +import { mkdtemp, realpath, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { basename, join } from 'node:path'; +import { pathToFileURL } from 'node:url'; +import { describe, test } from 'node:test'; +import { RequestError, type ContentBlock } from '@agentclientprotocol/sdk'; +import { MAX_ATTACHMENT_BYTES, MAX_ATTACHMENT_COUNT } from '@maka/core/attachments'; +import { mapAcpPromptContent } from '../acp/prompt-content.js'; + +describe('ACP prompt content', () => { + test('rejects a FIFO without blocking the process', { + skip: process.platform === 'win32', + }, async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-acp-fifo-')); + const fifo = join(root, 'pipe'); + execFileSync('mkfifo', [fifo]); + const child = spawn( + process.execPath, + [ + '--input-type=module', + '-e', + ` + import assert from 'node:assert/strict'; + import { mapAcpPromptContent } from ${JSON.stringify(new URL('../acp/prompt-content.js', import.meta.url).href)}; + await assert.rejects(mapAcpPromptContent([{ type: 'resource_link', name: 'pipe', uri: process.argv[1] }]), + { data: { field: 'prompt', reason: 'resource_not_file' } }); + `, + pathToFileURL(fifo).href, + ], + { stdio: 'pipe' }, + ); + let stderr = ''; + child.stderr.on('data', (data) => { + stderr += data; + }); + const timeout = setTimeout(() => child.kill('SIGKILL'), 3000); + try { + const code = await new Promise((resolve, reject) => { + child.once('error', reject); + child.once('exit', resolve); + }); + assert.equal(code, 0, `FIFO reader must reject and exit: ${stderr}`); + } finally { + clearTimeout(timeout); + child.kill(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('joins ordered text blocks with paragraph separators', async () => { + assert.deepEqual( + await mapAcpPromptContent([ + { type: 'text', text: 'first' }, + { type: 'text', text: 'second' }, + ]), + { text: 'first\n\nsecond' }, + ); + }); + + test('maps an ordinary local resource link to an external-file attachment', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-acp-prompt-')); + const path = join(root, 'note.txt'); + await writeFile(path, 'hello'); + const uri = pathToFileURL(path).href; + try { + assert.deepEqual( + await mapAcpPromptContent([ + { type: 'text', text: 'read this' }, + { type: 'resource_link', uri, name: 'note.txt', mimeType: 'text/plain' }, + ]), + { + text: `read this\n\n${uri}`, + displayText: 'read this', + attachments: [ + { + kind: 'other', + name: 'note.txt', + mimeType: 'text/plain', + bytes: 5, + ref: { kind: 'external_file', absolutePath: await realpath(path) }, + }, + ], + }, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + test('keeps a resource-only prompt model-visible while its display text is empty', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-acp-prompt-')); + const path = join(root, 'image.bin'); + await writeFile(path, Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a])); + const uri = pathToFileURL(path).href; + try { + const mapped = await mapAcpPromptContent([ + { type: 'resource_link', uri, name: basename(path), mimeType: 'application/octet-stream' }, + ]); + assert.equal(mapped.text, uri); + assert.equal(mapped.displayText, ''); + assert.equal(mapped.attachments?.[0]?.kind, 'image'); + assert.equal(mapped.attachments?.[0]?.mimeType, 'image/png'); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + test('rejects unadvertised content kinds and non-local or non-file resources', async () => { + for (const prompt of [ + [{ type: 'image', data: '', mimeType: 'image/png' }], + [{ type: 'audio', data: '', mimeType: 'audio/wav' }], + [{ type: 'resource', resource: { uri: 'file:///tmp/x', text: 'x' } }], + [{ type: 'resource_link', uri: 'https://example.com/x', name: 'x' }], + [{ type: 'resource_link', uri: 'file:///tmp', name: 'tmp' }], + ] as ContentBlock[][]) { + await assert.rejects(mapAcpPromptContent(prompt), invalidPromptContent); + } + }); + + test('enforces the shared attachment count and size limits before admission', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-acp-prompt-')); + const path = join(root, 'small.txt'); + await writeFile(path, 'x'); + const resource = { + type: 'resource_link' as const, + uri: pathToFileURL(path).href, + name: 'small.txt', + }; + try { + await assert.rejects( + mapAcpPromptContent(Array.from({ length: MAX_ATTACHMENT_COUNT + 1 }, () => resource)), + invalidPromptContent, + ); + await assert.rejects( + mapAcpPromptContent([resource], { + openFile: async () => ({ + size: MAX_ATTACHMENT_BYTES + 1, + isFile: true, + prefix: new Uint8Array(), + canonicalPath: path, + }), + }), + invalidPromptContent, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); +}); + +function invalidPromptContent(error: unknown): boolean { + return error instanceof RequestError && error.code === -32602; +} diff --git a/packages/cli/src/__tests__/acp-session-event-mapper.test.ts b/packages/cli/src/__tests__/acp-session-event-mapper.test.ts new file mode 100644 index 0000000000..1726992b84 --- /dev/null +++ b/packages/cli/src/__tests__/acp-session-event-mapper.test.ts @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import type { SessionEvent } from '@maka/core/events'; +import type { SessionNotification } from '@agentclientprotocol/sdk'; +import { AcpSessionEventMapper } from '../acp/session-event-mapper.js'; + +describe('ACP Session event mapper', () => { + test('streams text and thinking while deduplicating matching completion events', async () => { + const notifications: SessionNotification[] = []; + const mapper = eventMapper(notifications); + + await mapper.accept(event({ type: 'text_delta', messageId: 'answer', text: 'hel' })); + await mapper.accept(event({ type: 'text_delta', messageId: 'answer', text: 'lo' })); + await mapper.accept(event({ type: 'text_complete', messageId: 'answer', text: 'hello' })); + await mapper.accept(event({ type: 'thinking_delta', messageId: 'thought', text: 'hmm' })); + await mapper.accept(event({ type: 'thinking_complete', messageId: 'thought', text: 'hmm' })); + + assert.deepEqual( + notifications.map(({ update }) => update), + [ + chunk('agent_message_chunk', 'answer', 'hel'), + chunk('agent_message_chunk', 'answer', 'lo'), + chunk('agent_thought_chunk', 'thought', 'hmm'), + ], + ); + }); + + test('rejects non-prefix revisions instead of reporting a second message or success', async () => { + for (const kind of ['text', 'thinking'] as const) { + for (const text of ['new', '']) { + const notifications: SessionNotification[] = []; + const mapper = eventMapper(notifications); + await mapper.accept(event({ type: `${kind}_delta`, messageId: 'answer', text: 'old' })); + await assert.rejects( + mapper.accept(event({ type: `${kind}_complete`, messageId: 'answer', text })), + { data: { source: 'adapter', code: 'unsupported_stream_revision' } }, + ); + await assert.rejects(mapper.accept(event({ type: 'complete', stopReason: 'end_turn' }))); + assert.deepEqual( + notifications.map(({ update }) => update), + [chunk(kind === 'text' ? 'agent_message_chunk' : 'agent_thought_chunk', 'answer', 'old')], + ); + } + } + }); + + test('serializes canonical transcript replacement with live notifications', async () => { + const notifications: SessionNotification[] = []; + let releaseFirst!: () => void; + const firstPending = new Promise((resolve) => { + releaseFirst = resolve; + }); + let calls = 0; + const mapper = new AcpSessionEventMapper({ + sessionId: 'session-1', + notify: async (notification) => { + calls += 1; + if (calls === 1) await firstPending; + notifications.push(notification); + }, + }); + + const live = mapper.accept(event({ type: 'text_delta', messageId: 'answer', text: 'old' })); + const replacement = mapper.replaceTranscript('turn-1', [ + { + type: 'assistant', + id: 'answer', + turnId: 'turn-1', + ts: 2, + text: 'older', + modelId: 'model', + }, + ]); + releaseFirst(); + await Promise.all([live, replacement]); + + assert.equal(notifications.length, 2); + const update = notifications[1]?.update; + assert.equal(update?.sessionUpdate, 'agent_message_chunk'); + if (update?.sessionUpdate !== 'agent_message_chunk') return; + assert.equal(update.content.type === 'text' && update.content.text, 'er'); + assert.equal(update.messageId, 'answer'); + }); + + test('rejects a canonical message that clears already delivered thinking', async () => { + const notifications: SessionNotification[] = []; + const mapper = eventMapper(notifications); + await mapper.accept(event({ type: 'thinking_delta', messageId: 'answer', text: 'old' })); + await assert.rejects( + mapper.replaceTranscript('turn-1', [ + { + type: 'assistant', + id: 'answer', + turnId: 'turn-1', + ts: 2, + text: 'answer', + modelId: 'model', + }, + ]), + { data: { source: 'adapter', code: 'unsupported_stream_revision' } }, + ); + await assert.rejects(mapper.accept(event({ type: 'complete', stopReason: 'end_turn' }))); + assert.equal(notifications.length, 1); + }); + + test('ends on authoritative abort and nonrecoverable error but not recoverable errors', async () => { + const failed = eventMapper([]); + assert.equal( + await failed.accept(event({ type: 'error', recoverable: true, message: 'retry' })), + undefined, + ); + assert.equal( + await failed.accept(event({ type: 'error', recoverable: false, message: 'failed' })), + 'end_turn', + ); + assert.equal( + await failed.accept(event({ type: 'complete', stopReason: 'end_turn' })), + 'end_turn', + ); + assert.equal( + await eventMapper([]).accept(event({ type: 'abort', reason: 'crash' })), + 'end_turn', + ); + }); + + test('emits exactly one terminal result', async () => { + const mapper = eventMapper([]); + assert.equal( + await mapper.accept(event({ type: 'complete', stopReason: 'max_tokens' })), + 'end_turn', + ); + assert.equal(await mapper.accept(event({ type: 'abort', reason: 'crash' })), 'end_turn'); + assert.equal(await mapper.cancel(), 'end_turn'); + + const cancelled = eventMapper([]); + assert.equal(await cancelled.cancel(), 'cancelled'); + assert.equal( + await cancelled.accept(event({ type: 'complete', stopReason: 'end_turn' })), + 'cancelled', + ); + }); +}); + +function eventMapper(notifications: SessionNotification[]): AcpSessionEventMapper { + return new AcpSessionEventMapper({ + sessionId: 'session-1', + notify: async (notification) => void notifications.push(notification), + }); +} + +function chunk( + sessionUpdate: 'agent_message_chunk' | 'agent_thought_chunk', + messageId: string, + text: string, +) { + return { sessionUpdate, content: { type: 'text' as const, text }, messageId }; +} + +function event>(value: T): SessionEvent { + return { id: 'event', turnId: 'turn-1', ts: 1, ...value } as unknown as SessionEvent; +} diff --git a/packages/cli/src/__tests__/acp-session-registry.test.ts b/packages/cli/src/__tests__/acp-session-registry.test.ts index bb37240505..ee916851f9 100644 --- a/packages/cli/src/__tests__/acp-session-registry.test.ts +++ b/packages/cli/src/__tests__/acp-session-registry.test.ts @@ -25,9 +25,12 @@ import { describe, test } from 'node:test'; import { RequestError, type NewSessionRequest, + type SessionNotification, type SessionConfigOption, type SetSessionConfigOptionRequest, } from '@agentclientprotocol/sdk'; +import type { SessionEvent } from '@maka/core/events'; +import type { StoredMessage } from '@maka/core/session'; import { THINKING_LEVELS, type ThinkingLevel } from '@maka/core/model-thinking'; import { RuntimeHostOperationError, @@ -35,9 +38,16 @@ import { } from '@maka/runtime-host/client'; import { SESSION_CATALOG_CWD_MAX_BYTES, + SESSION_CONTINUITY_SCHEMA_VERSION, type SessionCatalogProjection, + type SessionContinuitySnapshot, } from '@maka/runtime-host/protocol'; -import { AcpSessionRegistry, type AcpSessionRegistryConnection } from '../acp/session-registry.js'; +import { + AcpSessionRegistry, + type AcpSessionAttachment, + type AcpSessionAttachmentOpenInput, + 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; @@ -130,6 +140,15 @@ describe('ACP Session registry', () => { value: 'bypass', }), ], + [ + 'turn.start', + () => + registry.prompt( + { sessionId: 'session-closed', prompt: [{ type: 'text', text: 'hello' }] }, + promptContext([]), + ), + ], + ['session.close', () => registry.close({ sessionId: 'session-closed' })], ] as const) { await assert.rejects(request(), (error: unknown) => { assert.ok(error instanceof RequestError); @@ -332,6 +351,871 @@ describe('ACP Session registry', () => { await registry.dispose(); }); + test('rejects unsupported prompt content before attaching or starting a Turn', async () => { + let attachmentOpens = 0; + const turnRequests: string[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + turnRequests.push(operation); + return catalogSession('session-prompt-validation'); + }, + }), + newSessionId: () => 'session-prompt-validation', + openSessionAttachment: async () => { + attachmentOpens += 1; + return new FakeAcpSessionAttachment('session-prompt-validation'); + }, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + turnRequests.length = 0; + + await assertInvalidParams( + registry.prompt( + { + sessionId: 'session-prompt-validation', + prompt: [{ type: 'image', data: '', mimeType: 'image/png' }], + }, + promptContext([]), + ), + { field: 'prompt', reason: 'unsupported_content_type' }, + ); + + assert.equal(attachmentOpens, 0); + assert.deepEqual(turnRequests, []); + await registry.dispose(); + }); + + test('shares a concurrent first attachment and starts event consumption before turn.start', async () => { + const notifications: SessionNotification[] = []; + const attachment = new FakeAcpSessionAttachment('session-concurrent-prompt'); + const attachGate = deferred(); + let attachmentOpens = 0; + const startedTurnIds: string[] = []; + const turnIds = ['turn-a', 'turn-b']; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return catalogSession('session-concurrent-prompt'); + if (operation === 'turn.start') { + const turnId = (input as { turnId: string }).turnId; + assert.equal(attachment.nextCalls(turnId), 1); + startedTurnIds.push(turnId); + queueMicrotask(() => { + attachment.emit( + turnId, + sessionEvent(turnId, { + type: 'text_complete', + messageId: `message-${turnId}`, + text: turnId, + }), + ); + attachment.emit( + turnId, + sessionEvent(turnId, { type: 'complete', stopReason: 'end_turn' }), + ); + attachment.finish(turnId); + }); + return { + kind: 'started', + turn: { + sessionId: 'session-concurrent-prompt', + turnId, + runId: `run-${turnId}`, + status: 'running', + }, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; + } + throw new Error(`Unexpected operation ${operation}`); + }, + }), + newSessionId: () => 'session-concurrent-prompt', + newTurnId: () => turnIds.shift()!, + openSessionAttachment: async () => { + attachmentOpens += 1; + return attachGate.promise; + }, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + const first = registry.prompt( + { sessionId: 'session-concurrent-prompt', prompt: [{ type: 'text', text: 'one' }] }, + promptContext(notifications), + ); + const second = registry.prompt( + { sessionId: 'session-concurrent-prompt', prompt: [{ type: 'text', text: 'two' }] }, + promptContext(notifications), + ); + await waitFor(() => attachmentOpens === 1); + attachGate.resolve(attachment); + + assert.deepEqual(await Promise.all([first, second]), [ + { stopReason: 'end_turn' }, + { stopReason: 'end_turn' }, + ]); + assert.deepEqual(new Set(startedTurnIds), new Set(['turn-a', 'turn-b'])); + assert.equal(attachmentOpens, 1); + assert.deepEqual( + new Set( + notifications.flatMap(({ update }) => + update.sessionUpdate === 'agent_message_chunk' && update.content.type === 'text' + ? [update.content.text] + : [], + ), + ), + new Set(['turn-a', 'turn-b']), + ); + await registry.dispose(); + assert.equal(attachment.closeCalls, 1); + }); + + test('latches cancellation while the initial attachment is pending and never dispatches', async () => { + const attachment = new FakeAcpSessionAttachment('session-cancel-before-attach'); + const attachGate = deferred(); + let turnStarts = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') + return catalogSession('session-cancel-before-attach'); + if (operation === 'turn.start') turnStarts += 1; + return {}; + }, + }), + newSessionId: () => 'session-cancel-before-attach', + newTurnId: () => 'turn-cancelled', + openSessionAttachment: async () => attachGate.promise, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry.prompt( + { + sessionId: 'session-cancel-before-attach', + prompt: [{ type: 'text', text: 'cancel me' }], + }, + promptContext([]), + ); + await registry.cancel({ sessionId: 'session-cancel-before-attach' }); + attachGate.resolve(attachment); + + assert.deepEqual(await prompt, { stopReason: 'cancelled' }); + assert.equal(turnStarts, 0); + await registry.dispose(); + }); + + test('waits for the live root identity before issuing exactly one turn.stop', async () => { + const attachment = new FakeAcpSessionAttachment('session-cancel-live'); + const startGate = deferred(); + const stopInputs: unknown[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return catalogSession('session-cancel-live'); + if (operation === 'turn.start') return startGate.promise; + if (operation === 'turn.stop') { + stopInputs.push(input); + return { + sessionId: 'session-cancel-live', + turnId: 'turn-live', + runId: 'run-live', + status: 'cancelled', + terminalEventId: 'terminal-live', + abortSource: 'user', + }; + } + throw new Error(`Unexpected operation ${operation}`); + }, + }), + newSessionId: () => 'session-cancel-live', + newTurnId: () => 'turn-live', + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry.prompt( + { sessionId: 'session-cancel-live', prompt: [{ type: 'text', text: 'run' }] }, + promptContext([]), + ); + await waitFor(() => attachment.nextCalls('turn-live') === 1); + const cancel = registry.cancel({ sessionId: 'session-cancel-live' }); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(stopInputs, []); + + attachment.setRoot({ + sessionId: 'session-cancel-live', + turnId: 'turn-live', + runId: 'run-live', + status: 'running', + }); + startGate.resolve({ + kind: 'started', + turn: { + sessionId: 'session-cancel-live', + turnId: 'turn-live', + runId: 'run-live', + status: 'running', + }, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }); + await cancel; + await registry.cancel({ sessionId: 'session-cancel-live' }); + assert.deepEqual(await prompt, { stopReason: 'cancelled' }); + assert.deepEqual(stopInputs, [ + { sessionId: 'session-cancel-live', turnId: 'turn-live', runId: 'run-live' }, + ]); + await registry.dispose(); + }); + + for (const timing of ['before interruption', 'after interruption'] as const) { + for (const recovery of [ + 'subscription', + 'query', + 'not-found', + 'terminal', + 'shutdown', + ] as const) { + test(`retains cancellation ${timing} until unknown admission resolves via ${recovery}`, async () => { + const sessionId = 'session-unknown-start'; + const turn = { + sessionId, + turnId: 'turn-unknown', + runId: 'run-recovered', + status: 'running' as const, + }; + const attachment = new FakeAcpSessionAttachment(sessionId); + const start = deferred(); + const query = deferred(); + const stopInputs: unknown[] = []; + let starts = 0; + let queries = 0; + let settled = false; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return catalogSession(sessionId); + if (operation === 'turn.start') { + starts += 1; + return start.promise; + } + if (operation === 'turn.query') { + assert.deepEqual(input, { sessionId, turnId: turn.turnId }); + queries += 1; + return query.promise; + } + if (operation === 'turn.stop') { + stopInputs.push(input); + attachment.setRoot({ + ...turn, + status: 'cancelled', + terminalEventId: 'terminal-unknown', + abortSource: 'user', + }); + return attachment.snapshot.rootTurn; + } + throw new Error(`Unexpected operation ${operation}`); + }, + }), + newSessionId: () => sessionId, + newTurnId: () => turn.turnId, + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry + .prompt({ sessionId, prompt: [{ type: 'text', text: 'run' }] }, promptContext([])) + .then((result) => { + settled = true; + return result; + }); + await waitFor(() => attachment.nextCalls(turn.turnId) === 1); + const cancel = () => + recovery === 'shutdown' ? registry.dispose() : registry.cancel({ sessionId }); + let cancellation = timing === 'before interruption' ? cancel() : undefined; + start.reject( + new RuntimeHostRequestInterruptedError( + 'turn.start', + 'command', + 'dispatched', + 'connection_lost', + ), + ); + await waitFor(() => queries === 1); + cancellation ??= cancel(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(settled, false); + assert.deepEqual(stopInputs, []); + if (recovery === 'subscription') { + // A transient query failure and an unrelated root do not retire or + // redirect the original cancellation intent. + query.reject( + new RuntimeHostRequestInterruptedError('turn.query', 'query', 'dispatched', 'timeout'), + ); + attachment.setRoot({ ...turn, turnId: 'other-turn', runId: 'other-run' }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(settled, false); + assert.deepEqual(stopInputs, []); + attachment.setRoot(turn); + } else if (recovery === 'not-found') { + query.reject( + new RuntimeHostOperationError('turn.query', 'not_found', 'Turn was not admitted'), + ); + } else if (recovery === 'terminal') { + query.resolve({ ...turn, status: 'completed', terminalEventId: 'terminal-unknown' }); + } else { + if (recovery === 'shutdown') assert.equal(attachment.closeCalls, 1); + query.resolve(turn); + } + await cancellation; + assert.deepEqual(await prompt, { stopReason: 'cancelled' }); + assert.deepEqual( + stopInputs, + recovery === 'not-found' || recovery === 'terminal' + ? [] + : [{ sessionId, turnId: turn.turnId, runId: turn.runId }], + ); + assert.equal(starts, 1); + assert.equal(queries, 1); + await registry.dispose(); + }); + } + } + + test('shutdown stops a late admitted start after observation has closed', async () => { + const sessionId = 'session-late-start'; + const turn = { sessionId, turnId: 'turn-late', runId: 'run-late', status: 'running' as const }; + const start = deferred(); + const stop = deferred(); + const calls: string[] = []; + const attachment = new FakeAcpSessionAttachment(sessionId); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return catalogSession(sessionId); + if (operation === 'turn.start') return start.promise; + if (operation === 'turn.stop') { + assert.deepEqual(input, { sessionId, turnId: turn.turnId, runId: turn.runId }); + calls.push('stop'); + return stop.promise; + } + throw new Error(`Unexpected operation ${operation}`); + }, + close: async () => { + calls.push('connection.close'); + }, + }), + newSessionId: () => sessionId, + newTurnId: () => turn.turnId, + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry.prompt( + { sessionId, prompt: [{ type: 'text', text: 'run' }] }, + promptContext([]), + ); + await waitFor(() => attachment.nextCalls(turn.turnId) === 1); + const disposal = registry.dispose(); + await waitFor(() => attachment.closeCalls === 1); + await new Promise((resolve) => setImmediate(resolve)); + start.resolve({ + kind: 'started', + turn, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }); + try { + await waitFor(() => calls.includes('stop')); + assert.deepEqual(calls, ['stop']); + } finally { + stop.resolve({ ...turn, status: 'cancelled' }); + await disposal; + await prompt; + } + assert.deepEqual(await prompt, { stopReason: 'cancelled' }); + assert.deepEqual(calls, ['stop', 'connection.close']); + }); + + for (const timing of ['before start returns', 'after start returns'] as const) { + for (const action of ['cancel', 'abort'] as const) { + test(`${action} completes the prompt when Stop delivery rejects ${timing} without another event`, async (t) => { + const diagnostic = t.mock.method(console, 'error', () => undefined); + const start = deferred(); + const abort = new AbortController(); + const sessionId = 'session-stop-reject'; + const turn = { + sessionId, + turnId: 'turn-reject', + runId: 'run-reject', + status: 'running' as const, + }; + const failure = new Error('Stop delivery failed'); + const attachment = new FakeAcpSessionAttachment(sessionId); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession(sessionId); + if (operation === 'turn.start') return start.promise; + if (operation === 'turn.stop') throw failure; + throw new Error(`Unexpected operation ${operation}`); + }, + }), + newSessionId: () => sessionId, + newTurnId: () => turn.turnId, + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + let outcome: unknown; + const prompt = registry + .prompt( + { sessionId, prompt: [{ type: 'text', text: 'run' }] }, + { ...promptContext([]), signal: abort.signal }, + ) + .then( + (result) => { + outcome = result; + }, + (error: unknown) => { + outcome = error; + }, + ); + await waitFor(() => attachment.nextCalls(turn.turnId) === 1); + const started = { + kind: 'started', + turn, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; + if (timing === 'after start returns') { + start.resolve(started); + await new Promise((resolve) => setImmediate(resolve)); + } + attachment.setRoot(turn); + const cancellation = action === 'cancel' ? registry.cancel({ sessionId }) : abort.abort(); + await waitFor(() => diagnostic.mock.callCount() === 1); + start.resolve(started); + await cancellation; + try { + await waitFor(() => outcome !== undefined); + assert.deepEqual(outcome, { stopReason: 'cancelled' }); + assert.deepEqual(diagnostic.mock.calls[0]?.arguments, [ + '[acp] Host Stop delivery failed:', + failure, + ]); + assert.equal(attachment.closeCalls, 0); + assert.equal(attachment.snapshot.rootTurn?.status, 'running'); + } finally { + await registry.dispose(); + await prompt; + } + }); + } + } + + test('close removes ownership immediately and still closes attachment after stop failure', async () => { + const attachment = new FakeAcpSessionAttachment('session-close-live'); + const stopFailure = new Error('stop failed'); + let turnStarted = false; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession('session-close-live'); + if (operation === 'turn.start') { + turnStarted = true; + return { + kind: 'started', + turn: { + sessionId: 'session-close-live', + turnId: 'turn-close', + runId: 'run-close', + status: 'running', + }, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; + } + if (operation === 'turn.stop') throw stopFailure; + if (operation === 'session.catalog.query') { + return { + kind: 'page', + revision: SESSION_REVISION, + sessions: [catalogSession('session-close-live')], + nextCursor: null, + }; + } + throw new Error(`Unexpected operation ${operation}`); + }, + }), + newSessionId: () => 'session-close-live', + newTurnId: () => 'turn-close', + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry + .prompt( + { sessionId: 'session-close-live', prompt: [{ type: 'text', text: 'run' }] }, + promptContext([]), + ) + .catch((error: unknown) => error); + await waitFor(() => turnStarted); + attachment.setRoot({ + sessionId: 'session-close-live', + turnId: 'turn-close', + runId: 'run-close', + status: 'running', + }); + + const firstClose = registry.close({ sessionId: 'session-close-live' }); + const concurrentClose = registry.close({ sessionId: 'session-close-live' }); + await assertInvalidParams( + registry.prompt( + { sessionId: 'session-close-live', prompt: [{ type: 'text', text: 'late' }] }, + promptContext([]), + ), + { reason: 'unknown_session' }, + ); + const closeOutcomes = await Promise.allSettled([firstClose, concurrentClose]); + assert.deepEqual( + closeOutcomes.map((outcome) => + outcome.status === 'rejected' ? outcome.reason : outcome.value, + ), + [stopFailure, stopFailure], + ); + assert.deepEqual(await prompt, { stopReason: 'cancelled' }); + assert.equal(attachment.closeCalls, 1); + assert.deepEqual(await registry.list({}), { + sessions: [ + { + sessionId: 'session-close-live', + cwd: '/workspace', + title: 'session-close-live', + updatedAt: '1970-01-01T00:00:00.001Z', + }, + ], + }); + await assertInvalidParams(registry.close({ sessionId: 'session-close-live' }), { + reason: 'unknown_session', + }); + await registry.dispose(); + }); + + for (const action of ['cancel', 'close', 'dispose'] as const) { + test(`${action} stops an externally started root on an idle attachment`, async () => { + const sessionId = 'external-root'; + const attachment = new FakeAcpSessionAttachment(sessionId); + const calls: Array<{ operation: string; input: unknown }> = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + calls.push({ operation, input }); + if (operation === 'session.create') return catalogSession(sessionId); + if (operation === 'turn.start') { + attachment.emit( + 'local', + sessionEvent('local', { type: 'complete', stopReason: 'end_turn' }), + ); + return { kind: 'started' }; + } + if (operation === 'turn.stop') { + assert.equal(attachment.closeCalls, 0); + return {}; + } + throw new Error(operation); + }, + }), + newSessionId: () => sessionId, + newTurnId: () => 'local', + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + await registry.prompt( + { sessionId, prompt: [{ type: 'text', text: 'hello' }] }, + promptContext([]), + ); + attachment.setRoot({ + sessionId, + turnId: 'external', + runId: 'external-run', + status: 'running', + }); + try { + if (action === 'dispose') await registry.dispose(); + else await registry[action]({ sessionId }); + assert.deepEqual( + calls.filter(({ operation }) => operation === 'turn.stop'), + [ + { + operation: 'turn.stop', + input: { sessionId, turnId: 'external', runId: 'external-run' }, + }, + ], + ); + } finally { + attachment.setRoot(null); + await registry.dispose(); + } + }); + } + + for (const source of ['complete', 'recovery'] as const) { + test(`fails a ${source} rewrite, stops its exact root, and permits another prompt`, async () => { + const sessionId = 'rewrite'; + const attachment = new FakeAcpSessionAttachment(sessionId); + const notifications: SessionNotification[] = []; + const stops: unknown[] = []; + let turnNumber = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return catalogSession(sessionId); + if (operation === 'turn.start') { + const { turnId } = input as { turnId: string }; + attachment.setRoot({ + sessionId, + turnId, + runId: `run-${turnId}`, + status: 'running', + }); + if (turnId === 'turn-1') { + attachment.emit( + turnId, + sessionEvent(turnId, { type: 'text_delta', messageId: 'answer', text: 'old' }), + ); + } else { + attachment.emit( + turnId, + sessionEvent(turnId, { type: 'complete', stopReason: 'end_turn' }), + ); + } + return { kind: 'started' }; + } + if (operation === 'turn.stop') { + stops.push(input); + return {}; + } + throw new Error(operation); + }, + }), + newSessionId: () => sessionId, + newTurnId: () => `turn-${++turnNumber}`, + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry.prompt( + { sessionId, prompt: [{ type: 'text', text: 'hello' }] }, + promptContext(notifications), + ); + const rejected = assert.rejects(prompt, { + data: { source: 'adapter', code: 'unsupported_stream_revision' }, + }); + await waitFor(() => notifications.length === 1); + if (source === 'complete') { + attachment.emit( + 'turn-1', + sessionEvent('turn-1', { type: 'text_complete', messageId: 'answer', text: '' }), + ); + } else { + attachment.replaceTranscript('turn-1', [ + { + type: 'assistant', + id: 'answer', + turnId: 'turn-1', + ts: 1, + text: 'new', + modelId: 'default', + }, + ]); + } + try { + await rejected; + assert.deepEqual(stops, [{ sessionId, turnId: 'turn-1', runId: 'run-turn-1' }]); + assert.equal(notifications.length, 1); + assert.deepEqual( + await registry.prompt( + { sessionId, prompt: [{ type: 'text', text: 'next' }] }, + promptContext([]), + ), + { stopReason: 'end_turn' }, + ); + } finally { + attachment.setRoot(null); + await registry.dispose(); + } + }); + } + + test('retires a failed attachment so the next prompt opens a fresh one', async () => { + const first = new FakeAcpSessionAttachment('session-reattach'); + const second = new FakeAcpSessionAttachment('session-reattach'); + let attachmentOpens = 0; + let starts = 0; + const stops: unknown[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return catalogSession('session-reattach'); + if (operation === 'turn.stop') { + stops.push(input); + return {}; + } + if (operation !== 'turn.start') throw new Error(`Unexpected operation ${operation}`); + starts += 1; + const turnId = (input as { turnId: string }).turnId; + const attachment = starts === 1 ? first : second; + queueMicrotask(() => { + if (starts === 1) { + attachment.failAttachment(new Error('subscription failed')); + } else { + attachment.emit( + turnId, + sessionEvent(turnId, { type: 'complete', stopReason: 'end_turn' }), + ); + attachment.finish(turnId); + } + }); + return { + kind: 'started', + turn: { + sessionId: 'session-reattach', + turnId, + runId: `run-${turnId}`, + status: 'running', + }, + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; + }, + }), + newSessionId: () => 'session-reattach', + newTurnId: (() => { + const ids = ['turn-first', 'turn-second']; + return () => ids.shift()!; + })(), + openSessionAttachment: async (input) => { + attachmentOpens += 1; + return (attachmentOpens === 1 ? first : second).bind(input); + }, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + + await assert.rejects( + registry.prompt( + { sessionId: 'session-reattach', prompt: [{ type: 'text', text: 'first' }] }, + promptContext([]), + ), + /subscription failed/u, + ); + assert.deepEqual( + await registry.prompt( + { sessionId: 'session-reattach', prompt: [{ type: 'text', text: 'second' }] }, + promptContext([]), + ), + { stopReason: 'end_turn' }, + ); + assert.equal(attachmentOpens, 2); + assert.deepEqual(stops, [ + { sessionId: 'session-reattach', turnId: 'turn-first', runId: 'run-turn-first' }, + ]); + await registry.dispose(); + }); + + for (const action of ['close', 'shutdown', 'failure'] as const) { + test(`handles ${action} before attachment open settles without starting a Turn`, async () => { + const attachment = new FakeAcpSessionAttachment('pending'); + const gate = deferred(); + let opening = false; + let starts = 0; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession('pending'); + starts += 1; + throw new Error('unexpected Turn admission'); + }, + }), + newSessionId: () => 'pending', + openSessionAttachment: async (input) => { + attachment.bind(input); + opening = true; + return gate.promise; + }, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry.prompt( + { sessionId: 'pending', prompt: [{ type: 'text', text: 'hello' }] }, + promptContext([]), + ); + const outcome = prompt.then( + (result) => result, + (error: unknown) => error, + ); + await waitFor(() => opening); + const closing = + action === 'close' + ? registry.close({ sessionId: 'pending' }) + : action === 'shutdown' + ? registry.dispose() + : Promise.resolve(); + if (action === 'failure') attachment.failAttachment(new Error('early subscription EOF')); + gate.resolve(attachment); + await closing; + const result = await outcome; + if (action === 'failure') assert.ok(result instanceof RequestError); + else assert.deepEqual(result, { stopReason: 'cancelled' }); + assert.equal(starts, 0); + assert.equal(attachment.closeCalls, 1); + await registry.dispose(); + }); + } + + test('shutdown cancels active prompts and closes attachments before the shared Host', async () => { + const lifecycle: string[] = []; + const startGate = deferred(); + const attachment = new FakeAcpSessionAttachment('session-shutdown', () => { + lifecycle.push('attachment.close'); + }); + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession('session-shutdown'); + if (operation === 'turn.start') return startGate.promise; + throw new Error(`Unexpected operation ${operation}`); + }, + close: async () => { + lifecycle.push('connection.close'); + }, + }), + newSessionId: () => 'session-shutdown', + newTurnId: () => 'turn-shutdown', + openSessionAttachment: async (input) => attachment.bind(input), + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry.prompt( + { sessionId: 'session-shutdown', prompt: [{ type: 'text', text: 'run' }] }, + promptContext([]), + ); + await waitFor(() => attachment.nextCalls('turn-shutdown') === 1); + + const disposal = registry.dispose(); + await waitFor(() => attachment.closeCalls === 1); + startGate.reject(new Error('start request interrupted')); + await disposal; + + assert.deepEqual(await prompt, { stopReason: 'cancelled' }); + assert.deepEqual(lifecycle, ['attachment.close', 'connection.close']); + await assert.rejects( + registry.list({}), + (error: unknown) => + error instanceof RequestError && + (error.data as { code?: string }).code === 'registry_closed', + ); + }); + test('returns projected configuration and owns only a representable successful create', async () => { const requests: Array<{ operation: string; input: unknown }> = []; let subscriptionOpens = 0; @@ -441,7 +1325,7 @@ describe('ACP Session registry', () => { await registry.dispose(); }); - test('does not grant ownership after failed or legacy creates', async () => { + test('keeps failed creates unowned and returns committed IDs even for unsupported projections', async () => { for (const [name, createOutcome] of [ [ 'failed', @@ -471,6 +1355,15 @@ describe('ACP Session registry', () => { newSessionId: () => sessionId, }); + if (!(createOutcome instanceof Error)) { + assert.deepEqual(await registry.create({ cwd: '/workspace', mcpServers: [] }), { + sessionId, + }); + await registry.close({ sessionId }); + assert.equal(requests, 1); + await registry.dispose(); + continue; + } await assert.rejects(registry.create({ cwd: '/workspace', mcpServers: [] })); await assertInvalidParams( registry.setConfigOption({ @@ -485,6 +1378,323 @@ describe('ACP Session registry', () => { } }); + test('returns the committed ID on catalog failure without admitting mutations during projection', async () => { + const catalog = deferred(); + let projecting = false; + const connection = fakeConnection({ request: async () => catalogSession('created') }); + const request = connection.request; + connection.request = (async (operation, input) => { + if (operation === 'connection.catalog.query') { + projecting = true; + return catalog.promise; + } + return request(operation, input); + }) as AcpSessionRegistryConnection['request']; + const registry = new AcpSessionRegistry({ + connect: async () => connection, + newSessionId: () => 'created', + }); + const creation = registry.create({ cwd: '/workspace', mcpServers: [] }); + await waitFor(() => projecting); + await assertInvalidParams( + registry.setConfigOption({ + sessionId: 'created', + configId: 'permission_mode', + value: 'bypass', + }), + { reason: 'unknown_session' }, + ); + catalog.reject(new Error('catalog unavailable')); + assert.deepEqual(await creation, { sessionId: 'created' }); + assert.deepEqual(await registry.close({ sessionId: 'created' }), {}); + await registry.dispose(); + }); + + test('publishes complete external options in order, including model changes, and stops after close', async () => { + const sessionId = 'external-options'; + const attachment = new FakeAcpSessionAttachment(sessionId); + let session = catalogSession(sessionId); + const notifications: SessionNotification[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return session; + if (operation === 'session.catalog.query') return { kind: 'session', session }; + if (operation === 'session.configuration.update') { + session = { + ...session, + ...(input as { patch: object }).patch, + revision: session.revision + 1, + }; + attachment.setMetadataRevision(session.revision); + return { kind: 'committed', session }; + } + if (operation === 'turn.start') { + attachment.emit( + 'turn', + sessionEvent('turn', { type: 'complete', stopReason: 'end_turn' }), + ); + return { kind: 'started' }; + } + throw new Error(operation); + }, + }), + newSessionId: () => sessionId, + newTurnId: () => 'turn', + openSessionAttachment: async (input) => { + attachment.bind(input); + attachment.setMetadataRevision(1); + return attachment; + }, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + await registry.prompt( + { sessionId, prompt: [{ type: 'text', text: 'hello' }] }, + promptContext(notifications), + ); + session = { ...session, revision: 2, model: 'non-reasoning' }; + attachment.setMetadataRevision(2); + await waitFor(() => notifications.length === 1); + const removed = notifications[0]!.update; + assert.equal(removed.sessionUpdate, 'config_option_update'); + if (removed.sessionUpdate !== 'config_option_update') assert.fail(); + assert.deepEqual( + removed.configOptions, + configOptions({}).filter(({ id }) => id !== 'thinking_level'), + ); + session = { ...session, revision: 3, model: 'default', thinkingLevel: 'high' }; + attachment.setMetadataRevision(3); + await waitFor(() => notifications.length === 2); + const added = notifications[1]!.update; + assert.equal(added.sessionUpdate, 'config_option_update'); + if (added.sessionUpdate !== 'config_option_update') assert.fail(); + assert.deepEqual(added.configOptions, configOptions({ thinking_level: 'high' })); + const configured = await registry.setConfigOption({ + sessionId, + configId: 'permission_mode', + value: 'bypass', + }); + assert.deepEqual(notifications[2]!.update, { + sessionUpdate: 'config_option_update', + configOptions: configured.configOptions, + }); + await registry.close({ sessionId }); + session = { ...session, revision: 5, model: 'non-reasoning' }; + attachment.setMetadataRevision(5); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(notifications.length, 3); + await registry.dispose(); + }); + + for (const replaceAttachment of [false, true]) { + test(`orders pending configuration responses before updates across ${replaceAttachment ? 'replacement' : 'first'} attachment`, async () => { + const sessionId = 'configuration-attachment-race'; + let session = catalogSession(sessionId); + let attachment: FakeAcpSessionAttachment | undefined; + let turn = 0; + let holdProjection = false; + const projectionStarted = deferred(); + const releaseProjection = deferred(); + const delivered: Array<[string, string | boolean]> = []; + const connection = fakeConnection({ + request: async (operation, input) => { + if (operation === 'session.create') return session; + if (operation === 'session.catalog.query') return { kind: 'session', session }; + if (operation === 'session.configuration.update') { + session = { + ...session, + ...(input as { patch: object }).patch, + revision: session.revision + 1, + }; + attachment?.setMetadataRevision(session.revision); + return { kind: 'committed', session }; + } + if (operation === 'turn.start') { + const { turnId } = input as { turnId: string }; + attachment!.emit( + turnId, + sessionEvent(turnId, { type: 'complete', stopReason: 'end_turn' }), + ); + return { kind: 'started' }; + } + throw new Error(operation); + }, + }); + const request = connection.request; + connection.request = (async (operation, input) => { + if (operation === 'connection.catalog.query' && holdProjection) { + holdProjection = false; + projectionStarted.resolve(); + await releaseProjection.promise; + } + return request(operation, input); + }) as AcpSessionRegistryConnection['request']; + const registry = new AcpSessionRegistry({ + connect: async () => connection, + newSessionId: () => sessionId, + newTurnId: () => `turn-${++turn}`, + openSessionAttachment: async (input) => { + attachment = new FakeAcpSessionAttachment(sessionId).bind(input); + attachment.setMetadataRevision(session.revision); + return attachment; + }, + }); + const prompt = () => + registry.prompt( + { sessionId, prompt: [{ type: 'text', text: 'hello' }] }, + { + signal: new AbortController().signal, + notify: async ({ update }) => { + if (update.sessionUpdate === 'config_option_update') { + delivered.push([ + 'notification', + update.configOptions.find(({ id }) => id === 'permission_mode')!.currentValue, + ]); + } + }, + }, + ); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + if (replaceAttachment) await prompt(); + holdProjection = true; + const setting = registry + .setConfigOption({ sessionId, configId: 'permission_mode', value: 'bypass' }) + .then(({ configOptions }) => { + delivered.push([ + 'response', + configOptions.find(({ id }) => id === 'permission_mode')!.currentValue, + ]); + }); + await projectionStarted.promise; + const previous = attachment; + if (replaceAttachment) previous!.failAttachment(new Error('subscription failed')); + const prompting = prompt(); + await waitFor(() => attachment !== undefined && attachment !== previous); + session = { ...session, revision: session.revision + 1, permissionMode: 'ask' }; + attachment!.setMetadataRevision(session.revision); + await new Promise((resolve) => setImmediate(resolve)); + releaseProjection.resolve(); + await Promise.all([setting, prompting]); + await waitFor(() => delivered.some(([kind]) => kind === 'notification')); + await registry.dispose(); + assert.deepEqual(delivered, [ + ['response', 'bypass'], + ['notification', 'ask'], + ]); + }); + } + + test('suppresses an external configuration projection that finishes after close', async () => { + const sessionId = 'closing-options'; + const attachment = new FakeAcpSessionAttachment(sessionId); + const read = deferred<{ kind: 'session'; session: SessionCatalogProjection }>(); + let reading = false; + const notifications: SessionNotification[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession(sessionId); + if (operation === 'session.catalog.query') { + reading = true; + return read.promise; + } + attachment.emit( + 'turn', + sessionEvent('turn', { type: 'complete', stopReason: 'end_turn' }), + ); + return { kind: 'started' }; + }, + }), + newSessionId: () => sessionId, + newTurnId: () => 'turn', + openSessionAttachment: async (input) => { + attachment.bind(input); + attachment.setMetadataRevision(1); + return attachment; + }, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + await registry.prompt( + { sessionId, prompt: [{ type: 'text', text: 'hello' }] }, + promptContext(notifications), + ); + attachment.setMetadataRevision(2); + await waitFor(() => reading); + await registry.close({ sessionId }); + read.resolve({ + kind: 'session', + session: catalogSession(sessionId, '/workspace', { revision: 2, permissionMode: 'bypass' }), + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(notifications, []); + await registry.dispose(); + }); + + test('closing an active prompt does not wait for a stalled configuration read', async () => { + const attachment = new FakeAcpSessionAttachment('stalled'); + const read = deferred<{ kind: 'session'; session: SessionCatalogProjection }>(); + let reading = false; + let started = false; + const notifications: SessionNotification[] = []; + const registry = new AcpSessionRegistry({ + connect: async () => + fakeConnection({ + request: async (operation) => { + if (operation === 'session.create') return catalogSession('stalled'); + if (operation === 'session.catalog.query') { + reading = true; + return read.promise; + } + if (operation === 'turn.stop') return {}; + started = true; + attachment.setRoot({ + sessionId: 'stalled', + turnId: 'turn', + runId: 'run', + status: 'running', + }); + attachment.setMetadataRevision(2); + attachment.emit( + 'turn', + sessionEvent('turn', { type: 'text_delta', messageId: 'answer', text: 'pending' }), + ); + return { kind: 'started' }; + }, + }), + newSessionId: () => 'stalled', + newTurnId: () => 'turn', + openSessionAttachment: async (input) => { + attachment.bind(input); + attachment.setMetadataRevision(1); + return attachment; + }, + }); + await registry.create({ cwd: '/workspace', mcpServers: [] }); + const prompt = registry.prompt( + { sessionId: 'stalled', prompt: [{ type: 'text', text: 'hello' }] }, + promptContext(notifications), + ); + await waitFor(() => reading && started); + let closed = false; + const closing = registry.close({ sessionId: 'stalled' }).then(() => { + closed = true; + }); + try { + await waitFor(() => closed); + assert.deepEqual(await prompt, { stopReason: 'cancelled' }); + assert.deepEqual(notifications, []); + } finally { + read.resolve({ + kind: 'session', + session: catalogSession('stalled', '/workspace', { revision: 2 }), + }); + await closing; + await registry.dispose(); + } + }); + test('rejects non-owned and invalid configuration requests before Host I/O', async () => { let requests = 0; const registry = new AcpSessionRegistry({ @@ -1404,14 +2614,170 @@ function fakeConnection( } = {}, ): AcpSessionRegistryConnection { return { + hostEpoch: 'host-1', request: async (operation, input) => operation === 'connection.catalog.query' ? connectionCatalogPage(overrides.thinkingLevels ?? THINKING_LEVELS) : (overrides.request?.(operation, input) ?? {}), + openSessionSubscription: async () => { + throw new Error('Unexpected recoverable subscription open'); + }, + openSessionSubscriptionOnce: async () => { + throw new Error('Unexpected initial subscription open'); + }, close: overrides.close ?? (async () => undefined), } as AcpSessionRegistryConnection; } +function promptContext(notifications: SessionNotification[]) { + return { + signal: new AbortController().signal, + notify: async (notification: SessionNotification) => void notifications.push(notification), + }; +} + +class FakeAcpSessionAttachment implements AcpSessionAttachment { + snapshot: SessionContinuitySnapshot; + closeCalls = 0; + #callbacks: AcpSessionAttachmentOpenInput | undefined; + readonly #streams = new Map(); + + constructor( + readonly sessionId: string, + readonly onClose: () => void = () => undefined, + ) { + this.snapshot = continuitySnapshot(sessionId); + } + + bind(input: AcpSessionAttachmentOpenInput): this { + this.#callbacks = input; + return this; + } + + eventsForTurn(turnId: string): AsyncIterable { + return this.#stream(turnId); + } + + failTurn(turnId: string, error: unknown): void { + this.#stream(turnId).fail(error); + } + + failAttachment(error: Error): void { + this.#callbacks?.onFailed(error); + for (const stream of this.#streams.values()) stream.fail(error); + } + + emit(turnId: string, event: SessionEvent): void { + this.#stream(turnId).push(event); + } + + finish(turnId: string): void { + this.#stream(turnId).finish(); + } + + nextCalls(turnId: string): number { + return this.#streams.get(turnId)?.nextCalls ?? 0; + } + + setRoot(rootTurn: SessionContinuitySnapshot['rootTurn']): void { + this.snapshot = { + ...this.snapshot, + projectionRevision: this.snapshot.projectionRevision + 1, + rootTurn, + }; + this.#callbacks?.onSnapshotChanged(this.snapshot); + } + + replaceTranscript(turnId: string, messages: readonly StoredMessage[]): void { + this.#callbacks?.onTranscriptReplaced(turnId, messages); + } + + setMetadataRevision(metadataRevision: number): void { + this.snapshot = { ...this.snapshot, session: { ...this.snapshot.session, metadataRevision } }; + this.#callbacks?.onSnapshotChanged(this.snapshot); + } + + async close(): Promise { + this.closeCalls += 1; + this.onClose(); + for (const stream of this.#streams.values()) stream.finish(); + } + + #stream(turnId: string): FakeEventStream { + let stream = this.#streams.get(turnId); + if (!stream) { + stream = new FakeEventStream(); + this.#streams.set(turnId, stream); + } + return stream; + } +} + +class FakeEventStream implements AsyncIterable, AsyncIterator { + readonly #events: SessionEvent[] = []; + readonly #waiters: Array<{ + resolve(value: IteratorResult): void; + reject(error: unknown): void; + }> = []; + nextCalls = 0; + #done = false; + + [Symbol.asyncIterator](): AsyncIterator { + return this; + } + + next(): Promise> { + this.nextCalls += 1; + const event = this.#events.shift(); + if (event) return Promise.resolve({ done: false, value: event }); + if (this.#done) return Promise.resolve({ done: true, value: undefined }); + return new Promise((resolve, reject) => this.#waiters.push({ resolve, reject })); + } + + push(event: SessionEvent): void { + const waiter = this.#waiters.shift(); + if (waiter) waiter.resolve({ done: false, value: event }); + else this.#events.push(event); + } + + fail(error: unknown): void { + this.#done = true; + for (const waiter of this.#waiters.splice(0)) waiter.reject(error); + } + + finish(): void { + this.#done = true; + for (const waiter of this.#waiters.splice(0)) { + waiter.resolve({ done: true, value: undefined }); + } + } +} + +function continuitySnapshot(sessionId: string): SessionContinuitySnapshot { + return { + schemaVersion: SESSION_CONTINUITY_SCHEMA_VERSION, + session: { + sessionId, + metadataRevision: 1, + status: 'active', + createdAt: 1, + isArchived: false, + }, + projectionRevision: 1, + rootTurn: null, + goal: null, + queue: { hostEpoch: 'host-1', queueRevision: 0, steering: [], followup: [] }, + interactions: { pending: [] }, + }; +} + +function sessionEvent>( + turnId: string, + value: T, +): SessionEvent { + return { id: `event-${turnId}`, turnId, ts: 1, ...value } as unknown as SessionEvent; +} + function connectionCatalogPage(thinkingLevels: readonly ThinkingLevel[]) { return { kind: 'page' as const, diff --git a/packages/cli/src/__tests__/acp-stdio-server.test.ts b/packages/cli/src/__tests__/acp-stdio-server.test.ts index 0114d9234e..060eb3cdee 100644 --- a/packages/cli/src/__tests__/acp-stdio-server.test.ts +++ b/packages/cli/src/__tests__/acp-stdio-server.test.ts @@ -42,7 +42,7 @@ describe('Maka ACP stdio server', () => { id: 1, result: { protocolVersion: 1, - agentCapabilities: { sessionCapabilities: { list: {} } }, + agentCapabilities: { sessionCapabilities: { list: {}, close: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: '0.2.0' }, }, @@ -239,12 +239,12 @@ describe('Maka ACP stdio server', () => { 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(methodFailure.error?.code, -32602); + assert.deepEqual(methodFailure.error?.data, { reason: 'unknown_session' }); assert.equal(harness.connectCalls(), 1); }); - test('keeps an unimplemented Session method Host-independent', async () => { + test('keeps close for an unknown Session Host-independent', async () => { const harness = createHarness([ `${JSON.stringify({ jsonrpc: '2.0', @@ -266,8 +266,8 @@ describe('Maka ACP stdio server', () => { .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(response.error?.code, -32602); + assert.deepEqual(response.error?.data, { reason: 'unknown_session' }); assert.equal(harness.connectCalls(), 0); }); }); @@ -306,9 +306,24 @@ function createHarness( connects += 1; if (options.connectError) throw options.connectError; return { - connection, + connection: { + ...connection, + reconnecting: true, + hostEpoch: connection.hostEpoch ?? 'host-1', + openSessionSubscription: + connection.openSessionSubscription?.bind(connection) ?? + (async () => { + throw new Error('Unexpected Session attachment'); + }), + openSessionSubscriptionOnce: + connection.openSessionSubscription?.bind(connection) ?? + (async () => { + throw new Error('Unexpected Session attachment'); + }), + subscribeConnectionAvailability: () => () => undefined, + }, close: () => connection.close(), - } as Awaited< + } as unknown as Awaited< ReturnType< typeof import('../runtime-host-cli-context.js').connectRuntimeHostCliConnection > diff --git a/packages/cli/src/__tests__/cli.test.ts b/packages/cli/src/__tests__/cli.test.ts index 09ed0228b9..6223153f07 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, session\/new, session\/list\)$/m, + /^ maka --acp Serve ACP v1 over stdio \(sessions, prompts, streaming, cancellation\)$/m, ); assert.match(help.text, /^ maka runtime-host serve /m); assert.doesNotMatch(help.text, /cli:dev/); diff --git a/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts b/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts index 77b8338f5d..f5e9aaf575 100644 --- a/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts +++ b/packages/cli/src/__tests__/tui-mcp-remote-publication.test.ts @@ -907,6 +907,9 @@ function reconnectingConnection(connection: RuntimeHostConnection) { return { ...connection, reconnecting: true as const, + openSessionSubscriptionOnce: ( + input: Parameters[0], + ) => connection.openSessionSubscription(input), subscribeConnectionAvailability: ( listener: (availability: { kind: 'connected'; diff --git a/packages/cli/src/acp/README.md b/packages/cli/src/acp/README.md new file mode 100644 index 0000000000..c130608eba --- /dev/null +++ b/packages/cli/src/acp/README.md @@ -0,0 +1,37 @@ + + +# ACP live Session behavior + +The adapter retains a Runtime Host subscription after the first prompt. Cancellation +and close use the subscription's current root identity, including a Turn started by +another Host client while the ACP attachment was idle. Close releases the subscription +and connection-local ownership; it does not delete or archive the durable Session. + +ACP v1 message chunks are append-only. Matching replay and prefix extensions are +supported. If a completed or recovered message changes text or thinking that was +already streamed (including clearing it), the adapter rejects the prompt with +JSON-RPC error `-32603` and `error.data.code: unsupported_stream_revision` and requests +a stop of that prompt's exact live root. It never represents a replacement by inventing +a new message ID or reports `end_turn` for that failed projection. The client may +still display the already delivered partial text; ACP v1 cannot retract it. The +Session remains owned and can accept another prompt or be closed. + +Local resource links must identify regular files. Filesystem admission rejects +non-regular files, including POSIX FIFOs, before reading their content. diff --git a/packages/cli/src/acp/maka-acp-agent.ts b/packages/cli/src/acp/maka-acp-agent.ts index 11926dc954..ac91c535f7 100644 --- a/packages/cli/src/acp/maka-acp-agent.ts +++ b/packages/cli/src/acp/maka-acp-agent.ts @@ -22,14 +22,17 @@ import type { AcpSessionRegistry } from './session-registry.js'; export interface MakaAcpAgentOptions { readonly version: string; - readonly sessionRegistry: Pick; + readonly sessionRegistry: Pick< + AcpSessionRegistry, + 'create' | 'list' | 'setConfigOption' | 'prompt' | 'cancel' | 'close' + >; } export function createMakaAcpAgent(options: MakaAcpAgentOptions): AgentApp { return agent({ name: 'maka' }) .onRequest(methods.agent.initialize, () => ({ protocolVersion: 1, - agentCapabilities: { sessionCapabilities: { list: {} } }, + agentCapabilities: { sessionCapabilities: { list: {}, close: {} } }, authMethods: [], agentInfo: { name: 'maka', title: 'Maka', version: options.version }, })) @@ -37,5 +40,15 @@ export function createMakaAcpAgent(options: MakaAcpAgentOptions): AgentApp { .onRequest(methods.agent.session.list, ({ params }) => options.sessionRegistry.list(params)) .onRequest(methods.agent.session.setConfigOption, ({ params }) => options.sessionRegistry.setConfigOption(params), - ); + ) + .onRequest(methods.agent.session.prompt, ({ params, signal, client }) => + options.sessionRegistry.prompt(params, { + signal, + notify: (notification) => client.notify(methods.client.session.update, notification), + }), + ) + .onNotification(methods.agent.session.cancel, ({ params }) => + options.sessionRegistry.cancel(params), + ) + .onRequest(methods.agent.session.close, ({ params }) => options.sessionRegistry.close(params)); } diff --git a/packages/cli/src/acp/prompt-content.ts b/packages/cli/src/acp/prompt-content.ts new file mode 100644 index 0000000000..17e2c96311 --- /dev/null +++ b/packages/cli/src/acp/prompt-content.ts @@ -0,0 +1,149 @@ +/* + * 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 { constants } from 'node:fs'; +import { open, realpath } from 'node:fs/promises'; +import { basename } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { RequestError, type ContentBlock } from '@agentclientprotocol/sdk'; +import { + attachmentKindFromMimeType, + MAX_ATTACHMENT_BYTES, + MAX_ATTACHMENT_COUNT, + PDF_HEADER_SCAN_BYTES, + resolveAttachmentMimeType, +} from '@maka/core/attachments'; +import type { AttachmentRef, MessageContent } from '@maka/core/events'; + +interface OpenedPromptFile { + readonly size: number; + readonly isFile: boolean; + readonly prefix: Uint8Array; + readonly canonicalPath: string; +} + +export interface AcpPromptContentDependencies { + readonly openFile?: (path: string) => Promise; +} + +export async function mapAcpPromptContent( + prompt: readonly ContentBlock[], + dependencies: AcpPromptContentDependencies = {}, +): Promise { + const resources = prompt.filter( + (block): block is Extract => + block.type === 'resource_link', + ); + if (resources.length > MAX_ATTACHMENT_COUNT) { + throw invalidPrompt('prompt', 'too_many_attachments'); + } + for (const block of prompt) { + if (block.type !== 'text' && block.type !== 'resource_link') { + throw invalidPrompt('prompt', 'unsupported_content_type'); + } + } + + const modelParts: string[] = []; + const displayParts: string[] = []; + const attachments: AttachmentRef[] = []; + for (const block of prompt) { + if (block.type === 'text') { + modelParts.push(block.text); + displayParts.push(block.text); + continue; + } + if (block.type !== 'resource_link') { + throw invalidPrompt('prompt', 'unsupported_content_type'); + } + const path = localFilePath(block.uri); + const file = await (dependencies.openFile ?? readPromptFile)(path).catch((error: unknown) => { + if (error instanceof RequestError) throw error; + throw invalidPrompt('prompt', 'resource_unreadable'); + }); + if (!file.isFile) throw invalidPrompt('prompt', 'resource_not_file'); + if (!Number.isSafeInteger(file.size) || file.size < 0 || file.size > MAX_ATTACHMENT_BYTES) { + throw invalidPrompt('prompt', 'resource_too_large'); + } + const name = block.name || basename(file.canonicalPath); + const mimeType = resolveAttachmentMimeType(file.prefix, block.mimeType ?? undefined, name); + modelParts.push(block.uri); + attachments.push({ + kind: attachmentKindFromMimeType(mimeType, name), + name, + mimeType, + bytes: file.size, + ref: { kind: 'external_file', absolutePath: file.canonicalPath }, + }); + } + const text = modelParts.join('\n\n'); + const displayText = displayParts.join('\n\n'); + return { + text, + ...(displayText !== text ? { displayText } : {}), + ...(attachments.length > 0 ? { attachments } : {}), + }; +} + +async function readPromptFile(path: string): Promise { + // A FIFO must not wait for a writer before we can reject it. + const handle = await open(path, constants.O_RDONLY | constants.O_NONBLOCK); + try { + const stats = await handle.stat(); + if (!stats.isFile()) throw invalidPrompt('prompt', 'resource_not_file'); + const prefix = Buffer.alloc(Math.min(PDF_HEADER_SCAN_BYTES, stats.size)); + const { bytesRead } = await handle.read(prefix, 0, prefix.length, 0); + return { + size: stats.size, + isFile: stats.isFile(), + prefix: prefix.subarray(0, bytesRead), + canonicalPath: await realpath(path), + }; + } finally { + await handle.close(); + } +} + +function localFilePath(uri: string): string { + let url: URL; + try { + url = new URL(uri); + } catch { + throw invalidPrompt('prompt', 'invalid_resource_uri'); + } + if ( + url.protocol !== 'file:' || + url.hostname !== '' || + url.username !== '' || + url.password !== '' || + url.port !== '' || + url.search !== '' || + url.hash !== '' + ) { + throw invalidPrompt('prompt', 'unsupported_resource_uri'); + } + try { + return fileURLToPath(url); + } catch { + throw invalidPrompt('prompt', 'invalid_resource_uri'); + } +} + +function invalidPrompt(field: string, reason: string): RequestError { + return RequestError.invalidParams({ field, reason }, 'Invalid ACP prompt content'); +} diff --git a/packages/cli/src/acp/session-event-mapper.ts b/packages/cli/src/acp/session-event-mapper.ts new file mode 100644 index 0000000000..8c71e1804f --- /dev/null +++ b/packages/cli/src/acp/session-event-mapper.ts @@ -0,0 +1,155 @@ +/* + * 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 { foldRuntimeHostAssistantDelta } from '@maka/runtime-host/adapter'; +import { + RequestError, + type SessionNotification, + type SessionUpdate, + type StopReason, +} from '@agentclientprotocol/sdk'; +import type { SessionEvent } from '@maka/core/events'; +import type { StoredMessage } from '@maka/core/session'; + +type StreamKind = 'text' | 'thinking'; + +export interface AcpSessionEventMapperOptions { + readonly sessionId: string; + readonly notify: (notification: SessionNotification) => Promise; +} + +/** Serializes one ACP prompt's live projection and terminal outcome. */ +export class AcpSessionEventMapper { + readonly #sessionId: string; + readonly #notify: (notification: SessionNotification) => Promise; + readonly #streams = new Map(); + #tail: Promise = Promise.resolve(); + #terminal: StopReason | undefined; + #failure: RequestError | undefined; + + constructor(options: AcpSessionEventMapperOptions) { + this.#sessionId = options.sessionId; + this.#notify = options.notify; + } + + accept(event: SessionEvent): Promise { + return this.#enqueue(async () => { + if (this.#failure) throw this.#failure; + if (this.#terminal) return this.#terminal; + switch (event.type) { + case 'text_delta': + await this.#acceptText( + 'text', + event.messageId, + deltaText(event, this.#streams.get(streamKey('text', event.messageId))), + ); + break; + case 'text_complete': + await this.#acceptText('text', event.messageId, event.text); + break; + case 'thinking_delta': + await this.#acceptText( + 'thinking', + event.messageId, + deltaText(event, this.#streams.get(streamKey('thinking', event.messageId))), + ); + break; + case 'thinking_complete': + await this.#acceptText('thinking', event.messageId, event.text); + break; + case 'complete': + this.#terminal = 'end_turn'; + break; + case 'error': + if (!event.recoverable) this.#terminal = 'end_turn'; + break; + case 'abort': + this.#terminal = 'end_turn'; + break; + default: + break; + } + return this.#terminal; + }); + } + + replaceTranscript(turnId: string, messages: readonly StoredMessage[]): Promise { + return this.#enqueue(async () => { + if (this.#failure) throw this.#failure; + if (this.#terminal) return; + for (const message of messages) { + if (message.turnId !== turnId || message.type !== 'assistant') continue; + await this.#acceptText('thinking', message.id, message.thinking?.text ?? ''); + await this.#acceptText('text', message.id, message.text); + } + }); + } + + cancel(): Promise { + return this.#enqueue(async () => { + this.#terminal ??= 'cancelled'; + return this.#terminal; + }); + } + + async #acceptText(kind: StreamKind, hostMessageId: string, nextText: string): Promise { + const key = streamKey(kind, hostMessageId); + const current = this.#streams.get(key) ?? ''; + if (!nextText.startsWith(current)) { + // ACP v1 chunks only append. A new message ID cannot retract prior output. + this.#failure = RequestError.internalError( + { source: 'adapter', code: 'unsupported_stream_revision' }, + 'Runtime Host revised streamed output that ACP v1 cannot replace; the prompt failed', + ); + throw this.#failure; + } + const chunk = nextText.slice(current.length); + this.#streams.set(key, nextText); + if (chunk.length === 0) return; + const update: SessionUpdate = { + sessionUpdate: kind === 'text' ? 'agent_message_chunk' : 'agent_thought_chunk', + content: { type: 'text', text: chunk }, + messageId: hostMessageId, + }; + await this.#notify({ sessionId: this.#sessionId, update }); + } + + #enqueue(operation: () => Promise): Promise { + const result = this.#tail.then(operation, operation); + this.#tail = result.then( + () => undefined, + () => undefined, + ); + return result; + } +} + +function deltaText( + event: Extract, + current = '', +): string { + return foldRuntimeHostAssistantDelta(current, { + startOffset: event.startOffset ?? current.length, + text: event.text, + }).text; +} + +function streamKey(kind: StreamKind, messageId: string): string { + return `${kind}:${messageId}`; +} diff --git a/packages/cli/src/acp/session-registry.ts b/packages/cli/src/acp/session-registry.ts index 7d8f28e109..64ad995390 100644 --- a/packages/cli/src/acp/session-registry.ts +++ b/packages/cli/src/acp/session-registry.ts @@ -22,14 +22,24 @@ import { realpath } from 'node:fs/promises'; import { isAbsolute, normalize } from 'node:path'; import { RequestError, + type CancelNotification, + type CloseSessionRequest, + type CloseSessionResponse, type ListSessionsRequest, type ListSessionsResponse, type NewSessionRequest, type NewSessionResponse, + type PromptRequest, + type PromptResponse, + type SessionNotification, type SessionConfigOption, type SetSessionConfigOptionRequest, type SetSessionConfigOptionResponse, + type StopReason, } from '@agentclientprotocol/sdk'; +import type { SessionEvent } from '@maka/core/events'; +import { isRuntimeHostTerminalTurn } from '@maka/runtime-host/adapter'; +import type { StoredMessage } from '@maka/core/session'; import { readRuntimeHostConnectionCatalog, readRuntimeHostSessionCatalogPage, @@ -37,16 +47,21 @@ import { RuntimeHostOperationError, RuntimeHostRequestInterruptedError, RuntimeHostSessionCatalogRevisionChangedError, - type RuntimeHostConnection, + type RuntimeHostReconnectingConnection, type RuntimeHostSessionCatalogPageCursor, } from '@maka/runtime-host/client'; import { SESSION_CATALOG_CURSOR_MAX_BYTES, SESSION_CATALOG_CWD_MAX_BYTES, + HOST_OPERATION_SPECS, type SessionCatalogProjection, + type SessionContinuitySnapshot, + type TurnSnapshot, } from '@maka/runtime-host/protocol'; +import { RuntimeHostSessionChannel } from '../runtime-host-session-channel.js'; import { RuntimeHostSessionUpdateError, + getRuntimeHostSession, requireRuntimeHostSessionProjection, updateRuntimeHostSession, } from '../runtime-host-session-update.js'; @@ -56,6 +71,8 @@ import { projectAcpSessionConfigOptions, validateAcpSessionConfigOptionRequest, } from './session-configuration.js'; +import { AcpSessionEventMapper } from './session-event-mapper.js'; +import { mapAcpPromptContent } from './prompt-content.js'; const ACP_SESSION_CURSOR_MAX_BYTES = 8 * 1024; @@ -63,25 +80,89 @@ type AcpSessionRegistryOperation = | 'connection.catalog.query' | 'session.create' | 'session.catalog.query' - | 'session.configuration.update'; -type AcpSessionRegistryLifecycleOperation = 'connect' | AcpSessionRegistryOperation; + | 'session.configuration.update' + | 'subscription.open' + | 'turn.start' + | 'turn.stop'; +type AcpSessionRegistryLifecycleOperation = + | 'connect' + | 'session.close' + | AcpSessionRegistryOperation; -export interface AcpSessionRegistryConnection { - readonly request: RuntimeHostConnection['request']; +export interface AcpSessionRegistryConnection + extends Pick< + RuntimeHostReconnectingConnection, + 'request' | 'openSessionSubscription' | 'openSessionSubscriptionOnce' | 'close' + > {} + +export interface AcpSessionAttachment { + readonly snapshot: SessionContinuitySnapshot; + eventsForTurn(turnId: string): AsyncIterable; + failTurn(turnId: string, error: unknown): void; close(): Promise; } +export interface AcpSessionAttachmentOpenInput { + readonly connection: AcpSessionRegistryConnection; + readonly sessionId: string; + readonly onSnapshotChanged: (snapshot: SessionContinuitySnapshot) => void; + readonly onTranscriptReplaced: (turnId: string, messages: readonly StoredMessage[]) => void; + readonly onFailed: (error: Error) => void; +} + +export interface AcpPromptContext { + readonly signal: AbortSignal; + readonly notify: (notification: SessionNotification) => Promise; +} + export interface AcpSessionRegistryOptions { readonly connect: (signal: AbortSignal) => Promise; readonly newSessionId?: () => string; + readonly newTurnId?: () => string; + readonly openSessionAttachment?: ( + input: AcpSessionAttachmentOpenInput, + ) => Promise; +} + +interface AcpAttachmentConfiguration { + readonly notify: AcpPromptContext['notify']; + readonly retired: Promise; + readonly retire: () => void; + tail: Promise; + metadataRevision?: number; + options?: string; + delivery?: Promise; +} + +interface ActiveAcpPrompt { + readonly sessionId: string; + readonly turnId: string; + readonly mapper: AcpSessionEventMapper; + readonly waiters: Set<() => void>; + attachment?: AcpSessionAttachment; + dispatchStarted: boolean; + admissionSettled: boolean; + startedTurn?: TurnSnapshot; + cancelled: boolean; + finished: boolean; + stopTask?: Promise; } /** Owns all Runtime Host resources associated with one ACP connection. */ export class AcpSessionRegistry { readonly #connect: (signal: AbortSignal) => Promise; readonly #newSessionId: () => string; + readonly #newTurnId: () => string; + readonly #openSessionAttachment: ( + input: AcpSessionAttachmentOpenInput, + ) => Promise; readonly #inFlightOperations = new Set>(); readonly #ownedSessionIds = new Set(); + readonly #attachments = new Map>(); + readonly #attachmentConfigurations = new Map(); + readonly #pendingConfigSets = new Map>>(); + readonly #activePrompts = new Map>(); + readonly #sessionCloseTasks = new Map>(); #connection: AcpSessionRegistryConnection | undefined; #connectTask: Promise | undefined; #connectAbortController: AbortController | undefined; @@ -92,6 +173,8 @@ export class AcpSessionRegistry { constructor(options: AcpSessionRegistryOptions) { this.#connect = options.connect; this.#newSessionId = options.newSessionId ?? randomUUID; + this.#newTurnId = options.newTurnId ?? randomUUID; + this.#openSessionAttachment = options.openSessionAttachment ?? openRuntimeHostSessionAttachment; } async create(params: NewSessionRequest): Promise { @@ -120,7 +203,58 @@ export class AcpSessionRegistry { } catch (error) { throw requestErrorFromConfigInput(error); } - return this.#track(this.#setConfigOption(params)); + const configuration = this.#attachmentConfigurations.get(params.sessionId); + const operation = this.#track( + configuration + ? this.#queueConfiguration(configuration, () => + this.#setConfigOption(params, configuration), + ) + : this.#setConfigOption(params), + ); + let pending = this.#pendingConfigSets.get(params.sessionId); + if (!pending) { + pending = new Set(); + this.#pendingConfigSets.set(params.sessionId, pending); + } + pending.add(operation); + try { + return await operation; + } finally { + pending.delete(operation); + if (pending.size === 0) this.#pendingConfigSets.delete(params.sessionId); + } + } + + async prompt(params: PromptRequest, context: AcpPromptContext): Promise { + this.#assertOpen('turn.start'); + this.#assertOwned(params.sessionId); + return this.#track(this.#prompt(params, context)); + } + + async cancel(params: CancelNotification): Promise { + if (this.#closing) return; + await this.#cancelSession(params.sessionId); + } + + async close(params: CloseSessionRequest): Promise { + this.#assertOpen('session.close'); + const existing = this.#sessionCloseTasks.get(params.sessionId); + if (existing) return existing; + this.#assertOwned(params.sessionId); + this.#ownedSessionIds.delete(params.sessionId); + const configuration = this.#attachmentConfigurations.get(params.sessionId); + const delivery = configuration?.delivery; + configuration?.retire(); + this.#attachmentConfigurations.delete(params.sessionId); + const task = this.#track(this.#closeSession(params.sessionId, delivery)); + this.#sessionCloseTasks.set(params.sessionId, task); + const forget = () => { + if (this.#sessionCloseTasks.get(params.sessionId) === task) { + this.#sessionCloseTasks.delete(params.sessionId); + } + }; + void task.then(forget, forget); + return task; } dispose(): Promise { @@ -130,6 +264,380 @@ export class AcpSessionRegistry { return this.#disposeTask; } + async #prompt(params: PromptRequest, context: AcpPromptContext): Promise { + const turnId = this.#newTurnId(); + const active: ActiveAcpPrompt = { + sessionId: params.sessionId, + turnId, + mapper: new AcpSessionEventMapper({ + sessionId: params.sessionId, + notify: async (notification) => { + const configuration = this.#attachmentConfigurations.get(params.sessionId); + if (configuration) await Promise.race([configuration.tail, configuration.retired]); + if (!this.#closing && this.#ownedSessionIds.has(params.sessionId)) { + await context.notify(notification); + } + }, + }), + waiters: new Set(), + dispatchStarted: false, + admissionSettled: false, + cancelled: false, + finished: false, + }; + this.#addActivePrompt(active); + const onAbort = () => { + void this.#cancelPrompt(active).catch(() => undefined); + }; + context.signal.addEventListener('abort', onAbort, { once: true }); + if (context.signal.aborted) onAbort(); + try { + const content = await mapAcpPromptContent(params.prompt); + let startInput; + try { + startInput = HOST_OPERATION_SPECS['turn.start'].decodeInput({ + sessionId: params.sessionId, + turnId, + content, + }); + } catch { + throw RequestError.invalidParams( + { field: 'prompt', reason: 'runtime_host_admission_rejected' }, + 'Prompt cannot be admitted by Runtime Host', + ); + } + if (active.cancelled) return { stopReason: await active.mapper.cancel() }; + + const connection = await this.#getConnection('subscription.open'); + let attachment: AcpSessionAttachment; + try { + attachment = await this.#ensureAttachment(params.sessionId, connection, context.notify); + } catch (error) { + if (active.cancelled) return { stopReason: await active.mapper.cancel() }; + throw error; + } + active.attachment = attachment; + this.#wake(active); + if (active.cancelled) return { stopReason: await active.mapper.cancel() }; + + const observation = this.#consumePromptEvents(active, attachment.eventsForTurn(turnId)); + // Mark the observer as handled immediately: turn.start may still be in flight + // when the live subscription reports a failure. + void observation.catch(() => undefined); + active.dispatchStarted = true; + this.#wake(active); + try { + const result = await connection.request('turn.start', startInput); + active.admissionSettled = true; + if (result.kind === 'started') active.startedTurn = result.turn; + this.#wake(active); + if (result.kind === 'blocked') { + const error = new Error('Runtime Host blocked the requested Turn'); + attachment.failTurn(turnId, error); + throw error; + } + } catch (error) { + // A lost dispatched response does not establish whether Host admitted + // this Turn. Retain this attempt until subscription or query facts do. + active.admissionSettled = !( + error instanceof RuntimeHostRequestInterruptedError && error.dispatch === 'dispatched' + ); + if (!active.admissionSettled) { + void connection.request('turn.query', { sessionId: active.sessionId, turnId }).then( + (turn) => { + active.startedTurn = turn; + active.admissionSettled = true; + this.#wake(active); + }, + (queryError: unknown) => { + // Only an authoritative absence settles unknown admission. A + // failed query must leave cancellation latched for recovery. + if ( + queryError instanceof RuntimeHostOperationError && + queryError.code === 'not_found' + ) { + active.admissionSettled = true; + } + this.#wake(active); + }, + ); + } + this.#wake(active); + attachment.failTurn(turnId, error); + if (!active.cancelled) throw requestErrorFromRuntimeHost(error, 'turn.start'); + } + + if (active.cancelled) { + await active.stopTask?.catch(() => undefined); + return { stopReason: await active.mapper.cancel() }; + } + const stopReason = await observation; + const configuration = this.#attachmentConfigurations.get(params.sessionId); + if (configuration) await Promise.race([configuration.tail, configuration.retired]); + return { stopReason }; + } catch (error) { + // A failed projection must not leave the corresponding Host Turn running. + active.stopTask ??= this.#stopPromptWhenObservable(active); + await active.stopTask.catch(() => undefined); + if (active.cancelled) return { stopReason: await active.mapper.cancel() }; + throw error; + } finally { + context.signal.removeEventListener('abort', onAbort); + active.finished = true; + this.#wake(active); + this.#removeActivePrompt(active); + } + } + + async #consumePromptEvents( + active: ActiveAcpPrompt, + events: AsyncIterable, + ): Promise { + try { + for await (const event of events) { + const terminal = await active.mapper.accept(event); + if (terminal) return terminal; + } + if (active.cancelled) return active.mapper.cancel(); + throw new Error('Runtime Host Turn observation ended without a terminal event'); + } catch (error) { + if (active.cancelled) return active.mapper.cancel(); + throw error; + } + } + + #cancelSession(sessionId: string): Promise[]> { + const active = [...(this.#activePrompts.get(sessionId) ?? [])]; + const cancellations = active.map((prompt) => this.#cancelPrompt(prompt)); + const attachment = this.#attachments.get(sessionId); + if (attachment) { + cancellations.push( + attachment.then( + async (opened) => { + const root = opened.snapshot.rootTurn; + // Local prompts already latch cancellation across pending turn.start. + // An idle attachment may also observe a Turn started by another client. + if ( + root && + !isRuntimeHostTerminalTurn(root) && + !active.some((prompt) => prompt.turnId === root.turnId) + ) { + await this.#connection?.request('turn.stop', { + sessionId: root.sessionId, + turnId: root.turnId, + runId: root.runId, + }); + } + }, + () => undefined, + ), + ); + } + return Promise.allSettled(cancellations); + } + + async #cancelPrompt(active: ActiveAcpPrompt): Promise { + active.cancelled = true; + active.stopTask ??= this.#stopPromptWhenObservable(active); + await Promise.all([ + active.mapper.cancel(), + active.stopTask.catch((error: unknown) => { + // End only this prompt's observation. Failed delivery does not establish + // a terminal Host Turn, and teardown still receives the original error. + active.attachment?.failTurn(active.turnId, error); + throw error; + }), + ]); + } + + async #stopPromptWhenObservable(active: ActiveAcpPrompt): Promise { + if (!active.dispatchStarted) return; + while (!active.finished) { + const observed = active.attachment?.snapshot.rootTurn; + // Subscription teardown can precede the start response. Keep the admitted + // identity until exact Stop completes, even when observation has ended. + const root = observed?.turnId === active.turnId ? observed : active.startedTurn; + if (root) { + if (isRuntimeHostTerminalTurn(root)) return; + const connection = this.#connection; + if (!connection) return; + try { + await connection.request('turn.stop', { + sessionId: root.sessionId, + turnId: root.turnId, + runId: root.runId, + }); + } catch (error) { + console.error('[acp] Host Stop delivery failed:', error); + throw error; + } + return; + } + if (active.admissionSettled) return; + await this.#waitForPromptChange(active); + } + } + + async #ensureAttachment( + sessionId: string, + connection: AcpSessionRegistryConnection, + notify: AcpPromptContext['notify'], + ): Promise { + const existing = this.#attachments.get(sessionId); + if (existing) return existing; + let retire!: () => void; + const retired = new Promise((resolve) => { + retire = resolve; + }); + const configuration: AcpAttachmentConfiguration = { + notify, + // Setters can outlive an absent or failed attachment. Their responses + // must precede refreshes delivered by the new attachment's queue. + tail: Promise.allSettled([...(this.#pendingConfigSets.get(sessionId) ?? [])]), + retired, + retire, + }; + this.#attachmentConfigurations.set(sessionId, configuration); + let task!: Promise; + let attachment: AcpSessionAttachment | undefined; + let earlyFailure: Error | undefined; + task = this.#openSessionAttachment({ + connection, + sessionId, + onSnapshotChanged: (snapshot) => { + this.#wakeSession(sessionId); + if (configuration.metadataRevision === undefined) { + configuration.metadataRevision = snapshot.session.metadataRevision; + return; + } + if (configuration.metadataRevision === snapshot.session.metadataRevision) return; + configuration.metadataRevision = snapshot.session.metadataRevision; + void this.#queueConfiguration(configuration, async () => { + if (!this.#configurationIsLive(sessionId, configuration)) return; + const session = await getRuntimeHostSession(connection, sessionId); + if (!session) throw unknownSessionError(); + const configOptions = await this.#projectConfigOptions(connection, session); + await this.#notifyConfiguration(sessionId, configuration, configOptions); + }).catch((error: unknown) => { + const failure = error instanceof Error ? error : new Error(String(error)); + if (attachment) this.#retireFailedAttachment(sessionId, task, attachment, failure); + else earlyFailure = failure; + }); + }, + onTranscriptReplaced: (turnId, messages) => { + for (const active of this.#activePrompts.get(sessionId) ?? []) { + if (active.turnId === turnId) { + void active.mapper.replaceTranscript(turnId, messages).catch((error: unknown) => { + active.attachment?.failTurn(turnId, error); + }); + } + } + }, + onFailed: (error) => { + if (!attachment) { + earlyFailure = error; + return; + } + this.#retireFailedAttachment(sessionId, task, attachment, error); + }, + }) + .then((opened) => { + attachment = opened; + if (earlyFailure) { + this.#retireFailedAttachment(sessionId, task, opened, earlyFailure); + throw earlyFailure; + } + if (this.#closing || !this.#ownedSessionIds.has(sessionId)) { + return opened.close().then(() => { + throw this.#closing ? registryClosedError('subscription.open') : unknownSessionError(); + }); + } + return opened; + }) + .catch((error: unknown) => { + if (this.#attachments.get(sessionId) === task) { + this.#attachments.delete(sessionId); + configuration.retire(); + this.#attachmentConfigurations.delete(sessionId); + } + if (error instanceof RequestError) throw error; + throw requestErrorFromRuntimeHost(error, 'subscription.open'); + }); + this.#attachments.set(sessionId, task); + return task; + } + + #retireFailedAttachment( + sessionId: string, + task: Promise, + attachment: AcpSessionAttachment, + error: Error, + ): void { + if (this.#attachments.get(sessionId) === task) { + this.#attachments.delete(sessionId); + this.#attachmentConfigurations.get(sessionId)?.retire(); + this.#attachmentConfigurations.delete(sessionId); + } + for (const active of this.#activePrompts.get(sessionId) ?? []) { + if (active.attachment !== attachment) continue; + attachment.failTurn(active.turnId, error); + this.#wake(active); + } + void attachment.close().catch(() => undefined); + } + + async #closeSession(sessionId: string, delivery?: Promise): Promise { + const cancellation = await this.#cancelSession(sessionId); + const attachmentTask = this.#attachments.get(sessionId); + this.#attachments.delete(sessionId); + let closeError: unknown; + if (attachmentTask) { + try { + // A rejected open has no retained resource; close still releases ownership. + const attachment = await attachmentTask.catch(() => undefined); + await attachment?.close(); + } catch (error) { + closeError = error; + } + } + await delivery; + const failedCancellation = cancellation.find( + (result): result is PromiseRejectedResult => result.status === 'rejected', + ); + if (failedCancellation) throw failedCancellation.reason; + if (closeError) throw closeError; + return {}; + } + + #addActivePrompt(active: ActiveAcpPrompt): void { + const prompts = this.#activePrompts.get(active.sessionId); + if (prompts) prompts.add(active); + else this.#activePrompts.set(active.sessionId, new Set([active])); + } + + #removeActivePrompt(active: ActiveAcpPrompt): void { + const prompts = this.#activePrompts.get(active.sessionId); + prompts?.delete(active); + if (prompts?.size === 0) this.#activePrompts.delete(active.sessionId); + } + + #wakeSession(sessionId: string): void { + for (const active of this.#activePrompts.get(sessionId) ?? []) this.#wake(active); + } + + #wake(active: ActiveAcpPrompt): void { + for (const resolve of active.waiters) resolve(); + active.waiters.clear(); + } + + #waitForPromptChange(active: ActiveAcpPrompt): Promise { + return new Promise((resolve) => active.waiters.add(resolve)); + } + + #assertOwned(sessionId: string): void { + if (!this.#ownedSessionIds.has(sessionId)) throw unknownSessionError(); + } + async #create(params: NewSessionRequest): Promise { const connection = await this.#getConnection('session.create'); const sessionId = this.#newSessionId(); @@ -143,19 +651,24 @@ export class AcpSessionRegistry { } catch (error) { throw requestErrorFromRuntimeHost(error, 'session.create', { sessionId }); } - let created: SessionCatalogProjection; + // Session creation has committed. Optional presentation failures must not + // turn that success into an unreachable durable Session. + let configOptions: SessionConfigOption[] | undefined; try { - created = requireRuntimeHostSessionProjection(result, 'session.create'); - } catch (error) { - throw requestErrorFromSessionUpdate(error, 'session.create', { sessionId }); + const created = requireRuntimeHostSessionProjection(result, 'session.create'); + configOptions = await this.#projectConfigOptions(connection, created); + } catch { + // The client can still prompt, configure, list, or close the returned ID. } - const configOptions = await this.#projectConfigOptions(connection, created); - this.#ownedSessionIds.add(sessionId); - return { sessionId, configOptions }; + // Do not admit mutations while projection is pending, or resurrect ownership + // if connection shutdown raced the successful Host creation. + if (!this.#closing) this.#ownedSessionIds.add(sessionId); + return { sessionId, ...(configOptions ? { configOptions } : {}) }; } async #setConfigOption( params: SetSessionConfigOptionRequest & { readonly value: string }, + configuration?: AcpAttachmentConfiguration, ): Promise { const connection = await this.#getConnection('session.configuration.update'); let committed: SessionCatalogProjection; @@ -171,13 +684,56 @@ export class AcpSessionRegistry { }), { operation: 'session.configuration.update', - assertRequestAllowed: () => this.#assertOpen('session.configuration.update'), + assertRequestAllowed: () => { + this.#assertOpen('session.configuration.update'); + this.#assertOwned(params.sessionId); + }, }, ); } catch (error) { throw requestErrorFromSessionUpdate(error, 'session.configuration.update'); } - return { configOptions: await this.#projectConfigOptions(connection, committed) }; + const configOptions = await this.#projectConfigOptions(connection, committed); + if (configuration) + await this.#notifyConfiguration(params.sessionId, configuration, configOptions); + return { configOptions }; + } + + #configurationIsLive(sessionId: string, configuration: AcpAttachmentConfiguration): boolean { + return ( + !this.#closing && + this.#ownedSessionIds.has(sessionId) && + this.#attachmentConfigurations.get(sessionId) === configuration + ); + } + + #queueConfiguration( + configuration: AcpAttachmentConfiguration, + operation: () => Promise, + ): Promise { + // Serialize asynchronous catalog projection and delivery, not Host frames: + // session-channel/projector remain the only subscription ordering authority. + // A local set emits its committed options before its response; subscription + // refreshes observed during that set follow its notification in this queue. + const result = configuration.tail.then(operation, operation); + configuration.tail = result.catch(() => undefined); + return result; + } + + async #notifyConfiguration( + sessionId: string, + configuration: AcpAttachmentConfiguration, + configOptions: SessionConfigOption[], + ): Promise { + if (!this.#configurationIsLive(sessionId, configuration)) return; + const options = JSON.stringify(configOptions); + if (configuration.options === options) return; + configuration.delivery = configuration.notify({ + sessionId, + update: { sessionUpdate: 'config_option_update', configOptions }, + }); + await configuration.delivery; + configuration.options = options; } async #projectConfigOptions( @@ -245,9 +801,20 @@ export class AcpSessionRegistry { } async #dispose(): Promise { - const connectionClose = this.#closeOwnedConnection(); - await Promise.allSettled([connectionClose]); - await Promise.allSettled([...this.#inFlightOperations]); + const sessionIds = new Set([...this.#activePrompts.keys(), ...this.#attachments.keys()]); + const cancellations = [...sessionIds].map((sessionId) => this.#cancelSession(sessionId)); + const attachments = [...this.#attachments.values()]; + this.#attachments.clear(); + const configurations = [...this.#attachmentConfigurations.values()]; + for (const configuration of configurations) configuration.retire(); + this.#attachmentConfigurations.clear(); + await Promise.allSettled(attachments.map(async (attachment) => (await attachment).close())); + await Promise.allSettled(cancellations); + await Promise.allSettled([this.#closeOwnedConnection()]); + await Promise.allSettled([ + ...this.#inFlightOperations, + ...configurations.map(({ tail }) => tail), + ]); this.#ownedSessionIds.clear(); } @@ -318,12 +885,44 @@ export class AcpSessionRegistry { } } - #assertOpen(operation: AcpSessionRegistryOperation): void { + #assertOpen(operation: AcpSessionRegistryLifecycleOperation): void { if (!this.#closing) return; throw registryClosedError(operation); } } +async function openRuntimeHostSessionAttachment( + input: AcpSessionAttachmentOpenInput, +): Promise { + const opened = await RuntimeHostSessionChannel.open({ + connection: input.connection, + openInitialSessionSubscription: input.connection.openSessionSubscriptionOnce.bind( + input.connection, + ), + sessionId: input.sessionId, + now: Date.now, + onTurnStarted: () => undefined, + onRuntimeResourceChanged: () => undefined, + onInteractionPending: () => undefined, + onInteractionResolved: () => undefined, + onTranscriptSettlement: () => undefined, + onTranscriptReplaced: input.onTranscriptReplaced, + onGoalChanged: () => undefined, + onSnapshotChanged: input.onSnapshotChanged, + onFailed: input.onFailed, + onRecovered: () => undefined, + }); + opened.channel.activate(); + return opened.channel; +} + +function unknownSessionError(): RequestError { + return RequestError.invalidParams( + { reason: 'unknown_session' }, + 'Session is not owned by this ACP connection', + ); +} + function registryClosedError(operation: AcpSessionRegistryLifecycleOperation): RequestError { return RequestError.internalError( { source: 'runtime_host', operation, code: 'registry_closed' }, diff --git a/packages/cli/src/acp/stdio-server.ts b/packages/cli/src/acp/stdio-server.ts index 27e2a1d985..e0c3988ac7 100644 --- a/packages/cli/src/acp/stdio-server.ts +++ b/packages/cli/src/acp/stdio-server.ts @@ -19,7 +19,10 @@ import { Readable, Writable } from 'node:stream'; import { ndJsonStream } from '@agentclientprotocol/sdk'; -import type { RuntimeHostConnection } from '@maka/runtime-host/client'; +import { + isRuntimeHostReconnectingConnection, + 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'; @@ -49,10 +52,15 @@ export async function runMakaAcpStdioServer( clientDataRoot: input.clientDataRoot, signal, }); + const connection = context.connection; + if (!isRuntimeHostReconnectingConnection(connection)) { + await context.close().catch(() => undefined); + throw new Error('ACP requires a reconnecting Runtime Host connection'); + } return { - request: context.connection.request.bind( - context.connection, - ) as RuntimeHostConnection['request'], + request: connection.request.bind(connection) as RuntimeHostConnection['request'], + openSessionSubscription: connection.openSessionSubscription.bind(connection), + openSessionSubscriptionOnce: connection.openSessionSubscriptionOnce.bind(connection), close: () => context.close(), }; }, diff --git a/packages/cli/src/cli-core.ts b/packages/cli/src/cli-core.ts index b625c4869b..5c3bbf875c 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, session/new, session/list)`, + ` ${cliCommand} --acp Serve ACP v1 over stdio (sessions, prompts, streaming, cancellation)`, ` ${cliCommand} run ... Run one non-interactive model turn`, ` ${cliCommand} activate ... Run one Cloud Session activation and emit JSONL`, ` ${cliCommand} -p ... Alias for ${cliCommand} run`, diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 03b57b0dd0..573cde5137 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -68,6 +68,8 @@ export interface RuntimeHostSessionChannelOpenResult { export interface RuntimeHostSessionChannelOptions { connection: Pick; + /** Optional opener pinned to the concrete Host connection used for first attachment. */ + openInitialSessionSubscription?: RuntimeHostConnection['openSessionSubscription']; sessionId: string; now: () => number; onTurnStarted: (turn: MakaPreparedSessionTurn) => void; @@ -163,7 +165,10 @@ export class RuntimeHostSessionChannel { static async open( options: RuntimeHostSessionChannelOptions, ): Promise { - const subscription = await options.connection.openSessionSubscription({ + const openInitial = + options.openInitialSessionSubscription ?? + options.connection.openSessionSubscription.bind(options.connection); + const subscription = await openInitial({ sessionId: options.sessionId, transcript: { kind: 'tail', diff --git a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts index 7aed4c5582..337cd01a71 100644 --- a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts +++ b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts @@ -264,6 +264,47 @@ test('a Session observation reopens safely after its first connection starts dra await connection.close(); }); +test('an initial Session observation stays pinned to the concrete connection it started on', async () => { + const first = connectionHarness( + 'first', + () => undefined, + async () => { + first.disconnect(); + throw new RuntimeHostOperationError( + 'subscription.open', + 'host_draining', + 'Runtime Host is draining', + ); + }, + ); + const replacement = connectionHarness( + 'replacement', + () => undefined, + async () => ({ subscriptionId: 'replacement-subscription' }), + ); + const reconnected = deferred(); + const connection = await createRuntimeHostReconnectingConnection({ + initialConnection: first.connection, + connect: async () => { + reconnected.resolve(); + return replacement.connection; + }, + }); + + await assert.rejects( + connection.openSessionSubscriptionOnce({ + sessionId: 'session-1', + transcript: { kind: 'none' }, + }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'host_draining', + ); + await reconnected.promise; + assert.equal(first.openedSubscriptions, 1); + assert.equal(replacement.openedSubscriptions, 0); + await connection.close(); +}); + test('a reconnecting Client rejects a different Host composition permanently', async () => { const first = connectionHarness('first', () => undefined); const replacement = connectionHarness('replacement', () => undefined, undefined, { diff --git a/packages/runtime-host/src/client/reconnecting-connection.ts b/packages/runtime-host/src/client/reconnecting-connection.ts index 46f1aa14cb..27e579d13e 100644 --- a/packages/runtime-host/src/client/reconnecting-connection.ts +++ b/packages/runtime-host/src/client/reconnecting-connection.ts @@ -46,6 +46,10 @@ import type { RuntimeHostSessionSubscription } from './session-subscription.js'; export interface RuntimeHostReconnectingConnection extends RuntimeHostConnection { readonly reconnecting: true; + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise; subscribeConnectionAvailability( listener: (availability: RuntimeHostConnectionAvailability) => void, ): () => void; @@ -212,6 +216,13 @@ class RuntimeHostReconnectingConnectionImpl implements RuntimeHostReconnectingCo } } + openSessionSubscriptionOnce( + input: SubscriptionOpenInput, + timeoutMs?: number, + ): Promise { + return this.#requireCurrent('subscription.open').openSessionSubscription(input, timeoutMs); + } + async replaceClientCapabilities( provider: ClientCapabilityProvider, timeoutMs?: number,