From 30078b30e6366b1a0bf584f86d7bd4ff2a02cdc5 Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Thu, 3 Sep 2026 20:48:41 +0800 Subject: [PATCH 1/2] fix(runtime-host): defer resource drain outside admission Observe hosted stop failures immediately so delayed child lookup cannot expose an unhandled rejection. Generated-by: Codex --- .../runtime-resource-coordinator.test.ts | 69 +++++++++++++++++-- .../src/server/failure-diagnostic.ts | 27 ++++++++ .../src/server/operation-dispatcher.ts | 11 +-- .../server/runtime-resource-coordinator.ts | 18 +++-- .../src/__tests__/session-manager.test.ts | 44 ++++++++++++ packages/runtime/src/session-manager.ts | 18 +++-- 6 files changed, 165 insertions(+), 22 deletions(-) create mode 100644 packages/runtime-host/src/server/failure-diagnostic.ts 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..0ed27f5509 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,64 @@ 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: () => { + 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'); @@ -705,9 +764,11 @@ describe('Host Runtime Resource coordinator', () => { }); function createHarness( - options: Pick< - HostRuntimeResourceCoordinatorInput, - 'resolveShell' | 'sessionAccessAuthority' + options: Partial< + Pick< + HostRuntimeResourceCoordinatorInput, + 'requestDrain' | 'resolveShell' | 'sessionAccessAuthority' + > > = {}, ) { let backgroundCompletion: ShellRunBashInput['onCompletion']; 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 ff6f237f9f..bc69fa9513 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, @@ -57,6 +55,7 @@ import { PLAN_OPERATION_SPECS } from '../protocol/plan.js'; import { PROJECT_CATALOG_OPERATION_SPECS } from '../protocol/project-catalog.js'; import { RUNTIME_POLICY_OPERATION_SPECS } from '../protocol/runtime-policy.js'; import { RUNTIME_RESOURCE_OPERATION_SPECS } from '../protocol/runtime-resource.js'; +import { boundedFailureDiagnostic } from './failure-diagnostic.js'; import { SCHEDULED_TASK_OPERATION_SPECS } from '../protocol/scheduled-task.js'; import { SESSION_CATALOG_OPERATION_SPECS } from '../protocol/session-catalog.js'; import { SESSION_CONTINUITY_OPERATION_SPECS } from '../protocol/session-continuity.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..15ad30b058 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, @@ -302,6 +303,7 @@ export class HostRuntimeResourceCoordinator if (context.principalKind === 'session_guest' && !guestGrantId) { return queryFailure('not_found', 'Session was not found'); } + let canonicalReadFailure: { readonly error: unknown } | undefined; const outcome: OperationOutcome<'runtime.resource.query'> = await this.#sessionAdmission.run( input.sessionId, async () => { @@ -311,7 +313,7 @@ export class HostRuntimeResourceCoordinator if (isSessionNotFoundError(error)) { return queryFailure('not_found', 'Session was not found'); } - this.#requestDrain(); + canonicalReadFailure = { error }; return queryFailure('internal_failure', 'Session state is unavailable'); } if (input.kind === 'get') { @@ -331,8 +333,8 @@ export class HostRuntimeResourceCoordinator resource: canonical, }), }; - } catch { - this.#requestDrain(); + } catch (error) { + canonicalReadFailure = { error }; return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); } } @@ -342,8 +344,8 @@ export class HostRuntimeResourceCoordinator if (context.principalKind === 'session_guest') { updates = updates.filter((update) => update.sessionId === input.sessionId); } - } catch { - this.#requestDrain(); + } catch (error) { + canonicalReadFailure = { error }; return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); } try { @@ -373,6 +375,12 @@ export class HostRuntimeResourceCoordinator } }, ); + if (canonicalReadFailure !== undefined) { + console.error( + `[runtime-host] canonical Runtime Resource read failed: ${boundedFailureDiagnostic(canonicalReadFailure.error)}`, + ); + this.#requestDrain(); + } return guestGrantId && this.#guestObservationGrantId(context, input.sessionId) !== guestGrantId ? queryFailure('not_found', 'Session was not found') : outcome; diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 8b8c0e0a17..b7fae7ae04 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -3231,6 +3231,50 @@ 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('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 e602dba5c4..9675162a6f 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -3585,9 +3585,11 @@ 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); + const ownStop = observeSettlement( + hostedAuthority + ? hostedAuthority.stopSession(sessionId, input) + : this.runtimeKernel.stopSession(sessionId, input), + ); let childStops: PromiseSettledResult[] = []; let childLookupError: unknown; try { @@ -3604,7 +3606,8 @@ export class SessionManager { } 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; @@ -5560,6 +5563,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) => From b9b0ecd4cae99c8939d5592a8573e2cb8dc1de7c Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Fri, 4 Sep 2026 14:30:49 +0800 Subject: [PATCH 2/2] fix(runtime-host): close reentrant drain review gaps Generated-by: Codex --- .../src/__tests__/host-kernel.test.ts | 32 +++++++++ .../runtime-resource-coordinator.test.ts | 35 +++++++-- .../__tests__/session-admission-gate.test.ts | 72 ++++++++++++++++++- .../runtime-host/src/server/host-kernel.ts | 7 +- .../src/server/operation-dispatcher.ts | 2 +- .../server/runtime-resource-coordinator.ts | 27 +++---- .../src/server/session-admission-gate.ts | 49 +++++++++++-- .../src/__tests__/session-manager.test.ts | 45 ++++++++++++ packages/runtime/src/session-manager.ts | 63 ++++++++-------- 9 files changed, 272 insertions(+), 60 deletions(-) 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 0ed27f5509..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'; @@ -264,9 +267,11 @@ describe('Host Runtime Resource coordinator', () => { let harness!: ReturnType; harness = createHarness({ requestDrain: () => { - harness.drainCount += 1; - drainAdmission = harness.sessionAdmission.run(SESSION_ID, async () => {}); - void drainAdmission.catch(() => {}); + runAfterCurrentSessionAdmission(() => { + harness.drainCount += 1; + drainAdmission = harness.sessionAdmission.run(SESSION_ID, async () => {}); + void drainAdmission.catch(() => {}); + }); }, }); harness.stateReadFailure = new Error('canonical state unavailable'); @@ -458,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 }); }); @@ -761,6 +767,21 @@ 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( @@ -777,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, @@ -890,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 35044a0c1c..406a487742 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(); @@ -147,3 +151,69 @@ 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']); +}); 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 bc69fa9513..06a3ae8d81 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -55,7 +55,6 @@ import { PLAN_OPERATION_SPECS } from '../protocol/plan.js'; import { PROJECT_CATALOG_OPERATION_SPECS } from '../protocol/project-catalog.js'; import { RUNTIME_POLICY_OPERATION_SPECS } from '../protocol/runtime-policy.js'; import { RUNTIME_RESOURCE_OPERATION_SPECS } from '../protocol/runtime-resource.js'; -import { boundedFailureDiagnostic } from './failure-diagnostic.js'; import { SCHEDULED_TASK_OPERATION_SPECS } from '../protocol/scheduled-task.js'; import { SESSION_CATALOG_OPERATION_SPECS } from '../protocol/session-catalog.js'; import { SESSION_CONTINUITY_OPERATION_SPECS } from '../protocol/session-continuity.js'; @@ -71,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'; diff --git a/packages/runtime-host/src/server/runtime-resource-coordinator.ts b/packages/runtime-host/src/server/runtime-resource-coordinator.ts index 15ad30b058..f5614c98e4 100644 --- a/packages/runtime-host/src/server/runtime-resource-coordinator.ts +++ b/packages/runtime-host/src/server/runtime-resource-coordinator.ts @@ -303,7 +303,6 @@ export class HostRuntimeResourceCoordinator if (context.principalKind === 'session_guest' && !guestGrantId) { return queryFailure('not_found', 'Session was not found'); } - let canonicalReadFailure: { readonly error: unknown } | undefined; const outcome: OperationOutcome<'runtime.resource.query'> = await this.#sessionAdmission.run( input.sessionId, async () => { @@ -313,8 +312,7 @@ export class HostRuntimeResourceCoordinator if (isSessionNotFoundError(error)) { return queryFailure('not_found', 'Session was not found'); } - canonicalReadFailure = { error }; - return queryFailure('internal_failure', 'Session state is unavailable'); + return this.#canonicalReadFailure(error, 'Session state is unavailable'); } if (input.kind === 'get') { try { @@ -334,8 +332,7 @@ export class HostRuntimeResourceCoordinator }), }; } catch (error) { - canonicalReadFailure = { error }; - return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); + return this.#canonicalReadFailure(error, 'Runtime Resource state is unavailable'); } } let updates: ShellRunUpdate[]; @@ -345,8 +342,7 @@ export class HostRuntimeResourceCoordinator updates = updates.filter((update) => update.sessionId === input.sessionId); } } catch (error) { - canonicalReadFailure = { error }; - return queryFailure('internal_failure', 'Runtime Resource state is unavailable'); + return this.#canonicalReadFailure(error, 'Runtime Resource state is unavailable'); } try { const resources = canonicalRuntimeResources(updates); @@ -375,17 +371,22 @@ export class HostRuntimeResourceCoordinator } }, ); - if (canonicalReadFailure !== undefined) { - console.error( - `[runtime-host] canonical Runtime Resource read failed: ${boundedFailureDiagnostic(canonicalReadFailure.error)}`, - ); - this.#requestDrain(); - } return guestGrantId && this.#guestObservationGrantId(context, input.sessionId) !== guestGrantId ? queryFailure('not_found', 'Session was not found') : 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 49dc631181..637d2a86f7 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(); @@ -97,7 +118,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); } @@ -137,7 +164,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, }); @@ -153,7 +184,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; @@ -179,8 +213,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 b7fae7ae04..0dd08b300c 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -3275,6 +3275,51 @@ describe('SessionManager child-session runtime primitive', () => { } }); + 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 9675162a6f..f8e8492893 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -3585,41 +3585,39 @@ export class SessionManager { const hostedAuthority = isRuntimeHostedRootAuthority(this.deps.messageAuthority) ? this.deps.messageAuthority : undefined; - const ownStop = observeSettlement( - hostedAuthority - ? hostedAuthority.stopSession(sessionId, input) - : this.runtimeKernel.stopSession(sessionId, input), + 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), ); - 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; - } - const ownStopResult = await ownStop; - if (ownStopResult.status === 'rejected') throw ownStopResult.reason; - const childStopError = childStops.find( - (result): result is PromiseRejectedResult => result.status === 'rejected', - )?.reason; - if (childLookupError !== undefined) throw childLookupError; - if (childStopError !== undefined) throw childStopError; } 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 { @@ -3627,16 +3625,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;