Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions apps/mobile/src/features/threads/use-composer-command-menu.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 };
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand All @@ -326,6 +334,7 @@ export function useComposerCommandMenu({
projectCwd,
);
if (!refreshed) retryLater();
else failedRefreshesRef.current = null;
}, retryLater);
}, [
draftMessage,
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/environment/ServerEnvironment.ts
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,7 @@ export const make = Effect.gen(function* () {
threadSnooze: true,
environmentThemes: true,
usageLimitSources: true,
providerWorkspaceSnapshotsOnDemand: true,
usagePriceOverrides: true,
usageModelAliases: true,
threadPinning: true,
Expand Down
21 changes: 20 additions & 1 deletion apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts
Original file line number Diff line number Diff line change
@@ -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 });

Expand All @@ -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)),
);
14 changes: 9 additions & 5 deletions apps/server/src/provider/ClaudeProvider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -431,18 +431,22 @@ export const probeClaudeWorkspaceSnapshot = Effect.fn("probeClaudeWorkspaceSnaps
): Effect.fn.Return<ProviderWorkspaceSnapshot, never, FileSystem.FileSystem | Path.Path> {
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,
slashCommands: dedupeSlashCommands([
COMPACT_SLASH_COMMAND,
...(capabilities?.slashCommands ?? []),
]),
slashCommandsPending: !capabilities,
slashCommandsPending: cwdExists && !capabilities,
};
});

Expand Down
92 changes: 92 additions & 0 deletions apps/server/src/provider/ProviderRegistry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>();
const releaseScan = yield* Deferred.make<void>();
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<void>(), 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");
Expand Down
41 changes: 29 additions & 12 deletions apps/server/src/provider/ProviderRegistry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -405,7 +406,7 @@ export const layer = Layer.effect(
cachedProviders.map((provider) => classifyCompatibility(provider, initialManifest)),
);
const workspaceRefreshesRef = yield* Ref.make<
ReadonlyMap<ProviderInstance, ReadonlyMap<string, symbol>>
ReadonlyMap<ProviderInstance, ReadonlyMap<string, Deferred.Deferred<void>>>
>(new Map());
const maintenanceActionStatesRef = yield* Ref.make<
ReadonlyMap<ProviderInstanceId, { readonly update?: ServerProviderUpdateState | undefined }>
Expand Down Expand Up @@ -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<void>();
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) =>
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading