diff --git a/apps/mobile/src/features/threads/use-composer-command-menu.ts b/apps/mobile/src/features/threads/use-composer-command-menu.ts index 8498d907f690..992a5838a66e 100644 --- a/apps/mobile/src/features/threads/use-composer-command-menu.ts +++ b/apps/mobile/src/features/threads/use-composer-command-menu.ts @@ -59,6 +59,8 @@ import type { ComposerCommandItem } from "./ComposerCommandPopover"; import { matchesSlashSkillQuery } from "./composerSlashSkillSearch"; const WORKSPACE_SNAPSHOT_RETRY_COOLDOWN_MS = 10_000; +// A cwd that never scans, such as a deleted worktree, would otherwise retry forever. +const WORKSPACE_SNAPSHOT_MAX_ATTEMPTS = 3; function composerSelectionAtEnd(draftMessage: string): ComposerEditorSelection { return { start: draftMessage.length, end: draftMessage.length }; @@ -252,6 +254,7 @@ export function useComposerCommandMenu({ ); // Record client request time to enforce TTL even when the client clock runs ahead of the server. const workspaceRefreshKeyRef = useRef<{ key: string; requestedAt: number } | null>(null); + const failedRefreshesRef = useRef<{ key: string; count: number } | null>(null); const [workspaceRefreshRetry, setWorkspaceRefreshRetry] = useState<{ key: string; notBefore: number; @@ -307,6 +310,11 @@ export function useComposerCommandMenu({ workspaceRefreshKeyRef.current = request; const retryLater = () => { if (workspaceRefreshKeyRef.current !== request) return; + const failed = failedRefreshesRef.current; + const count = failed?.key === key ? failed.count + 1 : 1; + failedRefreshesRef.current = { key, count }; + // Keeping the request marker lets the TTL hold off the next scan. + if (count >= WORKSPACE_SNAPSHOT_MAX_ATTEMPTS) return; workspaceRefreshKeyRef.current = null; setWorkspaceRefreshRetry({ key, @@ -326,6 +334,7 @@ export function useComposerCommandMenu({ projectCwd, ); if (!refreshed) retryLater(); + else failedRefreshesRef.current = null; }, retryLater); }, [ draftMessage, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 2bd56e30e9dc..4678aff95b73 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -239,6 +239,7 @@ export const make = Effect.gen(function* () { threadSnooze: true, environmentThemes: true, usageLimitSources: true, + providerWorkspaceSnapshotsOnDemand: true, usagePriceOverrides: true, usageModelAliases: true, threadPinning: true, diff --git a/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts b/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts index 80740b0a21d4..60a4371cb133 100644 --- a/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts +++ b/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts @@ -1,8 +1,15 @@ // @effect-diagnostics nodeBuiltinImport:off - cleanup uses Node's retrying rm, which the FileSystem service does not expose. +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { ClaudeSettings, type ServerProvider } from "@t3tools/contracts"; import { vi } from "vite-plus/test"; import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; -import { buildClaudeCapabilitiesProbeQueryOptions } from "./ClaudeProvider.ts"; +import { + buildClaudeCapabilitiesProbeQueryOptions, + probeClaudeWorkspaceSnapshot, +} from "./ClaudeProvider.ts"; vi.mock("@anthropic-ai/claude-agent-sdk", { spy: true }); @@ -20,3 +27,15 @@ it("drops strict MCP mode, and only that, next to an administrator's managed-mcp assert.deepEqual(options.mcpServers, {}); assert.equal(options.env?.ENABLE_CLAUDEAI_MCP_SERVERS, "false"); }); + +it.effect("reports a deleted worktree's scan as finished, not pending", () => + Effect.gen(function* () { + const snapshot = yield* probeClaudeWorkspaceSnapshot( + yield* Schema.decodeEffect(ClaudeSettings)({}), + { skills: [], slashCommands: [] } as unknown as ServerProvider, + "/nonexistent/t3-deleted-worktree", + { HOME: "/nonexistent/t3-home" }, + ); + assert.strictEqual(snapshot.slashCommandsPending, false); + }).pipe(Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/provider/ClaudeProvider.ts b/apps/server/src/provider/ClaudeProvider.ts index 4c6a86dbc381..5bc82a41d4ed 100644 --- a/apps/server/src/provider/ClaudeProvider.ts +++ b/apps/server/src/provider/ClaudeProvider.ts @@ -431,10 +431,14 @@ export const probeClaudeWorkspaceSnapshot = Effect.fn("probeClaudeWorkspaceSnaps ): Effect.fn.Return { if (!claudeSettings.enabled) return machineSnapshot; const skills = yield* discoverClaudeSkills(claudeSettings, cwd, environment); - const probeResult = yield* probeClaudeCapabilities(claudeSettings, environment, cwd, false).pipe( - Effect.result, - ); - const capabilities = Result.isSuccess(probeResult) ? probeResult.success : undefined; + 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)); + const probeResult = cwdExists + ? yield* probeClaudeCapabilities(claudeSettings, environment, cwd, false).pipe(Effect.result) + : undefined; + const capabilities = + probeResult && Result.isSuccess(probeResult) ? probeResult.success : undefined; return { ...machineSnapshot, skills, @@ -442,7 +446,7 @@ export const probeClaudeWorkspaceSnapshot = Effect.fn("probeClaudeWorkspaceSnaps COMPACT_SLASH_COMMAND, ...(capabilities?.slashCommands ?? []), ]), - slashCommandsPending: !capabilities, + slashCommandsPending: cwdExists && !capabilities, }; }); diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index f819fd64ca30..b30b9c061040 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -1246,6 +1246,98 @@ it.layer( ); }); + it.effect( + "a second request for a cwd being scanned gets its result, even if the first leaves", + () => + Effect.gen(function* () { + const driver = ProviderDriverKind.make("codex"); + const instanceId = ProviderInstanceId.make("codex"); + 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: [], + skills: [], + } as const satisfies ServerProvider; + const scanStarted = yield* Deferred.make(); + const releaseScan = yield* Deferred.make(); + const scans = yield* Ref.make(0); + const instance: ProviderInstance = { + instanceId, + driverKind: driver, + continuationIdentity: { driverKind: driver, continuationKey: "codex:instance:codex" }, + 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: () => + Ref.update(scans, (count) => count + 1).pipe( + Effect.andThen(Deferred.succeed(scanStarted, undefined)), + Effect.andThen(Deferred.await(releaseScan)), + Effect.as(provider), + ), + 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.provideMerge(layerInstanceRegistry), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-shared-workspace-scan-", + }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ).pipe(Scope.provide(scope)); + + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + const refresh = registry.refreshWorkspaceSnapshot({ instanceId, cwd: "/workspace" }); + const first = yield* Effect.forkChild(refresh); + yield* Deferred.await(scanStarted); + const second = yield* Effect.forkChild(refresh, { startImmediately: true }); + yield* Fiber.interrupt(first); + yield* Deferred.succeed(releaseScan, undefined); + assert.deepStrictEqual( + (yield* Fiber.join(second))[0]?.workspaceSnapshots?.map((snapshot) => snapshot.cwd), + ["/workspace"], + ); + assert.strictEqual(yield* Ref.get(scans), 1); + }).pipe(Effect.provide(runtimeServices)); + }), + ); + it.effect("publishes workspace skills while machine health is in error", () => Effect.gen(function* () { const driver = ProviderDriverKind.make("codex"); diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index 0096ca23c037..41b4e11f4888 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -12,6 +12,7 @@ import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -405,7 +406,7 @@ export const layer = Layer.effect( cachedProviders.map((provider) => classifyCompatibility(provider, initialManifest)), ); const workspaceRefreshesRef = yield* Ref.make< - ReadonlyMap> + ReadonlyMap>> >(new Map()); const maintenanceActionStatesRef = yield* Ref.make< ReadonlyMap @@ -922,21 +923,13 @@ export const layer = Layer.effect( const scannedAt = DateTime.formatIso(now); 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 (!input.fresh && current?.has(input.cwd)) return [false, refreshes] as const; - const next = new Map(refreshes); - next.set(instance, new Map(current).set(input.cwd, token)); - return [true, next] as const; - }); - if (!claimed) return yield* Ref.get(providersRef); + const token = yield* Deferred.make(); const refreshMachineSnapshot = input.fresh ? (instance.invalidateCaches ?? Effect.void).pipe( Effect.andThen(refreshInstance(input.instanceId)), ) : Effect.void; - return yield* refreshMachineSnapshot.pipe( + const runScan = refreshMachineSnapshot.pipe( Effect.andThen(instance.snapshotForCwd(input.cwd)), // Scans fail on error; returned inventory remains valid even when provider health reports an error. Effect.flatMap((scopedSnapshot) => @@ -974,9 +967,33 @@ export const layer = Layer.effect( if (nextCwds.size) next.set(instance, nextCwds); else next.delete(instance); return next; - }), + }).pipe(Effect.andThen(Deferred.succeed(token, undefined))), ), ); + // The claim and the fork happen together, so the token is always completed. The scan is + // detached so a disconnecting owner cannot cancel the result its waiters depend on. + const claim = yield* Effect.uninterruptible( + Effect.gen(function* () { + const inFlight = yield* Ref.modify(workspaceRefreshesRef, (refreshes) => { + const current = refreshes.get(instance); + const running = current?.get(input.cwd); + if (!input.fresh && running) return [running, refreshes] as const; + const next = new Map(refreshes); + next.set(instance, new Map(current).set(input.cwd, token)); + return [undefined, next] as const; + }); + if (inFlight) return { inFlight } as const; + return { + // Starting now attaches the scan's `ensuring` before anything can close the scope. + scan: yield* Effect.forkIn(runScan, serviceScope, { startImmediately: true }), + } as const; + }), + ); + // Wait for the running scan: on-demand clients get its result only in this reply. + if ("inFlight" in claim) { + return yield* Deferred.await(claim.inFlight).pipe(Effect.andThen(Ref.get(providersRef))); + } + return yield* Fiber.join(claim.scan); }); return { diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index a2c3910d4da5..49425668cd77 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -5,13 +5,17 @@ import { EnvironmentId, ProjectId, ThreadId, + ProviderDriverKind, + ProviderInstanceId, WS_METHODS, + WsServerRefreshProvidersRpc, WsSubscribeServerConfigRpc, type OrchestrationV2StoredEvent, type OrchestrationV2ThreadShell, type OrchestrationProjectShell, type ServerConfig, type ServerConfigStreamEvent, + type ServerProvider, } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; @@ -20,6 +24,7 @@ import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/http"; import { RpcGroup, RpcMessage, RpcServer } from "effect/rpc"; import * as SqlClient from "effect/sql/SqlClient"; @@ -632,6 +637,162 @@ it.effect( ), ); +it.effect("on-demand clients get provider config without workspace snapshots", () => + Effect.gen(function* () { + const instanceId = ProviderInstanceId.make("claudeAgent"); + const snapshotFor = (cwd: string, checkedAt: string) => ({ + cwd, + checkedAt, + slashCommands: [{ name: `${cwd.slice(1)}-command` }], + skills: [], + }); + const provider: ServerProvider = { + instanceId, + driver: ProviderDriverKind.make("claudeAgent"), + enabled: true, + installed: true, + version: "1.0.0", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-10-25T00:00:00.000Z", + models: [], + slashCommands: [{ name: "machine-command" }], + skills: [], + workspaceSnapshots: [ + snapshotFor("/a", "2026-10-25T00:00:00.000Z"), + snapshotFor("/b", "2026-10-25T00:00:00.000Z"), + ], + }; + const providerChanges = yield* Queue.unbounded>(); + const pulled = yield* Queue.unbounded(); + const responses = yield* Queue.unbounded(); + const configRpcServices = Layer.mergeAll( + layerUnusedConfigRpcServices, + Layer.mock(ExternalLauncher.ExternalLauncher)({ + resolveAvailableEditors: () => Effect.succeed([]), + resolveFileManagerRevealKind: () => Effect.succeed(undefined), + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([provider]), + streamChanges: Stream.fromQueue(providerChanges).pipe( + Stream.tap(() => Queue.offer(pulled, undefined)), + ), + refreshWorkspaceSnapshot: () => Effect.succeed([provider]), + }), + Layer.mock(UsageLimitSources.UsageLimitSources)({ + current: Effect.succeed([]), + streamChanges: Stream.make([]), + }), + Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({ + resolveTargets: () => Effect.succeed([]), + }), + Layer.mock(DirectEndpoints.DirectEndpoints)({ resolve: () => Effect.succeed([]) }), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + scratchRoot: Effect.succeedNone, + namedProjectsRoot: process.cwd(), + }), + Layer.mock(ServerEnvironment.ServerEnvironment)({ + getDescriptor: Effect.succeed({ + environmentId: EnvironmentId.make("config-test"), + label: "Config test", + platform: { os: "linux", arch: "x64" }, + serverVersion: "0.0.0", + capabilities: { repositoryIdentity: false }, + }), + }), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({ + getDescriptor: () => + Effect.succeed({ + policy: "loopback-browser", + bootstrapMethods: [], + sessionMethods: ["browser-session-cookie"], + sessionCookieName: "config-test-session", + }), + }), + ); + const server = yield* RpcServer.makeNoSerialization( + RpcGroup.make(WsSubscribeServerConfigRpc, WsServerRefreshProvidersRpc), + { + ...WS_RPC_SERVER_OPTIONS, + onFromServer: (response) => + response._tag === "Chunk" + ? Effect.forEach(response.values, (value) => Queue.offer(responses, value), { + discard: true, + }) + : response._tag === "Exit" && response.exit._tag === "Success" + ? Queue.offer(responses, response.exit.value) + : Effect.void, + }, + ).pipe( + Effect.provide( + layerWsRpc( + { + sessionId: AuthSessionId.make("config-test"), + subject: "config-test", + method: "browser-session-cookie", + scopes: [], + }, + {}, + {}, + yield* PreviewAutomationBroker.PreviewAutomationBroker, + yield* ServerBrowser.ServerBrowser, + ).pipe(Layer.provide(configRpcServices)), + ), + ); + const configRequest = RpcMessage.RequestId("config"); + yield* server.write(0, { + _tag: "Request", + id: configRequest, + tag: WS_METHODS.subscribeServerConfig, + payload: { workspaceSnapshotsOnDemand: true }, + headers: Headers.empty, + }); + const snapshot = (yield* Queue.take(responses)) as ServerConfigStreamEvent; + if (snapshot.type !== "snapshot") return yield* Effect.die("Missing initial snapshot"); + assert.isUndefined(snapshot.config.providers[0]?.workspaceSnapshots); + yield* server.write(0, { _tag: "Ack", requestId: configRequest }); + + // A rescan that only touches a snapshot sends nothing; the next real change does. + yield* Queue.offer(providerChanges, [ + { ...provider, workspaceSnapshots: [snapshotFor("/a", "2026-10-25T00:05:00.000Z")] }, + ]); + yield* Queue.take(pulled); + yield* TestClock.adjust("1 second"); + yield* Queue.offer(providerChanges, [{ ...provider, status: "warning" }]); + yield* Queue.take(pulled); + yield* TestClock.adjust("1 second"); + const update = (yield* Queue.take(responses)) as ServerConfigStreamEvent; + if (update.type !== "providerStatuses") return yield* Effect.die("Expected provider update"); + assert.equal(update.payload.providers[0]?.status, "warning"); + assert.isUndefined(update.payload.providers[0]?.workspaceSnapshots); + + yield* server.write(0, { + _tag: "Request", + id: RpcMessage.RequestId("refresh"), + tag: WS_METHODS.serverRefreshProviders, + payload: { instanceId, cwd: "/b", workspaceSnapshotsOnDemand: true }, + headers: Headers.empty, + }); + const refreshed = (yield* Queue.take(responses)) as { + readonly providers: ReadonlyArray; + }; + assert.deepEqual( + refreshed.providers.map((entry) => [entry.workspaceSnapshots, entry.slashCommands]), + [[[snapshotFor("/b", "2026-10-25T00:00:00.000Z")], []]], + ); + yield* server.write(0, { _tag: "Interrupt", requestId: configRequest, interruptors: [] }); + }).pipe( + Effect.provide( + Layer.mergeAll( + Layer.fresh(layerConfigSources), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ), +); + // Compaction keeps the oldest and newest event per message, so a cursor older // than the deleted rows still replays each message where the client would // first have seen it (reducers append on first sight) and reaches the same diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index a03176a455ea..e5ef646a615a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -82,6 +82,7 @@ import { type ServerConfig as ClientServerConfig, type ServerConfigStreamEvent, type ServerLifecycleStreamEvent, + type ServerProvider, type FilesystemBrowseFailure, FilesystemBrowseError, AssetWorkspaceContextNotFoundError, @@ -363,6 +364,10 @@ export const withLateEditorConfig = ( ); }; +function withoutWorkspaceSnapshots(providers: ReadonlyArray): ServerProvider[] { + return providers.map(({ workspaceSnapshots: _workspaceSnapshots, ...provider }) => provider); +} + function unexpectedCompatibilityError(error: never): never { throw new Error(`Unhandled compatibility error: ${String(error)}`); } @@ -1721,12 +1726,16 @@ export const layerWsRpc = ( const loadServerConfig = Effect.fn("ws.loadServerConfig")(function* (options: { readonly usageLimitsCommand: boolean; + readonly workspaceSnapshotsOnDemand: boolean; }) { const keybindingsConfig = yield* keybindings.loadConfigState; const currentProviders = yield* providerRegistry.getProviders; - const providers = options.usageLimitsCommand + const commandProviders = options.usageLimitsCommand ? withUsageLimitsCommands(currentProviders, yield* usageLimitSources.current) : currentProviders; + const providers = options.workspaceSnapshotsOnDemand + ? withoutWorkspaceSnapshots(commandProviders) + : commandProviders; const settings = ServerSettings.redactServerSettingsForClient( yield* serverSettings.getSettings, ); @@ -2207,7 +2216,8 @@ export const layerWsRpc = ( Effect.andThen(scheduledTasks.getWebhookDelivery(input)), ), [WS_METHODS.serverProbe]: (_input) => Effect.succeed({}), - [WS_METHODS.serverGetConfig]: (_input) => loadServerConfig({ usageLimitsCommand: false }), + [WS_METHODS.serverGetConfig]: (_input) => + loadServerConfig({ usageLimitsCommand: false, workspaceSnapshotsOnDemand: false }), [WS_METHODS.serverSearchAcpRegistry]: (input) => acpRegistryCatalog .search(input) @@ -2376,7 +2386,29 @@ export const layerWsRpc = ( providers = yield* providerRegistry.refreshInstance(instance.instanceId); } } - return { providers }; + if (input.workspaceSnapshotsOnDemand !== true) return { providers }; + const commandProviders = + input.usageLimitsCommand === true + ? withUsageLimitsCommands(providers, yield* usageLimitSources.current) + : providers; + const { instanceId, cwd } = input; + if (instanceId === undefined || cwd === undefined) { + return { providers: withoutWorkspaceSnapshots(commandProviders) }; + } + return { + providers: commandProviders + .filter((provider) => provider.instanceId === instanceId) + .map((provider) => ({ + ...provider, + // Callers read only the cwd snapshot; the config stream carries the rest. + models: [], + slashCommands: [], + skills: [], + workspaceSnapshots: (provider.workspaceSnapshots ?? []).filter( + (snapshot) => snapshot.cwd === cwd, + ), + })), + }; }), [WS_METHODS.mcpAppsCallTool]: (input) => mcpAppRequests.callTool(input), [WS_METHODS.mcpAppsToolInfo]: (input) => mcpAppRequests.toolInfo(input), @@ -3093,10 +3125,14 @@ export const layerWsRpc = ( Stream.unwrap( Effect.gen(function* () { const usageLimitsCommand = input.usageLimitsCommand === true; + const workspaceSnapshotsOnDemand = input.workspaceSnapshotsOnDemand === true; // Buffer changes through discovery and the initial snapshot's RPC ACK wait. const keybindingsChanges = yield* keybindings.subscribeChanges; const settingsChanges = yield* serverSettings.subscribeChanges; - const config = yield* loadServerConfig({ usageLimitsCommand }); + const config = yield* loadServerConfig({ + usageLimitsCommand, + workspaceSnapshotsOnDemand, + }); const keybindingsUpdates = keybindingsChanges.pipe( Stream.map((event) => ({ version: 1 as const, @@ -3118,8 +3154,15 @@ export const layerWsRpc = ( usageLimitsCommand ? sameUsageLimitCommandCoverage : () => true, ), ), - (providers, sources) => - usageLimitsCommand ? withUsageLimitsCommands(providers, sources) : providers, + (providers, sources) => { + const commandProviders = usageLimitsCommand + ? withUsageLimitsCommands(providers, sources) + : providers; + // Stripped before the de-dupe below, so snapshot-only changes send nothing. + return workspaceSnapshotsOnDemand + ? withoutWorkspaceSnapshots(commandProviders) + : commandProviders; + }, ).pipe( // Compare initial paired values with the snapshot; blindly dropping them loses changes between snapshot and subscription. (updates) => Stream.concat(rpcInitialItems([config.providers]), updates), diff --git a/apps/web/src/components/chat/ChatMarkdownMedia.tsx b/apps/web/src/components/chat/ChatMarkdownMedia.tsx index d74746378d93..03d18a55324b 100644 --- a/apps/web/src/components/chat/ChatMarkdownMedia.tsx +++ b/apps/web/src/components/chat/ChatMarkdownMedia.tsx @@ -202,6 +202,7 @@ export function ChatMarkdownImage(props: { alt={props.alt} data-markdown-copy={props.copyMarkdown} decoding="async" + loading="lazy" draggable={false} className={cn( CHAT_MARKDOWN_IMAGE_SIZE_CLASS_NAME, @@ -258,6 +259,7 @@ export function ChatMarkdownImage(props: { src={src} alt={props.alt} decoding="async" + loading="lazy" draggable={false} className="invisible absolute inset-0 size-full" {...imageEvents(src)} diff --git a/apps/web/src/components/chat/TimelineUserMessageRow.tsx b/apps/web/src/components/chat/TimelineUserMessageRow.tsx index 5a24df014aeb..e76116ba9121 100644 --- a/apps/web/src/components/chat/TimelineUserMessageRow.tsx +++ b/apps/web/src/components/chat/TimelineUserMessageRow.tsx @@ -499,6 +499,7 @@ function UserTimelineRow({ row }: { row: Extract diff --git a/apps/web/src/components/chat/useProviderWorkspaceSnapshotRefresh.ts b/apps/web/src/components/chat/useProviderWorkspaceSnapshotRefresh.ts index bda0641e9091..7256052cebbb 100644 --- a/apps/web/src/components/chat/useProviderWorkspaceSnapshotRefresh.ts +++ b/apps/web/src/components/chat/useProviderWorkspaceSnapshotRefresh.ts @@ -14,6 +14,8 @@ import { useEffect, useEffectEvent, useRef, useState } from "react"; import { useStableCallback } from "../../hooks/useStableCallback"; const WORKSPACE_SNAPSHOT_RETRY_COOLDOWN_MS = 10_000; +// A cwd that never scans, such as a deleted worktree, would otherwise retry forever. +const WORKSPACE_SNAPSHOT_MAX_ATTEMPTS = 3; /** * Keeps the selected provider's workspace snapshot fresh for the composer's @@ -32,6 +34,7 @@ export function useProviderWorkspaceSnapshotRefresh(input: { // The last scan this composer asked for. A request inside the TTL is not // repeated, so a client clock ahead of the server's cannot loop rescans. const workspaceRefreshKeyRef = useRef<{ key: string; requestedAt: number } | null>(null); + const failedRefreshesRef = useRef<{ key: string; count: number } | null>(null); const [workspaceRefreshRetry, setWorkspaceRefreshRetry] = useState<{ key: string; notBefore: number; @@ -96,6 +99,11 @@ export function useProviderWorkspaceSnapshotRefresh(input: { workspaceRefreshKeyRef.current = request; const retryLater = () => { if (workspaceRefreshKeyRef.current !== request) return; + const failed = failedRefreshesRef.current; + const count = failed?.key === key ? failed.count + 1 : 1; + failedRefreshesRef.current = { key, count }; + // Keeping the request marker lets the TTL hold off the next scan. + if (count >= WORKSPACE_SNAPSHOT_MAX_ATTEMPTS) return; workspaceRefreshKeyRef.current = null; setWorkspaceRefreshRetry({ key, @@ -115,6 +123,7 @@ export function useProviderWorkspaceSnapshotRefresh(input: { gitCwd, ); if (!refreshedSnapshot) retryLater(); + else failedRefreshesRef.current = null; }, retryLater); }; const refreshWorkspaceSnapshotOnChange = useEffectEvent(refreshWorkspaceSnapshotIfStale); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index ba30a863e939..ebd05d0b2d30 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -170,6 +170,11 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}), ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}), }; + // Config state merges the snapshots it fetches back in, so no subscriber needs them broadcast. + const onDemandSnapshots = (input: ServerConfigSubscriptionInput) => ({ + ...input, + workspaceSnapshotsOnDemand: true, + }); const connect = Effect.fnUntraced(function* (connection: PreparedConnection) { const via = connection.target._tag === "RelayConnectionTarget" ? "relay" : "direct"; @@ -248,7 +253,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( detail: `${connection.label} config subscription ended.`, }); const serverConfigSource = protocolClient[WS_METHODS.subscribeServerConfig]( - serverConfigInput, + onDemandSnapshots(serverConfigInput), ).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -387,7 +392,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( Effect.as( Equal.equals(input, serverConfigInput) ? serverConfigEvents - : protocolClient[WS_METHODS.subscribeServerConfig](input), + : protocolClient[WS_METHODS.subscribeServerConfig](onDemandSnapshots(input)), ), ), ); diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index 09b28c539d33..9ee4e55697fa 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -1,5 +1,6 @@ import { EnvironmentId, + ProviderInstanceId, ServerObservability, EnvironmentThemeId, UsageLimitSourceId, @@ -39,6 +40,8 @@ import { makeEnvironmentServerWelcomeState, makeEnvironmentServerConfigState, isLegacyUpdateHandoffLoss, + recordFetchedWorkspaceSnapshot, + withFetchedWorkspaceSnapshots, matchesServerUpdateReadyEvent, matchesServerUpdateResumeEvent, nudgeReconnectDuringUpdateRestart, @@ -937,6 +940,87 @@ describe("server state projection", () => { ).toBe(live); }); + it("merges fetched workspace snapshots only from servers that omit them", () => { + const claude = ProviderInstanceId.make("claudeAgent"); + const codex = ProviderInstanceId.make("codex"); + const snapshot = (cwd: string, command: string) => ({ + cwd, + checkedAt: "2026-10-25T00:00:00.000Z", + slashCommands: [{ name: command }], + skills: [], + }); + const provider = ( + instanceId: ProviderInstanceId, + workspaceSnapshots?: ReadonlyArray>, + slashCommands: ReadonlyArray<{ readonly name: string }> = [], + ) => + ({ + instanceId, + enabled: true, + slashCommands, + skills: [], + ...(workspaceSnapshots ? { workspaceSnapshots } : {}), + }) as unknown as ServerConfig["providers"][number]; + const configWith = (onDemand: boolean, providers: ServerConfig["providers"]) => + ({ + ...CONFIG, + environment: { capabilities: { providerWorkspaceSnapshotsOnDemand: onDemand } }, + providers, + }) as unknown as ServerConfig; + const claudeA = snapshot("/a", "claude"); + const record = ( + current: Parameters[0], + instanceId: ProviderInstanceId, + fresh: boolean, + reply: ServerConfig["providers"], + ) => + recordFetchedWorkspaceSnapshot( + current, + { instanceId, cwd: "/a", fresh }, + reply, + provider(instanceId), + ); + + let fetched = record([], claude, false, [provider(claude, [claudeA])]); + fetched = record(fetched, codex, false, [provider(codex, [snapshot("/a", "codex")])]); + const merged = withFetchedWorkspaceSnapshots( + configWith(true, [provider(claude), provider(codex)]), + fetched, + ); + expect(merged.providers[0]?.workspaceSnapshots).toEqual([claudeA]); + // An unchanged answer keeps the cache, so config readers do not re-render. + expect( + record(fetched, claude, false, [ + provider(claude, [JSON.parse(JSON.stringify(claudeA)) as typeof claudeA]), + ]), + ).toBe(fetched); + + // Health checks and usage updates leave the snapshot in place. + expect( + withFetchedWorkspaceSnapshots( + configWith(true, [ + { ...provider(claude), checkedAt: "2026-10-25T00:09:00.000Z", usageLimits: {} }, + ] as unknown as ServerConfig["providers"]), + fetched, + ).providers[0]?.workspaceSnapshots, + ).toEqual([claudeA]); + + // New provider-wide commands, such as a live session's, retire the old snapshot. + expect( + withFetchedWorkspaceSnapshots( + configWith(true, [provider(claude, undefined, [{ name: "live" }])]), + fetched, + ).providers[0]?.workspaceSnapshots, + ).toBeUndefined(); + + // Older servers send every snapshot and every drop; their config stands. + const legacy = configWith(false, [provider(claude)]); + expect(withFetchedWorkspaceSnapshots(legacy, fetched)).toBe(legacy); + + // A fresh scan drops the cwd's other provider snapshots, as the server does. + expect(record(fetched, claude, true, [])).toEqual([]); + }); + it.effect("starts from cached configuration and persists the live projection", () => Effect.gen(function* () { const events = yield* Queue.unbounded(); diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 7913f8fe84e2..0f917e81f1b9 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1,9 +1,11 @@ import { type EnvironmentId, + type ProviderInstanceId, type ServerConfig, type ServerConfigStreamEvent, type ServerLifecycleWelcomePayload, type ServerLifecycleStreamReadyEvent, + type ServerProviderWorkspaceSnapshot, type ServerSelfUpdateProgressEvent, type ServerSelfUpdateResult, shouldRecommendDatabaseCompaction, @@ -11,6 +13,7 @@ import { } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Duration from "effect/Duration"; +import * as Equal from "effect/Equal"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -42,6 +45,7 @@ import { runCachePersistence } from "./cachePersistence.ts"; import { isRpcClientError, request, + requestGuarded, runStream, subscribe, subscribeDynamicWithSession, @@ -51,6 +55,7 @@ import type { RpcSession } from "../rpc/session.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import { applyServerConfigProjection, + reuseUnchanged, type ServerConfigProjection, withoutEnvironmentThemes, } from "./serverConfigProjection.ts"; @@ -648,6 +653,92 @@ export function resolveServerConfigValue( return initialConfig ?? projection?.config ?? null; } +type ServerProviderEntry = ServerConfig["providers"][number]; + +/** + * The provider-wide fields a workspace snapshot builds on. A live session, a sign-out, or a + * /usage-limits change shows up here, so a snapshot fetched before it no longer applies. + */ +const snapshotBasis = (provider: ServerProviderEntry | undefined) => + provider && { + enabled: provider.enabled, + slashCommands: provider.slashCommands, + skills: provider.skills, + }; + +interface FetchedWorkspaceSnapshot { + readonly instanceId: ProviderInstanceId; + readonly snapshot: ServerProviderWorkspaceSnapshot; + readonly basis: ReturnType; +} + +const MAX_FETCHED_WORKSPACE_SNAPSHOTS = 32; + +/** + * Servers with `providerWorkspaceSnapshotsOnDemand` leave per-cwd snapshots out of the config, + * so the ones this client fetched are merged back in. Older servers send every snapshot and + * their drops, so their config is used as is. + */ +export function withFetchedWorkspaceSnapshots( + config: ServerConfig, + fetched: ReadonlyArray, +): ServerConfig { + if ( + fetched.length === 0 || + config.environment.capabilities.providerWorkspaceSnapshotsOnDemand !== true + ) { + return config; + } + return { + ...config, + providers: config.providers.map((provider) => { + // A dropped snapshot reads as lost, and the composer fetches it again. + const own = fetched.filter( + (entry) => + entry.instanceId === provider.instanceId && + Equal.equals(entry.basis, snapshotBasis(provider)), + ); + return own.length === 0 + ? provider + : { ...provider, workspaceSnapshots: own.map((entry) => entry.snapshot) }; + }), + }; +} + +export function recordFetchedWorkspaceSnapshot( + current: ReadonlyArray, + input: { readonly instanceId: ProviderInstanceId; readonly cwd: string; readonly fresh: boolean }, + providers: ServerConfig["providers"], + configProvider: ServerProviderEntry | undefined, +): ReadonlyArray { + const basis = snapshotBasis(configProvider); + const snapshot = providers + .find((provider) => provider.instanceId === input.instanceId) + ?.workspaceSnapshots?.find((candidate) => candidate.cwd === input.cwd); + const stored = current.find( + (entry) => entry.instanceId === input.instanceId && entry.snapshot.cwd === input.cwd, + ); + // A cached answer from the server would otherwise re-render every config reader. + if ( + !input.fresh && + snapshot && + stored && + Equal.equals(stored.snapshot, snapshot) && + Equal.equals(stored.basis, basis) + ) { + return current; + } + // A fresh scan drops the cwd's snapshots for every provider, as the server does. + const kept = current.filter( + (entry) => + entry.snapshot.cwd !== input.cwd || (!input.fresh && entry.instanceId !== input.instanceId), + ); + if (!snapshot) return kept.length === current.length ? current : kept; + return [...kept, { instanceId: input.instanceId, snapshot, basis }].slice( + -MAX_FETCHED_WORKSPACE_SNAPSHOTS, + ); +} + export function createServerEnvironmentAtoms( runtime: Atom.AtomRuntime< EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore | R, @@ -686,6 +777,12 @@ export function createServerEnvironmentAtoms( readonly environmentId: EnvironmentId; readonly input: EnvironmentRpcInput; }) => configProjectionFamily(target.environmentId); + const fetchedWorkspaceSnapshotsAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make>([]).pipe( + Atom.keepAlive, + Atom.withLabel(`environment-data:server:fetched-workspace-snapshots:${environmentId}`), + ), + ); const emptyConfigAtom = Atom.make(null).pipe( Atom.withLabel("environment-data:server:config:empty"), ); @@ -693,14 +790,24 @@ export function createServerEnvironmentAtoms( if (environmentId === null) { return emptyConfigAtom; } + let previousProviders: ServerConfig["providers"] | undefined; return Atom.make((get): ServerConfig | null => { const projection = Option.getOrNull( AsyncResult.value(get(configProjection({ environmentId, input: {} }))), ); - return resolveServerConfigValue( + const config = resolveServerConfigValue( projection, get(options.initialConfigValueAtom(environmentId)), ); + if (config === null) return null; + const merged = withFetchedWorkspaceSnapshots( + config, + get(fetchedWorkspaceSnapshotsAtom(environmentId)), + ); + if (merged === config) return config; + // Keep provider identities stable across config events; the projection already reuses the rest. + previousProviders = reuseUnchanged(previousProviders, merged.providers); + return { ...merged, providers: previousProviders }; }).pipe(Atom.withLabel(`environment-data:server:config:${environmentId}`)); }); const updateStateValueAtom = Atom.family((environmentId: EnvironmentId) => @@ -1227,6 +1334,27 @@ export function createServerEnvironmentAtoms( refreshProviders: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:refresh-providers", tag: WS_METHODS.serverRefreshProviders, + execute: (input) => + requestGuarded(WS_METHODS.serverRefreshProviders, { + ...input, + workspaceSnapshotsOnDemand: true, + ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}), + }), + onSuccess: ({ environmentId, input }, registry, result) => + Effect.sync(() => { + const { instanceId, cwd } = input; + if (instanceId === undefined || cwd === undefined) return; + registry.update(fetchedWorkspaceSnapshotsAtom(environmentId), (current) => + recordFetchedWorkspaceSnapshot( + current, + { instanceId, cwd, fresh: input.fresh === true }, + result.providers, + registry + .get(configValueAtom(environmentId)) + ?.providers.find((provider) => provider.instanceId === instanceId), + ), + ); + }), concurrency: { mode: "singleFlight", key: ({ environmentId, input }) => diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 771ade1902f2..7b47d88202c7 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -102,6 +102,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Clear published themes when absent; older servers never send updates. */ environmentThemes: Schema.optionalKey(Schema.Boolean), usageLimitSources: Schema.optionalKey(Schema.Boolean), + /** Honors `workspaceSnapshotsOnDemand`; older servers send every snapshot and drop them in the stream. */ + providerWorkspaceSnapshotsOnDemand: Schema.optionalKey(Schema.Boolean), usagePriceOverrides: Schema.optionalKey(Schema.Boolean), usageModelAliases: Schema.optionalKey(Schema.Boolean), threadPinning: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 088935c52917..4920a748d5b9 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -612,13 +612,21 @@ const WsServerGetConfigRpc = Rpc.make(WS_METHODS.serverGetConfig, { error: Schema.Union([KeybindingsConfigError, ServerSettingsError, EnvironmentAuthorizationError]), }); -const WsServerRefreshProvidersRpc = Rpc.make(WS_METHODS.serverRefreshProviders, { +export const WsServerRefreshProvidersRpc = Rpc.make(WS_METHODS.serverRefreshProviders, { payload: Schema.Struct({ instanceId: Schema.optional(ProviderInstanceId), cwd: Schema.optional(TrimmedNonEmptyString), fresh: Schema.optional(Schema.Boolean), /** Bypass model caches; explicit rediscovery may open sessions, background refreshes must not. */ refreshModels: Schema.optional(Schema.Boolean), + /** + * Reply without snapshots, or with only the requested instance's cwd snapshot. Snapshots then + * reach the client only through these replies, so another client's scan shows on its next + * refresh, at most one TTL later. + */ + workspaceSnapshotsOnDemand: Schema.optional(Schema.Boolean), + /** As on `subscribeServerConfig`: add /usage-limits to the replied snapshot. */ + usageLimitsCommand: Schema.optional(Schema.Boolean), }), success: ServerProviderUpdatedPayload, error: Schema.Union([EnvironmentAuthorizationError, ProviderSetupError]), @@ -1791,6 +1799,8 @@ export const WsSubscribeServerConfigRpc = Rpc.make(WS_METHODS.subscribeServerCon usageLimitSources: Schema.optional(Schema.Boolean), /** Advertise client handling before the server offers /usage-limits, or older clients send it as a prompt. */ usageLimitsCommand: Schema.optional(Schema.Boolean), + /** Omit per-cwd provider snapshots; the client fetches its cwd through `server.refreshProviders`. */ + workspaceSnapshotsOnDemand: Schema.optional(Schema.Boolean), }), success: ServerConfigStreamEvent, error: Schema.Union([KeybindingsConfigError, ServerSettingsError, EnvironmentAuthorizationError]),