diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 772a457dd3..f5cf1f9592 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -36,7 +36,13 @@ import type { AgentGraphIntentClaim, AgentGraphIntentClaimRequest, } from '@maka/core/agent-graph-control'; +import type { HostedUserQuestionSettlement } from '@maka/core/backend-types'; import type { ShellRunRecord } from '@maka/core/shell-run'; +import { waitFor as pollFor } from '@maka/core/test-only/async-primitives'; +import { + AgentGraphCoordinator, + agentGraphIdForRootSession, +} from '@maka/runtime/stream-graph-coordinator'; import { FAKE_ASK_USER_QUESTION_PROMPT, FAKE_HOLD_OPEN_PROMPT, @@ -74,7 +80,10 @@ import { stopOwnedWorkHubRoot, stopReplacedWorkHubRoot, } from '../server/execution-composition.js'; -import { waitFor as pollFor } from '@maka/core/test-only/async-primitives'; +import { RuntimeHostKernel, type RuntimeHostCompositionContext } from '../server/host-kernel.js'; +import { defineInteractiveRuntimeHostComposition } from '../server/host-composition.js'; +import { connectRuntimeHost, RuntimeHostOperationError } from '../client/index.js'; +import { RUNTIME_HOST_PROTOCOL_VERSION } from '../protocol/index.js'; import { readLedgerMessages } from './fixtures/ledger-transcript.js'; const require = createRequire(import.meta.url); @@ -2102,6 +2111,193 @@ test('production composition validates graph stop before aborting a claimed chil }); }); +test('interaction fail-stop stops graph operators through the kernel and releases ownership', { + timeout: 10_000, +}, async (t) => { + await withCompositionRoot(async ({ root, owner }) => { + const failure = new Error('backend continuation apply failed'); + let graph!: AgentGraphCoordinator; + const recoverGraph = AgentGraphCoordinator.prototype.recover; + t.mock.method( + AgentGraphCoordinator.prototype, + 'recover', + async function (this: AgentGraphCoordinator) { + graph = this; + return recoverGraph.call(this); + }, + ); + const published = deferred(); + const stopped = deferred(); + const stopObservations: Array<{ error?: unknown }> = []; + let settlement: HostedUserQuestionSettlement | undefined; + let retained = false; + let retainedAtShutdownRequest = false; + let captured!: Awaited>; + const host = await RuntimeHostKernel.start({ + owner, + idleGraceMs: 60_000, + shutdownGraceMs: 5_000, + composition: defineInteractiveRuntimeHostComposition(async (kernelContext) => { + captured = await createCapturedExecutionComposition(owner, { + context: { + ...kernelContext, + retainUntilProcessExit: () => { + retained = true; + kernelContext.retainUntilProcessExit(); + }, + requestDrain: () => { + retainedAtShutdownRequest = retained; + kernelContext.requestDrain(); + }, + }, + primaryBackendFactory: (backendContext) => { + const backend = new FakeBackend(backendContext); + const send = backend.send.bind(backend); + backend.send = async function* (input) { + const bridge = input.hostedInteraction; + assert.ok(bridge); + yield* send({ + ...input, + hostedInteraction: { + ...bridge, + admitUserQuestionRequest: async (request) => { + settlement = request.settlement; + await bridge.admitUserQuestionRequest({ + ...request, + settlement: { + ...request.settlement, + applyAnswer: async () => { + throw failure; + }, + }, + }); + published.resolve(request.request.requestId); + }, + }, + }); + }; + return backend; + }, + }); + return captured.composition; + }), + }); + const closed = host.closed.then( + () => undefined, + (error: unknown) => error, + ); + const connected = await connectRuntimeHost({ + rootPath: root, + protocol: { min: RUNTIME_HOST_PROTOCOL_VERSION, max: RUNTIME_HOST_PROTOCOL_VERSION }, + }); + assert.equal(connected.kind, 'connected'); + if (connected.kind !== 'connected') throw new Error('kernel connection unavailable'); + const { manager } = captured; + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + try { + const session = await manager.createSession({ + cwd: root, + llmConnectionId: FAKE_CONNECTION_ID, + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + }); + await graph.toolsForSession(session.id); + const turnId = 'interaction-drain-turn'; + // Prepare the held-open backend with a fixture residency. The answer below exercises + // kernel drain over UDS; poisoned root-execution settlement is a separate close path. + const started = await captured.composition.handlers['turn.start']( + { + sessionId: session.id, + turnId, + content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, + }, + { + hostEpoch: host.hostEpoch, + connectionId: 'interaction-drain-fixture', + principal: 'local_os_user', + acquireResidency: () => ({ release() {} }), + }, + ); + assert.equal(started.ok, true); + const interactionId = await published.promise; + const run = (await stores.runtimeEventStore.listSessionInvocations(session.id)).find( + (run) => run.turnId === turnId, + ); + assert.ok(run); + const operator = await manager.provisionAgentGraphOperator({ + graphId: agentGraphIdForRootSession(session.id), + workId: `graph_work_${'a'.repeat(32)}`, + operatorId: `graph_operator_${'b'.repeat(32)}`, + agentId: LOCAL_READ_AGENT_DEFINITION.id, + source: { + sessionId: session.id, + turnId, + runId: run.runId, + toolCallId: 'provision-for-drain', + }, + edges: [], + expectedScheduleRevision: 0, + }); + const stopSession = manager.stopSession.bind(manager); + t.mock.method( + manager, + 'stopSession', + async (sessionId: string, input: Parameters[1]) => { + if (sessionId !== operator.header.id) return stopSession(sessionId, input); + const observation: (typeof stopObservations)[number] = {}; + stopObservations.push(observation); + try { + await stopSession(sessionId, input); + } catch (error) { + observation.error = error; + throw error; + } finally { + stopped.resolve(); + } + }, + ); + await assert.rejects( + connected.connection.request('interaction.answer', { + sessionId: session.id, + interactionId, + answer: { kind: 'question', answers: ['邀请制', '本周', '是'] }, + }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === 'internal_failure', + ); + await stopped.promise; + assert.equal(retained, true); + assert.equal(retainedAtShutdownRequest, true); + assert.deepEqual(stopObservations, [{}]); + } finally { + // Release the injected backend waiter; fail-stop intentionally cannot apply its continuation. + await settlement?.applyClosure('turn_stopped'); + await connected.connection.close(); + void host.close().catch(() => undefined); + const closeError = await closed; + assert.ok( + closeError instanceof AggregateError, + `Unexpected shutdown result: ${String(closeError)}`, + ); + const errorTree = (error: unknown): string => + error instanceof AggregateError + ? [error.message, ...error.errors.map(errorTree)].join('\n') + : String(error); + // Poisoned compositions can aggregate other close errors; operator stop must not reenter admission. + const details = errorTree(closeError); + assert.match(details, /Interaction coordinator entered fail-stop/); + assert.doesNotMatch( + details, + /Cannot enter Session admission|termination required|shutdown deadline/i, + ); + const replacementOwner = await tryAcquireInteractiveRootOwner(owner.capability); + assert.ok(replacementOwner, 'kernel released exclusive root ownership'); + await replacementOwner.close(); + } + }); +}); + function compositionContext(owner: InteractiveRootOwner) { return { owner, @@ -2215,6 +2411,10 @@ async function seedLegacyFakeBackendSession( async function createCapturedExecutionComposition( owner: InteractiveRootOwner, options: { + readonly context?: Pick< + RuntimeHostCompositionContext, + 'retainUntilProcessExit' | 'requestDrain' + >; readonly safeBoundaryResume?: boolean; readonly primaryBackendFactory?: BackendFactory; readonly residencies?: HostResidencyRegistry; @@ -2248,6 +2448,7 @@ async function createCapturedExecutionComposition( residencies.acquire(label, kind), } : {}), + ...options.context, }, {}, { primaryBackendFactory }, diff --git a/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts b/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts index a9597a56bd..53f499143c 100644 --- a/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts @@ -227,7 +227,8 @@ describe('Host Runtime Resource coordinator', () => { assert.equal(!revoked.ok && revoked.error.code, 'not_found'); }); - test('drains for canonical state failure but keeps projection failure scoped to its query', async () => { + test('drains for canonical state failure but keeps projection failure scoped to its query', async (t) => { + t.mock.method(console, 'error', () => {}); const harness = createHarness(); harness.updates = [ resourceUpdate(0, { @@ -257,6 +258,39 @@ describe('Host Runtime Resource coordinator', () => { assert.equal(harness.terminateCount, 0); }); + test('logs a bounded redacted canonical state failure before draining', async (t) => { + const logs: string[] = []; + let drainCount = 0; + let logCountAtDrain = 0; + t.mock.method(console, 'error', (...args: unknown[]) => { + logs.push(args.map(String).join(' ')); + }); + const harness = createHarness({ + requestDrain: () => { + drainCount += 1; + logCountAtDrain = logs.length; + }, + }); + harness.stateReadFailure = new Error( + `canonical state unavailable api_key=sk-secretvalue123 ${'x'.repeat(16 * 1024)}`, + ); + + const result = await harness.coordinator.handlers['runtime.resource.query']( + { kind: 'list_start', sessionId: SESSION_ID }, + connection('connection-1'), + ); + + assert.equal(result.ok, false); + assert.equal(!result.ok && result.error.code, 'internal_failure'); + assert.equal(drainCount, 1); + assert.equal(logCountAtDrain, 1); + assert.equal(logs.length, 1); + assert.match(logs[0] ?? '', /canonical state unavailable/); + assert.match(logs[0] ?? '', /\[redacted\]/i); + assert.doesNotMatch(logs[0] ?? '', /sk-secretvalue123/); + assert.ok(Buffer.byteLength(logs[0] ?? '', 'utf8') < 9 * 1024); + }); + test('fences PTY control by connection and retains only exact sequence retries', async () => { const harness = createHarness(); const firstConnection = connection('connection-1'); @@ -399,6 +433,7 @@ describe('Host Runtime Resource coordinator', () => { assert.equal(started.ok, false); assert.ok(harness.lastBackgroundInput); assert.equal(harness.stopCount, 1); + assert.equal(harness.drainCount, 1); harness.finishBackground({ successful: false }); }); @@ -702,12 +737,55 @@ describe('Host Runtime Resource coordinator', () => { assert.equal(!missing.ok && missing.error.code, 'not_found'); assert.equal(harness.stopCount, 0); }); + + test('drains when the admitted mutable Session read fails', async () => { + const harness = createHarness(); + harness.sessionReadFailureAt = 2; + + const started = await harness.coordinator.handlers['runtime.resource.start']( + { sessionId: SESSION_ID, launchId: 'session-read-failure' }, + connection('connection-1'), + ); + + assert.equal(started.ok, false); + assert.equal(!started.ok && started.error.code, 'internal_failure'); + assert.equal(harness.drainCount, 1); + assert.equal(harness.lastBackgroundInput, undefined); + }); +}); + +test('rejects a queued resource start when drain detaches from the active Session admission', async () => { + const harness = createHarness(); + let release!: () => void; + let entered!: () => void; + const blocker = new Promise((resolve) => { + release = resolve; + }); + const started = new Promise((resolve) => { + entered = resolve; + }); + const active = harness.sessionAdmission.run(SESSION_ID, async () => { + entered(); + await blocker; + // Invoke from inside the active async context after the resource launch is queued. + harness.sessionAdmission.detach(() => harness.coordinator.beginDrain()); + }); + await started; + const resource = harness.coordinator.runBackgroundBash(backgroundInput()); + const observed = assert.rejects(resource, /Runtime resources are draining/); + release(); + await Promise.all([active, observed]); + assert.equal(harness.lastBackgroundInput, undefined); + assert.equal(harness.terminateCount, 1); + assert.equal(harness.activeResidencies, 0); }); function createHarness( - options: Pick< - HostRuntimeResourceCoordinatorInput, - 'resolveShell' | 'sessionAccessAuthority' + options: Partial< + Pick< + HostRuntimeResourceCoordinatorInput, + 'requestDrain' | 'resolveShell' | 'sessionAccessAuthority' + > > = {}, ) { let backgroundCompletion: ShellRunBashInput['onCompletion']; @@ -716,6 +794,8 @@ function createHarness( const state = { updates: [resourceUpdate(0)], sessionState: 'active' as 'active' | 'archived' | 'missing', + sessionReadCount: 0, + sessionReadFailureAt: undefined as number | undefined, writeCount: 0, stopCount: 0, terminateCount: 0, @@ -829,6 +909,10 @@ function createHarness( }, sessionHeaders: { readHeader: async (sessionId) => { + state.sessionReadCount += 1; + if (state.sessionReadCount === state.sessionReadFailureAt) { + throw new Error('Session state unavailable'); + } if (state.sessionState === 'missing') throw new SessionNotFoundError(sessionId); return { cwd: '/workspace', diff --git a/packages/runtime-host/src/__tests__/session-admission-gate.test.ts b/packages/runtime-host/src/__tests__/session-admission-gate.test.ts index f94d87a92f..626181be1d 100644 --- a/packages/runtime-host/src/__tests__/session-admission-gate.test.ts +++ b/packages/runtime-host/src/__tests__/session-admission-gate.test.ts @@ -173,3 +173,32 @@ test('work detached from an admission takes admissions of its own', async () => await detached; assert.deepEqual(order, ['active:start', 'active:end', 'detached:admitted']); }); + +test('detached stop waits for its Session admission to release', async () => { + const gate = new SessionAdmissionGate(); + const entered = deferred(); + const release = deferred(); + const order: string[] = []; + let stop!: Promise; + const active = gate.run('session', async () => { + order.push('active'); + stop = gate.detach(() => + gate.run('session', () => { + order.push('stop'); + }), + ); + entered.resolve(); + await release.promise; + order.push('released'); + }); + await entered.promise; + await new Promise((resolve) => setImmediate(resolve)); + try { + assert.deepEqual(order, ['active']); + } finally { + release.resolve(); + await active; + await stop; + } + assert.deepEqual(order, ['active', 'released', 'stop']); +}); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index cfb2070c78..b4a5942f5d 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -704,7 +704,7 @@ export async function createExecutionRuntimeHostComposition( if (poisonFailure) return; poisonFailure = error; context.retainUntilProcessExit(); - beginDrain(); + // Route poison through the kernel; the composition drain entry detaches admission. context.requestDrain(); }, onSandboxBoundarySettled: (sessionId) => @@ -1178,7 +1178,7 @@ export async function createExecutionRuntimeHostComposition( poisonFailure = error; runtimePolicyActivation.poison(); context.retainUntilProcessExit(); - beginDrain(); + // Route poison through the kernel; the composition drain entry detaches admission. context.requestDrain(); }, ...dependencies.oauthAuthorization, @@ -2185,7 +2185,9 @@ export async function createExecutionRuntimeHostComposition( releaseConnection: (connectionId: string) => { for (const module of domainModules) module.releaseConnection?.(connectionId); }, - beginDrain, + // Drain may stop graph operators while its caller still owns a Session admission. + // Leave that context; each stop still waits on its own Session queue. + beginDrain: () => sessionAdmission.detach(beginDrain), recover, startMaintenance: () => storageMaintenance.start(), close, diff --git a/packages/runtime-host/src/server/failure-diagnostic.ts b/packages/runtime-host/src/server/failure-diagnostic.ts new file mode 100644 index 0000000000..28e869d492 --- /dev/null +++ b/packages/runtime-host/src/server/failure-diagnostic.ts @@ -0,0 +1,27 @@ +/* + * 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 { truncateUtf8 } from '@maka/core/diagnostic-log'; +import { redactSecrets } from '@maka/core/redaction'; + +export function boundedFailureDiagnostic(error: unknown): string { + const details = + error instanceof Error ? error.stack || `${error.name}: ${error.message}` : String(error); + return truncateUtf8(redactSecrets(details), 8 * 1024, '\n'); +} diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index bbde94a645..18e5e7f61b 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -17,8 +17,6 @@ * under the License. */ -import { truncateUtf8 } from '@maka/core/diagnostic-log'; -import { redactSecrets } from '@maka/core/redaction'; import type { RootTurnAdmissionAuthorization } from '@maka/storage/execution-stores'; import { HOST_OPERATION_SPECS, @@ -73,6 +71,7 @@ import { USAGE_PRICING_OPERATION_SPECS } from '../protocol/usage-pricing.js'; import { WEB_SEARCH_OPERATION_SPECS } from '../protocol/web-search.js'; import { WORKHUB_COORDINATION_OPERATION_SPECS } from '../protocol/workhub-coordination.js'; import { PLUGIN_PLATFORM_OPERATION_SPECS } from '../protocol/plugin-platform.js'; +import { boundedFailureDiagnostic } from './failure-diagnostic.js'; import { createPeerMeshOperationHandlers } from './peer-mesh-authority.js'; import type { RuntimeHostConnectionAuthority } from './connection-authority.js'; @@ -369,7 +368,7 @@ async function dispatchTypedOperation( outcome = decodeOperationOutcome(request.operation, await handler(request.input, context)); } catch (error) { console.error( - `[runtime-host] unexpected ${request.operation} failure: ${boundedUnexpectedFailure(error)}`, + `[runtime-host] unexpected ${request.operation} failure: ${boundedFailureDiagnostic(error)}`, ); return operationFailureResponse( request as RequestFrame, @@ -391,9 +390,3 @@ async function dispatchTypedOperation( error: outcome.error, }; } - -function boundedUnexpectedFailure(error: unknown): string { - const details = - error instanceof Error ? error.stack || `${error.name}: ${error.message}` : String(error); - return truncateUtf8(redactSecrets(details), 8 * 1024, '\n'); -} diff --git a/packages/runtime-host/src/server/runtime-resource-coordinator.ts b/packages/runtime-host/src/server/runtime-resource-coordinator.ts index 95499ef4ad..f5614c98e4 100644 --- a/packages/runtime-host/src/server/runtime-resource-coordinator.ts +++ b/packages/runtime-host/src/server/runtime-resource-coordinator.ts @@ -60,6 +60,7 @@ import type { RuntimeResourceOperationHandlerMap, } from './operation-dispatcher.js'; import type { RuntimeHostAccessAuthority } from './access-authority.js'; +import { boundedFailureDiagnostic } from './failure-diagnostic.js'; import { SessionAdmissionGate } from './session-admission-gate.js'; import { boundedRuntimeResourceSnapshot, @@ -311,8 +312,7 @@ export class HostRuntimeResourceCoordinator if (isSessionNotFoundError(error)) { return queryFailure('not_found', 'Session was not found'); } - this.#requestDrain(); - return queryFailure('internal_failure', 'Session state is unavailable'); + return this.#canonicalReadFailure(error, 'Session state is unavailable'); } if (input.kind === 'get') { try { @@ -331,9 +331,8 @@ export class HostRuntimeResourceCoordinator resource: canonical, }), }; - } catch { - this.#requestDrain(); - return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); + } catch (error) { + return this.#canonicalReadFailure(error, 'Runtime Resource state is unavailable'); } } let updates: ShellRunUpdate[]; @@ -342,9 +341,8 @@ export class HostRuntimeResourceCoordinator if (context.principalKind === 'session_guest') { updates = updates.filter((update) => update.sessionId === input.sessionId); } - } catch { - this.#requestDrain(); - return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); + } catch (error) { + return this.#canonicalReadFailure(error, 'Runtime Resource state is unavailable'); } try { const resources = canonicalRuntimeResources(updates); @@ -378,6 +376,17 @@ export class HostRuntimeResourceCoordinator : outcome; } + #canonicalReadFailure( + error: unknown, + message: string, + ): OperationOutcome<'runtime.resource.query'> { + console.error( + `[runtime-host] canonical Runtime Resource read failed: ${boundedFailureDiagnostic(error)}`, + ); + this.#requestDrain(); + return queryFailure('internal_failure', message); + } + #guestObservationGrantId(context: ConnectionContext, sessionId: string): string | undefined { if (context.principalKind !== 'session_guest') return; return this.#sessionAccessAuthority?.activeSessionGrant( diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 124d79ff28..6a10b404e8 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -3292,6 +3292,73 @@ describe('SessionManager child-session runtime primitive', () => { assert.strictEqual(childTwoResult.status, 'cancelled'); }); + for (const { name, stopOptions, stop } of [ + { + name: 'observes a rejected hosted stop while child lookup is pending', + stopOptions: (rejectStop: () => Promise) => { + const authority = hostedRootAuthority(); + authority.stopSession = rejectStop; + return { messageAuthority: authority }; + }, + stop: (manager: SessionManager) => + manager.stopSession('session-1', { source: 'stop_button' }), + }, + { + name: 'observes a rejected direct stop while hosted child lookup is pending', + stopOptions: (rejectStop: () => Promise) => ({ + runtimeKernel: { stopSession: rejectStop } as unknown as RuntimeKernelLike, + messageAuthority: hostedRootAuthority(), + }), + stop: (manager: SessionManager) => + manager.deliverHostedRootStop('session-1', { source: 'stop_button' }), + }, + ]) { + test(name, async () => { + const store = new MemorySessionStore(); + const listStarted = makeGate(); + const releaseList = makeGate(); + const childLookupError = new Error('child lookup failed'); + store.list = async () => { + listStarted.release(); + await releaseList.promise; + throw childLookupError; + }; + const runStore = new MemoryAgentRunStore(); + const ownStopError = new Error('own stop rejected'); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends: new BackendRegistry(), + ...stopOptions(async () => { + throw ownStopError; + }), + newId: nextId(), + now: nextNow(350), + }); + const unhandled: unknown[] = []; + const onUnhandledRejection = (reason: unknown) => { + if (reason === ownStopError) unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandledRejection); + + const stopping = stop(manager); + const stopRejection = assert.rejects(stopping, (error: unknown) => error === ownStopError); + try { + await listStarted.promise; + await new Promise((resolve) => setImmediate(resolve)); + assert.deepStrictEqual(unhandled, []); + } finally { + releaseList.release(); + try { + await stopRejection; + } finally { + process.off('unhandledRejection', onUnhandledRejection); + } + } + }); + } + test('startup recovery repairs an interrupted child inline run only in the child session', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 40000d95d0..78740026ba 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -3595,38 +3595,39 @@ export class SessionManager { const hostedAuthority = isRuntimeHostedRootAuthority(this.deps.messageAuthority) ? this.deps.messageAuthority : undefined; - const ownStop = hostedAuthority - ? hostedAuthority.stopSession(sessionId, input) - : this.runtimeKernel.stopSession(sessionId, input); - let childStops: PromiseSettledResult[] = []; - let childLookupError: unknown; - try { - const children = await this.listChildSessions(sessionId); - childStops = await Promise.allSettled( - children - .filter((child) => child.subagentParent?.lifecycle === 'foreground') - .map((child) => - hostedAuthority - ? hostedAuthority.stopSession(child.id, input) - : this.runtimeKernel.stopSession(child.id, input), - ), - ); - } catch (error) { - childLookupError = error; - } - await ownStop; - const childStopError = childStops.find( - (result): result is PromiseRejectedResult => result.status === 'rejected', - )?.reason; - if (childLookupError !== undefined) throw childLookupError; - if (childStopError !== undefined) throw childStopError; + await this.#stopSessionTree( + sessionId, + hostedAuthority + ? hostedAuthority.stopSession(sessionId, input) + : this.runtimeKernel.stopSession(sessionId, input), + (childSessionId) => + hostedAuthority + ? hostedAuthority.stopSession(childSessionId, input) + : this.runtimeKernel.stopSession(childSessionId, input), + ); } async deliverHostedRootStop(sessionId: string, input: StopSessionInput = {}): Promise { - const ownStop = this.runtimeKernel.stopSession(sessionId, input); const authority = isRuntimeHostedRootAuthority(this.deps.messageAuthority) ? this.deps.messageAuthority : undefined; + await this.#stopSessionTree( + sessionId, + this.runtimeKernel.stopSession(sessionId, input), + (childSessionId) => + authority + ? authority.stopSession(childSessionId, input) + : this.runtimeKernel.stopSession(childSessionId, input), + ); + } + + async #stopSessionTree( + sessionId: string, + ownStop: Promise, + stopChild: (childSessionId: string) => Promise, + ): Promise { + // Observe immediately while child lookup runs; await below still propagates the original error. + void ownStop.catch(() => undefined); let childStops: PromiseSettledResult[] = []; let childLookupError: unknown; try { @@ -3634,11 +3635,7 @@ export class SessionManager { childStops = await Promise.allSettled( children .filter((child) => child.subagentParent?.lifecycle === 'foreground') - .map((child) => - authority - ? authority.stopSession(child.id, input) - : this.runtimeKernel.stopSession(child.id, input), - ), + .map((child) => stopChild(child.id)), ); } catch (error) { childLookupError = error;