Skip to content

fix(server): refresh workspace skills with provider status #11532

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yashranaway wants to merge 2 commits into from
Closed

fix(server): refresh workspace skills with provider status #11532

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b194e63
fix(server): refresh workspace skills with provider status
yashranaway Sep 13, 2026
c4eb8b6
fix(server): discard refreshes from replaced providers
yashranaway Sep 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 46 additions & 15 deletions apps/server/src/provider/Layers/ProviderRegistry.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1413,7 +1413,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te
}),
);

it.effect("deduplicates cwd probes and clears snapshots when an instance rebuilds", () =>
it.effect("refreshes and deduplicates workspace snapshots", () =>
Effect.gen(function* () {
const driver = ProviderDriverKind.make("codex");
const instanceId = ProviderInstanceId.make("codex");
Expand Down Expand Up @@ -1446,6 +1446,10 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te
const returnPendingSnapshot = yield* Ref.make(true);
const probeStarted = yield* Deferred.make<void>();
const releaseProbe = yield* Deferred.make<void>();
const machineChanges = yield* PubSub.unbounded<ServerProvider>();
const holdRefresh = yield* Ref.make(false);
const refreshStarted = yield* Deferred.make<void>();
const releaseRefresh = yield* Deferred.make<void>();
const makeInstance = (
provider: ServerProvider,
snapshotForCwd: NonNullable<ProviderInstance["snapshotForCwd"]>,
Expand All @@ -1467,8 +1471,14 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te
}),
),
getSnapshot: Effect.succeed(provider),
refresh: Effect.succeed(provider),
streamChanges: Stream.empty,
refresh: Effect.gen(function* () {
if (yield* Ref.get(holdRefresh)) {
yield* Deferred.succeed(refreshStarted, undefined);
yield* Deferred.await(releaseRefresh);
}
return provider;
}),
streamChanges: Stream.fromPubSub(machineChanges),
applyUsageLimits: () => Effect.void,
},
snapshotForCwd,
Expand Down Expand Up @@ -1544,9 +1554,12 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te
.pipe(Effect.forkChild);
yield* Effect.yieldNow;
assert.strictEqual(yield* Ref.get(snapshotCalls), 2);
yield* registry.refreshInstance(instanceId);
yield* Deferred.succeed(releaseProbe, undefined);
yield* Fiber.join(firstRefresh);
yield* Fiber.join(duplicateRefresh);
assert.deepStrictEqual((yield* registry.getProviders)[0]?.workspaceSnapshots, []);
yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" });
const published = yield* Fiber.join(workspaceUpdate);
assert.strictEqual(published._tag, "Some");
const providers = yield* registry.getProviders;
Expand All @@ -1556,21 +1569,39 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te
scopedProvider.skills,
);
yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" });
assert.strictEqual(yield* Ref.get(snapshotCalls), 2);
assert.strictEqual(yield* Ref.get(snapshotCalls), 3);
yield* registry.refresh();
assert.deepStrictEqual((yield* registry.getProviders)[0]?.workspaceSnapshots, []);
const lateProvider = { ...machineProvider, checkedAt: "2026-06-10T00:01:30.000Z" };
const lateUpdate = yield* registry.streamChanges.pipe(
Stream.filter((providers) => providers[0]?.checkedAt === lateProvider.checkedAt),
Stream.runHead,
Effect.forkChild,
);
yield* Effect.yieldNow;
yield* PubSub.publish(machineChanges, lateProvider);
yield* Fiber.join(lateUpdate);
assert.deepStrictEqual((yield* registry.getProviders)[0]?.workspaceSnapshots, []);
yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" });
assert.strictEqual(yield* Ref.get(snapshotCalls), 4);

yield* Ref.set(holdRefresh, true);
const staleRefresh = yield* registry.refreshInstance(instanceId).pipe(Effect.forkChild);
yield* Deferred.await(refreshStarted);
const replacementUpdate = yield* registry.streamChanges.pipe(
Stream.filter((providers) => providers[0]?.checkedAt === rebuiltProvider.checkedAt),
Stream.runHead,
Effect.forkChild,
);
yield* Effect.yieldNow;
yield* Ref.set(instancesRef, [rebuiltInstance]);
yield* PubSub.publish(registryChanges, undefined);
let rebuilt = yield* registry.getProviders;
for (
let attempt = 0;
attempt < 50 && rebuilt[0]?.checkedAt !== rebuiltProvider.checkedAt;
attempt += 1
) {
yield* Effect.yieldNow;
rebuilt = yield* registry.getProviders;
}
assert.strictEqual(rebuilt[0]?.checkedAt, rebuiltProvider.checkedAt);
assert.strictEqual(rebuilt[0]?.workspaceSnapshots, undefined);
yield* Fiber.join(replacementUpdate);
yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" });
const beforeStaleResult = yield* registry.getProviders;
yield* Deferred.succeed(releaseRefresh, undefined);
yield* Fiber.join(staleRefresh);
assert.deepStrictEqual(yield* registry.getProviders, beforeStaleResult);
}).pipe(Effect.provide(runtimeServices));
}),
);
Expand Down
86 changes: 56 additions & 30 deletions apps/server/src/provider/Layers/ProviderRegistry.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,10 @@ const snapshotInstanceKey = (provider: ServerProvider): ProviderInstanceId => {
// after `ProviderInstanceRegistry` rebuilds an instance (e.g. because
// its settings changed), a fresh source rides the new PubSub instead
// of a closed one.
const buildSnapshotSource = (instance: ProviderInstance): ProviderSnapshotSource => ({
type LiveProviderSnapshotSource = ProviderSnapshotSource & { readonly instance: ProviderInstance };

const buildSnapshotSource = (instance: ProviderInstance): LiveProviderSnapshotSource => ({
instance,
instanceId: instance.instanceId,
driverKind: instance.driverKind,
getSnapshot: instance.snapshot.getSnapshot,
Expand Down Expand Up @@ -358,7 +361,7 @@ export const ProviderRegistryLive = Layer.effect(
);
const providersRef = yield* Ref.make<ReadonlyArray<ServerProvider>>(cachedProviders);
const workspaceRefreshesRef = yield* Ref.make<
ReadonlyMap<ProviderInstance, ReadonlySet<string>>
ReadonlyMap<ProviderInstance, ReadonlyMap<string, symbol>>
>(new Map());
const maintenanceActionStatesRef = yield* Ref.make<
ReadonlyMap<ProviderInstanceId, { readonly update?: ServerProviderUpdateState | undefined }>
Expand All @@ -375,7 +378,7 @@ export const ProviderRegistryLive = Layer.effect(
// interleave two passes clobbering each other's fiber bookkeeping.
const syncSemaphore = yield* Semaphore.make(1);

const getLiveSources: Effect.Effect<ReadonlyArray<ProviderSnapshotSource>> = Ref.get(
const getLiveSources: Effect.Effect<ReadonlyArray<LiveProviderSnapshotSource>> = Ref.get(
liveSubsRef,
).pipe(Effect.map((map) => Array.from(map.values(), buildSnapshotSource)));

Expand Down Expand Up @@ -522,14 +525,27 @@ export const ProviderRegistryLive = Layer.effect(
);

const refreshOneSource = Effect.fn("refreshOneSource")(function* (
providerSource: ProviderSnapshotSource,
providerSource: LiveProviderSnapshotSource,
) {
return yield* providerSource.refresh.pipe(
Effect.flatMap((nextProvider) =>
correlateSnapshotWithSource(providerSource, nextProvider).pipe(
Effect.flatMap(syncProvider),
),
),
const nextProvider = yield* providerSource.refresh;
const correlated = yield* correlateSnapshotWithSource(providerSource, nextProvider);
const instance = yield* instanceRegistry.getInstance(providerSource.instanceId);
if (instance !== providerSource.instance) return yield* Ref.get(providersRef);
// Explicit refreshes invalidate workspace catalogs. Background snapshots
// still retain them, and active clients request their cwd again on demand.
const cached = (yield* Ref.get(providersRef)).find(
(provider) => provider.instanceId === providerSource.instanceId,
)?.workspaceSnapshots;
let hadPending = false;
if (instance) {
hadPending = yield* Ref.modify(workspaceRefreshesRef, (refreshes) => {
const next = new Map(refreshes);
next.delete(instance);
return [refreshes.has(instance), next] as const;
});
}
return yield* syncProvider(
cached !== undefined || hadPending ? { ...correlated, workspaceSnapshots: [] } : correlated,
Comment thread
yashranaway marked this conversation as resolved.
Resolved
);
});

Expand Down Expand Up @@ -818,11 +834,12 @@ export const ProviderRegistryLive = Layer.effect(
}
const instance = yield* instanceRegistry.getInstance(input.instanceId);
if (!instance?.snapshotForCwd) return providers;
const token = Symbol();
const claimed = yield* Ref.modify(workspaceRefreshesRef, (refreshes) => {
const current = refreshes.get(instance);
if (current?.has(input.cwd)) return [false, refreshes] as const;
const next = new Map(refreshes);
next.set(instance, new Set(current).add(input.cwd));
next.set(instance, new Map(current).set(input.cwd, token));
return [true, next] as const;
});
if (!claimed) return yield* Ref.get(providersRef);
Expand All @@ -831,31 +848,40 @@ export const ProviderRegistryLive = Layer.effect(
scopedSnapshot.status === "error"
? Ref.get(providersRef)
: instanceRegistry.getInstance(input.instanceId).pipe(
Effect.flatMap((currentInstance) => {
if (currentInstance !== instance) return Ref.get(providersRef);
return Ref.modify(providersRef, (currentProviders) => {
const nextProviders = currentProviders.map((candidate) =>
candidate.instanceId === input.instanceId &&
!candidate.workspaceSnapshots?.some((s) => s.cwd === input.cwd)
? upsertProviderWorkspaceSnapshot(candidate, input.cwd, scopedSnapshot)
: candidate,
Effect.flatMap((currentInstance) =>
Effect.gen(function* () {
const refreshes = yield* Ref.get(workspaceRefreshesRef);
if (
currentInstance !== instance ||
refreshes.get(instance)?.get(input.cwd) !== token
) {
return yield* Ref.get(providersRef);
}
return yield* Ref.modify(providersRef, (currentProviders) => {
const nextProviders = currentProviders.map((candidate) =>
candidate.instanceId === input.instanceId &&
!candidate.workspaceSnapshots?.some((s) => s.cwd === input.cwd)
? upsertProviderWorkspaceSnapshot(candidate, input.cwd, scopedSnapshot)
: candidate,
);
return [[currentProviders, nextProviders] as const, nextProviders];
}).pipe(
Effect.tap(([previousProviders, nextProviders]) =>
haveProvidersChanged(previousProviders, nextProviders)
? PubSub.publish(changesPubSub, nextProviders)
: Effect.void,
),
Effect.map(([, nextProviders]) => nextProviders),
);
return [[currentProviders, nextProviders] as const, nextProviders];
}).pipe(
Effect.tap(([previousProviders, nextProviders]) =>
haveProvidersChanged(previousProviders, nextProviders)
? PubSub.publish(changesPubSub, nextProviders)
: Effect.void,
),
Effect.map(([, nextProviders]) => nextProviders),
);
}),
}),
),
),
),
Effect.ensuring(
Ref.update(workspaceRefreshesRef, (refreshes) => {
if (refreshes.get(instance)?.get(input.cwd) !== token) return refreshes;
const next = new Map(refreshes);
const current = new Set(next.get(instance));
const current = new Map(next.get(instance));
current.delete(input.cwd);
if (current.size) next.set(instance, current);
else next.delete(instance);
Expand Down
Loading