fix(runtime-host): defer resource drain outside admission

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<void> | undefined;
+    let harness!: ReturnType<typeof createHarness>;
+    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'];

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<diagnostic truncated>');
+}

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<K extends OperationKey>(
     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<K extends OperationKey>(
         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<diagnostic truncated>');
-}

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;

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<void>((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();

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<void>[] = [];
     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<T>(items: readonly T[], max: number): T[] {
   return items.slice(items.length - max);
 }
 
+function observeSettlement<T>(promise: Promise<T>): Promise<PromiseSettledResult<T>> {
+  return promise.then(
+    (value) => ({ status: 'fulfilled', value }),
+    (reason: unknown) => ({ status: 'rejected', reason }),
+  );
+}
+
 function shellRunBashToolCallIds(messages: readonly StoredMessage[]): Set<string> {
   return new Set(
     messages.flatMap((message) =>