diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 1cf9eb26b03c0..5882611e073ed 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -1270,6 +1270,9 @@ it.layer( }), ), getSnapshot: Ref.get(currentProvider), + isAwaitingFirstCheck: Ref.get(currentProvider).pipe( + Effect.map((provider) => provider === pendingProvider), + ), refresh: Ref.get(nextProvider).pipe( Effect.tap((provider) => Ref.set(currentProvider, provider)), ), @@ -1303,6 +1306,23 @@ it.layer( ? retainedModels : [customModel, ...cachedProvider.models]; assert.deepStrictEqual((yield* registry.getProviders)[0]?.models, expectedModels); + if (!restarted) { + const probeFields = ({ status, installed, auth, version }: ServerProvider) => ({ + status, + installed, + auth, + version, + }); + // Boot reads the probe state from the cache, not the pending snapshot, and never persists the pending one. + assert.deepStrictEqual( + probeFields((yield* registry.getProviders)[0]!), + probeFields(cachedProvider), + ); + assert.deepStrictEqual( + probeFields((yield* readProviderStatusCache(filePath))!), + probeFields(cachedProvider), + ); + } yield* registry.refreshInstance(instance.instanceId); assert.deepStrictEqual( @@ -1335,6 +1355,84 @@ it.layer( ), ), ); + it.effect("keeps a check that finished before boot over the cached status", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const filePath = yield* resolveProviderStatusCachePath({ + cacheDir: config.providerStatusCacheDir, + instanceId: cachedProvider.instanceId, + }); + const instance = { + instanceId: cachedProvider.instanceId, + driverKind: cachedProvider.driver, + continuationIdentity: { + driverKind: cachedProvider.driver, + continuationKey: "codex:instance:codex-personal", + }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: cachedProvider.driver, + packageName: null, + }), + ), + getSnapshot: Effect.succeed(failedProvider), + isAwaitingFirstCheck: Effect.succeed(false), + refresh: Effect.succeed(failedProvider), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + } satisfies ProviderInstance; + const layerInstanceRegistry = Layer.succeed( + ProviderInstanceRegistry.ProviderInstanceRegistry, + { + getInstance: (id) => + Effect.succeed(id === instance.instanceId ? instance : undefined), + listInstances: Effect.succeed([instance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: Effect.flatMap(PubSub.unbounded(), PubSub.subscribe), + }, + ); + + // First boot has no cache file; the second has a stale ready one. + for (const hasCache of [false, true]) { + if (hasCache) { + yield* writeProviderStatusCache({ + filePath, + provider: { ...cachedProvider, status: "ready" }, + }); + } + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + assert.strictEqual((yield* registry.getProviders)[0]?.status, failedProvider.status); + assert.strictEqual( + (yield* readProviderStatusCache(filePath))?.status, + failedProvider.status, + ); + }).pipe( + Effect.provide( + ProviderRegistry.layer.pipe( + Layer.provide(layerInstanceRegistry), + Layer.provide(layerUnusedMutator), + ), + ), + Effect.scoped, + ); + } + }).pipe( + Effect.provide( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-codex-probed-before-boot-", + }).pipe(Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))), + ), + ), + ); }); describe("Antigravity model inventories", () => { diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index fe5375136dc25..b36363de36f3a 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -386,10 +386,17 @@ export const layer = Layer.effect( } fallbackByInstance.set(source.instanceId, provider); } + // Read after the snapshots: a snapshot read as awaiting is the pre-check placeholder. + const awaitingBootInstances = new Set(); + for (const instance of bootInstances) { + if (yield* instance.snapshot.isAwaitingFirstCheck ?? Effect.succeed(false)) { + awaitingBootInstances.add(instance); + } + } - const cachedProviders = yield* Effect.forEach( + const bootProviders = yield* Effect.forEach( bootSources, - (source) => + (source, index) => Effect.gen(function* () { const filePath = yield* resolveProviderStatusCachePath({ cacheDir: config.providerStatusCacheDir, @@ -399,11 +406,15 @@ export const layer = Layer.effect( if (fallbackProvider === undefined) { return undefined; } + // A completed check is seeded by the boot upsert instead, which persists it. + const placeholder = awaitingBootInstances.has(bootInstances[index]!) + ? fallbackProvider + : undefined; return yield* readProviderStatusCache(filePath).pipe( Effect.provideService(FileSystem.FileSystem, fileSystem), Effect.flatMap((cachedProvider) => { if (cachedProvider === undefined) { - return Effect.void.pipe(Effect.as(undefined as ServerProvider | undefined)); + return Effect.succeed(placeholder); } const correlation = { cachedProvider, @@ -416,7 +427,7 @@ export const layer = Layer.effect( cachedInstanceId: cachedProvider.instanceId ?? null, driver: source.driverKind, cachedDriver: cachedProvider.driver ?? null, - }).pipe(Effect.as(undefined as ServerProvider | undefined)); + }).pipe(Effect.as(placeholder)); } return Effect.succeed(hydrateCachedProvider(correlation)); }), @@ -441,7 +452,7 @@ export const layer = Layer.effect( ModelManifest.BUNDLED_MODEL_MANIFEST.compatibility, ); const providersRef = yield* Ref.make>( - cachedProviders.map((provider) => classifyCompatibility(provider, initialManifest)), + bootProviders.map((provider) => classifyCompatibility(provider, initialManifest)), ); const workspaceRefreshesRef = yield* Ref.make< ReadonlyMap>> @@ -551,14 +562,8 @@ export const layer = Layer.effect( }); const compatibilityRefreshRunning = yield* Ref.make(false); - const syncProvider = Effect.fn("syncProvider")(function* ( - provider: ServerProvider, - options?: { - readonly publish?: boolean; - }, - ) { - const providers = yield* upsertProviders([provider], options); - // Reclassify the current model after fetch; republishing the old probe could overwrite newer health results. + // Reclassify the current model after fetch; republishing the old probe could overwrite newer health results. + const refreshCompatibility = Effect.gen(function* () { if (!(yield* Ref.getAndSet(compatibilityRefreshRunning, true))) { yield* manifestService.refresh.pipe( Effect.andThen(upsertProviders([], { persist: false })), @@ -566,6 +571,15 @@ export const layer = Layer.effect( Effect.forkIn(serviceScope), ); } + }); + const syncProvider = Effect.fn("syncProvider")(function* ( + provider: ServerProvider, + options?: { + readonly publish?: boolean; + }, + ) { + const providers = yield* upsertProviders([provider], options); + yield* refreshCompatibility; return providers; }); @@ -861,6 +875,13 @@ export const layer = Layer.effect( Effect.gen(function* () { const source = buildSnapshotSource(instance); const provider = yield* source.getSnapshot; + // Boot shows the cached status until this instance's first check publishes a result. + if ( + awaitingBootInstances.has(instance) && + (yield* instance.snapshot.isAwaitingFirstCheck ?? Effect.succeed(false)) + ) { + return yield* refreshCompatibility; + } yield* correlateSnapshotWithSource(source, provider).pipe( Effect.flatMap(syncProvider), ); @@ -918,7 +939,10 @@ export const layer = Layer.effect( Scope.provide(serviceScope), ); - yield* upsertProviders(fallbackProviders, { publish: false }); + yield* upsertProviders( + fallbackProviders.filter((_, index) => !awaitingBootInstances.has(bootInstances[index]!)), + { publish: false }, + ); const instanceChanges = yield* instanceRegistry.subscribeChanges; yield* syncLiveSources; yield* Stream.runForEach( diff --git a/packages/provider-core/src/server/managedProvider.test.ts b/packages/provider-core/src/server/managedProvider.test.ts index c9e5a6e01390d..7a7e08ceddfb7 100644 --- a/packages/provider-core/src/server/managedProvider.test.ts +++ b/packages/provider-core/src/server/managedProvider.test.ts @@ -272,6 +272,7 @@ describe("makeManagedServerProvider", () => { const initial = yield* provider.getSnapshot; assert.deepStrictEqual(initial, initialSnapshot); + assert.strictEqual(yield* provider.isAwaitingFirstCheck!, true); const updatesFiber = yield* Stream.take(provider.streamChanges, 1).pipe( Stream.runCollect, @@ -287,6 +288,7 @@ describe("makeManagedServerProvider", () => { assert.deepStrictEqual(updates, [refreshedSnapshot]); assert.deepStrictEqual(latest, refreshedSnapshot); + assert.strictEqual(yield* provider.isAwaitingFirstCheck!, false); assert.strictEqual(yield* Ref.get(checkCalls), 1); }), ).pipe(Effect.provide(layerAlwaysRunTest)), diff --git a/packages/provider-core/src/server/managedProvider.ts b/packages/provider-core/src/server/managedProvider.ts index a766c95d62535..03cd1491806fe 100644 --- a/packages/provider-core/src/server/managedProvider.ts +++ b/packages/provider-core/src/server/managedProvider.ts @@ -28,6 +28,7 @@ interface ProviderSnapshotState { readonly enrichmentGeneration: number; // null preserves an explicitly cleared plan across probes and enrichment. readonly usageAuthLabel: string | null | undefined; + readonly checked: boolean; } function withUsageLimits( @@ -79,6 +80,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( snapshot: initialSnapshot, enrichmentGeneration: 0, usageAuthLabel: undefined, + checked: false, }); const settingsRef = yield* Ref.make(initialSettings); const enrichmentFiberRef = yield* Ref.make | null>(null); @@ -218,7 +220,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( ); return [ { snapshot, generation }, - { snapshot, enrichmentGeneration: generation, usageAuthLabel }, + { snapshot, enrichmentGeneration: generation, usageAuthLabel, checked: true }, ] as const; }, ); @@ -425,6 +427,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( return { resolveMaintenance: input.resolveMaintenance, getSnapshot: Ref.get(snapshotStateRef).pipe(Effect.map((state) => state.snapshot)), + isAwaitingFirstCheck: Ref.get(snapshotStateRef).pipe(Effect.map((state) => !state.checked)), refresh: input .resolveMaintenance({ fresh: true }) .pipe(Effect.andThen(refreshSnapshot), Effect.tapError(Effect.logError), Effect.orDie), diff --git a/packages/provider-core/src/server/snapshot.ts b/packages/provider-core/src/server/snapshot.ts index 6cb7ea15c5b10..2af61f3adcd07 100644 --- a/packages/provider-core/src/server/snapshot.ts +++ b/packages/provider-core/src/server/snapshot.ts @@ -21,6 +21,8 @@ export interface ManagedServerProvider { readonly fresh?: boolean; }) => Effect.Effect; readonly getSnapshot: Effect.Effect; + /** True while getSnapshot still returns the placeholder from before the first completed check. */ + readonly isAwaitingFirstCheck?: Effect.Effect; readonly refresh: Effect.Effect; readonly refreshUsage?: Effect.Effect; readonly streamChanges: Stream.Stream;