fix(runtime-host): detach composition drain from admission

packages/runtime-host/src/__tests__/execution-composition.test.ts

@@ -36,7 +36,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 { 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,
@@ -69,7 +76,8 @@ import {
   stopOwnedWorkHubRoot,
   stopReplacedWorkHubRoot,
 } 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';
@@ -1984,6 +1992,181 @@ 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<string>();
+    const stopped = deferred<void>();
+    const stopObservations: Array<{ outsideAdmission: boolean; error?: unknown }> = [];
+    let settlement: HostedUserQuestionSettlement | undefined;
+    let retained = false;
+    let retainedAtShutdownRequest = false;
+    let requestedInsideAdmission = false;
+    let composition!: Awaited<ReturnType<typeof createExecutionRuntimeHostComposition>>;
+    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);
+              },
+            },
+          });
+        };
+        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<SessionManager['stopSession']>[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,
@@ -2097,6 +2280,10 @@ async function seedLegacyFakeBackendSession(
 async function createCapturedExecutionComposition(
   owner: InteractiveRootOwner,
   options: {
+    readonly context?: Pick<
+      RuntimeHostCompositionContext,
+      'retainUntilProcessExit' | 'requestDrain'
+    >;
     readonly safeBoundaryResume?: boolean;
     readonly primaryBackendFactory?: BackendFactory;
     readonly residencies?: HostResidencyRegistry;
@@ -2125,6 +2312,7 @@ async function createCapturedExecutionComposition(
       {
         ...compositionContext(owner),
         ...(residencies ? { acquireResidency: (label: string) => residencies.acquire(label) } : {}),
+        ...options.context,
       },
       {},
       { primaryBackendFactory },

packages/runtime-host/src/__tests__/host-kernel.test.ts

@@ -87,6 +87,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,
@@ -958,6 +959,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' });