diff --git a/packages/runtime-host/src/__tests__/host-kernel.test.ts b/packages/runtime-host/src/__tests__/host-kernel.test.ts index 2e330f305a..cd7a823bb8 100644 --- a/packages/runtime-host/src/__tests__/host-kernel.test.ts +++ b/packages/runtime-host/src/__tests__/host-kernel.test.ts @@ -86,6 +86,7 @@ import { import type { RuntimeHostCompositionSource } from '../server/host-composition.js'; import { createUnavailableDomainOperationHandlers } from '../server/operation-dispatcher.js'; import { HostChangeFeed } from '../server/host-change-feed.js'; +import { SessionAdmissionGate } from '../server/session-admission-gate.js'; import { FramedTransport, RuntimeHostTransportError } from '../transport/framed-transport.js'; import { prepareStorageRootControlDirectory, @@ -946,6 +947,37 @@ describe('non-serving Runtime Host kernel', () => { }); }); + test('requestDrain leaves an active Session admission before beginning composition drain', async () => { + await withHostPaths(async (paths) => { + const capability = await resolveStorageRoot({ path: paths.root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + let context: RuntimeHostCompositionContext | undefined; + let drainCalls = 0; + const host = await RuntimeHostKernel.start({ + owner, + idleGraceMs: 10_000, + composition: defineInteractiveRuntimeHostComposition(async (value) => { + context = value; + return testComposition({ + beginDrain: () => { + drainCalls += 1; + }, + }); + }), + }); + const admission = new SessionAdmissionGate(); + + await admission.run('session', () => { + context?.requestDrain(); + assert.equal(drainCalls, 0); + }); + + assert.equal(drainCalls, 1); + await host.closed; + }); + }); + test('execution settlement can exclude environment resources without releasing Host ownership', async () => { await withHostPaths(async (paths) => { const capability = await resolveStorageRoot({ path: paths.root, kind: 'interactive' }); 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..e2de7181c8 100644 --- a/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts @@ -34,7 +34,10 @@ import { HostRuntimeResourceCoordinator, type HostRuntimeResourceCoordinatorInput, } from '../server/runtime-resource-coordinator.js'; -import { SessionAdmissionGate } from '../server/session-admission-gate.js'; +import { + runAfterCurrentSessionAdmission, + SessionAdmissionGate, +} from '../server/session-admission-gate.js'; const SESSION_ID = 'session-1'; const RUNTIME_REF = 'maka://runtime/background-tasks/shell-1'; @@ -227,7 +230,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 +261,66 @@ describe('Host Runtime Resource coordinator', () => { assert.equal(harness.terminateCount, 0); }); + test('requests a canonical state drain only after leaving Session admission', async (t) => { + t.mock.method(console, 'error', () => {}); + let drainAdmission: Promise | undefined; + let harness!: ReturnType; + harness = createHarness({ + requestDrain: () => { + runAfterCurrentSessionAdmission(() => { + harness.drainCount += 1; + drainAdmission = harness.sessionAdmission.run(SESSION_ID, async () => {}); + void drainAdmission.catch(() => {}); + }); + }, + }); + harness.stateReadFailure = new Error('canonical state unavailable'); + + 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(harness.drainCount, 1); + assert.ok(drainAdmission, 'the canonical read failure requests a drain'); + await assert.doesNotReject(drainAdmission); + }); + + 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 +463,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 +767,29 @@ 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); + }); }); function createHarness( - options: Pick< - HostRuntimeResourceCoordinatorInput, - 'resolveShell' | 'sessionAccessAuthority' + options: Partial< + Pick< + HostRuntimeResourceCoordinatorInput, + 'requestDrain' | 'resolveShell' | 'sessionAccessAuthority' + > > = {}, ) { let backgroundCompletion: ShellRunBashInput['onCompletion']; @@ -716,6 +798,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 +913,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 84d9e743ed..c734ec68dd 100644 --- a/packages/runtime-host/src/__tests__/session-admission-gate.test.ts +++ b/packages/runtime-host/src/__tests__/session-admission-gate.test.ts @@ -20,7 +20,11 @@ import { deferred } from '@maka/core/test-only/async-primitives'; import assert from 'node:assert/strict'; import { test } from 'node:test'; -import { SessionAdmissionGate } from '../server/session-admission-gate.js'; +import { + runAfterCurrentSessionAdmission, + SessionAdmissionGate, + type SessionAdmissionLease, +} from '../server/session-admission-gate.js'; test('serializes operations for one Session', async () => { const gate = new SessionAdmissionGate(); @@ -148,6 +152,72 @@ test('rejects accidental admission re-entry instead of deadlocking', async () => }); }); +test('runs outside-admission work synchronously when no admission is active', () => { + const gate = new SessionAdmissionGate(); + let ran = false; + + runAfterCurrentSessionAdmission(() => { + ran = true; + }); + + assert.equal(ran, true); +}); + +test('runs outside-admission work after release and before the next queued admission', async () => { + const gate = new SessionAdmissionGate(); + const entered = deferred(); + const release = deferred(); + const order: string[] = []; + + const active = gate.run('session', async () => { + order.push('active:start'); + runAfterCurrentSessionAdmission(() => { + order.push('after-release'); + }); + entered.resolve(); + await release.promise; + order.push('active:end'); + }); + await entered.promise; + const queued = gate.run('session', () => { + order.push('queued'); + }); + + assert.deepEqual(order, ['active:start']); + release.resolve(); + await Promise.all([active, queued]); + assert.deepEqual(order, ['active:start', 'active:end', 'after-release', 'queued']); +}); + +test('tracks admitted work started outside the owning async chain until release', async () => { + const gate = new SessionAdmissionGate(); + const leaseReady = deferred(); + const release = deferred(); + const order: string[] = []; + + const active = gate.run('session', async (lease) => { + order.push('active:start'); + leaseReady.resolve(lease); + await release.promise; + order.push('active:end'); + }); + const lease = await leaseReady.promise; + await gate.runAdmitted('session', lease, () => { + order.push('admitted'); + runAfterCurrentSessionAdmission(() => { + order.push('after-release'); + }); + }); + const queued = gate.run('session', () => { + order.push('queued'); + }); + + assert.deepEqual(order, ['active:start', 'admitted']); + release.resolve(); + await Promise.all([active, queued]); + assert.deepEqual(order, ['active:start', 'admitted', 'active:end', 'after-release', 'queued']); +}); + test('work detached from an admission takes admissions of its own', async () => { const gate = new SessionAdmissionGate(); const release = deferred(); @@ -172,3 +242,20 @@ test('work detached from an admission takes admissions of its own', async () => await detached; assert.deepEqual(order, ['active:start', 'active:end', 'detached:admitted']); }); + +test('treats detached work as outside the current admission', async () => { + const gate = new SessionAdmissionGate(); + const order: string[] = []; + + await gate.run('session', async () => { + order.push('active:start'); + await gate.detach(async () => { + runAfterCurrentSessionAdmission(() => { + order.push('detached:outside'); + }); + }); + order.push('active:end'); + }); + + assert.deepEqual(order, ['active:start', 'detached:outside', 'active:end']); +}); 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/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index 9655398dda..af8525befb 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -94,6 +94,7 @@ import { import { HostResidencyRegistry } from './host-residency-registry.js'; import type { PeerMeshNode } from '../peer-mesh/node.js'; import { createPeerMeshOperationHandlers } from './peer-mesh-authority.js'; +import { runAfterCurrentSessionAdmission } from './session-admission-gate.js'; const DEFAULT_IDLE_GRACE_MS = 30_000; const DEFAULT_HANDSHAKE_TIMEOUT_MS = 5_000; @@ -342,9 +343,11 @@ export class RuntimeHostKernel { this.#cancelIdle(); this.#cancelInitialConnectionDeadline(); this.#armShutdownDeadline(); - this.#beginCompositionDrain(); } - this.#commitRequestedShutdownIfQuiescent(); + runAfterCurrentSessionAdmission(() => { + this.#beginCompositionDrain(); + this.#commitRequestedShutdownIfQuiescent(); + }); } async #start(): Promise { diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index ff6f237f9f..06a3ae8d81 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, @@ -72,6 +70,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'; @@ -365,7 +364,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, @@ -387,9 +386,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-host/src/server/session-admission-gate.ts b/packages/runtime-host/src/server/session-admission-gate.ts index 69576581d3..ad5e9b80cb 100644 --- a/packages/runtime-host/src/server/session-admission-gate.ts +++ b/packages/runtime-host/src/server/session-admission-gate.ts @@ -27,6 +27,7 @@ export interface SessionAdmissionLease { interface SessionAdmissionContext { readonly sessionIds: ReadonlySet; + readonly afterRelease: Set<() => void>; active: boolean; } @@ -41,6 +42,26 @@ type SessionAdmissionTaskResult = | { readonly ok: true } | { readonly ok: false; readonly error: unknown }; +const currentSessionAdmissions = new AsyncLocalStorage(); + +/** Run immediately outside admission, or after every active admission in this async chain releases. */ +export function runAfterCurrentSessionAdmission(operation: () => void): void { + const activeAdmissions = [ + ...new Set((currentSessionAdmissions.getStore() ?? []).filter((context) => context.active)), + ]; + if (activeAdmissions.length === 0) { + operation(); + return; + } + + let remaining = activeAdmissions.length; + const afterRelease = () => { + remaining -= 1; + if (remaining === 0) operation(); + }; + for (const context of activeAdmissions) context.afterRelease.add(afterRelease); +} + export class SessionAdmissionGate { readonly #tails = new Map>(); readonly #context = new AsyncLocalStorage(); @@ -91,7 +112,15 @@ export class SessionAdmissionGate { * Turn reaching its own. Leaving the context here settles that by saying so. */ detach(operation: () => T): T { - return this.#context.exit(operation); + const context = this.#context.getStore(); + return this.#context.exit(() => { + const inheritedAdmissions = currentSessionAdmissions.getStore() ?? []; + if (!context || !inheritedAdmissions.includes(context)) return operation(); + return currentSessionAdmissions.run( + inheritedAdmissions.filter((admission) => admission !== context), + operation, + ); + }); } runAdmitted( @@ -110,7 +139,13 @@ export class SessionAdmissionGate { let task: Promise; try { - task = Promise.resolve(this.#context.run(state.context, operation)); + const inheritedAdmissions = currentSessionAdmissions.getStore() ?? []; + const admissions = inheritedAdmissions.includes(state.context) + ? inheritedAdmissions + : [...inheritedAdmissions, state.context]; + task = Promise.resolve( + currentSessionAdmissions.run(admissions, () => this.#context.run(state.context, operation)), + ); } catch (error) { task = Promise.reject(error); } @@ -150,7 +185,11 @@ export class SessionAdmissionGate { } const ownedSessionIds = new Set(sessionIds); - const context: SessionAdmissionContext = { sessionIds: ownedSessionIds, active: true }; + const context: SessionAdmissionContext = { + sessionIds: ownedSessionIds, + afterRelease: new Set(), + active: true, + }; const lease: SessionAdmissionLease = Object.freeze({ [sessionAdmissionLeaseBrand]: true as const, }); @@ -166,7 +205,10 @@ export class SessionAdmissionGate { let operationError: unknown; let operationFailed = false; try { - result = await this.#context.run(context, () => operation(lease)); + const inheritedAdmissions = currentSessionAdmissions.getStore() ?? []; + result = await currentSessionAdmissions.run([...inheritedAdmissions, context], () => + this.#context.run(context, () => operation(lease)), + ); } catch (error) { operationFailed = true; operationError = error; @@ -192,8 +234,13 @@ export class SessionAdmissionGate { context.active = false; this.#leases.delete(lease); release(); - for (const [sessionId, tail] of tails) { - if (this.#tails.get(sessionId) === tail) this.#tails.delete(sessionId); + try { + for (const operation of context.afterRelease) operation(); + } finally { + context.afterRelease.clear(); + for (const [sessionId, tail] of tails) { + if (this.#tails.get(sessionId) === tail) this.#tails.delete(sessionId); + } } } } diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 3acbdb5837..a837fa0f8e 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -3265,6 +3265,95 @@ describe('SessionManager child-session runtime primitive', () => { assert.strictEqual(childTwoResult.status, 'cancelled'); }); + test('observes a rejected hosted stop while child lookup is pending', 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 authority = hostedRootAuthority(); + const ownStopError = new Error('hosted stop rejected'); + authority.stopSession = async () => { + throw ownStopError; + }; + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends: new BackendRegistry(), + messageAuthority: authority, + newId: nextId(), + now: nextNow(350), + }); + const unhandled: unknown[] = []; + const onUnhandledRejection = (reason: unknown) => { + if (reason === ownStopError) unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandledRejection); + + const stopping = manager.stopSession('session-1', { source: 'stop_button' }); + 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(); + await stopRejection; + process.off('unhandledRejection', onUnhandledRejection); + } + }); + + test('observes a rejected direct stop while hosted child lookup is pending', 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('direct stop rejected'); + const manager = new SessionManager({ + store, + runStore, + runtimeEventStore: runStore, + backends: new BackendRegistry(), + runtimeKernel: { + stopSession: async () => { + throw ownStopError; + }, + } as unknown as RuntimeKernelLike, + messageAuthority: hostedRootAuthority(), + newId: nextId(), + now: nextNow(375), + }); + const unhandled: unknown[] = []; + const onUnhandledRejection = (reason: unknown) => { + if (reason === ownStopError) unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandledRejection); + + const stopping = manager.deliverHostedRootStop('session-1', { source: 'stop_button' }); + 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(); + await stopRejection; + 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 91bd2d1021..60f8c7c9ee 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -3593,38 +3593,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, + stopOwn: () => Promise, + stopChild: (childSessionId: string) => Promise, + ): Promise { + const ownStop = observeSettlement(stopOwn()); let childStops: PromiseSettledResult[] = []; let childLookupError: unknown; try { @@ -3632,16 +3633,13 @@ 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; } - await ownStop; + const ownStopResult = await ownStop; + if (ownStopResult.status === 'rejected') throw ownStopResult.reason; const childStopError = childStops.find( (result): result is PromiseRejectedResult => result.status === 'rejected', )?.reason; @@ -5531,6 +5529,13 @@ function tail(items: readonly T[], max: number): T[] { return items.slice(items.length - max); } +function observeSettlement(promise: Promise): Promise> { + return promise.then( + (value) => ({ status: 'fulfilled', value }), + (reason: unknown) => ({ status: 'rejected', reason }), + ); +} + function shellRunBashToolCallIds(messages: readonly StoredMessage[]): Set { return new Set( messages.flatMap((message) =>