feat(server): gate work during update drain

apps/server/src/actionResume/ActionResume.test.ts

@@ -27,6 +27,7 @@ import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSna
 import * as ThreadActionResume from "../orchestration/ThreadActionResume.ts";
 import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts";
 import * as TerminalManager from "../terminal/Manager.ts";
+import { UpdateDrainAdmission } from "../updateDrain/UpdateDrainAdmission.ts";
 import * as ActionResume from "./ActionResume.ts";
 
 const threadId = ThreadId.make("thread-action-resume");
@@ -114,6 +115,8 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up
   const timeline: string[] = [];
   let terminalStatus: "running" | "exited" = "running";
   let failWrite = false;
+  let admissionClosed = false;
+  const admittedKinds: string[] = [];
   let terminalListener: ((event: TerminalEvent) => Effect.Effect<void>) | undefined;
 
   const dependencies = Layer.mergeAll(
@@ -169,6 +172,14 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up
       } as never,
     ]),
     ThreadActionResume.layer,
+    Layer.mock(UpdateDrainAdmission)({
+      admit: (kind, effect) =>
+        Effect.sync(() => admittedKinds.push(kind)).pipe(
+          Effect.andThen(
+            admissionClosed ? Effect.die("update drain is closed in this test") : effect,
+          ),
+        ),
+    }),
     NodeServices.layer,
   );
 
@@ -280,6 +291,35 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up
       .pipe(Effect.flip);
     assert.equal(earlyExit.reason, "launch_failed");
     assert.equal(registry.getLatest(threadId)?.runId, failedState?.runId);
+
+    terminalStatus = "running";
+    const blocked = yield* service.runProjectActionAndResume(
+      { threadId, providerInstanceId },
+      "qa",
+    );
+    admissionClosed = true;
+    yield* terminalListener!({
+      type: "exited",
+      threadId,
+      terminalId: blocked.terminalId,
+      exitCode: 0,
+      exitSignal: null,
+    });
+    assert.equal(dispatched.filter((command) => command.type === "thread.turn.start").length, 1);
+    assert.equal(admittedKinds.at(-1), "thread-turn");
+    assert.deepInclude(registry.getLatest(threadId), {
+      outcome: "succeeded",
+      delivery: "pending",
+    });
+
+    admissionClosed = false;
+    yield* service.retryPendingFollowUps;
+    yield* service.retryPendingFollowUps;
+    assert.equal(dispatched.filter((command) => command.type === "thread.turn.start").length, 2);
+    assert.deepInclude(registry.getLatest(threadId), {
+      outcome: "succeeded",
+      delivery: "delivered",
+    });
   }).pipe(Effect.provide(ActionResume.layer.pipe(Layer.provideMerge(dependencies))), Effect.scoped);
 });
 
@@ -420,6 +460,9 @@ it.effect("requires an explicit resume after a running Action is found on startu
         } as never,
       ]),
       ThreadActionResume.layer,
+      Layer.mock(UpdateDrainAdmission)({
+        admit: (_kind, effect) => effect,
+      }),
       NodeServices.layer,
     );
 

apps/server/src/actionResume/ActionResume.ts

@@ -37,6 +37,7 @@ import { OrchestrationEngineService } from "../orchestration/Services/Orchestrat
 import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";
 import { ThreadActionResumeService } from "../orchestration/ThreadActionResume.ts";
 import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts";
+import { UpdateDrainAdmission } from "../updateDrain/UpdateDrainAdmission.ts";
 
 export const ACTION_RESUME_ACTIVITY_KIND = "action.resume.lifecycle";
 
@@ -74,6 +75,7 @@ export class ActionResume extends Context.Service<
     readonly cancelByArchive: (threadId: ThreadId) => Effect.Effect<void>;
     readonly resumeInterrupted: (threadId: ThreadId) => Effect.Effect<void, ActionResumeError>;
     readonly discardInterrupted: (threadId: ThreadId) => Effect.Effect<void, ActionResumeError>;
+    readonly retryPendingFollowUps: Effect.Effect<void>;
     readonly countRunning: Effect.Effect<number>;
   }
 >()("t3/actionResume/ActionResume") {}
@@ -262,6 +264,7 @@ const make = Effect.gen(function* () {
   const registry = yield* ThreadActionResumeService;
   const terminals = yield* TerminalManager.TerminalManager;
   const providers = yield* ProviderRegistry;
+  const admission = yield* UpdateDrainAdmission;
   const mutex = yield* Semaphore.make(1);
   const decodeState = Schema.decodeUnknownEffect(ActionResumeState);
   const outputCaptureByRunId = new Map<string, ActionOutputCapture>();
@@ -376,7 +379,7 @@ const make = Effect.gen(function* () {
     mutex.withPermits(1)(deliverPendingUnlocked(threadId));
 
   const deliverPending = (threadId: ThreadId) =>
-    attemptDeliverPending(threadId).pipe(
+    admission.admit("thread-turn", attemptDeliverPending(threadId)).pipe(
       Effect.catchCause((cause) =>
         Effect.logWarning("Action follow-up delivery failed; it remains pending", {
           threadId,
@@ -385,6 +388,14 @@ const make = Effect.gen(function* () {
       ),
     );
 
+  const retryPendingFollowUps = Effect.suspend(() =>
+    Effect.forEach(
+      registry.listLatest().filter((state) => state.delivery === "pending"),
+      (state) => deliverPending(state.threadId),
+      { concurrency: 1, discard: true },
+    ),
+  );
+
   const finishUnlocked = Effect.fn("ActionResume.finishUnlocked")(function* (
     input: FinishActionInput,
   ) {
@@ -778,6 +789,7 @@ const make = Effect.gen(function* () {
       resumeInterruptedImpl(threadId).pipe(mapActionResumeError("resume the interrupted Action")),
     discardInterrupted: (threadId) =>
       discardInterruptedImpl(threadId).pipe(mapActionResumeError("discard the interrupted Action")),
+    retryPendingFollowUps,
     countRunning: Effect.sync(() => registry.countRunning()),
   });
 });

apps/server/src/auth/RpcAuthorization.test.ts

@@ -40,6 +40,9 @@ describe("RPC authorization scopes", () => {
     expect(requiredScopeForRpcMethod(WS_METHODS.serverCancelUpdateDrain)).toBe(
       AuthOrchestrationOperateScope,
     );
+    expect(requiredScopeForRpcMethod(WS_METHODS.serverClaimUpdateActivation)).toBe(
+      AuthOrchestrationOperateScope,
+    );
   });
 
   it("allows relay status reads without granting relay installation access", () => {

apps/server/src/auth/RpcAuthorization.ts

@@ -52,6 +52,7 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.serverGetBackgroundPolicy]: AuthOrchestrationReadScope,
   [WS_METHODS.serverStartUpdateDrain]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverCancelUpdateDrain]: AuthOrchestrationOperateScope,
+  [WS_METHODS.serverClaimUpdateActivation]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverGetUpdateDrainStatus]: AuthOrchestrationReadScope,
   [WS_METHODS.cloudGetRelayClientStatus]: AuthRelayReadScope,
   [WS_METHODS.cloudInstallRelayClient]: AuthRelayWriteScope,

apps/server/src/mcp/McpHttpServer.test.ts

@@ -1,7 +1,15 @@
 import { expect, it } from "@effect/vitest";
 import { NodeHttpServer } from "@effect/platform-node";
 import * as NodeServices from "@effect/platform-node/NodeServices";
-import { EnvironmentId, PreviewTabId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
+import {
+  EnvironmentId,
+  PreviewTabId,
+  ProviderInstanceId,
+  ThreadId,
+  UpdateDrainAdmissionError,
+  UpdateDrainRequestId,
+  UpdateDrainTargetVersion,
+} from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Stream from "effect/Stream";
@@ -11,6 +19,8 @@ import { HttpBody, HttpClient, HttpRouter, HttpServerResponse } from "effect/uns
 import * as McpHttpServer from "./McpHttpServer.ts";
 import * as McpInvocationContext from "./McpInvocationContext.ts";
 import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
+import { ActionResume } from "../actionResume/ActionResume.ts";
+import { UpdateDrainAdmission } from "../updateDrain/UpdateDrainAdmission.ts";
 
 const environmentId = EnvironmentId.make("environment-mcp-test");
 const threadId = ThreadId.make("thread-mcp-test");
@@ -39,6 +49,51 @@ const TestLayer = McpHttpServer.PreviewToolkitRegistrationLive.pipe(
   Layer.provideMerge(PreviewAutomationBroker.layer.pipe(Layer.provide(NodeServices.layer))),
 );
 
+it.effect("rejects MCP action launch while update drain admission is closed", () =>
+  Effect.gen(function* () {
+    let launched = false;
+    const maintenance = new UpdateDrainAdmissionError({
+      reason: "update_draining",
+      requestId: UpdateDrainRequestId.make("mcp-drain"),
+      targetVersion: UpdateDrainTargetVersion.make("1.2.3"),
+      message: "LastCode is draining for an update.",
+    });
+    const layer = McpHttpServer.ActionResumeToolkitRegistrationLive.pipe(
+      Layer.provideMerge(McpServer.McpServer.layer),
+      Layer.provideMerge(
+        Layer.mock(ActionResume)({
+          runProjectActionAndResume: () =>
+            Effect.sync(() => {
+              launched = true;
+              throw new Error("action launch should not run");
+            }),
+        }),
+      ),
+      Layer.provideMerge(
+        Layer.mock(UpdateDrainAdmission)({
+          admit: () => Effect.fail(maintenance),
+        }),
+      ),
+    );
+
+    const result = yield* Effect.gen(function* () {
+      const server = yield* McpServer.McpServer;
+      return yield* server
+        .callTool({ name: "run_project_action_and_resume", arguments: { actionId: "qa" } })
+        .pipe(
+          Effect.provideService(McpInvocationContext.McpInvocationContext, {
+            ...invocation,
+            capabilities: new Set(["action-resume"] as const),
+          }),
+          Effect.provideService(McpSchema.McpServerClient, client),
+        );
+    }).pipe(Effect.provide(layer));
+
+    expect(result.isError).toBe(true);
+    expect(launched).toBe(false);
+  }),
+);
+
 it("normalizes empty successful notification responses to accepted", () => {
   const notificationResponse = McpHttpServer.normalizeMcpHttpResponse(
     HttpServerResponse.text("", { status: 200, contentType: "application/json" }),

apps/server/src/mcp/toolkits/actionResume/handlers.ts

@@ -1,46 +1,69 @@
 import { ActionResumeError } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 
 import { ActionResume } from "../../../actionResume/ActionResume.ts";
+import { UpdateDrainAdmission } from "../../../updateDrain/UpdateDrainAdmission.ts";
 import * as McpInvocationContext from "../../McpInvocationContext.ts";
 import { ActionResumeToolkit } from "./tools.ts";
 
-const handlers = {
-  list_project_actions: () =>
-    Effect.gen(function* () {
-      const invocation = yield* McpInvocationContext.requireMcpCapability("action-resume");
-      const service = yield* Effect.serviceOption(ActionResume);
-      if (Option.isNone(service)) {
-        return yield* new ActionResumeError({
-          reason: "internal_error",
-          message: "Action resume is unavailable in this server runtime.",
-        });
-      }
-      const actions = yield* service.value.listProjectActions({
-        threadId: invocation.threadId,
-        providerInstanceId: invocation.providerInstanceId,
-      });
-      return { actions };
-    }),
-  run_project_action_and_resume: ({ actionId }) =>
-    Effect.gen(function* () {
-      const invocation = yield* McpInvocationContext.requireMcpCapability("action-resume");
-      const service = yield* Effect.serviceOption(ActionResume);
-      if (Option.isNone(service)) {
-        return yield* new ActionResumeError({
-          reason: "internal_error",
-          message: "Action resume is unavailable in this server runtime.",
-        });
-      }
-      return yield* service.value.runProjectActionAndResume(
-        {
+const duringDrain = (message: string) =>
+  new ActionResumeError({
+    reason: "internal_error",
+    message,
+  });
+
+const makeHandlers = (admission: UpdateDrainAdmission["Service"]) =>
+  ({
+    list_project_actions: () =>
+      Effect.gen(function* () {
+        const invocation = yield* McpInvocationContext.requireMcpCapability("action-resume");
+        const service = yield* Effect.serviceOption(ActionResume);
+        if (Option.isNone(service)) {
+          return yield* new ActionResumeError({
+            reason: "internal_error",
+            message: "Action resume is unavailable in this server runtime.",
+          });
+        }
+        const actions = yield* service.value.listProjectActions({
           threadId: invocation.threadId,
           providerInstanceId: invocation.providerInstanceId,
-        },
-        actionId,
-      );
-    }),
-} satisfies Parameters<typeof ActionResumeToolkit.toLayer>[0];
+        });
+        return { actions };
+      }),
+    run_project_action_and_resume: ({ actionId }) =>
+      Effect.gen(function* () {
+        const invocation = yield* McpInvocationContext.requireMcpCapability("action-resume");
+        const service = yield* Effect.serviceOption(ActionResume);
+        if (Option.isNone(service)) {
+          return yield* new ActionResumeError({
+            reason: "internal_error",
+            message: "Action resume is unavailable in this server runtime.",
+          });
+        }
+        return yield* admission
+          .admit(
+            "action-resume",
+            service.value.runProjectActionAndResume(
+              {
+                threadId: invocation.threadId,
+                providerInstanceId: invocation.providerInstanceId,
+              },
+              actionId,
+            ),
+          )
+          .pipe(
+            Effect.catchTags({
+              UpdateDrainAdmissionError: (error) => Effect.fail(duringDrain(error.message)),
+              UpdateDrainError: (error) => Effect.fail(duringDrain(error.message)),
+            }),
+          );
+      }),
+  }) satisfies Parameters<typeof ActionResumeToolkit.toLayer>[0];
 
-export const ActionResumeToolkitHandlersLive = ActionResumeToolkit.toLayer(handlers);
+export const ActionResumeToolkitHandlersLive = Layer.unwrap(
+  UpdateDrainAdmission.pipe(
+    Effect.map((admission) => ActionResumeToolkit.toLayer(makeHandlers(admission))),
+  ),
+);