From 46c4b2fb6f6a27bf7449089d43ef184f76d647f4 Mon Sep 17 00:00:00 2001 From: Katya Date: Fri, 2 Oct 2026 22:09:08 +0200 Subject: [PATCH 1/5] fix: rescan workspace skills after session start hooks run --- .../Adapters/ClaudeAdapterV2.test.ts | 64 +++++++++ .../Adapters/ClaudeAdapterV2.ts | 25 +++- .../src/provider/Drivers/ClaudeDriver.ts | 12 +- .../provider/Layers/ProviderRegistry.test.ts | 135 ++++++++++++++++++ .../src/provider/Layers/ProviderRegistry.ts | 19 ++- apps/server/src/provider/ProviderDriver.ts | 7 + 6 files changed, 257 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 467d58702df0..7cc58d5b4c62 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -2006,6 +2006,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { readonly close?: (sdkMessages: Queue.Queue) => Effect.Effect; readonly interrupt?: Effect.Effect; readonly environment?: NodeJS.ProcessEnv; + readonly onSessionInit?: (cwd: string) => Effect.Effect; }) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -2036,6 +2037,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { fileSystem, path: yield* Path.Path, idAllocator, + ...(options?.onSessionInit === undefined ? {} : { onSessionInit: options.onSessionInit }), continuationRequests: { offer: (request) => Effect.sync(() => { @@ -2389,6 +2391,68 @@ describe("ClaudeAdapterV2 background wake turns", () => { }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), ); + it.effect("reports the session cwd once per query, when the CLI's first init arrives", () => + Effect.gen(function* () { + const initCwds: Array = []; + const harness = yield* makeWakeHarnessWithOptions({ + onSessionInit: (cwd) => + Effect.sync(() => { + initCwds.push(cwd); + }), + }); + const now = yield* DateTime.now; + const cwd = "/synthetic/worktree"; + const runtimePolicy = ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd, + }); + const init = (uuid: string) => + claudeSdkFrame({ + type: "system", + subtype: "init", + // The CLI may report a resolved path, never the one the client keys on. + cwd: "/private/synthetic/worktree", + session_id: WAKE_NATIVE_SESSION, + uuid, + }); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-init-1"), + text: "First.", + attachments: [], + runtimePolicy, + }), + ); + yield* harness.offerAndWait(init("00000000-0000-4000-8000-000000000701")); + assert.deepEqual(initCwds, [cwd]); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000702", result: "One." }), + ); + yield* Queue.take(harness.terminalReceipts); + + // The CLI repeats init on every turn of one process; hooks ran once. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-init-2"), + providerTurnOrdinal: 2, + text: "Second.", + attachments: [], + runtimePolicy, + }), + ); + yield* harness.offerAndWait(init("00000000-0000-4000-8000-000000000703")); + assert.deepEqual(initCwds, [cwd]); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + it.effect("keeps usage warnings and provisioned overage silent", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 48deeb4ae38f..66885f425bde 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -2660,6 +2660,7 @@ interface ClaudeLiveQueryContext { readonly query: ClaudeAgentSdkQuerySession; readonly queryPolicyKey: string; readonly selectionKey: string; + readonly cwd: string | null; readonly closed: Deferred.Deferred; // Whether this CLI process echoes a prompt's uuid on the first frame of // the turn answering it ("early") or only on its result. Learned from the @@ -2667,6 +2668,7 @@ interface ClaudeLiveQueryContext { // uuid before any echo, so it echoes, but a resume's own turns can still // run ahead of that prompt. promptEchoMode: "unknown" | "acknowledged" | "early" | "result_only"; + sessionInitReported: boolean; } interface ActiveClaudeToolCall { @@ -2868,6 +2870,12 @@ export interface ClaudeAdapterV2Options { readonly queryRunner: ClaudeAgentSdkQueryRunnerShape; readonly scopedLimitNames?: Ref.Ref; readonly onUsageLimits?: ServerProviderShape["applyUsageLimits"]; + /** + * Called with the query's cwd when its CLI process first reports `init`, + * which comes after SessionStart hooks have run. Once per query: the CLI + * repeats `init` on every turn. + */ + readonly onSessionInit?: (cwd: string) => Effect.Effect; /** Sink for wake-turn continuation requests; defaults to dropping them. */ readonly continuationRequests?: { readonly offer: ( @@ -5313,6 +5321,16 @@ export function makeClaudeAdapterV2( } const message = input.message; + if ( + message.type === "system" && + message.subtype === "init" && + !liveQuery.sessionInitReported + ) { + liveQuery.sessionInitReported = true; + if (liveQuery.cwd !== null && adapterOptions.onSessionInit) { + yield* adapterOptions.onSessionInit(liveQuery.cwd); + } + } // Before any routing: a Monitor started during an idle wake turn // reports its task before the drain replays the tool call. yield* trackClaudeMonitorCalls(message); @@ -6873,8 +6891,10 @@ export function makeClaudeAdapterV2( query: querySession, queryPolicyKey, selectionKey: compiledSelection.queryIdentity, + cwd: turnInput.runtimePolicy.cwd, closed, promptEchoMode: "unknown", + sessionInitReported: false, }; yield* Ref.set(queryContext, context); yield* querySession.messages.pipe( @@ -7577,7 +7597,10 @@ export type ClaudeAdapterV2DriverEnv = export const createClaudeAdapterV2 = Effect.fn("ClaudeAdapterV2Driver.create")( function* ( input: ProviderAdapterDriverCreateInput, - hooks: Pick = {}, + hooks: Pick< + ClaudeAdapterV2Options, + "scopedLimitNames" | "onUsageLimits" | "onSessionInit" + > = {}, ) { const { instanceId, environment, enabled, config } = input; const fileSystem = yield* FileSystem.FileSystem; diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index 1673d3e6d639..82366506b91c 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -19,7 +19,9 @@ import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import { HttpClient } from "effect/unstable/http"; import { ChildProcessSpawner } from "effect/unstable/process"; @@ -165,6 +167,9 @@ export const ClaudeDriver: ProviderDriver = { }); const scopedLimitNames = yield* makeClaudeScopedLimitNames; + // Sessions report here once their startup hooks have run; the registry + // listens and rescans the cwd. With no listener a report is dropped. + const workspaceRescans = yield* PubSub.unbounded(); const orchestrationAdapter = yield* createClaudeAdapterV2( { instanceId, @@ -174,7 +179,11 @@ export const ClaudeDriver: ProviderDriver = { enabled, config, }, - { scopedLimitNames, onUsageLimits: (update) => snapshot.applyUsageLimits(update) }, + { + scopedLimitNames, + onUsageLimits: (update) => snapshot.applyUsageLimits(update), + onSessionInit: (cwd) => PubSub.publish(workspaceRescans, cwd).pipe(Effect.asVoid), + }, ).pipe( Effect.mapError( (cause) => @@ -347,6 +356,7 @@ export const ClaudeDriver: ProviderDriver = { enabled, snapshot, invalidateCaches: Cache.invalidateAll(capabilitiesProbeCache), + workspaceRescans: Stream.fromPubSub(workspaceRescans), snapshotForCwd: (cwd: string) => !effectiveConfig.enabled ? snapshot.getSnapshot diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 332a5e931d29..60e7ecca9263 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -9,8 +9,10 @@ import * as Effect from "effect/Effect"; import * as Path from "effect/Path"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; @@ -39,6 +41,7 @@ import { applyServerSettingsPatch } from "@t3tools/shared/serverSettings"; import { checkCodexProviderStatus, type CodexAppServerProviderSnapshot } from "./CodexProvider.ts"; import { checkClaudeProviderStatus } from "./ClaudeProvider.ts"; +import { discoverClaudeSkills } from "../Drivers/ClaudeSkills.ts"; import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts"; import * as AntigravityInstallation from "../AntigravityInstallation.ts"; import * as ModelManifest from "../ModelManifest.ts"; @@ -1761,6 +1764,138 @@ it.layer(Layer.mergeAll(TestNodeServices, ServerSettingsModule.layerTest(), Test }), ); + it.effect( + "rescans a held workspace when its instance reports a started session", + () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-provider-registry-session-rescan-", + }); + const homePath = path.join(root, "claude-home"); + const workspace = path.join(root, "worktree"); + yield* fileSystem.makeDirectory(workspace, { recursive: true }); + const driver = ProviderDriverKind.make("claudeAgent"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + const machineProvider = { + instanceId, + driver, + status: "ready", + enabled: true, + installed: true, + auth: { status: "authenticated" }, + checkedAt: "2026-06-10T00:00:00.000Z", + version: "1.0.0", + models: [], + slashCommands: [], + skills: [], + } as const satisfies ServerProvider; + const cacheInvalidations = yield* Ref.make(0); + const rescans = yield* Queue.unbounded(); + const instance: ProviderInstance = { + instanceId, + driverKind: driver, + continuationIdentity: { + driverKind: driver, + continuationKey: "claudeAgent:instance:claudeAgent", + }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + getSnapshot: Effect.succeed(machineProvider), + refresh: Effect.succeed(machineProvider), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + snapshotForCwd: (cwd) => + discoverClaudeSkills({ homePath }, cwd).pipe( + Effect.map((skills) => ({ ...machineProvider, skills })), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + ), + invalidateCaches: Ref.update(cacheInvalidations, (count) => count + 1), + workspaceRescans: Stream.fromQueue(rescans), + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + }; + const registryChanges = yield* PubSub.unbounded(); + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtimeServices = yield* Layer.build( + ProviderRegistryLive.pipe( + Layer.provideMerge( + Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { + getInstance: (requestedId) => + Effect.succeed(requestedId === instanceId ? instance : undefined), + listInstances: Effect.succeed([instance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.fromPubSub(registryChanges), + subscribeChanges: PubSub.subscribe(registryChanges), + }), + ), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-session-rescan-", + }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ).pipe(Scope.provide(scope)); + + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + const workspaceSkills = () => + registry.getProviders.pipe( + Effect.map((providers) => + providers[0]?.workspaceSnapshots + ?.find((snapshot) => snapshot.cwd === workspace) + ?.skills.map((skill) => skill.name), + ), + ); + + // The composer scans as soon as the worktree exists, before a + // SessionStart hook has linked the project skills in. + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: workspace }); + assert.deepStrictEqual(yield* workspaceSkills(), []); + const skillDir = path.join(workspace, ".claude", "skills", "deploy"); + yield* fileSystem.makeDirectory(skillDir, { recursive: true }); + yield* fileSystem.writeFileString( + path.join(skillDir, "SKILL.md"), + [ + "---", + "name: deploy", + "description: Deploy the app.", + "---", + "", + "# Deploy", + ].join("\n"), + ); + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: workspace }); + assert.deepStrictEqual(yield* workspaceSkills(), []); + + const update = yield* registry.streamChanges.pipe(Stream.runHead, Effect.forkChild); + yield* Effect.yieldNow; + yield* Queue.offer(rescans, workspace); + assert.strictEqual((yield* Fiber.join(update))._tag, "Some"); + assert.deepStrictEqual(yield* workspaceSkills(), ["deploy"]); + // A session start rescans the one cwd; it is not the explicit + // refresh that re-reads the machine snapshot. + assert.strictEqual(yield* Ref.get(cacheInvalidations), 0); + }).pipe(Effect.provide(runtimeServices)); + }).pipe(Effect.provide(NodeServices.layer)), + ), + { timeout: 10_000 }, + ); + it.effect("refreshes OpenCode catalogs and preserves other providers", () => Effect.gen(function* () { const codexDriver = ProviderDriverKind.make("codex"); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.ts b/apps/server/src/provider/Layers/ProviderRegistry.ts index de476ca0299a..82cc1cfc1a43 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.ts @@ -749,6 +749,15 @@ export const ProviderRegistryLive = Layer.effect( yield* Stream.runForEach(source.streamChanges, (provider) => correlateSnapshotWithSource(source, provider).pipe(Effect.flatMap(syncProvider)), ).pipe(Effect.forkScoped); + if (instance.workspaceRescans) { + yield* Stream.runForEach(instance.workspaceRescans, (cwd) => + refreshWorkspaceSnapshot({ + instanceId: instance.instanceId, + cwd, + rescan: true, + }).pipe(Effect.ignoreCause({ log: true })), + ).pipe(Effect.forkScoped); + } } yield* Effect.yieldNow; @@ -900,7 +909,11 @@ export const ProviderRegistryLive = Layer.effect( readonly instanceId: ProviderInstanceId; readonly cwd: string; readonly fresh?: boolean; + // Scan again although the cwd has a snapshot, without the machine-wide + // work of a fresh scan. For a change the instance itself reported. + readonly rescan?: boolean; }) { + const forced = input.fresh === true || input.rescan === true; // Fresh scans drop other instances' snapshots for this cwd first, so a // composer on one of them scans again on next use, even when this // instance is gone or cannot be scanned. @@ -918,7 +931,7 @@ export const ProviderRegistryLive = Layer.effect( const workspaceSnapshotOf = (candidate: ServerProvider | undefined) => candidate?.workspaceSnapshots?.find((s) => s.cwd === input.cwd); const scannedFrom = workspaceSnapshotOf(provider); - if (!provider || !provider.enabled || (!input.fresh && scannedFrom)) { + if (!provider || !provider.enabled || (!forced && scannedFrom)) { return providers; } const instance = yield* instanceRegistry.getInstance(input.instanceId); @@ -930,8 +943,8 @@ export const ProviderRegistryLive = Layer.effect( next.set(instance, new Set(current).add(input.cwd)); return [true, next] as const; }); - // A fresh scan never joins a running one, which may predate the change. - if (!claimed && !input.fresh) return yield* Ref.get(providersRef); + // A forced scan never joins a running one, which may predate the change. + if (!claimed && !forced) return yield* Ref.get(providersRef); // Fresh scans also re-read the machine snapshot: Claude's plugin // commands come from it, not from the cwd scan. const refreshMachineSnapshot = input.fresh diff --git a/apps/server/src/provider/ProviderDriver.ts b/apps/server/src/provider/ProviderDriver.ts index 4f92df965c8a..b10613b1905c 100644 --- a/apps/server/src/provider/ProviderDriver.ts +++ b/apps/server/src/provider/ProviderDriver.ts @@ -35,6 +35,7 @@ import type { import type * as Effect from "effect/Effect"; import type * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; +import type * as Stream from "effect/Stream"; import type { TextGeneration } from "../textGeneration/TextGeneration.ts"; import type { ProviderAdapterV2Shape } from "../orchestration-v2/ProviderAdapter.ts"; @@ -77,6 +78,12 @@ export interface ProviderInstance { readonly enabled: boolean; readonly snapshot: ServerProviderShape; readonly snapshotForCwd?: (cwd: string) => Effect.Effect; + /** + * Workspaces whose skills may have changed since the registry scanned them, + * such as a session whose startup hooks have just finished. The registry + * rescans each emitted cwd through `snapshotForCwd`. + */ + readonly workspaceRescans?: Stream.Stream; readonly refreshModels?: () => Effect.Effect; /** Invalidate T3-owned discovery caches before an explicit provider refresh. */ readonly invalidateCaches?: Effect.Effect; From 3c69a972c39488f34c7a0d3ca1c02f506f5e13f2 Mon Sep 17 00:00:00 2001 From: Katya Date: Fri, 2 Oct 2026 22:34:14 +0200 Subject: [PATCH 2/5] fix: keep, order and end session-start rescans --- .../src/provider/Drivers/ClaudeDriver.ts | 13 +- .../ProviderInstanceRegistryLive.test.ts | 155 ++++++++++++++++++ .../provider/Layers/ProviderRegistry.test.ts | 114 +++++++++++++ .../src/provider/Layers/ProviderRegistry.ts | 5 +- 4 files changed, 281 insertions(+), 6 deletions(-) diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index 82366506b91c..cf71f01f9324 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -14,12 +14,13 @@ */ import { ClaudeSettings, ProviderDriverKind } from "@t3tools/contracts"; import * as Cache from "effect/Cache"; +import type * as Cause from "effect/Cause"; import * as Duration from "effect/Duration"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; -import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { HttpClient } from "effect/unstable/http"; @@ -168,8 +169,10 @@ export const ClaudeDriver: ProviderDriver = { const scopedLimitNames = yield* makeClaudeScopedLimitNames; // Sessions report here once their startup hooks have run; the registry - // listens and rescans the cwd. With no listener a report is dropped. - const workspaceRescans = yield* PubSub.unbounded(); + // listens and rescans the cwd. A queue keeps a report made before the + // registry subscribes, and ending it with the instance ends the listener. + const workspaceRescans = yield* Queue.unbounded(); + yield* Effect.addFinalizer(() => Queue.end(workspaceRescans)); const orchestrationAdapter = yield* createClaudeAdapterV2( { instanceId, @@ -182,7 +185,7 @@ export const ClaudeDriver: ProviderDriver = { { scopedLimitNames, onUsageLimits: (update) => snapshot.applyUsageLimits(update), - onSessionInit: (cwd) => PubSub.publish(workspaceRescans, cwd).pipe(Effect.asVoid), + onSessionInit: (cwd) => Queue.offer(workspaceRescans, cwd).pipe(Effect.asVoid), }, ).pipe( Effect.mapError( @@ -356,7 +359,7 @@ export const ClaudeDriver: ProviderDriver = { enabled, snapshot, invalidateCaches: Cache.invalidateAll(capabilitiesProbeCache), - workspaceRescans: Stream.fromPubSub(workspaceRescans), + workspaceRescans: Stream.fromQueue(workspaceRescans), snapshotForCwd: (cwd: string) => !effectiveConfig.enabled ? snapshot.getSnapshot diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index 4d383a7dd16b..414589a7a095 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -27,23 +27,34 @@ import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; */ import { describe, expect, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; import { EnvironmentId, + MessageId, + NodeId, type ClaudeSettings, type CodexSettings, type CursorSettings, type GrokSettings, type OpenCodeSettings, + ProjectId, ProviderDriverKind, type ProviderInstanceConfigMap, ProviderInstanceId, + ProviderSessionId, + RunAttemptId, + RunId, + ThreadId, } from "@t3tools/contracts"; import { HostProcessPlatform, isHostWindows } from "@t3tools/shared/hostProcess"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import { HttpClient, HttpClientResponse } from "effect/unstable/http"; @@ -52,6 +63,8 @@ import * as AntigravityInstallation from "../AntigravityInstallation.ts"; import * as ServerConfig from "../../config.ts"; import { expandHomePath } from "../../pathExpansion.ts"; import * as ServerSettings from "../../serverSettings.ts"; +import { ClaudeAgentSdkQueryRunner } from "../../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; +import { ProviderAdapterV2RuntimePolicy } from "../../orchestration-v2/ProviderAdapter.ts"; import { ClaudeDriver, type ClaudeDriverEnv } from "../Drivers/ClaudeDriver.ts"; import { CodexDriver, type CodexDriverEnv } from "../Drivers/CodexDriver.ts"; import { CursorDriver, type CursorDriverEnv } from "../Drivers/CursorDriver.ts"; @@ -822,4 +835,146 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { ); }).pipe(Effect.provide(testLayer)), ); + + it.live( + "keeps a Claude session start reported before anyone listens and ends with the instance", + () => + Effect.gen(function* () { + const sdkMessages = yield* Queue.unbounded(); + const processed = new WeakMap>(); + const queryRunner = ClaudeAgentSdkQueryRunner.of({ + allocateSessionId: Effect.succeed("native-thread-workspace-rescan"), + open: () => + Effect.succeed({ + // The next pull happens after the adapter handled the previous frame. + messages: Stream.fromQueue(sdkMessages).pipe( + Stream.flatMap((message) => + Stream.make(message).pipe( + Stream.concat( + Stream.fromEffect( + Effect.suspend(() => { + const handled = processed.get(message); + return handled === undefined + ? Effect.void + : Deferred.succeed(handled, undefined); + }), + ).pipe(Stream.drain), + ), + ), + ), + ), + offer: () => Effect.void, + setModel: () => Effect.void, + interrupt: Effect.void, + close: Effect.void, + }), + forkSession: () => Effect.die("unused forkSession"), + subagentLaunchToolUseId: () => Effect.succeed(null), + assertComplete: Effect.void, + }); + const claudeId = ProviderInstanceId.make("claude_rescan"); + const claudeConfig = { + [claudeId]: { + driver: ProviderDriverKind.make("claudeAgent"), + enabled: false, + config: makeClaudeConfig({}), + }, + } satisfies ProviderInstanceConfigMap; + const { registry, mutator } = yield* makeProviderInstanceRegistry({ + drivers: [ClaudeDriver], + configMap: claudeConfig, + }).pipe(Effect.provideService(ClaudeAgentSdkQueryRunner, queryRunner)); + const instance = yield* registry.getInstance(claudeId); + expect(instance?.workspaceRescans).toBeDefined(); + + const cwd = "/synthetic/worktree"; + const threadId = ThreadId.make("thread-workspace-rescan"); + const modelSelection = { instanceId: claudeId, model: "claude-sonnet-4-6" } as const; + const runtimePolicy = ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd, + }); + const runtime = yield* instance!.orchestrationAdapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-workspace-rescan"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + const attemptId = RunAttemptId.make("attempt-workspace-rescan"); + yield* runtime.startTurn({ + appThread: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project-workspace-rescan"), + title: "Workspace rescan", + providerInstanceId: claudeId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: providerThread.id, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + threadId, + runId: RunId.make("run-workspace-rescan"), + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId, + rootNodeId: NodeId.make("node-workspace-rescan"), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: MessageId.make("message-workspace-rescan"), + text: "Start.", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + const init = { + type: "system", + subtype: "init", + cwd, + session_id: "native-thread-workspace-rescan", + uuid: "00000000-0000-4000-8000-000000000801", + } as unknown as SDKMessage; + const handled = yield* Deferred.make(); + processed.set(init, handled); + yield* Queue.offer(sdkMessages, init); + yield* Deferred.await(handled); + + // The registry has not subscribed to the stream yet, so the report + // must have been kept for it. + const first = yield* instance!.workspaceRescans!.pipe( + Stream.take(1), + Stream.runCollect, + Effect.timeout("5 seconds"), + ); + expect(first).toEqual([cwd]); + + // Removing the instance closes its scope, which ends the stream. + const listener = yield* instance!.workspaceRescans!.pipe(Stream.runDrain, Effect.forkChild); + yield* Effect.yieldNow; + yield* mutator.reconcile({}); + yield* Fiber.join(listener).pipe(Effect.timeout("5 seconds")); + }).pipe(Effect.provide(testLayer)), + ); }); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 60e7ecca9263..03b543201022 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -1896,6 +1896,120 @@ it.layer(Layer.mergeAll(TestNodeServices, ServerSettingsModule.layerTest(), Test { timeout: 10_000 }, ); + it.effect("lets a rescan overwrite the snapshot of an older scan that committed first", () => + Effect.gen(function* () { + const driver = ProviderDriverKind.make("claudeAgent"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + const machineProvider = { + instanceId, + driver, + status: "ready", + enabled: true, + installed: true, + auth: { status: "authenticated" }, + checkedAt: "2026-06-10T00:00:00.000Z", + version: "1.0.0", + models: [], + slashCommands: [], + skills: [], + } as const satisfies ServerProvider; + const skillsBefore = [ + { name: "before", path: "/workspace/before/SKILL.md", enabled: true }, + ]; + const skillsAfter = [{ name: "after", path: "/workspace/after/SKILL.md", enabled: true }]; + // The first scan to start reads the folder before the hooks ran, the + // second after. Each waits for its own release, so the older one can + // commit first. + const scans = yield* Ref.make(0); + const gates = [yield* Deferred.make(), yield* Deferred.make()] as const; + const started = yield* Queue.unbounded(); + const rescans = yield* Queue.unbounded(); + const instance: ProviderInstance = { + instanceId, + driverKind: driver, + continuationIdentity: { + driverKind: driver, + continuationKey: "claudeAgent:instance:claudeAgent", + }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + getSnapshot: Effect.succeed(machineProvider), + refresh: Effect.succeed(machineProvider), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + snapshotForCwd: () => + Effect.gen(function* () { + const index = (yield* Ref.getAndUpdate(scans, (count) => count + 1)) as 0 | 1; + yield* Queue.offer(started, index); + yield* Deferred.await(gates[index]); + return { ...machineProvider, skills: index === 0 ? skillsBefore : skillsAfter }; + }), + workspaceRescans: Stream.fromQueue(rescans), + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + }; + const registryChanges = yield* PubSub.unbounded(); + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtimeServices = yield* Layer.build( + ProviderRegistryLive.pipe( + Layer.provideMerge( + Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { + getInstance: (requestedId) => + Effect.succeed(requestedId === instanceId ? instance : undefined), + listInstances: Effect.succeed([instance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.fromPubSub(registryChanges), + subscribeChanges: PubSub.subscribe(registryChanges), + }), + ), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-rescan-race-", + }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ).pipe(Scope.provide(scope)); + + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + const composerScan = yield* registry + .refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" }) + .pipe(Effect.forkChild); + assert.strictEqual(yield* Queue.take(started), 0); + yield* Queue.offer(rescans, "/workspace"); + assert.strictEqual(yield* Queue.take(started), 1); + const rescanned = yield* registry.streamChanges.pipe( + Stream.filter( + (providers) => + providers[0]?.workspaceSnapshots?.some((s) => s.skills === skillsAfter) === true, + ), + Stream.runHead, + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* Deferred.succeed(gates[0], undefined); + yield* Fiber.join(composerScan); + yield* Deferred.succeed(gates[1], undefined); + assert.strictEqual((yield* Fiber.join(rescanned))._tag, "Some"); + assert.deepStrictEqual( + (yield* registry.getProviders)[0]?.workspaceSnapshots?.map((s) => s.skills), + [skillsAfter], + ); + }).pipe(Effect.provide(runtimeServices)); + }), + ); + it.effect("refreshes OpenCode catalogs and preserves other providers", () => Effect.gen(function* () { const codexDriver = ProviderDriverKind.make("codex"); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.ts b/apps/server/src/provider/Layers/ProviderRegistry.ts index 82cc1cfc1a43..23358f2caa3d 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.ts @@ -962,10 +962,13 @@ export const ProviderRegistryLive = Layer.effect( if (currentInstance !== instance) return Ref.get(providersRef); // Write only if the cwd's snapshot did not change during the // scan. A session event or another scan that landed first is newer. + // A rescan starts after the change it reports, so a scan that + // started earlier and landed first is the stale one. return updateProviders((currentProviders) => currentProviders.map((candidate) => candidate.instanceId === input.instanceId && - Equal.equals(workspaceSnapshotOf(candidate), scannedFrom) + (input.rescan === true || + Equal.equals(workspaceSnapshotOf(candidate), scannedFrom)) ? upsertProviderWorkspaceSnapshot(candidate, input.cwd, scopedSnapshot) : candidate, ), From b1f33922f89531cc8ee4f0d5f8a8d190a29a726f Mon Sep 17 00:00:00 2001 From: Katya Date: Fri, 2 Oct 2026 23:31:17 +0200 Subject: [PATCH 3/5] fix: rescan only workspaces a client asked about --- .../provider/Layers/ProviderRegistry.test.ts | 96 +++++++++++++++++++ .../src/provider/Layers/ProviderRegistry.ts | 11 +++ 2 files changed, 107 insertions(+) diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 03b543201022..05dbf0db1bcd 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -2010,6 +2010,102 @@ it.layer(Layer.mergeAll(TestNodeServices, ServerSettingsModule.layerTest(), Test }), ); + it.effect("does not scan a rescan for a cwd no client asked about", () => + Effect.gen(function* () { + const driver = ProviderDriverKind.make("claudeAgent"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + const machineProvider = { + instanceId, + driver, + status: "ready", + enabled: true, + installed: true, + auth: { status: "authenticated" }, + checkedAt: "2026-06-10T00:00:00.000Z", + version: "1.0.0", + models: [], + slashCommands: [], + skills: [], + } as const satisfies ServerProvider; + const scannedCwds = yield* Ref.make>([]); + const rescans = yield* Queue.unbounded(); + const instance: ProviderInstance = { + instanceId, + driverKind: driver, + continuationIdentity: { + driverKind: driver, + continuationKey: "claudeAgent:instance:claudeAgent", + }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + getSnapshot: Effect.succeed(machineProvider), + refresh: Effect.succeed(machineProvider), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + snapshotForCwd: (cwd) => + Ref.updateAndGet(scannedCwds, (cwds) => [...cwds, cwd]).pipe( + // Each scan differs, so every commit is a visible change. + Effect.map((cwds) => ({ + ...machineProvider, + skills: [{ name: `scan-${cwds.length}`, path: `${cwd}/SKILL.md`, enabled: true }], + })), + ), + workspaceRescans: Stream.fromQueue(rescans), + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + }; + const registryChanges = yield* PubSub.unbounded(); + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtimeServices = yield* Layer.build( + ProviderRegistryLive.pipe( + Layer.provideMerge( + Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { + getInstance: (requestedId) => + Effect.succeed(requestedId === instanceId ? instance : undefined), + listInstances: Effect.succeed([instance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.fromPubSub(registryChanges), + subscribeChanges: PubSub.subscribe(registryChanges), + }), + ), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-unrequested-rescan-", + }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ).pipe(Scope.provide(scope)); + + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/requested" }); + // Sessions of threads nobody has open report their cwd too; those + // must not add entries that evict a composer's snapshot. + const updated = yield* registry.streamChanges.pipe(Stream.runHead, Effect.forkChild); + yield* Effect.yieldNow; + yield* Queue.offer(rescans, "/unrequested"); + yield* Queue.offer(rescans, "/requested"); + assert.strictEqual((yield* Fiber.join(updated))._tag, "Some"); + assert.deepStrictEqual(yield* Ref.get(scannedCwds), ["/requested", "/requested"]); + assert.deepStrictEqual( + (yield* registry.getProviders)[0]?.workspaceSnapshots?.map((s) => s.cwd), + ["/requested"], + ); + }).pipe(Effect.provide(runtimeServices)); + }), + ); + it.effect("refreshes OpenCode catalogs and preserves other providers", () => Effect.gen(function* () { const codexDriver = ProviderDriverKind.make("codex"); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.ts b/apps/server/src/provider/Layers/ProviderRegistry.ts index 23358f2caa3d..7ede09de1cdf 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.ts @@ -936,6 +936,17 @@ export const ProviderRegistryLive = Layer.effect( } const instance = yield* instanceRegistry.getInstance(input.instanceId); if (!instance?.snapshotForCwd) return providers; + // Snapshots go to every client and are capped, so a rescan only + // refreshes a cwd a client asked about: one it holds a snapshot for, or + // whose first scan is still running and may have read the folder too + // early. Sessions of threads nobody has open add nothing. + if ( + input.rescan === true && + !scannedFrom && + !(yield* Ref.get(workspaceRefreshesRef)).get(instance)?.has(input.cwd) + ) { + return providers; + } const claimed = yield* Ref.modify(workspaceRefreshesRef, (refreshes) => { const current = refreshes.get(instance); if (current?.has(input.cwd)) return [false, refreshes] as const; From 8a4800fb74d444f094a67a7d30a1f05cd7b84d76 Mon Sep 17 00:00:00 2001 From: Katya Date: Sat, 3 Oct 2026 00:16:09 +0200 Subject: [PATCH 4/5] fix: order workspace scans and start the rescan consumer late --- .../provider/Layers/ProviderRegistry.test.ts | 170 ++++++++++++ .../src/provider/Layers/ProviderRegistry.ts | 259 ++++++++++-------- 2 files changed, 317 insertions(+), 112 deletions(-) diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 05dbf0db1bcd..5f7d08b5955b 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -2106,6 +2106,176 @@ it.layer(Layer.mergeAll(TestNodeServices, ServerSettingsModule.layerTest(), Test }), ); + // A Claude-shaped instance whose workspace scans follow a script: each + // call takes the next entry, waits for its gate, then returns its skills. + const makeScriptedWorkspaceRegistry = Effect.fn("makeScriptedWorkspaceRegistry")(function* ( + script: ReadonlyArray<{ + readonly skills: string; + readonly gate?: Deferred.Deferred; + }>, + rescans: Queue.Queue, + ) { + const driver = ProviderDriverKind.make("claudeAgent"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + const machineProvider = { + instanceId, + driver, + status: "ready", + enabled: true, + installed: true, + auth: { status: "authenticated" }, + checkedAt: "2026-06-10T00:00:00.000Z", + version: "1.0.0", + models: [], + slashCommands: [], + skills: [], + } as const satisfies ServerProvider; + const next = yield* Ref.make(0); + const started = yield* Queue.unbounded(); + const instance: ProviderInstance = { + instanceId, + driverKind: driver, + continuationIdentity: { + driverKind: driver, + continuationKey: "claudeAgent:instance:claudeAgent", + }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + getSnapshot: Effect.succeed(machineProvider), + refresh: Effect.succeed(machineProvider), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + snapshotForCwd: (cwd) => + Effect.gen(function* () { + const index = yield* Ref.getAndUpdate(next, (count) => count + 1); + const step = script[index]; + if (step === undefined) return yield* Effect.die(`unscripted scan ${index}`); + yield* Queue.offer(started, index); + if (step.gate) yield* Deferred.await(step.gate); + return { + ...machineProvider, + skills: [ + { name: step.skills, path: `${cwd}/${step.skills}/SKILL.md`, enabled: true }, + ], + }; + }), + workspaceRescans: Stream.fromQueue(rescans), + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + }; + const registryChanges = yield* PubSub.unbounded(); + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtimeServices = yield* Layer.build( + ProviderRegistryLive.pipe( + Layer.provideMerge( + Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { + getInstance: (requestedId) => + Effect.succeed(requestedId === instanceId ? instance : undefined), + listInstances: Effect.succeed([instance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.fromPubSub(registryChanges), + subscribeChanges: PubSub.subscribe(registryChanges), + }), + ), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-provider-registry-scripted-" }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ).pipe(Scope.provide(scope)); + const heldSkills = (cwd: string) => + Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + const providers = yield* registry.getProviders; + return providers[0]?.workspaceSnapshots + ?.find((snapshot) => snapshot.cwd === cwd) + ?.skills.map((item) => item.name); + }); + return { instanceId, runtimeServices, started, heldSkills }; + }); + + it.effect( + "keeps a fresh scan that started after a rescan when the rescan finishes last", + () => + Effect.gen(function* () { + const rescanGate = yield* Deferred.make(); + const freshGate = yield* Deferred.make(); + const rescans = yield* Queue.unbounded(); + const { instanceId, runtimeServices, started, heldSkills } = + yield* makeScriptedWorkspaceRegistry( + [ + { skills: "other-initial" }, + { skills: "initial" }, + { skills: "rescan", gate: rescanGate }, + { skills: "fresh", gate: freshGate }, + { skills: "other-rescan" }, + ], + rescans, + ); + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/other" }); + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" }); + yield* Queue.offer(rescans, "/workspace"); + assert.strictEqual(yield* Queue.takeN(started, 3).pipe(Effect.map((n) => n[2])), 2); + // The fresh scan starts after the rescan and commits first. + const fresh = yield* registry + .refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace", fresh: true }) + .pipe(Effect.forkChild); + assert.strictEqual(yield* Queue.take(started), 3); + yield* Deferred.succeed(freshGate, undefined); + yield* Fiber.join(fresh); + assert.deepStrictEqual(yield* heldSkills("/workspace"), ["fresh"]); + yield* Deferred.succeed(rescanGate, undefined); + // Reports are handled in order, so once /other was rescanned the + // rescan of /workspace has finished. + const settled = yield* registry.streamChanges.pipe(Stream.runHead, Effect.forkChild); + yield* Effect.yieldNow; + yield* Queue.offer(rescans, "/other"); + yield* Fiber.join(settled); + assert.deepStrictEqual(yield* heldSkills("/other"), ["other-rescan"]); + assert.deepStrictEqual(yield* heldSkills("/workspace"), ["fresh"]); + }).pipe(Effect.provide(runtimeServices)); + }), + ); + + it.effect( + "serves a session start reported before the registry finished building", + () => + Effect.gen(function* () { + const rescans = yield* Queue.unbounded(); + yield* Queue.offer(rescans, "/workspace"); + const { instanceId, runtimeServices, heldSkills } = + yield* makeScriptedWorkspaceRegistry( + [{ skills: "initial" }, { skills: "rescan" }], + rescans, + ); + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" }); + const rescanned = yield* registry.streamChanges.pipe( + Stream.runHead, + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* Queue.offer(rescans, "/workspace"); + assert.strictEqual((yield* Fiber.join(rescanned))._tag, "Some"); + assert.deepStrictEqual(yield* heldSkills("/workspace"), ["rescan"]); + }).pipe(Effect.provide(runtimeServices)); + }), + { timeout: 10_000 }, + ); + it.effect("refreshes OpenCode catalogs and preserves other providers", () => Effect.gen(function* () { const codexDriver = ProviderDriverKind.make("codex"); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.ts b/apps/server/src/provider/Layers/ProviderRegistry.ts index 7ede09de1cdf..ffedffa7570f 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.ts @@ -422,6 +422,13 @@ export const ProviderRegistryLive = Layer.effect( const workspaceRefreshesRef = yield* Ref.make< ReadonlyMap> >(new Map()); + // Orders workspace scans by when they started, so a rescan cannot replace + // what a scan that started after it already committed. Per instance, the + // newest committed start for each cwd, capped like the snapshots are. + const scanStartsRef = yield* Ref.make(0); + const committedScanStartsRef = yield* Ref.make< + ReadonlyMap> + >(new Map()); const maintenanceActionStatesRef = yield* Ref.make< ReadonlyMap >(new Map()); @@ -663,6 +670,146 @@ export const ProviderRegistryLive = Layer.effect( return yield* instance.snapshot.resolveMaintenance(options); }); + // Defined before `syncLiveSources`: the rescan consumers it forks may + // already have a report waiting and must find these initialized. + const updateProviders = ( + update: (providers: ReadonlyArray) => ReadonlyArray, + ) => + Ref.modify(providersRef, (currentProviders) => { + const nextProviders = update(currentProviders); + return [[currentProviders, nextProviders] as const, nextProviders]; + }).pipe( + Effect.tap(([previousProviders, nextProviders]) => + haveProvidersChanged(previousProviders, nextProviders) + ? PubSub.publish(changesPubSub, nextProviders) + : Effect.void, + ), + Effect.map(([, nextProviders]) => nextProviders), + ); + + const refreshWorkspaceSnapshot = Effect.fn("refreshWorkspaceSnapshot")(function* (input: { + readonly instanceId: ProviderInstanceId; + readonly cwd: string; + readonly fresh?: boolean; + // Scan again although the cwd has a snapshot, without the machine-wide + // work of a fresh scan. For a change the instance itself reported. + readonly rescan?: boolean; + }) { + const forced = input.fresh === true || input.rescan === true; + // Fresh scans drop other instances' snapshots for this cwd first, so a + // composer on one of them scans again on next use, even when this + // instance is gone or cannot be scanned. + if (input.fresh) { + yield* updateProviders((providers) => + providers.map((candidate) => + candidate.instanceId === input.instanceId + ? candidate + : dropProviderWorkspaceSnapshot(candidate, input.cwd), + ), + ); + } + const providers = yield* Ref.get(providersRef); + const provider = providers.find((candidate) => candidate.instanceId === input.instanceId); + const workspaceSnapshotOf = (candidate: ServerProvider | undefined) => + candidate?.workspaceSnapshots?.find((s) => s.cwd === input.cwd); + const scannedFrom = workspaceSnapshotOf(provider); + if (!provider || !provider.enabled || (!forced && scannedFrom)) { + return providers; + } + const instance = yield* instanceRegistry.getInstance(input.instanceId); + if (!instance?.snapshotForCwd) return providers; + // Snapshots go to every client and are capped, so a rescan only + // refreshes a cwd a client asked about: one it holds a snapshot for, or + // whose first scan is still running and may have read the folder too + // early. Sessions of threads nobody has open add nothing. + if ( + input.rescan === true && + !scannedFrom && + !(yield* Ref.get(workspaceRefreshesRef)).get(instance)?.has(input.cwd) + ) { + return providers; + } + 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)); + return [true, next] as const; + }); + // A forced scan never joins a running one, which may predate the change. + if (!claimed && !forced) return yield* Ref.get(providersRef); + const startedAt = yield* Ref.updateAndGet(scanStartsRef, (count) => count + 1); + // Fresh scans also re-read the machine snapshot: Claude's plugin + // commands come from it, not from the cwd scan. + const refreshMachineSnapshot = input.fresh + ? (instance.invalidateCaches ?? Effect.void).pipe( + Effect.andThen(refreshInstance(input.instanceId)), + ) + : Effect.void; + return yield* refreshMachineSnapshot.pipe( + Effect.andThen(instance.snapshotForCwd(input.cwd)), + Effect.flatMap((scopedSnapshot) => + scopedSnapshot.status === "error" + ? Ref.get(providersRef) + : instanceRegistry.getInstance(input.instanceId).pipe( + Effect.flatMap((currentInstance) => { + if (currentInstance !== instance) return Ref.get(providersRef); + // Write only if the cwd's snapshot did not change during the + // scan. A session event or another scan that landed first is newer. + // A rescan starts after the change it reports, so a scan that + // started earlier and landed first is the stale one. It still + // loses to a scan that started after it and landed first. + return Ref.modify(committedScanStartsRef, (committed) => { + const forInstance = committed.get(input.instanceId); + const latest = forInstance?.get(input.cwd) ?? 0; + const cwds = new Map(forInstance); + cwds.delete(input.cwd); + cwds.set(input.cwd, Math.max(latest, startedAt)); + for (const oldest of cwds.keys()) { + if (cwds.size <= MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER) break; + cwds.delete(oldest); + } + return [ + input.rescan === true && latest > startedAt, + new Map(committed).set(input.instanceId, cwds), + ] as const; + }).pipe( + Effect.flatMap((superseded) => + superseded + ? Ref.get(providersRef) + : updateProviders((currentProviders) => + currentProviders.map((candidate) => + candidate.instanceId === input.instanceId && + (input.rescan === true || + Equal.equals(workspaceSnapshotOf(candidate), scannedFrom)) + ? upsertProviderWorkspaceSnapshot( + candidate, + input.cwd, + scopedSnapshot, + ) + : candidate, + ), + ), + ), + ); + }), + ), + ), + Effect.ensuring( + claimed + ? Ref.update(workspaceRefreshesRef, (refreshes) => { + const next = new Map(refreshes); + const current = new Set(next.get(instance)); + current.delete(input.cwd); + if (current.size) next.set(instance, current); + else next.delete(instance); + return next; + }) + : Effect.void, + ), + ); + }); + /** * Diff the aggregator's live-source set against the current * `ProviderInstanceRegistry` and: @@ -890,118 +1037,6 @@ export const ProviderRegistryLive = Layer.effect( return yield* Ref.get(providersRef); }); - const updateProviders = ( - update: (providers: ReadonlyArray) => ReadonlyArray, - ) => - Ref.modify(providersRef, (currentProviders) => { - const nextProviders = update(currentProviders); - return [[currentProviders, nextProviders] as const, nextProviders]; - }).pipe( - Effect.tap(([previousProviders, nextProviders]) => - haveProvidersChanged(previousProviders, nextProviders) - ? PubSub.publish(changesPubSub, nextProviders) - : Effect.void, - ), - Effect.map(([, nextProviders]) => nextProviders), - ); - - const refreshWorkspaceSnapshot = Effect.fn("refreshWorkspaceSnapshot")(function* (input: { - readonly instanceId: ProviderInstanceId; - readonly cwd: string; - readonly fresh?: boolean; - // Scan again although the cwd has a snapshot, without the machine-wide - // work of a fresh scan. For a change the instance itself reported. - readonly rescan?: boolean; - }) { - const forced = input.fresh === true || input.rescan === true; - // Fresh scans drop other instances' snapshots for this cwd first, so a - // composer on one of them scans again on next use, even when this - // instance is gone or cannot be scanned. - if (input.fresh) { - yield* updateProviders((providers) => - providers.map((candidate) => - candidate.instanceId === input.instanceId - ? candidate - : dropProviderWorkspaceSnapshot(candidate, input.cwd), - ), - ); - } - const providers = yield* Ref.get(providersRef); - const provider = providers.find((candidate) => candidate.instanceId === input.instanceId); - const workspaceSnapshotOf = (candidate: ServerProvider | undefined) => - candidate?.workspaceSnapshots?.find((s) => s.cwd === input.cwd); - const scannedFrom = workspaceSnapshotOf(provider); - if (!provider || !provider.enabled || (!forced && scannedFrom)) { - return providers; - } - const instance = yield* instanceRegistry.getInstance(input.instanceId); - if (!instance?.snapshotForCwd) return providers; - // Snapshots go to every client and are capped, so a rescan only - // refreshes a cwd a client asked about: one it holds a snapshot for, or - // whose first scan is still running and may have read the folder too - // early. Sessions of threads nobody has open add nothing. - if ( - input.rescan === true && - !scannedFrom && - !(yield* Ref.get(workspaceRefreshesRef)).get(instance)?.has(input.cwd) - ) { - return providers; - } - 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)); - return [true, next] as const; - }); - // A forced scan never joins a running one, which may predate the change. - if (!claimed && !forced) return yield* Ref.get(providersRef); - // Fresh scans also re-read the machine snapshot: Claude's plugin - // commands come from it, not from the cwd scan. - const refreshMachineSnapshot = input.fresh - ? (instance.invalidateCaches ?? Effect.void).pipe( - Effect.andThen(refreshInstance(input.instanceId)), - ) - : Effect.void; - return yield* refreshMachineSnapshot.pipe( - Effect.andThen(instance.snapshotForCwd(input.cwd)), - Effect.flatMap((scopedSnapshot) => - scopedSnapshot.status === "error" - ? Ref.get(providersRef) - : instanceRegistry.getInstance(input.instanceId).pipe( - Effect.flatMap((currentInstance) => { - if (currentInstance !== instance) return Ref.get(providersRef); - // Write only if the cwd's snapshot did not change during the - // scan. A session event or another scan that landed first is newer. - // A rescan starts after the change it reports, so a scan that - // started earlier and landed first is the stale one. - return updateProviders((currentProviders) => - currentProviders.map((candidate) => - candidate.instanceId === input.instanceId && - (input.rescan === true || - Equal.equals(workspaceSnapshotOf(candidate), scannedFrom)) - ? upsertProviderWorkspaceSnapshot(candidate, input.cwd, scopedSnapshot) - : candidate, - ), - ); - }), - ), - ), - Effect.ensuring( - claimed - ? Ref.update(workspaceRefreshesRef, (refreshes) => { - const next = new Map(refreshes); - const current = new Set(next.get(instance)); - current.delete(input.cwd); - if (current.size) next.set(instance, current); - else next.delete(instance); - return next; - }) - : Effect.void, - ), - ); - }); - return { getProviders: Ref.get(providersRef), refresh: (provider?: ProviderDriverKind) => From 93cf8fa3758a700ec6664aa0e5492d7d3a338119 Mon Sep 17 00:00:00 2001 From: Katya Date: Sun, 11 Oct 2026 13:21:54 +0200 Subject: [PATCH 5/5] fix: count a workspace scan as landed only when it writes --- .../src/provider/ProviderRegistry.test.ts | 62 ++++++++++++ apps/server/src/provider/ProviderRegistry.ts | 95 ++++++++++++------- 2 files changed, 122 insertions(+), 35 deletions(-) diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 5b6f914c2ce0..1f60a1e85233 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -2370,6 +2370,68 @@ it.layer( }), ); + it.effect("keeps a rescan alive when a later scan was rejected by the snapshot guard", () => + Effect.gen(function* () { + const firstGate = yield* Deferred.make(); + const rescanGate = yield* Deferred.make(); + const freshGate = yield* Deferred.make(); + const rescans = yield* Queue.unbounded(); + const { instanceId, runtimeServices, started, heldSkills } = + yield* makeScriptedWorkspaceRegistry( + [ + { skills: "other-initial" }, + { skills: "first", gate: firstGate }, + { skills: "rescan", gate: rescanGate }, + { skills: "fresh", gate: freshGate }, + { skills: "other-rescan" }, + ], + rescans, + ); + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + yield* registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/other" }); + assert.strictEqual(yield* Queue.take(started), 0); + // The first scan of /workspace is still running when the rescan and + // a fresh scan start, so the fresh scan began with no snapshot. + const first = yield* registry + .refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" }) + .pipe(Effect.forkChild); + assert.strictEqual(yield* Queue.take(started), 1); + yield* Queue.offer(rescans, "/workspace"); + assert.strictEqual(yield* Queue.take(started), 2); + const fresh = yield* registry + .refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace", fresh: true }) + .pipe(Effect.forkChild); + assert.strictEqual(yield* Queue.take(started), 3); + // The first scan lands, so the fresh scan finds the snapshot changed + // under it and writes nothing. + yield* Deferred.succeed(firstGate, undefined); + yield* Fiber.join(first); + yield* Deferred.succeed(freshGate, undefined); + yield* Fiber.join(fresh); + assert.deepStrictEqual(yield* heldSkills("/workspace"), ["first"]); + // A scan that wrote nothing has not committed, so the rescan that + // started before it is not superseded. + yield* Deferred.succeed(rescanGate, undefined); + const settled = yield* registry.streamChanges.pipe( + Stream.filter( + (providers) => + providers[0]?.workspaceSnapshots?.some( + (snapshot) => + snapshot.cwd === "/other" && snapshot.skills[0]?.name === "other-rescan", + ) === true, + ), + Stream.runHead, + Effect.forkChild, + ); + yield* Effect.yieldNow; + yield* Queue.offer(rescans, "/other"); + yield* Fiber.join(settled); + assert.deepStrictEqual(yield* heldSkills("/workspace"), ["rescan"]); + }).pipe(Effect.provide(runtimeServices)); + }), + ); + it.effect( "serves a session start reported before the registry finished building", () => diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index 93ef26402376..628a3f23f6e0 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -813,6 +813,14 @@ export const layer = Layer.effect( // Defined before `syncLiveSources`: the rescan consumers it forks may // already have a report waiting and must find these initialized. + const publishIfChanged = ( + previousProviders: ReadonlyArray, + nextProviders: ReadonlyArray, + ) => + haveProvidersChanged(previousProviders, nextProviders) + ? PubSub.publish(changesPubSub, nextProviders) + : Effect.void; + const updateProviders = ( update: (providers: ReadonlyArray) => ReadonlyArray, ) => @@ -821,9 +829,7 @@ export const layer = Layer.effect( return [[currentProviders, nextProviders] as const, nextProviders]; }).pipe( Effect.tap(([previousProviders, nextProviders]) => - haveProvidersChanged(previousProviders, nextProviders) - ? PubSub.publish(changesPubSub, nextProviders) - : Effect.void, + publishIfChanged(previousProviders, nextProviders), ), Effect.map(([, nextProviders]) => nextProviders), ); @@ -909,39 +915,58 @@ export const layer = Layer.effect( // scan. A session event or another scan that landed first is newer. // A rescan starts after the change it reports, so a scan that // started earlier and landed first is the stale one. It still - // loses to a scan that started after it and landed first. - return Ref.modify(committedScanStartsRef, (committed) => { - const forInstance = committed.get(input.instanceId); - const latest = forInstance?.get(input.cwd) ?? 0; - const cwds = new Map(forInstance); - cwds.delete(input.cwd); - cwds.set(input.cwd, Math.max(latest, startedAt)); - for (const oldest of cwds.keys()) { - if (cwds.size <= MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER) break; - cwds.delete(oldest); + // loses to a scan that started after it and landed first. Only a + // scan that wrote counts as landed: one the guard rejected must + // not make a rescan that started before it look superseded. + return Effect.gen(function* () { + // Read, write and commit run with no yield between them, and + // the publish comes last: a yield after the write but before + // the commit would let an older rescan pass the check and + // overwrite this snapshot. + const latest = + (yield* Ref.get(committedScanStartsRef)) + .get(input.instanceId) + ?.get(input.cwd) ?? 0; + if (input.rescan === true && latest > startedAt) { + return yield* Ref.get(providersRef); + } + const [previousProviders, nextProviders, written] = yield* Ref.modify( + providersRef, + (currentProviders) => { + let written = false; + const nextProviders = currentProviders.map((candidate) => { + if ( + candidate.instanceId !== input.instanceId || + (input.rescan !== true && + !Equal.equals(workspaceSnapshotOf(candidate), scannedFrom)) + ) { + return candidate; + } + written = true; + return upsertProviderWorkspaceSnapshot(candidate, input.cwd, { + ...scopedSnapshot, + checkedAt: scannedAt, + }); + }); + return [[currentProviders, nextProviders, written] as const, nextProviders]; + }, + ); + if (written) { + yield* Ref.update(committedScanStartsRef, (committed) => { + const forInstance = committed.get(input.instanceId); + const cwds = new Map(forInstance); + cwds.delete(input.cwd); + cwds.set(input.cwd, Math.max(forInstance?.get(input.cwd) ?? 0, startedAt)); + for (const oldest of cwds.keys()) { + if (cwds.size <= MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER) break; + cwds.delete(oldest); + } + return new Map(committed).set(input.instanceId, cwds); + }); } - return [ - input.rescan === true && latest > startedAt, - new Map(committed).set(input.instanceId, cwds), - ] as const; - }).pipe( - Effect.flatMap((superseded) => - superseded - ? Ref.get(providersRef) - : updateProviders((currentProviders) => - currentProviders.map((candidate) => - candidate.instanceId === input.instanceId && - (input.rescan === true || - Equal.equals(workspaceSnapshotOf(candidate), scannedFrom)) - ? upsertProviderWorkspaceSnapshot(candidate, input.cwd, { - ...scopedSnapshot, - checkedAt: scannedAt, - }) - : candidate, - ), - ), - ), - ); + yield* publishIfChanged(previousProviders, nextProviders); + return nextProviders; + }); }), ), ),