authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-04 17:46:12-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 17:46:12-04:00
log98a29cbaa1ccf8d8afb6d35e3e1d925ff9b5fa90
tree94d043a9376b756e81ad9f65ea54a711aaf5358c
parent91c66ac43ddad4f8697887009746f4de11736cb9
signature Signed by PGP key B5690EEEBB952194

fix: address usage limits and merge settlement regressions (#9784)

9 files changed, 308 insertions(+), 93 deletions(-)

apps/mobile/src/connection/runtime.ts+4-1
...@@ -30,7 +30,10 @@ type ConnectionLayerSource =...@@ -30,7 +30,10 @@ type ConnectionLayerSource =
30 | typeof mobileBackgroundActivityObserverLayer30 | typeof mobileBackgroundActivityObserverLayer
31 | typeof mobileBackgroundActivityReporterLayer;31 | typeof mobileBackgroundActivityReporterLayer;
3232
33const providedClientConnectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe(33const providedClientConnectionLayer = Layer.merge(
34 Connection.layerWithOptions({ usageLimitSources: true }),
35 snapshotLoaderLayer,
36).pipe(
34 Layer.provideMerge(37 Layer.provideMerge(
35 Layer.mergeAll(38 Layer.mergeAll(
36 runtimeContextLayer,39 runtimeContextLayer,
apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts+42
...@@ -1,4 +1,9 @@...@@ -1,4 +1,9 @@
1// @effect-diagnostics nodeBuiltinImport:off - cleanup uses Node's retrying rm, which the FileSystem service does not expose.1// @effect-diagnostics nodeBuiltinImport:off - cleanup uses Node's retrying rm, which the FileSystem service does not expose.
2import * as ClaudeSdk from "@anthropic-ai/claude-agent-sdk";
3import { vi } from "vite-plus/test";
4import * as Deferred from "effect/Deferred";
5import * as Fiber from "effect/Fiber";
6import * as TestClock from "effect/testing/TestClock";
2import { ClaudeSettings } from "@t3tools/contracts";7import { ClaudeSettings } from "@t3tools/contracts";
3import * as NodeFSP from "node:fs/promises";8import * as NodeFSP from "node:fs/promises";
4import * as NodeServices from "@effect/platform-node/NodeServices";9import * as NodeServices from "@effect/platform-node/NodeServices";
...@@ -14,6 +19,8 @@ import {...@@ -14,6 +19,8 @@ import {
14 probeClaudeCapabilities,19 probeClaudeCapabilities,
15} from "./ClaudeProvider.ts";20} from "./ClaudeProvider.ts";
1621
22vi.mock("@anthropic-ai/claude-agent-sdk", { spy: true });
23
17const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings);24const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings);
1825
19it("isolates Claude capability probes without dropping workspace setting sources", () => {26it("isolates Claude capability probes without dropping workspace setting sources", () => {
...@@ -181,3 +188,38 @@ it.layer(NodeServices.layer)("Claude capability probe SDK boundary", (it) => {...@@ -181,3 +188,38 @@ it.layer(NodeServices.layer)("Claude capability probe SDK boundary", (it) => {
181 }).pipe(Effect.scoped),188 }).pipe(Effect.scoped),
182 );189 );
183});190});
191
192it.effect("preserves initialized capabilities when optional usage times out", () =>
193 Effect.gen(function* () {
194 const usageStarted = yield* Deferred.make<void>();
195 let abortSignal: AbortSignal | undefined;
196 const query = vi.spyOn(ClaudeSdk, "query").mockImplementation(({ options }) => {
197 abortSignal = options?.abortController?.signal;
198 return {
199 initializationResult: async () => ({
200 account: { email: "dev@example.com", subscriptionType: "pro", tokenSource: "oauth" },
201 commands: [{ name: "review", description: "Review changes", argumentHint: "[path]" }],
202 }),
203 usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: () => {
204 Deferred.doneUnsafe(usageStarted, Effect.void);
205 return new Promise(() => {});
206 },
207 } as ReturnType<typeof ClaudeSdk.query>;
208 });
209 yield* Effect.addFinalizer(() => Effect.sync(() => query.mockRestore()));
210 const probe = yield* probeClaudeCapabilities(
211 decodeClaudeSettings({ binaryPath: "claude" }),
212 ).pipe(Effect.forkChild);
213 yield* Deferred.await(usageStarted);
214 yield* TestClock.adjust("4 seconds");
215 const capabilities = yield* Fiber.join(probe);
216 assert.equal(capabilities?.email, "dev@example.com");
217 assert.equal(capabilities?.subscriptionType, "pro");
218 assert.equal(capabilities?.tokenSource, "oauth");
219 assert.deepEqual(capabilities?.slashCommands, [
220 { name: "review", description: "Review changes", input: { hint: "[path]" } },
221 ]);
222 assert.equal(capabilities?.usage, undefined);
223 assert.equal(abortSignal?.aborted, true);
224 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
225);
apps/server/src/provider/Layers/ClaudeProvider.ts+33-30
...@@ -355,44 +355,47 @@ const probeClaudeCapabilities = (...@@ -355,44 +355,47 @@ const probeClaudeCapabilities = (
355 }),355 }),
356 });356 });
357 const init = await q.initializationResult();357 const init = await q.initializationResult();
358 // Usage is a second control round trip on the same process; a failure358 return { q, init };
359 // there must not cost the slash commands and account we already have.
360 const usage = await q.usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET().then(
361 (response) => ({
362 rate_limits_available: response.rate_limits_available,
363 rate_limits: response.rate_limits,
364 }),
365 () => undefined,
366 );
367 const account = init.account as
368 | {
369 readonly email?: string;
370 readonly subscriptionType?: string;
371 readonly tokenSource?: string;
372 readonly apiProvider?: string;
373 }
374 | undefined;
375 return {
376 email: account?.email,
377 subscriptionType: account?.subscriptionType,
378 tokenSource: account?.tokenSource,
379 apiProvider: account?.apiProvider,
380 slashCommands: parseClaudeInitializationCommands(init.commands),
381 ...(usage ? { usage } : {}),
382 } satisfies ClaudeCapabilitiesProbe;
383 });359 });
384 }).pipe(360 }).pipe(
361 Effect.timeout(CAPABILITIES_PROBE_TIMEOUT_MS),
362 Effect.flatMap(({ q, init }) =>
363 Effect.gen(function* () {
364 // Usage has its own deadline so a slow optional request cannot discard initialization.
365 const usageResult = yield* Effect.tryPromise(() =>
366 q.usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET(),
367 ).pipe(Effect.timeout(DEFAULT_TIMEOUT_MS), Effect.result);
368 const usage = Result.isSuccess(usageResult)
369 ? {
370 rate_limits_available: usageResult.success.rate_limits_available,
371 rate_limits: usageResult.success.rate_limits,
372 }
373 : undefined;
374 const account = init.account as
375 | {
376 readonly email?: string;
377 readonly subscriptionType?: string;
378 readonly tokenSource?: string;
379 readonly apiProvider?: string;
380 }
381 | undefined;
382 return {
383 email: account?.email,
384 subscriptionType: account?.subscriptionType,
385 tokenSource: account?.tokenSource,
386 apiProvider: account?.apiProvider,
387 slashCommands: parseClaudeInitializationCommands(init.commands),
388 ...(usage ? { usage } : {}),
389 } satisfies ClaudeCapabilitiesProbe;
390 }),
391 ),
385 Effect.ensuring(392 Effect.ensuring(
386 Effect.sync(() => {393 Effect.sync(() => {
387 if (!abort.signal.aborted) abort.abort();394 if (!abort.signal.aborted) abort.abort();
388 }),395 }),
389 ),396 ),
390 Effect.timeoutOption(CAPABILITIES_PROBE_TIMEOUT_MS),
391 Effect.result,397 Effect.result,
392 Effect.map((result) => {398 Effect.map((result) => (Result.isSuccess(result) ? result.success : undefined)),
393 if (Result.isFailure(result)) return undefined;
394 return Option.isSome(result.success) ? result.success.value : undefined;
395 }),
396 );399 );
397};400};
398401
apps/server/src/pullRequest/PullRequestService.test.ts+22-2
...@@ -1005,10 +1005,12 @@ it.effect("refuses an action the host never claimed it could run", () =>...@@ -1005,10 +1005,12 @@ it.effect("refuses an action the host never claimed it could run", () =>
1005 }),1005 }),
1006);1006);
10071007
1008it.effect("publishes a successful merge for immediate settlement", () =>1008it.effect("publishes a merge for immediate settlement only after host confirmation", () =>
1009 Effect.scoped(1009 Effect.scoped(
1010 Effect.gen(function* () {1010 Effect.gen(function* () {
1011 const mergedAt = "2026-09-03T02:00:00.000Z";1011 const mergedAt = "2026-09-03T02:00:00.000Z";
1012 let state: "open" | "merged" = "open";
1013 let confirmationFails = false;
1012 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };1014 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
1013 const service = yield* makeService({1015 const service = yield* makeService({
1014 projects: [1016 projects: [
...@@ -1016,7 +1018,17 @@ it.effect("publishes a successful merge for immediate settlement", () =>...@@ -1016,7 +1018,17 @@ it.effect("publishes a successful merge for immediate settlement", () =>
1016 ],1018 ],
1017 providers: [1019 providers: [
1018 fakeProvider("github", {1020 fakeProvider("github", {
1019 runAction: () => TestClock.setTime(Date.parse(mergedAt)),1021 getChangeRequestSummary: () =>
1022 confirmationFails
1023 ? Effect.fail(
1024 new PullRequestProviderError({
1025 provider: "github",
1026 operation: "getChangeRequestSummary",
1027 reason: "failed",
1028 detail: "HTTP 504",
1029 }),
1030 )
1031 : Effect.succeed({ ...changeRequest(1, mergedAt), state }),
1020 }),1032 }),
1021 ],1033 ],
1022 });1034 });
...@@ -1025,6 +1037,13 @@ it.effect("publishes a successful merge for immediate settlement", () =>...@@ -1025,6 +1037,13 @@ it.effect("publishes a successful merge for immediate settlement", () =>
1025 Effect.forkChild({ startImmediately: true }),1037 Effect.forkChild({ startImmediately: true }),
1026 );1038 );
10271039
1040 // Queueing succeeds while the host still reports an open PR.
1041 yield* service.runAction({ ...reference, action: "merge" });
1042 confirmationFails = true;
1043 yield* service.runAction({ ...reference, action: "merge" });
1044 confirmationFails = false;
1045 state = "merged";
1046 yield* TestClock.setTime(Date.parse(mergedAt));
1028 yield* service.runAction({1047 yield* service.runAction({
1029 ...reference,1048 ...reference,
1030 repository: " ACME/WEB ",1049 repository: " ACME/WEB ",
...@@ -1965,6 +1984,7 @@ it.effect("refuses a merge strategy the host does not offer", () =>...@@ -1965,6 +1984,7 @@ it.effect("refuses a merge strategy the host does not offer", () =>
1965 review: FULL_REVIEW,1984 review: FULL_REVIEW,
1966 reviewers: FULL_REVIEWERS,1985 reviewers: FULL_REVIEWERS,
1967 },1986 },
1987 getChangeRequestSummary: () => Effect.succeed(changeRequest(1, "2026-07-02T00:00:00Z")),
1968 runAction: (input) => {1988 runAction: (input) => {
1969 ranWith = input.mergeMethod ?? "merge";1989 ranWith = input.mergeMethod ?? "merge";
1970 return Effect.void;1990 return Effect.void;
apps/server/src/pullRequest/PullRequestService.ts+9
...@@ -2432,6 +2432,15 @@ export const make = Effect.gen(function* () {...@@ -2432,6 +2432,15 @@ export const make = Effect.gen(function* () {
2432 bumpRefEpoch({ ...input, repository });2432 bumpRefEpoch({ ...input, repository });
2433 listingsEpoch = ++epochCounter;2433 listingsEpoch = ++epochCounter;
2434 if (input.action === "merge") {2434 if (input.action === "merge") {
2435 // A successful merge action can merely enqueue the PR or enable auto-merge.
2436 const confirmed = yield* summaryUncached({ ...input, repository }).pipe(
2437 Effect.catch((error) =>
2438 Effect.logWarning("failed to confirm pull request merge", { error }).pipe(
2439 Effect.as(null),
2440 ),
2441 ),
2442 );
2443 if (confirmed?.state !== "merged") return;
2435 yield* PubSub.publish(mergedPullRequests, {2444 yield* PubSub.publish(mergedPullRequests, {
2436 projectId: input.projectId,2445 projectId: input.projectId,
2437 repository,2446 repository,
apps/web/src/components/usage/UsagePage.tsx+7-7
...@@ -1,3 +1,4 @@...@@ -1,3 +1,4 @@
1import { useAtomValue } from "@effect/atom-react";
1import type { UsageProviderKind } from "@t3tools/contracts";2import type { UsageProviderKind } from "@t3tools/contracts";
2import { CheckIcon, RefreshCwIcon, XIcon } from "lucide-react";3import { CheckIcon, RefreshCwIcon, XIcon } from "lucide-react";
3import { useMemo, useState } from "react";4import { useMemo, useState } from "react";
...@@ -6,7 +7,7 @@ import type { DailyTotals, HourlyTotals } from "@t3tools/shared/usageMerge";...@@ -6,7 +7,7 @@ import type { DailyTotals, HourlyTotals } from "@t3tools/shared/usageMerge";
67
7import { isElectron } from "../../env";8import { isElectron } from "../../env";
8import { cn } from "../../lib/utils";9import { cn } from "../../lib/utils";
9import { usePrimaryEnvironmentId } from "../../state/environments";10import { environmentPresentations } from "../../state/presentation";
10import { serverEnvironment } from "../../state/server";11import { serverEnvironment } from "../../state/server";
11import { useUsage, type EnvironmentUsageStatus } from "../../state/usage";12import { useUsage, type EnvironmentUsageStatus } from "../../state/usage";
12import { useAtomCommand } from "../../state/use-atom-command";13import { useAtomCommand } from "../../state/use-atom-command";
...@@ -69,7 +70,7 @@ export function UsagePage() {...@@ -69,7 +70,7 @@ export function UsagePage() {
69 const { days: windowDays, window } = windowSelection;70 const { days: windowDays, window } = windowSelection;
70 const isPast24Hours = windowDays === 1;71 const isPast24Hours = windowDays === 1;
71 const { merged, environments, isPending, isPartial, refresh } = useUsage(window);72 const { merged, environments, isPending, isPartial, refresh } = useUsage(window);
72 const primaryEnvironmentId = usePrimaryEnvironmentId();73 const presentations = useAtomValue(environmentPresentations.presentationsAtom);
73 const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, {74 const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, {
74 reportFailure: false,75 reportFailure: false,
75 });76 });
...@@ -115,12 +116,11 @@ export function UsagePage() {...@@ -115,12 +116,11 @@ export function UsagePage() {
115 });116 });
116 };117 };
117 const refreshWindow = () => {118 const refreshWindow = () => {
118 // On Limits the button re-probes every provider (and usage-limit source)
119 // on the primary environment; the live snapshots then flow in over the
120 // config stream, so nothing else needs to move.
121 if (showingLimits) {119 if (showingLimits) {
122 if (primaryEnvironmentId) {120 for (const [environmentId, presentation] of presentations) {
123 void refreshProviders({ environmentId: primaryEnvironmentId, input: {} });121 if (presentation.connection.phase === "connected" && presentation.serverConfig !== null) {
122 void refreshProviders({ environmentId, input: {} });
123 }
124 }124 }
125 return;125 return;
126 }126 }
apps/web/src/connection/runtime.ts+1-1
...@@ -31,7 +31,7 @@ type ConnectionLayerSource =...@@ -31,7 +31,7 @@ type ConnectionLayerSource =
31 | typeof backgroundActivityReporterLayer;31 | typeof backgroundActivityReporterLayer;
3232
33const providedClientConnectionLayer = Layer.merge(33const providedClientConnectionLayer = Layer.merge(
34 Connection.layerWithOptions({ environmentThemes: true }),34 Connection.layerWithOptions({ environmentThemes: true, usageLimitSources: true }),
35 snapshotLoaderLayer,35 snapshotLoaderLayer,
36).pipe(36).pipe(
37 Layer.provideMerge(37 Layer.provideMerge(
packages/client-runtime/src/rpc/session.test.ts+168-49
...@@ -8,6 +8,7 @@ import {...@@ -8,6 +8,7 @@ import {
8 ServerConfigStreamEvent,8 ServerConfigStreamEvent,
9 type ServerConfigStreamEvent as ServerConfigStreamEventType,9 type ServerConfigStreamEvent as ServerConfigStreamEventType,
10 WS_METHODS,10 WS_METHODS,
11 UsageLimitSourceId,
11} from "@t3tools/contracts";12} from "@t3tools/contracts";
12import { describe, expect, it } from "@effect/vitest";13import { describe, expect, it } from "@effect/vitest";
13import * as Cause from "effect/Cause";14import * as Cause from "effect/Cause";
...@@ -174,6 +175,29 @@ const THEME_SERVER_CONFIG: ServerConfigType = {...@@ -174,6 +175,29 @@ const THEME_SERVER_CONFIG: ServerConfigType = {
174 },175 },
175};176};
176const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG);177const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG);
178const SOURCE_SERVER_CONFIG: ServerConfigType = {
179 ...THEME_SERVER_CONFIG,
180 environment: {
181 ...THEME_SERVER_CONFIG.environment,
182 capabilities: { ...THEME_SERVER_CONFIG.environment.capabilities, usageLimitSources: true },
183 },
184};
185const SOURCE_EVENT: ServerConfigStreamEventType = {
186 version: 1,
187 type: "usageLimitSourcesUpdated",
188 payload: {
189 sources: [
190 {
191 id: UsageLimitSourceId.make("proxy"),
192 kind: "cliproxy",
193 label: "Proxy",
194 checkedAt: "2026-09-04T00:00:00Z",
195 accounts: [],
196 },
197 ],
198 },
199};
200
177const LEGACY_SERVER_CONFIG = {201const LEGACY_SERVER_CONFIG = {
178 ...ENCODED_SERVER_CONFIG,202 ...ENCODED_SERVER_CONFIG,
179 environment: {203 environment: {
...@@ -402,52 +426,133 @@ describe("RpcSessionFactory", () => {...@@ -402,52 +426,133 @@ describe("RpcSessionFactory", () => {
402 ),426 ),
403 );427 );
404428
405 it.effect("shares only a config subscription with the same theme opt-in", () =>429 for (const options of [
430 { environmentThemes: true },
431 { usageLimitSources: true },
432 { environmentThemes: true, usageLimitSources: true },
433 ]) {
434 it.effect(
435 `shares only a config subscription with the same opt-ins: ${JSON.stringify(options)}`,
436 () =>
437 Effect.scoped(
438 Effect.gen(function* () {
439 const { factory, sockets } = yield* makeFactory(options);
440 const session = yield* factory.connect(PREPARED);
441 const readyFiber = yield* Effect.forkChild(session.ready);
442 const socket = yield* awaitSocket(sockets);
443 socket.open();
444 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, options);
445 yield* Fiber.join(readyFiber);
446
447 const shared = yield* session.subscribeServerConfig(options).pipe(Stream.runHead);
448 expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } });
449 expect(
450 socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest),
451 ).toHaveLength(1);
452
453 const fallbackFiber = yield* session
454 .subscribeServerConfig({})
455 .pipe(Stream.runHead, Effect.forkChild);
456 const fallbackRequest = yield* awaitRequest(socket, 1);
457 expect(fallbackRequest).toMatchObject({
458 tag: WS_METHODS.subscribeServerConfig,
459 payload: {},
460 });
461 socket.serverMessage(
462 encodeJson({
463 _tag: "Chunk",
464 requestId: fallbackRequest.id,
465 values: [
466 {
467 version: 1,
468 type: "snapshot",
469 config: ENCODED_THEME_SERVER_CONFIG,
470 },
471 ],
472 }),
473 );
474 expect(yield* Fiber.join(fallbackFiber)).toMatchObject({
475 _tag: "Some",
476 value: { type: "snapshot" },
477 });
478 }),
479 ),
480 );
481 }
482
483 it.effect.each([
484 { usageLimitSources: true },
485 { environmentThemes: true, usageLimitSources: true },
486 ])("replays usage sources, removal, and capability downgrade with %j", (options) =>
406 Effect.scoped(487 Effect.scoped(
407 Effect.gen(function* () {488 Effect.gen(function* () {
408 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });489 const { factory, sockets } = yield* makeFactory(options);
409 const session = yield* factory.connect(PREPARED);490 const session = yield* factory.connect(PREPARED);
410 const readyFiber = yield* Effect.forkChild(session.ready);491 const ready = yield* Effect.forkChild(session.ready);
411 const socket = yield* awaitSocket(sockets);492 const socket = yield* awaitSocket(sockets);
412 socket.open();493 socket.open();
413 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {494 yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), options);
414 environmentThemes: true,495 yield* Fiber.join(ready);
415 });496 const observed = yield* Queue.unbounded<ServerConfigStreamEventType>();
416 yield* Fiber.join(readyFiber);497 yield* session.subscribeServerConfig(options).pipe(
417498 Stream.runForEach((event) => Queue.offer(observed, event)),
418 const shared = yield* session499 Effect.forkChild,
419 .subscribeServerConfig({ environmentThemes: true })500 );
420 .pipe(Stream.runHead);501 expect((yield* Queue.take(observed)).type).toBe("snapshot");
421 expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } });502 const themes: ServerConfigStreamEventType[] = options.environmentThemes
503 ? [{ version: 1, type: "environmentThemesUpdated", payload: { themes: [] } }]
504 : [];
505 for (const event of themes) {
506 yield* publishConfigEvents(socket, [event]);
507 expect(yield* Queue.take(observed)).toEqual(event);
508 }
509 const events: ServerConfigStreamEventType[] = [
510 SOURCE_EVENT,
511 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
512 SOURCE_EVENT,
513 { version: 1, type: "snapshot", config: THEME_SERVER_CONFIG },
514 ];
515 for (const event of events) {
516 yield* publishConfigEvents(socket, [event]);
517 expect(yield* Queue.take(observed)).toEqual(event);
518 const started = yield* Deferred.make<void>();
519 const replay = yield* session.subscribeServerConfig(options).pipe(
520 Stream.tap(() => Deferred.succeed(started, undefined)),
521 Stream.takeUntil((item) => item.type === "keybindingsUpdated"),
522 Stream.runCollect,
523 Effect.forkChild,
524 );
525 yield* Deferred.await(started);
526 // A live end marker makes a missing or stale replay event fail without a timeout.
527 const marker: ServerConfigStreamEventType = {
528 version: 1,
529 type: "keybindingsUpdated",
530 payload: { keybindings: [], issues: [] },
531 };
532 yield* publishConfigEvents(socket, [marker]);
533 expect(yield* Queue.take(observed)).toEqual(marker);
534 const replayed = Array.from(yield* Fiber.join(replay));
535 expect(replayed.slice(1)).toEqual([
536 ...themes,
537 ...(event.type === "snapshot" ? [] : [event]),
538 marker,
539 ]);
540 let projection = applyServerConfigProjection(Option.none(), {
541 version: 1,
542 type: "snapshot",
543 config: SOURCE_SERVER_CONFIG,
544 });
545 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
546 for (const item of replayed) projection = applyServerConfigProjection(projection, item);
547 expect(Option.getOrThrow(projection).config.usageLimitSources).toEqual(
548 event.type === "usageLimitSourcesUpdated" && event.payload.sources.length > 0
549 ? event.payload.sources
550 : undefined,
551 );
552 }
422 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(553 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
423 1,554 1,
424 );555 );
425
426 const fallbackFiber = yield* session
427 .subscribeServerConfig({})
428 .pipe(Stream.runHead, Effect.forkChild);
429 const fallbackRequest = yield* awaitRequest(socket, 1);
430 expect(fallbackRequest).toMatchObject({
431 tag: WS_METHODS.subscribeServerConfig,
432 payload: {},
433 });
434 socket.serverMessage(
435 encodeJson({
436 _tag: "Chunk",
437 requestId: fallbackRequest.id,
438 values: [
439 {
440 version: 1,
441 type: "snapshot",
442 config: ENCODED_THEME_SERVER_CONFIG,
443 },
444 ],
445 }),
446 );
447 expect(yield* Fiber.join(fallbackFiber)).toMatchObject({
448 _tag: "Some",
449 value: { type: "snapshot" },
450 });
451 }),556 }),
452 ),557 ),
453 );558 );
...@@ -546,16 +651,20 @@ describe("RpcSessionFactory", () => {...@@ -546,16 +651,20 @@ describe("RpcSessionFactory", () => {
546 ),651 ),
547 );652 );
548653
549 it.effect("recovers a slow subscriber after it misses theme deletion", () =>654 it.effect("recovers a slow subscriber after it misses theme and usage-source deletion", () =>
550 Effect.scoped(655 Effect.scoped(
551 Effect.gen(function* () {656 Effect.gen(function* () {
552 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });657 const { factory, sockets } = yield* makeFactory({
658 environmentThemes: true,
659 usageLimitSources: true,
660 });
553 const session = yield* factory.connect(PREPARED);661 const session = yield* factory.connect(PREPARED);
554 const readyFiber = yield* Effect.forkChild(session.ready);662 const readyFiber = yield* Effect.forkChild(session.ready);
555 const socket = yield* awaitSocket(sockets);663 const socket = yield* awaitSocket(sockets);
556 socket.open();664 socket.open();
557 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {665 yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), {
558 environmentThemes: true,666 environmentThemes: true,
667 usageLimitSources: true,
559 });668 });
560 yield* Fiber.join(readyFiber);669 yield* Fiber.join(readyFiber);
561670
...@@ -563,7 +672,7 @@ describe("RpcSessionFactory", () => {...@@ -563,7 +672,7 @@ describe("RpcSessionFactory", () => {
563 const releaseSlowSubscriber = yield* Deferred.make<void>();672 const releaseSlowSubscriber = yield* Deferred.make<void>();
564 let firstEvent = true;673 let firstEvent = true;
565 const slowSubscriber = yield* session674 const slowSubscriber = yield* session
566 .subscribeServerConfig({ environmentThemes: true })675 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
567 .pipe(676 .pipe(
568 Stream.mapEffect((event) => {677 Stream.mapEffect((event) => {
569 if (!firstEvent) return Effect.succeed(event);678 if (!firstEvent) return Effect.succeed(event);
...@@ -573,7 +682,7 @@ describe("RpcSessionFactory", () => {...@@ -573,7 +682,7 @@ describe("RpcSessionFactory", () => {
573 Effect.as(event),682 Effect.as(event),
574 );683 );
575 }),684 }),
576 Stream.take(3),685 Stream.take(4),
577 Stream.runCollect,686 Stream.runCollect,
578 Effect.forkChild,687 Effect.forkChild,
579 );688 );
...@@ -612,12 +721,18 @@ describe("RpcSessionFactory", () => {...@@ -612,12 +721,18 @@ describe("RpcSessionFactory", () => {
612 type: "settingsUpdated",721 type: "settingsUpdated",
613 payload: { settings: DEFAULT_SERVER_SETTINGS },722 payload: { settings: DEFAULT_SERVER_SETTINGS },
614 }));723 }));
615 const allEvents = [...themeEvents, ...settingsEvents];724 const sourceEvents: ServerConfigStreamEventType[] = [
725 SOURCE_EVENT,
726 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
727 ];
728 const allEvents = [...themeEvents, ...sourceEvents, ...settingsEvents];
616 const observedByFastSubscriber = yield* Queue.unbounded<ServerConfigStreamEventType>();729 const observedByFastSubscriber = yield* Queue.unbounded<ServerConfigStreamEventType>();
617 yield* session.subscribeServerConfig({ environmentThemes: true }).pipe(730 yield* session
618 Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)),731 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
619 Effect.forkChild,732 .pipe(
620 );733 Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)),
734 Effect.forkChild,
735 );
621 expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot");736 expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot");
622 for (const event of allEvents) {737 for (const event of allEvents) {
623 yield* publishConfigEvents(socket, [event]);738 yield* publishConfigEvents(socket, [event]);
...@@ -630,19 +745,23 @@ describe("RpcSessionFactory", () => {...@@ -630,19 +745,23 @@ describe("RpcSessionFactory", () => {
630 "snapshot",745 "snapshot",
631 "snapshot",746 "snapshot",
632 "environmentThemesUpdated",747 "environmentThemesUpdated",
748 "usageLimitSourcesUpdated",
633 ]);749 ]);
634 expect(recovered[2]).toMatchObject({ payload: { themes: [] } });750 expect(recovered[2]).toMatchObject({ payload: { themes: [] } });
751 expect(recovered[3]).toMatchObject({ payload: { sources: [] } });
635752
636 let projection = applyServerConfigProjection(Option.none(), {753 let projection = applyServerConfigProjection(Option.none(), {
637 version: 1,754 version: 1,
638 type: "snapshot",755 type: "snapshot",
639 config: THEME_SERVER_CONFIG,756 config: SOURCE_SERVER_CONFIG,
640 });757 });
641 projection = applyServerConfigProjection(projection, themeEvents[0]!);758 projection = applyServerConfigProjection(projection, themeEvents[0]!);
759 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
642 for (const event of recovered.slice(1)) {760 for (const event of recovered.slice(1)) {
643 projection = applyServerConfigProjection(projection, event);761 projection = applyServerConfigProjection(projection, event);
644 }762 }
645 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();763 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
764 expect(Option.getOrThrow(projection).config.usageLimitSources).toBeUndefined();
646 }),765 }),
647 ),766 ),
648 );767 );
packages/client-runtime/src/rpc/session.ts+22-3
...@@ -54,6 +54,7 @@ export interface RpcSession {...@@ -54,6 +54,7 @@ export interface RpcSession {
5454
55export interface RpcSessionOptions {55export interface RpcSessionOptions {
56 readonly environmentThemes?: boolean;56 readonly environmentThemes?: boolean;
57 readonly usageLimitSources?: boolean;
57}58}
5859
59export class RpcSessionFactory extends Context.Service<60export class RpcSessionFactory extends Context.Service<
...@@ -83,11 +84,16 @@ type EnvironmentThemesUpdatedEvent = Extract<...@@ -83,11 +84,16 @@ type EnvironmentThemesUpdatedEvent = Extract<
83 ServerConfigStreamEvent,84 ServerConfigStreamEvent,
84 { readonly type: "environmentThemesUpdated" }85 { readonly type: "environmentThemesUpdated" }
85>;86>;
87type UsageLimitSourcesUpdatedEvent = Extract<
88 ServerConfigStreamEvent,
89 { readonly type: "usageLimitSourcesUpdated" }
90>;
8691
87interface ServerConfigReplayState {92interface ServerConfigReplayState {
88 readonly projection: ServerConfigProjection;93 readonly projection: ServerConfigProjection;
89 readonly revision: number;94 readonly revision: number;
90 readonly themesEvent: EnvironmentThemesUpdatedEvent | undefined;95 readonly themesEvent: EnvironmentThemesUpdatedEvent | undefined;
96 readonly sourcesEvent: UsageLimitSourcesUpdatedEvent | undefined;
91}97}
9298
93interface BufferedServerConfigEvent {99interface BufferedServerConfigEvent {
...@@ -104,7 +110,11 @@ function serverConfigReplayEvents(...@@ -104,7 +110,11 @@ function serverConfigReplayEvents(
104 type: "snapshot" as const,110 type: "snapshot" as const,
105 config: withoutEnvironmentThemes(state.projection.config),111 config: withoutEnvironmentThemes(state.projection.config),
106 };112 };
107 return state.themesEvent === undefined ? [snapshot] : [snapshot, state.themesEvent];113 return [
114 snapshot,
115 ...(state.themesEvent === undefined ? [] : [state.themesEvent]),
116 ...(state.sourcesEvent === undefined ? [] : [state.sourcesEvent]),
117 ];
108}118}
109119
110function mapSessionRpcError(120function mapSessionRpcError(
...@@ -134,8 +144,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (...@@ -134,8 +144,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
134 options: RpcSessionOptions = {},144 options: RpcSessionOptions = {},
135) {145) {
136 const webSocketConstructor = yield* Socket.WebSocketConstructor;146 const webSocketConstructor = yield* Socket.WebSocketConstructor;
137 const serverConfigInput: ServerConfigSubscriptionInput =147 const serverConfigInput: ServerConfigSubscriptionInput = {
138 options.environmentThemes === true ? { environmentThemes: true } : {};148 ...(options.environmentThemes === true ? { environmentThemes: true } : {}),
149 ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}),
150 };
139151
140 const connect = Effect.fnUntraced(function* (connection: PreparedConnection) {152 const connect = Effect.fnUntraced(function* (connection: PreparedConnection) {
141 yield* Effect.annotateCurrentSpan({153 yield* Effect.annotateCurrentSpan({
...@@ -218,6 +230,13 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (...@@ -218,6 +230,13 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
218 event.config.environment.capabilities.environmentThemes !== true230 event.config.environment.capabilities.environmentThemes !== true
219 ? undefined231 ? undefined
220 : Option.getOrUndefined(current)?.themesEvent,232 : Option.getOrUndefined(current)?.themesEvent,
233 sourcesEvent:
234 event.type === "usageLimitSourcesUpdated"
235 ? event
236 : event.type === "snapshot" &&
237 event.config.environment.capabilities.usageLimitSources !== true
238 ? undefined
239 : Option.getOrUndefined(current)?.sourcesEvent,
221 } satisfies ServerConfigReplayState;240 } satisfies ServerConfigReplayState;
222 return [241 return [
223 Option.some({ event, replay: next, revision: next.revision }),242 Option.some({ event, replay: next, revision: next.revision }),