From 1cbbaa05d8c9b15cf534c34a4c0f0973db1f2ba0 Mon Sep 17 00:00:00 2001 From: Patrick Roza Date: Sun, 26 Jul 2026 08:16:15 +0200 Subject: [PATCH 1/5] fix(client-runtime): re-read network status when the app resumes (upstream #4528) Imported from https://github.com/pingdotgg/t3code/pull/4528 --- .../src/connection/connectivity.test.ts | 298 ++++++++++++++++++ .../src/connection/connectivity.ts | 84 ++++- .../src/connection/registry.test.ts | 53 +++- .../client-runtime/src/connection/registry.ts | 12 +- .../src/connection/supervisor.test.ts | 54 +++- .../src/connection/supervisor.ts | 29 +- .../client-runtime/src/relay/discovery.ts | 16 +- 7 files changed, 516 insertions(+), 30 deletions(-) create mode 100644 packages/client-runtime/src/connection/connectivity.test.ts diff --git a/packages/client-runtime/src/connection/connectivity.test.ts b/packages/client-runtime/src/connection/connectivity.test.ts new file mode 100644 index 00000000000..7e607ee30d0 --- /dev/null +++ b/packages/client-runtime/src/connection/connectivity.test.ts @@ -0,0 +1,298 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import * as Connectivity from "./connectivity.ts"; +import type { NetworkStatus } from "./model.ts"; +import * as ConnectionWakeups from "./wakeups.ts"; + +const makeHarness = Effect.fn("TestConnectivityHarness.make")(function* (options?: { + readonly status?: Effect.Effect; + readonly initialStatus?: NetworkStatus; +}) { + const liveStatus = yield* Ref.make(options?.initialStatus ?? "online"); + const reported = yield* SubscriptionRef.make<{ + readonly sequence: number; + readonly status: NetworkStatus; + }>({ sequence: 0, status: options?.initialStatus ?? "online" }); + const wakeups = yield* SubscriptionRef.make(0); + const applied = yield* Ref.make>([]); + + const connectivity = Connectivity.Connectivity.of({ + status: options?.status ?? Ref.get(liveStatus), + changes: SubscriptionRef.changes(reported).pipe( + Stream.drop(1), + Stream.map((event) => event.status), + ), + }); + + const wakeupService = ConnectionWakeups.ConnectionWakeups.of({ + changes: SubscriptionRef.changes(wakeups).pipe( + Stream.drop(1), + Stream.map(() => "application-active" as const), + ), + }); + + return { + connectivity, + wakeups: wakeupService, + applied, + setLiveStatus: (status: NetworkStatus) => Ref.set(liveStatus, status), + // Emits a listener event, as a platform connectivity listener would. + report: (status: NetworkStatus) => + Ref.set(liveStatus, status).pipe( + Effect.andThen( + SubscriptionRef.update(reported, (event) => ({ + sequence: event.sequence + 1, + status, + })), + ), + ), + resume: SubscriptionRef.update(wakeups, (count) => count + 1), + apply: (status: NetworkStatus) => Ref.update(applied, (statuses) => [...statuses, status]), + }; +}); + +// The forked listeners subscribe after `followNetworkStatus` returns, so repeat +// the trigger until its effect is observed rather than racing the first one. +const untilApplied = Effect.fn("TestConnectivityHarness.untilApplied")(function* ( + trigger: Effect.Effect, + applied: Ref.Ref>, + expected: number, +) { + for (let attempt = 0; attempt < 100; attempt += 1) { + yield* trigger; + yield* Effect.yieldNow; + if ((yield* Ref.get(applied)).length >= expected) { + return yield* Ref.get(applied); + } + } + return yield* Effect.die(new Error("The expected network status was never applied.")); +}); + +describe("followNetworkStatus", () => { + it.effect("applies statuses reported by the platform listener", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* Connectivity.followNetworkStatus({ + connectivity: harness.connectivity, + wakeups: harness.wakeups, + apply: harness.apply, + }); + + const applied = yield* untilApplied(harness.report("offline"), harness.applied, 1); + expect(applied).toEqual(["offline"]); + }), + ), + ); + + it.effect("applies a resumed read when the listener missed a transition", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ initialStatus: "offline" }); + yield* Connectivity.followNetworkStatus({ + connectivity: harness.connectivity, + wakeups: harness.wakeups, + apply: harness.apply, + }); + + // The device came back online while suspended and the listener never + // reported the transition. + yield* harness.setLiveStatus("online"); + const applied = yield* untilApplied(harness.resume, harness.applied, 1); + expect(applied).toEqual(["online"]); + }), + ), + ); + + it.effect("discards a resumed read that a repeated report raced", () => + Effect.scoped( + Effect.gen(function* () { + const readStarted = yield* Deferred.make(); + const releaseRead = yield* Deferred.make(); + const harness = yield* makeHarness({ + initialStatus: "online", + // The resume samples a brief opposite state. + status: Deferred.succeed(readStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseRead)), + Effect.as("offline" as const), + ), + }); + yield* Connectivity.followNetworkStatus({ + connectivity: harness.connectivity, + wakeups: harness.wakeups, + apply: harness.apply, + }); + + // Apply "online" first so the listener's later repeat of it is genuinely + // redundant rather than a new status. + yield* untilApplied(harness.report("online"), harness.applied, 1); + yield* harness.resume.pipe( + Effect.andThen(Effect.yieldNow), + Effect.repeat({ until: () => Deferred.isDone(readStarted) }), + ); + + // The repeat carries no new status, but it proves the listener spoke + // after this read began, so the read is stale even though the status it + // returns differs. Applying it would strand consumers on "offline" while + // the platform reports "online" — the very failure this helper exists to + // prevent. + yield* harness.report("online"); + yield* Effect.yieldNow; + yield* Deferred.succeed(releaseRead, undefined); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + + expect(yield* Ref.get(harness.applied)).toEqual(["online"]); + }), + ), + ); + + it.effect("still deduplicates a repeated report before it reaches consumers", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ initialStatus: "online" }); + yield* Connectivity.followNetworkStatus({ + connectivity: harness.connectivity, + wakeups: harness.wakeups, + apply: harness.apply, + }); + + yield* untilApplied(harness.report("offline"), harness.applied, 1); + // Counting the repeat for staleness must not turn it into a redundant + // consumer update. + yield* harness.report("offline"); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + expect(yield* Ref.get(harness.applied)).toEqual(["offline"]); + + // A genuine transition after the repeat still lands. + yield* untilApplied(harness.report("online"), harness.applied, 2); + expect(yield* Ref.get(harness.applied)).toEqual(["offline", "online"]); + }), + ), + ); + + it.effect("does not start a second status read while one is in flight", () => + Effect.scoped( + Effect.gen(function* () { + const firstRead = yield* Deferred.make(); + const readCount = yield* Ref.make(0); + const harness = yield* makeHarness({ + initialStatus: "unknown", + status: Ref.updateAndGet(readCount, (count) => count + 1).pipe( + Effect.andThen(Deferred.await(firstRead)), + ), + }); + yield* Connectivity.followNetworkStatus({ + connectivity: harness.connectivity, + wakeups: harness.wakeups, + apply: harness.apply, + }); + + yield* harness.resume.pipe( + Effect.andThen(Effect.yieldNow), + Effect.repeat({ + until: () => Ref.get(readCount).pipe(Effect.map((count) => count >= 1)), + }), + ); + + // Resumes are consumed sequentially, so further resumes cannot start a + // read that races the one already in flight. Overlapping snapshots + // therefore cannot apply out of order. + yield* harness.resume; + yield* harness.resume; + yield* Effect.yieldNow; + expect(yield* Ref.get(readCount)).toBe(1); + + yield* Deferred.succeed(firstRead, "online"); + yield* Effect.yieldNow; + expect(yield* Ref.get(harness.applied)).toEqual(["online"]); + }), + ), + ); + + it.effect("discards a resumed read that a newer reported change superseded", () => + Effect.scoped( + Effect.gen(function* () { + const readStarted = yield* Deferred.make(); + const releaseRead = yield* Deferred.make(); + const harness = yield* makeHarness({ + initialStatus: "offline", + status: Deferred.succeed(readStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseRead)), + Effect.as("online" as const), + ), + }); + yield* Connectivity.followNetworkStatus({ + connectivity: harness.connectivity, + wakeups: harness.wakeups, + apply: harness.apply, + }); + + // Resume, and hold the status read open so the listener can report a + // newer transition while that read is still in flight. + yield* harness.resume.pipe( + Effect.andThen(Effect.yieldNow), + Effect.repeat({ until: () => Deferred.isDone(readStarted) }), + ); + yield* untilApplied(harness.report("offline"), harness.applied, 1); + yield* Deferred.succeed(releaseRead, undefined); + yield* Effect.yieldNow; + + // The stale "online" snapshot must not land on top of the newer event. + expect(yield* Ref.get(harness.applied)).toEqual(["offline"]); + }), + ), + ); + + it.effect("keeps a reported change from interleaving with a resumed apply", () => + Effect.scoped( + Effect.gen(function* () { + const applyStarted = yield* Deferred.make(); + const releaseApply = yield* Deferred.make(); + const harness = yield* makeHarness({ initialStatus: "offline" }); + // Holds the resumed apply open once it begins, so a reported change has + // a window to interleave between the guard and its apply. + const gatedApply = (status: NetworkStatus) => + status === "online" + ? Deferred.succeed(applyStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseApply)), + Effect.andThen(harness.apply(status)), + ) + : harness.apply(status); + + yield* Connectivity.followNetworkStatus({ + connectivity: harness.connectivity, + wakeups: harness.wakeups, + apply: gatedApply, + }); + + yield* harness.setLiveStatus("online"); + yield* harness.resume.pipe( + Effect.andThen(Effect.yieldNow), + Effect.repeat({ until: () => Deferred.isDone(applyStarted) }), + ); + + // The listener reports a newer transition mid-apply. + yield* harness.report("offline"); + yield* Effect.yieldNow; + yield* Deferred.succeed(releaseApply, undefined); + yield* Effect.yieldNow.pipe( + Effect.repeat({ + until: () => + Ref.get(harness.applied).pipe(Effect.map((statuses) => statuses.length >= 2)), + }), + ); + + // The reported change has to land last; applying it before the resumed + // status finished would leave the stale "online" as the final word. + expect(yield* Ref.get(harness.applied)).toEqual(["online", "offline"]); + }), + ), + ); +}); diff --git a/packages/client-runtime/src/connection/connectivity.ts b/packages/client-runtime/src/connection/connectivity.ts index 6b40680ce35..f3ce89bf553 100644 --- a/packages/client-runtime/src/connection/connectivity.ts +++ b/packages/client-runtime/src/connection/connectivity.ts @@ -1,9 +1,13 @@ import * as Context from "effect/Context"; -import type * as Effect from "effect/Effect"; +import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; -import type * as Stream from "effect/Stream"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; import type { NetworkStatus } from "./model.ts"; +import * as ConnectionWakeups from "./wakeups.ts"; export class Connectivity extends Context.Service< Connectivity, @@ -17,3 +21,79 @@ export const make = (service: Connectivity["Service"]) => Connectivity.of(servic export const layer = (service: Connectivity["Service"]) => Layer.succeed(Connectivity, make(service)); + +/** + * Applies every reported connectivity change, plus a freshly read status each + * time the application resumes. + * + * Platform listeners drop transitions while an app is suspended, so a consumer + * that follows `changes` alone keeps a stale status until the next real + * transition — which left mobile stranded on "offline" until it was restarted. + * Reading the status is asynchronous, so a read that started before a newer + * reported change is discarded instead of applied over it. + */ +export const followNetworkStatus = Effect.fnUntraced(function* (options: { + readonly connectivity: Connectivity["Service"]; + readonly wakeups: ConnectionWakeups.ConnectionWakeups["Service"]; + readonly apply: (status: NetworkStatus) => Effect.Effect; +}) { + // Counts every report the listener delivers, including one that repeats the + // status already in effect. Such a repeat carries no new status, but it does + // prove the listener spoke more recently than a resume read still in flight, + // so it has to invalidate that read: otherwise a snapshot taken during a brief + // opposite state would land afterwards and overwrite the real status. + const reportCount = yield* Ref.make(0); + // Tracks what was last handed to `options.apply` purely to keep repeats from + // reaching consumers. Deduplication is deliberately kept separate from the + // staleness guard above. + const appliedStatus = yield* Ref.make>(Option.none()); + // Counting a report and applying it has to be indivisible with respect to the + // resume branch's guard. Otherwise a change landing between that guard and its + // apply would be overwritten by the older read. + const applyLock = yield* Semaphore.make(1); + + const applyStatus = Effect.fnUntraced(function* (status: NetworkStatus) { + const changed = yield* Ref.modify(appliedStatus, (current) => + Option.isSome(current) && current.value === status + ? ([false, current] as const) + : ([true, Option.some(status)] as const), + ); + if (changed) { + yield* options.apply(status); + } + }); + + yield* options.connectivity.changes.pipe( + Stream.runForEach((status) => + applyLock.withPermits(1)( + Ref.update(reportCount, (count) => count + 1).pipe(Effect.andThen(applyStatus(status))), + ), + ), + // Subscribe before returning so a transition reported while this is still + // being set up is not dropped. + Effect.forkScoped({ startImmediately: true }), + ); + + yield* options.wakeups.changes.pipe( + Stream.runForEach((reason) => + reason === "application-active" + ? Effect.gen(function* () { + // `runForEach` is sequential, so resume reads cannot overlap: a + // second resume does not start a read until this one has applied. + const startedAt = yield* Ref.get(reportCount); + const status = yield* options.connectivity.status; + yield* applyLock.withPermits(1)( + Effect.gen(function* () { + // Re-read under the permit: any report that arrived while the + // read was in flight is newer, so this result is stale. + if ((yield* Ref.get(reportCount)) === startedAt) { + yield* applyStatus(status); + } + }), + ); + }) + : Effect.void, + ), + Effect.forkScoped({ startImmediately: true }), + ); +}); diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index c75cc2fad4b..3e40927aebe 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -267,8 +267,12 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( Ref.update(ownedDataClears, (environmentIds) => [...environmentIds, environmentId]), }); const networkStatus = yield* SubscriptionRef.make<"unknown" | "offline" | "online">("online"); + // Status reads come from a separate ref so a test can simulate a platform + // listener that missed a transition while the app was suspended. + const liveNetworkStatus = yield* Ref.make<"unknown" | "offline" | "online">("online"); + const wakeups = yield* SubscriptionRef.make(0); const connectivity = Connectivity.Connectivity.of({ - status: SubscriptionRef.get(networkStatus), + status: Ref.get(liveNetworkStatus), changes: SubscriptionRef.changes(networkStatus), }); const profileStore = ConnectionProfileStore.ConnectionProfileStore.of({ @@ -375,7 +379,12 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( Layer.succeed(Connectivity.Connectivity, connectivity), Layer.succeed( ConnectionWakeups.ConnectionWakeups, - ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.never }), + ConnectionWakeups.ConnectionWakeups.of({ + changes: SubscriptionRef.changes(wakeups).pipe( + Stream.drop(1), + Stream.map(() => "application-active" as const), + ), + }), ), Layer.succeed(ConnectionDriver.ConnectionDriver, driver), cacheLayer, @@ -398,6 +407,11 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( storedRemoteTokens, disconnectedSshTargets, networkStatus, + // Changes the network the device is actually on without emitting a change + // event, mimicking a listener that was suspended while backgrounded. + setLiveNetworkStatus: (status: "unknown" | "offline" | "online") => + Ref.set(liveNetworkStatus, status), + resume: SubscriptionRef.update(wakeups, (count) => count + 1), }; }); @@ -456,6 +470,41 @@ describe("EnvironmentRegistry", () => { }), ); + it.effect("recovers the network status a suspended listener never reported", () => + Effect.gen(function* () { + const harness = yield* makeHarness([]); + + yield* Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const offline = yield* Effect.forkChild( + SubscriptionRef.changes(registry.networkStatus).pipe( + Stream.filter((status) => status === "offline"), + Stream.runHead, + ), + ); + yield* SubscriptionRef.set(harness.networkStatus, "offline"); + yield* Fiber.join(offline); + + // The device came back online while backgrounded and the listener never + // reported the transition, so only a fresh read can observe it. + yield* harness.setLiveNetworkStatus("online"); + // The forked listener subscribes after `start`, so repeat the resume + // until it is observed rather than racing the first one. + yield* harness.resume.pipe( + Effect.andThen(Effect.yieldNow), + Effect.repeat({ + while: () => + SubscriptionRef.get(registry.networkStatus).pipe( + Effect.map((status) => status !== "online"), + ), + }), + ); + + expect(yield* SubscriptionRef.get(registry.networkStatus)).toBe("online"); + }).pipe(Effect.provide(harness.layer), Effect.scoped); + }), + ); + it.effect("starts persisted environments independently", () => Effect.gen(function* () { const bothLoadsStarted = yield* Deferred.make(); diff --git a/packages/client-runtime/src/connection/registry.ts b/packages/client-runtime/src/connection/registry.ts index a3a36272f32..2bc5076a297 100644 --- a/packages/client-runtime/src/connection/registry.ts +++ b/packages/client-runtime/src/connection/registry.ts @@ -652,10 +652,14 @@ export const make = Effect.gen(function* () { ), ), ); - yield* connectivity.changes.pipe( - Stream.runForEach((status) => SubscriptionRef.set(networkStatus, status)), - Effect.forkScoped, - ); + // `networkStatus` feeds the connection UI and readiness checks, so it has to + // recover from a transition dropped while the app was suspended just like the + // supervisors do. Otherwise a reconnected environment still reads as offline. + yield* Connectivity.followNetworkStatus({ + connectivity, + wakeups, + apply: (status) => SubscriptionRef.set(networkStatus, status), + }); return EnvironmentRegistry.of({ entries, diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index f3901e42251..95df5de21a6 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -116,9 +116,13 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: readonly ready?: (attempt: number) => Effect.Effect; readonly probe?: (attempt: number) => Effect.Effect; }) { - const networkStatus = yield* SubscriptionRef.make( + // `reportedNetworkStatus` drives the change stream while `liveNetworkStatus` + // answers status reads, so a test can simulate a platform listener that missed + // a transition while the app was suspended. + const reportedNetworkStatus = yield* SubscriptionRef.make( options?.networkStatus ?? "online", ); + const liveNetworkStatus = yield* Ref.make(options?.networkStatus ?? "online"); const prepareCount = yield* Ref.make(0); const sessionCount = yield* Ref.make(0); const releaseCount = yield* Ref.make(0); @@ -134,8 +138,8 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: >([]); const connectivity = Connectivity.Connectivity.of({ - status: SubscriptionRef.get(networkStatus), - changes: SubscriptionRef.changes(networkStatus), + status: Ref.get(liveNetworkStatus), + changes: SubscriptionRef.changes(reportedNetworkStatus), }); const prepare = Effect.fn("TestConnectionDriver.prepare")(function* (target: ConnectionTarget) { @@ -197,7 +201,13 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: prepareCount, sessionCount, releaseCount, - setNetworkStatus: (status: NetworkStatus) => SubscriptionRef.set(networkStatus, status), + setNetworkStatus: (status: NetworkStatus) => + Ref.set(liveNetworkStatus, status).pipe( + Effect.andThen(SubscriptionRef.set(reportedNetworkStatus, status)), + ), + // Changes the network the device is actually on without emitting a change + // event, mimicking a listener that was suspended while backgrounded. + setNetworkStatusWithoutNotifying: (status: NetworkStatus) => Ref.set(liveNetworkStatus, status), wake: (reason: "application-active" | "credentials-changed") => SubscriptionRef.update(wakeups, (event) => ({ sequence: event.sequence + 1, @@ -311,6 +321,42 @@ describe("EnvironmentSupervisor", () => { }), ); + it.effect("recovers from a network change the platform dropped while suspended", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ networkStatus: "offline" }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "offline"); + + // The device regained connectivity while backgrounded, but the listener + // was suspended and never reported the transition. + yield* harness.setNetworkStatusWithoutNotifying("online"); + + // The supervisor reaches its offline state before the forked wakeup + // listener subscribes, so repeat the resume until it is observed. + yield* harness.wake("application-active").pipe( + Effect.andThen(Effect.yieldNow), + Effect.repeat({ + while: () => + SubscriptionRef.get(supervisor.state).pipe( + Effect.map((state) => state.phase === "offline"), + ), + }), + ); + + const ready = yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(ready).toMatchObject({ + desired: true, + network: "online", + phase: "connected", + lastFailure: null, + }); + expect(yield* Ref.get(harness.prepareCount)).toBe(1); + }), + ); + it.effect("retries forever with exponential backoff capped at sixteen seconds", () => Effect.gen(function* () { const harness = yield* makeHarness({ diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index d9efcd4263a..5d0c63358c3 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -667,18 +667,23 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } }); - yield* connectivity.changes.pipe( - Stream.runForEach((network) => - Ref.modify(intent, (current) => - current.network === network ? [false, current] : ([true, { ...current, network }] as const), - ).pipe( - Effect.flatMap((changed) => - changed ? signal({ _tag: "NetworkChanged", network }) : Effect.void, - ), - ), - ), - Effect.forkScoped, - ); + const applyNetworkStatus = Effect.fnUntraced(function* (network: NetworkStatus) { + const changed = yield* Ref.modify(intent, (current) => + current.network === network ? [false, current] : ([true, { ...current, network }] as const), + ); + if (changed) { + yield* signal({ _tag: "NetworkChanged", network }); + } + }); + + // The offline branch of `run` only waits for signals and re-reads the same + // cached network value, so a transition dropped while the app was suspended + // would otherwise strand this supervisor until the app restarted. + yield* Connectivity.followNetworkStatus({ + connectivity, + wakeups, + apply: applyNetworkStatus, + }); yield* wakeups.changes.pipe( Stream.runForEach((reason) => signal({ _tag: "Wakeup", reason })), Effect.forkScoped, diff --git a/packages/client-runtime/src/relay/discovery.ts b/packages/client-runtime/src/relay/discovery.ts index 855bb2654ed..4c58121742f 100644 --- a/packages/client-runtime/src/relay/discovery.ts +++ b/packages/client-runtime/src/relay/discovery.ts @@ -309,9 +309,15 @@ export const make = Effect.fn("RelayEnvironmentDiscovery.make")(function* () { ), ); - yield* connectivity.changes.pipe( - Stream.changes, - Stream.runForEach((networkStatus) => + // Follow resumes too: a transition dropped while the app was suspended would + // otherwise leave `offline` set, so the environment list stayed stale until + // another network event or a manual refresh. `followNetworkStatus` only + // reports actual changes, so a resume that reads the current status does not + // trigger a refresh. + yield* Connectivity.followNetworkStatus({ + connectivity, + wakeups, + apply: (networkStatus) => networkStatus === "offline" ? SubscriptionRef.update(state, (current) => ({ ...current, @@ -321,9 +327,7 @@ export const make = Effect.fn("RelayEnvironmentDiscovery.make")(function* () { : Ref.get(hasRefreshed).pipe( Effect.flatMap((shouldRefresh) => (shouldRefresh ? refresh : Effect.void)), ), - ), - Effect.forkScoped, - ); + }); yield* wakeups.changes.pipe( Stream.runForEach((reason) => reason === "credentials-changed" From 57fe51ec7990f19965c76fcb0fa38264a1a36ef0 Mon Sep 17 00:00:00 2001 From: Patrick Roza Date: Sun, 26 Jul 2026 08:16:27 +0200 Subject: [PATCH 2/5] perf(server): compress and cache packaged web assets (upstream #4516) Imported from https://github.com/pingdotgg/t3code/pull/4516 --- apps/server/src/http.ts | 86 +++++++- apps/server/src/staticAssetDelivery.test.ts | 223 ++++++++++++++++++++ apps/server/src/staticAssetDelivery.ts | 185 ++++++++++++++++ 3 files changed, 490 insertions(+), 4 deletions(-) create mode 100644 apps/server/src/staticAssetDelivery.test.ts create mode 100644 apps/server/src/staticAssetDelivery.ts diff --git a/apps/server/src/http.ts b/apps/server/src/http.ts index b9bb40f372d..69de791be35 100644 --- a/apps/server/src/http.ts +++ b/apps/server/src/http.ts @@ -37,6 +37,13 @@ import { } from "./auth/http.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import { browserApiCorsAllowedHeaders, browserApiCorsAllowedMethods } from "./httpCors.ts"; +import { + contentCacheKey, + isCompressibleContentType, + makeStaticCompressionCache, + negotiateStaticEncoding, + resolveStaticCacheControl, +} from "./staticAssetDelivery.ts"; const OTLP_TRACES_PROXY_PATH = "/api/observability/v1/traces"; const LOOPBACK_HOSTNAMES = new Set(["127.0.0.1", "::1", "localhost"]); @@ -204,6 +211,8 @@ export const assetRouteLayer = HttpRouter.add( }), ); +const staticCompressionCache = makeStaticCompressionCache(); + export const staticAndDevRouteLayer = HttpRouter.add( "GET", "*", @@ -273,9 +282,18 @@ export const staticAndDevRouteLayer = HttpRouter.add( if (!indexData) { return HttpServerResponse.text("Not Found", { status: 404 }); } - return HttpServerResponse.uint8Array(indexData, { - status: 200, + return yield* respondWithStaticFile({ + data: indexData, contentType: "text/html; charset=utf-8", + // The SPA fallback always revalidates so a new build is picked up. + cacheControl: resolveStaticCacheControl("index.html"), + // Every deep link lands here, so the document is worth compressing. + // This is the one path whose file keeps a stable name across builds, + // so it is keyed by content: metadata read separately from the bytes + // could describe a different build than the one being served. The + // document is small enough for hashing it to be cheap. + cacheKey: contentCacheKey(indexData), + acceptEncoding: request.headers["accept-encoding"], }); } @@ -285,9 +303,69 @@ export const staticAndDevRouteLayer = HttpRouter.add( return HttpServerResponse.text("Internal Server Error", { status: 500 }); } - return HttpServerResponse.uint8Array(data, { - status: 200, + return yield* respondWithStaticFile({ + data, contentType, + cacheControl: resolveStaticCacheControl(staticRelativePath), + cacheKey: staticCacheKey(filePath, fileInfo), + acceptEncoding: request.headers["accept-encoding"], }); }), ); + +/** + * Identifies a build of a file for the compression cache, without hashing + * payloads that can run to megabytes. The stat is taken before the bytes are + * read, so a rebuild landing between the two can only orphan an entry under + * the superseded key, never publish those bytes under the newer one. Files + * whose contents change without their name are keyed by content instead; the + * rest carry a content hash in their filename already. + */ +function staticCacheKey(filePath: string, info: FileSystem.File.Info): string { + const mtimeMs = info.mtime.pipe( + Option.map((mtime) => mtime.getTime()), + Option.getOrElse(() => 0), + ); + return `${filePath} ${mtimeMs} ${info.size}`; +} + +const respondWithStaticFile = Effect.fn("staticAndDevRoute.respond")(function* (input: { + readonly data: Uint8Array; + readonly contentType: string; + readonly cacheControl: string; + /** Null for the SPA fallback, whose bytes are re-read on every request. */ + readonly cacheKey: string | null; + readonly acceptEncoding: string | undefined; +}) { + const headers: Record = { + "Cache-Control": input.cacheControl, + }; + + const encoding = isCompressibleContentType(input.contentType) + ? negotiateStaticEncoding(input.acceptEncoding) + : null; + if (isCompressibleContentType(input.contentType)) { + // Announce negotiation even when this client took the identity encoding, + // so shared caches do not hand a compressed body to a client that + // cannot read it. + headers["Vary"] = "Accept-Encoding"; + } + + const cacheKey = input.cacheKey; + const compressed = + encoding && cacheKey !== null + ? yield* Effect.tryPromise(() => + staticCompressionCache.get({ cacheKey, data: input.data, encoding }), + ).pipe(Effect.orElseSucceed(() => null)) + : null; + + if (compressed && encoding) { + headers["Content-Encoding"] = encoding; + } + + return HttpServerResponse.uint8Array(compressed ?? input.data, { + status: 200, + contentType: input.contentType, + headers, + }); +}); diff --git a/apps/server/src/staticAssetDelivery.test.ts b/apps/server/src/staticAssetDelivery.test.ts new file mode 100644 index 00000000000..9f3eff34e33 --- /dev/null +++ b/apps/server/src/staticAssetDelivery.test.ts @@ -0,0 +1,223 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { + contentCacheKey, + isCompressibleContentType, + makeStaticCompressionCache, + negotiateStaticEncoding, + resolveStaticCacheControl, +} from "./staticAssetDelivery.ts"; + +describe("contentCacheKey", () => { + const encode = (value: string) => new TextEncoder().encode(value); + + it("keys identical content the same way", () => { + expect(contentCacheKey(encode("a"))).toBe( + contentCacheKey(encode("a")), + ); + }); + + it("separates content of the same length", () => { + // A rebuild can reuse a filename, a size, and even a timestamp, so the + // key has to come from the bytes themselves. + expect(contentCacheKey(encode("a"))).not.toBe( + contentCacheKey(encode("b")), + ); + }); +}); + +describe("resolveStaticCacheControl", () => { + it("marks content-hashed bundle assets immutable", () => { + expect(resolveStaticCacheControl("assets/index-DxV9k2Qp.js")).toBe( + "public, max-age=31536000, immutable", + ); + expect(resolveStaticCacheControl("assets/style-a1b2c3d4.css")).toBe( + "public, max-age=31536000, immutable", + ); + }); + + it("revalidates entry documents and unhashed files", () => { + expect(resolveStaticCacheControl("index.html")).toBe("no-cache"); + expect(resolveStaticCacheControl("/index.html")).toBe("no-cache"); + // No hash means a rebuild reuses the name, so it must not be pinned. + expect(resolveStaticCacheControl("assets/logo.svg")).toBe("no-cache"); + expect(resolveStaticCacheControl("favicon.ico")).toBe("no-cache"); + }); +}); + +describe("negotiateStaticEncoding", () => { + it("prefers brotli when the client accepts both", () => { + expect(negotiateStaticEncoding("gzip, deflate, br")).toBe("br"); + }); + + it("falls back to gzip when brotli is absent", () => { + expect(negotiateStaticEncoding("gzip, deflate")).toBe("gzip"); + }); + + it("returns null when nothing usable is offered", () => { + expect(negotiateStaticEncoding(undefined)).toBeNull(); + expect(negotiateStaticEncoding("")).toBeNull(); + expect(negotiateStaticEncoding("deflate")).toBeNull(); + }); + + it("honors explicit refusals expressed as q=0", () => { + expect(negotiateStaticEncoding("br;q=0, gzip")).toBe("gzip"); + expect(negotiateStaticEncoding("gzip;q=0, br;q=0")).toBeNull(); + }); + + it("accepts a wildcard offer", () => { + expect(negotiateStaticEncoding("*")).toBe("br"); + }); +}); + +describe("isCompressibleContentType", () => { + it("compresses text and structured payloads", () => { + expect(isCompressibleContentType("text/html; charset=utf-8")).toBe(true); + expect(isCompressibleContentType("application/javascript")).toBe(true); + expect(isCompressibleContentType("image/svg+xml")).toBe(true); + }); + + it("leaves already-compressed binaries alone", () => { + expect(isCompressibleContentType("image/png")).toBe(false); + expect(isCompressibleContentType("font/woff2")).toBe(false); + expect(isCompressibleContentType("application/octet-stream")).toBe(false); + }); +}); + +describe("makeStaticCompressionCache", () => { + const bundle = new TextEncoder().encode("export const value = 1;\n".repeat(500)); + + it("compresses a payload once and reuses the result", async () => { + let compressCalls = 0; + const cache = makeStaticCompressionCache(async (data) => { + compressCalls += 1; + return data.subarray(0, 64); + }); + + const first = await cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "gzip" }); + const second = await cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "gzip" }); + + expect(first).not.toBeNull(); + expect(second).toBe(first); + expect(compressCalls).toBe(1); + }); + + it("compresses once for a burst of concurrent requests", async () => { + let compressCalls = 0; + let release = () => {}; + const started = new Promise((resolve) => { + release = resolve; + }); + const cache = makeStaticCompressionCache(async (data) => { + compressCalls += 1; + await started; + return data.subarray(0, 64); + }); + + const requests = [ + cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "br" }), + cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "br" }), + cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "br" }), + ]; + release(); + const results = await Promise.all(requests); + + expect(compressCalls).toBe(1); + expect(results[1]).toBe(results[0]); + expect(results[2]).toBe(results[0]); + }); + + it("retries after a failed compression instead of caching the failure", async () => { + let compressCalls = 0; + const cache = makeStaticCompressionCache(async (data) => { + compressCalls += 1; + if (compressCalls === 1) throw new Error("zlib buffer error"); + return data.subarray(0, 64); + }); + + await expect( + cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "gzip" }), + ).rejects.toThrow("zlib buffer error"); + + const retried = await cache.get({ + cacheKey: "bundle.js 1 100", + data: bundle, + encoding: "gzip", + }); + + expect(compressCalls).toBe(2); + expect(retried).not.toBeNull(); + }); + + it("recompresses when the file changes underneath the same path", async () => { + let compressCalls = 0; + const cache = makeStaticCompressionCache(async (data) => { + compressCalls += 1; + return data.subarray(0, 64); + }); + + await cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "gzip" }); + await cache.get({ cacheKey: "bundle.js 2 140", data: bundle, encoding: "gzip" }); + + expect(compressCalls).toBe(2); + }); + + it("keeps brotli and gzip results apart", async () => { + const cache = makeStaticCompressionCache(async (data, encoding) => + new TextEncoder().encode(`${encoding}:${data.byteLength}`), + ); + + const brotli = await cache.get({ cacheKey: "bundle.js 1 100", data: bundle, encoding: "br" }); + const gzipped = await cache.get({ + cacheKey: "bundle.js 1 100", + data: bundle, + encoding: "gzip", + }); + + expect(new TextDecoder().decode(brotli ?? new Uint8Array())).toContain("br:"); + expect(new TextDecoder().decode(gzipped ?? new Uint8Array())).toContain("gzip:"); + }); + + it("skips payloads too small to be worth compressing", async () => { + let compressCalls = 0; + const cache = makeStaticCompressionCache(async (data) => { + compressCalls += 1; + return data; + }); + + const result = await cache.get({ + cacheKey: "tiny.js 1 8", + data: new TextEncoder().encode("const a=1;"), + encoding: "gzip", + }); + + expect(result).toBeNull(); + expect(compressCalls).toBe(0); + }); + + it("declines a result that did not get smaller", async () => { + const cache = makeStaticCompressionCache(async (data) => new Uint8Array(data.byteLength + 32)); + + const result = await cache.get({ + cacheKey: "incompressible.bin 1 100", + data: bundle, + encoding: "gzip", + }); + + expect(result).toBeNull(); + expect(cache.retainedByteLength).toBe(0); + }); + + it("really shrinks a realistic bundle with the default compressor", async () => { + const cache = makeStaticCompressionCache(); + + const compressed = await cache.get({ + cacheKey: "real.js 1 100", + data: bundle, + encoding: "br", + }); + + expect(compressed).not.toBeNull(); + expect((compressed as Uint8Array).byteLength).toBeLessThan(bundle.byteLength / 2); + }); +}); diff --git a/apps/server/src/staticAssetDelivery.ts b/apps/server/src/staticAssetDelivery.ts new file mode 100644 index 00000000000..fa7d41ec255 --- /dev/null +++ b/apps/server/src/staticAssetDelivery.ts @@ -0,0 +1,185 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeCrypto from "node:crypto"; +import * as NodeZlib from "node:zlib"; +import * as NodeUtil from "node:util"; + +/** + * Delivery policy for the packaged web bundle. Vite emits content-hashed + * files under `assets/`, so those are immutable and safe to cache forever, + * while `index.html` and the SPA fallback must revalidate or a new build is + * never picked up. + */ +const IMMUTABLE_CACHE_CONTROL = "public, max-age=31536000, immutable"; +const REVALIDATE_CACHE_CONTROL = "no-cache"; + +/** Vite content hashes are at least 8 url-safe characters before the extension. */ +const HASHED_ASSET_PATTERN = /-[A-Za-z0-9_-]{8,}\.[A-Za-z0-9]+$/; + +/** + * Compressing tiny payloads costs more than it saves, and the header + * overhead can make the response larger than the original. + */ +const MIN_COMPRESSIBLE_BYTES = 1024; + +/** Total size of compressed payloads retained across all static files. */ +const COMPRESSED_CACHE_MAX_BYTES = 64 * 1024 * 1024; + +const COMPRESSIBLE_CONTENT_TYPES = [ + "text/", + "application/javascript", + "application/json", + "application/manifest+json", + "application/wasm", + "image/svg+xml", +]; + +const gzip = NodeUtil.promisify(NodeZlib.gzip); +const brotliCompress = NodeUtil.promisify(NodeZlib.brotliCompress); + +export type StaticContentEncoding = "br" | "gzip"; + +export function resolveStaticCacheControl(relativePath: string): string { + const normalized = relativePath.replaceAll("\\", "/").replace(/^\/+/, ""); + return normalized.startsWith("assets/") && HASHED_ASSET_PATTERN.test(normalized) + ? IMMUTABLE_CACHE_CONTROL + : REVALIDATE_CACHE_CONTROL; +} + +/** + * Cache key derived from the bytes being served rather than from file + * metadata. Used where the served content and the metadata could otherwise + * disagree, which would let one request's compression be reused for + * different content. Only worth it for small payloads, since it hashes the + * whole buffer on every request. + */ +export function contentCacheKey(data: Uint8Array): string { + return NodeCrypto.createHash("sha1").update(data).digest("hex"); +} + +export function isCompressibleContentType(contentType: string): boolean { + const normalized = contentType.toLowerCase(); + return COMPRESSIBLE_CONTENT_TYPES.some((prefix) => normalized.startsWith(prefix)); +} + +/** + * Pick an encoding from `Accept-Encoding`, preferring Brotli. Entries with an + * explicit `q=0` are refusals, and `identity;q=0` plus an unsupported codec + * list leaves nothing usable, so the caller falls back to the raw bytes. + */ +export function negotiateStaticEncoding( + acceptEncoding: string | undefined, +): StaticContentEncoding | null { + if (!acceptEncoding) return null; + + const acceptedQualityByCodec = new Map(); + for (const entry of acceptEncoding.split(",")) { + const [rawCodec = "", ...parameters] = entry.trim().split(";"); + const codec = rawCodec.trim().toLowerCase(); + if (!codec) continue; + const quality = parameters + .map((parameter) => /^\s*q=([0-9.]+)\s*$/i.exec(parameter)) + .find((match) => match !== null)?.[1]; + acceptedQualityByCodec.set(codec, quality === undefined ? 1 : Number.parseFloat(quality)); + } + + const accepts = (codec: StaticContentEncoding) => { + const quality = acceptedQualityByCodec.get(codec) ?? acceptedQualityByCodec.get("*"); + return quality !== undefined && quality > 0; + }; + + if (accepts("br")) return "br"; + if (accepts("gzip")) return "gzip"; + return null; +} + +async function compressBytes( + data: Uint8Array, + encoding: StaticContentEncoding, +): Promise { + if (encoding === "gzip") { + return new Uint8Array(await gzip(data)); + } + return new Uint8Array( + await brotliCompress(data, { + params: { + // Default quality 11 costs seconds on a multi-megabyte bundle. 5 is + // the usual static-asset sweet spot, and every result is cached. + [NodeZlib.constants.BROTLI_PARAM_QUALITY]: 5, + [NodeZlib.constants.BROTLI_PARAM_SIZE_HINT]: data.byteLength, + }, + }), + ); +} + +/** + * Compress once per (file, encoding) and reuse the result. Packaged assets + * never change while the server runs, so recompressing a multi-megabyte + * bundle for every remote request is pure waste. Entries are keyed by mtime + * and size so a dev-server rebuild is not served stale. + */ +export function makeStaticCompressionCache(compress = compressBytes) { + const compressedByKey = new Map(); + /** + * Compressions already running. A cold start requests the bundle, its CSS, + * and its chunks at once, and browsers retry, so without this the same + * multi-megabyte payload is compressed once per concurrent request. + */ + const inFlightByKey = new Map>(); + let retainedBytes = 0; + + const evictOldest = (incomingBytes: number) => { + while (retainedBytes + incomingBytes > COMPRESSED_CACHE_MAX_BYTES) { + const oldestKey = compressedByKey.keys().next().value; + if (oldestKey === undefined) return; + retainedBytes -= compressedByKey.get(oldestKey)?.byteLength ?? 0; + compressedByKey.delete(oldestKey); + } + }; + + const compressAndRetain = async ( + key: string, + data: Uint8Array, + encoding: StaticContentEncoding, + ): Promise => { + const compressed = await compress(data, encoding); + // A payload that grew under compression is not worth serving encoded. + if (compressed.byteLength >= data.byteLength) return null; + + if (compressed.byteLength <= COMPRESSED_CACHE_MAX_BYTES) { + evictOldest(compressed.byteLength); + compressedByKey.set(key, compressed); + retainedBytes += compressed.byteLength; + } + return compressed; + }; + + return { + get(input: { + readonly cacheKey: string; + readonly data: Uint8Array; + readonly encoding: StaticContentEncoding; + }): Promise { + if (input.data.byteLength < MIN_COMPRESSIBLE_BYTES) return Promise.resolve(null); + + const key = `${input.encoding} ${input.cacheKey}`; + const cached = compressedByKey.get(key); + if (cached) return Promise.resolve(cached); + + const pending = inFlightByKey.get(key); + if (pending) return pending; + + // Clear the in-flight entry however this settles, so a failed + // compression is retried rather than remembered as permanently pending. + const compression = compressAndRetain(key, input.data, input.encoding).finally(() => { + inFlightByKey.delete(key); + }); + inFlightByKey.set(key, compression); + return compression; + }, + get retainedByteLength(): number { + return retainedBytes; + }, + }; +} + +export type StaticCompressionCache = ReturnType; From ddb67bb953a76164a2914d1f6338e932dd16be54 Mon Sep 17 00:00:00 2001 From: Patrick Roza Date: Sun, 26 Jul 2026 08:16:33 +0200 Subject: [PATCH 3/5] perf(server): back off repeated checkpoint capture failures (upstream #4517) Imported from https://github.com/pingdotgg/t3code/pull/4517 --- .../src/checkpointing/CaptureBackoff.test.ts | 191 ++++++++++++++++++ .../src/checkpointing/CaptureBackoff.ts | 146 +++++++++++++ .../CheckpointCaptureBackoff.test.ts | 168 +++++++++++++++ .../src/checkpointing/CheckpointStore.ts | 39 +++- 4 files changed, 542 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/checkpointing/CaptureBackoff.test.ts create mode 100644 apps/server/src/checkpointing/CaptureBackoff.ts create mode 100644 apps/server/src/checkpointing/CheckpointCaptureBackoff.test.ts diff --git a/apps/server/src/checkpointing/CaptureBackoff.test.ts b/apps/server/src/checkpointing/CaptureBackoff.test.ts new file mode 100644 index 00000000000..6e9a4c64f3d --- /dev/null +++ b/apps/server/src/checkpointing/CaptureBackoff.test.ts @@ -0,0 +1,191 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { cooldownForFailureCount, makeCaptureBackoff } from "./CaptureBackoff.ts"; + +const MINUTE = 60_000; +const CWD = "/repo/workspace"; + +describe("cooldownForFailureCount", () => { + it("tolerates transient failures before opening a cooldown", () => { + expect(cooldownForFailureCount(1)).toBe(0); + expect(cooldownForFailureCount(2)).toBe(0); + }); + + it("backs off further the longer capture keeps failing", () => { + expect(cooldownForFailureCount(3)).toBe(5 * MINUTE); + expect(cooldownForFailureCount(4)).toBe(10 * MINUTE); + expect(cooldownForFailureCount(5)).toBe(20 * MINUTE); + }); + + it("caps the cooldown so a workspace is retried eventually", () => { + expect(cooldownForFailureCount(20)).toBe(60 * MINUTE); + expect(cooldownForFailureCount(500)).toBe(60 * MINUTE); + }); +}); + +describe("makeCaptureBackoff", () => { + it("does not skip before the failure threshold is reached", () => { + const backoff = makeCaptureBackoff(); + + backoff.recordFailure(CWD, 0, "timeout"); + backoff.recordFailure(CWD, 1_000, "timeout"); + + expect(backoff.beginAttempt(CWD, 2_000).skip).toBe(false); + }); + + it("skips and replays the recorded failure once capture keeps failing", () => { + const backoff = makeCaptureBackoff(); + + for (const at of [0, 1_000, 2_000]) { + backoff.recordFailure(CWD, at, "git add timed out"); + } + + const decision = backoff.beginAttempt(CWD, 3_000); + expect(decision.skip).toBe(true); + expect(decision.lastError).toBe("git add timed out"); + expect(decision.remainingMs).toBeGreaterThan(0); + }); + + it("retries again once the cooldown elapses", () => { + const backoff = makeCaptureBackoff(); + + for (const at of [0, 0, 0]) { + backoff.recordFailure(CWD, at, "timeout"); + } + + expect(backoff.beginAttempt(CWD, 5 * MINUTE - 1).skip).toBe(true); + expect(backoff.beginAttempt(CWD, 5 * MINUTE).skip).toBe(false); + }); + + it("clears the record after a capture succeeds", () => { + const backoff = makeCaptureBackoff(); + + for (const at of [0, 0, 0]) { + backoff.recordFailure(CWD, at, "timeout"); + } + expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true); + + backoff.recordSuccess(CWD); + + expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(false); + expect(backoff.trackedWorkspaceCount).toBe(0); + }); + + it("tracks each workspace independently", () => { + const backoff = makeCaptureBackoff(); + const healthy = "/repo/other"; + + for (const at of [0, 0, 0]) { + backoff.recordFailure(CWD, at, "timeout"); + } + + expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true); + expect(backoff.beginAttempt(healthy, 1_000).skip).toBe(false); + }); + + it("bounds how many workspaces it retains", () => { + const backoff = makeCaptureBackoff(); + + for (let index = 0; index < 400; index += 1) { + backoff.recordFailure(`/repo/workspace-${index}`, index, "timeout"); + } + + expect(backoff.trackedWorkspaceCount).toBe(256); + // The oldest entries are the ones dropped: an evicted workspace starts + // counting again from one, a retained one continues. + expect(backoff.recordFailure("/repo/workspace-0", 1_000, "timeout").consecutiveFailures).toBe( + 1, + ); + expect(backoff.recordFailure("/repo/workspace-399", 1_000, "timeout").consecutiveFailures).toBe( + 2, + ); + }); + + it("keeps a repeatedly failing workspace alive through eviction pressure", () => { + const backoff = makeCaptureBackoff(); + + // A workspace in active use fails on every turn while many one-off + // workspaces churn past it. + for (let index = 0; index < 400; index += 1) { + backoff.recordFailure(CWD, index, "timeout"); + backoff.recordFailure(`/repo/other-${index}`, index, "timeout"); + } + + expect(backoff.beginAttempt(CWD, 400).skip).toBe(true); + }); + + it("keeps a workspace that is only being skipped safe from eviction", () => { + const backoff = makeCaptureBackoff(); + + for (const at of [0, 0, 0]) { + backoff.recordFailure(CWD, at, "timeout"); + } + + // A skipped workspace never calls recordFailure again, so only the skip + // itself can keep it ahead of churn from unrelated workspaces. + for (let index = 0; index < 400; index += 1) { + expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true); + backoff.recordFailure(`/repo/other-${index}`, index, "timeout"); + } + + expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(true); + }); + + it("does not reserve for a workspace that has not reached the threshold", () => { + const backoff = makeCaptureBackoff(); + + backoff.recordFailure(CWD, 0, "timeout"); + + // One transient failure must not suppress a capture running alongside it. + expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(false); + expect(backoff.beginAttempt(CWD, 1_000).skip).toBe(false); + expect(backoff.beginAttempt(CWD, 2_000).skip).toBe(false); + }); + + it("releases only one caller when the cooldown expires", () => { + const backoff = makeCaptureBackoff(); + + for (const at of [0, 0, 0]) { + backoff.recordFailure(CWD, at, "timeout"); + } + + // Threads sharing a workspace can complete turns together, and only the + // first past the cooldown should pay for the capture. + const released = [ + backoff.beginAttempt(CWD, 5 * MINUTE), + backoff.beginAttempt(CWD, 5 * MINUTE), + backoff.beginAttempt(CWD, 5 * MINUTE), + ].filter((decision) => !decision.skip); + + expect(released).toHaveLength(1); + }); + + it("lets the reservation lapse when an attempt never reports back", () => { + const backoff = makeCaptureBackoff(); + + for (const at of [0, 0, 0]) { + backoff.recordFailure(CWD, at, "timeout"); + } + backoff.beginAttempt(CWD, 5 * MINUTE); + + // An interrupted capture reports neither success nor failure, so the + // reservation must expire on its own rather than wedge the workspace. + expect(backoff.beginAttempt(CWD, 5 * MINUTE + 30_000).skip).toBe(true); + expect(backoff.beginAttempt(CWD, 6 * MINUTE + 1).skip).toBe(false); + }); + + it("keeps extending the cooldown while failures continue", () => { + const backoff = makeCaptureBackoff(); + + for (const at of [0, 0, 0]) { + backoff.recordFailure(CWD, at, "timeout"); + } + const firstRemaining = backoff.beginAttempt(CWD, 0).remainingMs; + + // The next attempt after the cooldown fails again, so the wait grows. + backoff.recordFailure(CWD, 5 * MINUTE, "timeout"); + const secondRemaining = backoff.beginAttempt(CWD, 5 * MINUTE).remainingMs; + + expect(secondRemaining).toBeGreaterThan(firstRemaining); + }); +}); diff --git a/apps/server/src/checkpointing/CaptureBackoff.ts b/apps/server/src/checkpointing/CaptureBackoff.ts new file mode 100644 index 00000000000..d53a8278e68 --- /dev/null +++ b/apps/server/src/checkpointing/CaptureBackoff.ts @@ -0,0 +1,146 @@ +/** + * Consecutive-failure backoff for workspace checkpoint capture. + * + * Capture runs a full-tree `git add -A` under a temporary index after every + * completed turn. On a repository large enough to exceed the VCS process + * timeout, that capture can never succeed, so the unguarded retry pegs a CPU + * core for as long as any thread is in use and litters `.git/objects/pack` + * with `tmp_pack_*` files from each killed process. + * + * After a few consecutive failures for a workspace, capture is skipped for a + * growing cooldown instead of being retried every turn. Any success clears + * the record, so a transient failure (a lock held by a concurrent git + * command) costs nothing. + * + * @module CaptureBackoff + */ + +/** Failures tolerated before a workspace enters cooldown. */ +const FAILURE_THRESHOLD = 3; +const BASE_COOLDOWN_MS = 5 * 60_000; +const MAX_COOLDOWN_MS = 60 * 60_000; + +/** + * Only a workspace that later succeeds clears its own record, so a workspace + * that fails once and is then abandoned would otherwise be retained for the + * lifetime of the server. Bound the tracked set and evict least-recently + * touched entries; dropping one only costs a workspace its failure history. + */ +const MAX_TRACKED_WORKSPACES = 256; + +/** + * How long the first caller past an expired cooldown holds the retry to + * itself. Threads sharing a workspace complete turns independently, so + * without this they would all be released at once and launch the same + * expensive capture together. Comfortably longer than the VCS process + * timeout that kills a stuck capture, and it lapses on its own, so an + * attempt that never reports back cannot wedge the workspace. + */ +const ATTEMPT_RESERVATION_MS = 60_000; + +export interface CaptureBackoffDecision { + readonly skip: boolean; + /** Milliseconds left in the cooldown, for logging. Zero when not skipping. */ + readonly remainingMs: number; + /** + * The failure that opened the cooldown. Replayed instead of inventing a new + * error, so callers keep seeing the real reason capture is unavailable and + * the error channel is unchanged. + */ + readonly lastError: E | null; +} + +export function cooldownForFailureCount(consecutiveFailures: number): number { + if (consecutiveFailures < FAILURE_THRESHOLD) return 0; + const doublings = consecutiveFailures - FAILURE_THRESHOLD; + // Clamp the exponent before shifting so a long-lived workspace cannot + // overflow into a negative or infinite cooldown. + const scale = 2 ** Math.min(doublings, 10); + return Math.min(BASE_COOLDOWN_MS * scale, MAX_COOLDOWN_MS); +} + +export interface CaptureFailureOutcome { + readonly consecutiveFailures: number; + /** Zero while the workspace is still under the failure threshold. */ + readonly cooldownMs: number; +} + +interface WorkspaceRecord { + consecutiveFailures: number; + skipUntilMs: number; + lastError: E; +} + +/** + * Tracks capture health per workspace. Callers ask whether to skip, then + * report the outcome of any capture they actually ran. + */ +export function makeCaptureBackoff() { + const recordByCwd = new Map>(); + + /** Move a record to the most-recent position so eviction sees real usage. */ + const touch = (cwd: string, record: WorkspaceRecord) => { + recordByCwd.delete(cwd); + recordByCwd.set(cwd, record); + }; + + return { + /** + * Decide whether this caller should run a capture. Mutating: a caller + * released past an expired cooldown reserves the attempt, and any read + * refreshes eviction recency so a workspace being actively skipped is not + * evicted by churn from unrelated workspaces. + */ + beginAttempt(cwd: string, nowMs: number): CaptureBackoffDecision { + const record = recordByCwd.get(cwd); + if (!record) { + return { skip: false, remainingMs: 0, lastError: null }; + } + + touch(cwd, record); + // Below the threshold a workspace has no cooldown at all, and its zero + // deadline must not read as one that just expired: reserving there + // would let a single transient failure suppress a concurrent capture. + if (record.consecutiveFailures < FAILURE_THRESHOLD) { + return { skip: false, remainingMs: 0, lastError: null }; + } + + if (nowMs < record.skipUntilMs) { + return { + skip: true, + remainingMs: record.skipUntilMs - nowMs, + lastError: record.lastError, + }; + } + + record.skipUntilMs = nowMs + ATTEMPT_RESERVATION_MS; + return { skip: false, remainingMs: 0, lastError: null }; + }, + + recordSuccess(cwd: string): void { + recordByCwd.delete(cwd); + }, + + recordFailure(cwd: string, nowMs: number, error: E): CaptureFailureOutcome { + const consecutiveFailures = (recordByCwd.get(cwd)?.consecutiveFailures ?? 0) + 1; + const cooldownMs = cooldownForFailureCount(consecutiveFailures); + touch(cwd, { + consecutiveFailures, + skipUntilMs: cooldownMs === 0 ? 0 : nowMs + cooldownMs, + lastError: error, + }); + + while (recordByCwd.size > MAX_TRACKED_WORKSPACES) { + const oldestCwd = recordByCwd.keys().next().value; + if (oldestCwd === undefined) break; + recordByCwd.delete(oldestCwd); + } + + return { consecutiveFailures, cooldownMs }; + }, + + get trackedWorkspaceCount(): number { + return recordByCwd.size; + }, + }; +} diff --git a/apps/server/src/checkpointing/CheckpointCaptureBackoff.test.ts b/apps/server/src/checkpointing/CheckpointCaptureBackoff.test.ts new file mode 100644 index 00000000000..ab1b3351a97 --- /dev/null +++ b/apps/server/src/checkpointing/CheckpointCaptureBackoff.test.ts @@ -0,0 +1,168 @@ +import { it } from "@effect/vitest"; +import { ThreadId, VcsProcessTimeoutError } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import { describe, expect } from "vite-plus/test"; + +import * as CheckpointStore from "./CheckpointStore.ts"; +import { checkpointRefForThreadTurn } from "./Utils.ts"; +import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; +import type * as VcsDriver from "../vcs/VcsDriver.ts"; + +const CWD = "/repo/huge-monorepo"; +const THREAD_ID = ThreadId.make("thread-checkpoint-backoff"); + +const captureTimeout = new VcsProcessTimeoutError({ + operation: "GitVcsDriver.checkpoints.captureCheckpoint", + command: "git add", + cwd: CWD, + timeoutMs: 30_000, +}); + +/** + * A driver whose capture always exceeds the process timeout, matching a + * repository too large for a full-tree `git add -A` to finish in time. + */ +function makeAlwaysTimingOutRegistry( + captureAttempts: { count: number }, + options: { readonly succeedOnAttempt?: number } = {}, +) { + const checkpoints = { + captureCheckpoint: () => + Effect.suspend(() => { + captureAttempts.count += 1; + return captureAttempts.count === options.succeedOnAttempt + ? Effect.void + : Effect.fail(captureTimeout); + }), + hasCheckpointRef: () => Effect.succeed(false), + restoreCheckpoint: () => Effect.succeed(false), + diffCheckpoints: () => Effect.succeed(""), + deleteCheckpointRefs: () => Effect.void, + } as unknown as VcsDriver.VcsCheckpointOps; + + const handle = { + kind: "git" as const, + repository: { + kind: "git" as const, + rootPath: CWD, + metadataPath: `${CWD}/.git`, + freshness: { source: "cache" as const, checkedAt: 0 }, + }, + driver: { checkpoints } as unknown as VcsDriver.VcsDriver["Service"], + } as unknown as VcsDriverRegistry.VcsDriverHandle; + + return Layer.succeed( + VcsDriverRegistry.VcsDriverRegistry, + VcsDriverRegistry.VcsDriverRegistry.of({ + get: () => Effect.succeed(handle.driver), + detect: () => Effect.succeed(handle), + resolve: () => Effect.succeed(handle), + }), + ); +} + +describe("checkpoint capture backoff", () => { + it.effect("stops re-running a capture that keeps timing out", () => { + const captureAttempts = { count: 0 }; + + return Effect.gen(function* () { + const store = yield* CheckpointStore.CheckpointStore; + const capture = (turn: number) => + store + .captureCheckpoint({ + cwd: CWD, + checkpointRef: checkpointRefForThreadTurn(THREAD_ID, turn), + }) + .pipe(Effect.flip); + + // The first three turns each pay for a real capture attempt. + for (let turn = 1; turn <= 3; turn += 1) { + const error = yield* capture(turn); + expect(error._tag).toBe("VcsProcessTimeoutError"); + } + expect(captureAttempts.count).toBe(3); + + // Every later turn inside the cooldown replays the recorded failure + // without spawning git again. The cooldown is minutes long, so the rest + // of this test runs well inside it. + for (let turn = 4; turn <= 20; turn += 1) { + const error = yield* capture(turn); + expect(error).toBe(captureTimeout); + } + expect(captureAttempts.count).toBe(3); + }).pipe( + Effect.provide( + CheckpointStore.layer.pipe(Layer.provide(makeAlwaysTimingOutRegistry(captureAttempts))), + ), + ); + }); + + it.effect("does not hold a workspace when the driver cannot be resolved", () => { + const captureAttempts = { count: 0 }; + const registryFailure = new VcsProcessTimeoutError({ + operation: "VcsDriverRegistry.resolve", + command: "git rev-parse", + cwd: CWD, + timeoutMs: 5_000, + }); + const failingRegistry = Layer.succeed( + VcsDriverRegistry.VcsDriverRegistry, + VcsDriverRegistry.VcsDriverRegistry.of({ + get: () => Effect.fail(registryFailure), + detect: () => Effect.fail(registryFailure), + resolve: () => Effect.fail(registryFailure), + }) as never, + ); + + return Effect.gen(function* () { + const store = yield* CheckpointStore.CheckpointStore; + const capture = (turn: number) => + store + .captureCheckpoint({ + cwd: CWD, + checkpointRef: checkpointRefForThreadTurn(THREAD_ID, turn), + }) + .pipe(Effect.flip); + + // Failing before the driver runs still counts as a failure, so the + // first two turns stay under the threshold and keep retrying rather + // than being held by a stale reservation. + for (let turn = 1; turn <= 2; turn += 1) { + const error = yield* capture(turn); + expect(error).toBe(registryFailure); + } + expect(captureAttempts.count).toBe(0); + }).pipe(Effect.provide(CheckpointStore.layer.pipe(Layer.provide(failingRegistry)))); + }); + + it.effect("keeps capturing for a workspace that recovers", () => { + const captureAttempts = { count: 0 }; + + return Effect.gen(function* () { + const store = yield* CheckpointStore.CheckpointStore; + const capture = (turn: number) => + store.captureCheckpoint({ + cwd: CWD, + checkpointRef: checkpointRefForThreadTurn(THREAD_ID, turn), + }); + + // Two failures stay under the threshold, and the success that follows + // clears them, so a later isolated failure does not open a cooldown. + yield* capture(1).pipe(Effect.flip); + yield* capture(2).pipe(Effect.flip); + yield* capture(3); + yield* capture(4).pipe(Effect.flip); + yield* capture(5).pipe(Effect.flip); + yield* capture(6).pipe(Effect.flip); + + expect(captureAttempts.count).toBe(6); + }).pipe( + Effect.provide( + CheckpointStore.layer.pipe( + Layer.provide(makeAlwaysTimingOutRegistry(captureAttempts, { succeedOnAttempt: 3 })), + ), + ), + ); + }); +}); diff --git a/apps/server/src/checkpointing/CheckpointStore.ts b/apps/server/src/checkpointing/CheckpointStore.ts index f13aa4572c1..2dd6ab71dbd 100644 --- a/apps/server/src/checkpointing/CheckpointStore.ts +++ b/apps/server/src/checkpointing/CheckpointStore.ts @@ -14,10 +14,12 @@ * @module CheckpointStore */ import { VcsUnsupportedOperationError, type CheckpointRef } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import { makeCaptureBackoff } from "./CaptureBackoff.ts"; import type { CheckpointStoreError } from "./Errors.ts"; import type { VcsCheckpointOps } from "../vcs/VcsDriver.ts"; import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; @@ -98,6 +100,7 @@ export class CheckpointStore extends Context.Service< export const make = Effect.gen(function* () { const vcsRegistry = yield* VcsDriverRegistry.VcsDriverRegistry; + const captureBackoff = makeCaptureBackoff(); const resolveCheckpoints = Effect.fn("CheckpointStore.resolveCheckpoints")(function* ( operation: string, @@ -122,8 +125,40 @@ export const make = Effect.gen(function* () { const captureCheckpoint: CheckpointStore["Service"]["captureCheckpoint"] = Effect.fn( "captureCheckpoint", )(function* (input) { - const checkpoints = yield* resolveCheckpoints("CheckpointStore.captureCheckpoint", input.cwd); - return yield* checkpoints.captureCheckpoint(input); + const startedAt = yield* Clock.currentTimeMillis; + const decision = captureBackoff.beginAttempt(input.cwd, startedAt); + if (decision.skip && decision.lastError) { + // Spawning git here would repeat a capture that cannot succeed, so + // replay the failure the caller already handles instead of burning a + // CPU core and leaving another orphaned tmp_pack behind. + yield* Effect.logDebug("Skipping checkpoint capture during failure cooldown", { + cwdLength: input.cwd.length, + remainingMs: decision.remainingMs, + }); + return yield* Effect.fail(decision.lastError); + } + + // Resolving the driver sits inside the reported scope so that a failure + // before the capture runs still replaces this caller's reservation with a + // real outcome, rather than leaving the workspace held until it lapses. + return yield* Effect.gen(function* () { + const checkpoints = yield* resolveCheckpoints("CheckpointStore.captureCheckpoint", input.cwd); + return yield* checkpoints.captureCheckpoint(input); + }).pipe( + Effect.tap(() => Effect.sync(() => captureBackoff.recordSuccess(input.cwd))), + Effect.tapError((error) => + Effect.gen(function* () { + const failedAt = yield* Clock.currentTimeMillis; + const outcome = captureBackoff.recordFailure(input.cwd, failedAt, error); + yield* Effect.logWarning("Checkpoint capture failed", { + cwdLength: input.cwd.length, + consecutiveFailures: outcome.consecutiveFailures, + cooldownMs: outcome.cooldownMs, + error, + }); + }), + ), + ); }); const hasCheckpointRef: CheckpointStore["Service"]["hasCheckpointRef"] = Effect.fn( From 91aa0b1d7880e2c88c4bbde406c58ff5e3c8abc6 Mon Sep 17 00:00:00 2001 From: Patrick Roza Date: Sun, 26 Jul 2026 08:16:59 +0200 Subject: [PATCH 4/5] fix(server): preserve manually renamed thread titles (upstream #4558) Imported from https://github.com/pingdotgg/t3code/pull/4558\n\nAdapted to retain our provider restart-recovery constants while replacing the local default-title check with the shared policy. --- .../Layers/ProviderCommandReactor.ts | 7 +- .../Layers/ProviderRuntimeIngestion.test.ts | 217 +++++++++++++++++- .../Layers/ProviderRuntimeIngestion.ts | 18 +- apps/web/src/components/CommandPalette.tsx | 13 +- packages/shared/package.json | 4 + packages/shared/src/threadTitle.test.ts | 64 ++++++ packages/shared/src/threadTitle.ts | 17 ++ 7 files changed, 326 insertions(+), 14 deletions(-) create mode 100644 packages/shared/src/threadTitle.test.ts create mode 100644 packages/shared/src/threadTitle.ts diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index b6588cbea37..1d2290f8914 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -13,6 +13,7 @@ import { type RuntimeMode, type TurnId, } from "@t3tools/contracts"; +import { isDefaultThreadTitle } from "@t3tools/shared/threadTitle"; import { isTemporaryWorktreeBranch, WORKTREE_BRANCH_PREFIX } from "@t3tools/shared/git"; import * as Cache from "effect/Cache"; import * as Cause from "effect/Cause"; @@ -106,7 +107,6 @@ const turnStartKeyForEvent = (event: ProviderIntentEvent): string => const HANDLED_TURN_START_KEY_MAX = 10_000; const HANDLED_TURN_START_KEY_TTL = Duration.minutes(30); const DEFAULT_RUNTIME_MODE: RuntimeMode = "full-access"; -const DEFAULT_THREAD_TITLE = "New thread"; const STARTUP_RECOVERY_CONCURRENCY = 4; export const RESTART_RECOVERY_CONTINUATION_INSTRUCTION = @@ -128,14 +128,13 @@ export function providerErrorLabelFromInstanceHint(input: { } function canReplaceThreadTitle(currentTitle: string, titleSeed?: string): boolean { - const trimmedCurrentTitle = currentTitle.trim(); - if (trimmedCurrentTitle === DEFAULT_THREAD_TITLE) { + if (isDefaultThreadTitle(currentTitle)) { return true; } const trimmedTitleSeed = titleSeed?.trim(); return trimmedTitleSeed !== undefined && trimmedTitleSeed.length > 0 - ? trimmedCurrentTitle === trimmedTitleSeed + ? currentTitle.trim() === trimmedTitleSeed : false; } diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index b057f5afddb..c52c6cacf1d 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -2849,7 +2849,7 @@ describe("ProviderRuntimeIngestion", () => { const thread = await waitForThread( harness.readModel, (entry) => - entry.title === "Renamed by provider" && + entry.title === "Thread" && entry.activities.some( (activity: ProviderRuntimeTestActivity) => activity.kind === "turn.plan.updated", ) && @@ -2864,7 +2864,7 @@ describe("ProviderRuntimeIngestion", () => { ), ); - expect(thread.title).toBe("Renamed by provider"); + expect(thread.title).toBe("Thread"); const planActivity = thread.activities.find( (activity: ProviderRuntimeTestActivity) => activity.id === "evt-turn-plan-updated", @@ -2905,6 +2905,219 @@ describe("ProviderRuntimeIngestion", () => { expect(checkpoint?.checkpointRef).toBe("provider-diff:evt-turn-diff-updated"); }); + effectIt.effect("applies provider thread.metadata.updated when thread title is the default", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const now = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-default"), + threadId: ThreadId.make("thread-default"), + projectId: asProjectId("project-1"), + title: "New thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + + harness.emit({ + type: "thread.metadata.updated", + eventId: asEventId("evt-thread-metadata-default"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-default"), + payload: { + name: "Provider default title", + metadata: { source: "provider" }, + }, + }); + + yield* Effect.promise(() => harness.drain()); + + const thread = yield* Effect.promise(() => + waitForThread( + harness.readModel, + (entry) => entry.title === "Provider default title", + 2000, + asThreadId("thread-default"), + ), + ); + + expect(thread.title).toBe("Provider default title"); + }), + ); + + effectIt.effect( + "rejects provider thread.metadata.updated when thread title is already customized", + () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const now = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.meta.update", + commandId: CommandId.make("cmd-thread-custom-title"), + threadId: ThreadId.make("thread-1"), + title: "My custom title", + }); + + harness.emit({ + type: "thread.metadata.updated", + eventId: asEventId("evt-thread-metadata-custom"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + name: "Provider override attempt", + metadata: { source: "provider" }, + }, + }); + + yield* Effect.promise(() => harness.drain()); + + yield* Effect.promise(() => + waitForThread( + harness.readModel, + (entry) => entry.id === "thread-1" && entry.title === "My custom title", + ), + ); + + const readModel = yield* Effect.promise(() => harness.readModel()); + const thread = readModel.threads.find((entry) => entry.id === "thread-1"); + expect(thread?.title).toBe("My custom title"); + }), + ); + + effectIt.effect("skips provider thread.metadata.updated when payload.name is missing", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const now = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-no-name"), + threadId: ThreadId.make("thread-no-name"), + projectId: asProjectId("project-1"), + title: "New thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + + harness.emit({ + type: "thread.metadata.updated", + eventId: asEventId("evt-thread-metadata-no-name"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-no-name"), + payload: {}, + }); + + yield* Effect.promise(() => harness.drain()); + + const readModel = yield* Effect.promise(() => harness.readModel()); + const thread = readModel.threads.find((entry) => entry.id === "thread-no-name"); + expect(thread?.title).toBe("New thread"); + }), + ); + + effectIt.effect("skips provider thread.metadata.updated when payload.name is empty string", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const now = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-empty"), + threadId: ThreadId.make("thread-empty-name"), + projectId: asProjectId("project-1"), + title: "New thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + + harness.emit({ + type: "thread.metadata.updated", + eventId: asEventId("evt-thread-metadata-empty"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-empty-name"), + payload: { + name: "", + }, + }); + + yield* Effect.promise(() => harness.drain()); + + const readModel = yield* Effect.promise(() => harness.readModel()); + const thread = readModel.threads.find((entry) => entry.id === "thread-empty-name"); + expect(thread?.title).toBe("New thread"); + }), + ); + + effectIt.effect( + "skips provider thread.metadata.updated when payload.name sanitizes to empty", + () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => createHarness()); + const now = "2026-01-01T00:00:00.000Z"; + + yield* harness.engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create-whitespace"), + threadId: ThreadId.make("thread-whitespace-name"), + projectId: asProjectId("project-1"), + title: "New thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + + harness.emit({ + type: "thread.metadata.updated", + eventId: asEventId("evt-thread-metadata-whitespace"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-whitespace-name"), + payload: { + name: " ", + }, + }); + + yield* Effect.promise(() => harness.drain()); + + const readModel = yield* Effect.promise(() => harness.readModel()); + const thread = readModel.threads.find((entry) => entry.id === "thread-whitespace-name"); + expect(thread?.title).toBe("New thread"); + }), + ); + it("projects context window updates into normalized thread activities", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index d901340e3ac..1e51b30968c 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -26,6 +26,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import { isDefaultThreadTitle, sanitizeTitle } from "@t3tools/shared/threadTitle"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts"; @@ -1744,12 +1745,17 @@ const make = Effect.gen(function* () { } if (event.type === "thread.metadata.updated" && event.payload.name) { - yield* orchestrationEngine.dispatch({ - type: "thread.meta.update", - commandId: yield* providerCommandId(event, "thread-meta-update"), - threadId: thread.id, - title: event.payload.name, - }); + if (isDefaultThreadTitle(thread.title)) { + const sanitized = sanitizeTitle(event.payload.name); + if (sanitized.length > 0) { + yield* orchestrationEngine.dispatch({ + type: "thread.meta.update", + commandId: yield* providerCommandId(event, "thread-meta-update"), + threadId: thread.id, + title: sanitized, + }); + } + } } if (event.type === "turn.diff.updated") { diff --git a/apps/web/src/components/CommandPalette.tsx b/apps/web/src/components/CommandPalette.tsx index a1d54e8cfc3..37d3830d016 100644 --- a/apps/web/src/components/CommandPalette.tsx +++ b/apps/web/src/components/CommandPalette.tsx @@ -59,6 +59,7 @@ import { useAtomCommand } from "../state/use-atom-command"; import { useAtomQueryRunner } from "../state/use-atom-query-runner"; import { useEnvironments, usePrimaryEnvironmentId } from "../state/environments"; import { useProjects, useThreadShells } from "../state/entities"; +import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/models"; import { resolveThreadActionProjectRef, startNewThreadFromContext } from "../lib/chatThreadActions"; import { appendBrowsePathSegment, @@ -468,6 +469,14 @@ function CommandPaletteDialog(props: { ); } +function renderThreadLeadingContent(thread: EnvironmentThreadShell) { + return ; +} + +function renderThreadTrailingContent(thread: EnvironmentThreadShell) { + return ; +} + function OpenCommandPaletteDialog(props: { readonly openIntent: CommandPaletteOpenIntent | null; readonly setOpen: (open: boolean) => void; @@ -853,8 +862,8 @@ function OpenCommandPaletteDialog(props: { projectTitleById, sortOrder: clientSettings.sidebarThreadSortOrder, icon: , - renderLeadingContent: (thread) => , - renderTrailingContent: (thread) => , + renderLeadingContent: renderThreadLeadingContent, + renderTrailingContent: renderThreadTrailingContent, runThread: async (thread) => { await navigate({ to: "/$environmentId/$threadId", diff --git a/packages/shared/package.json b/packages/shared/package.json index 7d0cc5eb820..5757143dca0 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -163,6 +163,10 @@ "types": "./src/terminalLabels.ts", "import": "./src/terminalLabels.ts" }, + "./threadTitle": { + "types": "./src/threadTitle.ts", + "import": "./src/threadTitle.ts" + }, "./relayClient": { "types": "./src/relayClient.ts", "import": "./src/relayClient.ts" diff --git a/packages/shared/src/threadTitle.test.ts b/packages/shared/src/threadTitle.test.ts new file mode 100644 index 00000000000..a211c9796a9 --- /dev/null +++ b/packages/shared/src/threadTitle.test.ts @@ -0,0 +1,64 @@ +import { describe, it, expect } from "vite-plus/test"; +import { isDefaultThreadTitle, sanitizeTitle, DEFAULT_THREAD_TITLE } from "./threadTitle.ts"; + +describe("isDefaultThreadTitle", () => { + it("returns true for the default title", () => { + expect(isDefaultThreadTitle("New thread")).toBe(true); + }); + + it("returns true for the default title with whitespace", () => { + expect(isDefaultThreadTitle(" New thread ")).toBe(true); + }); + + it("returns false for a custom title", () => { + expect(isDefaultThreadTitle("My custom title")).toBe(false); + }); + + it("returns false for empty string", () => { + expect(isDefaultThreadTitle("")).toBe(false); + }); + + it("returns false for null", () => { + expect(isDefaultThreadTitle(null)).toBe(false); + }); + + it("returns false for undefined", () => { + expect(isDefaultThreadTitle(undefined)).toBe(false); + }); +}); + +describe("sanitizeTitle", () => { + it("preserves a normal title", () => { + expect(sanitizeTitle("Hello World")).toBe("Hello World"); + }); + + it("trims whitespace", () => { + expect(sanitizeTitle(" hello ")).toBe("hello"); + }); + + it("strips control characters", () => { + expect(sanitizeTitle("hello\x00world")).toBe("helloworld"); + expect(sanitizeTitle("hello\x1Fworld")).toBe("helloworld"); + }); + + it("returns empty string for whitespace-only input", () => { + expect(sanitizeTitle(" ")).toBe(""); + expect(sanitizeTitle(" \t ")).toBe(""); + }); + + it("returns empty string for control-char-only input", () => { + expect(sanitizeTitle("\x00")).toBe(""); + expect(sanitizeTitle("\x00\x00")).toBe(""); + }); + + it("truncates to MAX_TITLE_LENGTH", () => { + const long = "a".repeat(1000); + expect(sanitizeTitle(long).length).toBe(500); + }); +}); + +describe("DEFAULT_THREAD_TITLE", () => { + it("equals New thread", () => { + expect(DEFAULT_THREAD_TITLE).toBe("New thread"); + }); +}); diff --git a/packages/shared/src/threadTitle.ts b/packages/shared/src/threadTitle.ts new file mode 100644 index 00000000000..ee35dde673f --- /dev/null +++ b/packages/shared/src/threadTitle.ts @@ -0,0 +1,17 @@ +const MAX_TITLE_LENGTH = 500; + +export const DEFAULT_THREAD_TITLE = "New thread"; + +export function isDefaultThreadTitle(title: string | null | undefined): boolean { + return (title ?? "").trim() === DEFAULT_THREAD_TITLE; +} + +export function sanitizeTitle(title: string): string { + return title + .replace(/./g, (c) => { + const code = c.charCodeAt(0); + return code > 0x1f || code === 0x09 || code === 0x0a || code === 0x0d ? c : ""; + }) + .slice(0, MAX_TITLE_LENGTH) + .trim(); +} From 81fd8bd23ce578baf1664a12a51b5e6a0af462db Mon Sep 17 00:00:00 2001 From: Patrick Roza Date: Sun, 26 Jul 2026 08:17:03 +0200 Subject: [PATCH 5/5] fix(web): avoid woke status overlap after snooze closes (upstream #4539) Imported from https://github.com/pingdotgg/t3code/pull/4539 --- apps/web/src/components/SidebarV2.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index 48bae67b553..be4437b5bff 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -879,10 +879,10 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { ) : ( )} - +