perf(server): defer maintenance until initial client delivery

apps/server/src/orchestration-v2/PullRequestSyncReactor.ts

@@ -33,7 +33,7 @@ import * as Stream from "effect/Stream";
 import * as GitManager from "../git/GitManager.ts";
 import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground, forkParked } from "../serverActivation.ts";
 import * as Orchestrator from "./Orchestrator.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 import { isTerminalRunStatus } from "./ThreadManagementService.ts";
@@ -845,7 +845,7 @@ export const make = Effect.gen(function* () {
         }
       }).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))),
     );
-    yield* forkParked(
+    yield* forkBackground(
       Effect.gen(function* () {
         yield* worker.enqueue("all");
         yield* worker.drain;

apps/server/src/orchestration-v2/ThreadPullRequestService.ts

@@ -28,7 +28,7 @@ import * as ProjectStore from "./ProjectStore.ts";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
 import { makeLinkedThreadsChangeDetector } from "../pullRequest/linkedThreads.ts";
 import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground, forkParked } from "../serverActivation.ts";
 import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts";
 import { makeNoPullRequestRechecks } from "./noPullRequestRechecks.ts";
 import * as Orchestrator from "./Orchestrator.ts";
@@ -488,7 +488,7 @@ export const make = Effect.gen(function* () {
         Effect.forever,
       ),
     );
-    yield* forkParked(
+    yield* forkBackground(
       Effect.gen(function* () {
         yield* worker.enqueue({ threadId: null, refresh: false, backfill: true });
         yield* worker.drain;

apps/server/src/orchestration-v2/ThreadSettlementService.ts

@@ -30,7 +30,7 @@ import * as GitManager from "../git/GitManager.ts";
 import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
 import * as ServerSettings from "../serverSettings.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground, forkParked } from "../serverActivation.ts";
 import * as TerminalManager from "../terminal/Manager.ts";
 import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
 import * as ProjectStore from "./ProjectStore.ts";
@@ -806,7 +806,7 @@ export const make = Effect.gen(function* () {
     const events = orchestrator.streamDomainEvents;
     const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
     let lastSettlementSettings = autoSettlementSettingsKey(initialSettings);
-    yield* forkParked(
+    yield* forkBackground(
       Effect.gen(function* () {
         yield* enqueueFullSweep();
         yield* fullSweeps.drain;

apps/server/src/provider/ProviderHostLive.ts

@@ -10,6 +10,7 @@ import { resolveAttachmentPath } from "../attachmentStore.ts";
 import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
 import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
 import * as ServerConfig from "../config.ts";
+import * as ServerActivation from "../serverActivation.ts";
 import * as ServerLifecycleEvents from "../serverLifecycleEvents.ts";
 import * as ServerSettings from "../serverSettings.ts";
 import * as DocumentText from "./DocumentText.ts";
@@ -28,6 +29,7 @@ export const layer = Layer.effect(
     const lifecycleEvents = yield* Effect.serviceOption(
       ServerLifecycleEvents.ServerLifecycleEvents,
     );
+    const background = yield* ServerActivation.ServerBackgroundStart;
     return ProviderHost.ProviderHost.of({
       paths: {
         cwd: config.cwd,
@@ -45,7 +47,12 @@ export const layer = Layer.effect(
       shouldRunBackgroundWork: backgroundPolicy.shouldRunScopeWork,
       ...Option.match(lifecycleEvents, {
         onNone: () => ({}),
-        onSome: (events) => ({ awaitServerReady: events.awaitReady }),
+        onSome: (events) => ({
+          awaitServerReady:
+            background === undefined
+              ? events.awaitReady
+              : Effect.andThen(events.awaitReady, background.awaitTurn),
+        }),
       }),
       resolveAttachmentPath: (attachment) =>
         resolveAttachmentPath({ attachmentsDir: config.attachmentsDir, attachment }),

apps/server/src/provider/ProviderRegistry.ts

@@ -122,6 +122,11 @@ export class ProviderRegistry extends Context.Service<
     }) => Effect.Effect<ReadonlyArray<ServerProvider>>;
 
     readonly streamChanges: Stream.Stream<ReadonlyArray<ServerProvider>>;
+    readonly subscribeChanges: Effect.Effect<
+      Stream.Stream<ReadonlyArray<ServerProvider>>,
+      never,
+      Scope.Scope
+    >;
   }
 >()("t3/provider/ProviderRegistry") {}
 
@@ -1221,6 +1226,9 @@ export const layer = Layer.effect(
       get streamChanges() {
         return Stream.fromPubSub(changesPubSub);
       },
+      get subscribeChanges() {
+        return Effect.map(PubSub.subscribe(changesPubSub), Stream.fromSubscription);
+      },
     } satisfies ProviderRegistry["Service"];
   }),
 );

apps/server/src/provider/providerMaintenanceRunner.test.ts

@@ -227,6 +227,7 @@ function makeRegistry(
         Effect.succeed(lifecycleFor(provider)),
       setProviderMaintenanceActionState,
       streamChanges: Stream.empty,
+      subscribeChanges: Effect.succeed(Stream.empty),
     };
 
     return {

apps/server/src/provider/testUtils/providerRegistryMock.ts

@@ -21,6 +21,7 @@ const makeProviderRegistryMock = (
     Effect.succeed(makeManualOnlyProviderMaintenanceCapabilities({ provider, packageName: null })),
   setProviderMaintenanceActionState: () => Effect.succeed(providers),
   streamChanges: Stream.empty,
+  subscribeChanges: Effect.succeed(Stream.empty),
 });
 
 export const layer = (providers: ReadonlyArray<ServerProvider> = []) =>

apps/server/src/pullRequest/PullRequestReadCache.ts

@@ -23,7 +23,7 @@ import * as PersistedCache from "effect/persistence/PersistedCache";
 import * as Persistence from "effect/persistence/Persistence";
 import * as ServerConfig from "../config.ts";
 import { detachStackFrame } from "../git/detachStackFrame.ts";
-import { forkParked } from "../serverActivation.ts";
+import { forkBackground } from "../serverActivation.ts";
 
 const CONCURRENT_READS = 512;
 /** Invalidation excludes only reads whose scopes share its stripe, not every PR read. */
@@ -243,7 +243,7 @@ export const layer = Layer.unwrap(
       Layer.provide(
         // Memoize cleanup acquisition with the store; Layer.tap runs per consumer.
         Layer.effectDiscard(
-          forkParked(
+          forkBackground(
             pruneExpiredEntryFiles(directory).pipe(
               Effect.catch((cause) =>
                 Effect.logWarning("Failed to prune PR cache files", { cause }),

apps/server/src/server.ts

@@ -898,7 +898,13 @@ const layerMakeServer = Layer.unwrap(
     const ownership = yield* ServerOwnership.acquireServerOwnership(config.serverRuntimeStatePath);
     const activation = yield* Deferred.make<void>();
     const awaitActivation = Deferred.await(activation);
-    const layerActivation = Layer.succeed(ServerActivation.ServerActivation, awaitActivation);
+    const layerActivation = Layer.merge(
+      Layer.succeed(ServerActivation.ServerActivation, awaitActivation),
+      Layer.effect(
+        ServerActivation.ServerBackgroundStart,
+        ServerActivation.makeBackgroundStart(awaitActivation),
+      ),
+    );
     const runtimeStateParked = yield* Deferred.make<void>();
     const tailscaleParked = yield* Deferred.make<void>();
     const cloudLinkParked = yield* Deferred.make<void>();

apps/server/src/serverActivation.test.ts

@@ -0,0 +1,39 @@
+import { expect, it } from "@effect/vitest";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
+import * as TestClock from "effect/testing/TestClock";
+
+import * as ServerActivation from "./serverActivation.ts";
+
+it.effect("keeps maintenance parked until activation and client delivery", () =>
+  Effect.gen(function* () {
+    const activation = yield* Deferred.make<void>();
+    const ran = yield* Deferred.make<void>();
+    const background = yield* ServerActivation.makeBackgroundStart(Deferred.await(activation));
+    yield* ServerActivation.forkBackground(Deferred.succeed(ran, undefined)).pipe(
+      Effect.provideService(ServerActivation.ServerActivation, Deferred.await(activation)),
+      Effect.provideService(ServerActivation.ServerBackgroundStart, background),
+    );
+
+    yield* TestClock.adjust("1 minute");
+    expect(yield* Deferred.isDone(ran)).toBe(false);
+    yield* Deferred.succeed(activation, undefined);
+    yield* TestClock.adjust("10 seconds");
+    expect(yield* Deferred.isDone(ran)).toBe(false);
+
+    yield* background.markClientServed;
+    yield* Deferred.await(ran);
+  }),
+);
+
+it.effect("releases background work after the fallback when no client connects", () =>
+  Effect.gen(function* () {
+    const background = yield* ServerActivation.makeBackgroundStart(Effect.void);
+    const waiter = yield* background.awaitTurn.pipe(Effect.forkScoped);
+    yield* TestClock.adjust("29 seconds");
+    expect(waiter.pollUnsafe()).toBeUndefined();
+    yield* TestClock.adjust("1 second");
+    yield* Fiber.join(waiter);
+  }),
+);

apps/server/src/serverActivation.ts

@@ -1,7 +1,7 @@
 import * as Context from "effect/Context";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
-import type * as Fiber from "effect/Fiber";
+import * as Fiber from "effect/Fiber";
 import type * as Scope from "effect/Scope";
 
 export class ServerActivation extends Context.Reference<Effect.Effect<void> | undefined>(
@@ -27,3 +27,41 @@ export const forkParkedFiber = Effect.fn("serverActivation.forkParkedFiber")(fun
 export const forkParked = <A, E, R>(
   effect: Effect.Effect<A, E, R>,
 ): Effect.Effect<void, never, Scope.Scope | R> => forkParkedFiber(effect).pipe(Effect.asVoid);
+
+interface BackgroundStart {
+  readonly awaitTurn: Effect.Effect<void>;
+  readonly markClientServed: Effect.Effect<void>;
+}
+
+export class ServerBackgroundStart extends Context.Reference<BackgroundStart | undefined>(
+  "t3/serverBackgroundStart",
+  { defaultValue: () => undefined },
+) {}
+
+export const makeBackgroundStart = Effect.fn("serverActivation.makeBackgroundStart")(function* (
+  activation: Effect.Effect<void>,
+) {
+  const clientServed = yield* Deferred.make<void>();
+  // Headless environments still run maintenance even if no client connects.
+  const ready = yield* activation.pipe(
+    Effect.andThen(Deferred.await(clientServed).pipe(Effect.timeoutOption("30 seconds"))),
+    Effect.forkScoped,
+  );
+  return {
+    awaitTurn: Fiber.join(ready).pipe(Effect.asVoid),
+    markClientServed: Deferred.succeed(clientServed, undefined).pipe(Effect.asVoid),
+  } satisfies BackgroundStart;
+});
+
+export const awaitBackgroundTurn = Effect.gen(function* () {
+  const background = yield* ServerBackgroundStart;
+  if (background !== undefined) yield* background.awaitTurn;
+});
+
+export const markClientServed = Effect.gen(function* () {
+  const background = yield* ServerBackgroundStart;
+  if (background !== undefined) yield* background.markClientServed;
+});
+
+export const forkBackground = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+  forkParked(Effect.andThen(awaitBackgroundTurn, effect));

apps/server/src/serverRuntimeStartup.ts

@@ -45,7 +45,7 @@ import * as ProjectStore from "./orchestration-v2/ProjectStore.ts";
 import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts";
 import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
 import * as ServerSettings from "./serverSettings.ts";
-import { forkParked, forkParkedFiber } from "./serverActivation.ts";
+import { forkBackground, forkParked, forkParkedFiber } from "./serverActivation.ts";
 import * as AnalyticsService from "./telemetry/AnalyticsService.ts";
 import { getServerOwners } from "./telemetry/Identify.ts";
 import * as ServerEnvironment from "./environment/ServerEnvironment.ts";
@@ -614,7 +614,7 @@ const make = Effect.fn("serverRuntimeStartup.make")(function* (options?: Startup
         : importPendingTranscripts
     ).pipe(forkParked);
 
-    yield* forkParked(
+    yield* forkBackground(
       Effect.gen(function* () {
         yield* Effect.logDebug("startup phase: recording startup heartbeat");
         yield* recordStartupHeartbeat.pipe(
@@ -623,6 +623,10 @@ const make = Effect.fn("serverRuntimeStartup.make")(function* (options?: Startup
           }),
           Effect.ignoreCause({ log: true }),
         );
+      }),
+    );
+    yield* forkParked(
+      Effect.gen(function* () {
         if (serverConfig.startupPresentation === "headless") {
           yield* Effect.logDebug("startup phase: headless access info");
           const accessInfo = yield* issueHeadlessServeAccessInfo();

apps/server/src/storageCleanup.ts

@@ -49,7 +49,7 @@ import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManag
 import { threadHasQueuedTurnStart } from "./orchestration-v2/ThreadSettlementService.ts";
 import { MAX_OVERDUE_ARM_MS } from "./orchestration-v2/UsageLimitRecoveryWorker.ts";
 import * as WorktreeBranchSync from "./orchestration-v2/WorktreeBranchSync.ts";
-import { forkParked } from "./serverActivation.ts";
+import { forkBackground, forkParked } from "./serverActivation.ts";
 import * as Settings from "./serverSettings.ts";
 import * as TerminalManager from "./terminal/Manager.ts";
 import { isFilesystemRoot, managedWorktreesDirectories } from "./worktreesDirectory.ts";
@@ -1800,7 +1800,7 @@ const make = Effect.gen(function* () {
     const changes = yield* settingsService.subscribeChanges;
     const events = engine.streamDomainEvents;
     let lastSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
-    yield* forkParked(
+    yield* forkBackground(
       requestSweep.pipe(
         Effect.andThen(worker.drain),
         Effect.repeat(Schedule.spaced("1 hour")),

apps/server/src/telemetry/AnalyticsService.ts

@@ -20,6 +20,7 @@ import * as HttpClientRequest from "effect/http/HttpClientRequest";
 import * as HttpClientResponse from "effect/http/HttpClientResponse";
 
 import packageJson from "../../package.json" with { type: "json" };
+import { awaitBackgroundTurn } from "../serverActivation.ts";
 import * as ServerConfig from "../config.ts";
 import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
 import { getTelemetryIdentity, type TelemetryOwner } from "./Identify.ts";
@@ -352,7 +353,7 @@ export const make = Effect.gen(function* () {
 
   yield* Effect.forever(Effect.sleep(FLUSH_INTERVAL_MS).pipe(Effect.andThen(flushWhenDue)), {
     disableYield: true,
-  }).pipe(Effect.forkScoped);
+  }).pipe((effect) => Effect.andThen(awaitBackgroundTurn, effect), Effect.forkScoped);
 
   yield* Effect.addFinalizer(() =>
     flush.pipe(Effect.timeout(SHUTDOWN_FLUSH_TIMEOUT), Effect.ignore),

apps/server/src/ws.test.ts

@@ -906,6 +906,7 @@ it.effect.each([
       const settings = yield* ServerSettings.ServerSettingsService;
       const keybindingsSubscribed = yield* Deferred.make<void>();
       const settingsSubscribed = yield* Deferred.make<void>();
+      const providersSubscribed = yield* Deferred.make<void>();
       const keybindingsChanges = keybindings.subscribeChanges.pipe(
         Effect.tap(() => Deferred.succeed(keybindingsSubscribed, undefined)),
       );
@@ -940,7 +941,9 @@ it.effect.each([
         }),
         Layer.mock(ProviderRegistry.ProviderRegistry)({
           getProviders: Effect.succeed([]),
-          streamChanges: Stream.never,
+          subscribeChanges: Deferred.succeed(providersSubscribed, undefined).pipe(
+            Effect.as(Stream.never),
+          ),
         }),
         Layer.mock(UsageLimitSources.UsageLimitSources)({ streamChanges: Stream.make([]) }),
         Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({
@@ -1045,6 +1048,8 @@ it.effect.each([
       });
       yield* Deferred.await(discoveryStarted);
       assert.isTrue(yield* Deferred.isDone(remoteDiscoveryStarted));
+      // Startup probes are released by this delivery, so their results must already be buffered.
+      assert.isTrue(yield* Deferred.isDone(providersSubscribed));
 
       const duringDiscovery = yield* settings.updateSettings({ defaultAutoPull: true });
       yield* Deferred.succeed(finishDiscovery, undefined);
@@ -1317,8 +1322,8 @@ it.effect("on-demand clients get provider config without workspace snapshots", (
       }),
       Layer.mock(ProviderRegistry.ProviderRegistry)({
         getProviders: Effect.sync(() => registryProviders),
-        streamChanges: Stream.fromQueue(providerChanges).pipe(
-          Stream.tap(() => Queue.offer(pulled, undefined)),
+        subscribeChanges: Effect.succeed(
+          Stream.fromQueue(providerChanges).pipe(Stream.tap(() => Queue.offer(pulled, undefined))),
         ),
         refresh: () => Effect.sync(() => registryProviders),
         refreshWorkspaceSnapshot: () => Effect.sync(() => registryProviders),

apps/server/src/ws.ts

@@ -183,6 +183,7 @@ import * as ProviderAuthService from "./provider/ProviderAuthService.ts";
 import * as ProviderInstallation from "./provider/providerInstallation.ts";
 import * as ServerSelfUpdate from "./cloud/selfUpdate.ts";
 import * as NpmRelease from "./cloud/npmRelease.ts";
+import * as ServerActivation from "./serverActivation.ts";
 import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
 import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
 import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver";
@@ -3315,6 +3316,7 @@ export const layerWsRpc = (
               // Buffer changes through discovery and the initial snapshot's send.
               const keybindingsChanges = yield* keybindings.subscribeChanges;
               const settingsChanges = yield* serverSettings.subscribeChanges;
+              const providerChanges = yield* providerRegistry.subscribeChanges;
               const remoteTargets = yield* Effect.forkScoped(remoteOpenTargets.resolveTargets(), {
                 startImmediately: true,
               });
@@ -3343,10 +3345,7 @@ export const layerWsRpc = (
               );
               const providerStatuses = Stream.zipLatestWith(
                 // Seed the changes-only stream so early source refreshes pair with current providers.
-                Stream.concat(
-                  Stream.fromEffect(providerRegistry.getProviders),
-                  providerRegistry.streamChanges,
-                ),
+                Stream.concat(Stream.fromEffect(providerRegistry.getProviders), providerChanges),
                 usageLimitSources.streamChanges.pipe(
                   Stream.changesWith(
                     usageLimitsCommand ? sameUsageLimitCommandCoverage : () => true,
@@ -3443,9 +3442,16 @@ export const layerWsRpc = (
                 rpcInitialItems([
                   { version: 1 as const, type: "snapshot" as const, config: initialConfig },
                 ]),
-                input.providerStatusDeltas === true
-                  ? updates.pipe(withProviderStatusDeltas(initialConfig.providers))
-                  : updates,
+                // Deferred maintenance starts once a client has its first snapshot.
+                Stream.unwrap(
+                  ServerActivation.markClientServed.pipe(
+                    Effect.as(
+                      input.providerStatusDeltas === true
+                        ? updates.pipe(withProviderStatusDeltas(initialConfig.providers))
+                        : updates,
+                    ),
+                  ),
+                ),
               );
             }),
           ),