From 98a29cbaa1ccf8d8afb6d35e3e1d925ff9b5fa90 Mon Sep 17 00:00:00 2001 From: maria Date: Fri, 4 Sep 2026 17:46:12 -0400 Subject: [PATCH 01/12] fix: address usage limits and merge settlement regressions (#9784) --- 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 | 14 +- apps/web/src/connection/runtime.ts | 2 +- .../client-runtime/src/rpc/session.test.ts | 217 ++++++++++++++---- packages/client-runtime/src/rpc/session.ts | 25 +- 9 files changed, 308 insertions(+), 93 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 a96e4ac37410..c73a9a9ffb39 100644 --- a/apps/web/src/components/usage/UsagePage.tsx +++ b/apps/web/src/components/usage/UsagePage.tsx @@ -1,3 +1,4 @@ +import { useAtomValue } from "@effect/atom-react"; import type { UsageProviderKind } from "@t3tools/contracts"; import { CheckIcon, RefreshCwIcon, XIcon } from "lucide-react"; import { useMemo, useState } from "react"; @@ -6,7 +7,7 @@ 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 { serverEnvironment } from "../../state/server"; import { useUsage, type EnvironmentUsageStatus } from "../../state/usage"; import { useAtomCommand } from "../../state/use-atom-command"; @@ -69,7 +70,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 presentations = useAtomValue(environmentPresentations.presentationsAtom); const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, { reportFailure: false, }); @@ -115,12 +116,11 @@ 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, presentation] of presentations) { + if (presentation.connection.phase === "connected" && presentation.serverConfig !== null) { + 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..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: { @@ -402,52 +426,133 @@ describe("RpcSessionFactory", () => { ), ); - it.effect("shares only a config subscription with the same theme opt-in", () => + 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" }, + }); + }), + ), + ); + } + + 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({ environmentThemes: true }); + const { factory, sockets } = yield* makeFactory(options); const session = yield* factory.connect(PREPARED); - const readyFiber = yield* Effect.forkChild(session.ready); + const ready = 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" } }); + 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, ); - - 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" }, - }); }), ), ); @@ -546,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); @@ -563,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); @@ -573,7 +682,7 @@ describe("RpcSessionFactory", () => { Effect.as(event), ); }), - Stream.take(3), + Stream.take(4), Stream.runCollect, Effect.forkChild, ); @@ -612,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]); @@ -630,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 7d975be5c9d3..d98a88100daa 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< @@ -83,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 { @@ -104,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( @@ -134,8 +144,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({ @@ -218,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 }), From 389bbcc8d9dd5c463668944f7093052301c8a850 Mon Sep 17 00:00:00 2001 From: maria Date: Fri, 4 Sep 2026 17:47:20 -0400 Subject: [PATCH 02/12] fix(web): give toggle thumbs consistent inset spacing (#9805) --- apps/web/src/components/ui/switch.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/web/src/components/ui/switch.tsx b/apps/web/src/components/ui/switch.tsx index 0d45c88861e1..267b314caa40 100644 --- a/apps/web/src/components/ui/switch.tsx +++ b/apps/web/src/components/ui/switch.tsx @@ -12,7 +12,7 @@ function Switch({ return ( From 5a2f3ebf6ee6eab6124ad68272c2fb5d74bc0f5d Mon Sep 17 00:00:00 2001 From: maria Date: Fri, 4 Sep 2026 17:51:40 -0400 Subject: [PATCH 03/12] fix(web): use segmented controls for mode switches (#9781) --- apps/web/src/components/DiffPanel.tsx | 9 ++-- apps/web/src/components/GitActionsControl.tsx | 42 +++++++++---------- .../pullRequest/PullRequestCodeTab.tsx | 9 ++-- .../pullRequest/PullRequestMarkdownEditor.tsx | 32 +++++++------- .../settings/DiagnosticsSettings.tsx | 27 ++++++------ .../settings/ProviderSettingsPanel.tsx | 25 ++++++----- .../settings/ResourceTelemetryDiagnostics.tsx | 25 +++++------ .../components/settings/ThemeEditorPanel.tsx | 31 ++++++++------ .../settings/providerSettingsTabs.ts | 10 ----- apps/web/src/components/ui/toggle-group.tsx | 2 +- 10 files changed, 106 insertions(+), 106 deletions(-) delete mode 100644 apps/web/src/components/settings/providerSettingsTabs.ts diff --git a/apps/web/src/components/DiffPanel.tsx b/apps/web/src/components/DiffPanel.tsx index bad0e1b4ffe0..87f5f1ba56f8 100644 --- a/apps/web/src/components/DiffPanel.tsx +++ b/apps/web/src/components/DiffPanel.tsx @@ -800,8 +800,9 @@ export default function DiffPanel({ )} { const next = value[0]; @@ -810,10 +811,10 @@ export default function DiffPanel({ } }} > - + - + diff --git a/apps/web/src/components/GitActionsControl.tsx b/apps/web/src/components/GitActionsControl.tsx index 341444324e73..b7fc811d5127 100644 --- a/apps/web/src/components/GitActionsControl.tsx +++ b/apps/web/src/components/GitActionsControl.tsx @@ -42,6 +42,7 @@ import { Radio as RadioPrimitive } from "@base-ui/react/radio"; import { AzureDevOpsIcon, BitbucketIcon, GitHubIcon, GitLabIcon } from "~/components/Icons"; import { RadioGroup } from "~/components/ui/radio-group"; import { Spinner } from "~/components/ui/spinner"; +import { toggleVariants } from "~/components/ui/toggle"; import { cn } from "~/lib/utils"; import { buildGitActionProgressStages, @@ -831,32 +832,29 @@ function PublishRepositoryDialog(props: PublishRepositoryDialogProps) { Protocol - setPublishProtocol(value as SourceControlCloneProtocol) - } + onValueChange={(protocol) => { + if (protocol === "ssh" || protocol === "https") { + setPublishProtocol(protocol); + } + }} aria-labelledby="publish-protocol-label" disabled={publishRepositoryAction.isPending} - className="grid grid-cols-2 gap-2" > - {(["ssh", "https"] as const).map((value) => { - const isSelected = publishProtocol === value; - return ( - - {value === "ssh" ? "SSH" : "HTTPS"} - - ); - })} + {(["ssh", "https"] as const).map((protocol) => ( + + {protocol.toUpperCase()} + + ))} diff --git a/apps/web/src/components/pullRequest/PullRequestCodeTab.tsx b/apps/web/src/components/pullRequest/PullRequestCodeTab.tsx index 506d59dc6457..aa2d278ce249 100644 --- a/apps/web/src/components/pullRequest/PullRequestCodeTab.tsx +++ b/apps/web/src/components/pullRequest/PullRequestCodeTab.tsx @@ -1138,8 +1138,9 @@ export function PullRequestCodeTab({ ) : null} { const next = value[0]; @@ -1148,10 +1149,10 @@ export function PullRequestCodeTab({ } }} > - + - + diff --git a/apps/web/src/components/pullRequest/PullRequestMarkdownEditor.tsx b/apps/web/src/components/pullRequest/PullRequestMarkdownEditor.tsx index 5fbffdd08d3d..fde29d023774 100644 --- a/apps/web/src/components/pullRequest/PullRequestMarkdownEditor.tsx +++ b/apps/web/src/components/pullRequest/PullRequestMarkdownEditor.tsx @@ -5,6 +5,7 @@ import { cn } from "~/lib/utils"; import { Button } from "../ui/button"; import { Textarea } from "../ui/textarea"; +import { Toggle, ToggleGroup } from "../ui/toggle-group"; import { PullRequestMarkdown } from "./PullRequestMarkdown"; /** @@ -64,24 +65,19 @@ export function PullRequestMarkdownEditor({ onCancel(); }} > -
- - -
+ { + const mode = next[0]; + if (mode === "write" || mode === "preview") setPreview(mode === "preview"); + }} + > + Write + Preview + {preview ? (
{empty ? ( diff --git a/apps/web/src/components/settings/DiagnosticsSettings.tsx b/apps/web/src/components/settings/DiagnosticsSettings.tsx index b53dc5b8cb38..0b23fb2d2072 100644 --- a/apps/web/src/components/settings/DiagnosticsSettings.tsx +++ b/apps/web/src/components/settings/DiagnosticsSettings.tsx @@ -36,6 +36,7 @@ import { usePrimaryEnvironment } from "../../state/environments"; import { useCopyToClipboard } from "../../hooks/useCopyToClipboard"; import { Button } from "../ui/button"; import { ScrollArea } from "../ui/scroll-area"; +import { Toggle, ToggleGroup } from "../ui/toggle-group"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { toastManager } from "../ui/toast"; import { ExpandableText } from "./ExpandableText"; @@ -599,21 +600,23 @@ function ResourceHistoryWindowSelector({ onSelect: (windowMs: number) => void; }) { return ( -
+ { + const selected = RESOURCE_HISTORY_WINDOWS.find( + (option) => String(option.windowMs) === next[0], + ); + if (selected) onSelect(selected.windowMs); + }} + > {RESOURCE_HISTORY_WINDOWS.map((option) => ( - + ))} -
+ ); } diff --git a/apps/web/src/components/settings/ProviderSettingsPanel.tsx b/apps/web/src/components/settings/ProviderSettingsPanel.tsx index 1bb1b512e4a2..e5575ee9f785 100644 --- a/apps/web/src/components/settings/ProviderSettingsPanel.tsx +++ b/apps/web/src/components/settings/ProviderSettingsPanel.tsx @@ -72,6 +72,7 @@ import { NumberFieldInput, } from "../ui/number-field"; import { ScrollArea } from "../ui/scroll-area"; +import { Toggle, ToggleGroup } from "../ui/toggle-group"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { stackedThreadToast, toastManager } from "../ui/toast"; import { AddProviderInstanceDialog } from "./AddProviderInstanceDialog"; @@ -80,7 +81,6 @@ import { ProviderInstanceCard } from "./ProviderInstanceCard"; import { UsageProviderSettings } from "./UsageProviderSettings"; import { ProviderSetupSection, readAntigravityAuthMethod } from "./ProviderSetupSection"; import { DRIVER_OPTIONS, getDriverOption } from "./providerDriverMeta"; -import { providerSettingsTabClassName } from "./providerSettingsTabs"; import { searchableSetting } from "./settingsSearch"; import { backgroundActivityOverrideSettings, @@ -326,22 +326,25 @@ function ProviderSettingsPanelContent(target: ProviderSettingsTarget) { const deviceTabs = !onlyPrimaryDevice && options.length > 0 ? ( -
+ { + const environment = options.find((option) => option.environmentId === next[0]); + if (environment) setSelectedEnvironmentId(environment.environmentId); + }} + > {options.map((environment) => { const machine = resolveEnvironmentMachineKind(environment.serverConfig); - const selected = environment.environmentId === effectiveEnvironmentId; const detail = providerEnvironmentDetail(environment); const statusText = connectionStatusTitle(environment.connection); return ( setSelectedEnvironmentId(environment.environmentId)} - > + {detail}, {statusText} - + } /> @@ -366,7 +369,7 @@ function ProviderSettingsPanelContent(target: ProviderSettingsTarget) { ); })} -
+ ) : null; diff --git a/apps/web/src/components/settings/ResourceTelemetryDiagnostics.tsx b/apps/web/src/components/settings/ResourceTelemetryDiagnostics.tsx index 741cd3c3f37f..ca934952f8a9 100644 --- a/apps/web/src/components/settings/ResourceTelemetryDiagnostics.tsx +++ b/apps/web/src/components/settings/ResourceTelemetryDiagnostics.tsx @@ -45,6 +45,7 @@ import { useAtomCommand } from "../../state/use-atom-command"; import { formatRelativeTime } from "../../timestampFormat"; import { Button } from "../ui/button"; import { ScrollArea } from "../ui/scroll-area"; +import { Toggle, ToggleGroup } from "../ui/toggle-group"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { toastManager } from "../ui/toast"; import { @@ -378,21 +379,21 @@ function HistoryWindowSelector({ onSelect: (windowMs: number) => void; }) { return ( -
+ { + const selected = HISTORY_WINDOWS.find((option) => String(option.windowMs) === next[0]); + if (selected) onSelect(selected.windowMs); + }} + > {HISTORY_WINDOWS.map((option) => ( - + ))} -
+ ); } diff --git a/apps/web/src/components/settings/ThemeEditorPanel.tsx b/apps/web/src/components/settings/ThemeEditorPanel.tsx index 4e85bc8806e8..02d1dad458f1 100644 --- a/apps/web/src/components/settings/ThemeEditorPanel.tsx +++ b/apps/web/src/components/settings/ThemeEditorPanel.tsx @@ -38,6 +38,7 @@ import { cn } from "../../lib/utils"; import { Button } from "../ui/button"; import { Input } from "../ui/input"; import { Switch } from "../ui/switch"; +import { Toggle, ToggleGroup } from "../ui/toggle-group"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { getThemeRoleLabel, ThemeColorField } from "./ThemeColorPicker"; import { @@ -942,24 +943,17 @@ export function ThemeEditorPanel({ ); const renderAppearanceButton = (appearance: ThemeAppearance) => { - const isActive = activeAppearance === appearance; const lockReason = appearanceLockReason(appearance); // A locked mode stays hoverable so the tooltip can say why it is off; // a real disabled attribute would swallow the pointer events. const button = ( - + ); if (lockReason === null) return button; return ( @@ -973,10 +967,23 @@ export function ThemeEditorPanel({ const renderAppearanceButtons = () => (
Appearance -
+ { + const appearance = next[0]; + if ( + (appearance === "light" || appearance === "dark") && + appearanceLockReason(appearance) === null + ) { + setActiveAppearance(appearance); + } + }} + > {renderAppearanceButton("light")} {renderAppearanceButton("dark")} -
+
); diff --git a/apps/web/src/components/settings/providerSettingsTabs.ts b/apps/web/src/components/settings/providerSettingsTabs.ts deleted file mode 100644 index 1c4e828bacd9..000000000000 --- a/apps/web/src/components/settings/providerSettingsTabs.ts +++ /dev/null @@ -1,10 +0,0 @@ -import { cn } from "../../lib/utils"; - -export function providerSettingsTabClassName(selected: boolean): string { - return cn( - "relative flex h-full shrink-0 cursor-pointer items-center rounded-sm px-3 text-xs font-medium outline-none transition-colors focus-visible:ring-2 focus-visible:ring-ring focus-visible:ring-inset", - selected - ? "text-foreground after:absolute after:inset-x-0 after:bottom-0 after:h-0.5 after:bg-primary" - : "text-muted-foreground hover:text-foreground", - ); -} diff --git a/apps/web/src/components/ui/toggle-group.tsx b/apps/web/src/components/ui/toggle-group.tsx index ae10fe81611f..23501ec7a96c 100644 --- a/apps/web/src/components/ui/toggle-group.tsx +++ b/apps/web/src/components/ui/toggle-group.tsx @@ -16,7 +16,7 @@ const ToggleGroupContext = React.createContext Date: Fri, 4 Sep 2026 23:59:27 +0200 Subject: [PATCH 04/12] test(web): isolate usage page button mock from shared workers UsagePage.test mocks Button as a host-element string. Under isolate:false that stub leaked into ComposerControl and dropped size classes. --- apps/web/vite.config.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/apps/web/vite.config.ts b/apps/web/vite.config.ts index 2e5fad8591b6..63e51d5372e2 100644 --- a/apps/web/vite.config.ts +++ b/apps/web/vite.config.ts @@ -125,10 +125,17 @@ const isolatedUnitTestFiles = [ // because it does not share registries; the fork does. "src/components/diffs/StyledDiffCodeView.test.tsx", "src/components/settings/AddProviderInstanceDialog.environment.test.tsx", + // Mocks `../ui/button` as a host-element string; under isolate:false that + // stub leaks into later files (ComposerControl then renders variant as a + // DOM attribute and drops size classes). + "src/components/settings/ProjectIconPickerDialog.test.tsx", + "src/components/settings/ProviderSettingsPanel.environment.test.tsx", // Mocks `../ui/toast`; under isolate:false an earlier file binds the real // toast manager and the release-link error toast is never recorded. "src/components/sidebar/SidebarUpdateReleaseNotes.test.tsx", - "src/components/settings/ProviderSettingsPanel.environment.test.tsx", + // Mocks `react` useState and `../ui/button` as "button"; same isolate:false + // Button leak as ProjectIconPickerDialog.test.tsx. + "src/components/usage/UsagePage.test.tsx", "src/connection/storage.test.ts", "src/contextMenuFallback.test.ts", "src/environments/primary/bootstrap.test.ts", From b906ce2d73d025e801877f42a35b7a2f5629806f Mon Sep 17 00:00:00 2001 From: maria Date: Fri, 4 Sep 2026 18:35:42 -0400 Subject: [PATCH 05/12] fix(server): recover opted-in threads after machine restarts (#9803) --- .../settings/DesktopClientSettings.test.ts | 1 - .../provider/Layers/ProviderService.test.ts | 119 ++++ .../src/provider/Layers/ProviderService.ts | 17 + .../serverRuntimeStartup.reconcile.test.ts | 629 +++++++++++++----- apps/server/src/serverRuntimeStartup.ts | 83 ++- .../components/ServerUpdateAction.test.tsx | 3 +- .../web/src/components/ServerUpdateAction.tsx | 5 +- .../components/settings/SettingsPanels.tsx | 9 +- .../src/components/settings/settingsSearch.ts | 6 +- docs/user/updating.md | 10 +- packages/contracts/src/settings.ts | 9 +- 11 files changed, 708 insertions(+), 183 deletions(-) diff --git a/apps/desktop/src/settings/DesktopClientSettings.test.ts b/apps/desktop/src/settings/DesktopClientSettings.test.ts index 28cce3cfb507..0d9ddc8fde91 100644 --- a/apps/desktop/src/settings/DesktopClientSettings.test.ts +++ b/apps/desktop/src/settings/DesktopClientSettings.test.ts @@ -26,7 +26,6 @@ const clientSettings: ClientSettings = { confirmThreadArchive: true, confirmThreadDelete: false, confirmThreadUnpin: false, - continueThreadsAfterServerUpdate: true, contextWindowMeterEnabled: false, composerCollapseOnBlur: false, composerCollapseOnScroll: true, diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 5e59960685a7..95e5a7983346 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -452,6 +452,125 @@ function makeProviderServiceLayer( }; } +for (const [enabled, completed] of [ + [false, false], + [true, false], + [true, true], +] as const) { + it.effect( + `persists shutdown recovery before stopping providers when enabled=${enabled}, completed=${completed}`, + () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const persistence = yield* Layer.build( + ProviderSessionDirectoryLive.pipe( + Layer.provide( + ProviderSessionRuntime.layer.pipe(Layer.provide(SqlitePersistenceMemory)), + ), + ), + ); + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory.pipe( + Effect.provide(persistence), + ); + const threadId = asThreadId("shutdown-recovery"); + const turnId = asTurnId("shutdown-recovery-turn"); + const scope = yield* Scope.make(); + const services = yield* Layer.build( + makeProviderServiceLive().pipe( + Layer.provide( + Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, directory), + ), + Layer.provide( + Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistry, + makeStaticInstanceRegistry([[codexInstanceId, codex.adapter]]), + ), + ), + Layer.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: enabled })), + Layer.provide(serverConfigTestLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ), + ).pipe(Scope.provide(scope)); + const provider = yield* ProviderService.ProviderService.pipe(Effect.provide(services)); + const session = yield* provider.startSession(threadId, { + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + threadId, + runtimeMode: "full-access", + }); + codex.listSessions.mockReturnValue( + Effect.succeed([ + { + ...session, + status: completed ? "ready" : "running", + activeTurnId: completed ? undefined : turnId, + }, + ]), + ); + const pending = yield* directory.getBinding(threadId); + assert(Option.isSome(pending)); + yield* directory.upsert({ + ...pending.value, + runtimePayload: { activeTurnId: null, continueAfterServerUpdate: turnId }, + }); + const accepted = yield* provider.sendTurn({ threadId, continuation: true }); + const admitted = yield* directory.getBinding(threadId); + assert(Option.isSome(admitted)); + assert.propertyVal(admitted.value.runtimePayload, "activeTurnId", accepted.turnId); + assert.propertyVal(admitted.value.runtimePayload, "continueAfterServerUpdate", null); + if (completed) { + // Updates can mark an already-admitted turn immediately before it finishes. + yield* directory.upsert({ + ...admitted.value, + runtimePayload: { + continueAfterServerUpdate: accepted.turnId, + continueAfterServerUpdatePrepared: null, + }, + }); + } + const markers: unknown[] = []; + codex.stopAll.mockImplementation(() => + Effect.gen(function* () { + const binding = yield* directory.getBinding(threadId); + assert(Option.isSome(binding)); + markers.push(binding.value.runtimePayload); + }).pipe(Effect.orDie), + ); + yield* Scope.close(scope, Exit.void); + const binding = yield* directory.getBinding(threadId); + assert(Option.isSome(binding)); + assert.equal(codex.stopAll.mock.calls.length, 1); + assert.deepStrictEqual(binding.value.resumeCursor, session.resumeCursor); + assert.equal(binding.value.status, "stopped"); + assert.propertyVal(markers[0], "activeTurnId", completed ? null : turnId); + if (enabled && !completed) { + assert.propertyVal(markers[0], "continueAfterServerUpdate", turnId); + assert.propertyVal(binding.value.runtimePayload, "continueAfterServerUpdate", turnId); + } else if (completed) { + assert.propertyVal( + binding.value.runtimePayload, + "continueAfterServerUpdate", + accepted.turnId, + ); + assert.propertyVal( + binding.value.runtimePayload, + "continueAfterServerUpdatePrepared", + null, + ); + } else { + assert.propertyVal(markers[0], "continueAfterServerUpdate", null); + assert.propertyVal(binding.value.runtimePayload, "continueAfterServerUpdate", null); + } + }).pipe(Effect.provide(NodeServices.layer)), + ); +} + it.effect("ProviderServiceLive catches stopAll failures during shutdown", () => Effect.gen(function* () { const codex = makeFakeCodexAdapter(); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 2098cacda5eb..cf9d9b395d5b 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -223,6 +223,7 @@ function toRuntimePayloadFromSession( session: ProviderSession, extra?: { readonly modelSelection?: unknown; + readonly continueAfterServerUpdate?: TurnId; readonly lastRuntimeEvent?: string; readonly lastRuntimeEventAt?: string; }, @@ -232,6 +233,9 @@ function toRuntimePayloadFromSession( model: session.model ?? null, activeTurnId: session.activeTurnId ?? null, lastError: session.lastError ?? null, + ...(extra?.continueAfterServerUpdate !== undefined + ? { continueAfterServerUpdate: extra.continueAfterServerUpdate } + : {}), ...(extra?.modelSelection !== undefined ? { modelSelection: extra.modelSelection } : {}), ...(extra?.lastRuntimeEvent !== undefined ? { lastRuntimeEvent: extra.lastRuntimeEvent } : {}), ...(extra?.lastRuntimeEventAt !== undefined @@ -831,6 +835,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( threadId: ThreadId, extra?: { readonly modelSelection?: unknown; + readonly continueAfterServerUpdate?: TurnId; readonly lastRuntimeEvent?: string; readonly lastRuntimeEventAt?: string; }, @@ -1433,6 +1438,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( runtimePayload: { ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}), activeTurnId: turn.turnId, + // Admission and marker consumption must survive the same restart. + continueAfterServerUpdate: null, + continueAfterServerUpdatePrepared: null, lastRuntimeEvent: "provider.sendTurn", lastRuntimeEventAt: yield* nowIso, }, @@ -1733,6 +1741,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( status: "stopped", runtimePayload: { activeTurnId: null, + continueAfterServerUpdate: null, + continueAfterServerUpdatePrepared: null, }, }); yield* analytics.record("provider.session.stopped", { @@ -1942,6 +1952,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ); const runStopAll = Effect.fn("runStopAll")(function* () { + const continueAfterRestart = yield* serverSettings.getSettings.pipe( + Effect.map((settings) => settings.continueThreadsAfterServerUpdate), + Effect.orElseSucceed(() => false), + ); const properties = yield* Ref.modify(turnAnalytics, (state) => { const completed: Array>> = []; for (const [sessionKey, session] of state.sessions) { @@ -1969,6 +1983,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( yield* Effect.forEach(activeSessions, (session) => Effect.flatMap(nowIso, (lastRuntimeEventAt) => upsertSessionBinding(session, session.threadId, { + ...(continueAfterRestart && session.status === "running" && session.activeTurnId + ? { continueAfterServerUpdate: session.activeTurnId } + : {}), lastRuntimeEvent: "provider.stopAll", lastRuntimeEventAt, }), diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 39c284330d46..2c95a6163acd 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -1,6 +1,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { type OrchestrationCommand, + type OrchestrationSessionStatus, ProviderDriverKind, ProviderInstanceId, type ProviderSendTurnInput, @@ -10,15 +11,21 @@ import { import { assert, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; import { OrchestrationCommandInvariantError } from "./orchestration/Errors.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; -import { ProviderSessionDirectoryPersistenceError } from "./provider/Errors.ts"; +import { + ProviderSessionDirectoryPersistenceError, + ProviderSessionNotFoundError, +} from "./provider/Errors.ts"; import * as ProviderService from "./provider/Services/ProviderService.ts"; import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts"; +import { ServerActivation } from "./serverActivation.ts"; +import * as ServerSettings from "./serverSettings.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; const providerInstanceId = ProviderInstanceId.make("codex"); @@ -26,7 +33,7 @@ const updatedAt = "2026-08-20T12:00:00.000Z"; const makeThread = ( id: string, - status: "starting" | "running" | "ready" | "stopped" | "error", + status: OrchestrationSessionStatus, activeTurnId: TurnId | null = null, archivedAt: string | null = null, deletedAt: string | null = null, @@ -73,6 +80,7 @@ const queryWithThreads = (threads: ReadonlyArray>) const runReconciliation = (input: { readonly threads: ReadonlyArray>; + readonly continueAfterRestart?: boolean; readonly liveThreadIds?: ReadonlyArray; readonly providerService?: ProviderService.ProviderService["Service"]; readonly directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"]; @@ -97,7 +105,14 @@ const runReconciliation = (input: { subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }), - Effect.provide(NodeServices.layer), + Effect.provide( + Layer.mergeAll( + ServerSettings.layerTest({ + continueThreadsAfterServerUpdate: input.continueAfterRestart ?? false, + }), + NodeServices.layer, + ), + ), ); it.effect("marks active running sessions that have persisted resume state", () => { @@ -138,7 +153,7 @@ it.effect("marks active running sessions that have persisted resume state", () = upsert: (binding) => Effect.sync(() => upserts.push(binding)), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }), Effect.tap((marked) => Effect.sync(() => { @@ -147,169 +162,186 @@ it.effect("marks active running sessions that have persisted resume state", () = assert.deepStrictEqual(upserts[0]?.runtimePayload, { activeTurnId: "turn-mark-active", continueAfterServerUpdate: active.session.activeTurnId, + continueAfterServerUpdatePrepared: null, }); }), ), ); }); -it.effect("continues marked sessions after activation with provider-specific input", () => - Effect.gen(function* () { - const codex = makeThread( - "thread-continue-codex", - "running", - TurnId.make("turn-continue-codex"), - ); - const fallback = makeThread("thread-continue-fallback", "starting"); - const fallbackContinuationTurnId = TurnId.make("turn-continue-fallback"); - const fallbackProviderInstanceId = ProviderInstanceId.make("claudeAgent"); - const continuationSent = yield* Deferred.make(); - const continuationCleared = yield* Deferred.make(); - const sends: ProviderSendTurnInput[] = []; - const dispatched: OrchestrationCommand[] = []; - const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; - const bindings = new Map( - [codex, fallback].map((thread) => [ - thread.id, - { - threadId: thread.id, - provider: - thread.id === codex.id - ? ProviderDriverKind.make("codex") - : ProviderDriverKind.make("claudeAgent"), - providerInstanceId: - thread.id === codex.id ? providerInstanceId : fallbackProviderInstanceId, - status: "running" as const, - runtimePayload: { - continueAfterServerUpdate: - thread.id === codex.id ? codex.session.activeTurnId : fallbackContinuationTurnId, +it.effect.each(["marked update", "opt-in restart"] as const)( + "continues %s sessions after activation with provider-specific input", + (recovery) => + Effect.gen(function* () { + const codex = makeThread( + "thread-continue-codex", + "running", + TurnId.make("turn-continue-codex"), + ); + const fallbackContinuationTurnId = TurnId.make("turn-continue-fallback"); + const fallback = makeThread( + "thread-continue-fallback", + recovery === "marked update" ? "starting" : "running", + recovery === "marked update" ? null : fallbackContinuationTurnId, + ); + const fallbackProviderInstanceId = ProviderInstanceId.make("claudeAgent"); + const continuationSent = yield* Deferred.make(); + const continuationCleared = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + const dispatched: OrchestrationCommand[] = []; + const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; + const bindings = new Map( + [codex, fallback].map((thread) => [ + thread.id, + { + threadId: thread.id, + provider: + thread.id === codex.id + ? ProviderDriverKind.make("codex") + : ProviderDriverKind.make("claudeAgent"), + providerInstanceId: + thread.id === codex.id ? providerInstanceId : fallbackProviderInstanceId, + status: "running" as const, + resumeCursor: { threadId: thread.id }, + runtimePayload: + recovery === "marked update" + ? { + continueAfterServerUpdate: + thread.id === codex.id + ? codex.session.activeTurnId + : fallbackContinuationTurnId, + } + : { activeTurnId: thread.session.activeTurnId }, }, - }, - ]), - ); - const providerService: ProviderService.ProviderService["Service"] = { - ...makeProviderService(), - getCapabilities: (instanceId) => - Effect.succeed({ - sessionModelSwitch: "in-session", - ...(instanceId === providerInstanceId ? { promptlessTurnContinuation: true } : {}), - }), - sendTurn: (input) => - Effect.gen(function* () { - sends.push(input); - if (sends.length === 2) { - yield* Deferred.succeed(continuationSent, undefined); - } - return { - threadId: input.threadId, - turnId: TurnId.make(`continued-${String(input.threadId)}`), - }; - }), - }; - - yield* runReconciliation({ - threads: [codex, fallback], - providerService, - directory: { - getBinding: (threadId) => - Effect.sync(() => { - const binding = bindings.get(threadId); - return binding === undefined ? Option.none() : Option.some(binding); + ]), + ); + const providerService: ProviderService.ProviderService["Service"] = { + ...makeProviderService(), + getCapabilities: (instanceId) => + Effect.succeed({ + sessionModelSwitch: "in-session", + ...(instanceId === providerInstanceId ? { promptlessTurnContinuation: true } : {}), }), - upsert: (binding) => - Effect.sync(() => { - bindings.set(binding.threadId, binding); - upserts.push(binding); - const clearedCount = upserts.filter((candidate) => { - const payload = candidate.runtimePayload; - return ( - payload !== null && - typeof payload === "object" && - !Array.isArray(payload) && - "continueAfterServerUpdate" in payload && - payload.continueAfterServerUpdate === null - ); - }).length; - return clearedCount === 1; - }).pipe( - Effect.flatMap((firstMarkerCleared) => - firstMarkerCleared ? Deferred.succeed(continuationCleared, undefined) : Effect.void, + sendTurn: (input) => + Effect.gen(function* () { + sends.push(input); + if (sends.length === 2) { + yield* Deferred.succeed(continuationSent, undefined); + } + return { + threadId: input.threadId, + turnId: TurnId.make(`continued-${String(input.threadId)}`), + }; + }), + }; + + yield* runReconciliation({ + threads: [codex, fallback], + continueAfterRestart: recovery === "opt-in restart", + providerService, + directory: { + getBinding: (threadId) => + Effect.sync(() => { + const binding = bindings.get(threadId); + return binding === undefined ? Option.none() : Option.some(binding); + }), + upsert: (binding) => + Effect.sync(() => { + bindings.set(binding.threadId, binding); + upserts.push(binding); + const clearedCount = upserts.filter((candidate) => { + const payload = candidate.runtimePayload; + return ( + payload !== null && + typeof payload === "object" && + !Array.isArray(payload) && + "continueAfterServerUpdate" in payload && + payload.continueAfterServerUpdate === null + ); + }).length; + return clearedCount === 1; + }).pipe( + Effect.flatMap((firstMarkerCleared) => + firstMarkerCleared ? Deferred.succeed(continuationCleared, undefined) : Effect.void, + ), ), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: (command) => + Effect.sync(() => dispatched.push(command)).pipe( + Effect.as({ sequence: dispatched.length }), ), - getProvider: () => Effect.die("unused"), - listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), - }, - dispatch: (command) => - Effect.sync(() => dispatched.push(command)).pipe( - Effect.as({ sequence: dispatched.length }), - ), - }); - yield* Deferred.await(continuationSent); - yield* Deferred.await(continuationCleared); + }); + yield* Deferred.await(continuationSent); + yield* Deferred.await(continuationCleared); - assert.deepStrictEqual( - sends.toSorted((left, right) => String(left.threadId).localeCompare(String(right.threadId))), - [ - { threadId: codex.id, continuation: true, interactionMode: "default" }, - { - threadId: fallback.id, - input: "Continue where you left off.", - interactionMode: "default", - }, - ], - ); - assert.deepStrictEqual( - dispatched.map((command) => - command.type === "thread.session.set" - ? { - threadId: command.threadId, - status: command.session.status, - activeTurnId: command.session.activeTurnId, - } - : null, - ), - [ - { - threadId: codex.id, - status: "starting", - activeTurnId: null, - }, - { - threadId: fallback.id, - status: "starting", - activeTurnId: fallback.session.activeTurnId, - }, - ], - ); - for (const [thread, continuationTurnId] of [ - [codex, codex.session.activeTurnId], - [fallback, fallbackContinuationTurnId], - ] as const) { assert.deepStrictEqual( - upserts - .filter((binding) => binding.threadId === thread.id) - .map((binding) => binding.runtimePayload)[0], - { - continueAfterServerUpdate: continuationTurnId, - activeTurnId: null, - }, + sends.toSorted((left, right) => + String(left.threadId).localeCompare(String(right.threadId)), + ), + [ + { threadId: codex.id, continuation: true, interactionMode: "default" }, + { + threadId: fallback.id, + input: "Continue where you left off.", + interactionMode: "default", + }, + ], + ); + assert.deepStrictEqual( + dispatched.map((command) => + command.type === "thread.session.set" + ? { + threadId: command.threadId, + status: command.session.status, + activeTurnId: command.session.activeTurnId, + } + : null, + ), + [ + { + threadId: codex.id, + status: "starting", + activeTurnId: null, + }, + { + threadId: fallback.id, + status: "starting", + activeTurnId: null, + }, + ], ); - } - assert.equal( - upserts.some((binding) => { - const payload = binding.runtimePayload; - return ( - payload !== null && - typeof payload === "object" && - !Array.isArray(payload) && - "continueAfterServerUpdate" in payload && - payload.continueAfterServerUpdate === null + for (const [thread, continuationTurnId] of [ + [codex, codex.session.activeTurnId], + [fallback, fallbackContinuationTurnId], + ] as const) { + assert.deepStrictEqual( + upserts + .filter((binding) => binding.threadId === thread.id) + .map((binding) => binding.runtimePayload)[0], + { + continueAfterServerUpdate: continuationTurnId, + continueAfterServerUpdatePrepared: true, + activeTurnId: null, + }, ); - }), - true, - ); - }), + } + assert.equal( + upserts.some((binding) => { + const payload = binding.runtimePayload; + return ( + payload !== null && + typeof payload === "object" && + !Array.isArray(payload) && + "continueAfterServerUpdate" in payload && + payload.continueAfterServerUpdate === null + ); + }), + true, + ); + }), ); it.effect("does not continue archived or deleted marked sessions", () => { @@ -361,7 +393,7 @@ it.effect("does not continue archived or deleted marked sessions", () => { upsert: () => Effect.void, getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }, dispatch: (command) => Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })), @@ -416,7 +448,7 @@ it.effect("retries continuation preparation before settling a persistent failure upsert: () => Effect.void, getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }, dispatch: (command) => { if (command.type !== "thread.session.set") { @@ -487,7 +519,7 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio upsert: (binding) => Effect.sync(() => upserts.push(binding)), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }, dispatch: (command) => Effect.sync(() => dispatched.push(command)).pipe(Effect.as({ sequence: dispatched.length })), @@ -521,6 +553,7 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio activeTurnId: null, unrelated: binding.threadId, continueAfterServerUpdate: null, + continueAfterServerUpdatePrepared: null, } : { activeTurnId: null, unrelated: binding.threadId }, ); @@ -565,7 +598,7 @@ it.effect( upsert: () => Effect.fail(writeFailure), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }, dispatch: (command) => Effect.sync(() => dispatched.push(command)).pipe( @@ -602,7 +635,7 @@ it.effect("retries failed projections and continues after a persistent failure", upsert: () => Effect.void, getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }, dispatch: (command) => { if (command.type !== "thread.session.set") { @@ -651,7 +684,7 @@ it.effect("does not fail startup when the live provider session inventory cannot upsert: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, @@ -662,7 +695,283 @@ it.effect("does not fail startup when the live provider session inventory cannot subscribeDomainEvents: Effect.succeed(Stream.empty), latestSequence: Effect.succeed(0), }), - Effect.provide(NodeServices.layer), + Effect.provide(Layer.mergeAll(NodeServices.layer, ServerSettings.layerTest())), Effect.tap(() => Effect.sync(() => assert.equal(queried, false))), ); }); + +for (const scenario of [ + "disabled", + "stopped projection", + "finished projection", + "stopped binding", + "finished binding", + "missing cursor", + "mismatched turn", + "marked without cursor", + "marked stopped projection", + "marked superseded turn", +] as const) { + it.effect(`does not recover an interrupted session with ${scenario}`, () => { + const turnId = TurnId.make("turn-excluded-recovery"); + const thread = makeThread( + "thread-excluded-recovery", + scenario.includes("stopped projection") + ? "stopped" + : scenario === "finished projection" + ? "ready" + : scenario === "marked superseded turn" + ? "starting" + : "running", + scenario === "marked superseded turn" ? null : turnId, + ); + const dispatched: OrchestrationCommand[] = []; + const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; + return runReconciliation({ + threads: [thread], + continueAfterRestart: scenario !== "disabled", + directory: { + getBinding: () => + Effect.succeed( + Option.some({ + threadId: thread.id, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + status: scenario === "stopped binding" ? "stopped" : "running", + ...(scenario.includes("cursor") ? {} : { resumeCursor: { threadId: thread.id } }), + runtimePayload: { + activeTurnId: + scenario === "finished binding" + ? null + : scenario === "mismatched turn" || scenario === "marked superseded turn" + ? "another-turn" + : turnId, + ...(scenario.startsWith("marked") ? { continueAfterServerUpdate: turnId } : {}), + }, + }), + ), + upsert: (binding) => + Effect.sync(() => { + upserts.push(binding); + }), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: (command) => + Effect.sync(() => { + dispatched.push(command); + return { sequence: dispatched.length }; + }), + }).pipe( + Effect.tap(() => + Effect.sync(() => { + assert.deepStrictEqual( + dispatched.map( + (command) => command.type === "thread.session.set" && command.session.status, + ), + ["error"], + ); + assert.deepStrictEqual( + upserts.map((binding) => binding.status), + ["stopped"], + ); + }), + ), + ); + }); +} + +for (const preparedStatus of [ + "starting", + "ready", + "ready with failed scan", + "completed after update marking", +] as const) { + it.effect(`recovers again if startup exits with a prepared ${preparedStatus} session`, () => + Effect.gen(function* () { + const turnId = TurnId.make("turn-interrupted-startup"); + const thread = makeThread("thread-interrupted-startup", "running", turnId); + const activation = yield* Deferred.make(); + const cleared = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + let binding: ProviderSessionDirectory.ProviderRuntimeBinding = { + threadId: thread.id, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + status: "running", + resumeCursor: { threadId: thread.id }, + runtimePayload: { activeTurnId: turnId }, + }; + const input = { + threads: [thread], + continueAfterRestart: true, + providerService: { + ...makeProviderService(), + getCapabilities: () => + Effect.succeed({ + sessionModelSwitch: "in-session" as const, + promptlessTurnContinuation: true, + }), + sendTurn: (input: ProviderSendTurnInput) => + Effect.sync(() => { + sends.push(input); + return { threadId: input.threadId, turnId: TurnId.make("turn-recovered") }; + }), + }, + directory: { + getBinding: () => Effect.sync(() => Option.some(binding)), + upsert: (next: ProviderSessionDirectory.ProviderRuntimeBinding) => + Effect.gen(function* () { + binding = next; + if (binding.status !== "starting" || sends.length === 0) return; + yield* Deferred.succeed(cleared, undefined); + }), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => + preparedStatus === "ready with failed scan" + ? Effect.fail( + new ProviderSessionDirectoryPersistenceError({ + operation: "listBindings", + detail: "unreadable unrelated binding", + }), + ) + : Effect.sync(() => [{ ...binding, lastSeenAt: "2026-01-01T00:00:00.000Z" }]), + }, + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + if (command.type === "thread.session.set") { + thread.session.status = command.session.status; + thread.session.activeTurnId = command.session.activeTurnId; + } + return { sequence: 1 }; + }), + }; + + yield* runReconciliation(input).pipe( + Effect.provideService(ServerActivation, Deferred.await(activation)), + Effect.scoped, + ); + assert.deepStrictEqual(sends, []); + assert.equal(thread.session.status, "starting"); + assert.equal(thread.session.activeTurnId, null); + assert.deepStrictEqual(binding.runtimePayload, { + activeTurnId: null, + continueAfterServerUpdate: turnId, + continueAfterServerUpdatePrepared: true, + }); + + if (preparedStatus === "completed after update marking") { + thread.session.status = "ready"; + binding = { + ...binding, + status: "stopped", + runtimePayload: { + activeTurnId: null, + continueAfterServerUpdate: turnId, + continueAfterServerUpdatePrepared: null, + }, + }; + yield* runReconciliation(input); + assert.deepStrictEqual(sends, []); + assert.equal(thread.session.status, "ready"); + return; + } + thread.session.status = + preparedStatus === "ready with failed scan" ? "ready" : preparedStatus; + yield* runReconciliation(input); + yield* Deferred.await(cleared); + assert.deepStrictEqual(sends, [ + { threadId: thread.id, continuation: true, interactionMode: "default" }, + ]); + assert.deepStrictEqual(binding.runtimePayload, { + activeTurnId: null, + continueAfterServerUpdate: null, + continueAfterServerUpdatePrepared: null, + }); + }), + ); +} + +it.effect("settles failed opt-in recovery without retrying the provider turn", () => + Effect.gen(function* () { + const turnId = TurnId.make("turn-failed-recovery"); + const thread = makeThread("thread-failed-recovery", "running", turnId); + const settled = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + const dispatched: OrchestrationCommand[] = []; + const preparedPayloads: unknown[] = []; + let binding: ProviderSessionDirectory.ProviderRuntimeBinding = { + threadId: thread.id, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + status: "running", + resumeCursor: { threadId: thread.id }, + runtimePayload: { activeTurnId: turnId }, + }; + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: true, + providerService: { + ...makeProviderService(), + getCapabilities: () => + Effect.succeed({ sessionModelSwitch: "in-session", promptlessTurnContinuation: true }), + sendTurn: (input) => + Effect.gen(function* () { + sends.push(input); + preparedPayloads.push(binding.runtimePayload); + return yield* Effect.fail( + new ProviderSessionNotFoundError({ threadId: input.threadId }), + ); + }), + }, + directory: { + getBinding: () => Effect.sync(() => Option.some(binding)), + upsert: (next) => + Effect.sync(() => { + binding = next; + }), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: (command) => + Effect.gen(function* () { + dispatched.push(command); + if (command.type === "thread.session.set" && command.session.status === "error") { + yield* Deferred.succeed(settled, undefined); + } + return { sequence: dispatched.length }; + }), + }); + yield* Deferred.await(settled); + assert.equal(sends.length, 1); + assert.deepStrictEqual(preparedPayloads, [ + { + activeTurnId: null, + continueAfterServerUpdate: turnId, + continueAfterServerUpdatePrepared: true, + }, + ]); + assert.deepStrictEqual( + dispatched.map( + (command) => + command.type === "thread.session.set" && { + status: command.session.status, + activeTurnId: command.session.activeTurnId, + }, + ), + [ + { status: "starting", activeTurnId: null }, + { status: "error", activeTurnId: null }, + ], + ); + assert.equal(binding.status, "stopped"); + assert.deepStrictEqual(binding.runtimePayload, { + activeTurnId: null, + continueAfterServerUpdate: null, + continueAfterServerUpdatePrepared: null, + }); + }), +); diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 064796b2810b..a34d1bdbde91 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -394,6 +394,7 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* ( runtimePayload: { ...readRuntimePayload(binding.value.runtimePayload), [SERVER_UPDATE_CONTINUATION_KEY]: activeTurnId, + continueAfterServerUpdatePrepared: null, }, }); marked.push(thread.id); @@ -423,6 +424,7 @@ const clearContinuationMarkers = ( runtimePayload: { ...readRuntimePayload(binding.runtimePayload), [SERVER_UPDATE_CONTINUATION_KEY]: null, + continueAfterServerUpdatePrepared: null, }, }), }), @@ -443,17 +445,56 @@ export const reconcileProviderSessions = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; const providerService = yield* ProviderService.ProviderService; const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const settings = yield* ServerSettings.ServerSettingsService; + const continueAfterRestart = yield* settings.getSettings.pipe( + Effect.map((value) => value.continueThreadsAfterServerUpdate), + Effect.catch((cause) => + Effect.logWarning("could not read restart continuation preference", { cause }).pipe( + Effect.as(false), + ), + ), + ); const liveThreadIds = new Set( (yield* providerService.listSessions()).map((session) => session.threadId), ); const { threads } = yield* query.getCommandReadModel(); + // Provider startup can report ready before the continuation is submitted. + // Find those markers in one read rather than querying every idle thread. + const preparedThreadIds = new Set( + (yield* directory.listBindings().pipe( + Effect.catch((cause) => + Effect.logWarning("failed to read prepared provider continuations", { cause }).pipe( + Effect.andThen( + Effect.forEach( + threads.filter( + (thread) => thread.session?.status === "ready" && !liveThreadIds.has(thread.id), + ), + (thread) => + directory.getBinding(thread.id).pipe(Effect.orElseSucceed(() => Option.none())), + ), + ), + Effect.map((bindings) => + bindings.flatMap((binding) => (Option.isSome(binding) ? [binding.value] : [])), + ), + ), + ), + )) + .filter( + (binding) => + readServerUpdateContinuationTurnId(binding.runtimePayload) !== null && + readRuntimePayload(binding.runtimePayload).activeTurnId === null && + readRuntimePayload(binding.runtimePayload).continueAfterServerUpdatePrepared === true, + ) + .map((binding) => binding.threadId), + ); const orphanedThreads = threads.filter( (thread) => thread.session !== null && (thread.session.status === "starting" || thread.session.status === "running" || - thread.session.activeTurnId !== null) && + thread.session.activeTurnId !== null || + (thread.session.status === "ready" && preparedThreadIds.has(thread.id))) && !liveThreadIds.has(thread.id), ); @@ -479,7 +520,27 @@ export const reconcileProviderSessions = Effect.gen(function* () { : null; const continuationMarked = continuationTurnId !== null && - (session.activeTurnId === null || continuationTurnId === session.activeTurnId); + (session.activeTurnId === null || continuationTurnId === session.activeTurnId) && + Option.isSome(binding) && + (readRuntimePayload(binding.value.runtimePayload).activeTurnId == null || + readRuntimePayload(binding.value.runtimePayload).activeTurnId === continuationTurnId); + const preparedWhileReady = + session.status === "ready" && + session.activeTurnId === null && + continuationMarked && + Option.isSome(binding) && + readRuntimePayload(binding.value.runtimePayload).activeTurnId === null && + readRuntimePayload(binding.value.runtimePayload).continueAfterServerUpdatePrepared === true; + // Abrupt shutdowns cannot write an update marker. Require both durable + // records to agree on an unfinished turn before recovering one implicitly. + const interruptedByRestart = + continueAfterRestart && + session.status === "running" && + session.activeTurnId !== null && + Option.isSome(binding) && + binding.value.status === "running" && + binding.value.resumeCursor != null && + readRuntimePayload(binding.value.runtimePayload).activeTurnId === session.activeTurnId; const settleAsError = (lastError: string) => Effect.gen(function* () { yield* Effect.gen(function* () { @@ -490,7 +551,12 @@ export const reconcileProviderSessions = Effect.gen(function* () { runtimePayload: { ...readRuntimePayload(binding.value.runtimePayload), activeTurnId: null, - ...(continuationMarkerPresent ? { [SERVER_UPDATE_CONTINUATION_KEY]: null } : {}), + ...(continuationMarkerPresent || interruptedByRestart + ? { + [SERVER_UPDATE_CONTINUATION_KEY]: null, + continueAfterServerUpdatePrepared: null, + } + : {}), }, }); } @@ -535,7 +601,9 @@ export const reconcileProviderSessions = Effect.gen(function* () { if ( Option.isSome(binding) && - continuationMarked && + (continuationMarked || interruptedByRestart) && + (session.status === "running" || session.status === "starting" || preparedWhileReady) && + binding.value.resumeCursor != null && thread.archivedAt === null && thread.deletedAt === null ) { @@ -545,6 +613,9 @@ export const reconcileProviderSessions = Effect.gen(function* () { status: "starting", runtimePayload: { ...readRuntimePayload(binding.value.runtimePayload), + // Keep recovery durable if this process also exits before sending. + [SERVER_UPDATE_CONTINUATION_KEY]: session.activeTurnId ?? continuationTurnId, + continueAfterServerUpdatePrepared: true, activeTurnId: null, }, }); @@ -608,12 +679,12 @@ export const reconcileProviderSessions = Effect.gen(function* () { } return; } - yield* Effect.logWarning("failed to continue provider session after server update", { + yield* Effect.logWarning("failed to continue provider session after server restart", { threadId: thread.id, cause: continuationExit.cause, }); yield* settleAsError( - "Could not continue this thread after the server update. Send a new message to continue.", + "Could not continue this thread after the server restart. Send a new message to continue.", ).pipe(Effect.ignoreCause); }), ); diff --git a/apps/web/src/components/ServerUpdateAction.test.tsx b/apps/web/src/components/ServerUpdateAction.test.tsx index 584078c12bf4..67236361f677 100644 --- a/apps/web/src/components/ServerUpdateAction.test.tsx +++ b/apps/web/src/components/ServerUpdateAction.test.tsx @@ -15,7 +15,8 @@ vi.mock("~/hooks/useCopyToClipboard", () => ({ useCopyToClipboard: () => ({ copyToClipboard: vi.fn() }), })); vi.mock("~/hooks/useSettings", () => ({ - useClientSettings: ( + useEnvironmentSettings: ( + _environmentId: EnvironmentId, selector: (settings: { continueThreadsAfterServerUpdate: boolean }) => unknown, ) => selector({ continueThreadsAfterServerUpdate: testState.continueThreadsAfterServerUpdate }), })); diff --git a/apps/web/src/components/ServerUpdateAction.tsx b/apps/web/src/components/ServerUpdateAction.tsx index 71b974416dd3..1845ada716b6 100644 --- a/apps/web/src/components/ServerUpdateAction.tsx +++ b/apps/web/src/components/ServerUpdateAction.tsx @@ -8,7 +8,7 @@ import type { ComponentProps } from "react"; import { requestConfirmDialog } from "~/confirmDialog"; import { useCopyToClipboard } from "~/hooks/useCopyToClipboard"; -import { useClientSettings } from "~/hooks/useSettings"; +import { useEnvironmentSettings } from "~/hooks/useSettings"; import { serverEnvironment } from "~/state/server"; import { useAtomCommand } from "~/state/use-atom-command"; import { manualServerUpdateCommand } from "~/versionSkew"; @@ -99,7 +99,8 @@ export function ServerUpdateAction({ readonly size?: ComponentProps["size"]; }) { const isDesktopAppUpdate = selfUpdate === "desktop-managed"; - const continueThreadsAfterServerUpdate = useClientSettings( + const continueThreadsAfterServerUpdate = useEnvironmentSettings( + environmentId, (settings) => settings.continueThreadsAfterServerUpdate, ); const updateServer = useAtomCommand(serverEnvironment.updateServer, { diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index b0344b640393..66f7691a4a07 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -564,7 +564,7 @@ export function useSettingsRestore(onRestored?: () => void) { : []), ...(settings.continueThreadsAfterServerUpdate !== DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate - ? ["Continue threads after server updates"] + ? ["Continue threads after restarts"] : []), ...(isBackgroundActivityDirty ? ["Background activity"] : []), ...(settings.defaultThreadEnvMode !== DEFAULT_UNIFIED_SETTINGS.defaultThreadEnvMode @@ -2438,12 +2438,13 @@ export function GeneralSettingsPanel() { updateSettings({ continueThreadsAfterServerUpdate: @@ -2459,7 +2460,7 @@ export function GeneralSettingsPanel() { onCheckedChange={(checked) => updateSettings({ continueThreadsAfterServerUpdate: Boolean(checked) }) } - aria-label="Continue threads after server updates" + aria-label="Continue threads after restarts" /> } /> diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index 22e2204d8a27..7358ed8f9a17 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -218,9 +218,11 @@ export const SETTINGS_SEARCH_ITEMS = [ }, { id: "continue-threads-after-server-update", - title: "Continue threads after server updates", + title: "Continue threads after restarts", to: "/settings/general", - searchTerms: ["resume running active work restart desktop update automatically"], + searchTerms: [ + "resume running active interrupted work restart reboot machine crash desktop update automatically", + ], }, { id: "background-activity", diff --git a/docs/user/updating.md b/docs/user/updating.md index 8ccefb4e5157..14500cbe4620 100644 --- a/docs/user/updating.md +++ b/docs/user/updating.md @@ -10,9 +10,13 @@ notice. Server updates restart the connection and can interrupt active agents and terminal commands. Saved threads, settings, and project files remain. -**Settings → General → Continue threads after server updates** is off by default. -Enable it to resume supported active threads once the replacement server is -ready. Terminal commands may still be interrupted. +**Settings → General → Continue threads after restarts** is off by default. +Enable it for each environment to resume supported active threads after an +update, crash, or machine restart. T3 Code must start again on that machine; +the setting does not enable automatic startup. Terminal commands may still be +interrupted, and threads without saved provider resume state need a new message. +If you previously enabled continuation for updates, enable this environment +setting once to allow recovery without a connected client. ## Update a connected server diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index cd4b12713edf..51fbc812225c 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -255,9 +255,6 @@ export const ClientSettingsSchema = Schema.Struct({ confirmThreadArchive: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), confirmThreadDelete: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), confirmThreadUnpin: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), - continueThreadsAfterServerUpdate: Schema.Boolean.pipe( - Schema.withDecodingDefault(Effect.succeed(false)), - ), dismissedProviderUpdateNotificationKeys: Schema.Array(TrimmedNonEmptyString).pipe( Schema.withDecodingDefault(Effect.succeed([])), ), @@ -835,6 +832,10 @@ export const ServerSettings = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed(false)), ), enableProviderUpdateChecks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), + // Retain the update-era key; recovery now needs an environment-owned opt-in. + continueThreadsAfterServerUpdate: Schema.Boolean.pipe( + Schema.withDecodingDefault(Effect.succeed(false)), + ), /** * Whether agents may drive the in-app preview browser. Turning this off * withholds the MCP credential, so the `t3-code` server (and with it every @@ -1105,6 +1106,7 @@ export const ServerSettingsPatch = Schema.Struct({ // Server settings enableLegacyTokenStreaming: Schema.optionalKey(Schema.Boolean), enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean), + continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean), enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean), sidebarAutoSettleAfterDays: Schema.optionalKey(Schema.NullOr(SidebarAutoSettleAfterDays)), sidebarAutoSettleOnMerge: Schema.optionalKey(Schema.Boolean), @@ -1181,7 +1183,6 @@ export const ClientSettingsPatch = Schema.Struct({ confirmThreadArchive: Schema.optionalKey(Schema.Boolean), confirmThreadDelete: Schema.optionalKey(Schema.Boolean), confirmThreadUnpin: Schema.optionalKey(Schema.Boolean), - continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean), diffIgnoreWhitespace: Schema.optionalKey(Schema.Boolean), diffLayout: Schema.optionalKey(DiffLayout), environmentIdentificationMode: Schema.optionalKey(EnvironmentIdentificationMode), From 3610791955d70d7e8a68809a110d2bccdb2baaf6 Mon Sep 17 00:00:00 2001 From: T3 Code PR Stack <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 5 Sep 2026 00:56:55 +0200 Subject: [PATCH 06/12] test(mobile): stop requiring identical first-pass shiki token splits highlightSourceFile and highlightCodeSnippet can tokenize at different granularity before the grammar is warm. Both still have to reconstruct the source and emit colored tokens. --- .../features/review/shikiReviewHighlighter.test.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/apps/mobile/src/features/review/shikiReviewHighlighter.test.ts b/apps/mobile/src/features/review/shikiReviewHighlighter.test.ts index be723040152a..f136fc236f73 100644 --- a/apps/mobile/src/features/review/shikiReviewHighlighter.test.ts +++ b/apps/mobile/src/features/review/shikiReviewHighlighter.test.ts @@ -158,8 +158,17 @@ describe("highlightSourceFile", () => { .join(""), ).toBe(source); expect(highlighted.flat().some((token) => token.color !== null)).toBe(true); + const snippet = await highlighter.highlightCodeSnippet({ + code: source, + language: "ts", + theme: "dark", + }); expect( - await highlighter.highlightCodeSnippet({ code: source, language: "ts", theme: "dark" }), - ).toEqual(highlighted); + snippet + .flat() + .map((token) => token.content) + .join(""), + ).toBe(source); + expect(snippet.flat().some((token) => token.color !== null)).toBe(true); }); }); From dd7bc147f799f290eb58578a7f81643ecf9ad52e Mon Sep 17 00:00:00 2001 From: maria Date: Fri, 4 Sep 2026 19:01:45 -0400 Subject: [PATCH 07/12] fix(web): simplify changed files into a persistent folder tree (#9821) --- .../components/chat/ChangedFilesTree.test.tsx | 34 ++-- .../src/components/chat/ChangedFilesTree.tsx | 158 ++++-------------- .../components/chat/MessagesTimeline.test.tsx | 3 +- .../src/components/chat/MessagesTimeline.tsx | 16 +- apps/web/src/uiStateStore.test.ts | 7 +- apps/web/src/uiStateStore.ts | 5 +- 6 files changed, 57 insertions(+), 166 deletions(-) diff --git a/apps/web/src/components/chat/ChangedFilesTree.test.tsx b/apps/web/src/components/chat/ChangedFilesTree.test.tsx index 4998c40b0c55..899714ed55a1 100644 --- a/apps/web/src/components/chat/ChangedFilesTree.test.tsx +++ b/apps/web/src/components/chat/ChangedFilesTree.test.tsx @@ -10,26 +10,21 @@ describe("ChangedFilesCard", () => { {}} onToggleAllDirectories={() => {}} onOpenTurnDiff={() => {}} />, ); - expect(markup).toContain('data-changed-files-state="expanded"'); - expect(markup).toContain('aria-expanded="true"'); - expect(markup).toContain('aria-label="Collapse all folders"'); + expect(markup).toContain('data-changed-files-state="tree"'); expect(markup).toContain('aria-label="Open diff"'); expect(markup).toContain('role="group" aria-label="2 additions, 1 deletions"'); expect(markup).toContain("1 changed file"); expect(markup).not.toContain("1 changed files"); }); - it("renders a scope and representative-file preview for a large latest change", () => { + it("shows collapsed folders and root files together", () => { const markup = renderToStaticMarkup( { }, { path: "README.md", kind: "modified", additions: 3, deletions: 0 }, ]} - expanded={false} - showCompactPreview allDirectoriesExpanded={false} resolvedTheme="light" - onExpandedChange={() => {}} onToggleAllDirectories={() => {}} onOpenTurnDiff={() => {}} />, ); - expect(markup).toContain('data-changed-files-state="preview"'); + expect(markup).toContain('data-changed-files-state="tree"'); expect(markup).toContain('aria-expanded="false"'); - expect(markup).toContain("apps"); - expect(markup).toContain("2 files"); - expect(markup).toContain("packages"); - expect(markup).toContain("root"); - expect(markup).toContain("App.tsx"); - expect(markup).toContain("git.ts"); + expect(markup).toContain("apps/web/src"); + expect(markup).not.toContain("App.tsx"); + expect(markup).toContain("packages/shared/src"); + expect(markup).not.toContain("git.ts"); expect(markup).toContain("README.md"); - expect(markup).toContain("Show all 4 files"); + expect(markup).not.toContain("Show all"); expect(markup).not.toContain("App.test.tsx"); }); - it("keeps older collapsed changes to a one-line receipt", () => { + it("keeps the folder tree visible when folders are collapsed", () => { const markup = renderToStaticMarkup( {}} onToggleAllDirectories={() => {}} onOpenTurnDiff={() => {}} />, ); - expect(markup).toContain('data-changed-files-state="collapsed"'); + expect(markup).toContain('data-changed-files-state="tree"'); expect(markup).toContain("1 changed file"); + expect(markup).toContain("apps/web/src"); expect(markup).not.toContain("Show all"); expect(markup).not.toContain("App.tsx"); }); diff --git a/apps/web/src/components/chat/ChangedFilesTree.tsx b/apps/web/src/components/chat/ChangedFilesTree.tsx index 906bf4c34cb4..5ac9f09613f3 100644 --- a/apps/web/src/components/chat/ChangedFilesTree.tsx +++ b/apps/web/src/components/chat/ChangedFilesTree.tsx @@ -19,99 +19,59 @@ import { DiffStatLabel, hasNonZeroStat } from "./DiffStatLabel"; import { PierreEntryIcon } from "./PierreEntryIcon"; import { Button } from "../ui/button"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; -import { - changedFileName, - selectChangedFilePreview, - summarizeChangedFileScopes, -} from "./changedFilesPresentation"; const EMPTY_DIRECTORY_OVERRIDES: Record = {}; export const ChangedFilesCard = memo(function ChangedFilesCard(props: { turnId: TurnId; files: ReadonlyArray; - expanded: boolean; - showCompactPreview: boolean; allDirectoriesExpanded: boolean; resolvedTheme: "light" | "dark"; - onExpandedChange: (expanded: boolean) => void; onToggleAllDirectories: () => void; onOpenTurnDiff: (turnId: TurnId, filePath?: string) => void; }) { const { turnId, files, - expanded, - showCompactPreview, allDirectoriesExpanded, resolvedTheme, - onExpandedChange, onToggleAllDirectories, onOpenTurnDiff, } = props; const summaryStat = useMemo(() => summarizeTurnDiffStats(files), [files]); - const scopeSummary = useMemo(() => summarizeChangedFileScopes(files), [files]); - const previewFiles = useMemo(() => selectChangedFilePreview(files), [files]); - const compactPreviewVisible = showCompactPreview && !expanded; + const hasDirectories = files.some((file) => /[/\\]/.test(file.path)); return (
- -
- {expanded ? ( + {hasNonZeroStat(summaryStat) && ( + + )} +
+
+ {hasDirectories && ( - ) : null} + )} onOpenTurnDiff(turnId, files[0]?.path)} /> @@ -150,61 +110,14 @@ export const ChangedFilesCard = memo(function ChangedFilesCard(props: {
- {expanded ? ( - - ) : compactPreviewVisible ? ( -
-

- {scopeSummary.map((scope, index) => ( - - {index > 0 ? : null} - {scope.label} - - {scope.fileCount} file{scope.fileCount === 1 ? "" : "s"} - - - ))} -

-
- {previewFiles.map((file) => ( - - onOpenTurnDiff(turnId, file.path)} - /> - } - > - - {changedFileName(file.path)} - - {file.path} - - ))} - -
-
- ) : null} +
); }); @@ -261,7 +174,8 @@ export const ChangedFilesTree = memo(function ChangedFilesTree(props: { {isExpanded && ( -
- {node.children.map((childNode) => renderTreeNode(childNode, depth + 1))} -
+
{node.children.map((childNode) => renderTreeNode(childNode, depth + 1))}
)}
); @@ -299,7 +211,7 @@ export const ChangedFilesTree = memo(function ChangedFilesTree(props: {