From 3fff4449c2ff189102e50ef95de0be55c94116bd Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 1 Sep 2026 01:15:23 +0800 Subject: [PATCH 1/9] feat(runtime): withdraw producer-owned forms Let an exact hosted Run withdraw one form without closing its surrounding Turn. Commit producer cancellation through the existing InteractionStore authority, preserve an already-claimed Run closure, and compose provider-local cancellation with the Tool invocation signal. Part of #4364. Generated-by: OpenAI Codex --- packages/core/src/backend-types.ts | 2 + packages/core/src/interaction.ts | 1 + .../__tests__/interaction-coordinator.test.ts | 20 +++- .../src/server/interaction-coordinator.ts | 55 +++++++++ .../src/__tests__/fake-backend.test.ts | 1 + .../__tests__/interaction-authority.test.ts | 2 + .../runtime-kernel-interaction.test.ts | 2 + .../session-manager-terminal-ledger.test.ts | 1 + .../src/__tests__/session-manager.test.ts | 1 + .../tool-runtime-form-interaction.test.ts | 108 ++++++++++++++---- .../tool-runtime-sandbox-boundary.test.ts | 3 + packages/runtime/src/interaction-authority.ts | 5 + packages/runtime/src/tool-runtime.ts | 45 ++++++-- 13 files changed, 208 insertions(+), 38 deletions(-) diff --git a/packages/core/src/backend-types.ts b/packages/core/src/backend-types.ts index a3ffdcae98..7b9d45854f 100644 --- a/packages/core/src/backend-types.ts +++ b/packages/core/src/backend-types.ts @@ -160,6 +160,8 @@ export interface HostedInteractionBridge { request: FormRequestEvent; settlement: HostedFormSettlement; }): Promise; + /** Withdraw one exact producer-owned form without closing the surrounding Run. */ + withdrawFormRequest(requestId: string): Promise; admitSandboxBoundaryRequest(input: { request: SandboxBoundaryRequestEvent; settlement: HostedSandboxBoundarySettlement; diff --git a/packages/core/src/interaction.ts b/packages/core/src/interaction.ts index 383e61f5e7..a05a4cde46 100644 --- a/packages/core/src/interaction.ts +++ b/packages/core/src/interaction.ts @@ -72,6 +72,7 @@ export const INTERACTION_FORM_VALUE_MAX_BYTES = 2_048; export const INTERACTION_CLOSURE_REASONS = [ 'turn_stopped', 'turn_terminal', + 'producer_cancelled', 'timed_out', 'host_restarted', 'provider_disconnected', diff --git a/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts b/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts index 3b978c79bd..52ebb394de 100644 --- a/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts @@ -230,13 +230,27 @@ describe('HostInteractionCoordinator', () => { closure: (reason) => closures.push(reason), }), }); - await owner.close('turn_terminal'); - assert.deepEqual(closures, ['turn_terminal']); + await owner.withdrawFormRequest('form_2'); + assert.deepEqual(closures, ['producer_cancelled']); assert.deepEqual((await store.readInteraction('form_2'))?.outcome?.outcome, { kind: 'closure', - reason: 'turn_terminal', + reason: 'producer_cancelled', committedAt: 102, }); + + await owner.acceptFormRequest({ + request: formEvent('form_3', 12), + continuation: formContinuation('form_3', { + closure: (reason) => closures.push(reason), + }), + }); + await owner.close('turn_terminal'); + assert.deepEqual(closures, ['producer_cancelled', 'turn_terminal']); + assert.deepEqual((await store.readInteraction('form_3'))?.outcome?.outcome, { + kind: 'closure', + reason: 'turn_terminal', + committedAt: 103, + }); owner.release(); await coordinator.close(); }); diff --git a/packages/runtime-host/src/server/interaction-coordinator.ts b/packages/runtime-host/src/server/interaction-coordinator.ts index 589517d04c..be899c983f 100644 --- a/packages/runtime-host/src/server/interaction-coordinator.ts +++ b/packages/runtime-host/src/server/interaction-coordinator.ts @@ -254,6 +254,7 @@ export class HostInteractionCoordinator implements RuntimeInteractionAuthority { ) => this.#acceptUserQuestionRequest(run, input), acceptFormRequest: (input: Parameters[0]) => this.#acceptFormRequest(run, input), + withdrawFormRequest: (requestId: string) => this.#withdrawFormRequest(run, requestId), acceptSandboxBoundaryRequest: ( input: Parameters[0], ) => this.#acceptSandboxBoundaryRequest(run, input), @@ -969,6 +970,60 @@ export class HostInteractionCoordinator implements RuntimeInteractionAuthority { } } + #withdrawFormRequest(run: BoundRun, requestId: string): Promise { + try { + this.#assertOwnedRun(run); + this.#throwIfPoisoned(); + if (run.released) { + throw this.#poison( + new RuntimeInteractionInvariantError( + `Released Interaction Run ${run.runId} cannot withdraw a form`, + ), + ); + } + return observed( + this.#sessionAdmission.run(run.sessionId, async (admission) => { + this.#throwIfPoisoned(); + // A whole-Run stop/terminal closure that already claimed ownership + // remains the reason for every still-pending Interaction in that Run. + if (run.closure) return; + const record = await this.#readInteraction(requestId); + // Cancellation may win while admission is still proving publication. + // The producer will also observe its abort and must not publish afterward. + if (!record) return; + if (!sameRun(record.request, run) || record.request.request.kind !== 'form') { + throw this.#poison( + new RuntimeInteractionInvariantError( + `Interaction Run ${run.runId} cannot withdraw form ${requestId}`, + ), + ); + } + // A canonical user answer or Run closure that won the Session admission + // race stays authoritative. + if (record.outcome) return; + const entry = this.#requireLiveStored(record.request); + if (entry.kind !== 'form' || entry.run !== run) { + throw this.#poison( + new RuntimeInteractionInvariantError( + `Form ${requestId} is not owned by Interaction Run ${run.runId}`, + ), + ); + } + const outcome = await this.#commitOutcome(record.request, { + kind: 'closure', + reason: 'producer_cancelled', + committedAt: this.#now(), + }); + await this.#refreshCanonicalContinuity(run.sessionId, admission); + this.#throwIfPoisoned(); + await this.#applyAndDelete(entry, outcome); + }), + ); + } catch (error) { + return rejected(error); + } + } + #claimRunClosure(run: BoundRun, reason: RuntimeInteractionRunClosureReason): RunClosure { if (run.closure) return run.closure; diff --git a/packages/runtime/src/__tests__/fake-backend.test.ts b/packages/runtime/src/__tests__/fake-backend.test.ts index 0673427f59..c61f3447e5 100644 --- a/packages/runtime/src/__tests__/fake-backend.test.ts +++ b/packages/runtime/src/__tests__/fake-backend.test.ts @@ -45,6 +45,7 @@ test('Fake question publication waits for exact hosted admission', async () => { await allowAdmission.promise; }, acceptFormRequest: async () => {}, + withdrawFormRequest: async () => {}, close: async () => {}, release: () => {}, }), diff --git a/packages/runtime/src/__tests__/interaction-authority.test.ts b/packages/runtime/src/__tests__/interaction-authority.test.ts index 25cbb72508..ebca8c44c3 100644 --- a/packages/runtime/src/__tests__/interaction-authority.test.ts +++ b/packages/runtime/src/__tests__/interaction-authority.test.ts @@ -91,6 +91,7 @@ describe('Runtime Interaction authority seam', () => { acceptSandboxBoundaryRequest: async () => {}, acceptUserQuestionRequest: async () => {}, acceptFormRequest: async () => {}, + withdrawFormRequest: async () => {}, close: async (reason) => { log.push(`close:${reason}`); }, @@ -603,6 +604,7 @@ function authority( release: () => {}, ...overrides, acceptFormRequest: overrides.acceptFormRequest ?? (async () => {}), + withdrawFormRequest: overrides.withdrawFormRequest ?? (async () => {}), }), }; } diff --git a/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts b/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts index 8f3ce30567..2216a52160 100644 --- a/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts +++ b/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts @@ -101,6 +101,7 @@ describe('RuntimeKernel Interaction close cleanup', () => { acceptSandboxBoundaryRequest: async () => {}, acceptUserQuestionRequest: async () => {}, acceptFormRequest: async () => {}, + withdrawFormRequest: async () => {}, close: async () => { closeCalls += 1; closeStarted.resolve(); @@ -485,6 +486,7 @@ function runtimeFixture(options: RuntimeFixtureOptions = {}): { acceptSandboxBoundaryRequest: async () => {}, acceptUserQuestionRequest: async () => {}, acceptFormRequest: async () => {}, + withdrawFormRequest: async () => {}, close: async () => { markCloseStarted(); if (options.deferredClose) await closeReleased; diff --git a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts index 6da3e1c27b..9c343b10b4 100644 --- a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts +++ b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts @@ -2926,6 +2926,7 @@ function hostedInteractionAuthority(): RuntimeInteractionAuthority { acceptSandboxBoundaryRequest: async () => {}, acceptUserQuestionRequest: async () => {}, acceptFormRequest: async () => {}, + withdrawFormRequest: async () => {}, close: async () => {}, release: () => {}, }), diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index cc9d56622f..a0babc9ecb 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -14763,6 +14763,7 @@ function testInteractionAuthority(): RuntimeInteractionAuthority { acceptSandboxBoundaryRequest: async () => {}, acceptUserQuestionRequest: async () => {}, acceptFormRequest: async () => {}, + withdrawFormRequest: async () => {}, close: async () => {}, release: () => {}, }), diff --git a/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts b/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts index af70508124..72653b191a 100644 --- a/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts +++ b/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts @@ -19,6 +19,7 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; +import type { HostedFormSettlement } from '@maka/core/backend-types'; import type { SessionEvent } from '@maka/core/events'; import type { SessionHeader } from '@maka/core/session'; import { z } from 'zod'; @@ -49,7 +50,7 @@ function header(): SessionHeader { }; } -function formTool(): MakaTool> { +function formTool(cancellationSignal?: AbortSignal): MakaTool> { return { name: 'SyntheticForm', description: 'Exercise the provider-neutral form seam.', @@ -57,30 +58,33 @@ function formTool(): MakaTool> { nesting: 'direct_only', impl: (_input, context) => { if (!context.requestUserForm) throw new Error('Form Interaction is unavailable'); - return context.requestUserForm({ - message: 'Choose deployment settings', - requester: { name: 'deploy', source: 'Synthetic provider' }, - fields: [ - { - kind: 'integer', - name: 'replicas', - label: 'Replicas', - required: true, - minimum: 1, - maximum: 10, - }, - { - kind: 'multi_select', - name: 'regions', - label: 'Regions', - required: false, - options: [ - { value: 'us', label: 'US' }, - { value: 'eu', label: 'EU' }, - ], - }, - ], - }); + return context.requestUserForm( + { + message: 'Choose deployment settings', + requester: { name: 'deploy', source: 'Synthetic provider' }, + fields: [ + { + kind: 'integer', + name: 'replicas', + label: 'Replicas', + required: true, + minimum: 1, + maximum: 10, + }, + { + kind: 'multi_select', + name: 'regions', + label: 'Regions', + required: false, + options: [ + { value: 'us', label: 'US' }, + { value: 'eu', label: 'EU' }, + ], + }, + ], + }, + cancellationSignal ? { cancellationSignal } : undefined, + ); }, }; } @@ -197,4 +201,58 @@ describe('ToolRuntime form Interaction', () => { false, ); }); + + test('withdraws the exact hosted form when its producer is cancelled', async () => { + const events: SessionEvent[] = []; + const producer = new AbortController(); + let admitted: { requestId: string; settlement: HostedFormSettlement } | undefined; + const withdrawals: string[] = []; + const toolRuntime = createTestToolRuntime({ + sessionId: 'session-1', + header: header(), + connection: { providerType: 'openai', slug: 'c' } as never, + modelId: 'm', + appendMessage: async () => {}, + newId: (() => { + let id = 0; + return () => `id-${++id}`; + })(), + now: () => 1, + getPermissionPauseTarget: () => null, + hostedInteraction: { + sessionId: 'session-1', + turnId: 'turn-1', + runId: 'run-1', + admitUserQuestionRequest: async () => { + throw new Error('Unexpected user question'); + }, + admitFormRequest: async (input) => { + admitted = { requestId: input.request.requestId, settlement: input.settlement }; + }, + withdrawFormRequest: async (requestId) => { + withdrawals.push(requestId); + await admitted?.settlement.applyClosure('producer_cancelled'); + }, + admitSandboxBoundaryRequest: async () => { + throw new Error('Unexpected sandbox boundary'); + }, + }, + }); + const pending = toolRuntime.settleToolCall({ + tool: formTool(producer.signal), + turnId: 'turn-1', + toolCallId: 'tool-1', + input: {}, + abortSignal: new AbortController().signal, + eventSink: sink(events), + }); + while (!admitted) await new Promise((resolve) => setImmediate(resolve)); + + producer.abort(new DOMException('Provider invocation ended', 'AbortError')); + await pending; + + assert.deepEqual(withdrawals, [admitted.requestId]); + assert.equal(toolRuntime.pendingUserFormCount(), 0); + assert.equal(events.filter((event) => event.type === 'form_answer_ack').length, 0); + }); }); diff --git a/packages/runtime/src/__tests__/tool-runtime-sandbox-boundary.test.ts b/packages/runtime/src/__tests__/tool-runtime-sandbox-boundary.test.ts index acab112c5b..e4a71162ac 100644 --- a/packages/runtime/src/__tests__/tool-runtime-sandbox-boundary.test.ts +++ b/packages/runtime/src/__tests__/tool-runtime-sandbox-boundary.test.ts @@ -222,6 +222,9 @@ describe('ToolRuntime session sandbox boundary', () => { admitFormRequest: async () => { throw new Error('Unexpected user form'); }, + withdrawFormRequest: async () => { + throw new Error('Unexpected user form withdrawal'); + }, admitSandboxBoundaryRequest: async ({ request, settlement }) => { admittedRequest = request; captured = settlement; diff --git a/packages/runtime/src/interaction-authority.ts b/packages/runtime/src/interaction-authority.ts index 374c132705..b4463624f2 100644 --- a/packages/runtime/src/interaction-authority.ts +++ b/packages/runtime/src/interaction-authority.ts @@ -120,6 +120,7 @@ export interface RuntimeInteractionRunFacet RuntimeInteractionRunIdentity {} export interface RuntimeInteractionRunOwner extends RuntimeInteractionRunFacet { + withdrawFormRequest(requestId: string): Promise; close(reason: RuntimeInteractionRunClosureReason): Promise; release(): void; } @@ -383,6 +384,10 @@ export class RuntimeInteractionRunBinding implements HostedInteractionBridge { } } + withdrawFormRequest(requestId: string): Promise { + return this.owner.withdrawFormRequest(requestId); + } + async admitSandboxBoundaryRequest(input: { request: SandboxBoundaryRequestEvent; settlement: HostedSandboxBoundarySettlement; diff --git a/packages/runtime/src/tool-runtime.ts b/packages/runtime/src/tool-runtime.ts index b542c7a6b3..326ba4d550 100644 --- a/packages/runtime/src/tool-runtime.ts +++ b/packages/runtime/src/tool-runtime.ts @@ -310,7 +310,10 @@ export interface MakaToolContext { view?: 'result' | 'events' | 'runtime_events' | 'all'; }) => Promise; askUserQuestion?: (questions: UserQuestion[]) => Promise; - requestUserForm?: (form: InteractionFormInput) => Promise; + requestUserForm?: ( + form: InteractionFormInput, + options?: { readonly cancellationSignal?: AbortSignal }, + ) => Promise; requestSandboxBoundary?: ( expansion: SandboxBoundaryExpansion, justification: string, @@ -1722,8 +1725,15 @@ export class ToolRuntime { }), askUserQuestion: (questions) => this.askUserQuestion(turnId, toolUseId, questions, ctx.abortSignal, queue), - requestUserForm: (form) => - this.requestUserForm(turnId, toolUseId, form, ctx.abortSignal, queue), + requestUserForm: (form, options) => + this.requestUserForm( + turnId, + toolUseId, + form, + ctx.abortSignal, + queue, + options?.cancellationSignal, + ), requestSandboxBoundary: (expansion, justification) => this.requestSandboxBoundary( turnId, @@ -2634,8 +2644,10 @@ export class ToolRuntime { form: InteractionFormInput, abortSignal: AbortSignal, queue: DurableSessionEventSink, + producerCancellationSignal?: AbortSignal, ): Promise { - throwIfAborted(abortSignal); + const interactionSignal = composeChildAbortSignal(abortSignal, producerCancellationSignal); + throwIfAborted(interactionSignal); const hostedRun = this.interactionRun(); const requestId = this.input.newId(); const request = projectInteractionFormRequest({ toolUseId, ...form }); @@ -2649,7 +2661,18 @@ export class ToolRuntime { this.userForms.reject(requestId, abortErrorFromSignal(abortSignal)); this.finishDeferredFormTurnClosure(); }; + let producerWithdrawal: Promise | undefined; + const onProducerCancellation = (): void => { + if (abortSignal.aborted) return; + if (hostedRun) { + producerWithdrawal ??= hostedRun.withdrawFormRequest(requestId); + } else if (producerCancellationSignal) { + this.userForms.reject(requestId, abortErrorFromSignal(producerCancellationSignal)); + this.finishDeferredFormTurnClosure(); + } + }; abortSignal.addEventListener('abort', onAbort, { once: true }); + producerCancellationSignal?.addEventListener('abort', onProducerCancellation, { once: true }); if (hostedRun) void parked.catch(() => undefined); try { const requestEvent: FormRequestEvent = { @@ -2667,9 +2690,9 @@ export class ToolRuntime { const settlement = this.createFormSettlement(turnId, requestId); const admission = hostedRun.admitFormRequest({ request: requestEvent, settlement }); try { - await racePromiseWithAbort(admission, abortSignal); + await racePromiseWithAbort(admission, interactionSignal); } catch (error) { - if (abortSignal.aborted) { + if (interactionSignal.aborted) { void admission.catch((admissionError) => { this.userForms.reject( requestId, @@ -2682,7 +2705,7 @@ export class ToolRuntime { ); this.finishDeferredFormTurnClosure(); }); - throw abortErrorFromSignal(abortSignal); + throw abortErrorFromSignal(interactionSignal); } this.userForms.reject( requestId, @@ -2701,10 +2724,10 @@ export class ToolRuntime { ); } } - throwIfAborted(abortSignal); + throwIfAborted(interactionSignal); queue.push(requestEvent); - const response = await racePromiseWithAbort(parked, abortSignal); - throwIfAborted(abortSignal); + const response = await racePromiseWithAbort(parked, interactionSignal); + throwIfAborted(interactionSignal); const answerAck: FormAnswerAckEvent = { type: 'form_answer_ack', id: this.input.newId(), @@ -2720,6 +2743,8 @@ export class ToolRuntime { : { action: response.action }; } finally { abortSignal.removeEventListener('abort', onAbort); + producerCancellationSignal?.removeEventListener('abort', onProducerCancellation); + if (producerWithdrawal) await producerWithdrawal; } } From 998ee6c239aa0478439be986ef172d7a0f557cf9 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 1 Sep 2026 01:27:04 +0800 Subject: [PATCH 2/9] feat(runtime-host): carry nested capability forms Add strict request/result frames and expose one provider-neutral requestInteraction callback for admitted Client Capability invocations. Keep correlation inside the client channel and publish a new compatibility epoch for peers that understand the round trip.\n\nPart of #4364.\n\nGenerated-by: OpenAI Codex --- .../client-capability-channel.test.ts | 177 ++++++++++++++++++ .../client-capability-protocol.test.ts | 88 +++++++++ .../src/__tests__/protocol.test.ts | 4 + .../src/client/client-capability-channel.ts | 102 ++++++++++ .../src/client/client-capability.ts | 3 + .../src/protocol/client-capability.ts | 92 ++++++++- packages/runtime-host/src/protocol/index.ts | 4 +- 7 files changed, 467 insertions(+), 3 deletions(-) diff --git a/packages/runtime-host/src/__tests__/client-capability-channel.test.ts b/packages/runtime-host/src/__tests__/client-capability-channel.test.ts index fd48cffabb..ff69d4a4e2 100644 --- a/packages/runtime-host/src/__tests__/client-capability-channel.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-channel.test.ts @@ -301,3 +301,180 @@ test('Client Capability channel forwards admitted tool progress before the resul ]); channel.close(new Error('test complete')); }); + +test('Client Capability channel correlates one admitted nested form before the final result', async () => { + let registrationId = ''; + const written: unknown[] = []; + let channel!: ClientCapabilityChannel; + const provider: ClientCapabilityProvider = { + offers: () => [ + { + offerId: 'fixture', + version: '0', + affinity: 'call', + hostPathAccess: 'none', + label: 'Fixture', + tools: [{ serverId: 'fixture', name: 'deploy', inputSchema: { type: 'object' } }], + }, + ], + call: async (_frame, options) => { + await options.accept(); + const answer = await options.requestInteraction({ + message: 'Choose a target', + requester: { name: 'deploy', source: 'Fixture' }, + fields: [ + { + kind: 'single_select', + name: 'target', + label: 'Target', + required: true, + options: [ + { value: 'staging', label: 'Staging' }, + { value: 'production', label: 'Production' }, + ], + }, + ], + }); + assert.deepEqual(answer, { action: 'accept', values: { target: 'staging' } }); + return { content: [{ type: 'text', text: 'deployed' }] }; + }, + }; + channel = new ClientCapabilityChannel({ + write: async (frame) => { + written.push(frame); + if (frame.kind === 'client.capability.accepted') { + queueMicrotask(() => + channel.accept({ + kind: 'client.capability.admitted', + invocationId: frame.invocationId, + }), + ); + } else if (frame.kind === 'client.capability.interaction_request') { + queueMicrotask(() => + channel.accept({ + kind: 'client.capability.interaction_result', + invocationId: frame.invocationId, + interactionId: frame.interactionId, + result: { action: 'accept', values: { target: 'staging' } }, + }), + ); + } + }, + replace: async (input) => { + registrationId = input.registrationId; + return { registrationId, revision: 1 }; + }, + unregister: async (input) => ({ registrationId: input.registrationId, revision: 2 }), + onFailure: (error) => { + throw error; + }, + }); + await channel.replace(provider, 1_000); + channel.accept({ + kind: 'client.capability.call', + invocationId: 'nested-form', + registrationId, + offerId: 'fixture', + serverId: 'fixture', + toolName: 'deploy', + arguments: {}, + sessionId: 'session', + turnId: 'turn', + toolCallId: 'tool-call', + }); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); + + const interaction = written.find( + (frame) => + typeof frame === 'object' && + frame !== null && + 'kind' in frame && + frame.kind === 'client.capability.interaction_request', + ); + assert.ok(interaction); + assert.deepEqual(written.at(-1), { + kind: 'client.capability.result', + invocationId: 'nested-form', + result: { content: [{ type: 'text', text: 'deployed' }] }, + }); + channel.accept({ kind: 'client.capability.release', invocationId: 'nested-form' }); + channel.close(new Error('test complete')); +}); + +test('Client Capability release rejects a pending nested form', async () => { + let registrationId = ''; + let interactionStarted!: () => void; + const started = new Promise((resolve) => { + interactionStarted = resolve; + }); + let observedError: unknown; + let channel!: ClientCapabilityChannel; + const provider: ClientCapabilityProvider = { + offers: () => [ + { + offerId: 'fixture', + version: '0', + affinity: 'call', + hostPathAccess: 'none', + label: 'Fixture', + tools: [{ serverId: 'fixture', name: 'deploy', inputSchema: { type: 'object' } }], + }, + ], + call: async (_frame, options) => { + await options.accept(); + try { + await options.requestInteraction({ + message: 'Choose a target', + requester: { name: 'deploy' }, + fields: [{ kind: 'boolean', name: 'confirm', label: 'Confirm', required: true }], + }); + return { content: [] }; + } catch (error) { + observedError = error; + throw error; + } + }, + }; + channel = new ClientCapabilityChannel({ + write: async (frame) => { + if (frame.kind === 'client.capability.accepted') { + queueMicrotask(() => + channel.accept({ + kind: 'client.capability.admitted', + invocationId: frame.invocationId, + }), + ); + } else if (frame.kind === 'client.capability.interaction_request') { + interactionStarted(); + } + }, + replace: async (input) => { + registrationId = input.registrationId; + return { registrationId, revision: 1 }; + }, + unregister: async (input) => ({ registrationId: input.registrationId, revision: 2 }), + onFailure: (error) => { + throw error; + }, + }); + await channel.replace(provider, 1_000); + channel.accept({ + kind: 'client.capability.call', + invocationId: 'released-form', + registrationId, + offerId: 'fixture', + serverId: 'fixture', + toolName: 'deploy', + arguments: {}, + sessionId: 'session', + turnId: 'turn', + toolCallId: 'tool-call', + }); + await started; + + channel.accept({ kind: 'client.capability.release', invocationId: 'released-form' }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(observedError instanceof Error && observedError.name, 'AbortError'); + channel.close(new Error('test complete')); +}); diff --git a/packages/runtime-host/src/__tests__/client-capability-protocol.test.ts b/packages/runtime-host/src/__tests__/client-capability-protocol.test.ts index 1f79a4b7a3..b5cfcdf54a 100644 --- a/packages/runtime-host/src/__tests__/client-capability-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-protocol.test.ts @@ -136,6 +136,94 @@ describe('Client Capability protocol', () => { invocationId: 'invocation', }, ); + assert.deepEqual( + decodeClientFrame({ + kind: 'client.capability.interaction_request', + invocationId: 'invocation', + interactionId: 'provider-form-1', + request: { + message: 'Choose a target', + requester: { name: 'deploy', source: 'Fixture' }, + fields: [ + { + kind: 'single_select', + name: 'target', + label: 'Target', + required: true, + options: [ + { value: 'staging', label: 'Staging' }, + { value: 'production', label: 'Production' }, + ], + }, + ], + }, + }), + { + kind: 'client.capability.interaction_request', + invocationId: 'invocation', + interactionId: 'provider-form-1', + request: { + message: 'Choose a target', + requester: { name: 'deploy', source: 'Fixture' }, + fields: [ + { + kind: 'single_select', + name: 'target', + label: 'Target', + required: true, + options: [ + { value: 'staging', label: 'Staging' }, + { value: 'production', label: 'Production' }, + ], + }, + ], + }, + }, + ); + assert.deepEqual( + decodeHostFrame({ + kind: 'client.capability.interaction_result', + invocationId: 'invocation', + interactionId: 'provider-form-1', + result: { action: 'accept', values: { target: 'staging' } }, + }), + { + kind: 'client.capability.interaction_result', + invocationId: 'invocation', + interactionId: 'provider-form-1', + result: { action: 'accept', values: { target: 'staging' } }, + }, + ); + }); + + test('rejects malformed nested Client Capability interactions at the codec', () => { + assert.throws( + () => + decodeClientFrame({ + kind: 'client.capability.interaction_request', + invocationId: 'invocation', + interactionId: 'provider-form-1', + request: { + message: 'Invalid duplicate fields', + requester: { name: 'fixture' }, + fields: [ + { kind: 'boolean', name: 'same', label: 'First', required: true }, + { kind: 'boolean', name: 'same', label: 'Second', required: true }, + ], + }, + }), + (error: unknown) => error instanceof RuntimeHostProtocolError, + ); + assert.throws( + () => + decodeHostFrame({ + kind: 'client.capability.interaction_result', + invocationId: 'invocation', + interactionId: 'provider-form-1', + result: { action: 'cancel', values: {} }, + }), + (error: unknown) => error instanceof RuntimeHostProtocolError, + ); }); test('keeps Host services open-world and outside model tool offers', () => { diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index baf7270a1b..6f8210429e 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -211,6 +211,10 @@ describe('Runtime Host bootstrap protocol', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 38); }); + test('publishes a new compatibility epoch for nested Client Capability interactions', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 81); + }); + test('publishes a new compatibility epoch for onboarding endpoint overrides', () => { // Epoch 44 peers reject the required `baseUrl` and `connectionId` on // onboarding inputs, and the `base_url_not_configured` / diff --git a/packages/runtime-host/src/client/client-capability-channel.ts b/packages/runtime-host/src/client/client-capability-channel.ts index 5ed354567e..4467f961ae 100644 --- a/packages/runtime-host/src/client/client-capability-channel.ts +++ b/packages/runtime-host/src/client/client-capability-channel.ts @@ -18,9 +18,11 @@ */ import { randomUUID } from 'node:crypto'; +import type { InteractionFormInput, InteractionFormResult } from '@maka/core/interaction'; import { CLIENT_CAPABILITY_MAX_RESULT_BYTES, CLIENT_CAPABILITY_RESULT_CHUNK_MAX_BYTES, + decodeClientCapabilityClientFrame, decodeClientCapabilityReplaceInput, decodeClientCapabilityResult, type ClientCapabilityAdmissionEvidence, @@ -45,6 +47,7 @@ interface ClientCapabilityRegistration { interface ClientCapabilityInvocation { readonly controller: AbortController; admission?: ClientCapabilityAdmission; + interaction?: ClientCapabilityPendingInteraction; released: boolean; } @@ -54,6 +57,20 @@ interface ClientCapabilityAdmission { reject(error: unknown): boolean; } +interface ClientCapabilityPendingInteraction { + readonly interactionId: string; + readonly promise: Promise< + Extract['result'] + >; + resolve( + result: Extract< + ClientCapabilityHostFrame, + { kind: 'client.capability.interaction_result' } + >['result'], + ): boolean; + reject(error: unknown): boolean; +} + export interface ClientCapabilityChannelOptions { readonly write: (frame: ClientCapabilityClientFrame) => Promise; readonly replace: ( @@ -163,6 +180,7 @@ export class ClientCapabilityChannel { new DOMException('Client Capability invocation was cancelled', 'AbortError'), ); invocation.admission?.reject(capabilityInvocationAbortReason(invocation)); + invocation.interaction?.reject(capabilityInvocationAbortReason(invocation)); return; } case 'client.capability.release': { @@ -173,6 +191,7 @@ export class ClientCapabilityChannel { new DOMException('Client Capability invocation was released', 'AbortError'), ); invocation.admission?.reject(capabilityInvocationAbortReason(invocation)); + invocation.interaction?.reject(capabilityInvocationAbortReason(invocation)); this.#invocations.delete(frame.invocationId); return; } @@ -187,6 +206,18 @@ export class ClientCapabilityChannel { } return; } + case 'client.capability.interaction_result': { + const invocation = this.#invocations.get(frame.invocationId); + const interaction = invocation?.interaction; + if ( + !interaction || + interaction.interactionId !== frame.interactionId || + !interaction.resolve(frame.result) + ) { + throw new Error('Runtime Host returned an unmatched capability interaction result'); + } + return; + } } } @@ -197,6 +228,7 @@ export class ClientCapabilityChannel { invocation.released = true; invocation.controller.abort(error); invocation.admission?.reject(error); + invocation.interaction?.reject(error); } this.#invocations.clear(); const providers = new Set( @@ -285,6 +317,7 @@ export class ClientCapabilityChannel { readonly signal: AbortSignal; accept(evidence: ClientCapabilityAdmissionEvidence): Promise; progress(current: number, total: number): void; + requestInteraction(form: InteractionFormInput): Promise; }) => Promise>, ): Promise { let accepted = false; @@ -329,14 +362,52 @@ export class ClientCapabilityChannel { }) .catch((error: unknown) => this.#options.onFailure(asError(error))); }; + const requestInteraction = async ( + request: InteractionFormInput, + ): Promise => { + if (!accepted) { + throw new Error('Client Capability interaction requires an admitted invocation'); + } + if (invocation.released) throw capabilityInvocationAbortReason(invocation); + if (invocation.interaction) { + throw new Error('Client Capability invocation already has a pending interaction'); + } + const interactionId = randomUUID(); + const frame = decodeClientCapabilityClientFrame({ + kind: 'client.capability.interaction_request', + invocationId, + interactionId, + request, + }); + if (frame.kind !== 'client.capability.interaction_request') { + throw new Error('Client Capability interaction request was not canonical'); + } + const interaction = createClientCapabilityPendingInteraction(interactionId); + invocation.interaction = interaction; + try { + try { + await this.#options.write(frame); + } catch (error) { + interaction.reject(error); + throw error; + } + return await interaction.promise; + } finally { + if (invocation.interaction === interaction) invocation.interaction = undefined; + } + }; const result = decodeClientCapabilityResult( await execute({ signal: invocation.controller.signal, accept, progress, + requestInteraction, }), ); if (invocation.released) return; + if (invocation.interaction) { + throw new Error('Client Capability provider returned with a pending interaction'); + } await accept({ kind: 'none' }); await this.#sendResult(invocationId, result, invocation); } catch (error) { @@ -448,6 +519,37 @@ function createClientCapabilityAdmission(): ClientCapabilityAdmission { }; } +function createClientCapabilityPendingInteraction( + interactionId: string, +): ClientCapabilityPendingInteraction { + let state: 'pending' | 'resolved' | 'rejected' = 'pending'; + let resolvePromise!: (result: InteractionFormResult) => void; + let rejectPromise!: (error: unknown) => void; + const promise = new Promise((resolve, reject) => { + resolvePromise = resolve; + rejectPromise = reject; + }); + // A buggy provider may start but not await the callback. The invocation still + // fails closed, while channel teardown must not create an unhandled rejection. + void promise.catch(() => undefined); + return { + interactionId, + promise, + resolve: (result) => { + if (state !== 'pending') return false; + state = 'resolved'; + resolvePromise(result); + return true; + }, + reject: (error) => { + if (state !== 'pending') return false; + state = 'rejected'; + rejectPromise(error); + return true; + }, + }; +} + function capabilityInvocationAbortReason(invocation: ClientCapabilityInvocation): unknown { return ( invocation.controller.signal.reason ?? diff --git a/packages/runtime-host/src/client/client-capability.ts b/packages/runtime-host/src/client/client-capability.ts index e91ff2cf58..f0a4946333 100644 --- a/packages/runtime-host/src/client/client-capability.ts +++ b/packages/runtime-host/src/client/client-capability.ts @@ -17,6 +17,7 @@ * under the License. */ +import type { InteractionFormInput, InteractionFormResult } from '@maka/core/interaction'; import type { ClientCapabilityCallFrame, ClientCapabilityCallResult, @@ -38,6 +39,8 @@ export interface ClientCapabilityProvider { accept(evidence: ClientCapabilityAdmissionEvidence): Promise; /** Publish bounded live progress after admission. */ progress?(current: number, total: number): void; + /** Request one Host-owned form after the invocation is admitted. */ + requestInteraction(form: InteractionFormInput): Promise; }, ): Promise; callService?( diff --git a/packages/runtime-host/src/protocol/client-capability.ts b/packages/runtime-host/src/protocol/client-capability.ts index ce51734621..404f6eac43 100644 --- a/packages/runtime-host/src/protocol/client-capability.ts +++ b/packages/runtime-host/src/protocol/client-capability.ts @@ -18,6 +18,12 @@ */ import { TOOL_ACTIVITY_KINDS, type ToolActivityKind } from '@maka/core/events'; +import { + decodeInteractionAnswer, + projectInteractionFormRequest, + type InteractionFormInput, + type InteractionFormResult, +} from '@maka/core/interaction'; import { assertExactKeys, requireCount, @@ -177,13 +183,21 @@ export interface ClientCapabilityAdmittedFrame { readonly invocationId: string; } +export interface ClientCapabilityInteractionResultFrame { + readonly kind: 'client.capability.interaction_result'; + readonly invocationId: string; + readonly interactionId: string; + readonly result: InteractionFormResult; +} + export type ClientCapabilityHostFrame = | ClientCapabilityCallFrame | ClientCapabilityServiceCallFrame | ClientCapabilityCancelFrame | ClientCapabilityReleaseFrame | ClientCapabilityRegistrationReleaseFrame - | ClientCapabilityAdmittedFrame; + | ClientCapabilityAdmittedFrame + | ClientCapabilityInteractionResultFrame; export interface ClientCapabilityAcceptedFrame { readonly kind: 'client.capability.accepted'; @@ -234,6 +248,13 @@ export interface ClientCapabilityResultChunkFrame { readonly data: string; } +export interface ClientCapabilityInteractionRequestFrame { + readonly kind: 'client.capability.interaction_request'; + readonly invocationId: string; + readonly interactionId: string; + readonly request: InteractionFormInput; +} + export type ClientCapabilityClientFrame = | ClientCapabilityAcceptedFrame | ClientCapabilityRejectedFrame @@ -241,7 +262,8 @@ export type ClientCapabilityClientFrame = | ClientCapabilityProgressFrame | ClientCapabilityResultFrame | ClientCapabilityResultStartFrame - | ClientCapabilityResultChunkFrame; + | ClientCapabilityResultChunkFrame + | ClientCapabilityInteractionRequestFrame; export const CLIENT_CAPABILITY_OPERATION_SPECS = { 'client.capability.replace': defineHostPathOperation< @@ -485,6 +507,19 @@ export function decodeClientCapabilityClientFrame(value: unknown): ClientCapabil data, }; } + case 'client.capability.interaction_request': + assertExactKeys(frame, 'Client Capability interaction request frame', [ + 'kind', + 'invocationId', + 'interactionId', + 'request', + ]); + return { + kind: frame.kind, + invocationId: requireEntityId(frame.invocationId, 'invocationId'), + interactionId: requireEntityId(frame.interactionId, 'interactionId'), + request: decodeClientCapabilityFormRequest(frame.request), + }; default: throw invalidProtocolFrame('Invalid Client Capability client frame kind'); } @@ -583,6 +618,19 @@ export function decodeClientCapabilityHostFrame(value: unknown): ClientCapabilit kind: frame.kind, invocationId: requireEntityId(frame.invocationId, 'invocationId'), }; + case 'client.capability.interaction_result': + assertExactKeys(frame, 'Client Capability interaction result frame', [ + 'kind', + 'invocationId', + 'interactionId', + 'result', + ]); + return { + kind: frame.kind, + invocationId: requireEntityId(frame.invocationId, 'invocationId'), + interactionId: requireEntityId(frame.interactionId, 'interactionId'), + result: decodeClientCapabilityFormResult(frame.result), + }; case 'client.capability.registration_release': assertExactKeys(frame, 'Client Capability registration release frame', [ 'kind', @@ -614,6 +662,44 @@ export function decodeClientCapabilityResult(value: unknown): ClientCapabilityCa }; } +function decodeClientCapabilityFormRequest(value: unknown): InteractionFormInput { + const record = requireExactRecord(value, 'Client Capability form request', [ + 'message', + 'requester', + 'fields', + ]); + let request: ReturnType; + try { + request = projectInteractionFormRequest({ + toolUseId: 'client-capability-interaction', + message: record.message as string, + requester: record.requester as InteractionFormInput['requester'], + fields: record.fields as InteractionFormInput['fields'], + }); + } catch { + throw invalidProtocolFrame('Invalid Client Capability form request'); + } + return { + message: request.message, + requester: request.requester, + fields: request.fields, + }; +} + +function decodeClientCapabilityFormResult(value: unknown): InteractionFormResult { + const record = requireRecord(value, 'Client Capability form result'); + let answer: ReturnType; + try { + answer = decodeInteractionAnswer({ kind: 'form', ...record }); + } catch { + throw invalidProtocolFrame('Invalid Client Capability form result'); + } + if (answer.kind !== 'form') throw invalidProtocolFrame('Invalid Client Capability form result'); + return answer.action === 'accept' + ? { action: 'accept', values: answer.values } + : { action: answer.action }; +} + function decodeClientCapabilityOffer(value: unknown): ClientCapabilityOffer { const record = requireRecord(value, 'Client Capability offer'); assertOptionalExactKeys( @@ -1131,6 +1217,7 @@ const CLIENT_CAPABILITY_CLIENT_FRAME_KINDS = new Set([ @@ -1140,4 +1227,5 @@ const CLIENT_CAPABILITY_HOST_FRAME_KINDS = new Set Date: Tue, 1 Sep 2026 01:44:09 +0800 Subject: [PATCH 3/9] 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 + ...e-host-capability-provider-command.test.ts | 1 + .../client-capability-channel.test.ts | 4 +- ...ient-capability-interaction-broker.test.ts | 397 ++++++++++++++++++ .../server/client-capability-coordinator.ts | 9 +- .../client-capability-invocation-broker.ts | 320 ++++++++++++-- .../runtime/src/__tests__/mcp-tools.test.ts | 38 ++ packages/runtime/src/mcp-tools.ts | 13 + 9 files changed, 742 insertions(+), 43 deletions(-) create mode 100644 packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts 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 da5c76d48b..65ee0b6da9 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 @@ -1214,6 +1214,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 98d97e5ba9..89658d460a 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 @@ -602,6 +602,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; @@ -723,5 +724,6 @@ async function call( return provider.call(frame, { signal: new AbortController().signal, accept: async (evidence) => accept(evidence), + 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 6df9f8bd49..e5cf430c3e 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 @@ -149,6 +149,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-channel.test.ts b/packages/runtime-host/src/__tests__/client-capability-channel.test.ts index ff69d4a4e2..337c27e998 100644 --- a/packages/runtime-host/src/__tests__/client-capability-channel.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-channel.test.ts @@ -318,7 +318,7 @@ test('Client Capability channel correlates one admitted nested form before the f }, ], call: async (_frame, options) => { - await options.accept(); + await options.accept({ kind: 'none' }); const answer = await options.requestInteraction({ message: 'Choose a target', requester: { name: 'deploy', source: 'Fixture' }, @@ -422,7 +422,7 @@ test('Client Capability release rejects a pending nested form', async () => { }, ], call: async (_frame, options) => { - await options.accept(); + await options.accept({ kind: 'none' }); try { await options.requestInteraction({ message: 'Choose a target', diff --git a/packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts b/packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts new file mode 100644 index 0000000000..b127b290cb --- /dev/null +++ b/packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts @@ -0,0 +1,397 @@ +/* + * 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, + admissionEvidence: { kind: 'none' }, + }); + await flush(); + 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; + 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, + admissionEvidence: { kind: 'none' }, + }); + await flush(); + 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((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((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, + admissionEvidence: { kind: 'none' }, + }); + await flush(); + 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((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((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, + admissionEvidence: { kind: 'none' }, + }); + await flush(); + 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((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((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, + admissionEvidence: { kind: 'none' }, + }); + await flush(); + 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 { + 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 315c440086..d68f33c511 100644 --- a/packages/runtime-host/src/server/client-capability-coordinator.ts +++ b/packages/runtime-host/src/server/client-capability-coordinator.ts @@ -785,16 +785,16 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService } releaseConnection(connectionId: string): Promise { - 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; 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 { @@ -1052,6 +1052,8 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService options.context, options.signal, options.timeoutMs ?? DEFAULT_CALL_TIMEOUT_MS, + undefined, + options.requestInteraction, ); try { const evidence = await prepared.waitUntilAccepted(); @@ -1108,6 +1110,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 a4c7db15b3..045d3d6256 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, @@ -81,11 +82,19 @@ interface InvocationState void; onProgress?: (current: number, total: number) => void; + readonly requestInteraction?: ClientCapabilityInteractionHandler; readonly timeoutMs: number; readonly providerAvailability: AbortController; - timer: NodeJS.Timeout | undefined; + cancelTimer?: () => void; + interaction?: InvocationInteraction; acceptedSettled: boolean; - phase: 'dispatched' | 'accepted' | 'admitted' | 'chunks'; + phase: + | 'dispatched' + | 'accepted' + | 'admitted' + | 'awaiting_interaction' + | 'delivering_interaction_result' + | 'chunks'; progress?: { current: number; total: number }; chunks?: { readonly byteLength: number; @@ -95,6 +104,19 @@ interface InvocationState; + readonly resolveDone: () => void; + terminal?: { readonly error: Error; readonly releaseRemote: boolean }; +} + +type ClientCapabilityInteractionHandler = ( + form: InteractionFormInput, + options?: { readonly cancellationSignal?: AbortSignal }, +) => Promise; + export interface PreparedClientCapabilityInvocation { readonly invocationId: string; /** Resolves once the provider has parsed the call and is waiting at its admission cut. */ @@ -112,6 +134,7 @@ export interface ClientCapabilityInvocationBrokerOptions< > { readonly senderFor: (connectionId: string) => ClientCapabilityConnectionSender | undefined; readonly onRegistrationIdle: (registration: Registration) => void; + readonly scheduleTimeout?: (callback: () => void, timeoutMs: number) => () => void; } export class ClientCapabilityInvocationBroker< @@ -119,12 +142,21 @@ export class ClientCapabilityInvocationBroker< > { readonly #senderFor: ClientCapabilityInvocationBrokerOptions['senderFor']; readonly #onRegistrationIdle: ClientCapabilityInvocationBrokerOptions['onRegistrationIdle']; + readonly #scheduleTimeout: NonNullable< + ClientCapabilityInvocationBrokerOptions['scheduleTimeout'] + >; readonly #invocations = new Map>(); readonly #retiredInvocationIds = new Set(); constructor(options: ClientCapabilityInvocationBrokerOptions) { this.#senderFor = options.senderFor; this.#onRegistrationIdle = options.onRegistrationIdle; + this.#scheduleTimeout = + options.scheduleTimeout ?? + ((callback, timeoutMs) => { + const timer = setTimeout(callback, timeoutMs); + return () => clearTimeout(timer); + }); } async invoke( @@ -135,6 +167,7 @@ export class ClientCapabilityInvocationBroker< signal: AbortSignal | undefined, timeoutMs: number, onProgress?: (current: number, total: number) => void, + requestInteraction?: ClientCapabilityInteractionHandler, ): Promise { const prepared = this.prepare( registration, @@ -144,6 +177,7 @@ export class ClientCapabilityInvocationBroker< signal, timeoutMs, onProgress, + requestInteraction, ); await prepared.waitUntilAccepted(); return prepared.admit(); @@ -157,8 +191,15 @@ export class ClientCapabilityInvocationBroker< signal: AbortSignal | undefined, timeoutMs: number, onProgress?: (current: number, total: number) => void, + requestInteraction?: ClientCapabilityInteractionHandler, ): PreparedClientCapabilityInvocation { - return this.#prepare(registration, signal, timeoutMs, onProgress, (invocationId) => ({ + return this.#prepare( + registration, + signal, + timeoutMs, + onProgress, + requestInteraction, + (invocationId) => ({ kind: 'client.capability.call', invocationId, registrationId: registration.registrationId, @@ -170,7 +211,8 @@ export class ClientCapabilityInvocationBroker< turnId: context.turnId, toolCallId: context.toolCallId, ...(binding.hostPathAccess === 'cwd' ? { cwd: context.cwd } : {}), - })); + }), + ); } async invokeService( @@ -204,7 +246,7 @@ export class ClientCapabilityInvocationBroker< signal: AbortSignal | undefined, timeoutMs: number, ): PreparedClientCapabilityInvocation { - return this.#prepare(registration, signal, timeoutMs, undefined, (invocationId) => ({ + return this.#prepare(registration, signal, timeoutMs, undefined, undefined, (invocationId) => ({ kind: 'client.capability.service_call', invocationId, registrationId: registration.registrationId, @@ -220,6 +262,7 @@ export class ClientCapabilityInvocationBroker< signal: AbortSignal | undefined, timeoutMs: number, onProgress: ((current: number, total: number) => void) | undefined, + requestInteraction: ClientCapabilityInteractionHandler | undefined, frameFor: (invocationId: string) => ClientCapabilityHostFrame, ): PreparedClientCapabilityInvocation { const sender = this.#senderFor(registration.connectionId); @@ -253,16 +296,17 @@ 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' || invocation.phase === 'accepted' ? asError(abortReason(signal)) : new ToolOutcomeUnknownError( 'Client Capability invocation was cancelled after admission', - ), - true, - ); + ); + if (invocation.interaction) { + this.#terminateInteraction(invocation, error, true, true); + } else { + this.#settle(invocation, undefined, error, true); + } } : undefined; const invocation: InvocationState = { @@ -275,14 +319,14 @@ export class ClientCapabilityInvocationBroker< signal, onAbort, onProgress, + requestInteraction, timeoutMs, providerAvailability: new AbortController(), - timer: undefined, acceptedSettled: false, phase: 'dispatched', }; this.#invocations.set(invocationId, invocation); - this.#startTimeout(invocation); + this.#armTimer(invocation); if (onAbort) signal?.addEventListener('abort', onAbort, { once: true }); void sender.send(frameFor(invocationId)).catch(() => { const current = this.#invocations.get(invocationId); @@ -314,7 +358,7 @@ export class ClientCapabilityInvocationBroker< if (invocation.phase === 'accepted') { invocation.onProgress = onProgress ?? invocation.onProgress; invocation.phase = 'admitted'; - this.#startTimeout(invocation); + this.#armTimer(invocation); const currentSender = this.#senderFor(invocation.registration.connectionId); if (!currentSender) { this.#settle( @@ -380,8 +424,7 @@ export class ClientCapabilityInvocationBroker< throw new Error('Client Capability invocation was accepted more than once'); } invocation.phase = 'accepted'; - if (invocation.timer) clearTimeout(invocation.timer); - invocation.timer = undefined; + this.#clearTimer(invocation); invocation.acceptedSettled = true; invocation.resolveAccepted(frame.admissionEvidence); return; @@ -398,18 +441,31 @@ export class ClientCapabilityInvocationBroker< ); return; case 'client.capability.failed': - if (invocation.phase !== 'admitted' && invocation.phase !== 'chunks') { + if (invocation.phase === 'dispatched' || invocation.phase === 'accepted') { throw new Error('Client Capability failure arrived before admission'); } - 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 !== 'admitted' && invocation.phase !== 'chunks') { + if ( + invocation.phase !== 'admitted' && + invocation.phase !== 'delivering_interaction_result' && + invocation.phase !== 'chunks' + ) { throw new Error('Client Capability progress arrived before admission'); } if ( @@ -422,14 +478,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 !== 'admitted') { + 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 !== 'admitted' && + invocation.phase !== 'delivering_interaction_result' + ) { throw new Error('Client Capability result arrived outside the admitted phase'); } this.#settle(invocation, frame.result, undefined, true); return; case 'client.capability.result_start': - if (invocation.phase !== 'admitted') { + 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 !== 'admitted' && + invocation.phase !== 'delivering_interaction_result' + ) { throw new Error('Client Capability result chunks started outside the admitted phase'); } invocation.phase = 'chunks'; @@ -445,7 +534,8 @@ export class ClientCapabilityInvocationBroker< } } - releaseConnection(connectionId: string): void { + async releaseConnection(connectionId: string): Promise { + const interactions: Promise[] = []; for (const invocation of [...this.#invocations.values()]) { if (invocation.registration.connectionId !== connectionId) continue; if (invocation.phase === 'dispatched' || invocation.phase === 'accepted') { @@ -456,9 +546,7 @@ export class ClientCapabilityInvocationBroker< ), ); } - this.#settle( - invocation, - undefined, + const error = invocation.phase === 'dispatched' || invocation.phase === 'accepted' ? new ClientCapabilityInvocationError( 'capability_lost', @@ -466,10 +554,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 { @@ -514,11 +607,157 @@ export class ClientCapabilityInvocationBroker< this.#settle(invocation, decodeClientCapabilityResult(decoded), undefined, true); } - #startTimeout(invocation: InvocationState): void { - if (invocation.timer) clearTimeout(invocation.timer); - invocation.timer = setTimeout(() => { + #acceptInteraction( + invocation: InvocationState, + interactionId: string, + request: InteractionFormInput, + ): void { + if ( + (invocation.phase !== 'admitted' && + 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 admitted 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((resolve) => { + resolveDone = resolve; + }), + resolveDone: () => resolveDone(), + }; + invocation.interaction = interaction; + void this.#runInteraction(invocation, interaction, request); + } + + async #runInteraction( + invocation: InvocationState, + interaction: InvocationInteraction, + request: InteractionFormInput, + ): Promise { + 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 = interaction.terminal; + 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 = 'admitted'; + 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, + 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, + interaction: InvocationInteraction, + ): boolean { + return ( + this.#invocations.get(invocation.invocationId) === invocation && + invocation.interaction === interaction + ); + } + + #armTimer(invocation: InvocationState): void { + this.#clearTimer(invocation); + invocation.cancelTimer = this.#scheduleTimeout(() => { const current = this.#invocations.get(invocation.invocationId); - if (!current) return; + if (current !== invocation) return; const sender = this.#senderFor(current.registration.connectionId); void sender ?.send({ kind: 'client.capability.cancel', invocationId: current.invocationId }) @@ -537,6 +776,11 @@ export class ClientCapabilityInvocationBroker< }, invocation.timeoutMs); } + #clearTimer(invocation: InvocationState): void { + invocation.cancelTimer?.(); + invocation.cancelTimer = undefined; + } + #settle( invocation: InvocationState, result: ClientCapabilityCallResult | undefined, @@ -545,7 +789,7 @@ export class ClientCapabilityInvocationBroker< ): void { if (this.#invocations.get(invocation.invocationId) !== invocation) return; this.#invocations.delete(invocation.invocationId); - if (invocation.timer) clearTimeout(invocation.timer); + this.#clearTimer(invocation); if (invocation.onAbort && invocation.signal) { invocation.signal.removeEventListener('abort', invocation.onAbort); } 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 a7c48941db..c3d53fa419 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 { PermissionMode, ToolCategory } from '@maka/core/permission'; import type { ExecutionBoundary } from '@maka/core/sandbox-boundary'; import type { ToolRecoveryMode } from '@maka/core/runtime-event'; @@ -66,6 +67,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; } export interface McpToolInvocationContext { @@ -171,6 +176,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), From 94576f18317834266b4434c50d0922dfbd6091ba Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 1 Sep 2026 16:14:31 +0800 Subject: [PATCH 4/9] fix(runtime-host): order nested form cleanup --- .../client-capability-coordinator.test.ts | 74 +++++++++++++++++++ .../server/client-capability-coordinator.ts | 17 +++-- .../tool-runtime-form-interaction.test.ts | 61 +++++++++++++++ packages/runtime/src/tool-runtime.ts | 15 +++- 4 files changed, 158 insertions(+), 9 deletions(-) diff --git a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts index c34e4611f8..039c08c480 100644 --- a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts @@ -1900,6 +1900,80 @@ test('service-only registration lifecycle does not invalidate model backends', a assert.equal(modelToolChanges, 2); }); +test('close waits for nested Client Capability interaction cleanup', async () => { + const coordinator = createCoordinator(); + let connection!: ClientCapabilityConnection; + let interactionStarted!: () => void; + const started = new Promise((resolve) => { + interactionStarted = resolve; + }); + let finishCleanup!: () => void; + const cleanup = new Promise((resolve) => { + finishCleanup = resolve; + }); + connection = coordinator.attachConnection(clientCapabilityConnectionIdentity('connection-a'), { + send: async (frame) => { + if (frame.kind === 'client.capability.call') { + connection.accept({ + kind: 'client.capability.accepted', + invocationId: frame.invocationId, + admissionEvidence: { kind: 'none' }, + }); + } else if (frame.kind === 'client.capability.admitted') { + connection.accept({ + kind: 'client.capability.interaction_request', + invocationId: frame.invocationId, + interactionId: 'interaction-a', + request: { + message: 'Choose a target', + requester: { name: 'deploy' }, + fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }], + }, + }); + } + }, + }); + await replace(coordinator, 'connection-a', 'registration-a', 'deploy'); + assert.deepEqual(await coordinator.bindSession('session-a', 'connection-a'), { ok: true }); + const snapshot = coordinator.snapshotForSession('session-a'); + assert.ok(snapshot); + const call = Promise.resolve( + snapshot.tools[0]!.impl({}, { + sessionId: 'session-a', + turnId: 'turn-a', + cwd: '/tmp', + toolCallId: 'tool-call-a', + abortSignal: new AbortController().signal, + emitOutput: () => undefined, + requestUserForm: async (_form, options) => { + interactionStarted(); + const signal = options?.cancellationSignal; + assert.ok(signal); + if (!signal.aborted) { + await new Promise((resolve) => + signal.addEventListener('abort', () => resolve(), { once: true }), + ); + } + await cleanup; + throw signal.reason; + }, + }), + ); + void call.catch(() => undefined); + await started; + snapshot.release(); + + let closed = false; + const closing = coordinator.close().then(() => { + closed = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(closed, false); + finishCleanup(); + await closing; + await assert.rejects(call, ToolOutcomeUnknownError); +}); + async function invoke(tool: NonNullable>): Promise { return tool.impl( {}, diff --git a/packages/runtime-host/src/server/client-capability-coordinator.ts b/packages/runtime-host/src/server/client-capability-coordinator.ts index d68f33c511..9b83c2cfa4 100644 --- a/packages/runtime-host/src/server/client-capability-coordinator.ts +++ b/packages/runtime-host/src/server/client-capability-coordinator.ts @@ -789,12 +789,15 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService const connection = this.#connections.get(connectionId); if (!connection) return invocationCleanup; let task!: Promise; - task = this.#activation - .runMutation(() => this.#releaseConnectionState(connection)) + task = Promise.all([ + invocationCleanup, + this.#activation.runMutation(() => this.#releaseConnectionState(connection)), + ]) + .then(() => undefined) .finally(() => this.#pendingConnectionReleases.delete(task)); this.#pendingConnectionReleases.add(task); void task.catch(() => undefined); - return Promise.all([invocationCleanup, task]).then(() => undefined); + return task; } #releaseConnectionState(connection: ClientProviderConnection): void { @@ -827,10 +830,10 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService async close(): Promise { this.beginDrain(); - for (const connectionId of [...this.#connections.keys()]) { - this.releaseConnection(connectionId); - } - await Promise.allSettled([...this.#pendingConnectionReleases]); + const releases = [...this.#connections.keys()].map((connectionId) => + this.releaseConnection(connectionId), + ); + await Promise.allSettled(releases); this.#invocations.close(); this.#sessions.clear(); } diff --git a/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts b/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts index 72653b191a..8959f36e04 100644 --- a/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts +++ b/packages/runtime/src/__tests__/tool-runtime-form-interaction.test.ts @@ -255,4 +255,65 @@ describe('ToolRuntime form Interaction', () => { assert.equal(toolRuntime.pendingUserFormCount(), 0); assert.equal(events.filter((event) => event.type === 'form_answer_ack').length, 0); }); + + test('waits for hosted admission before withdrawing a cancelled producer form', async () => { + const events: SessionEvent[] = []; + const producer = new AbortController(); + let admitted: { requestId: string; settlement: HostedFormSettlement } | undefined; + let finishAdmission!: () => void; + const admissionGate = new Promise((resolve) => { + finishAdmission = resolve; + }); + const withdrawals: string[] = []; + const toolRuntime = createTestToolRuntime({ + sessionId: 'session-1', + header: header(), + connection: { providerType: 'openai', slug: 'c' } as never, + modelId: 'm', + appendMessage: async () => {}, + newId: (() => { + let id = 0; + return () => `id-${++id}`; + })(), + now: () => 1, + getPermissionPauseTarget: () => null, + hostedInteraction: { + sessionId: 'session-1', + turnId: 'turn-1', + runId: 'run-1', + admitUserQuestionRequest: async () => { + throw new Error('Unexpected user question'); + }, + admitFormRequest: async (input) => { + admitted = { requestId: input.request.requestId, settlement: input.settlement }; + await admissionGate; + }, + withdrawFormRequest: async (requestId) => { + withdrawals.push(requestId); + await admitted?.settlement.applyClosure('producer_cancelled'); + }, + admitSandboxBoundaryRequest: async () => { + throw new Error('Unexpected sandbox boundary'); + }, + }, + }); + const pending = toolRuntime.settleToolCall({ + tool: formTool(producer.signal), + turnId: 'turn-1', + toolCallId: 'tool-1', + input: {}, + abortSignal: new AbortController().signal, + eventSink: sink(events), + }); + while (!admitted) await new Promise((resolve) => setImmediate(resolve)); + + producer.abort(new DOMException('Provider invocation ended', 'AbortError')); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(withdrawals, []); + finishAdmission(); + await pending; + + assert.deepEqual(withdrawals, [admitted.requestId]); + assert.equal(toolRuntime.pendingUserFormCount(), 0); + }); }); diff --git a/packages/runtime/src/tool-runtime.ts b/packages/runtime/src/tool-runtime.ts index 326ba4d550..a12fdce73e 100644 --- a/packages/runtime/src/tool-runtime.ts +++ b/packages/runtime/src/tool-runtime.ts @@ -2661,11 +2661,19 @@ export class ToolRuntime { this.userForms.reject(requestId, abortErrorFromSignal(abortSignal)); this.finishDeferredFormTurnClosure(); }; + let hostedAdmission: Promise | undefined; let producerWithdrawal: Promise | undefined; const onProducerCancellation = (): void => { if (abortSignal.aborted) return; if (hostedRun) { - producerWithdrawal ??= hostedRun.withdrawFormRequest(requestId); + producerWithdrawal ??= Promise.resolve().then(async () => { + try { + await hostedAdmission; + } catch { + return; + } + await hostedRun.withdrawFormRequest(requestId); + }); } else if (producerCancellationSignal) { this.userForms.reject(requestId, abortErrorFromSignal(producerCancellationSignal)); this.finishDeferredFormTurnClosure(); @@ -2688,7 +2696,10 @@ export class ToolRuntime { }; if (hostedRun) { const settlement = this.createFormSettlement(turnId, requestId); - const admission = hostedRun.admitFormRequest({ request: requestEvent, settlement }); + const admission = Promise.resolve().then(() => + hostedRun.admitFormRequest({ request: requestEvent, settlement }), + ); + hostedAdmission = admission; try { await racePromiseWithAbort(admission, interactionSignal); } catch (error) { From 02ebd57428eee6cde1f3415eace573c639736fa6 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 1 Sep 2026 16:16:06 +0800 Subject: [PATCH 5/9] test(desktop): complete capability interaction fake --- apps/desktop/src/main/__tests__/browser-tools.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/desktop/src/main/__tests__/browser-tools.test.ts b/apps/desktop/src/main/__tests__/browser-tools.test.ts index 98edf823c8..d5e6736ea7 100644 --- a/apps/desktop/src/main/__tests__/browser-tools.test.ts +++ b/apps/desktop/src/main/__tests__/browser-tools.test.ts @@ -287,6 +287,7 @@ describe('browser tool execution', () => { { signal: new AbortController().signal, accept: async () => undefined, + requestInteraction: async () => assert.fail('Unexpected provider interaction'), }, ); assert.equal(resolved, 2); From f2ad0c96bd5c3b790a5c46c4d8e6a0abdcbfe8c9 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 1 Sep 2026 16:18:02 +0800 Subject: [PATCH 6/9] fix(runtime-host): forward forms after capability admission --- .../server/client-capability-coordinator.ts | 8 ++-- .../client-capability-invocation-broker.ts | 10 +++-- .../runtime/src/__tests__/mcp-tools.test.ts | 43 +++++++++++++++++++ packages/runtime/src/mcp-tools.ts | 7 +++ 4 files changed, 61 insertions(+), 7 deletions(-) diff --git a/packages/runtime-host/src/server/client-capability-coordinator.ts b/packages/runtime-host/src/server/client-capability-coordinator.ts index 9b83c2cfa4..6aa5014cc7 100644 --- a/packages/runtime-host/src/server/client-capability-coordinator.ts +++ b/packages/runtime-host/src/server/client-capability-coordinator.ts @@ -1055,8 +1055,6 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService options.context, options.signal, options.timeoutMs ?? DEFAULT_CALL_TIMEOUT_MS, - undefined, - options.requestInteraction, ); try { const evidence = await prepared.waitUntilAccepted(); @@ -1074,7 +1072,8 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService ); if (!target) { return { - execute: ({ emitProgress } = {}) => prepared.admit(emitProgress), + execute: ({ emitProgress, requestInteraction } = {}) => + prepared.admit(emitProgress, requestInteraction), cancel: () => prepared.cancel(), }; } @@ -1095,7 +1094,8 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService } } return { - execute: ({ emitProgress } = {}) => prepared.admit(emitProgress), + execute: ({ emitProgress, requestInteraction } = {}) => + prepared.admit(emitProgress, requestInteraction), cancel: () => prepared.cancel(), }; } catch (error) { 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 045d3d6256..80b29c092b 100644 --- a/packages/runtime-host/src/server/client-capability-invocation-broker.ts +++ b/packages/runtime-host/src/server/client-capability-invocation-broker.ts @@ -82,7 +82,7 @@ interface InvocationState void; onProgress?: (current: number, total: number) => void; - readonly requestInteraction?: ClientCapabilityInteractionHandler; + requestInteraction?: ClientCapabilityInteractionHandler; readonly timeoutMs: number; readonly providerAvailability: AbortController; cancelTimer?: () => void; @@ -122,7 +122,10 @@ export interface PreparedClientCapabilityInvocation { /** Resolves once the provider has parsed the call and is waiting at its admission cut. */ waitUntilAccepted(): Promise; /** Crosses the admission cut and returns the provider result. */ - admit(onProgress?: (current: number, total: number) => void): Promise; + admit( + onProgress?: (current: number, total: number) => void, + requestInteraction?: ClientCapabilityInteractionHandler, + ): Promise; /** Cancels an accepted call that will not cross the admission cut. */ cancel(): void; /** Aborts when the provider connection disappears before this call is admitted. */ @@ -351,12 +354,13 @@ export class ClientCapabilityInvocationBroker< providerSignal: this.#invocations.get(invocationId)?.providerAvailability.signal ?? AbortSignal.abort(), waitUntilAccepted: () => accepted, - admit: async (onProgress) => { + admit: async (onProgress, requestInteraction) => { await accepted; const invocation = this.#invocations.get(invocationId); if (!invocation) return result; if (invocation.phase === 'accepted') { invocation.onProgress = onProgress ?? invocation.onProgress; + invocation.requestInteraction = requestInteraction ?? invocation.requestInteraction; invocation.phase = 'admitted'; this.#armTimer(invocation); const currentSender = this.#senderFor(invocation.registration.connectionId); diff --git a/packages/runtime/src/__tests__/mcp-tools.test.ts b/packages/runtime/src/__tests__/mcp-tools.test.ts index 5769c21995..acb102faac 100644 --- a/packages/runtime/src/__tests__/mcp-tools.test.ts +++ b/packages/runtime/src/__tests__/mcp-tools.test.ts @@ -135,6 +135,49 @@ test('buildMcpTools carries the Runtime-owned form callback to the provider', as ); }); +test('prepared MCP execution receives the Runtime-owned form callback after admission', async () => { + const toolBinding = binding('prepared-form-binding'); + const provider: McpToolProvider = { + toolSnapshot: () => ({ + revision: 1, + tools: [boundTool(descriptor('client', 'deploy'), toolBinding)], + }), + prepareTool: async () => ({ + execute: async (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 }], + }); + assert.deepEqual(answer, { action: 'accept', values: { target: 'staging' } }); + return { content: [] }; + }, + cancel: () => undefined, + }), + callTool: async () => assert.fail('Prepared provider must not use direct callTool'), + }; + const [tool] = buildMcpTools(provider); + assert.ok(tool?.prepareExecution); + const controller = new AbortController(); + const prepared = await tool.prepareExecution({}, { + sessionId: 'session', + turnId: 'turn', + cwd: '/workspace', + toolCallId: 'tool-call', + abortSignal: controller.signal, + }); + await prepared.execute({ + sessionId: 'session', + turnId: 'turn', + cwd: '/workspace', + toolCallId: 'tool-call', + abortSignal: controller.signal, + emitOutput: () => undefined, + requestUserForm: async () => ({ 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 c3d53fa419..13a4ee5225 100644 --- a/packages/runtime/src/mcp-tools.ts +++ b/packages/runtime/src/mcp-tools.ts @@ -58,6 +58,7 @@ export interface McpToolProvider { export interface McpPreparedToolCall { execute(options?: { readonly emitProgress?: (current: number, total: number) => void; + readonly requestInteraction?: McpToolCallOptions['requestInteraction']; }): Promise; cancel(): Promise | void; } @@ -142,6 +143,12 @@ export function buildMcpTools( ...(executionContext.emitProgress ? { emitProgress: executionContext.emitProgress } : {}), + ...(executionContext.requestUserForm + ? { + requestInteraction: (form, interactionOptions) => + executionContext.requestUserForm!(form, interactionOptions), + } + : {}), }), cancel: () => prepared.cancel(), }; From 3f16930f5d8ca80953b9da11c76f8f62a339dee8 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 1 Sep 2026 16:23:00 +0800 Subject: [PATCH 7/9] fix(runtime-host): await prior capability releases --- .../src/__tests__/client-capability-coordinator.test.ts | 5 ++++- .../runtime-host/src/server/client-capability-coordinator.ts | 1 + 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts index 039c08c480..95550ce5c8 100644 --- a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts @@ -1963,6 +1963,9 @@ test('close waits for nested Client Capability interaction cleanup', async () => await started; snapshot.release(); + const connectionClosing = connection.close(); + await new Promise((resolve) => setImmediate(resolve)); + let closed = false; const closing = coordinator.close().then(() => { closed = true; @@ -1970,7 +1973,7 @@ test('close waits for nested Client Capability interaction cleanup', async () => await new Promise((resolve) => setImmediate(resolve)); assert.equal(closed, false); finishCleanup(); - await closing; + await Promise.all([connectionClosing, closing]); await assert.rejects(call, ToolOutcomeUnknownError); }); diff --git a/packages/runtime-host/src/server/client-capability-coordinator.ts b/packages/runtime-host/src/server/client-capability-coordinator.ts index 6aa5014cc7..d836f6d286 100644 --- a/packages/runtime-host/src/server/client-capability-coordinator.ts +++ b/packages/runtime-host/src/server/client-capability-coordinator.ts @@ -834,6 +834,7 @@ export class HostClientCapabilityCoordinator implements ClientCapabilityService this.releaseConnection(connectionId), ); await Promise.allSettled(releases); + await Promise.allSettled([...this.#pendingConnectionReleases]); this.#invocations.close(); this.#sessions.clear(); } From b5fe32ea41b4c8f6e34c7644611370b8eaf7b71a Mon Sep 17 00:00:00 2001 From: me2seeks Date: Wed, 2 Sep 2026 22:04:34 +0800 Subject: [PATCH 8/9] style: format nested capability form files The nested capability form sources predate the formatter rules now on main; rebase onto the current parent and reformat so the changed-file biome gate passes again. No semantic change. --- .../client-capability-coordinator.test.ts | 41 ++++++++++--------- .../client-capability-invocation-broker.ts | 25 ++++++----- .../runtime/src/__tests__/mcp-tools.test.ts | 17 ++++---- 3 files changed, 44 insertions(+), 39 deletions(-) diff --git a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts index 95550ce5c8..8032f806be 100644 --- a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts @@ -1938,26 +1938,29 @@ test('close waits for nested Client Capability interaction cleanup', async () => const snapshot = coordinator.snapshotForSession('session-a'); assert.ok(snapshot); const call = Promise.resolve( - snapshot.tools[0]!.impl({}, { - sessionId: 'session-a', - turnId: 'turn-a', - cwd: '/tmp', - toolCallId: 'tool-call-a', - abortSignal: new AbortController().signal, - emitOutput: () => undefined, - requestUserForm: async (_form, options) => { - interactionStarted(); - const signal = options?.cancellationSignal; - assert.ok(signal); - if (!signal.aborted) { - await new Promise((resolve) => - signal.addEventListener('abort', () => resolve(), { once: true }), - ); - } - await cleanup; - throw signal.reason; + snapshot.tools[0]!.impl( + {}, + { + sessionId: 'session-a', + turnId: 'turn-a', + cwd: '/tmp', + toolCallId: 'tool-call-a', + abortSignal: new AbortController().signal, + emitOutput: () => undefined, + requestUserForm: async (_form, options) => { + interactionStarted(); + const signal = options?.cancellationSignal; + assert.ok(signal); + if (!signal.aborted) { + await new Promise((resolve) => + signal.addEventListener('abort', () => resolve(), { once: true }), + ); + } + await cleanup; + throw signal.reason; + }, }, - }), + ), ); void call.catch(() => undefined); await started; 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 80b29c092b..b8a2669241 100644 --- a/packages/runtime-host/src/server/client-capability-invocation-broker.ts +++ b/packages/runtime-host/src/server/client-capability-invocation-broker.ts @@ -203,17 +203,17 @@ export class ClientCapabilityInvocationBroker< 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 } : {}), + 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 } : {}), }), ); } @@ -617,8 +617,7 @@ export class ClientCapabilityInvocationBroker< request: InteractionFormInput, ): void { if ( - (invocation.phase !== 'admitted' && - invocation.phase !== 'delivering_interaction_result') || + (invocation.phase !== 'admitted' && invocation.phase !== 'delivering_interaction_result') || invocation.interaction ) { if (invocation.interaction) { diff --git a/packages/runtime/src/__tests__/mcp-tools.test.ts b/packages/runtime/src/__tests__/mcp-tools.test.ts index acb102faac..0d020f69f5 100644 --- a/packages/runtime/src/__tests__/mcp-tools.test.ts +++ b/packages/runtime/src/__tests__/mcp-tools.test.ts @@ -160,13 +160,16 @@ test('prepared MCP execution receives the Runtime-owned form callback after admi const [tool] = buildMcpTools(provider); assert.ok(tool?.prepareExecution); const controller = new AbortController(); - const prepared = await tool.prepareExecution({}, { - sessionId: 'session', - turnId: 'turn', - cwd: '/workspace', - toolCallId: 'tool-call', - abortSignal: controller.signal, - }); + const prepared = await tool.prepareExecution( + {}, + { + sessionId: 'session', + turnId: 'turn', + cwd: '/workspace', + toolCallId: 'tool-call', + abortSignal: controller.signal, + }, + ); await prepared.execute({ sessionId: 'session', turnId: 'turn', From e27d47d363584428e35e29c81afcf3f675adbf4c Mon Sep 17 00:00:00 2001 From: me2seeks Date: Thu, 3 Sep 2026 14:11:38 +0800 Subject: [PATCH 9/9] test: give nested form fixtures an explicit string bound Admission now proves every legal answer serializes, so a string field without maxLength is no longer admissible. Bound the fixtures to keep them representative of forms a provider can actually publish. --- .../src/__tests__/client-capability-coordinator.test.ts | 4 +++- .../client-capability-interaction-broker.test.ts | 4 ++-- packages/runtime/src/__tests__/mcp-tools.test.ts | 8 ++++++-- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts index 8032f806be..adb437d2eb 100644 --- a/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts @@ -1927,7 +1927,9 @@ test('close waits for nested Client Capability interaction cleanup', async () => request: { message: 'Choose a target', requester: { name: 'deploy' }, - fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }], + fields: [ + { kind: 'string', name: 'target', label: 'Target', required: true, maxLength: 256 }, + ], }, }); } diff --git a/packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts b/packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts index b127b290cb..31e23f4d73 100644 --- a/packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts +++ b/packages/runtime-host/src/__tests__/client-capability-interaction-broker.test.ts @@ -91,7 +91,7 @@ test('Client Capability nested interaction pauses and rearms the execution timeo request: { message: 'Choose a target', requester: { name: 'deploy' }, - fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }], + fields: [{ kind: 'string', name: 'target', label: 'Target', required: true, maxLength: 256 }], }, }); assert.equal(timers.activeCount(), 0); @@ -156,7 +156,7 @@ test('Client Capability accepts a final result while interaction delivery is sti request: { message: 'Choose a target', requester: { name: 'deploy' }, - fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }], + fields: [{ kind: 'string', name: 'target', label: 'Target', required: true, maxLength: 256 }], }, }); diff --git a/packages/runtime/src/__tests__/mcp-tools.test.ts b/packages/runtime/src/__tests__/mcp-tools.test.ts index 0d020f69f5..878154d13f 100644 --- a/packages/runtime/src/__tests__/mcp-tools.test.ts +++ b/packages/runtime/src/__tests__/mcp-tools.test.ts @@ -107,7 +107,9 @@ test('buildMcpTools carries the Runtime-owned form callback to the provider', as { message: 'Choose a target', requester: { name: 'deploy' }, - fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }], + fields: [ + { kind: 'string', name: 'target', label: 'Target', required: true, maxLength: 256 }, + ], }, { cancellationSignal: cancellation.signal }, ); @@ -148,7 +150,9 @@ test('prepared MCP execution receives the Runtime-owned form callback after admi const answer = await options.requestInteraction({ message: 'Choose a target', requester: { name: 'deploy' }, - fields: [{ kind: 'string', name: 'target', label: 'Target', required: true }], + fields: [ + { kind: 'string', name: 'target', label: 'Target', required: true, maxLength: 256 }, + ], }); assert.deepEqual(answer, { action: 'accept', values: { target: 'staging' } }); return { content: [] };