From 9589d3b683e5e83b9c62f6ba751a1e92c65cacb1 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 4 Sep 2026 21:08:57 +0000 Subject: [PATCH 1/3] fix: address usage limits and merge settlement regressions --- apps/mobile/src/connection/runtime.ts | 5 +- .../Layers/ClaudeCapabilitiesProbe.test.ts | 42 ++++++++ .../src/provider/Layers/ClaudeProvider.ts | 63 +++++------ .../pullRequest/PullRequestService.test.ts | 24 ++++- .../src/pullRequest/PullRequestService.ts | 9 ++ apps/web/src/components/usage/UsagePage.tsx | 39 +++++-- apps/web/src/connection/runtime.ts | 2 +- .../client-runtime/src/rpc/session.test.ts | 100 +++++++++--------- packages/client-runtime/src/rpc/session.ts | 7 +- 9 files changed, 200 insertions(+), 91 deletions(-) diff --git a/apps/mobile/src/connection/runtime.ts b/apps/mobile/src/connection/runtime.ts index ee224ce9f6ed..662a4dcdf70c 100644 --- a/apps/mobile/src/connection/runtime.ts +++ b/apps/mobile/src/connection/runtime.ts @@ -30,7 +30,10 @@ type ConnectionLayerSource = | typeof mobileBackgroundActivityObserverLayer | typeof mobileBackgroundActivityReporterLayer; -const providedClientConnectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe( +const providedClientConnectionLayer = Layer.merge( + Connection.layerWithOptions({ usageLimitSources: true }), + snapshotLoaderLayer, +).pipe( Layer.provideMerge( Layer.mergeAll( runtimeContextLayer, diff --git a/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts b/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts index f167955fbaec..232b8cc02d00 100644 --- a/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts +++ b/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts @@ -1,4 +1,9 @@ // @effect-diagnostics nodeBuiltinImport:off - cleanup uses Node's retrying rm, which the FileSystem service does not expose. +import * as ClaudeSdk from "@anthropic-ai/claude-agent-sdk"; +import { vi } from "vite-plus/test"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; import { ClaudeSettings } from "@t3tools/contracts"; import * as NodeFSP from "node:fs/promises"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -14,6 +19,8 @@ import { probeClaudeCapabilities, } from "./ClaudeProvider.ts"; +vi.mock("@anthropic-ai/claude-agent-sdk", { spy: true }); + const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); it("isolates Claude capability probes without dropping workspace setting sources", () => { @@ -181,3 +188,38 @@ it.layer(NodeServices.layer)("Claude capability probe SDK boundary", (it) => { }).pipe(Effect.scoped), ); }); + +it.effect("preserves initialized capabilities when optional usage times out", () => + Effect.gen(function* () { + const usageStarted = yield* Deferred.make(); + let abortSignal: AbortSignal | undefined; + const query = vi.spyOn(ClaudeSdk, "query").mockImplementation(({ options }) => { + abortSignal = options?.abortController?.signal; + return { + initializationResult: async () => ({ + account: { email: "dev@example.com", subscriptionType: "pro", tokenSource: "oauth" }, + commands: [{ name: "review", description: "Review changes", argumentHint: "[path]" }], + }), + usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: () => { + Deferred.doneUnsafe(usageStarted, Effect.void); + return new Promise(() => {}); + }, + } as ReturnType; + }); + yield* Effect.addFinalizer(() => Effect.sync(() => query.mockRestore())); + const probe = yield* probeClaudeCapabilities( + decodeClaudeSettings({ binaryPath: "claude" }), + ).pipe(Effect.forkChild); + yield* Deferred.await(usageStarted); + yield* TestClock.adjust("4 seconds"); + const capabilities = yield* Fiber.join(probe); + assert.equal(capabilities?.email, "dev@example.com"); + assert.equal(capabilities?.subscriptionType, "pro"); + assert.equal(capabilities?.tokenSource, "oauth"); + assert.deepEqual(capabilities?.slashCommands, [ + { name: "review", description: "Review changes", input: { hint: "[path]" } }, + ]); + assert.equal(capabilities?.usage, undefined); + assert.equal(abortSignal?.aborted, true); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/provider/Layers/ClaudeProvider.ts b/apps/server/src/provider/Layers/ClaudeProvider.ts index e4ec8c522da7..e3d2c6ab565d 100644 --- a/apps/server/src/provider/Layers/ClaudeProvider.ts +++ b/apps/server/src/provider/Layers/ClaudeProvider.ts @@ -355,44 +355,47 @@ const probeClaudeCapabilities = ( }), }); const init = await q.initializationResult(); - // Usage is a second control round trip on the same process; a failure - // there must not cost the slash commands and account we already have. - const usage = await q.usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET().then( - (response) => ({ - rate_limits_available: response.rate_limits_available, - rate_limits: response.rate_limits, - }), - () => undefined, - ); - const account = init.account as - | { - readonly email?: string; - readonly subscriptionType?: string; - readonly tokenSource?: string; - readonly apiProvider?: string; - } - | undefined; - return { - email: account?.email, - subscriptionType: account?.subscriptionType, - tokenSource: account?.tokenSource, - apiProvider: account?.apiProvider, - slashCommands: parseClaudeInitializationCommands(init.commands), - ...(usage ? { usage } : {}), - } satisfies ClaudeCapabilitiesProbe; + return { q, init }; }); }).pipe( + Effect.timeout(CAPABILITIES_PROBE_TIMEOUT_MS), + Effect.flatMap(({ q, init }) => + Effect.gen(function* () { + // Usage has its own deadline so a slow optional request cannot discard initialization. + const usageResult = yield* Effect.tryPromise(() => + q.usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET(), + ).pipe(Effect.timeout(DEFAULT_TIMEOUT_MS), Effect.result); + const usage = Result.isSuccess(usageResult) + ? { + rate_limits_available: usageResult.success.rate_limits_available, + rate_limits: usageResult.success.rate_limits, + } + : undefined; + const account = init.account as + | { + readonly email?: string; + readonly subscriptionType?: string; + readonly tokenSource?: string; + readonly apiProvider?: string; + } + | undefined; + return { + email: account?.email, + subscriptionType: account?.subscriptionType, + tokenSource: account?.tokenSource, + apiProvider: account?.apiProvider, + slashCommands: parseClaudeInitializationCommands(init.commands), + ...(usage ? { usage } : {}), + } satisfies ClaudeCapabilitiesProbe; + }), + ), Effect.ensuring( Effect.sync(() => { if (!abort.signal.aborted) abort.abort(); }), ), - Effect.timeoutOption(CAPABILITIES_PROBE_TIMEOUT_MS), Effect.result, - Effect.map((result) => { - if (Result.isFailure(result)) return undefined; - return Option.isSome(result.success) ? result.success.value : undefined; - }), + Effect.map((result) => (Result.isSuccess(result) ? result.success : undefined)), ); }; diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 1f2ff59a94d3..43ba73d325ef 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -1005,10 +1005,12 @@ it.effect("refuses an action the host never claimed it could run", () => }), ); -it.effect("publishes a successful merge for immediate settlement", () => +it.effect("publishes a merge for immediate settlement only after host confirmation", () => Effect.scoped( Effect.gen(function* () { const mergedAt = "2026-09-03T02:00:00.000Z"; + let state: "open" | "merged" = "open"; + let confirmationFails = false; const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; const service = yield* makeService({ projects: [ @@ -1016,7 +1018,17 @@ it.effect("publishes a successful merge for immediate settlement", () => ], providers: [ fakeProvider("github", { - runAction: () => TestClock.setTime(Date.parse(mergedAt)), + getChangeRequestSummary: () => + confirmationFails + ? Effect.fail( + new PullRequestProviderError({ + provider: "github", + operation: "getChangeRequestSummary", + reason: "failed", + detail: "HTTP 504", + }), + ) + : Effect.succeed({ ...changeRequest(1, mergedAt), state }), }), ], }); @@ -1025,6 +1037,13 @@ it.effect("publishes a successful merge for immediate settlement", () => Effect.forkChild({ startImmediately: true }), ); + // Queueing succeeds while the host still reports an open PR. + yield* service.runAction({ ...reference, action: "merge" }); + confirmationFails = true; + yield* service.runAction({ ...reference, action: "merge" }); + confirmationFails = false; + state = "merged"; + yield* TestClock.setTime(Date.parse(mergedAt)); yield* service.runAction({ ...reference, repository: " ACME/WEB ", @@ -1965,6 +1984,7 @@ it.effect("refuses a merge strategy the host does not offer", () => review: FULL_REVIEW, reviewers: FULL_REVIEWERS, }, + getChangeRequestSummary: () => Effect.succeed(changeRequest(1, "2026-07-02T00:00:00Z")), runAction: (input) => { ranWith = input.mergeMethod ?? "merge"; return Effect.void; diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index 6a37ed935848..ceba7ce32e04 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -2432,6 +2432,15 @@ export const make = Effect.gen(function* () { bumpRefEpoch({ ...input, repository }); listingsEpoch = ++epochCounter; if (input.action === "merge") { + // A successful merge action can merely enqueue the PR or enable auto-merge. + const confirmed = yield* summaryUncached({ ...input, repository }).pipe( + Effect.catch((error) => + Effect.logWarning("failed to confirm pull request merge", { error }).pipe( + Effect.as(null), + ), + ), + ); + if (confirmed?.state !== "merged") return; yield* PubSub.publish(mergedPullRequests, { projectId: input.projectId, repository, diff --git a/apps/web/src/components/usage/UsagePage.tsx b/apps/web/src/components/usage/UsagePage.tsx index 4b62df374122..d99faa1c646f 100644 --- a/apps/web/src/components/usage/UsagePage.tsx +++ b/apps/web/src/components/usage/UsagePage.tsx @@ -1,4 +1,7 @@ +import { useAtomValue } from "@effect/atom-react"; import type { UsageProviderKind } from "@t3tools/contracts"; +import * as Option from "effect/Option"; +import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { CheckIcon, RefreshCwIcon, XIcon } from "lucide-react"; import { useMemo, useState } from "react"; @@ -6,7 +9,8 @@ import type { DailyTotals, HourlyTotals } from "@t3tools/shared/usageMerge"; import { isElectron } from "../../env"; import { cn } from "../../lib/utils"; -import { usePrimaryEnvironmentId } from "../../state/environments"; +import { environmentPresentations } from "../../state/presentation"; +import { environmentSession } from "../../state/session"; import { serverEnvironment } from "../../state/server"; import { useUsage, type EnvironmentUsageStatus } from "../../state/usage"; import { useAtomCommand } from "../../state/use-atom-command"; @@ -57,6 +61,30 @@ const WINDOW_OPTIONS = [ { days: 90, label: "90 days" }, ] as const; +const limitsRefreshEnvironmentsAtom = Atom.family((enabled: boolean) => + Atom.make((get) => + !enabled + ? [] + : [...get(environmentPresentations.presentationsAtom)] + .filter(([environmentId, presentation]) => { + if ( + presentation.connection.phase !== "connected" || + presentation.serverConfig === null + ) { + return false; + } + const session = Option.getOrNull( + AsyncResult.value(get(environmentSession.sessionStateAtom(environmentId))), + ); + return ( + session?.authenticated === true && + (session.scopes === undefined || session.scopes.includes("orchestration:operate")) + ); + }) + .map(([environmentId]) => environmentId), + ), +); + export function UsagePage() { const [windowSelection, setWindowSelection] = useState(() => ({ days: 30, @@ -68,7 +96,7 @@ export function UsagePage() { const { days: windowDays, window } = windowSelection; const isPast24Hours = windowDays === 1; const { merged, environments, isPending, isPartial, refresh } = useUsage(window); - const primaryEnvironmentId = usePrimaryEnvironmentId(); + const limitsRefreshEnvironments = useAtomValue(limitsRefreshEnvironmentsAtom(showingLimits)); const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, { reportFailure: false, }); @@ -114,12 +142,9 @@ export function UsagePage() { }); }; const refreshWindow = () => { - // On Limits the button re-probes every provider (and usage-limit source) - // on the primary environment; the live snapshots then flow in over the - // config stream, so nothing else needs to move. if (showingLimits) { - if (primaryEnvironmentId) { - void refreshProviders({ environmentId: primaryEnvironmentId, input: {} }); + for (const environmentId of limitsRefreshEnvironments) { + void refreshProviders({ environmentId, input: {} }); } return; } diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index 06c8bf0ccfed..eacce33a816c 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -31,7 +31,7 @@ type ConnectionLayerSource = | typeof backgroundActivityReporterLayer; const providedClientConnectionLayer = Layer.merge( - Connection.layerWithOptions({ environmentThemes: true }), + Connection.layerWithOptions({ environmentThemes: true, usageLimitSources: true }), snapshotLoaderLayer, ).pipe( Layer.provideMerge( diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index f8f940bc551f..cc4cb1a0e179 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -402,55 +402,59 @@ describe("RpcSessionFactory", () => { ), ); - it.effect("shares only a config subscription with the same theme opt-in", () => - Effect.scoped( - Effect.gen(function* () { - const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); - const session = yield* factory.connect(PREPARED); - const readyFiber = yield* Effect.forkChild(session.ready); - const socket = yield* awaitSocket(sockets); - socket.open(); - yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, { - environmentThemes: true, - }); - yield* Fiber.join(readyFiber); - - const shared = yield* session - .subscribeServerConfig({ environmentThemes: true }) - .pipe(Stream.runHead); - expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } }); - expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength( - 1, - ); - - const fallbackFiber = yield* session - .subscribeServerConfig({}) - .pipe(Stream.runHead, Effect.forkChild); - const fallbackRequest = yield* awaitRequest(socket, 1); - expect(fallbackRequest).toMatchObject({ - tag: WS_METHODS.subscribeServerConfig, - payload: {}, - }); - socket.serverMessage( - encodeJson({ - _tag: "Chunk", - requestId: fallbackRequest.id, - values: [ - { - version: 1, - type: "snapshot", - config: ENCODED_THEME_SERVER_CONFIG, - }, - ], + for (const options of [ + { environmentThemes: true }, + { usageLimitSources: true }, + { environmentThemes: true, usageLimitSources: true }, + ]) { + it.effect( + `shares only a config subscription with the same opt-ins: ${JSON.stringify(options)}`, + () => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(options); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, options); + yield* Fiber.join(readyFiber); + + const shared = yield* session.subscribeServerConfig(options).pipe(Stream.runHead); + expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } }); + expect( + socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest), + ).toHaveLength(1); + + const fallbackFiber = yield* session + .subscribeServerConfig({}) + .pipe(Stream.runHead, Effect.forkChild); + const fallbackRequest = yield* awaitRequest(socket, 1); + expect(fallbackRequest).toMatchObject({ + tag: WS_METHODS.subscribeServerConfig, + payload: {}, + }); + socket.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: fallbackRequest.id, + values: [ + { + version: 1, + type: "snapshot", + config: ENCODED_THEME_SERVER_CONFIG, + }, + ], + }), + ); + expect(yield* Fiber.join(fallbackFiber)).toMatchObject({ + _tag: "Some", + value: { type: "snapshot" }, + }); }), - ); - expect(yield* Fiber.join(fallbackFiber)).toMatchObject({ - _tag: "Some", - value: { type: "snapshot" }, - }); - }), - ), - ); + ), + ); + } it.effect("replays theme updates and deletion as authoritative events", () => Effect.scoped( diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 7d975be5c9d3..8a0bd4225772 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -54,6 +54,7 @@ export interface RpcSession { export interface RpcSessionOptions { readonly environmentThemes?: boolean; + readonly usageLimitSources?: boolean; } export class RpcSessionFactory extends Context.Service< @@ -134,8 +135,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( options: RpcSessionOptions = {}, ) { const webSocketConstructor = yield* Socket.WebSocketConstructor; - const serverConfigInput: ServerConfigSubscriptionInput = - options.environmentThemes === true ? { environmentThemes: true } : {}; + const serverConfigInput: ServerConfigSubscriptionInput = { + ...(options.environmentThemes === true ? { environmentThemes: true } : {}), + ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}), + }; const connect = Effect.fnUntraced(function* (connection: PreparedConnection) { yield* Effect.annotateCurrentSpan({ From a38cd0de3cb98adddfbfcb0250ce224f533bc833 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 4 Sep 2026 21:20:10 +0000 Subject: [PATCH 2/3] fix(web): refresh limits without waiting for session lookup --- apps/web/src/components/usage/UsagePage.tsx | 35 +++------------------ 1 file changed, 5 insertions(+), 30 deletions(-) diff --git a/apps/web/src/components/usage/UsagePage.tsx b/apps/web/src/components/usage/UsagePage.tsx index d99faa1c646f..dce6bc6e5220 100644 --- a/apps/web/src/components/usage/UsagePage.tsx +++ b/apps/web/src/components/usage/UsagePage.tsx @@ -1,7 +1,5 @@ import { useAtomValue } from "@effect/atom-react"; import type { UsageProviderKind } from "@t3tools/contracts"; -import * as Option from "effect/Option"; -import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { CheckIcon, RefreshCwIcon, XIcon } from "lucide-react"; import { useMemo, useState } from "react"; @@ -10,7 +8,6 @@ import type { DailyTotals, HourlyTotals } from "@t3tools/shared/usageMerge"; import { isElectron } from "../../env"; import { cn } from "../../lib/utils"; import { environmentPresentations } from "../../state/presentation"; -import { environmentSession } from "../../state/session"; import { serverEnvironment } from "../../state/server"; import { useUsage, type EnvironmentUsageStatus } from "../../state/usage"; import { useAtomCommand } from "../../state/use-atom-command"; @@ -61,30 +58,6 @@ const WINDOW_OPTIONS = [ { days: 90, label: "90 days" }, ] as const; -const limitsRefreshEnvironmentsAtom = Atom.family((enabled: boolean) => - Atom.make((get) => - !enabled - ? [] - : [...get(environmentPresentations.presentationsAtom)] - .filter(([environmentId, presentation]) => { - if ( - presentation.connection.phase !== "connected" || - presentation.serverConfig === null - ) { - return false; - } - const session = Option.getOrNull( - AsyncResult.value(get(environmentSession.sessionStateAtom(environmentId))), - ); - return ( - session?.authenticated === true && - (session.scopes === undefined || session.scopes.includes("orchestration:operate")) - ); - }) - .map(([environmentId]) => environmentId), - ), -); - export function UsagePage() { const [windowSelection, setWindowSelection] = useState(() => ({ days: 30, @@ -96,7 +69,7 @@ export function UsagePage() { const { days: windowDays, window } = windowSelection; const isPast24Hours = windowDays === 1; const { merged, environments, isPending, isPartial, refresh } = useUsage(window); - const limitsRefreshEnvironments = useAtomValue(limitsRefreshEnvironmentsAtom(showingLimits)); + const presentations = useAtomValue(environmentPresentations.presentationsAtom); const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, { reportFailure: false, }); @@ -143,8 +116,10 @@ export function UsagePage() { }; const refreshWindow = () => { if (showingLimits) { - for (const environmentId of limitsRefreshEnvironments) { - void refreshProviders({ environmentId, input: {} }); + for (const [environmentId, presentation] of presentations) { + if (presentation.connection.phase === "connected" && presentation.serverConfig !== null) { + void refreshProviders({ environmentId, input: {} }); + } } return; } From 6908eb0f86b30cf71f5ea48567ad278c6577c256 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 4 Sep 2026 21:38:17 +0000 Subject: [PATCH 3/3] fix(client-runtime): replay usage limit sources for shared subscribers --- .../client-runtime/src/rpc/session.test.ts | 137 ++++++++++++++++-- packages/client-runtime/src/rpc/session.ts | 18 ++- 2 files changed, 143 insertions(+), 12 deletions(-) diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index cc4cb1a0e179..f463ec280da1 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -8,6 +8,7 @@ import { ServerConfigStreamEvent, type ServerConfigStreamEvent as ServerConfigStreamEventType, WS_METHODS, + UsageLimitSourceId, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Cause from "effect/Cause"; @@ -174,6 +175,29 @@ const THEME_SERVER_CONFIG: ServerConfigType = { }, }; const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG); +const SOURCE_SERVER_CONFIG: ServerConfigType = { + ...THEME_SERVER_CONFIG, + environment: { + ...THEME_SERVER_CONFIG.environment, + capabilities: { ...THEME_SERVER_CONFIG.environment.capabilities, usageLimitSources: true }, + }, +}; +const SOURCE_EVENT: ServerConfigStreamEventType = { + version: 1, + type: "usageLimitSourcesUpdated", + payload: { + sources: [ + { + id: UsageLimitSourceId.make("proxy"), + kind: "cliproxy", + label: "Proxy", + checkedAt: "2026-09-04T00:00:00Z", + accounts: [], + }, + ], + }, +}; + const LEGACY_SERVER_CONFIG = { ...ENCODED_SERVER_CONFIG, environment: { @@ -456,6 +480,83 @@ describe("RpcSessionFactory", () => { ); } + it.effect.each([ + { usageLimitSources: true }, + { environmentThemes: true, usageLimitSources: true }, + ])("replays usage sources, removal, and capability downgrade with %j", (options) => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(options); + const session = yield* factory.connect(PREPARED); + const ready = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), options); + yield* Fiber.join(ready); + const observed = yield* Queue.unbounded(); + yield* session.subscribeServerConfig(options).pipe( + Stream.runForEach((event) => Queue.offer(observed, event)), + Effect.forkChild, + ); + expect((yield* Queue.take(observed)).type).toBe("snapshot"); + const themes: ServerConfigStreamEventType[] = options.environmentThemes + ? [{ version: 1, type: "environmentThemesUpdated", payload: { themes: [] } }] + : []; + for (const event of themes) { + yield* publishConfigEvents(socket, [event]); + expect(yield* Queue.take(observed)).toEqual(event); + } + const events: ServerConfigStreamEventType[] = [ + SOURCE_EVENT, + { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } }, + SOURCE_EVENT, + { version: 1, type: "snapshot", config: THEME_SERVER_CONFIG }, + ]; + for (const event of events) { + yield* publishConfigEvents(socket, [event]); + expect(yield* Queue.take(observed)).toEqual(event); + const started = yield* Deferred.make(); + const replay = yield* session.subscribeServerConfig(options).pipe( + Stream.tap(() => Deferred.succeed(started, undefined)), + Stream.takeUntil((item) => item.type === "keybindingsUpdated"), + Stream.runCollect, + Effect.forkChild, + ); + yield* Deferred.await(started); + // A live end marker makes a missing or stale replay event fail without a timeout. + const marker: ServerConfigStreamEventType = { + version: 1, + type: "keybindingsUpdated", + payload: { keybindings: [], issues: [] }, + }; + yield* publishConfigEvents(socket, [marker]); + expect(yield* Queue.take(observed)).toEqual(marker); + const replayed = Array.from(yield* Fiber.join(replay)); + expect(replayed.slice(1)).toEqual([ + ...themes, + ...(event.type === "snapshot" ? [] : [event]), + marker, + ]); + let projection = applyServerConfigProjection(Option.none(), { + version: 1, + type: "snapshot", + config: SOURCE_SERVER_CONFIG, + }); + projection = applyServerConfigProjection(projection, SOURCE_EVENT); + for (const item of replayed) projection = applyServerConfigProjection(projection, item); + expect(Option.getOrThrow(projection).config.usageLimitSources).toEqual( + event.type === "usageLimitSourcesUpdated" && event.payload.sources.length > 0 + ? event.payload.sources + : undefined, + ); + } + expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength( + 1, + ); + }), + ), + ); + it.effect("replays theme updates and deletion as authoritative events", () => Effect.scoped( Effect.gen(function* () { @@ -550,16 +651,20 @@ describe("RpcSessionFactory", () => { ), ); - it.effect("recovers a slow subscriber after it misses theme deletion", () => + it.effect("recovers a slow subscriber after it misses theme and usage-source deletion", () => Effect.scoped( Effect.gen(function* () { - const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); + const { factory, sockets } = yield* makeFactory({ + environmentThemes: true, + usageLimitSources: true, + }); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); - yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, { + yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), { environmentThemes: true, + usageLimitSources: true, }); yield* Fiber.join(readyFiber); @@ -567,7 +672,7 @@ describe("RpcSessionFactory", () => { const releaseSlowSubscriber = yield* Deferred.make(); let firstEvent = true; const slowSubscriber = yield* session - .subscribeServerConfig({ environmentThemes: true }) + .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true }) .pipe( Stream.mapEffect((event) => { if (!firstEvent) return Effect.succeed(event); @@ -577,7 +682,7 @@ describe("RpcSessionFactory", () => { Effect.as(event), ); }), - Stream.take(3), + Stream.take(4), Stream.runCollect, Effect.forkChild, ); @@ -616,12 +721,18 @@ describe("RpcSessionFactory", () => { type: "settingsUpdated", payload: { settings: DEFAULT_SERVER_SETTINGS }, })); - const allEvents = [...themeEvents, ...settingsEvents]; + const sourceEvents: ServerConfigStreamEventType[] = [ + SOURCE_EVENT, + { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } }, + ]; + const allEvents = [...themeEvents, ...sourceEvents, ...settingsEvents]; const observedByFastSubscriber = yield* Queue.unbounded(); - yield* session.subscribeServerConfig({ environmentThemes: true }).pipe( - Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)), - Effect.forkChild, - ); + yield* session + .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true }) + .pipe( + Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)), + Effect.forkChild, + ); expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot"); for (const event of allEvents) { yield* publishConfigEvents(socket, [event]); @@ -634,19 +745,23 @@ describe("RpcSessionFactory", () => { "snapshot", "snapshot", "environmentThemesUpdated", + "usageLimitSourcesUpdated", ]); expect(recovered[2]).toMatchObject({ payload: { themes: [] } }); + expect(recovered[3]).toMatchObject({ payload: { sources: [] } }); let projection = applyServerConfigProjection(Option.none(), { version: 1, type: "snapshot", - config: THEME_SERVER_CONFIG, + config: SOURCE_SERVER_CONFIG, }); projection = applyServerConfigProjection(projection, themeEvents[0]!); + projection = applyServerConfigProjection(projection, SOURCE_EVENT); for (const event of recovered.slice(1)) { projection = applyServerConfigProjection(projection, event); } expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined(); + expect(Option.getOrThrow(projection).config.usageLimitSources).toBeUndefined(); }), ), ); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 8a0bd4225772..d98a88100daa 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -84,11 +84,16 @@ type EnvironmentThemesUpdatedEvent = Extract< ServerConfigStreamEvent, { readonly type: "environmentThemesUpdated" } >; +type UsageLimitSourcesUpdatedEvent = Extract< + ServerConfigStreamEvent, + { readonly type: "usageLimitSourcesUpdated" } +>; interface ServerConfigReplayState { readonly projection: ServerConfigProjection; readonly revision: number; readonly themesEvent: EnvironmentThemesUpdatedEvent | undefined; + readonly sourcesEvent: UsageLimitSourcesUpdatedEvent | undefined; } interface BufferedServerConfigEvent { @@ -105,7 +110,11 @@ function serverConfigReplayEvents( type: "snapshot" as const, config: withoutEnvironmentThemes(state.projection.config), }; - return state.themesEvent === undefined ? [snapshot] : [snapshot, state.themesEvent]; + return [ + snapshot, + ...(state.themesEvent === undefined ? [] : [state.themesEvent]), + ...(state.sourcesEvent === undefined ? [] : [state.sourcesEvent]), + ]; } function mapSessionRpcError( @@ -221,6 +230,13 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( event.config.environment.capabilities.environmentThemes !== true ? undefined : Option.getOrUndefined(current)?.themesEvent, + sourcesEvent: + event.type === "usageLimitSourcesUpdated" + ? event + : event.type === "snapshot" && + event.config.environment.capabilities.usageLimitSources !== true + ? undefined + : Option.getOrUndefined(current)?.sourcesEvent, } satisfies ServerConfigReplayState; return [ Option.some({ event, replay: next, revision: next.revision }),