feat(prime): install managed distributions safely

apps/mobile/src/features/connection/ConnectionEnvironmentRow.tsx

@@ -1,7 +1,8 @@
 import { SymbolView } from "../../components/AppSymbol";
 import { connectionStatusText } from "@t3tools/client-runtime/connection";
 import type { AtomCommandResult } from "@t3tools/client-runtime/state/runtime";
-import type { EnvironmentId } from "@t3tools/contracts";
+import { useAtomValue } from "@effect/atom-react";
+import type { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts";
 import * as Cause from "effect/Cause";
 import { AsyncResult } from "effect/unstable/reactivity";
 import { useCallback, useState } from "react";
@@ -12,6 +13,8 @@ import { AppText as Text, AppTextInput as TextInput } from "../../components/App
 import { cn } from "../../lib/cn";
 import { copyTextWithHaptic } from "../../lib/copyTextWithHaptic";
 import type { ConnectedEnvironmentSummary } from "../../state/remote-runtime-types";
+import { serverEnvironment } from "../../state/server";
+import { useEnvironmentQuery } from "../../state/query";
 import { ConnectionStatusDot } from "./ConnectionStatusDot";
 
 function connectionStatusLabel(environment: ConnectedEnvironmentSummary): string | null {
@@ -22,6 +25,82 @@ function connectionStatusLabel(environment: ConnectedEnvironmentSummary): string
   });
 }
 
+function PrimeHostMaintenanceInstanceStatus(props: {
+  readonly environmentId: EnvironmentId;
+  readonly instanceId: ProviderInstanceId;
+  readonly label: string;
+  readonly distributionMessage: string | null;
+}) {
+  const { data, error, isPending } = useEnvironmentQuery(
+    serverEnvironment.primeManagedMaintenance({
+      environmentId: props.environmentId,
+      input: { instanceId: props.instanceId },
+    }),
+  );
+  const operation = data?.scheduled ?? data?.operation ?? null;
+  return (
+    <View className="gap-1 rounded-[14px] border border-input-border bg-input px-3.5 py-3">
+      <Text className="text-xs font-t3-bold text-foreground">{props.label}</Text>
+      <Text className="text-xs leading-normal text-foreground-muted">
+        {data?.message ??
+          (isPending ? "Reading host maintenance status." : (error ?? "Status unavailable."))}
+      </Text>
+      {props.distributionMessage ? (
+        <Text className="text-xs leading-normal text-foreground-muted">
+          {props.distributionMessage}
+        </Text>
+      ) : null}
+      {operation ? (
+        <Text
+          className={cn(
+            "text-xs leading-normal",
+            operation.status === "failed" ? "text-adaptive-rose-500-400" : "text-foreground-muted",
+          )}
+        >
+          {operation.status.replaceAll("-", " ")} · {operation.message}
+        </Text>
+      ) : null}
+      {data?.guidance ? (
+        <Text className="text-xs leading-normal text-adaptive-amber-600-400">{data.guidance}</Text>
+      ) : null}
+    </View>
+  );
+}
+
+function PrimeHostMaintenanceStatus(props: { readonly environmentId: EnvironmentId }) {
+  const config = useAtomValue(serverEnvironment.configValueAtom(props.environmentId));
+  const primeProviders =
+    config?.providers.filter((provider) => provider.driver === "primeAgent") ?? [];
+  return (
+    <View className="gap-2 border-t border-border pt-3">
+      <Text className="text-2xs font-t3-bold tracking-[0.8px] uppercase text-foreground-muted">
+        Prime host maintenance
+      </Text>
+      {primeProviders.length > 0 ? (
+        primeProviders.map((provider) => (
+          <PrimeHostMaintenanceInstanceStatus
+            key={provider.instanceId}
+            environmentId={props.environmentId}
+            instanceId={provider.instanceId}
+            label={provider.displayName ?? "Prime Agent"}
+            distributionMessage={provider.distribution?.message ?? null}
+          />
+        ))
+      ) : (
+        <Text className="text-xs leading-normal text-foreground-muted">
+          {config === null
+            ? "Connect to read Prime maintenance status."
+            : "This environment reports no configured Prime Agent instance."}
+        </Text>
+      )}
+      <Text className="text-xs leading-normal text-foreground-muted">
+        Install, update, rollback, switch back, and cleanup are host operations. Open Provider
+        Settings in Pylon web or desktop for this environment. Active work is never interrupted.
+      </Text>
+    </View>
+  );
+}
+
 export function ConnectionEnvironmentRow(props: {
   readonly environment: ConnectedEnvironmentSummary;
   readonly expanded: boolean;
@@ -163,6 +242,8 @@ export function ConnectionEnvironmentRow(props: {
             </>
           )}
 
+          <PrimeHostMaintenanceStatus environmentId={props.environment.environmentId} />
+
           <View className="flex-row justify-end gap-2">
             {props.environment.isRelayManaged ? null : (
               <Pressable

apps/server/src/auth/RpcAuthorization.ts

@@ -51,6 +51,8 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.providerSetSessionAutoCompaction]: AuthOrchestrationOperateScope,
   [WS_METHODS.providerRefineSessionHarness]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverUpdateProvider]: AuthOrchestrationOperateScope,
+  [WS_METHODS.serverGetPrimeManagedMaintenance]: AuthOrchestrationReadScope,
+  [WS_METHODS.serverRunPrimeManagedMaintenance]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverStartProviderLogin]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverSubmitProviderLoginCode]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverCancelProviderLogin]: AuthOrchestrationOperateScope,

apps/server/src/provider/Drivers/PrimeAgentDriver.ts

@@ -44,6 +44,7 @@ import {
   makePrimeDistributionNetworkDependencies,
 } from "../prime/PrimeAgentDistributionVerifier.ts";
 import { makePrimeAgentDaemonManager } from "../prime/PrimeAgentDaemonManager.ts";
+import { resolvePrimeManagedBuildReceiptTarget } from "../prime/PrimeAgentManagedToolStore.ts";
 import {
   defaultProviderContinuationIdentity,
   type ProviderDriver,
@@ -177,19 +178,23 @@ export const PrimeAgentDriver: ProviderDriver<PrimeAgentSettings, PrimeAgentDriv
             }),
           );
           const publicPackage = yield* locatePrimeAgentPublicPackage(executablePath);
-          return yield* Effect.promise(() =>
-            inspectPrimeAgentDistribution(
+          return yield* Effect.promise(async () => {
+            const managedReceipt = await resolvePrimeManagedBuildReceiptTarget({
+              stateDir: serverConfig.stateDir,
+              packageRoot: publicPackage.packageRoot,
+            });
+            return await inspectPrimeAgentDistribution(
               {
-                stateDir: serverConfig.stateDir,
-                instanceId,
+                stateDir: managedReceipt?.stateDir ?? serverConfig.stateDir,
+                instanceId: managedReceipt?.instanceId ?? instanceId,
                 packageRoot: publicPackage.packageRoot,
                 platform: hostPlatform,
                 checkedAt: snapshot.checkedAt,
                 ...(enableUpdateChecks === undefined ? {} : { enableUpdateChecks }),
               },
               { loadLatestVerifiedPublication },
-            ),
-          );
+            );
+          });
         }).pipe(
           Effect.catchCause(() =>
             Effect.succeed({

apps/server/src/provider/Layers/ProviderService.test.ts

@@ -1160,6 +1160,94 @@ it.effect(
     }).pipe(Effect.provide(NodeServices.layer)),
 );
 
+it.effect("ProviderServiceLive fences starts and inventories exact instance quiescence", () => {
+  const codex = makeFakeCodexAdapter();
+  const cursor = makeFakeCodexAdapter(CURSOR_DRIVER);
+  const registry = makeAdapterRegistryMock({
+    [CODEX_DRIVER]: codex.adapter,
+    [CURSOR_DRIVER]: cursor.adapter,
+  });
+  const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
+    Layer.provide(SqlitePersistenceMemory),
+  );
+  const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer));
+  const providerLayer = Layer.merge(
+    makeProviderServiceLive().pipe(
+      Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)),
+      Layer.provide(directoryLayer),
+      Layer.provide(defaultServerSettingsLayer),
+      Layer.provide(serverConfigTestLayer),
+      Layer.provide(AnalyticsService.layerTest),
+      Layer.provide(
+        Layer.succeed(
+          ProviderEventLoggers.ProviderEventLoggers,
+          ProviderEventLoggers.NoOpProviderEventLoggers,
+        ),
+      ),
+    ),
+    directoryLayer,
+  );
+
+  return Effect.gen(function* () {
+    const provider = yield* ProviderService.ProviderService;
+    const activeThread = asThreadId("thread-maintenance-active");
+    yield* provider.startSession(activeThread, {
+      provider: CODEX_DRIVER,
+      providerInstanceId: codexInstanceId,
+      threadId: activeThread,
+      cwd: "/tmp/project-maintenance-active",
+      runtimeMode: "full-access",
+    });
+    assert.deepInclude(yield* provider.reserveProviderMaintenance!(codexInstanceId), {
+      status: "busy",
+    });
+
+    const quiescentInstanceId = ProviderInstanceId.make("cursor");
+    const reserved = yield* provider.reserveProviderMaintenance!(quiescentInstanceId);
+    assert.equal(reserved.status, "reserved");
+    if (reserved.status !== "reserved") return;
+    const fencedThread = asThreadId("thread-maintenance-fenced");
+    const fencedError = yield* provider
+      .startSession(fencedThread, {
+        provider: CURSOR_DRIVER,
+        providerInstanceId: quiescentInstanceId,
+        threadId: fencedThread,
+        cwd: "/tmp/project-maintenance-fenced",
+        runtimeMode: "full-access",
+      })
+      .pipe(Effect.flip);
+    assert.instanceOf(fencedError, ProviderValidationError);
+    assert.include(fencedError.message, "fenced for scheduled host maintenance");
+
+    const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
+    const recoveryThread = asThreadId("thread-maintenance-recovery-fenced");
+    yield* directory.upsert({
+      threadId: recoveryThread,
+      provider: CURSOR_DRIVER,
+      providerInstanceId: quiescentInstanceId,
+      runtimeMode: "full-access",
+    });
+    cursor.startSession.mockClear();
+    const recoveryError = yield* provider
+      .sendTurn({ threadId: recoveryThread, input: "must not recover", attachments: [] })
+      .pipe(Effect.flip);
+    assert.instanceOf(recoveryError, ProviderValidationError);
+    assert.include(recoveryError.message, "fenced for scheduled host maintenance");
+    assert.equal(cursor.startSession.mock.calls.length, 0);
+
+    yield* provider.releaseProviderMaintenance!(reserved.reservation);
+    yield* provider.startSession(fencedThread, {
+      provider: CURSOR_DRIVER,
+      providerInstanceId: quiescentInstanceId,
+      threadId: fencedThread,
+      cwd: "/tmp/project-maintenance-fenced",
+      runtimeMode: "full-access",
+    });
+    yield* provider.stopSession({ threadId: fencedThread });
+    yield* provider.stopSession({ threadId: activeThread });
+  }).pipe(Effect.provide(Layer.merge(providerLayer, NodeServices.layer)));
+});
+
 routing.layer("ProviderServiceLive routing", (it) => {
   it.effect("reclaims start reservations after long historical-thread churn", () =>
     Effect.gen(function* () {

apps/server/src/provider/Layers/ProviderService.ts

@@ -348,6 +348,27 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
         Effect.asVoid,
       )
     : Effect.void;
+  const instanceMaintenanceState = yield* SynchronizedRef.make(
+    new Map<ProviderInstanceId, { readonly pendingStarts: number; readonly fenceToken?: string }>(),
+  );
+  const beginInstanceStart = (instanceId: ProviderInstanceId) =>
+    SynchronizedRef.modify(instanceMaintenanceState, (current) => {
+      const state = current.get(instanceId) ?? { pendingStarts: 0 };
+      if (state.fenceToken !== undefined) return [false, current] as const;
+      const next = new Map(current);
+      next.set(instanceId, { ...state, pendingStarts: state.pendingStarts + 1 });
+      return [true, next] as const;
+    });
+  const finishInstanceStart = (instanceId: ProviderInstanceId) =>
+    SynchronizedRef.update(instanceMaintenanceState, (current) => {
+      const state = current.get(instanceId);
+      if (!state) return current;
+      const next = new Map(current);
+      const pendingStarts = Math.max(0, state.pendingStarts - 1);
+      if (pendingStarts === 0 && state.fenceToken === undefined) next.delete(instanceId);
+      else next.set(instanceId, { ...state, pendingStarts });
+      return next;
+    });
   const reserveStartSession = (threadId: ThreadId) =>
     SynchronizedRef.modify(startReservations, (current) => {
       const previous = current.get(threadId);
@@ -882,10 +903,16 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
       } as const;
     }
 
+    if (!(yield* beginInstanceStart(instanceId))) {
+      return yield* toValidationError(
+        input.operation,
+        `Provider instance '${instanceId}' is fenced for scheduled host maintenance.`,
+      );
+    }
     const recovered = yield* recoverSessionForThread({
       binding,
       operation: input.operation,
-    });
+    }).pipe(Effect.ensuring(finishInstanceStart(instanceId)));
     return {
       adapter: recovered.adapter,
       instanceId,
@@ -942,6 +969,12 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
         "ProviderService.startSession",
         parsed,
       );
+      if (!(yield* beginInstanceStart(resolvedInstanceId))) {
+        return yield* toValidationError(
+          "ProviderService.startSession",
+          `Provider instance '${resolvedInstanceId}' is fenced for scheduled host maintenance.`,
+        );
+      }
       let metricProvider = parsed.provider ?? String(resolvedInstanceId);
       yield* Effect.annotateCurrentSpan({
         "provider.operation": "start-session",
@@ -1130,7 +1163,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
             }),
           ),
         )
-        .pipe(Effect.ensuring(releaseStartReservation(threadId, reservation.token)));
+        .pipe(
+          Effect.ensuring(releaseStartReservation(threadId, reservation.token)),
+          Effect.ensuring(finishInstanceStart(resolvedInstanceId)),
+        );
     },
   );
 
@@ -2382,6 +2418,61 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
     },
   );
 
+  const releaseProviderMaintenance: ProviderServiceMethod<"releaseProviderMaintenance"> = (
+    reservation,
+  ) =>
+    SynchronizedRef.update(instanceMaintenanceState, (current) => {
+      const entry = [...current.entries()].find(
+        ([, state]) => state.fenceToken === reservation.token,
+      );
+      if (!entry) return current;
+      const [instanceId, state] = entry;
+      const next = new Map(current);
+      if (state.pendingStarts === 0) next.delete(instanceId);
+      else next.set(instanceId, { pendingStarts: state.pendingStarts });
+      return next;
+    });
+
+  const reserveProviderMaintenance: ProviderServiceMethod<"reserveProviderMaintenance"> = Effect.fn(
+    "reserveProviderMaintenance",
+  )(function* (instanceId) {
+    const token = NodeCrypto.randomUUID();
+    const fenced = yield* SynchronizedRef.modify(instanceMaintenanceState, (current) => {
+      const state = current.get(instanceId) ?? { pendingStarts: 0 };
+      if (state.fenceToken !== undefined || state.pendingStarts > 0) {
+        return [false, current] as const;
+      }
+      const next = new Map(current);
+      next.set(instanceId, { pendingStarts: 0, fenceToken: token });
+      return [true, next] as const;
+    });
+    if (!fenced) {
+      return {
+        status: "busy",
+        reasons: ["a provider session start or another maintenance reservation is pending"],
+      } as const;
+    }
+    const reservation = { token };
+    const sessions = yield* listSessionsForInstance(instanceId).pipe(
+      Effect.onError(() => releaseProviderMaintenance(reservation)),
+    );
+    const activeIncarnation = [...currentSessionIncarnations.values()].some(
+      (incarnation) => incarnation.instanceId === instanceId,
+    );
+    if (sessions.length > 0 || activeIncarnation) {
+      yield* releaseProviderMaintenance(reservation);
+      return {
+        status: "busy",
+        reasons: [
+          sessions.some((session) => session.activeTurnId !== undefined)
+            ? "an active or admitted provider turn exists"
+            : "an active provider session or owned runtime exists",
+        ],
+      } as const;
+    }
+    return { status: "reserved", reservation } as const;
+  });
+
   const getCapabilities: ProviderServiceMethod<"getCapabilities"> = (instanceId) =>
     registry.getByInstance(instanceId).pipe(Effect.map((adapter) => adapter.capabilities));
 
@@ -2560,6 +2651,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
     listSessions,
     getSessionContinuation,
     listSessionsForInstance,
+    reserveProviderMaintenance,
+    releaseProviderMaintenance,
     getCapabilities,
     getInstanceInfo,
     rollbackConversation,