You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by zh...@apache.org on 2026/08/31 17:51:13 UTC
(maka) 03/03: feat(runtime-host): broker nested capability forms
This is an automated email from the ASF dual-hosted git repository.
me2seeks pushed a commit to branch feat/4364-client-capability-form-interactions
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 36ec70c6dc2a36b1b87ce11a3bad6aef98eea57b
Author: me2seeks <me...@gmail.com>
AuthorDate: Tue Sep 1 01:44:09 2026 +0800
feat(runtime-host): broker nested capability forms
Route Client Capability interaction requests through the Runtime-owned form callback. Pause provider execution time only while the canonical form is pending, bound result delivery, and rearm a fresh execution timeout after delivery.\n\nClose the exact producer-owned form before settling provider failure, cancellation, or connection loss, while preserving Runtime Host as the only Interaction authority.\n\nPart of #4364.\n\nGenerated-by: OpenAI Codex
---
.../runtime-host-desktop-candidate.test.ts | 1 +
.../runtime-host-native-capabilities.test.ts | 2 +
...untime-host-capability-provider-command.test.ts | 1 +
.../client-capability-invocation-broker.test.ts | 372 ++++++++++++++++++++
.../src/server/client-capability-coordinator.ts | 7 +-
.../server/client-capability-invocation-broker.ts | 387 +++++++++++++++++----
packages/runtime/src/__tests__/mcp-tools.test.ts | 38 ++
packages/runtime/src/mcp-tools.ts | 13 +
8 files changed, 753 insertions(+), 68 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts
index 26222c86ad..b0cd16f52d 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts
@@ -1207,6 +1207,7 @@ function connectionHarness(
return provider.call(frame, {
signal: new AbortController().signal,
accept: async () => undefined,
+ requestInteraction: async () => assert.fail('Unexpected provider interaction'),
});
},
disconnect: () => resolveClosed?.(),
diff --git a/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts b/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts
index 07fb18e7c1..916199a755 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts
@@ -528,6 +528,7 @@ test('forwards Host cancellation to an admitted Desktop invocation', async () =>
const inFlight = provider.call(capabilityFrame(), {
signal: controller.signal,
accept: async () => undefined,
+ requestInteraction: async () => assert.fail('Unexpected provider interaction'),
});
await started;
@@ -649,5 +650,6 @@ async function call(
return provider.call(frame, {
signal: new AbortController().signal,
accept: async () => accept(),
+ requestInteraction: async () => assert.fail('Unexpected provider interaction'),
});
}
diff --git a/packages/cli/src/__tests__/runtime-host-capability-provider-command.test.ts b/packages/cli/src/__tests__/runtime-host-capability-provider-command.test.ts
index c059040227..8eb4338adb 100644
--- a/packages/cli/src/__tests__/runtime-host-capability-provider-command.test.ts
+++ b/packages/cli/src/__tests__/runtime-host-capability-provider-command.test.ts
@@ -87,6 +87,7 @@ test('MCP capability publication freezes an accepted callable tool snapshot', as
accept: async () => {
accepted = true;
},
+ requestInteraction: async () => assert.fail('Unexpected provider interaction'),
},
);
assert.deepEqual(result, { content: [{ type: 'text', text: '{"path":"README.md"}' }] });
diff --git a/packages/runtime-host/src/__tests__/client-capability-invocation-broker.test.ts b/packages/runtime-host/src/__tests__/client-capability-invocation-broker.test.ts
new file mode 100644
index 0000000000..16f3804c94
--- /dev/null
+++ b/packages/runtime-host/src/__tests__/client-capability-invocation-broker.test.ts
@@ -0,0 +1,372 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { ToolOutcomeUnknownError } from '@maka/core/events';
+import type { ClientCapabilityHostFrame } from '../protocol/index.js';
+import {
+ ClientCapabilityInvocationBroker,
+ ClientCapabilityInvocationError,
+ type ClientCapabilityInvocationBinding,
+ type ClientCapabilityInvocationRegistration,
+} from '../server/client-capability-invocation-broker.js';
+
+const REGISTRATION: ClientCapabilityInvocationRegistration = {
+ connectionId: 'connection-a',
+ registrationId: 'registration-a',
+};
+
+const BINDING: ClientCapabilityInvocationBinding = {
+ offerId: 'offer-a',
+ hostPathAccess: 'none',
+ descriptor: {
+ serverId: 'fixture',
+ name: 'deploy',
+ inputSchema: { type: 'object' },
+ },
+};
+
+const CONTEXT = {
+ sessionId: 'session-a',
+ turnId: 'turn-a',
+ toolCallId: 'tool-call-a',
+ cwd: '/tmp',
+};
+
+test('Client Capability nested interaction pauses and rearms the execution timeout', async () => {
+ const sent: ClientCapabilityHostFrame[] = [];
+ const timers = createTimerHarness();
+ let answer!: (value: { action: 'accept'; values: { target: string } }) => void;
+ const broker = new ClientCapabilityInvocationBroker({
+ senderFor: () => ({ send: async (frame) => void sent.push(frame) }),
+ onRegistrationIdle: () => {},
+ scheduleTimeout: timers.schedule,
+ });
+ const result = broker.invoke(
+ REGISTRATION,
+ BINDING,
+ {},
+ CONTEXT,
+ undefined,
+ 1_000,
+ undefined,
+ async (_form, options) => {
+ assert.equal(options?.cancellationSignal?.aborted, false);
+ return new Promise((resolve) => {
+ answer = resolve;
+ });
+ },
+ );
+ await flush();
+ const invocationId = callInvocationId(sent);
+ assert.equal(timers.activeCount(), 1);
+
+ broker.accept('connection-a', { kind: 'client.capability.accepted', invocationId });
+ broker.accept('connection-a', {
+ kind: 'client.capability.interaction_request',
+ invocationId,
+ interactionId: 'interaction-a',
+ request: {
+ message: 'Choose a target',
+ requester: { name: 'deploy' },
+ fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }],
+ },
+ });
+ assert.equal(timers.activeCount(), 0);
+
+ answer({ action: 'accept', values: { target: 'staging' } });
+ await flush();
+ assert.deepEqual(sent.at(-1), {
+ kind: 'client.capability.interaction_result',
+ invocationId,
+ interactionId: 'interaction-a',
+ result: { action: 'accept', values: { target: 'staging' } },
+ });
+ assert.equal(timers.activeCount(), 1);
+
+ timers.fireActive();
+ await assert.rejects(result, (error: unknown) => error instanceof ToolOutcomeUnknownError);
+ assert.equal(timers.activeCount(), 0);
+ broker.close();
+});
+
+test('Client Capability accepts a final result while interaction delivery is still flushing', async () => {
+ const sent: ClientCapabilityHostFrame[] = [];
+ let broker!: ClientCapabilityInvocationBroker<ClientCapabilityInvocationRegistration>;
+ const brokerOptions = {
+ senderFor: () => ({
+ send: async (frame: ClientCapabilityHostFrame) => {
+ sent.push(frame);
+ if (frame.kind !== 'client.capability.interaction_result') return;
+ broker.accept('connection-a', {
+ kind: 'client.capability.result',
+ invocationId: frame.invocationId,
+ result: { content: [{ type: 'text', text: 'deployed' }] },
+ });
+ await flush();
+ },
+ }),
+ onRegistrationIdle: () => {},
+ };
+ broker = new ClientCapabilityInvocationBroker(brokerOptions);
+ const result = broker.invoke(
+ REGISTRATION,
+ BINDING,
+ {},
+ CONTEXT,
+ undefined,
+ 1_000,
+ undefined,
+ async () => ({ action: 'accept', values: { target: 'staging' } }),
+ );
+ await flush();
+ const invocationId = callInvocationId(sent);
+ broker.accept('connection-a', { kind: 'client.capability.accepted', invocationId });
+ broker.accept('connection-a', {
+ kind: 'client.capability.interaction_request',
+ invocationId,
+ interactionId: 'interaction-a',
+ request: {
+ message: 'Choose a target',
+ requester: { name: 'deploy' },
+ fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }],
+ },
+ });
+
+ assert.deepEqual(await result, { content: [{ type: 'text', text: 'deployed' }] });
+ broker.close();
+});
+
+test('Client Capability provider failure waits for pending interaction withdrawal', async () => {
+ const sent: ClientCapabilityHostFrame[] = [];
+ let finishWithdrawal!: () => void;
+ const withdrawal = new Promise<void>((resolve) => {
+ finishWithdrawal = resolve;
+ });
+ let producerCancelled = false;
+ const broker = new ClientCapabilityInvocationBroker({
+ senderFor: () => ({ send: async (frame) => void sent.push(frame) }),
+ onRegistrationIdle: () => {},
+ });
+ const result = broker.invoke(
+ REGISTRATION,
+ BINDING,
+ {},
+ CONTEXT,
+ undefined,
+ 1_000,
+ undefined,
+ async (_form, options) => {
+ const signal = options?.cancellationSignal;
+ assert.ok(signal);
+ await new Promise<void>((resolve) =>
+ signal.addEventListener(
+ 'abort',
+ () => {
+ producerCancelled = true;
+ resolve();
+ },
+ { once: true },
+ ),
+ );
+ await withdrawal;
+ throw signal.reason;
+ },
+ );
+ await flush();
+ const invocationId = callInvocationId(sent);
+ broker.accept('connection-a', { kind: 'client.capability.accepted', invocationId });
+ broker.accept('connection-a', {
+ kind: 'client.capability.interaction_request',
+ invocationId,
+ interactionId: 'interaction-a',
+ request: {
+ message: 'Confirm',
+ requester: { name: 'deploy' },
+ fields: [{ kind: 'boolean', name: 'confirm', label: 'Confirm', required: true }],
+ },
+ });
+ broker.accept('connection-a', {
+ kind: 'client.capability.failed',
+ invocationId,
+ message: 'provider stopped',
+ });
+ await flush();
+ assert.equal(producerCancelled, true);
+ assert.equal(
+ sent.some((frame) => frame.kind === 'client.capability.release'),
+ false,
+ );
+
+ finishWithdrawal();
+ await assert.rejects(
+ result,
+ (error: unknown) =>
+ error instanceof ClientCapabilityInvocationError &&
+ error.code === 'provider_failed' &&
+ error.message === 'provider stopped',
+ );
+ assert.equal(sent.at(-1)?.kind, 'client.capability.release');
+ broker.close();
+});
+
+test('Client Capability connection release waits for pending interaction withdrawal', async () => {
+ const sent: ClientCapabilityHostFrame[] = [];
+ let finishWithdrawal!: () => void;
+ const withdrawal = new Promise<void>((resolve) => {
+ finishWithdrawal = resolve;
+ });
+ const broker = new ClientCapabilityInvocationBroker({
+ senderFor: () => ({ send: async (frame) => void sent.push(frame) }),
+ onRegistrationIdle: () => {},
+ });
+ const result = broker.invoke(
+ REGISTRATION,
+ BINDING,
+ {},
+ CONTEXT,
+ undefined,
+ 1_000,
+ undefined,
+ async (_form, options) => {
+ const signal = options?.cancellationSignal;
+ assert.ok(signal);
+ await new Promise<void>((resolve) =>
+ signal.addEventListener('abort', () => resolve(), { once: true }),
+ );
+ await withdrawal;
+ throw signal.reason;
+ },
+ );
+ await flush();
+ const invocationId = callInvocationId(sent);
+ broker.accept('connection-a', { kind: 'client.capability.accepted', invocationId });
+ broker.accept('connection-a', {
+ kind: 'client.capability.interaction_request',
+ invocationId,
+ interactionId: 'interaction-a',
+ request: {
+ message: 'Confirm',
+ requester: { name: 'deploy' },
+ fields: [{ kind: 'boolean', name: 'confirm', label: 'Confirm', required: true }],
+ },
+ });
+ let released = false;
+ const release = broker.releaseConnection('connection-a').then(() => {
+ released = true;
+ });
+ await flush();
+ assert.equal(released, false);
+
+ finishWithdrawal();
+ await release;
+ await assert.rejects(result, /disconnected after accepting/);
+ broker.close();
+});
+
+test('Client Capability cancellation settles only after the nested interaction closes', async () => {
+ const sent: ClientCapabilityHostFrame[] = [];
+ const invocationController = new AbortController();
+ let finishWithdrawal!: () => void;
+ const withdrawal = new Promise<void>((resolve) => {
+ finishWithdrawal = resolve;
+ });
+ const broker = new ClientCapabilityInvocationBroker({
+ senderFor: () => ({ send: async (frame) => void sent.push(frame) }),
+ onRegistrationIdle: () => {},
+ });
+ const result = broker.invoke(
+ REGISTRATION,
+ BINDING,
+ {},
+ CONTEXT,
+ invocationController.signal,
+ 1_000,
+ undefined,
+ async (_form, options) => {
+ const signal = options?.cancellationSignal;
+ assert.ok(signal);
+ await new Promise<void>((resolve) =>
+ signal.addEventListener('abort', () => resolve(), { once: true }),
+ );
+ await withdrawal;
+ throw signal.reason;
+ },
+ );
+ await flush();
+ const invocationId = callInvocationId(sent);
+ broker.accept('connection-a', { kind: 'client.capability.accepted', invocationId });
+ broker.accept('connection-a', {
+ kind: 'client.capability.interaction_request',
+ invocationId,
+ interactionId: 'interaction-a',
+ request: {
+ message: 'Confirm',
+ requester: { name: 'deploy' },
+ fields: [{ kind: 'boolean', name: 'confirm', label: 'Confirm', required: true }],
+ },
+ });
+ invocationController.abort(new Error('stop'));
+ await flush();
+ assert.equal(
+ sent.some((frame) => frame.kind === 'client.capability.cancel'),
+ true,
+ );
+ assert.equal(
+ sent.some((frame) => frame.kind === 'client.capability.release'),
+ false,
+ );
+
+ finishWithdrawal();
+ await assert.rejects(result, (error: unknown) => error instanceof ToolOutcomeUnknownError);
+ assert.equal(sent.at(-1)?.kind, 'client.capability.release');
+ broker.close();
+});
+
+function callInvocationId(frames: readonly ClientCapabilityHostFrame[]): string {
+ const call = frames.find((frame) => frame.kind === 'client.capability.call');
+ assert.ok(call && call.kind === 'client.capability.call');
+ return call.invocationId;
+}
+
+function createTimerHarness(): {
+ readonly schedule: (callback: () => void) => () => void;
+ activeCount(): number;
+ fireActive(): void;
+} {
+ const active = new Set<() => void>();
+ return {
+ schedule: (callback) => {
+ active.add(callback);
+ return () => active.delete(callback);
+ },
+ activeCount: () => active.size,
+ fireActive: () => {
+ const callback = active.values().next().value;
+ assert.ok(callback);
+ active.delete(callback);
+ callback();
+ },
+ };
+}
+
+async function flush(): Promise<void> {
+ await new Promise((resolve) => setImmediate(resolve));
+}
diff --git a/packages/runtime-host/src/server/client-capability-coordinator.ts b/packages/runtime-host/src/server/client-capability-coordinator.ts
index f95b8d8c64..67b6665a7d 100644
--- a/packages/runtime-host/src/server/client-capability-coordinator.ts
+++ b/packages/runtime-host/src/server/client-capability-coordinator.ts
@@ -748,16 +748,16 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService
}
releaseConnection(connectionId: string): Promise<void> {
- this.#invocations.releaseConnection(connectionId);
+ const invocationCleanup = this.#invocations.releaseConnection(connectionId);
const connection = this.#connections.get(connectionId);
- if (!connection) return Promise.resolve();
+ if (!connection) return invocationCleanup;
let task!: Promise<void>;
task = this.#activation
.runMutation(() => this.#releaseConnectionState(connection))
.finally(() => this.#pendingConnectionReleases.delete(task));
this.#pendingConnectionReleases.add(task);
void task.catch(() => undefined);
- return task;
+ return Promise.all([invocationCleanup, task]).then(() => undefined);
}
#releaseConnectionState(connection: ClientProviderConnection): void {
@@ -1014,6 +1014,7 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService
options.signal,
options.timeoutMs ?? DEFAULT_CALL_TIMEOUT_MS,
options.emitProgress,
+ options.requestInteraction,
);
},
};
diff --git a/packages/runtime-host/src/server/client-capability-invocation-broker.ts b/packages/runtime-host/src/server/client-capability-invocation-broker.ts
index c77bc98bf3..6eff5b5d1d 100644
--- a/packages/runtime-host/src/server/client-capability-invocation-broker.ts
+++ b/packages/runtime-host/src/server/client-capability-invocation-broker.ts
@@ -19,6 +19,7 @@
import { randomUUID } from 'node:crypto';
import { ToolOutcomeUnknownError } from '@maka/core/events';
+import type { InteractionFormInput, InteractionFormResult } from '@maka/core/interaction';
import {
CLIENT_CAPABILITY_MAX_RESULT_BYTES,
CLIENT_CAPABILITY_RESULT_CHUNK_MAX_BYTES,
@@ -78,8 +79,16 @@ interface InvocationState<Registration extends ClientCapabilityInvocationRegistr
readonly signal?: AbortSignal;
readonly onAbort?: () => void;
readonly onProgress?: (current: number, total: number) => void;
- readonly timer: NodeJS.Timeout;
- phase: 'dispatched' | 'accepted' | 'chunks';
+ readonly requestInteraction?: ClientCapabilityInteractionHandler;
+ readonly timeoutMs: number;
+ cancelTimer?: () => void;
+ interaction?: InvocationInteraction;
+ phase:
+ | 'dispatched'
+ | 'accepted'
+ | 'awaiting_interaction'
+ | 'delivering_interaction_result'
+ | 'chunks';
progress?: { current: number; total: number };
chunks?: {
readonly byteLength: number;
@@ -89,11 +98,25 @@ interface InvocationState<Registration extends ClientCapabilityInvocationRegistr
};
}
+interface InvocationInteraction {
+ readonly interactionId: string;
+ readonly controller: AbortController;
+ readonly done: Promise<void>;
+ readonly resolveDone: () => void;
+ terminal?: { readonly error: Error; readonly releaseRemote: boolean };
+}
+
+type ClientCapabilityInteractionHandler = (
+ form: InteractionFormInput,
+ options?: { readonly cancellationSignal?: AbortSignal },
+) => Promise<InteractionFormResult>;
+
export interface ClientCapabilityInvocationBrokerOptions<
Registration extends ClientCapabilityInvocationRegistration,
> {
readonly senderFor: (connectionId: string) => ClientCapabilityConnectionSender | undefined;
readonly onRegistrationIdle: (registration: Registration) => void;
+ readonly scheduleTimeout?: (callback: () => void, timeoutMs: number) => () => void;
}
export class ClientCapabilityInvocationBroker<
@@ -101,12 +124,21 @@ export class ClientCapabilityInvocationBroker<
> {
readonly #senderFor: ClientCapabilityInvocationBrokerOptions<Registration>['senderFor'];
readonly #onRegistrationIdle: ClientCapabilityInvocationBrokerOptions<Registration>['onRegistrationIdle'];
+ readonly #scheduleTimeout: NonNullable<
+ ClientCapabilityInvocationBrokerOptions<Registration>['scheduleTimeout']
+ >;
readonly #invocations = new Map<string, InvocationState<Registration>>();
readonly #retiredInvocationIds = new Set<string>();
constructor(options: ClientCapabilityInvocationBrokerOptions<Registration>) {
this.#senderFor = options.senderFor;
this.#onRegistrationIdle = options.onRegistrationIdle;
+ this.#scheduleTimeout =
+ options.scheduleTimeout ??
+ ((callback, timeoutMs) => {
+ const timer = setTimeout(callback, timeoutMs);
+ return () => clearTimeout(timer);
+ });
}
invoke(
@@ -117,20 +149,28 @@ export class ClientCapabilityInvocationBroker<
signal: AbortSignal | undefined,
timeoutMs: number,
onProgress?: (current: number, total: number) => void,
+ requestInteraction?: ClientCapabilityInteractionHandler,
): Promise<ClientCapabilityCallResult> {
- return this.#invoke(registration, signal, timeoutMs, onProgress, (invocationId) => ({
- kind: 'client.capability.call',
- invocationId,
- registrationId: registration.registrationId,
- offerId: binding.offerId,
- serverId: binding.descriptor.serverId,
- toolName: binding.descriptor.name,
- arguments: args,
- sessionId: context.sessionId,
- turnId: context.turnId,
- toolCallId: context.toolCallId,
- ...(binding.hostPathAccess === 'cwd' ? { cwd: context.cwd } : {}),
- }));
+ return this.#invoke(
+ registration,
+ signal,
+ timeoutMs,
+ onProgress,
+ requestInteraction,
+ (invocationId) => ({
+ kind: 'client.capability.call',
+ invocationId,
+ registrationId: registration.registrationId,
+ offerId: binding.offerId,
+ serverId: binding.descriptor.serverId,
+ toolName: binding.descriptor.name,
+ arguments: args,
+ sessionId: context.sessionId,
+ turnId: context.turnId,
+ toolCallId: context.toolCallId,
+ ...(binding.hostPathAccess === 'cwd' ? { cwd: context.cwd } : {}),
+ }),
+ );
}
invokeService(
@@ -142,7 +182,7 @@ export class ClientCapabilityInvocationBroker<
signal: AbortSignal | undefined,
timeoutMs: number,
): Promise<ClientCapabilityCallResult> {
- return this.#invoke(registration, signal, timeoutMs, undefined, (invocationId) => ({
+ return this.#invoke(registration, signal, timeoutMs, undefined, undefined, (invocationId) => ({
kind: 'client.capability.service_call',
invocationId,
registrationId: registration.registrationId,
@@ -158,6 +198,7 @@ export class ClientCapabilityInvocationBroker<
signal: AbortSignal | undefined,
timeoutMs: number,
onProgress: ((current: number, total: number) => void) | undefined,
+ requestInteraction: ClientCapabilityInteractionHandler | undefined,
frameFor: (invocationId: string) => ClientCapabilityHostFrame,
): Promise<ClientCapabilityCallResult> {
const sender = this.#senderFor(registration.connectionId);
@@ -189,36 +230,19 @@ export class ClientCapabilityInvocationBroker<
const invocation = this.#invocations.get(invocationId);
if (!invocation) return;
void sender.send({ kind: 'client.capability.cancel', invocationId }).catch(() => {});
- this.#settle(
- invocation,
- undefined,
+ const error =
invocation.phase === 'dispatched'
? asError(abortReason(signal))
: new ToolOutcomeUnknownError(
'Client Capability invocation was cancelled after provider acceptance',
- ),
- true,
- );
+ );
+ if (invocation.interaction) {
+ this.#terminateInteraction(invocation, error, true, false);
+ } else {
+ this.#settle(invocation, undefined, error, true);
+ }
}
: undefined;
- const timer = setTimeout(() => {
- const invocation = this.#invocations.get(invocationId);
- if (!invocation) return;
- void sender.send({ kind: 'client.capability.cancel', invocationId }).catch(() => {});
- this.#settle(
- invocation,
- undefined,
- invocation.phase === 'dispatched'
- ? new ClientCapabilityInvocationError(
- 'timed_out',
- 'Client Capability invocation timed out before provider acceptance',
- )
- : new ToolOutcomeUnknownError(
- 'Client Capability invocation timed out after provider acceptance',
- ),
- true,
- );
- }, timeoutMs);
const invocation: InvocationState<Registration> = {
invocationId,
registration,
@@ -227,10 +251,12 @@ export class ClientCapabilityInvocationBroker<
signal,
onAbort,
onProgress,
- timer,
+ requestInteraction,
+ timeoutMs,
phase: 'dispatched',
};
this.#invocations.set(invocationId, invocation);
+ this.#armTimer(invocation);
if (onAbort) signal?.addEventListener('abort', onAbort, { once: true });
void sender.send(frameFor(invocationId)).catch(() => {
const current = this.#invocations.get(invocationId);
@@ -284,14 +310,14 @@ export class ClientCapabilityInvocationBroker<
.catch(() => {
const current = this.#invocations.get(invocation.invocationId);
if (!current) return;
- this.#settle(
- current,
- undefined,
- new ToolOutcomeUnknownError(
- 'Client Capability acceptance acknowledgement could not be delivered',
- ),
- false,
+ const error = new ToolOutcomeUnknownError(
+ 'Client Capability acceptance acknowledgement could not be delivered',
);
+ if (current.interaction) {
+ this.#terminateInteraction(current, error, false, true);
+ } else {
+ this.#settle(current, undefined, error, false);
+ }
});
return;
}
@@ -310,15 +336,28 @@ export class ClientCapabilityInvocationBroker<
if (invocation.phase === 'dispatched') {
throw new Error('Client Capability failure arrived before acceptance');
}
- this.#settle(
- invocation,
- undefined,
- new ClientCapabilityInvocationError('provider_failed', frame.message),
- true,
- );
+ if (invocation.interaction) {
+ this.#terminateInteraction(
+ invocation,
+ new ClientCapabilityInvocationError('provider_failed', frame.message),
+ true,
+ true,
+ );
+ } else {
+ this.#settle(
+ invocation,
+ undefined,
+ new ClientCapabilityInvocationError('provider_failed', frame.message),
+ true,
+ );
+ }
return;
case 'client.capability.progress':
- if (invocation.phase !== 'accepted' && invocation.phase !== 'chunks') {
+ if (
+ invocation.phase !== 'accepted' &&
+ invocation.phase !== 'delivering_interaction_result' &&
+ invocation.phase !== 'chunks'
+ ) {
throw new Error('Client Capability progress arrived before acceptance');
}
if (
@@ -331,14 +370,47 @@ export class ClientCapabilityInvocationBroker<
invocation.progress = { current: frame.current, total: frame.total };
invocation.onProgress?.(frame.current, frame.total);
return;
+ case 'client.capability.interaction_request':
+ this.#acceptInteraction(invocation, frame.interactionId, frame.request);
+ return;
case 'client.capability.result':
- if (invocation.phase !== 'accepted') {
+ if (invocation.phase === 'awaiting_interaction') {
+ this.#terminateInteraction(
+ invocation,
+ new ClientCapabilityInvocationError(
+ 'provider_failed',
+ 'Client Capability provider returned before its interaction completed',
+ ),
+ true,
+ true,
+ );
+ return;
+ }
+ if (
+ invocation.phase !== 'accepted' &&
+ invocation.phase !== 'delivering_interaction_result'
+ ) {
throw new Error('Client Capability result arrived outside the accepted phase');
}
this.#settle(invocation, frame.result, undefined, true);
return;
case 'client.capability.result_start':
- if (invocation.phase !== 'accepted') {
+ if (invocation.phase === 'awaiting_interaction') {
+ this.#terminateInteraction(
+ invocation,
+ new ClientCapabilityInvocationError(
+ 'provider_failed',
+ 'Client Capability provider returned before its interaction completed',
+ ),
+ true,
+ true,
+ );
+ return;
+ }
+ if (
+ invocation.phase !== 'accepted' &&
+ invocation.phase !== 'delivering_interaction_result'
+ ) {
throw new Error('Client Capability result chunks started outside the accepted phase');
}
invocation.phase = 'chunks';
@@ -354,12 +426,11 @@ export class ClientCapabilityInvocationBroker<
}
}
- releaseConnection(connectionId: string): void {
+ async releaseConnection(connectionId: string): Promise<void> {
+ const interactions: Promise<void>[] = [];
for (const invocation of [...this.#invocations.values()]) {
if (invocation.registration.connectionId !== connectionId) continue;
- this.#settle(
- invocation,
- undefined,
+ const error =
invocation.phase === 'dispatched'
? new ClientCapabilityInvocationError(
'capability_lost',
@@ -367,10 +438,15 @@ export class ClientCapabilityInvocationBroker<
)
: new ToolOutcomeUnknownError(
'Client Capability provider disconnected after accepting the call',
- ),
- false,
- );
+ );
+ if (invocation.interaction) {
+ interactions.push(invocation.interaction.done);
+ this.#terminateInteraction(invocation, error, false, true);
+ } else {
+ this.#settle(invocation, undefined, error, false);
+ }
}
+ await Promise.all(interactions);
}
holdsRegistration(registration: Registration): boolean {
@@ -415,6 +491,181 @@ export class ClientCapabilityInvocationBroker<
this.#settle(invocation, decodeClientCapabilityResult(decoded), undefined, true);
}
+ #acceptInteraction(
+ invocation: InvocationState<Registration>,
+ interactionId: string,
+ request: InteractionFormInput,
+ ): void {
+ if (
+ (invocation.phase !== 'accepted' && invocation.phase !== 'delivering_interaction_result') ||
+ invocation.interaction
+ ) {
+ if (invocation.interaction) {
+ this.#terminateInteraction(
+ invocation,
+ new ClientCapabilityInvocationError(
+ 'provider_failed',
+ 'Client Capability provider requested overlapping interactions',
+ ),
+ true,
+ true,
+ );
+ return;
+ }
+ throw new Error('Client Capability interaction arrived outside the accepted phase');
+ }
+ if (!invocation.requestInteraction) {
+ this.#settle(
+ invocation,
+ undefined,
+ new ClientCapabilityInvocationError(
+ 'provider_failed',
+ 'Client Capability interaction is unavailable for this invocation',
+ ),
+ true,
+ );
+ return;
+ }
+ this.#clearTimer(invocation);
+ invocation.phase = 'awaiting_interaction';
+ let resolveDone!: () => void;
+ const interaction: InvocationInteraction = {
+ interactionId,
+ controller: new AbortController(),
+ done: new Promise<void>((resolve) => {
+ resolveDone = resolve;
+ }),
+ resolveDone: () => resolveDone(),
+ };
+ invocation.interaction = interaction;
+ void this.#runInteraction(invocation, interaction, request);
+ }
+
+ async #runInteraction(
+ invocation: InvocationState<Registration>,
+ interaction: InvocationInteraction,
+ request: InteractionFormInput,
+ ): Promise<void> {
+ try {
+ const signal = invocation.signal
+ ? AbortSignal.any([invocation.signal, interaction.controller.signal])
+ : interaction.controller.signal;
+ const result = await invocation.requestInteraction!(request, {
+ cancellationSignal: signal,
+ });
+ if (!this.#isCurrentInteraction(invocation, interaction)) return;
+ const terminalBeforeSend = interactionTerminal(interaction);
+ if (terminalBeforeSend) {
+ this.#settle(
+ invocation,
+ undefined,
+ terminalBeforeSend.error,
+ terminalBeforeSend.releaseRemote,
+ );
+ return;
+ }
+ const sender = this.#senderFor(invocation.registration.connectionId);
+ if (!sender) {
+ this.#settle(
+ invocation,
+ undefined,
+ new ToolOutcomeUnknownError(
+ 'Client Capability provider disappeared before receiving an interaction result',
+ ),
+ false,
+ );
+ return;
+ }
+ invocation.interaction = undefined;
+ invocation.phase = 'delivering_interaction_result';
+ this.#armTimer(invocation);
+ await sender.send({
+ kind: 'client.capability.interaction_result',
+ invocationId: invocation.invocationId,
+ interactionId: interaction.interactionId,
+ result,
+ });
+ if (this.#invocations.get(invocation.invocationId) !== invocation) return;
+ if (invocation.phase !== 'delivering_interaction_result') return;
+ invocation.phase = 'accepted';
+ this.#armTimer(invocation);
+ } catch (error) {
+ if (this.#invocations.get(invocation.invocationId) !== invocation) return;
+ if (invocation.interaction !== interaction) {
+ if (invocation.phase !== 'delivering_interaction_result') return;
+ this.#settle(
+ invocation,
+ undefined,
+ new ToolOutcomeUnknownError(
+ `Client Capability interaction result could not be delivered: ${asError(error).message}`,
+ ),
+ false,
+ );
+ return;
+ }
+ this.#settle(
+ invocation,
+ undefined,
+ interaction.terminal?.error ?? asError(error),
+ interaction.terminal?.releaseRemote ?? true,
+ );
+ } finally {
+ interaction.resolveDone();
+ }
+ }
+
+ #terminateInteraction(
+ invocation: InvocationState<Registration>,
+ error: Error,
+ releaseRemote: boolean,
+ cancelProducer: boolean,
+ ): void {
+ const interaction = invocation.interaction;
+ if (!interaction || interaction.terminal) return;
+ interaction.terminal = { error, releaseRemote };
+ if (cancelProducer) interaction.controller.abort(error);
+ }
+
+ #isCurrentInteraction(
+ invocation: InvocationState<Registration>,
+ interaction: InvocationInteraction,
+ ): boolean {
+ return (
+ this.#invocations.get(invocation.invocationId) === invocation &&
+ invocation.interaction === interaction
+ );
+ }
+
+ #armTimer(invocation: InvocationState<Registration>): void {
+ this.#clearTimer(invocation);
+ invocation.cancelTimer = this.#scheduleTimeout(() => {
+ const current = this.#invocations.get(invocation.invocationId);
+ if (current !== invocation) return;
+ const sender = this.#senderFor(invocation.registration.connectionId);
+ void sender
+ ?.send({ kind: 'client.capability.cancel', invocationId: invocation.invocationId })
+ .catch(() => {});
+ this.#settle(
+ invocation,
+ undefined,
+ invocation.phase === 'dispatched'
+ ? new ClientCapabilityInvocationError(
+ 'timed_out',
+ 'Client Capability invocation timed out before provider acceptance',
+ )
+ : new ToolOutcomeUnknownError(
+ 'Client Capability invocation timed out after provider acceptance',
+ ),
+ true,
+ );
+ }, invocation.timeoutMs);
+ }
+
+ #clearTimer(invocation: InvocationState<Registration>): void {
+ invocation.cancelTimer?.();
+ invocation.cancelTimer = undefined;
+ }
+
#settle(
invocation: InvocationState<Registration>,
result: ClientCapabilityCallResult | undefined,
@@ -423,7 +674,7 @@ export class ClientCapabilityInvocationBroker<
): void {
if (this.#invocations.get(invocation.invocationId) !== invocation) return;
this.#invocations.delete(invocation.invocationId);
- clearTimeout(invocation.timer);
+ this.#clearTimer(invocation);
if (invocation.onAbort && invocation.signal) {
invocation.signal.removeEventListener('abort', invocation.onAbort);
}
@@ -455,6 +706,12 @@ function abortReason(signal: AbortSignal): unknown {
return signal.reason ?? new Error('Client Capability invocation cancelled');
}
+function interactionTerminal(
+ interaction: InvocationInteraction,
+): InvocationInteraction['terminal'] {
+ return interaction.terminal;
+}
+
function asError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error));
}
diff --git a/packages/runtime/src/__tests__/mcp-tools.test.ts b/packages/runtime/src/__tests__/mcp-tools.test.ts
index b3ffdeb2b3..5769c21995 100644
--- a/packages/runtime/src/__tests__/mcp-tools.test.ts
+++ b/packages/runtime/src/__tests__/mcp-tools.test.ts
@@ -97,6 +97,44 @@ test('buildMcpTools projects discovery, abort, and rich model output', async ()
assert.match(model?.value[2]?.type === 'text' ? model.value[2].text : '', /structuredContent/u);
});
+test('buildMcpTools carries the Runtime-owned form callback to the provider', async () => {
+ const cancellation = new AbortController();
+ const provider = fakeProvider(
+ [boundTool(descriptor('client', 'deploy'), binding('nested-form-binding'))],
+ async (_binding, _args, options) => {
+ assert.ok(options.requestInteraction);
+ const answer = await options.requestInteraction(
+ {
+ message: 'Choose a target',
+ requester: { name: 'deploy' },
+ fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }],
+ },
+ { cancellationSignal: cancellation.signal },
+ );
+ assert.deepEqual(answer, { action: 'accept', values: { target: 'staging' } });
+ return { content: [] };
+ },
+ );
+ const [tool] = buildMcpTools(provider);
+
+ await tool?.impl(
+ {},
+ {
+ sessionId: 'session',
+ turnId: 'turn',
+ cwd: '/workspace',
+ toolCallId: 'tool-call',
+ abortSignal: new AbortController().signal,
+ emitOutput() {},
+ requestUserForm: async (form, options) => {
+ assert.equal(form.message, 'Choose a target');
+ assert.equal(options?.cancellationSignal, cancellation.signal);
+ return { action: 'accept', values: { target: 'staging' } };
+ },
+ },
+ );
+});
+
test('Direct-mode MCP calls request managed network expansion before provider dispatch', async () => {
const sequence: string[] = [];
const boundary = createManagedExecutionBoundary(createWorkspaceWritePermissionProfile(), 0);
diff --git a/packages/runtime/src/mcp-tools.ts b/packages/runtime/src/mcp-tools.ts
index fce515ad42..c204f3761a 100644
--- a/packages/runtime/src/mcp-tools.ts
+++ b/packages/runtime/src/mcp-tools.ts
@@ -26,6 +26,7 @@ import type {
McpToolDescriptor,
McpToolSnapshot,
} from '@maka/core/mcp';
+import type { InteractionFormInput, InteractionFormResult } from '@maka/core/interaction';
import type { ToolCategory } from '@maka/core/permission';
import type { ToolRecoveryMode } from '@maka/core/runtime-event';
import type { ToolResultContentPart, ToolResultOutput } from './model-protocol.js';
@@ -53,6 +54,10 @@ export interface McpToolCallOptions {
readonly timeoutMs?: number;
readonly context: McpToolInvocationContext;
readonly emitProgress?: (current: number, total: number) => void;
+ readonly requestInteraction?: (
+ form: InteractionFormInput,
+ options?: { readonly cancellationSignal?: AbortSignal },
+ ) => Promise<InteractionFormResult>;
}
export interface McpToolInvocationContext {
@@ -125,6 +130,14 @@ export function buildMcpTools(
cwd: context.cwd,
},
...(context.emitProgress ? { emitProgress: context.emitProgress } : {}),
+ ...(context.requestUserForm
+ ? {
+ requestInteraction: (
+ form: InteractionFormInput,
+ interactionOptions?: { readonly cancellationSignal?: AbortSignal },
+ ) => context.requestUserForm!(form, interactionOptions),
+ }
+ : {}),
});
},
toModelOutput: ({ output }) => mcpResultToModelOutput(output),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org