From 0f04d1657d11b83321dfd4d9ea83210fcbae7eb4 Mon Sep 17 00:00:00 2001 From: bompus Date: Fri, 16 Oct 2026 19:24:42 -0600 Subject: [PATCH 1/4] fix(server): scrolling back through a long thread no longer rebuilds the whole thread index on every page Each history or timeline page read every turn item of the thread and rebuilt the visible-item index in JS, so a full scrollback cost time quadratic in thread length (13,000 items: 72 pages, 2.3 to 2.7 s). The index is now kept for the 4 most recently read threads and reused while the newest event of the thread and its fork sources is unchanged, using the same revision check thread find already relies on. --- .../src/orchestration-v2/ProjectionStore.ts | 52 +++++++++++++++---- .../src/orchestration-v2/ThreadFind.test.ts | 33 ++++++++++++ .../server/src/orchestration-v2/ThreadFind.ts | 22 +------- .../src/orchestration-v2/threadRevision.ts | 30 +++++++++++ 4 files changed, 107 insertions(+), 30 deletions(-) create mode 100644 apps/server/src/orchestration-v2/threadRevision.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 4222c3c27021a..22740eb341782 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -92,6 +92,7 @@ import { THREAD_HISTORY_PAGE_POLICY, OLDER_THREAD_USER_TURN_LIMIT, } from "./threadHistoryPaging.ts"; +import { makeThreadRevisionReader } from "./threadRevision.ts"; export class ProjectionStoreApplyEventError extends Schema.TaggedError()( "ProjectionStoreApplyEventError", @@ -4926,17 +4927,15 @@ export const layer: Layer.Layer = readonly item: TimelineIndexItem; readonly synthetic?: OrchestrationV2TurnItem; }; + type TimelineIndex = { + readonly records: ProjectionRecords<"runs" | "attempts">; + readonly local: ReadonlyArray; + readonly visible: ReadonlyArray; + }; const readTimelineIndex = ( threadId: ThreadId, seen: ReadonlySet, - ): Effect.Effect< - { - readonly records: ProjectionRecords<"runs" | "attempts">; - readonly local: ReadonlyArray; - readonly visible: ReadonlyArray; - }, - ProjectionStoreV2Error - > => + ): Effect.Effect => Effect.gen(function* () { const records = yield* getThreadRecords(threadId, ["runs", "attempts"]); const rows = yield* sql<{ @@ -5035,6 +5034,39 @@ export const layer: Layer.Layer = readTimelineIndex(threadId, new Set()).pipe(Effect.map((index) => index.visible)), ); + // Every history page needs the whole thread's index, so scrolling back through a long + // thread would rebuild it once per page. Entries are valid while the newest event of the + // thread and its fork sources is unchanged; a revision of 0 means no events to compare. + // A projection rebuild appends no events, so it needs a fresh store. + const readThreadRevision = yield* makeThreadRevisionReader; + const timelineIndexes = new Map< + ThreadId, + { readonly sequence: number; readonly index: TimelineIndex } + >(); + const TIMELINE_INDEX_CACHE_CAPACITY = 4; + const readCurrentTimelineIndex = (threadId: ThreadId) => + Effect.gen(function* () { + const sequence = yield* readThreadRevision([threadId]).pipe( + Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })), + ); + const cached = timelineIndexes.get(threadId); + if (cached !== undefined && cached.sequence === sequence && sequence > 0) { + timelineIndexes.delete(threadId); + timelineIndexes.set(threadId, cached); + return cached.index; + } + const index = yield* readTimelineIndex(threadId, new Set()); + timelineIndexes.delete(threadId); + if (sequence > 0) { + timelineIndexes.set(threadId, { sequence, index }); + if (timelineIndexes.size > TIMELINE_INDEX_CACHE_CAPACITY) { + const oldest = timelineIndexes.keys().next().value; + if (oldest !== undefined) timelineIndexes.delete(oldest); + } + } + return index; + }); + const getThreadHistoryPage: ProjectionStoreV2Shape["getThreadHistoryPage"] = ( threadId, cursor, @@ -5044,7 +5076,7 @@ export const layer: Layer.Layer = sql .withTransaction( Effect.gen(function* () { - const index = yield* readTimelineIndex(threadId, new Set()); + const index = yield* readCurrentTimelineIndex(threadId); if (index.records.thread.deletedAt !== null) return yield* new ProjectionStoreThreadNotFoundError({ threadId }); const revisions = yield* sql<{ @@ -5142,7 +5174,7 @@ export const layer: Layer.Layer = sql .withTransaction( Effect.gen(function* () { - const index = yield* readTimelineIndex(threadId, new Set()); + const index = yield* readCurrentTimelineIndex(threadId); const matching = index.visible .map((row, position) => ({ ...row, position })) .filter((row) => diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index a49fb3925cb55..73b2de5193a96 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -1003,4 +1003,37 @@ describe("V2 history materialization", () => { assert.deepEqual(sizes, [1, 1, 75, 3]); }).pipe(Effect.provide(layerTest)), ); + + it.effect("keeps cached timeline reads current as the thread and its fork source change", () => + Effect.gen(function* () { + const projection = yield* setup; + const parent = ThreadId.make("thread:parent"); + const parentRun = RunId.make("run:parent"); + yield* commit([thread(parent, projectId), run(parent, parentRun)]); + yield* putItems([{ ...item("p1", 1, "one", "assistant_message", parent), runId: parentRun }]); + const child = thread(threadId, projectId); + if (child.type !== "thread.created") return; + yield* commit([ + { + ...child, + id: EventId.make("child:fork"), + type: "thread.metadata-updated", + payload: { + ...child.payload, + forkedFrom: { type: "run", threadId: parent, runId: parentRun }, + }, + }, + ]); + yield* putItems([item("c1", 1, "local")]); + const totalItems = projection + .getTimelinePage(threadId, { view: "activity", limit: 100 }) + .pipe(Effect.map((page) => page.totalItems)); + const initial = yield* totalItems; + assert.equal(yield* totalItems, initial); + yield* putItems([item("c2", 2, "local again")]); + assert.equal(yield* totalItems, initial + 1); + yield* putItems([{ ...item("p2", 2, "two", "assistant_message", parent), runId: parentRun }]); + assert.equal(yield* totalItems, initial + 2); + }).pipe(Effect.provide(layerTest)), + ); }); diff --git a/apps/server/src/orchestration-v2/ThreadFind.ts b/apps/server/src/orchestration-v2/ThreadFind.ts index 6e48e0a651a97..695149633f22c 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.ts @@ -20,6 +20,7 @@ import { ProjectionStoreThreadNotFoundError, type ProjectionStoreV2Error, } from "./ProjectionStore.ts"; +import { makeThreadRevisionReader } from "./threadRevision.ts"; interface FindRow { readonly position: number; @@ -214,7 +215,6 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( }), ), ); - const encodeThreadIds = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(ThreadId))); const encodeSources = Schema.encodeEffect( Schema.fromJsonString(Schema.Array(Schema.Struct({ threadId: ThreadId, id: TurnItemId }))), ); @@ -282,25 +282,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* ( sourceThreadIds: readonly ThreadId[]; } >(); - const readRevision = Effect.fn("ThreadFind.readRevision")(function* ( - threadIds: readonly ThreadId[], - ) { - const sources = yield* encodeThreadIds(threadIds); - const revision = yield* sql<{ sequence: number }>` - WITH RECURSIVE source_threads(thread_id) AS ( - SELECT value FROM json_each(${sources}) - UNION - SELECT json_extract(t.payload_json, '$.forkedFrom.threadId') - FROM orchestration_v2_projection_threads t - JOIN source_threads s ON s.thread_id = t.thread_id - WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run' - ) - SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events - WHERE application_event_version = 2 AND aggregate_kind = 'thread' - AND stream_id IN (SELECT thread_id FROM source_threads) - `; - return revision[0]?.sequence ?? 0; - }); + const readRevision = yield* makeThreadRevisionReader; const snapshot = Effect.fn("ThreadFind.snapshot")( function* (input: OrchestrationV2SearchThreadInput) { const { threadId } = input; diff --git a/apps/server/src/orchestration-v2/threadRevision.ts b/apps/server/src/orchestration-v2/threadRevision.ts new file mode 100644 index 0000000000000..846d80e807201 --- /dev/null +++ b/apps/server/src/orchestration-v2/threadRevision.ts @@ -0,0 +1,30 @@ +import { ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; + +/** + * Latest event sequence across the given threads and every thread they fork + * from. Cached reads compare it with the value they were built at. + */ +export const makeThreadRevisionReader = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const encodeThreadIds = Schema.encodeEffect(Schema.fromJsonString(Schema.Array(ThreadId))); + return Effect.fn("ThreadFind.readRevision")(function* (threadIds: readonly ThreadId[]) { + const sources = yield* encodeThreadIds(threadIds); + const revision = yield* sql<{ sequence: number }>` + WITH RECURSIVE source_threads(thread_id) AS ( + SELECT value FROM json_each(${sources}) + UNION + SELECT json_extract(t.payload_json, '$.forkedFrom.threadId') + FROM orchestration_v2_projection_threads t + JOIN source_threads s ON s.thread_id = t.thread_id + WHERE json_extract(t.payload_json, '$.forkedFrom.type') = 'run' + ) + SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events + WHERE application_event_version = 2 AND aggregate_kind = 'thread' + AND stream_id IN (SELECT thread_id FROM source_threads) + `; + return revision[0]?.sequence ?? 0; + }); +}); From 820963755aa715e94855dd8478d3848ce24ab9ae Mon Sep 17 00:00:00 2001 From: Aaron Queen Date: Fri, 30 Oct 2026 13:39:19 -0600 Subject: [PATCH 2/4] fix(web): SSH-launched servers no longer show a manual update prompt (#23) The desktop app installs its own version on an SSH host and relaunches the server it started there whenever it connects, but that server reports no self-update path, so the update banner offered a relaunch command to run by hand. SSH connections to a standalone server now say the desktop app updates it when it reconnects, and the banner no longer counts them as needing a manual update. --- .../ServerUpdateAction.browser.test.tsx | 62 +++++++++++++++++++ .../web/src/components/ServerUpdateAction.tsx | 30 +++++++++ .../chat/useAutoBalanceUpdateBanner.tsx | 10 ++- .../sidebar/updateEverything.logic.test.ts | 3 + .../sidebar/updateEverything.logic.ts | 8 ++- .../components/sidebar/useUpdateEverything.ts | 3 + apps/web/src/connection/platform.ts | 3 + apps/web/src/state/desktopSshServerKind.ts | 60 ++++++++++++++++++ 8 files changed, 177 insertions(+), 2 deletions(-) create mode 100644 apps/web/src/state/desktopSshServerKind.ts diff --git a/apps/web/src/components/ServerUpdateAction.browser.test.tsx b/apps/web/src/components/ServerUpdateAction.browser.test.tsx index ed3a723652a56..96b22c3450e95 100644 --- a/apps/web/src/components/ServerUpdateAction.browser.test.tsx +++ b/apps/web/src/components/ServerUpdateAction.browser.test.tsx @@ -27,6 +27,19 @@ vi.mock(import("~/hooks/useCopyToClipboard"), async (original) => ({ isCopied: false, })) as never, })); +vi.mock(import("~/state/environments"), async (original) => ({ + ...(await original()), + useEnvironment: ((environmentId: EnvironmentId) => ({ + environmentId, + entry: { + target: { + _tag: environmentId.startsWith("env-ssh") + ? "SshConnectionTarget" + : "BearerConnectionTarget", + }, + }, + })) as never, +})); vi.mock(import("~/state/session"), async (original) => { const actual = await original(); return { @@ -58,6 +71,7 @@ vi.mock(import("./ui/toast"), async (original) => { import { resetConfirmDialogForTests } from "~/confirmDialog"; import { appAtomRegistry } from "~/rpc/atomRegistry"; +import { readStoredSshServerKinds, recordSshServerKind } from "~/state/desktopSshServerKind"; import { ServerUpdateAction, ServerUpdatesAction, @@ -171,6 +185,54 @@ describe("ServerUpdateAction", () => { }, ); + it.each(["button", "icon"] as const)( + "tells a server the desktop app launched over SSH that the desktop updates it (%s)", + async (appearance) => { + using _state = setup(); + recordSshServerKind("env-ssh-managed" as EnvironmentId, "managed"); + await using screen = await renderAction({ + environmentId: "env-ssh-managed" as EnvironmentId, + selfUpdate: null, + appearance, + }); + expect(button(screen)).toBeUndefined(); + const notice = + appearance === "icon" + ? screen.container.querySelector('[tabindex="0"]')?.getAttribute("aria-label") + : screen.container.textContent; + expect(notice).toContain("The desktop app updates this server when it reconnects."); + }, + ); + + it("restores the launch kind after a restart", () => { + window.localStorage.setItem( + "t3code:ssh-server-kinds:v1", + JSON.stringify({ "env-ssh-restored": "managed" }), + ); + expect(readStoredSshServerKinds()).toEqual({ "env-ssh-restored": "managed" }); + window.localStorage.removeItem("t3code:ssh-server-kinds:v1"); + }); + + it("still offers the relaunch command once the same server is paired by URL", async () => { + using _state = setup(); + recordSshServerKind("env-paired-by-url" as EnvironmentId, "managed"); + await using screen = await renderAction({ + environmentId: "env-paired-by-url" as EnvironmentId, + selfUpdate: null, + }); + expect(button(screen)).toBeDefined(); + }); + + it("still offers the relaunch command for an SSH server the desktop adopted rather than launched", async () => { + using _state = setup(); + recordSshServerKind("env-ssh-external" as EnvironmentId, "external"); + await using screen = await renderAction({ + environmentId: "env-ssh-external" as EnvironmentId, + selfUpdate: null, + }); + expect(button(screen)).toBeDefined(); + }); + it("disables the action once maintenance access is removed", async () => { using _state = setup(); await using screen = await renderAction(); diff --git a/apps/web/src/components/ServerUpdateAction.tsx b/apps/web/src/components/ServerUpdateAction.tsx index 141f139aefd7b..61606d386c9a6 100644 --- a/apps/web/src/components/ServerUpdateAction.tsx +++ b/apps/web/src/components/ServerUpdateAction.tsx @@ -23,6 +23,8 @@ import { type ComponentProps, useRef, useState } from "react"; import { requestConfirmDialog } from "~/confirmDialog"; import { useCopyToClipboard } from "~/hooks/useCopyToClipboard"; +import { useDesktopRelaunchesServer } from "~/state/desktopSshServerKind"; +import { useEnvironment } from "~/state/environments"; import { serverEnvironment, updateOutdatedServer } from "~/state/server"; import { appAtomRegistry } from "~/rpc/atomRegistry"; import { useAtomCommand } from "~/state/use-atom-command"; @@ -40,6 +42,8 @@ const UPDATE_STAGE_LABELS: Record = { }; const pendingUpdateEnvironmentIds = new Set(); +export const DESKTOP_RELAUNCH_NOTICE = "The desktop app updates this server when it reconnects."; + export function serverUpdateStageLabel(stage: ServerUpdateStage): string { return UPDATE_STAGE_LABELS[stage]; } @@ -250,6 +254,10 @@ export function ServerUpdateAction({ const sessionStateAtom = environmentSession.sessionStateAtom(environmentId); const canUpdate = canUpdateServer(useAtomValue(sessionStateAtom)); const update = useServerUpdate(); + const environment = useEnvironment(environmentId); + const desktopRelaunchesServer = useDesktopRelaunchesServer(); + const desktopRelaunched = + selfUpdate === null && environment !== null && desktopRelaunchesServer(environment); const { copyToClipboard } = useCopyToClipboard<{ command: string }>({ target: installation?.kind === "npm-global" ? "update command" : "relaunch command", onCopy: ({ command }) => { @@ -310,6 +318,28 @@ export function ServerUpdateAction({ return ; } + if (desktopRelaunched) { + return appearance === "icon" ? ( + + + } + > + + + {DESKTOP_RELAUNCH_NOTICE} + + ) : ( + {DESKTOP_RELAUNCH_NOTICE} + ); + } + const manualCommand = selfUpdate === null ? manualServerUpdateCommand(targetVersion, installation) : null; const actionLabel = diff --git a/apps/web/src/components/chat/useAutoBalanceUpdateBanner.tsx b/apps/web/src/components/chat/useAutoBalanceUpdateBanner.tsx index 60f93aa54b965..580b8480fc03e 100644 --- a/apps/web/src/components/chat/useAutoBalanceUpdateBanner.tsx +++ b/apps/web/src/components/chat/useAutoBalanceUpdateBanner.tsx @@ -2,6 +2,7 @@ import { useAtomValue } from "@effect/atom-react"; import type { ServerUpdateState } from "@t3tools/client-runtime/state/server"; import { useState } from "react"; +import { useDesktopRelaunchesServer } from "~/state/desktopSshServerKind"; import { environmentIdsKey } from "~/state/environmentAtomList"; import type { EnvironmentPresentation } from "~/state/environments"; import { environmentUpdateStatesAtom } from "~/state/server"; @@ -16,6 +17,7 @@ import { supportsDesktopAppUpdate, } from "~/versionSkew"; import { + DESKTOP_RELAUNCH_NOTICE, ServerUpdateAction, ServerUpdateProgress, ServerUpdatesAction, @@ -33,6 +35,7 @@ export function useAutoBalanceUpdateBanner( environmentIdsKey(environments.map(({ environmentId }) => environmentId)), ), ); + const desktopRelaunchesServer = useDesktopRelaunchesServer(); const states = environments.flatMap((environment, index) => { const state = updateStates[index]; return state ? [{ environment, state }] : []; @@ -65,6 +68,7 @@ export function useAutoBalanceUpdateBanner( targetVersion: state.status === "idle" ? mismatch!.clientVersion : state.targetVersion, connected: environment.connection.phase === "connected", remoteUpdate: selfUpdate !== null && (selfUpdate !== "desktop-managed" || desktopAppUpdate), + desktopRelaunched: selfUpdate === null && desktopRelaunchesServer(environment), state, dismissKey, }, @@ -74,7 +78,9 @@ export function useAutoBalanceUpdateBanner( const running = machines.filter((machine) => machine.state.status === "running").length; const failed = machines.filter((machine) => machine.state.status === "failed").length; - const manual = machines.filter((machine) => !machine.remoteUpdate).length; + const manual = machines.filter( + (machine) => !machine.remoteUpdate && !machine.desktopRelaunched, + ).length; const targets = machines.filter( (machine) => machine.connected && machine.remoteUpdate && machine.state.status !== "running", ); @@ -103,6 +109,8 @@ export function useAutoBalanceUpdateBanner(
{machine.serverLabel}
{machine.state.status !== "idle" ? ( + ) : machine.desktopRelaunched ? ( +
{DESKTOP_RELAUNCH_NOTICE}
) : !machine.remoteUpdate ? ( <>
Manual update required
diff --git a/apps/web/src/components/sidebar/updateEverything.logic.test.ts b/apps/web/src/components/sidebar/updateEverything.logic.test.ts index 62aef93572c4d..390da32b7669d 100644 --- a/apps/web/src/components/sidebar/updateEverything.logic.test.ts +++ b/apps/web/src/components/sidebar/updateEverything.logic.test.ts @@ -39,6 +39,7 @@ function machine( }, connected: true, hostedByThisApp: false, + desktopRelaunched: false, canMaintain: true, updateState: { status: "idle" }, ...overrides, @@ -67,6 +68,7 @@ describe("collectServerUpdateTargets", () => { machine("current", "0.0.50", "boot-service"), machine("current-offline", "0.0.50", "boot-service", { connected: false }), machine("manual", "0.0.40", undefined), + machine("ssh-managed", "0.0.40", undefined, { desktopRelaunched: true }), ], "0.0.50", true, @@ -83,6 +85,7 @@ describe("collectServerUpdateTargets", () => { ["no-scope", "no permission"], ["busy", "already updating"], ["manual", "needs a manual update"], + ["ssh-managed", "updates when the desktop app reconnects"], ]); }); diff --git a/apps/web/src/components/sidebar/updateEverything.logic.ts b/apps/web/src/components/sidebar/updateEverything.logic.ts index de21c535f517c..bdf2cbfe8e66e 100644 --- a/apps/web/src/components/sidebar/updateEverything.logic.ts +++ b/apps/web/src/components/sidebar/updateEverything.logic.ts @@ -24,6 +24,8 @@ export interface UpdateEverythingMachine { } | null; readonly connected: boolean; readonly hostedByThisApp: boolean; + /** The desktop app started this server over SSH and relaunches it on its own version at every connect. */ + readonly desktopRelaunched: boolean; readonly canMaintain: boolean; readonly updateState: ServerUpdateState; } @@ -101,7 +103,11 @@ export function collectServerUpdateTargets( const selfUpdate = resolveServerSelfUpdateCapability(config); const desktopAppUpdate = supportsDesktopAppUpdate(config); if (selfUpdate === null) { - skip("needs a manual update"); + skip( + machine.desktopRelaunched + ? "updates when the desktop app reconnects" + : "needs a manual update", + ); continue; } if (selfUpdate === "desktop-managed" && !desktopAppUpdate) { diff --git a/apps/web/src/components/sidebar/useUpdateEverything.ts b/apps/web/src/components/sidebar/useUpdateEverything.ts index d8ab3a5bce710..f77a4eb398478 100644 --- a/apps/web/src/components/sidebar/useUpdateEverything.ts +++ b/apps/web/src/components/sidebar/useUpdateEverything.ts @@ -4,6 +4,7 @@ import { PROVIDER_DISPLAY_NAMES } from "@t3tools/contracts"; import { APP_VERSION } from "~/branding"; import { isDesktopLocalConnectionTarget } from "~/connection/desktopLocal"; import { isElectron } from "~/env"; +import { useDesktopRelaunchesServer } from "~/state/desktopSshServerKind"; import { useEnvironments } from "~/state/environments"; import { useNpmReleasePublished } from "~/state/npmRelease"; import { environmentIdsKey } from "~/state/environmentAtomList"; @@ -56,6 +57,7 @@ export function useUpdateEverything( ) { const targetVersion = localUpdateVersion ?? APP_VERSION; const { environments } = useEnvironments(); + const desktopRelaunchesServer = useDesktopRelaunchesServer(); const environmentIds = environmentIdsKey(environments.map(({ environmentId }) => environmentId)); const sessionStates = useAtomValue(environmentSessionStatesAtom(environmentIds)); const updateStates = useAtomValue(environmentUpdateStatesAtom(environmentIds)); @@ -65,6 +67,7 @@ export function useUpdateEverything( serverConfig: environment.serverConfig, connected: environment.connection.phase === "connected", hostedByThisApp: isHostedByThisApp(environment), + desktopRelaunched: desktopRelaunchesServer(environment), canMaintain: canUpdateServer(sessionStates[index]!), updateState: updateStates[index]!, })); diff --git a/apps/web/src/connection/platform.ts b/apps/web/src/connection/platform.ts index e60ed172e9260..13ba8a91440d5 100644 --- a/apps/web/src/connection/platform.ts +++ b/apps/web/src/connection/platform.ts @@ -54,6 +54,7 @@ import { isHostedStaticApp } from "../hostedPairing"; import { isLocalEnvironmentDisabled } from "../localEnvironment"; import { randomUUID } from "../lib/utils"; import { appAtomRegistry } from "../rpc/atomRegistry"; +import { recordSshServerKind } from "../state/desktopSshServerKind"; import { acknowledgeRpcRequest, trackRpcRequestSent } from "../rpc/requestLatencyState"; import { desktopLocalConnectionId, @@ -252,6 +253,7 @@ export const provisionDesktopSshEnvironment = Effect.fn( }); }, }); + recordSshServerKind(descriptor.environmentId, bootstrap.remoteServerKind); return { environmentId: descriptor.environmentId, label: descriptor.label, @@ -291,6 +293,7 @@ export const prepareDesktopSshEnvironment = Effect.fn("web.connectionPlatform.ss try: () => bridge.fetchSshSessionState(bootstrap.httpBaseUrl, bearerToken), catch: sshPreparationError, }); + recordSshServerKind(input.expectedEnvironmentId, bootstrap.remoteServerKind); // Stored grants never widen; re-pair tokens missing newer standard scopes. const granted = session.permissions ?? session.scopes ?? []; if ( diff --git a/apps/web/src/state/desktopSshServerKind.ts b/apps/web/src/state/desktopSshServerKind.ts new file mode 100644 index 0000000000000..aad2a5bf972b8 --- /dev/null +++ b/apps/web/src/state/desktopSshServerKind.ts @@ -0,0 +1,60 @@ +import { useAtomValue } from "@effect/atom-react"; +import type { DesktopSshEnvironmentBootstrap, EnvironmentId } from "@t3tools/contracts"; +import { Atom } from "effect/reactivity"; +import * as Schema from "effect/Schema"; + +import { getLocalStorageItem, setLocalStorageItem } from "~/hooks/useLocalStorage"; +import { appAtomRegistry } from "~/rpc/atomRegistry"; +import type { EnvironmentPresentation } from "~/state/environments"; + +type SshServerKind = NonNullable; + +const STORAGE_KEY = "t3code:ssh-server-kinds:v1"; +const StoredKindsSchema = Schema.Record(Schema.String, Schema.Literals(["external", "managed"])); +type StoredKinds = typeof StoredKindsSchema.Type; + +// Cached server configs render before an SSH connection is prepared again, so +// the last kind seen must survive a restart. +export function readStoredSshServerKinds(): StoredKinds { + try { + return getLocalStorageItem(STORAGE_KEY, StoredKindsSchema) ?? {}; + } catch (error) { + console.error("Could not read SSH server kinds.", error); + return {}; + } +} + +const sshServerKindsAtom = Atom.make(readStoredSshServerKinds()).pipe( + Atom.keepAlive, + Atom.withLabel("desktop:ssh-server-kinds"), +); + +/** The desktop app relaunches a "managed" server on its own version at every connect; it leaves an "external" one alone. */ +export function recordSshServerKind( + environmentId: EnvironmentId, + kind: SshServerKind | undefined, +): void { + if (kind === undefined) return; + const kinds = appAtomRegistry.get(sshServerKindsAtom); + if (kinds[environmentId] === kind) return; + const next = { ...kinds, [environmentId]: kind }; + appAtomRegistry.set(sshServerKindsAtom, next); + try { + setLocalStorageItem(STORAGE_KEY, next, StoredKindsSchema); + } catch (error) { + console.error("Could not persist SSH server kinds.", error); + } +} + +/** + * Whether the desktop app started and relaunches this environment's server, + * so its updates need no manual step. Only an SSH connection qualifies: the + * same server paired by URL after its SSH environment was removed does not. + */ +export function useDesktopRelaunchesServer(): ( + environment: Pick, +) => boolean { + const kinds = useAtomValue(sshServerKindsAtom); + return ({ environmentId, entry }) => + entry.target._tag === "SshConnectionTarget" && kinds[environmentId] === "managed"; +} From 7761ba0ad779850076f64890e672b0e20d890563 Mon Sep 17 00:00:00 2001 From: Aaron Queen Date: Fri, 30 Oct 2026 15:59:47 -0600 Subject: [PATCH 3/4] perf(server): stop serializing terminal chunks just to measure them (#16) * perf(server): stop serializing terminal chunks just to measure them The terminal output window stringified every RPC chunk only to count its bytes, then the protocol serialized the same frame again for the wire. Windowed chunks carry terminal events, so charge them with the existing queuedSize model (envelope + payload bytes) instead; unrecognized values keep the exact JSON measurement. * fix(server): charge error and snapshot fields in the terminal byte budget Review finding on the size-estimate change: an error event was charged only the 128-byte envelope although its message is unbounded, so the window could retain eight large error chunks under its 64 KiB limit. queuedSize now charges every variable-size field (output data, error message, activity label, snapshot history/cwd/path/label), and the window fast path only applies to the known terminal event types; other shapes keep the exact JSON measurement. Adds a regression test where one 64 KiB error message fills the window. * chore(server): drop an effect-diagnostics directive that no longer applies * fix(server): charge thread and terminal ids in the terminal byte budget --- .../src/terminal/OutputProtocol.test.ts | 69 +++++++++++++++++++ apps/server/src/terminal/OutputProtocol.ts | 69 +++++++++++++++---- 2 files changed, 125 insertions(+), 13 deletions(-) diff --git a/apps/server/src/terminal/OutputProtocol.test.ts b/apps/server/src/terminal/OutputProtocol.test.ts index a804a6610b774..97d4f8c99a887 100644 --- a/apps/server/src/terminal/OutputProtocol.test.ts +++ b/apps/server/src/terminal/OutputProtocol.test.ts @@ -80,6 +80,75 @@ describe("terminal output window", () => { assert.equal((yield* Queue.take(responses))._tag, "Exit"); }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped), ); + + const bigChunkEvents = [ + { + name: "error messages", + event: { + threadId: "thread-1", + terminalId: "default", + type: "error" as const, + message: "e".repeat(64 * 1024), + }, + }, + { + name: "thread ids", + event: { + threadId: "t".repeat(64 * 1024), + terminalId: "default", + type: "output" as const, + data: "x", + }, + }, + ]; + + it.effect.each(bigChunkEvents)("charges $name against the pending-byte limit", ({ event }) => + Effect.gen(function* () { + const tag = WS_METHODS.terminalAttach; + const group = RpcGroup.make(Rpc.make(tag, { success: TerminalEventSchema, stream: true })); + const output = yield* Queue.unbounded>(); + const responses = yield* Queue.unbounded(); + const receive = yield* Deferred.make[0]>(); + const protocol = yield* RpcServer.Protocol.make((write) => + Effect.gen(function* () { + yield* Deferred.succeed(receive, write); + const serialization = yield* RpcSerialization.RpcSerialization; + return { + disconnects: yield* Queue.unbounded(), + send: (_clientId, response) => Queue.offer(responses, response), + end: () => Effect.void, + clientIds: Effect.succeed(new Set([0])), + initialMessage: Effect.succeedNone, + supportsAck: true, + supportsTransferables: false, + supportsSpanPropagation: false, + supportsNotifications: true, + codecFor: serialization.codecFor, + }; + }), + ); + yield* RpcServer.make(group).pipe( + Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))), + Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)), + Effect.forkScoped, + ); + const write = yield* Deferred.await(receive); + yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] }); + + yield* Queue.offer(output, event); + yield* TestClock.adjust(0); + assert.equal(yield* Queue.size(responses), 1); + yield* Queue.take(responses); + // One 64 KiB field fills the byte window on its own. + yield* Queue.offer(output, event); + yield* TestClock.adjust(0); + assert.equal(yield* Queue.size(responses), 0); + yield* write(0, { _tag: "Ack", requestId: "1" }); + assert.equal((yield* Queue.take(responses))._tag, "Chunk"); + yield* write(0, { _tag: "Interrupt", requestId: "1" }); + assert.equal((yield* Queue.take(responses))._tag, "Exit"); + }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped), + ); }); const output = (data: string): TerminalAttachStreamEvent => ({ diff --git a/apps/server/src/terminal/OutputProtocol.ts b/apps/server/src/terminal/OutputProtocol.ts index 1a8e93381aae2..d9fc13f72f182 100644 --- a/apps/server/src/terminal/OutputProtocol.ts +++ b/apps/server/src/terminal/OutputProtocol.ts @@ -8,10 +8,65 @@ import type { RpcServer } from "effect/rpc"; const MAX_PENDING_CHUNKS = 8; const MAX_PENDING_BYTES = 64 * 1024; +// Charge an envelope so tiny events cannot bypass the byte budget. +const EVENT_OVERHEAD = 128; const isFull = (sizes: number[]) => sizes.length >= MAX_PENDING_CHUNKS || sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES; +const byteLengthOf = (value: unknown) => (typeof value === "string" ? Buffer.byteLength(value) : 0); + +// Charges every variable-size field a terminal event can carry, so the byte +// budgets below cannot be bypassed by a large payload in an uncharged field. +const queuedSize = (event: TerminalEvent | TerminalAttachStreamEvent) => + EVENT_OVERHEAD + + ("threadId" in event ? byteLengthOf(event.threadId) + byteLengthOf(event.terminalId) : 0) + + (event.type === "output" + ? byteLengthOf(event.data) + : event.type === "error" + ? byteLengthOf(event.message) + : event.type === "activity" + ? byteLengthOf(event.label) + : "snapshot" in event + ? byteLengthOf(event.snapshot.threadId) + + byteLengthOf(event.snapshot.terminalId) + + byteLengthOf(event.snapshot.history) + + byteLengthOf(event.snapshot.cwd) + + byteLengthOf(event.snapshot.worktreePath) + + byteLengthOf(event.snapshot.label) + : 0); + +const TERMINAL_EVENT_TYPES = new Set([ + "started", + "output", + "exited", + "closed", + "error", + "cleared", + "restarted", + "activity", + "snapshot", +]); + +// Windowed chunks carry terminal events, so charge them the way the output +// queue below does instead of serializing the frame a second time just to +// measure it. Only the event shapes queuedSize measures take the fast path; +// anything else keeps the exact JSON measurement. +const chunkSize = (values: ReadonlyArray) => + values.reduce((total, value) => { + if (typeof value === "string") return total + EVENT_OVERHEAD + Buffer.byteLength(value); + if ( + typeof value === "object" && + value !== null && + "type" in value && + typeof value.type === "string" && + TERMINAL_EVENT_TYPES.has(value.type) + ) { + return total + queuedSize(value as TerminalEvent | TerminalAttachStreamEvent); + } + return total + Buffer.byteLength(JSON.stringify(value) ?? ""); + }, EVENT_OVERHEAD); + export function withTerminalOutputWindow( protocol: RpcServer.Protocol["Service"], ): RpcServer.Protocol["Service"] { @@ -50,9 +105,7 @@ export function withTerminalOutputWindow( } else if (response._tag === "Chunk") { const window = windows.get(`${clientId}:${response.requestId}`); if (window) { - // @effect-diagnostics-next-line preferSchemaOverJson:off - const size = Buffer.byteLength(JSON.stringify(response)); - window.push(size); + window.push(chunkSize(response.values)); if (!isFull(window)) { return send.pipe( Effect.andThen(() => @@ -71,8 +124,6 @@ export function withTerminalOutputWindow( } const MAX_QUEUED_OUTPUT = 8 * 1024 * 1024; -// Charge an envelope so tiny events cannot bypass the byte budget. -const EVENT_OVERHEAD = 128; type Listener = (event: A) => Effect.Effect; // Acquisition must register cleanup synchronously with installing the listener. @@ -81,14 +132,6 @@ type Subscribe = ( registerCleanup: (stop: () => void) => void, ) => Effect.Effect<() => void, E>; -const queuedSize = (event: TerminalEvent | TerminalAttachStreamEvent) => - EVENT_OVERHEAD + - (event.type === "output" - ? Buffer.byteLength(event.data) - : "snapshot" in event - ? Buffer.byteLength(event.snapshot.history) - : 0); - export function terminalOutputStream( subscribe: Subscribe, resync?: Subscribe, From e16be7de1609ead490db659bcea69456b9f96c76 Mon Sep 17 00:00:00 2001 From: Aaron Queen Date: Fri, 30 Oct 2026 18:47:51 -0600 Subject: [PATCH 4/4] test(server): type the terminal byte-budget cases with the attach stream schema (#27) --- apps/server/src/terminal/OutputProtocol.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/apps/server/src/terminal/OutputProtocol.test.ts b/apps/server/src/terminal/OutputProtocol.test.ts index 97d4f8c99a887..7872273d2863e 100644 --- a/apps/server/src/terminal/OutputProtocol.test.ts +++ b/apps/server/src/terminal/OutputProtocol.test.ts @@ -105,8 +105,11 @@ describe("terminal output window", () => { it.effect.each(bigChunkEvents)("charges $name against the pending-byte limit", ({ event }) => Effect.gen(function* () { const tag = WS_METHODS.terminalAttach; - const group = RpcGroup.make(Rpc.make(tag, { success: TerminalEventSchema, stream: true })); - const output = yield* Queue.unbounded>(); + const group = RpcGroup.make( + Rpc.make(tag, { success: TerminalAttachStreamEventSchema, stream: true }), + ); + const output = + yield* Queue.unbounded>(); const responses = yield* Queue.unbounded(); const receive = yield* Deferred.make[0]>(); const protocol = yield* RpcServer.Protocol.make((write) =>