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/6] 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/6] 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; From 3ae27121e005512723f2c64bea1ba5fbd6d82155 Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Mon, 7 Sep 2026 17:00:06 +0800 Subject: [PATCH 3/6] fix(runtime-host): route fail-stop drain through kernel Interaction poison and OAuth fatal callbacks started domain drain before the kernel could defer it past Session admission. Keep the immediate fail-stop latch and let the kernel own domain drain ordering. Cover interaction.answer continuation failure with the production composition, SQLite graph operator provisioning, and real SessionManager stop. Verify stop runs outside admission and the original failure remains canonical. Generated-by: Codex --- .../__tests__/execution-composition.test.ts | 202 +++++++++++++++++- .../src/server/execution-composition.ts | 4 +- 2 files changed, 200 insertions(+), 6 deletions(-) diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 3e5de76f15..7a35945e45 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -31,7 +31,14 @@ 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 { deferred, waitFor as pollFor } from '@maka/core/test-only/async-primitives'; +import { RuntimeInteractionFailStopError } from '@maka/runtime/interaction-authority'; +import { + AgentGraphCoordinator, + agentGraphIdForRootSession, +} from '@maka/runtime/stream-graph-coordinator'; import { FAKE_ASK_USER_QUESTION_PROMPT, FAKE_HOLD_OPEN_PROMPT, @@ -63,8 +70,10 @@ import { runtimeHostFilesystemWorkerRuntime, stopOwnedWorkHubRoot, stopReplacedWorkHubRoot, + type ExecutionRuntimeHostCompositionDependencies, } from '../server/execution-composition.js'; -import { waitFor as pollFor } from '@maka/core/test-only/async-primitives'; +import type { RuntimeHostCompositionContext } from '../server/host-kernel.js'; +import { runAfterCurrentSessionAdmission } from '../server/session-admission-gate.js'; const require = createRequire(import.meta.url); const FAKE_CONNECTION_ID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'; @@ -1468,6 +1477,183 @@ test('production composition validates graph stop before aborting a claimed chil }); }); +test('interaction fail-stop drains graph operators after answer admission releases', { + 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<{ outsideAdmission: boolean; error?: unknown }> = []; + let settlement: HostedUserQuestionSettlement | undefined; + let retained = false; + let retainedAtShutdownRequest = false; + let requestedInsideAdmission = false; + let composition!: Awaited>; + const captured = await createCapturedExecutionComposition(owner, { + context: { + retainUntilProcessExit: () => { + retained = true; + }, + requestDrain: () => { + retainedAtShutdownRequest = retained; + let released = false; + runAfterCurrentSessionAdmission(() => { + released = true; + composition.beginDrain(); + }); + requestedInsideAdmission ||= !released; + }, + }, + dependencies: { + 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; + }, + }, + }); + composition = captured.composition; + const { manager } = captured; + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + const client = { + hostEpoch: 'execution-composition-test', + connectionId: 'interaction-drain-client', + principal: 'local_os_user' as const, + acquireResidency: () => ({ release() {} }), + }; + 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'; + const started = await composition.handlers['turn.start']( + { + sessionId: session.id, + turnId, + content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, + }, + client, + ); + 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); + let outsideAdmission = false; + runAfterCurrentSessionAdmission(() => { + outsideAdmission = true; + }); + const observation: (typeof stopObservations)[number] = { outsideAdmission }; + stopObservations.push(observation); + try { + await stopSession(sessionId, input); + } catch (error) { + observation.error = error; + throw error; + } finally { + stopped.resolve(); + } + }, + ); + await assert.rejects( + composition.handlers['interaction.answer']( + { + sessionId: session.id, + interactionId, + answer: { kind: 'question', answers: ['邀请制', '本周', '是'] }, + }, + client, + ), + (error: unknown) => + error instanceof RuntimeInteractionFailStopError && error.authorityFailure === failure, + ); + await stopped.promise; + assert.equal(retained, true); + assert.equal(retainedAtShutdownRequest, true); + assert.equal(requestedInsideAdmission, true); + assert.deepEqual(stopObservations, [{ outsideAdmission: true }]); + await assert.rejects( + composition.handlers['interaction.answer']( + { + sessionId: session.id, + interactionId, + answer: { kind: 'question', answers: ['邀请制', '本周', '是'] }, + }, + client, + ), + RuntimeInteractionFailStopError, + ); + } finally { + // Release the injected backend waiter; fail-stop intentionally cannot apply its continuation. + await settlement?.applyClosure('turn_stopped'); + await assert.rejects( + composition.close(), + /Unable to close Runtime Host execution composition/, + ); + } + }); +}); + function compositionContext(owner: InteractiveRootOwner) { return { owner, @@ -1578,7 +1764,13 @@ async function seedLegacyFakeBackendSession( return sessionId; } -async function createCapturedExecutionComposition(owner: InteractiveRootOwner): Promise<{ +async function createCapturedExecutionComposition( + owner: InteractiveRootOwner, + options: { + context?: Partial; + dependencies?: ExecutionRuntimeHostCompositionDependencies; + } = {}, +): Promise<{ composition: Awaited>; manager: SessionManager; }> { @@ -1593,9 +1785,11 @@ async function createCapturedExecutionComposition(owner: InteractiveRootOwner): // own; the deterministic one arrives through the same `primaryBackendFactory` // seam the Desktop E2E run uses. const composition = await createExecutionRuntimeHostComposition( - compositionContext(owner), + { ...compositionContext(owner), ...options.context }, {}, - { primaryBackendFactory: (backendContext) => new FakeBackend(backendContext) }, + options.dependencies ?? { + primaryBackendFactory: (backendContext) => new FakeBackend(backendContext), + }, ); await composition.recover(); if (!manager) throw new Error('Production execution composition did not construct Runtime'); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 7c0f8f90e0..6604380148 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -675,7 +675,7 @@ export async function createExecutionRuntimeHostComposition( if (poisonFailure) return; poisonFailure = error; context.retainUntilProcessExit(); - beginDrain(); + // The kernel starts domain drain after the current Session admission releases. context.requestDrain(); }, onSandboxBoundarySettled: (sessionId) => @@ -1140,7 +1140,7 @@ export async function createExecutionRuntimeHostComposition( poisonFailure = error; runtimePolicyActivation.poison(); context.retainUntilProcessExit(); - beginDrain(); + // The kernel starts domain drain after the current Session admission releases. context.requestDrain(); }, ...dependencies.oauthAuthorization, From efbc4ee1a942cb035719b370d804d0ce57fb9dfe Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Mon, 7 Sep 2026 17:06:04 +0800 Subject: [PATCH 4/6] refactor(runtime-host): simplify drain release and stop observation Wait for each still-active inherited admission without a fan-in counter or deduplication set. Observe the original stop promise immediately and await it after child lookup, removing the single-use settlement wrapper and stop factory. Ablation checks: removing kernel deferral fails the kernel regression; removing rejection observation fails both stop paths; keeping only the newest admission fails overlapping detached publication with admission reentry. Preserve these safeguards and cover both ancestor/child release orders. Generated-by: Codex --- .../__tests__/session-admission-gate.test.ts | 47 ++++++++++++++++++- .../src/server/session-admission-gate.ts | 17 ++----- packages/runtime/src/session-manager.ts | 24 ++++------ 3 files changed, 59 insertions(+), 29 deletions(-) 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 8826d1072f..43fbf1a3b9 100644 --- a/packages/runtime-host/src/__tests__/session-admission-gate.test.ts +++ b/packages/runtime-host/src/__tests__/session-admission-gate.test.ts @@ -154,7 +154,6 @@ 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(() => { @@ -260,3 +259,49 @@ test('treats detached work as outside the current admission', async () => { assert.deepEqual(order, ['active:start', 'detached:outside', 'active:end']); }); + +for (const firstToRelease of ['outer', 'inner'] as const) { + test(`drain waits for overlapping detached admissions when ${firstToRelease} releases first`, async () => { + const gate = new SessionAdmissionGate(); + const entered = deferred(); + const releaseOuter = deferred(); + const releaseInner = deferred(); + const order: string[] = []; + let inner!: Promise; + let stop!: Promise; + const outer = gate.run('outer', async () => { + inner = gate.enqueueDetached('inner', async () => { + runAfterCurrentSessionAdmission(() => { + order.push('drain'); + stop = gate.run('operator', () => { + order.push('stop'); + }); + }); + entered.resolve(); + await releaseInner.promise; + order.push('inner'); + }); + await releaseOuter.promise; + order.push('outer'); + }); + await entered.promise; + if (firstToRelease === 'outer') { + releaseOuter.resolve(); + await outer; + } else { + releaseInner.resolve(); + await inner; + } + assert.deepEqual(order, [firstToRelease]); + releaseOuter.resolve(); + releaseInner.resolve(); + await Promise.all([outer, inner]); + await stop; + assert.deepEqual(order, [ + firstToRelease, + firstToRelease === 'outer' ? 'inner' : 'outer', + 'drain', + 'stop', + ]); + }); +} diff --git a/packages/runtime-host/src/server/session-admission-gate.ts b/packages/runtime-host/src/server/session-admission-gate.ts index 985b9df878..c0b70edddd 100644 --- a/packages/runtime-host/src/server/session-admission-gate.ts +++ b/packages/runtime-host/src/server/session-admission-gate.ts @@ -47,20 +47,13 @@ const currentSessionAdmissions = new AsyncLocalStorage void): void { - const activeAdmissions = [ - ...new Set((currentSessionAdmissions.getStore() ?? []).filter((context) => context.active)), - ]; - if (activeAdmissions.length === 0) { - operation(); - return; - } - - let remaining = activeAdmissions.length; + const admissions = currentSessionAdmissions.getStore() ?? []; const afterRelease = () => { - remaining -= 1; - if (remaining === 0) operation(); + const active = admissions.find((context) => context.active); + if (active) active.afterRelease.add(afterRelease); + else operation(); }; - for (const context of activeAdmissions) context.afterRelease.add(afterRelease); + afterRelease(); } export class SessionAdmissionGate { diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index bc9af2f900..dcf063bf87 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -3597,10 +3597,9 @@ export class SessionManager { : undefined; await this.#stopSessionTree( sessionId, - () => - hostedAuthority - ? hostedAuthority.stopSession(sessionId, input) - : this.runtimeKernel.stopSession(sessionId, input), + hostedAuthority + ? hostedAuthority.stopSession(sessionId, input) + : this.runtimeKernel.stopSession(sessionId, input), (childSessionId) => hostedAuthority ? hostedAuthority.stopSession(childSessionId, input) @@ -3614,7 +3613,7 @@ export class SessionManager { : undefined; await this.#stopSessionTree( sessionId, - () => this.runtimeKernel.stopSession(sessionId, input), + this.runtimeKernel.stopSession(sessionId, input), (childSessionId) => authority ? authority.stopSession(childSessionId, input) @@ -3624,10 +3623,11 @@ export class SessionManager { async #stopSessionTree( sessionId: string, - stopOwn: () => Promise, + ownStop: Promise, stopChild: (childSessionId: string) => Promise, ): Promise { - const ownStop = observeSettlement(stopOwn()); + // Observe immediately while child lookup runs; await below still propagates the original error. + void ownStop.catch(() => undefined); let childStops: PromiseSettledResult[] = []; let childLookupError: unknown; try { @@ -3640,8 +3640,7 @@ export class SessionManager { } catch (error) { childLookupError = error; } - const ownStopResult = await ownStop; - if (ownStopResult.status === 'rejected') throw ownStopResult.reason; + await ownStop; const childStopError = childStops.find( (result): result is PromiseRejectedResult => result.status === 'rejected', )?.reason; @@ -5531,13 +5530,6 @@ 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 4fb4577c94da5a617dd4547279af7cf3ce18afdb Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Mon, 7 Sep 2026 23:19:10 +0800 Subject: [PATCH 5/6] test: simplify drain and stop rejection coverage --- .../runtime-resource-coordinator.test.ts | 32 +--- .../src/__tests__/session-manager.test.ts | 150 ++++++++---------- 2 files changed, 65 insertions(+), 117 deletions(-) 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 e2de7181c8..ea8868e7c1 100644 --- a/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts @@ -34,10 +34,7 @@ import { HostRuntimeResourceCoordinator, type HostRuntimeResourceCoordinatorInput, } from '../server/runtime-resource-coordinator.js'; -import { - runAfterCurrentSessionAdmission, - SessionAdmissionGate, -} from '../server/session-admission-gate.js'; +import { SessionAdmissionGate } from '../server/session-admission-gate.js'; const SESSION_ID = 'session-1'; const RUNTIME_REF = 'maka://runtime/background-tasks/shell-1'; @@ -261,33 +258,6 @@ 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; diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index d39119608b..33660a5e6d 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -3290,94 +3290,72 @@ 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 () => { + 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; - }, - } 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); + }), + 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.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); - } - }); + 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(); From af12ded17e220d5bbcf1f45fa1f0f29e5879af6f Mon Sep 17 00:00:00 2001 From: sungl <81428141+Sun-GLiang@users.noreply.github.com> Date: Tue, 8 Sep 2026 10:08:02 +0800 Subject: [PATCH 6/6] fix(runtime-host): detach composition drain at admission boundary Reuse the existing Session admission detach boundary for drain. Keep per-Session stop serialization and verify real kernel dispatch without imposing an all-ancestor release contract. Generated-by: Codex --- .../__tests__/execution-composition.test.ts | 191 ++++++++++-------- .../src/__tests__/host-kernel.test.ts | 32 --- .../runtime-resource-coordinator.test.ts | 26 +++ .../__tests__/session-admission-gate.test.ts | 153 +++----------- .../src/server/execution-composition.ts | 8 +- .../runtime-host/src/server/host-kernel.ts | 7 +- .../src/server/session-admission-gate.ts | 53 +---- 7 files changed, 166 insertions(+), 304 deletions(-) diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 9c5eae436c..ed44b1832c 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -39,7 +39,6 @@ import type { 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 { RuntimeInteractionFailStopError } from '@maka/runtime/interaction-authority'; import { AgentGraphCoordinator, agentGraphIdForRootSession, @@ -76,8 +75,10 @@ import { stopOwnedWorkHubRoot, stopReplacedWorkHubRoot, } from '../server/execution-composition.js'; -import type { RuntimeHostCompositionContext } from '../server/host-kernel.js'; -import { runAfterCurrentSessionAdmission } from '../server/session-admission-gate.js'; +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'; const require = createRequire(import.meta.url); const FAKE_CONNECTION_ID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'; @@ -1992,7 +1993,7 @@ test('production composition validates graph stop before aborting a claimed chil }); }); -test('interaction fail-stop drains graph operators after answer admission releases', { +test('interaction fail-stop stops graph operators through the kernel and releases ownership', { timeout: 10_000, }, async (t) => { await withCompositionRoot(async ({ root, owner }) => { @@ -2009,65 +2010,72 @@ test('interaction fail-stop drains graph operators after answer admission releas ); const published = deferred(); const stopped = deferred(); - const stopObservations: Array<{ outsideAdmission: boolean; error?: unknown }> = []; + const stopObservations: Array<{ error?: unknown }> = []; let settlement: HostedUserQuestionSettlement | undefined; let retained = false; let retainedAtShutdownRequest = false; - let requestedInsideAdmission = false; - let composition!: Awaited>; - const captured = await createCapturedExecutionComposition(owner, { - context: { - retainUntilProcessExit: () => { - retained = true; - }, - requestDrain: () => { - retainedAtShutdownRequest = retained; - let released = false; - runAfterCurrentSessionAdmission(() => { - released = true; - composition.beginDrain(); - }); - requestedInsideAdmission ||= !released; - }, - }, - 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); - }, + 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(); }, - }); - }; - return backend; - }, + 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 }, }); - composition = captured.composition; + assert.equal(connected.kind, 'connected'); + if (connected.kind !== 'connected') throw new Error('kernel connection unavailable'); const { manager } = captured; const stores = await openInteractiveExecutionStoresForWrite(owner.lease); - const client = { - hostEpoch: 'execution-composition-test', - connectionId: 'interaction-drain-client', - principal: 'local_os_user' as const, - acquireResidency: () => ({ release() {} }), - }; try { const session = await manager.createSession({ cwd: root, @@ -2078,13 +2086,20 @@ test('interaction fail-stop drains graph operators after answer admission releas }); await graph.toolsForSession(session.id); const turnId = 'interaction-drain-turn'; - const started = await composition.handlers['turn.start']( + // 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 }, }, - client, + { + hostEpoch: host.hostEpoch, + connectionId: 'interaction-drain-fixture', + principal: 'local_os_user', + acquireResidency: () => ({ release() {} }), + }, ); assert.equal(started.ok, true); const interactionId = await published.promise; @@ -2112,11 +2127,7 @@ test('interaction fail-stop drains graph operators after answer admission releas 'stopSession', async (sessionId: string, input: Parameters[1]) => { if (sessionId !== operator.header.id) return stopSession(sessionId, input); - let outsideAdmission = false; - runAfterCurrentSessionAdmission(() => { - outsideAdmission = true; - }); - const observation: (typeof stopObservations)[number] = { outsideAdmission }; + const observation: (typeof stopObservations)[number] = {}; stopObservations.push(observation); try { await stopSession(sessionId, input); @@ -2129,40 +2140,42 @@ test('interaction fail-stop drains graph operators after answer admission releas }, ); await assert.rejects( - composition.handlers['interaction.answer']( - { - sessionId: session.id, - interactionId, - answer: { kind: 'question', answers: ['邀请制', '本周', '是'] }, - }, - client, - ), + connected.connection.request('interaction.answer', { + sessionId: session.id, + interactionId, + answer: { kind: 'question', answers: ['邀请制', '本周', '是'] }, + }), (error: unknown) => - error instanceof RuntimeInteractionFailStopError && error.authorityFailure === failure, + error instanceof RuntimeHostOperationError && error.code === 'internal_failure', ); await stopped.promise; assert.equal(retained, true); assert.equal(retainedAtShutdownRequest, true); - assert.equal(requestedInsideAdmission, true); - assert.deepEqual(stopObservations, [{ outsideAdmission: true }]); - await assert.rejects( - composition.handlers['interaction.answer']( - { - sessionId: session.id, - interactionId, - answer: { kind: 'question', answers: ['邀请制', '本周', '是'] }, - }, - client, - ), - RuntimeInteractionFailStopError, - ); + assert.deepEqual(stopObservations, [{}]); } finally { // Release the injected backend waiter; fail-stop intentionally cannot apply its continuation. await settlement?.applyClosure('turn_stopped'); - await assert.rejects( - composition.close(), - /Unable to close Runtime Host execution composition/, - ); + 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(); } }); }); diff --git a/packages/runtime-host/src/__tests__/host-kernel.test.ts b/packages/runtime-host/src/__tests__/host-kernel.test.ts index 8d8234e190..f3c007aa05 100644 --- a/packages/runtime-host/src/__tests__/host-kernel.test.ts +++ b/packages/runtime-host/src/__tests__/host-kernel.test.ts @@ -87,7 +87,6 @@ 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, @@ -959,37 +958,6 @@ 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 ea8868e7c1..53f499143c 100644 --- a/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/runtime-resource-coordinator.test.ts @@ -754,6 +754,32 @@ describe('Host Runtime Resource coordinator', () => { }); }); +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: Partial< Pick< 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 43fbf1a3b9..626181be1d 100644 --- a/packages/runtime-host/src/__tests__/session-admission-gate.test.ts +++ b/packages/runtime-host/src/__tests__/session-admission-gate.test.ts @@ -20,11 +20,7 @@ import { deferred } from '@maka/core/test-only/async-primitives'; import assert from 'node:assert/strict'; import { test } from 'node:test'; -import { - runAfterCurrentSessionAdmission, - SessionAdmissionGate, - type SessionAdmissionLease, -} from '../server/session-admission-gate.js'; +import { SessionAdmissionGate } from '../server/session-admission-gate.js'; test('serializes operations for one Session', async () => { const gate = new SessionAdmissionGate(); @@ -153,71 +149,6 @@ test('rejects accidental admission re-entry instead of deadlocking', async () => }); }); -test('runs outside-admission work synchronously when no admission is active', () => { - 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(); @@ -243,65 +174,31 @@ test('work detached from an admission takes admissions of its own', async () => assert.deepEqual(order, ['active:start', 'active:end', 'detached:admitted']); }); -test('treats detached work as outside the current admission', async () => { +test('detached stop waits for its Session admission to release', async () => { const gate = new SessionAdmissionGate(); + const entered = deferred(); + const release = deferred(); 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'); + 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'); }); - - assert.deepEqual(order, ['active:start', 'detached:outside', 'active:end']); -}); - -for (const firstToRelease of ['outer', 'inner'] as const) { - test(`drain waits for overlapping detached admissions when ${firstToRelease} releases first`, async () => { - const gate = new SessionAdmissionGate(); - const entered = deferred(); - const releaseOuter = deferred(); - const releaseInner = deferred(); - const order: string[] = []; - let inner!: Promise; - let stop!: Promise; - const outer = gate.run('outer', async () => { - inner = gate.enqueueDetached('inner', async () => { - runAfterCurrentSessionAdmission(() => { - order.push('drain'); - stop = gate.run('operator', () => { - order.push('stop'); - }); - }); - entered.resolve(); - await releaseInner.promise; - order.push('inner'); - }); - await releaseOuter.promise; - order.push('outer'); - }); - await entered.promise; - if (firstToRelease === 'outer') { - releaseOuter.resolve(); - await outer; - } else { - releaseInner.resolve(); - await inner; - } - assert.deepEqual(order, [firstToRelease]); - releaseOuter.resolve(); - releaseInner.resolve(); - await Promise.all([outer, inner]); + await entered.promise; + await new Promise((resolve) => setImmediate(resolve)); + try { + assert.deepEqual(order, ['active']); + } finally { + release.resolve(); + await active; await stop; - assert.deepEqual(order, [ - firstToRelease, - firstToRelease === 'outer' ? 'inner' : 'outer', - 'drain', - '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 edaee39899..a8c37b5daa 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -681,7 +681,7 @@ export async function createExecutionRuntimeHostComposition( if (poisonFailure) return; poisonFailure = error; context.retainUntilProcessExit(); - // The kernel starts domain drain after the current Session admission releases. + // Route poison through the kernel; the composition drain entry detaches admission. context.requestDrain(); }, onSandboxBoundarySettled: (sessionId) => @@ -1146,7 +1146,7 @@ export async function createExecutionRuntimeHostComposition( poisonFailure = error; runtimePolicyActivation.poison(); context.retainUntilProcessExit(); - // The kernel starts domain drain after the current Session admission releases. + // Route poison through the kernel; the composition drain entry detaches admission. context.requestDrain(); }, ...dependencies.oauthAuthorization, @@ -2117,7 +2117,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/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index 8d93b96820..ca9ffb71bf 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -96,7 +96,6 @@ import { HostResidencyRegistry } from './host-residency-registry.js'; import type { PeerMeshNode } from '../peer-mesh/node.js'; import { createPeerMeshOperationHandlers } from './peer-mesh-authority.js'; import { createHostResourceCollector } from './host-resource-collector.js'; -import { runAfterCurrentSessionAdmission } from './session-admission-gate.js'; const DEFAULT_IDLE_GRACE_MS = 30_000; const DEFAULT_HANDSHAKE_TIMEOUT_MS = 5_000; @@ -361,11 +360,9 @@ export class RuntimeHostKernel { this.#cancelIdle(); this.#cancelInitialConnectionDeadline(); this.#armShutdownDeadline(); - } - runAfterCurrentSessionAdmission(() => { this.#beginCompositionDrain(); - this.#commitRequestedShutdownIfQuiescent(); - }); + } + this.#commitRequestedShutdownIfQuiescent(); } async #start(): Promise { diff --git a/packages/runtime-host/src/server/session-admission-gate.ts b/packages/runtime-host/src/server/session-admission-gate.ts index c0b70edddd..162c2f7b3a 100644 --- a/packages/runtime-host/src/server/session-admission-gate.ts +++ b/packages/runtime-host/src/server/session-admission-gate.ts @@ -27,7 +27,6 @@ export interface SessionAdmissionLease { interface SessionAdmissionContext { readonly sessionIds: ReadonlySet; - readonly afterRelease: Set<() => void>; readonly lease: SessionAdmissionLease; active: boolean; } @@ -43,19 +42,6 @@ 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 admissions = currentSessionAdmissions.getStore() ?? []; - const afterRelease = () => { - const active = admissions.find((context) => context.active); - if (active) active.afterRelease.add(afterRelease); - else operation(); - }; - afterRelease(); -} - export class SessionAdmissionGate { readonly #tails = new Map>(); readonly #context = new AsyncLocalStorage(); @@ -115,15 +101,7 @@ export class SessionAdmissionGate { * Turn reaching its own. Leaving the context here settles that by saying so. */ detach(operation: () => T): T { - 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, - ); - }); + return this.#context.exit(operation); } runAdmitted( @@ -142,13 +120,7 @@ export class SessionAdmissionGate { let task: Promise; try { - 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)), - ); + task = Promise.resolve(this.#context.run(state.context, operation)); } catch (error) { task = Promise.reject(error); } @@ -191,12 +163,7 @@ export class SessionAdmissionGate { const lease: SessionAdmissionLease = Object.freeze({ [sessionAdmissionLeaseBrand]: true as const, }); - const context: SessionAdmissionContext = { - sessionIds: ownedSessionIds, - afterRelease: new Set(), - lease, - active: true, - }; + const context: SessionAdmissionContext = { sessionIds: ownedSessionIds, lease, active: true }; const state: SessionAdmissionLeaseState = { sessionIds: ownedSessionIds, context, @@ -209,10 +176,7 @@ export class SessionAdmissionGate { let operationError: unknown; let operationFailed = false; try { - const inheritedAdmissions = currentSessionAdmissions.getStore() ?? []; - result = await currentSessionAdmissions.run([...inheritedAdmissions, context], () => - this.#context.run(context, () => operation(lease)), - ); + result = await this.#context.run(context, () => operation(lease)); } catch (error) { operationFailed = true; operationError = error; @@ -238,13 +202,8 @@ export class SessionAdmissionGate { context.active = false; this.#leases.delete(lease); release(); - 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); - } + for (const [sessionId, tail] of tails) { + if (this.#tails.get(sessionId) === tail) this.#tails.delete(sessionId); } } }