diff --git a/apps/server/src/provider/ClaudeProvider.ts b/apps/server/src/provider/ClaudeProvider.ts index f969544fad064..70ebe8100a159 100644 --- a/apps/server/src/provider/ClaudeProvider.ts +++ b/apps/server/src/provider/ClaudeProvider.ts @@ -588,12 +588,15 @@ export const probeClaudeWorkspaceSnapshot = Effect.fn("probeClaudeWorkspaceSnaps machineSnapshot: ServerProvider, cwd: string, environment?: NodeJS.ProcessEnv, + publishPartial?: (snapshot: ProviderWorkspaceSnapshot) => Effect.Effect, ): Effect.fn.Return { if (!claudeSettings.enabled) return machineSnapshot; const skills = yield* discoverClaudeSkills(claudeSettings, cwd, environment); const fileSystem = yield* FileSystem.FileSystem; // A deleted worktree can never answer the probe. Pending would make clients rescan it forever. const cwdExists = yield* fileSystem.exists(cwd).pipe(Effect.orElseSucceed(() => true)); + // The capability probe starts the CLI and can take seconds. + if (cwdExists && publishPartial) yield* publishPartial({ ...machineSnapshot, skills }); const probeResult = cwdExists ? yield* probeClaudeCapabilities(claudeSettings, environment, cwd, false).pipe(Effect.result) : undefined; diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index af20b76754943..5495986bf5858 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -382,10 +382,16 @@ export const ClaudeDriver: ProviderDriver + snapshotForCwd: (cwd, publishPartial) => snapshot.getSnapshot.pipe( Effect.flatMap((machineSnapshot) => - probeClaudeWorkspaceSnapshot(effectiveConfig, machineSnapshot, cwd, processEnv), + probeClaudeWorkspaceSnapshot( + effectiveConfig, + machineSnapshot, + cwd, + processEnv, + publishPartial, + ), ), Effect.provideService(FileSystem.FileSystem, fileSystem), Effect.provideService(Path.Path, path), diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 564a90f40e6ce..d4c1fca6366dd 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -2486,6 +2486,147 @@ it.layer( }), ); + it.effect("publishes an evicted folder's skills before its slow command probe ends", () => + Effect.gen(function* () { + const driver = ProviderDriverKind.make("claudeAgent"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + const machineSkills = [{ name: "personal", path: "/home/me/SKILL.md", enabled: true }]; + const projectSkills = [{ name: "project", path: "/a/.claude/SKILL.md", enabled: true }]; + const provider = { + instanceId, + driver, + status: "ready", + enabled: true, + installed: true, + auth: { status: "authenticated" }, + checkedAt: "2026-06-10T00:00:00.000Z", + version: "1.0.0", + models: [], + slashCommands: [COMPACT_SLASH_COMMAND], + skills: machineSkills, + } as const satisfies ServerProvider; + const probeCommand = { name: "review", description: "Review changes" }; + const probeStarted = yield* Deferred.make(); + const releaseProbe = yield* Deferred.make(); + const slowProbeCwd = yield* Ref.make(null); + // Claude reads skills from disk quickly, then waits on a CLI probe for + // its commands. A driver may hand the registry what it already knows. + const snapshotForCwd = ( + cwd: string, + publishPartial?: (snapshot: ProviderWorkspaceSnapshot) => Effect.Effect, + ) => + Effect.gen(function* () { + const skills = cwd === "/a" ? projectSkills : []; + if ((yield* Ref.get(slowProbeCwd)) === cwd) { + if (publishPartial) yield* publishPartial({ ...provider, skills }); + yield* Deferred.succeed(probeStarted, undefined); + yield* Deferred.await(releaseProbe); + } + return { + ...provider, + skills, + slashCommands: [COMPACT_SLASH_COMMAND, probeCommand], + slashCommandsPending: false, + } satisfies ProviderWorkspaceSnapshot; + }); + const instance: ProviderInstance = { + instanceId, + driverKind: driver, + continuationIdentity: { driverKind: driver, continuationKey: "claude:instance:claude" }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + getSnapshot: Effect.succeed(provider), + refresh: Effect.succeed(provider), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + snapshotForCwd, + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + }; + const layerInstanceRegistry = Layer.succeed( + ProviderInstanceRegistry.ProviderInstanceRegistry, + { + getInstance: (requestedId) => + Effect.succeed(requestedId === instanceId ? instance : undefined), + listInstances: Effect.succeed([instance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: Effect.flatMap(PubSub.unbounded(), PubSub.subscribe), + }, + ); + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtimeServices = yield* Layer.build( + ProviderRegistry.layer.pipe( + Layer.provide(layerUnusedMutator), + Layer.provideMerge(layerInstanceRegistry), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-evicted-workspace-", + }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ).pipe(Scope.provide(scope)); + + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + const workspaceA = Effect.map(registry.getProviders, (providers) => + providers[0]?.workspaceSnapshots?.find((snapshot) => snapshot.cwd === "/a"), + ); + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/a" }); + assert.deepStrictEqual((yield* workspaceA)?.skills, projectSkills); + + // Sixteen other worktrees push /a out of the capped snapshot list, + // so clients fall back to the machine's personal skills for /a. + for (let index = 0; index < 16; index++) { + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: `/other-${index}` }); + } + assert.strictEqual(yield* workspaceA, undefined); + + yield* Ref.set(slowProbeCwd, "/a"); + // An on-demand client sees /a only in its reply, which comes back with the partial. + const reply = yield* registry.refreshWorkspaceSnapshot({ + instanceId, + cwd: "/a", + untilPartial: true, + }); + const replied = reply[0]?.workspaceSnapshots?.find((snapshot) => snapshot.cwd === "/a"); + assert.deepStrictEqual(replied?.skills, projectSkills); + assert.strictEqual(replied?.slashCommandsPending, true); + yield* Deferred.await(probeStarted); + // The client's follow-up request joins the running scan for the rest. + const rescan = yield* registry + .refreshWorkspaceSnapshot({ instanceId, cwd: "/a" }) + .pipe(Effect.forkChild({ startImmediately: true })); + + // The partial entry can be evicted again before the probe finishes. + for (let index = 0; index < 16; index++) { + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: `/later-${index}` }); + } + assert.strictEqual(yield* workspaceA, undefined); + + yield* Deferred.succeed(releaseProbe, undefined); + const settled = (yield* Fiber.join(rescan))[0]?.workspaceSnapshots?.find( + (snapshot) => snapshot.cwd === "/a", + ); + assert.deepStrictEqual(settled, yield* workspaceA); + assert.deepStrictEqual(settled?.skills, projectSkills); + assert.deepStrictEqual(settled?.slashCommands, [COMPACT_SLASH_COMMAND, probeCommand]); + assert.strictEqual(settled?.slashCommandsPending, undefined); + }).pipe(Effect.provide(runtimeServices)); + }), + ); + it.effect("rescans a folder whose model catalog could not be read before its TTL", () => Effect.gen(function* () { const driver = ProviderDriverKind.make("opencode"); diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index ba9ce9b2b6af7..d3807e8a5329d 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -89,6 +89,8 @@ export class ProviderRegistry extends Context.Service< /** Directory the driver scans; the snapshot stays keyed by `cwd`. Defaults to `cwd`. */ readonly scanCwd?: string; readonly fresh?: boolean; + /** Return once the scan publishes partial results; the scan keeps running to completion. */ + readonly untilPartial?: boolean; }) => Effect.Effect>; /** @@ -1041,6 +1043,7 @@ export const layer = Layer.effect( readonly cwd: string; readonly scanCwd?: string; readonly fresh?: boolean; + readonly untilPartial?: boolean; }) { if (input.fresh) { yield* updateProviders((providers) => @@ -1073,6 +1076,7 @@ export const layer = Layer.effect( const snapshotForCwd = instance?.snapshotForCwd; if (!instance || !snapshotForCwd) return providers; const token = yield* Deferred.make(); + const partialPublished = yield* Deferred.make>(); // A machine refresh can change this cwd's snapshot, so the scan publishes against the result. const refreshMachineSnapshot = input.fresh ? (instance.invalidateCaches ?? Effect.void).pipe( @@ -1085,39 +1089,55 @@ export const layer = Layer.effect( ), ) : Effect.succeed(scannedFrom); - const runScan = refreshMachineSnapshot.pipe( - Effect.flatMap((scanBase) => - Effect.map(snapshotForCwd(input.scanCwd ?? input.cwd), (scopedSnapshot) => ({ - scanBase, - scopedSnapshot, - })), - ), - // Scans fail on error; returned inventory remains valid even when provider health reports an error. - Effect.flatMap(({ scanBase, scopedSnapshot }) => - instanceRegistry.getInstance(input.instanceId).pipe( - Effect.flatMap((currentInstance) => { - if (currentInstance !== instance) return Ref.get(providersRef); - return updateProviders((currentProviders) => { - // Read ownership inside the update so a newer claim cannot race publication. - if (Ref.getUnsafe(workspaceRefreshesRef).get(instance)?.get(input.cwd) !== token) { - return currentProviders; + // Results replace what the scan started from or last published, so neither a partial + // publication nor its capacity eviction makes the final one look stale. + let scanBase = scannedFrom; + let publishedFrom = scannedFrom; + const publish = (scopedSnapshot: ProviderWorkspaceSnapshot) => + instanceRegistry.getInstance(input.instanceId).pipe( + Effect.flatMap((currentInstance) => { + if (currentInstance !== instance) return Ref.get(providersRef); + return updateProviders((currentProviders) => { + // Read ownership inside the update so a newer claim cannot race publication. + if (Ref.getUnsafe(workspaceRefreshesRef).get(instance)?.get(input.cwd) !== token) { + return currentProviders; + } + return currentProviders.map((candidate) => { + if (candidate.instanceId !== input.instanceId) return candidate; + const current = workspaceSnapshotOf(candidate); + if (!Equal.equals(current, publishedFrom) && !Equal.equals(current, scanBase)) { + return candidate; } - return currentProviders.map((candidate) => - candidate.instanceId === input.instanceId && - Equal.equals(workspaceSnapshotOf(candidate), scanBase) - ? upsertProviderWorkspaceSnapshot(candidate, input.cwd, { - ...scopedSnapshot, - checkedAt: scannedAt, - // Without driver command discovery, retain previous commands after failed health checks and retry. - slashCommandsPending: - scopedSnapshot.slashCommandsPending ?? scopedSnapshot.status === "error", - }) - : candidate, - ); + const next = upsertProviderWorkspaceSnapshot(candidate, input.cwd, { + ...scopedSnapshot, + checkedAt: scannedAt, + // Without driver command discovery, retain previous commands after failed health checks and retry. + slashCommandsPending: + scopedSnapshot.slashCommandsPending ?? scopedSnapshot.status === "error", + }); + publishedFrom = workspaceSnapshotOf(next); + return next; }); - }), + }); + }), + ); + const runScan = refreshMachineSnapshot.pipe( + Effect.tap((base) => + Effect.sync(() => { + scanBase = base; + publishedFrom = base; + }), + ), + Effect.andThen( + snapshotForCwd(input.scanCwd ?? input.cwd, (partial) => + publish({ ...partial, slashCommandsPending: true }).pipe( + Effect.flatMap((providers) => Deferred.succeed(partialPublished, providers)), + Effect.asVoid, + ), ), ), + // Scans fail on error; returned inventory remains valid even when provider health reports an error. + Effect.flatMap(publish), Effect.ensuring( Ref.update(workspaceRefreshesRef, (refreshes) => { const current = refreshes.get(instance); @@ -1154,7 +1174,10 @@ export const layer = Layer.effect( if ("inFlight" in claim) { return yield* Deferred.await(claim.inFlight).pipe(Effect.andThen(Ref.get(providersRef))); } - return yield* Fiber.join(claim.scan); + // Joining only stops waiting; the detached scan still publishes its final result. + return yield* input.untilPartial + ? Effect.raceFirst(Fiber.join(claim.scan), Deferred.await(partialPublished)) + : Fiber.join(claim.scan); }); const invalidateModelCaches = Effect.fn("ProviderRegistry.invalidateModelCaches")(function* ( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 66512ab6a22c0..df37c9abd6bfc 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2447,6 +2447,7 @@ export const layerWsRpc = ( cwd: input.cwd, scanCwd: yield* projectService.workspaceScanDirectory(input.cwd), fresh: input.fresh === true, + untilPartial: input.acceptPartialWorkspaceSnapshot === true, }) : input.reloadSecrets === true ? providerRegistry.reloadSecretsAndRefresh(input.instanceId) diff --git a/packages/client-runtime/src/providerSkills.test.ts b/packages/client-runtime/src/providerSkills.test.ts index acd4a4904c1d4..5b8c9d10c9ee7 100644 --- a/packages/client-runtime/src/providerSkills.test.ts +++ b/packages/client-runtime/src/providerSkills.test.ts @@ -205,7 +205,7 @@ describe("provider workspace snapshot refresh atom", () => { expect(run).toHaveBeenCalledTimes(1); expect(run).toHaveBeenCalledWith(expect.anything(), { environmentId, - input: { instanceId: provider.instanceId, cwd }, + input: { instanceId: provider.instanceId, cwd, acceptPartialWorkspaceSnapshot: true }, }); responses.push(() => withSnapshot("2026-01-01T00:00:00.000Z")); await vi.advanceTimersByTimeAsync(0); @@ -214,6 +214,32 @@ describe("provider workspace snapshot refresh atom", () => { unmount(); }); + it("joins a scan after its partial reply without spending the retry budget", async () => { + vi.useFakeTimers({ now: Date.parse("2026-01-01T00:00:00.000Z") }); + const partial = withSnapshot("2026-01-01T00:00:00.000Z", true); + const complete = withSnapshot("2026-01-01T00:00:00.000Z"); + // The first probe fails, so the joined request also comes back pending. + const { registry, atom, providersAtom, run, unmount } = setup(noSnapshot, [ + () => partial, + () => partial, + () => partial, + () => complete, + () => partial, + ]); + await vi.advanceTimersByTimeAsync(0); + expect(run).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(10_000); + registry.refresh(atom); + await vi.advanceTimersByTimeAsync(0); + expect(run).toHaveBeenCalledTimes(4); + expect(isProviderWorkspaceSnapshotScanning(registry.get(atom))).toBe(true); + registry.set(providersAtom, complete); + await vi.advanceTimersByTimeAsync(60_000); + expect(run).toHaveBeenCalledTimes(4); + expect(isProviderWorkspaceSnapshotScanning(registry.get(atom))).toBe(false); + unmount(); + }); + it.each(["already expired", "expires during discovery"])( "retries an unchanged catalog that %s after the short cooldown", async (expiry) => { diff --git a/packages/client-runtime/src/providerSkills.ts b/packages/client-runtime/src/providerSkills.ts index 0f91fc29f74e5..c5dfe565c6e80 100644 --- a/packages/client-runtime/src/providerSkills.ts +++ b/packages/client-runtime/src/providerSkills.ts @@ -170,6 +170,8 @@ export interface ProviderWorkspaceSnapshotRefreshState { readonly hadSnapshot: boolean; /** The server published a partial snapshot and is still discovering. */ readonly snapshotPending: boolean; + /** The last reply was an early partial, so the next request joins its scan. */ + readonly joining: boolean; } const initialProviderWorkspaceSnapshotRefreshState: ProviderWorkspaceSnapshotRefreshState = { @@ -178,6 +180,7 @@ const initialProviderWorkspaceSnapshotRefreshState: ProviderWorkspaceSnapshotRef failedAttempts: 0, hadSnapshot: false, snapshotPending: false, + joining: false, }; /** @@ -215,6 +218,7 @@ function planProviderWorkspaceSnapshotRefresh( : previous.failedAttempts, hadSnapshot: hasSnapshot, snapshotPending: pending, + joining: hasSnapshot ? false : previous.joining, }; const refresh = (requestedAt === null || now - requestedAt >= PROVIDER_WORKSPACE_SNAPSHOT_TTL_MS) && @@ -239,12 +243,15 @@ function failProviderWorkspaceSnapshotRefresh( // A newer request, or a snapshot loss, already superseded this one. if (state.requestedAt !== requestedAt) return state; const failedAttempts = state.failedAttempts + 1; - if (failedAttempts >= WORKSPACE_SNAPSHOT_MAX_ATTEMPTS) return { ...state, failedAttempts }; + if (failedAttempts >= WORKSPACE_SNAPSHOT_MAX_ATTEMPTS) { + return { ...state, failedAttempts, joining: false }; + } return { ...state, requestedAt: null, retryNotBefore: now + WORKSPACE_SNAPSHOT_RETRY_COOLDOWN_MS, failedAttempts, + joining: false, }; } @@ -295,7 +302,11 @@ export function createProviderWorkspaceSnapshotRefreshAtomFamily(input: { readonly refreshProviders: AtomCommand< { readonly environmentId: EnvironmentId; - readonly input: { readonly instanceId: ProviderInstanceId; readonly cwd: string }; + readonly input: { + readonly instanceId: ProviderInstanceId; + readonly cwd: string; + readonly acceptPartialWorkspaceSnapshot: boolean; + }; }, { readonly providers: ReadonlyArray }, unknown @@ -332,37 +343,44 @@ export function createProviderWorkspaceSnapshotRefreshAtomFamily(input: { } if (!plan.refresh) return plan.state; const registry = get.registry; - // Outlives this build on purpose: a rebuild must not drop the retry. - // Once nothing subscribes, the retry state dies with the node; writing - // would recreate and rebuild it, which scans again with no subscriber. - const retryLater = () => { + const joining = plan.state.joining; + // Outlives this build on purpose: a rebuild must not drop the retry. Once nothing + // subscribes, writing would recreate the node and scan with no subscriber. A partial + // reply arrives before the command probe ends: ask again at once to join that scan, + // and count only the joined reply as an attempt. + const retryLater = (pending: boolean) => { const node = registry.getNodes().get(refreshAtom); if (node === undefined || node.canBeRemoved) return; registry.set(refreshAtom, (state) => - // @effect-diagnostics-next-line globalDate:off - failProviderWorkspaceSnapshotRefresh(state, now, Date.now()), + pending && !joining && state.requestedAt === now + ? { ...state, requestedAt: null, joining: true } + : // @effect-diagnostics-next-line globalDate:off + failProviderWorkspaceSnapshotRefresh(state, now, Date.now()), ); }; void runAtomCommand( registry, input.refreshProviders, - { environmentId, input: { instanceId, cwd } }, + { environmentId, input: { instanceId, cwd, acceptPartialWorkspaceSnapshot: true } }, { reportFailure: false }, - ).then((result) => { - const refreshedProvider = - result._tag === "Success" ? findProvider(result.value.providers) : undefined; - const snapshot = - refreshedProvider && resolveProviderWorkspaceSnapshot(refreshedProvider, cwd); - // A new server timestamp proves progress even when the client clock is ahead. - // @effect-diagnostics-next-line globalDate:off - const completedAt = Date.now(); - const refreshed = - hasCompleteProviderWorkspaceSnapshot(refreshedProvider, cwd) && - (hasCurrentProviderWorkspaceSnapshot(refreshedProvider, cwd, completedAt) || - snapshot?.checkedAt !== - (provider && resolveProviderWorkspaceSnapshot(provider, cwd))?.checkedAt); - if (!refreshed) retryLater(); - }, retryLater); + ).then( + (result) => { + const refreshedProvider = + result._tag === "Success" ? findProvider(result.value.providers) : undefined; + const snapshot = + refreshedProvider && resolveProviderWorkspaceSnapshot(refreshedProvider, cwd); + // A new server timestamp proves progress even when the client clock is ahead. + // @effect-diagnostics-next-line globalDate:off + const completedAt = Date.now(); + const refreshed = + hasCompleteProviderWorkspaceSnapshot(refreshedProvider, cwd) && + (hasCurrentProviderWorkspaceSnapshot(refreshedProvider, cwd, completedAt) || + snapshot?.checkedAt !== + (provider && resolveProviderWorkspaceSnapshot(provider, cwd))?.checkedAt); + if (!refreshed) retryLater(snapshot?.slashCommandsPending === true); + }, + () => retryLater(false), + ); return plan.state; }, (ctx, update: Update) => { diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index adba6dbaca71b..c2fe573ade902 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -674,6 +674,11 @@ export const WsServerRefreshProvidersRpc = Rpc.make(WS_METHODS.serverRefreshProv * refresh, at most one TTL later. */ workspaceSnapshotsOnDemand: Schema.optional(Schema.Boolean), + /** + * Reply once the cwd scan publishes partial results, such as skills before commands. Sent by + * clients that ask again at once to join the scan; older clients wait for the full scan. + */ + acceptPartialWorkspaceSnapshot: Schema.optional(Schema.Boolean), /** As on `subscribeServerConfig`: add /usage-limits to the replied snapshot. */ usageLimitsCommand: Schema.optional(Schema.Boolean), /** As on `subscribeServerConfig`: a failed cwd scan replies with the machine catalog as a pending snapshot. */ diff --git a/packages/provider-core/src/server/driver.ts b/packages/provider-core/src/server/driver.ts index ac5ea46f39f16..bdf49b1c55524 100644 --- a/packages/provider-core/src/server/driver.ts +++ b/packages/provider-core/src/server/driver.ts @@ -47,6 +47,8 @@ export interface ProviderInstance { readonly snapshot: ManagedServerProvider; readonly snapshotForCwd?: ( cwd: string, + /** Publishes what a slow scan already knows, such as skills read before a CLI probe. */ + publishPartial?: (snapshot: ProviderWorkspaceSnapshot) => Effect.Effect, ) => Effect.Effect; readonly refreshModels?: () => Effect.Effect; readonly invalidateCaches?: Effect.Effect;