diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b5a0f2cfb5556..a751b0b1c2f63 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -110,6 +110,7 @@ import { THREAD_HISTORY_INLINE_WORK_ITEM_LIMIT, THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, } from "./threadHistoryPaging.ts"; +import { makeThreadRevisionReader } from "./threadRevision.ts"; export class ProjectionStoreApplyEventError extends Schema.TaggedError()( "ProjectionStoreApplyEventError", @@ -5653,18 +5654,18 @@ 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, includePersistentOutput?: boolean, - ) => Effect.Effect< - { - readonly records: ProjectionRecords<"runs" | "attempts">; - readonly local: ReadonlyArray; - readonly visible: ReadonlyArray; - }, - ProjectionStoreV2Error - > = Effect.fn("ProjectionStore.readTimelineIndex")( + ) => Effect.Effect = Effect.fn( + "ProjectionStore.readTimelineIndex", + )( function* (threadId: ThreadId, seen: ReadonlySet, includePersistentOutput = false) { const records = yield* getThreadRecords(threadId, ["runs", "attempts"]); const rows = yield* sql<{ @@ -5800,6 +5801,43 @@ 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< + string, + { readonly sequence: number; readonly index: TimelineIndex } + >(); + const TIMELINE_INDEX_CACHE_CAPACITY = 4; + // The index holds attempt and persistent-output fields only when includePersistentOutput + // is set, so the two variants of a thread are cached separately. + const readCurrentTimelineIndex = Effect.fn("ProjectionStore.readCurrentTimelineIndex")( + function* (threadId: ThreadId, includePersistentOutput = false) { + const key = `${threadId}:${includePersistentOutput}`; + const sequence = yield* readThreadRevision([threadId]).pipe( + Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })), + ); + const cached = timelineIndexes.get(key); + if (cached !== undefined && cached.sequence === sequence && sequence > 0) { + timelineIndexes.delete(key); + timelineIndexes.set(key, cached); + return cached.index; + } + const index = yield* readTimelineIndex(threadId, new Set(), includePersistentOutput); + timelineIndexes.delete(key); + if (sequence > 0) { + timelineIndexes.set(key, { 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, @@ -5810,7 +5848,7 @@ export const layer: Layer.Layer = sql .withTransaction( Effect.gen(function* () { - const index = yield* readTimelineIndex(threadId, new Set(), workMode !== undefined); + const index = yield* readCurrentTimelineIndex(threadId, workMode !== undefined); if (index.records.thread.deletedAt !== null) return yield* new ProjectionStoreThreadNotFoundError({ threadId }); const revisions = yield* sql<{ @@ -5927,7 +5965,7 @@ export const layer: Layer.Layer = sql .withTransaction( Effect.gen(function* () { - const index = yield* readTimelineIndex(threadId, new Set()); + const index = yield* readCurrentTimelineIndex(threadId); const attachmentOwners = options.attachmentId === undefined ? undefined diff --git a/apps/server/src/orchestration-v2/ThreadFind.test.ts b/apps/server/src/orchestration-v2/ThreadFind.test.ts index 6793421a64333..681875709c1ec 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.test.ts @@ -822,4 +822,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 ead68680a21c9..c49e8cd1c3794 100644 --- a/apps/server/src/orchestration-v2/ThreadFind.ts +++ b/apps/server/src/orchestration-v2/ThreadFind.ts @@ -21,6 +21,7 @@ import { ProjectionStoreThreadNotFoundError, type ProjectionStoreV2Error, } from "./ProjectionStore.ts"; +import { makeThreadRevisionReader } from "./threadRevision.ts"; interface FindRow { readonly position: number; @@ -183,7 +184,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 }))), ); @@ -252,28 +252,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 - -- The planner otherwise walks every V2 event backwards from the newest, - -- which takes seconds for a thread that has been idle a long time. - INDEXED BY idx_orch_events_stream_sequence - 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..3e747d7896d9f --- /dev/null +++ b/apps/server/src/orchestration-v2/threadRevision.ts @@ -0,0 +1,33 @@ +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 + -- The planner otherwise walks every V2 event backwards from the newest, + -- which takes seconds for a thread that has been idle a long time. + INDEXED BY idx_orch_events_stream_sequence + WHERE application_event_version = 2 AND aggregate_kind = 'thread' + AND stream_id IN (SELECT thread_id FROM source_threads) + `; + return revision[0]?.sequence ?? 0; + }); +}); diff --git a/apps/server/src/terminal/OutputProtocol.test.ts b/apps/server/src/terminal/OutputProtocol.test.ts index a804a6610b774..7872273d2863e 100644 --- a/apps/server/src/terminal/OutputProtocol.test.ts +++ b/apps/server/src/terminal/OutputProtocol.test.ts @@ -80,6 +80,78 @@ 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: 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) => + 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, 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 32a2811aca189..aef41ec5e8a08 100644 --- a/apps/web/src/components/ServerUpdateAction.tsx +++ b/apps/web/src/components/ServerUpdateAction.tsx @@ -24,6 +24,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"; @@ -44,6 +46,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]; } @@ -275,6 +279,10 @@ export function ServerUpdateAction({ const canUpdate = useAtomValue(sessionStateAtom, canUpdateServer); const updatesDenial = managedUpdatesDenial(useManagedPolicy(environmentId)); 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 }) => { @@ -347,6 +355,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 58328dbc1eeeb..346aeba3df950 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", ); @@ -82,6 +84,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 cbbd3baa7c776..9d2c99ed87e34 100644 --- a/apps/web/src/components/sidebar/updateEverything.logic.ts +++ b/apps/web/src/components/sidebar/updateEverything.logic.ts @@ -27,6 +27,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; } @@ -105,7 +107,11 @@ export function collectServerUpdateTargets( } 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 05127d9885768..0975928c1ac5a 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 { environmentIdsKey } from "~/state/environmentAtomList"; import { environmentUpdateStatesAtom } from "~/state/server"; @@ -54,6 +55,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)); @@ -63,6 +65,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"; +}