diff --git a/apps/desktop/scripts/main-process-bundle.test.mjs b/apps/desktop/scripts/main-process-bundle.test.mjs index 678b62f58aaca..f7513a66d931b 100644 --- a/apps/desktop/scripts/main-process-bundle.test.mjs +++ b/apps/desktop/scripts/main-process-bundle.test.mjs @@ -2,6 +2,7 @@ import * as NodeChildProcess from "node:child_process"; import * as NodeFSP from "node:fs/promises"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; +import * as NodeURL from "node:url"; import * as NodeVM from "node:vm"; import { build } from "vite-plus/pack"; import { assert, it } from "vite-plus/test"; @@ -110,6 +111,34 @@ void import("./linux.ts").then(({ result }) => process.emit("ready", result));`, } }); +it("keeps the real desktop chunks from requiring the application entry", async () => { + // conf can evict the entry from the require cache. Lazy chunks must not require it. + const directory = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-desktop-chunks-")); + try { + assert.ok(Array.isArray(desktopConfig.pack)); + const packConfig = desktopConfig.pack.find((config) => + Object.values(config.entry ?? {}).includes("src/main.ts"), + ); + assert.ok(packConfig); + await build({ + ...packConfig, + config: false, + cwd: NodeURL.fileURLToPath(new URL("..", import.meta.url)), + outDir: directory, + sourcemap: false, + onSuccess: undefined, + logLevel: "silent", + }); + for (const filename of await NodeFSP.readdir(directory)) { + if (!filename.endsWith(".cjs") || filename === "main.cjs") continue; + const source = await NodeFSP.readFile(NodePath.join(directory, filename), "utf8"); + assert.notMatch(source, /require\(["'`]\.\/(?:application|main)\.cjs["'`]\)/, filename); + } + } finally { + await NodeFSP.rm(directory, { recursive: true, force: true }); + } +}, 120_000); + it("loads the emitted packaged boot entry and backend cache preload", async () => { const directory = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-desktop-boot-")); try { diff --git a/apps/desktop/src/app/DesktopObservability.test.ts b/apps/desktop/src/app/DesktopObservability.test.ts index ef462f19a29b0..5f027c5fd8084 100644 --- a/apps/desktop/src/app/DesktopObservability.test.ts +++ b/apps/desktop/src/app/DesktopObservability.test.ts @@ -6,6 +6,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; import * as HttpClient from "effect/http/HttpClient"; import * as HttpClientResponse from "effect/http/HttpClientResponse"; @@ -76,18 +77,29 @@ interface ExportedRequest { readonly body: string; } -const layerCollector = (requests: Array) => +/** + * Answers every export with a 200 and keeps what was posted for assertions. + * `stallFirst` never answers the first export. + */ +const layerCollector = ( + requests: Array, + options?: { readonly stallFirst?: boolean }, +) => Layer.succeed( HttpClient.HttpClient, HttpClient.make((request) => - Effect.sync(() => { + Effect.suspend(() => { requests.push({ url: request.url, headers: request.headers, body: request.body._tag === "Uint8Array" ? new TextDecoder().decode(request.body.body) : "", }); - return HttpClientResponse.fromWeb(request, new Response(null, { status: 200 })); + return options?.stallFirst === true && requests.length === 1 + ? Effect.never + : Effect.succeed( + HttpClientResponse.fromWeb(request, new Response(null, { status: 200 })), + ); }), ), ); @@ -384,7 +396,7 @@ describe("DesktopObservability", () => { ), ); - it.effect("exports main process log records to the configured logs endpoint", () => { + it.effect("exports log records after a stalled collector times out", () => { const requests: Array = []; return Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -394,6 +406,7 @@ describe("DesktopObservability", () => { const layerEnvironment = layerEnvironmentFor(baseDir, true, { T3CODE_OTLP_LOGS_URL: "https://collector.example.com/v1/logs", T3CODE_OTLP_HEADERS: "x-scope=desktop", + T3CODE_OTLP_EXPORT_INTERVAL_MS: "1000", }); const tracePath = yield* Effect.gen(function* () { const environment = yield* DesktopEnvironment.DesktopEnvironment; @@ -401,14 +414,21 @@ describe("DesktopObservability", () => { }).pipe(Effect.provide(layerEnvironment)); yield* Effect.scoped( - Effect.logInfo("desktop log export").pipe( - Effect.withSpan("desktop-log-export-test"), + Effect.gen(function* () { + yield* Effect.logInfo("before the stall"); + for (let second = 0; second < 15; second += 1) yield* TestClock.adjust("1 second"); + yield* Effect.logInfo("desktop log export").pipe( + Effect.withSpan("desktop-log-export-test"), + ); + for (let second = 0; second < 5; second += 1) yield* TestClock.adjust("1 second"); + // Before shutdown, whose flush could hide a stuck exporter. + assert.isTrue(requests.some((request) => request.body.includes("desktop log export"))); + }).pipe( Effect.provide(DesktopObservability.layer.pipe(Layer.provideMerge(layerEnvironment))), ), ); - assert.lengthOf(requests, 1); - const [request] = requests; + const request = requests.find((value) => value.body.includes("desktop log export")); assert.strictEqual(request?.url, "https://collector.example.com/v1/logs"); assert.include(request?.body ?? "", "desktop log export"); assert.include(request?.body ?? "", "service.runtime"); @@ -424,7 +444,13 @@ describe("DesktopObservability", () => { assert.lengthOf(record?.events ?? [], 0); }).pipe( Effect.scoped, - Effect.provide(Layer.mergeAll(NodeServices.layer, layerCollector(requests), layerEmptyEnv)), + Effect.provide( + Layer.mergeAll( + NodeServices.layer, + layerCollector(requests, { stallFirst: true }), + layerEmptyEnv, + ), + ), ); }); diff --git a/apps/desktop/src/app/DesktopObservability.ts b/apps/desktop/src/app/DesktopObservability.ts index 299db7019dd7a..6660ac7b83cc4 100644 --- a/apps/desktop/src/app/DesktopObservability.ts +++ b/apps/desktop/src/app/DesktopObservability.ts @@ -27,6 +27,8 @@ import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as SynchronizedRef from "effect/SynchronizedRef"; import * as Tracer from "effect/Tracer"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientError from "effect/http/HttpClientError"; import { OtlpExporter, OtlpLogger, OtlpTracer } from "effect/observability"; import * as DesktopEnvironment from "./DesktopEnvironment.ts"; @@ -591,6 +593,33 @@ const layerBackendOutputLogFactory = Layer.effect( }), ); +// Bound both headers and body so a stalled collector cannot stop later exports. +const withOtlpTimeout = (client: HttpClient.HttpClient) => + client.pipe( + HttpClient.transform((response, request) => + response.pipe( + Effect.tap((received) => received.arrayBuffer), + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => + Effect.fail( + new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ + request, + description: "OTLP export timed out", + }), + }), + ), + }), + ), + ), + ); + +/** + * Logs and traces for the main process, assembled together because they share + * one read of the environment and Settings, and because a process gets exactly + * one logger set. + */ const layerTelemetry = Layer.unwrap( Effect.gen(function* () { const environment = yield* DesktopEnvironment.DesktopEnvironment; @@ -678,7 +707,11 @@ const layerTelemetry = Layer.unwrap( ); return layerOtelWarnings.pipe( - Layer.provideMerge(Layer.mergeAll(layerLogger, layerTracer)), + Layer.provideMerge( + Layer.mergeAll(layerLogger, layerTracer).pipe( + Layer.updateService(HttpClient.HttpClient, withOtlpTimeout), + ), + ), Layer.provide(OtelEnvironment.layerResourceAttributes(endpoints.resourceAttributes)), ); }), diff --git a/apps/desktop/src/main.ts b/apps/desktop/src/main.ts index 8d183a63e845e..85548a4f2da75 100644 --- a/apps/desktop/src/main.ts +++ b/apps/desktop/src/main.ts @@ -227,9 +227,10 @@ const layerDesktopClerk = DesktopClerk.layer.pipe( Layer.provideMerge(ElectronApp.layer), ); +// node:http bypasses environment proxies and keeps undici out of the main bundle. const layerDesktopApplicationRuntime = layerDesktopApplication.pipe( Layer.provideMerge(NodeServices.layer), - Layer.provideMerge(NodeHttpClient.layerUndici), + Layer.provideMerge(NodeHttpClient.layerNodeHttp), Layer.provideMerge(NetService.layer), Layer.provideMerge(layerElectron), ); diff --git a/apps/desktop/src/preview/Manager.ts b/apps/desktop/src/preview/Manager.ts index b4eca14575e79..8b88df29727be 100644 --- a/apps/desktop/src/preview/Manager.ts +++ b/apps/desktop/src/preview/Manager.ts @@ -587,6 +587,7 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function { readonly semaphore: Semaphore.Semaphore; users: number } >(); const tabLifecycleGenerations = new Map(); + let nextTabLifecycleGeneration = 0; const attempt = (errorContext: PreviewOperationContext, evaluate: () => A) => Effect.try({ @@ -1685,7 +1686,8 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function }, ); if (result.created) { - tabLifecycleGenerations.set(tabId, (tabLifecycleGenerations.get(tabId) ?? 0) + 1); + // Generations remain unique when a closed tab id is reused. + tabLifecycleGenerations.set(tabId, ++nextTabLifecycleGeneration); } yield* emit(tabId, result.state); return result.state; @@ -1724,6 +1726,7 @@ const makeNativeOperations = Effect.fn("PreviewManager.makeOperations")(function ] as const; }); if (Option.isNone(tab)) return; + tabLifecycleGenerations.delete(tabId); const closedTab = tab.value; if (closedTab.webContentsId != null) { yield* Effect.all( diff --git a/apps/desktop/src/snapShot/LinuxSnapShot.ts b/apps/desktop/src/snapShot/LinuxSnapShot.ts index 8603232e7e9fc..38cede44d4028 100644 --- a/apps/desktop/src/snapShot/LinuxSnapShot.ts +++ b/apps/desktop/src/snapShot/LinuxSnapShot.ts @@ -19,6 +19,9 @@ import { isKdeCaptureSession, type KdeCapturePaths } from "./KdeSnapShot.ts"; import { isGnomeCaptureSession, readPortalPng, resizeLinuxCapture } from "./linuxCaptureSession.ts"; export { readPortalPng, resizeLinuxCapture } from "./linuxCaptureSession.ts"; import { isHyprlandCaptureSession, type HyprlandCapturePaths } from "./HyprlandSnapShot.ts"; +// The entry chunk also imports NiriSnapShot, so a lazy import of it would +// require application.cjs again, and conf evicts that from the require cache. +import { captureNiriWindow, checkNiriCaptureSupport, niriSocketPath } from "./NiriSnapShot.ts"; const PORTAL = "org.freedesktop.portal.Desktop"; const PORTAL_PATH = "/org/freedesktop/portal/desktop"; @@ -365,7 +368,6 @@ export class LinuxCaptureConnection { export async function getLinuxCaptureSupport(appId: string) { if (isHyprlandCaptureSession()) return { linuxBackend: "hyprland" as const, linuxFeedbackAvailable: false }; - const { niriSocketPath, checkNiriCaptureSupport } = await import("./NiriSnapShot.ts"); const niri = niriSocketPath(); if (niri) { await checkNiriCaptureSupport(niri); @@ -392,7 +394,6 @@ export async function captureLinuxWindow( const { captureHyprlandWindow } = await import("./HyprlandSnapShot.ts"); return captureHyprlandWindow(hyprlandPaths, options); } - const { niriSocketPath, captureNiriWindow } = await import("./NiriSnapShot.ts"); const niri = niriSocketPath(); if (niri) return captureNiriWindow(niri); const connection = new LinuxCaptureConnection(); diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index 2107d9305e936..e0ba8e9ede222 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -1,6 +1,9 @@ +import * as Schema from "effect/Schema"; import { ORCHESTRATION_CACHE_SCHEMA_VERSION, ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, + StoredOrchestrationShellSnapshot, + StoredOrchestrationThreadSnapshot, } from "@t3tools/client-runtime/platform"; import { CommandId, @@ -27,6 +30,10 @@ const PROJECT_ID = ProjectId.make("project-1"); const THREAD_ID = ThreadId.make("thread-1"); const PROVIDER_INSTANCE_ID = ProviderInstanceId.make("codex"); const NOW = DateTime.makeUnsafe("2026-07-29T12:00:00.000Z"); +// How a bundle without the deferred-thread envelope decodes a cached thread. +const decodePlainStoredThread = Schema.decodeUnknownExit( + Schema.fromJsonString(StoredOrchestrationThreadSnapshot), +); const SHELL_SNAPSHOT: OrchestrationV2ShellSnapshot = { schemaVersion: 1, snapshotSequence: 3, @@ -246,6 +253,36 @@ describe("mobile SQLite environment cache store", () => { expect(memory.schemaVersions.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))).toBe( ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, ); + + // An older bundle reads deferred checkpoints' empty file lists as "no + // changes", so it must not be able to decode their entry. + yield* store.saveThread(ENVIRONMENT_ID, { + ...THREAD_SNAPSHOT, + projection: { + ...THREAD_SNAPSHOT.projection, + checkpoints: [ + { + id: "checkpoint-deferred" as never, + threadId: THREAD_ID, + scopeId: "scope" as never, + runId: null, + nodeId: "node" as never, + parentCheckpointId: null, + ordinalWithinScope: 1, + appRunOrdinal: 1, + ref: "refs/t3/deferred" as never, + status: "ready", + files: [], + deferredFileCount: 2, + capturedAt: NOW, + }, + ], + }, + }); + const deferredPayload = memory.values.get(cacheId(ENVIRONMENT_ID, "thread", THREAD_ID))!; + expect(decodePlainStoredThread(deferredPayload)._tag).toBe("Failure"); + const deferred = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); + expect(deferred.projection.checkpoints[0]?.deferredFileCount).toBe(2); }), ); diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index 8df7fc104df26..35fe3b1124901 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -33,12 +33,24 @@ const StoredVcsRefs = Schema.Struct({ refs: VcsListRefsResult, }); -const decodeStoredThreadSnapshot = Schema.decodeUnknownEffect( - Schema.fromJsonString(StoredOrchestrationThreadSnapshot), -); -const encodeStoredThreadSnapshot = Schema.encodeEffect( - Schema.fromJsonString(StoredOrchestrationThreadSnapshot), +// Deferred checkpoints have empty file lists that an older bundle, after an +// OTA rollback, would show as turns with no changes. Their entries are +// wrapped, so an older bundle fails to decode them, drops them and loads live. +const StoredThreadCache = Schema.Union([ + StoredOrchestrationThreadSnapshot, + Schema.Struct({ + format: Schema.Literal("deferred-thread-1"), + stored: StoredOrchestrationThreadSnapshot, + }), +]); +const decodeStoredThreadCache = Schema.decodeUnknownEffect( + Schema.fromJsonString(StoredThreadCache), ); +const decodeStoredThreadSnapshot = (raw: string) => + decodeStoredThreadCache(raw).pipe( + Effect.map((cached) => ("format" in cached ? cached.stored : cached)), + ); +const encodeStoredThreadSnapshot = Schema.encodeEffect(Schema.fromJsonString(StoredThreadCache)); const decodeStoredServerConfig = Schema.decodeUnknownEffect( Schema.fromJsonString(StoredServerConfig), ); @@ -146,12 +158,18 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { ), saveThread: Effect.fn("MobileEnvironmentCache.saveThread")(function* (environmentId, snapshot) { const threadId = snapshot.projection.thread.id; - const payload = yield* encodeStoredThreadSnapshot({ + const stored = { schemaVersion: ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION, environmentId, threadId, snapshot, - }).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause))); + }; + const deferred = snapshot.projection.checkpoints.some( + (checkpoint) => checkpoint.deferredFileCount !== undefined, + ); + const payload = yield* encodeStoredThreadSnapshot( + deferred ? { format: "deferred-thread-1", stored } : stored, + ).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause))); yield* database .saveCache( environmentId, diff --git a/apps/mobile/src/features/review/reviewModel.ts b/apps/mobile/src/features/review/reviewModel.ts index 7b82f02d006eb..88d34c4192058 100644 --- a/apps/mobile/src/features/review/reviewModel.ts +++ b/apps/mobile/src/features/review/reviewModel.ts @@ -99,7 +99,7 @@ function checkpointTitle(checkpoint: ThreadCheckpointSummary): string { } function checkpointSubtitle(checkpoint: ThreadCheckpointSummary): string { - const fileCount = checkpoint.files.length; + const fileCount = checkpoint.deferredFileCount ?? checkpoint.files.length; if (checkpoint.status !== "ready") { return `Diff ${checkpoint.status}`; } diff --git a/apps/server/src/bin.ts b/apps/server/src/bin.ts index c5a2cb18fd149..eacac6ee96088 100644 --- a/apps/server/src/bin.ts +++ b/apps/server/src/bin.ts @@ -35,6 +35,9 @@ if ( } if (process.argv.length === 3 && command === "--version") { process.stdout.write(`t3 v${packageJson.version}\n`); + } else if (command === "--bootstrap-fd" || command?.startsWith("--bootstrap-fd=")) { + const { runLightCli } = await import("./binLightCli.ts"); + await runLightCli("desktop"); } else if ( command === "pair" || command === "auth" || diff --git a/apps/server/src/binLightCli.ts b/apps/server/src/binLightCli.ts index 86db7f56039ee..c0d43d31b99dd 100644 --- a/apps/server/src/binLightCli.ts +++ b/apps/server/src/binLightCli.ts @@ -8,8 +8,22 @@ import packageJson from "../package.json" with { type: "json" }; import { sharedServerCommandFlags } from "./cli/config.ts"; export async function runLightCli( - command: "pair" | "auth" | "service" | "project" | "serve" | "start", + command: "pair" | "auth" | "service" | "project" | "serve" | "start" | "desktop", ) { + const root = Command.make("t3", { ...sharedServerCommandFlags }).pipe( + Command.withDescription("Run the T3 Code server."), + ); + if (command === "desktop") { + const { runDefaultServerCommand } = await import("./cli/server.ts"); + Command.run(root.pipe(Command.withHandler(runDefaultServerCommand)), { + version: packageJson.version, + }).pipe( + Effect.scoped, + Effect.provide(Layer.mergeAll(NodeServices.layer, NetService.layer)), + NodeRuntime.runMain, + ); + return; + } const selected = command === "pair" ? (await import("./cli/pair.ts")).pairCommand @@ -22,10 +36,7 @@ export async function runLightCli( : command === "serve" ? (await import("./cli/server.ts")).serveCommand : (await import("./cli/server.ts")).startCommand; - const cli = Command.make("t3", { ...sharedServerCommandFlags }).pipe( - Command.withDescription("Run the T3 Code server."), - Command.withSubcommands([selected]), - ); + const cli = root.pipe(Command.withSubcommands([selected])); Command.run(cli, { version: packageJson.version }).pipe( Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, NetService.layer)), diff --git a/apps/server/src/bootstrap.test.ts b/apps/server/src/bootstrap.test.ts index 0b14c7dd1e238..cc7faec574478 100644 --- a/apps/server/src/bootstrap.test.ts +++ b/apps/server/src/bootstrap.test.ts @@ -56,6 +56,54 @@ vi.mock("node:fs", async (importOriginal) => { }); const windowsHost = HostProcess.Platform.defaultValue() === "win32"; + +it.effect( + "closes inherited Electron descriptors while preserving explicit channels and Node fds", + () => + Effect.gen(function* () { + if ((yield* HostProcess.Platform) !== "linux") return; + const script = ` + import fs from 'node:fs'; import { syncBuiltinESMExports } from 'node:module'; + import * as Effect from ${JSON.stringify(import.meta.resolve("effect/Effect"))}; + import * as HostProcess from ${JSON.stringify(import.meta.resolve("@t3tools/shared/HostProcess"))}; + import { closeInheritedElectronDescriptors } from ${JSON.stringify(new URL("./bootstrap.ts", import.meta.url).href)}; + const ownFd = fs.openSync('/dev/null', 'r'), readFileSync = fs.readFileSync; + // A parent's POSIX lock metadata is absent from its inherited fd in the child. + fs.readFileSync = (path, ...args) => readFileSync(path, ...args) + + (path === '/proc/' + process.ppid + '/fdinfo/64' ? 'lock:\\t1: POSIX ADVISORY WRITE parent 0 EOF\\n' : ''); + syncBuiltinESMExports(); + await Effect.runPromise(closeInheritedElectronDescriptors([3]).pipe( + Effect.provideService(HostProcess.Platform, "linux"), + Effect.provideService(HostProcess.Environment, { ELECTRON_RUN_AS_NODE: "1" }), + )); + let inheritedClosed = false; + try { fs.fstatSync(64); } catch (error) { inheritedClosed = error.code === 'EBADF'; } + fs.fstatSync(ownFd); fs.fstatSync(65); fs.writeSync(3, 'preserved'); + console.log(inheritedClosed); + `; + const result = NodeChildProcess.spawnSync( + "bash", + [ + "-c", + // fd 65 opens after the fork, like Electron's native descriptors. + 'exec 64 { try { diff --git a/apps/server/src/bootstrap.ts b/apps/server/src/bootstrap.ts index 708a8fd9119bf..55567388834cf 100644 --- a/apps/server/src/bootstrap.ts +++ b/apps/server/src/bootstrap.ts @@ -150,6 +150,47 @@ const isUnavailableBootstrapFdError = Predicate.compose( (_) => _.code === "EBADF" || _.code === "ENOENT", ); +/** + * Chromium leaves inherited descriptors without close-on-exec in Electron's + * Node children. Electron also opens native descriptors without that flag, + * so close only descriptors whose information still matches the parent. + * Run after reading the envelope, before services or the spawn host start. + */ +export const closeInheritedElectronDescriptors = Effect.fn("closeInheritedElectronDescriptors")( + function* (preservedFds: ReadonlyArray) { + const platform = yield* HostProcess.Platform; + const environment = yield* HostProcess.Environment; + if (platform !== "linux" || environment.ELECTRON_RUN_AS_NODE !== "1") return; + const preserved = new Set([0, 1, 2, ...preservedFds]); + yield* Effect.sync(() => { + let descriptors: string[]; + let parentPid: string | undefined; + try { + descriptors = NodeFS.readdirSync("/proc/self/fd"); + parentPid = /^PPid:\s+(\d+)$/m.exec(NodeFS.readFileSync("/proc/self/status", "utf8"))?.[1]; + } catch { + return; + } + if (parentPid === undefined) return; + for (const descriptor of descriptors) { + const fd = Number(descriptor); + if (preserved.has(fd)) continue; + try { + const info = NodeFS.readFileSync(`/proc/self/fdinfo/${fd}`, "utf8"); + const flags = /^flags:\s+([0-7]+)$/m.exec(info)?.[1]; + if (flags === undefined || (Number.parseInt(flags, 8) & 0o2000000) !== 0) continue; + // POSIX locks belong to the parent process and are absent in child fdinfo. + const parentInfo = NodeFS.readFileSync(`/proc/${parentPid}/fdinfo/${fd}`, "utf8"); + if (parentInfo.replace(/^lock:.*\n/gm, "") !== info.replace(/^lock:.*\n/gm, "")) continue; + NodeFS.closeSync(fd); + } catch { + // A descriptor may already have closed; unavailable procfs is harmless. + } + } + }); + }, +); + const isFdReady = (fd: number) => Effect.try({ try: () => NodeFS.fstatSync(fd), diff --git a/apps/server/src/cli/config.ts b/apps/server/src/cli/config.ts index f9b8730b7d509..5b8e21dd57b00 100644 --- a/apps/server/src/cli/config.ts +++ b/apps/server/src/cli/config.ts @@ -21,7 +21,7 @@ import * as SchemaTransformation from "effect/SchemaTransformation"; import { Argument, Flag } from "effect/cli"; import * as CliError from "effect/cli/CliError"; -import { readBootstrapEnvelope } from "../bootstrap.ts"; +import { closeInheritedElectronDescriptors, readBootstrapEnvelope } from "../bootstrap.ts"; import * as ServerConfig from "../config.ts"; import { expandHomePath, resolveBaseDir } from "../os-jank.ts"; import { isProcessAlive, readPersistedServerRuntimeState } from "../serverRuntimeState.ts"; @@ -283,6 +283,15 @@ export const resolveServerConfig = ( ? yield* readBootstrapEnvelope(DesktopBackendBootstrap, bootstrapFd) : Option.none(); const bootstrap = Option.getOrUndefined(bootstrapEnvelope); + if (bootstrap !== undefined) { + yield* closeInheritedElectronDescriptors([ + bootstrapFd, + bootstrap.desktopTelemetryFd, + bootstrap.desktopTelemetryControlFd, + bootstrap.desktopBrowserFd, + bootstrap.desktopBrowserControlFd, + ]); + } const mode: ServerConfig.RuntimeMode = Option.getOrElse( resolveOptionPrecedence( diff --git a/apps/server/src/orchestration-v2/CheckpointService.test.ts b/apps/server/src/orchestration-v2/CheckpointService.test.ts index 3bb6d0be84528..1d7b73c07688e 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.test.ts @@ -18,7 +18,7 @@ import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as CheckpointService from "./CheckpointService.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; -it.effect.each([false, true, "interrupt"] as const)( +it.effect.each([false, true, "interrupt", "no-git", "probe-error", "probe-interrupt"] as const)( "materializes baseline, lookup fails=%s", (lookupFails) => { const scope: OrchestrationV2CheckpointScope = { @@ -48,14 +48,27 @@ it.effect.each([false, true, "interrupt"] as const)( ) : Effect.succeed(true), ); + const captureCheckpoint = vi.fn(() => Effect.void); const layerTest = CheckpointService.layer.pipe( Layer.provide( Layer.mergeAll( IdAllocator.layer, Layer.mock(CheckpointStore.CheckpointStore)({ - isGitRepository: () => Effect.succeed(true), + isGitRepository: () => + lookupFails === "probe-interrupt" + ? Effect.interrupt + : lookupFails === "probe-error" + ? Effect.fail( + new VcsProcessTimeoutError({ + operation: "probe", + command: "git", + cwd: scope.cwd, + timeoutMs: 1, + }), + ) + : Effect.succeed(lookupFails !== "no-git"), hasCheckpointRef, - captureCheckpoint: () => Effect.void, + captureCheckpoint, }), ), ), @@ -64,6 +77,15 @@ it.effect.each([false, true, "interrupt"] as const)( return Effect.gen(function* () { const checkpoints = yield* CheckpointService.CheckpointServiceV2; + if (["no-git", "probe-error", "probe-interrupt"].includes(String(lookupFails))) { + const exit = yield* Effect.exit( + checkpoints.captureBaseline({ scope, ordinalWithinScope: 0 }), + ); + if (lookupFails === "probe-interrupt") assert.isTrue(Exit.hasInterrupts(exit)); + else assert.isTrue(Exit.isSuccess(exit)); + assert.equal(captureCheckpoint.mock.calls.length, 0); + return; + } if (lookupFails === "interrupt") { const exit = yield* Effect.exit( checkpoints.materializeBaselineCheckpoint({ scope, ordinalWithinScope: 2 }), diff --git a/apps/server/src/orchestration-v2/CheckpointService.ts b/apps/server/src/orchestration-v2/CheckpointService.ts index b49da7740b82c..9698937dced16 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.ts @@ -268,6 +268,8 @@ export const layer: Layer.Layer< withWorkspaceLock( input.scope.cwd, Effect.gen(function* () { + // A failed probe (no Git installed, a timeout) means no baseline, + // the same answer the turn-end capture gets. if (!(yield* isGitCheckpointable(input.scope.cwd))) { return; } diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a150d4d51ddb6..8c21be98b69a8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -308,6 +308,7 @@ export interface OrchestratorV2Shape { cursor: string, throughEntryId?: string, conversationOnly?: boolean, + includeSupport?: boolean, ) => Effect.Effect; readonly getTimelinePage: ( threadId: ThreadId, @@ -11402,9 +11403,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (cause) => new OrchestratorProjectionError({ threadId: input.threadId, cause }), ), ), - getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly, includeSupport) => projectionStore - .getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly) + .getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly, includeSupport) .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))), getTimelinePage: (threadId, options) => projectionStore diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 2d0da51502981..fc264faf589ed 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1165,6 +1165,15 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ]; startIds.push(rows[1]!.turn_item_id); yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`; + yield* sql`INSERT INTO orchestration_v2_projection_checkpoints VALUES ( + ${`checkpoint:${run}`}, ${threadId}, ${`scope:${run}`}, ${runId}, ${`node:notification-turn-pages:${run}`}, + NULL, 1, ${run}, 'ready', ${nowIso}, json_object( + 'id', ${`checkpoint:${run}`}, 'threadId', ${threadId}, 'scopeId', ${`scope:${run}`}, + 'runId', ${runId}, 'nodeId', ${`node:notification-turn-pages:${run}`}, 'parentCheckpointId', NULL, + 'ordinalWithinScope', 1, 'appRunOrdinal', ${run}, 'ref', ${`refs/t3/notification-turn-pages/${run}`}, + 'status', 'ready', 'capturedAt', ${nowIso}, + 'files', json('[{"path":"a.ts","kind":"modified","additions":1,"deletions":0}]')) + )`; } const { statements, tracer } = traceSqlStatements(); @@ -1215,6 +1224,50 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { cursor = page.nextCursor; } assert.deepEqual(loaded, allIds); + // Exercise the same user/agent input cohorts with this existing run fixture. + yield* sql`UPDATE orchestration_v2_projection_turn_items + SET type = 'user_message', payload_json = json_set(payload_json, + '$.type', 'user_message', '$.createdBy', 'agent', '$.creationSource', 'server', + '$.inputIntent', 'queued_turn', '$.messageId', turn_item_id, '$.text', 'wake', '$.attachments', json('[]')) + WHERE turn_item_id IN (SELECT value FROM json_each(${encodeUnknownJsonString(startIds)})) AND type = 'notification'`; + const viewport = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 77, + deferHistoricalSupport: true, + }); + const boundedViewport = buildBoundedThreadProjection({ ...viewport, viewportWindow: true }); + assert.deepEqual( + boundedViewport.projection.checkpoints + .map((checkpoint) => checkpoint.appRunOrdinal) + .toSorted((a, b) => Number(a) - Number(b)), + Array.from({ length: 152 }, (_, index) => index + 261), + ); + assert.deepInclude( + boundedViewport.projection.checkpoints.find( + (checkpoint) => checkpoint.appRunOrdinal === 393, + ), + { + files: [], + deferredFileCount: 1, + }, + ); + assert.lengthOf( + boundedViewport.projection.checkpoints.find( + (checkpoint) => checkpoint.appRunOrdinal === 412, + )!.files, + 1, + ); + // No user turn in the latest 152 starts: only the viewport's checkpoint stays listed. + yield* sql`UPDATE orchestration_v2_projection_turn_items + SET payload_json = json_set(payload_json, '$.createdBy', 'agent') + WHERE thread_id = ${threadId} AND type = 'user_message'`; + const agent = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 1, + deferHistoricalSupport: true, + }); + assert.deepEqual( + agent.projection.checkpoints.map((checkpoint) => checkpoint.appRunOrdinal), + [412], + ); }), ); @@ -1443,6 +1496,25 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); assert.lengthOf(sharedNodeSnapshot.projection.visibleTurnItems, 76); assert.deepEqual(sharedNodeSnapshot.projection.nodes, snapshot.projection.nodes); + + const promptId = TurnItemId.make("item:bounded-node-history:prompt"); + yield* sql`INSERT INTO orchestration_v2_projection_turn_items + SELECT ${promptId}, thread_id, run_id, ${rootNodeId}, provider_thread_id, + provider_turn_id, parent_item_id, 0, 'user_message', status, updated_at, + json_set(payload_json, '$.id', ${promptId}, '$.nodeId', ${rootNodeId}, '$.ordinal', 0, + '$.type', 'user_message', '$.messageId', 'prompt', '$.text', 'Start the long run', + '$.attachments', json('[]'), '$.inputIntent', 'turn_start', '$.createdBy', 'user', '$.creationSource', 'web') + FROM orchestration_v2_projection_turn_items WHERE turn_item_id = ${duplicateItemId}`; + const viewport = yield* projectionStore.getThreadSnapshotWindow(threadId, { + rowLimit: 75, + deferHistoricalSupport: true, + }); + assert.strictEqual(viewport.projection.visibleTurnItems[0]?.sourceItemId, promptId); + assert.lengthOf(viewport.projection.visibleTurnItems, 76); + const legacy = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 75 }); + assert.isFalse( + legacy.projection.visibleTurnItems.some((row) => row.sourceItemId === promptId), + ); }), ); @@ -5837,12 +5909,32 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { yield* sql`UPDATE orchestration_v2_projection_threads SET payload_json = json_set(payload_json, '$.pullRequests', json('[]')) WHERE thread_id = ${threadId}`; + yield* store.apply({ + id: EventId.make("event:older-writer:visited"), + type: "thread.visit-recorded", + threadId, + occurredAt: at, + payload: { lastVisitedAt: at }, + }); assert.deepEqual(yield* store.getThreadsWithPullRequests(), []); - // Its write marks the read projection stale, so shells read the payload it wrote. + // Its write marked the read projection stale. The visit rebuilt it from + // that payload, so shells read what it wrote without decoding the thread. + const [read] = yield* sql<{ readonly source_stale: number }>` + SELECT source_stale FROM orchestration_v2_projection_thread_reads + WHERE thread_id = ${threadId}`; + assert.equal(read?.source_stale, 0); assert.deepEqual( (yield* store.getThreadShell(threadId, { compactPullRequests: true }))?.pullRequests, [], ); + yield* store.apply({ + id: EventId.make("event:older-writer:marked-unread"), + type: "thread.visit-recorded", + threadId, + occurredAt: at, + payload: { lastVisitedAt: null }, + }); + assert.equal((yield* store.getThreadShell(threadId))?.lastVisitedAt, null); yield* sql`UPDATE orchestration_v2_projection_threads SET payload_json = ${stored!.payload_json} WHERE thread_id = ${threadId}`; assert.equal((yield* store.getThreadsWithPullRequests()).length, 1); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9aee61eea53ec..f1c318b20efd4 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -383,6 +383,7 @@ export interface ProjectionStoreV2Shape { cursor: string, throughEntryId?: string, conversationOnly?: boolean, + includeSupport?: boolean, ) => Effect.Effect; readonly getTimelinePage: ( threadId: ThreadId, @@ -548,6 +549,8 @@ export interface ProjectionStoreV2Shape { options: { readonly rowLimit: number; readonly userTurnLimit?: number | undefined; + /** Historical checkpoints hydrate with their rows instead of node ancestry. */ + readonly deferHistoricalSupport?: boolean | undefined; readonly anchorItemId?: TurnItemId | undefined; readonly anchorThreadId?: ThreadId | undefined; readonly requiredRunId?: RunId | undefined; @@ -1074,9 +1077,6 @@ const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema const encodeThreadPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema), ); -const encodeThreadUpdatedAt = Schema.encodeEffect( - OrchestrationV2AppThreadJsonSchema.fields.updatedAt, -); const encodeRunPayload = Schema.encodeEffect(Schema.fromJsonString(OrchestrationV2RunJsonSchema)); const encodeRestartCancelledBackgroundWork = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2RunJsonSchema.fields.restartCancelledBackgroundWork), @@ -1841,6 +1841,60 @@ export const layer: Layer.Layer = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; + // Field events already carry decoded values. Patch JSON in SQLite rather than + // decoding and encoding every saved PR and stack for a timestamp change. + const patchThreadFields = ( + threadId: ThreadId, + fields: { + readonly updatedAt?: string; + readonly lastVisitedAt?: string | null; + readonly activeProviderThreadId?: ProviderThreadId; + }, + ) => + Effect.gen(function* () { + const readRows = yield* sql<{ readonly source_stale: number }>` + SELECT source_stale FROM orchestration_v2_projection_thread_reads + WHERE thread_id = ${threadId} + `; + const patches = Object.entries(fields); + yield* sql` + UPDATE orchestration_v2_projection_threads + SET payload_json = json_set(payload_json, ${sql.join( + ", ", + false, + )(patches.map(([key, value]) => sql`${`$.${key}`}, ${value}`))}) + ${fields.updatedAt === undefined ? sql`` : sql`, updated_at = ${fields.updatedAt}`} + ${fields.activeProviderThreadId === undefined ? sql`` : sql`, active_provider_thread_id = ${fields.activeProviderThreadId}`} + WHERE thread_id = ${threadId} + `; + // An older build may have changed other fields. Only restore a read row + // that was current before the canonical write marked it stale. + if (readRows[0]?.source_stale === 0) { + const readPatches = sql.join( + ", ", + false, + )(patches.map(([key, value]) => sql`${`$.thread.${key}`}, ${value}`)); + yield* sql` + UPDATE orchestration_v2_projection_thread_reads + SET payload_json = json_set(payload_json, ${readPatches}), + compact_payload_json = json_set(compact_payload_json, ${readPatches}), + source_stale = 0 + WHERE thread_id = ${threadId} + `; + } else { + // A stale or missing read row is rebuilt from the whole thread once, + // so readers stop decoding the full payload. + const rows = yield* sql` + SELECT payload_json FROM orchestration_v2_projection_threads + WHERE thread_id = ${threadId} + `; + const row = rows[0]; + if (row !== undefined) { + yield* upsertThreadRead(yield* decodeThreadPayload(row.payload_json)); + } + } + }); + const upsertThreadRead = (thread: OrchestrationV2AppThread) => Effect.gen(function* () { const payload = yield* encodeThreadReadPayload(thread); @@ -1954,8 +2008,16 @@ export const layer: Layer.Layer = } // These events carry one field rather than the thread, so they patch the stored row. // A link sync keeps the thread's activity timestamp, just like its legacy event. - case "thread.pull-request-link-synced": case "thread.visit-recorded": { + yield* patchThreadFields(event.threadId, { + lastVisitedAt: + event.payload.lastVisitedAt === null + ? null + : DateTime.formatIso(event.payload.lastVisitedAt), + }); + break; + } + case "thread.pull-request-link-synced": { const rows = yield* sql` SELECT payload_json FROM orchestration_v2_projection_threads @@ -1965,10 +2027,7 @@ export const layer: Layer.Layer = const row = rows[0]; if (row === undefined) break; const thread = yield* decodeThreadPayload(row.payload_json); - const updatedThread = - event.type === "thread.visit-recorded" - ? { ...thread, lastVisitedAt: event.payload.lastVisitedAt } - : applyThreadPullRequestLinkSync(thread, event.payload); + const updatedThread = applyThreadPullRequestLinkSync(thread, event.payload); const payloadJson = yield* encodeThreadPayload(updatedThread); yield* sql` UPDATE orchestration_v2_projection_threads @@ -2317,31 +2376,10 @@ export const layer: Layer.Layer = event.payload.appThreadId !== null && !isQueuedProviderThreadPlaceholder(event.payload) ) { - const threadRows = yield* sql` - SELECT payload_json - FROM orchestration_v2_projection_threads - WHERE thread_id = ${event.payload.appThreadId} - LIMIT 1 - `; - const threadRow = threadRows[0]; - if (threadRow !== undefined) { - const thread = yield* decodeThreadPayload(threadRow.payload_json); - const updatedThread = { - ...thread, - activeProviderThreadId: event.payload.id, - updatedAt: event.payload.updatedAt, - }; - const updatedThreadPayloadJson = yield* encodeThreadPayload(updatedThread); - yield* sql` - UPDATE orchestration_v2_projection_threads - SET - active_provider_thread_id = ${event.payload.id}, - updated_at = ${stringField(parseEncodedPayload(updatedThreadPayloadJson), "updatedAt")}, - payload_json = ${updatedThreadPayloadJson} - WHERE thread_id = ${event.payload.appThreadId} - `; - yield* upsertThreadRead(updatedThread); - } + yield* patchThreadFields(event.payload.appThreadId, { + activeProviderThreadId: event.payload.id, + updatedAt: DateTime.formatIso(event.payload.updatedAt), + }); } break; } @@ -2765,14 +2803,9 @@ export const layer: Layer.Layer = event.type !== "thread.model-selection-updated" && event.type !== "thread.provider-switched" ) { - const updatedAt = yield* encodeThreadUpdatedAt(event.occurredAt); - yield* sql` - UPDATE orchestration_v2_projection_threads - SET - updated_at = ${updatedAt}, - payload_json = json_set(payload_json, '$.updatedAt', ${updatedAt}) - WHERE thread_id = ${event.threadId} - `; + yield* patchThreadFields(event.threadId, { + updatedAt: DateTime.formatIso(event.occurredAt), + }); } }).pipe( Effect.mapError( @@ -2789,6 +2822,7 @@ export const layer: Layer.Layer = window?: { readonly rowLimit: number; readonly userTurnLimit?: number | undefined; + readonly deferHistoricalSupport?: boolean | undefined; readonly anchorItemId?: TurnItemId | undefined; readonly requiredRunId?: RunId | undefined; readonly suppressLocal?: boolean | undefined; @@ -2975,6 +3009,21 @@ export const layer: Layer.Layer = ) AND request.type = 'run_interrupt_request' UNION + -- A viewport can start inside a long turn; keep that turn's prompt. + SELECT start.item_rowid, start.ordinal, start.turn_item_id + FROM ( + SELECT rowid AS item_rowid, ordinal, turn_item_id + FROM orchestration_v2_projection_turn_items + WHERE ${window.deferHistoricalSupport ? 1 : 0} = 1 + AND run_id = ( + SELECT run_id FROM selected ORDER BY ordinal ASC, turn_item_id ASC LIMIT 1 + ) + AND type = 'user_message' + AND json_extract(payload_json, '$.inputIntent') IN ('turn_start', 'queued_turn') + ORDER BY ordinal ASC + LIMIT 1 + ) AS start + UNION SELECT latest.item_rowid, latest.ordinal, latest.turn_item_id FROM ( SELECT rowid AS item_rowid, ordinal, turn_item_id @@ -3062,6 +3111,9 @@ export const layer: Layer.Layer = const cohortMessageIds = cohortJson("messageId"); const cohortPlanIds = cohortJson("planId"); const cohortCheckpointIds = cohortJson("checkpointId"); + const checkpointNodeIds = window?.deferHistoricalSupport + ? cohortJson("nodeId") + : cohortNodeIds; const cohortHandoffIds = cohortJson("contextHandoffId"); const [ @@ -3323,12 +3375,58 @@ export const layer: Layer.Layer = ORDER BY scope_id ASC, ordinal_within_scope ASC ` : sql` - SELECT payload_json FROM orchestration_v2_projection_checkpoints + SELECT CASE + WHEN status IN ('pending','capturing') + OR checkpoint_id IN (SELECT value FROM json_each(${cohortCheckpointIds})) + OR run_id IN (SELECT value FROM json_each(${cohortRunIds})) + OR node_id IN (SELECT value FROM json_each(${checkpointNodeIds})) + THEN payload_json + ELSE json_set(payload_json, '$.files', json('[]'), + '$.deferredFileCount', json_array_length(payload_json, '$.files')) + END AS payload_json + FROM orchestration_v2_projection_checkpoints WHERE thread_id = ${threadId} AND (status IN ('pending','capturing') OR checkpoint_id IN (SELECT value FROM json_each(${cohortCheckpointIds})) OR run_id IN (SELECT value FROM json_each(${cohortRunIds})) - OR node_id IN (SELECT value FROM json_each(${cohortNodeIds}))) + OR node_id IN (SELECT value FROM json_each(${checkpointNodeIds})) + -- A viewport keeps the runs of the latest user turns listed for + -- the Diff panel and Review, bounded like a turn-aligned window: + -- from the run of the twelfth latest user turn (a page plus + -- two), or every run when there are fewer. Wake-up runs ride + -- along without using up the turn budget. With no user turn + -- among the latest turns, only the window's own runs are listed. + -- Their file lists arrive with the rows' history page. + OR (${window.deferHistoricalSupport ? 1 : 0} = 1 AND app_run_ordinal >= ( + WITH turn_starts AS ( + SELECT run.ordinal AS run_ordinal, + json_extract(item.payload_json, '$.createdBy') = 'user' AS by_user + FROM orchestration_v2_projection_turn_items AS item + JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id + WHERE item.thread_id = ${threadId} + AND item.type = 'user_message' + AND json_extract(item.payload_json, '$.inputIntent') IN ('turn_start', 'queued_turn') + AND run.status <> 'rolled_back' + AND NOT ( + json_extract(item.payload_json, '$.inputIntent') = 'queued_turn' + AND run.status = 'cancelled' + ) + ORDER BY item.ordinal DESC + LIMIT ${THREAD_HISTORY_MAX_RAW_TURNS + 2} + ) + SELECT CASE + WHEN TOTAL(by_user) = 0 THEN NULL + ELSE MAX( + COALESCE(( + SELECT run_ordinal FROM turn_starts WHERE by_user + ORDER BY run_ordinal DESC + LIMIT 1 OFFSET ${THREAD_HISTORY_PAGE_POLICY.maxUserTurns + 1} + ), 0), + CASE WHEN COUNT(*) >= ${THREAD_HISTORY_MAX_RAW_TURNS + 2} THEN MIN(run_ordinal) ELSE 0 END + ) + END + FROM turn_starts + ))) ORDER BY scope_id ASC, ordinal_within_scope ASC `, fields !== undefined && !fields.includes("contextHandoffs") @@ -3449,6 +3547,7 @@ export const layer: Layer.Layer = window?: { readonly rowLimit: number; readonly userTurnLimit?: number | undefined; + readonly deferHistoricalSupport?: boolean | undefined; readonly anchorItemId?: TurnItemId | undefined; readonly requiredRunId?: RunId | undefined; readonly suppressLocal?: boolean | undefined; @@ -3563,6 +3662,7 @@ export const layer: Layer.Layer = return { rowLimit: window.rowLimit, userTurnLimit: window.userTurnLimit, + deferHistoricalSupport: window.deferHistoricalSupport, requiredRunId: forkedFrom.runId, suppressLocal: anchor === undefined || anchorIsInDescendant, ...(anchor === undefined || anchorIsInDescendant @@ -5104,6 +5204,7 @@ export const layer: Layer.Layer = cursor, throughEntryId, conversationOnly, + includeSupport, ) => sql .withTransaction( @@ -5133,6 +5234,29 @@ export const layer: Layer.Layer = catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), }); const end = (page.items.at(-1)?.canonicalPosition ?? -1) + 1; + const supportProjection = + includeSupport && page.items.length > 0 + ? yield* readProjection(threadId, new Set(), { + rowLimit: page.items.length, + deferHistoricalSupport: true, + ...(page.items.at(-1)!.sourceThreadId === threadId + ? { anchorItemId: page.items.at(-1)!.sourceItemId } + : {}), + historyAnchor: { + threadId: page.items.at(-1)!.sourceThreadId, + itemId: page.items.at(-1)!.sourceItemId, + }, + }) + : undefined; + const support = + supportProjection === undefined + ? undefined + : { + nodes: supportProjection.nodes, + checkpoints: supportProjection.checkpoints, + runs: supportProjection.runs, + attempts: supportProjection.attempts, + }; const hasUserTurns = index.visible .slice(0, end) .some((row) => isThreadHistoryUserTurn(row.item)); @@ -5179,6 +5303,7 @@ export const layer: Layer.Layer = ...page, snapshotSequence, items: items.map((row, position) => ({ ...row, position })), + ...(support === undefined ? {} : { support }), }; const bounded = selectRecentTimelineWindow({ items, @@ -5197,6 +5322,7 @@ export const layer: Layer.Layer = : bounded.items, nextCursor: bounded.hasMoreHistory ? bounded.nextCursor : page.nextCursor, hasMoreHistory: bounded.hasMoreHistory || page.hasMoreHistory, + ...(support === undefined ? {} : { support }), }; }), ) @@ -5291,6 +5417,7 @@ export const layer: Layer.Layer = const projection = yield* readProjection(threadId, new Set(), { rowLimit: options.rowLimit, userTurnLimit: options.userTurnLimit, + deferHistoricalSupport: options.deferHistoricalSupport, ...(historyAnchor?.threadId === threadId ? { anchorItemId: historyAnchor.itemId } : {}), @@ -6821,7 +6948,7 @@ export const layerMemory: Layer.Layer = Layer.effect( ), ), ), - getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly, includeSupport) => service.getThreadSnapshot(threadId).pipe( Effect.flatMap(({ projection, snapshotSequence }) => Effect.gen(function* () { @@ -6844,6 +6971,16 @@ export const layerMemory: Layer.Layer = Layer.effect( items: conversationOnly ? page.items.filter((row) => isConversationHistoryItem(row.item)) : page.items, + ...(includeSupport + ? { + support: { + nodes: projection.nodes, + checkpoints: projection.checkpoints, + runs: projection.runs, + attempts: projection.attempts, + }, + } + : {}), }; }, catch: (cause) => new ProjectionStoreReadError({ threadId, cause }), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index f1de494ae190c..c7899e6560573 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -32,6 +32,7 @@ import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; +import * as CheckpointService from "./CheckpointService.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -98,6 +99,7 @@ it("does not commit running state when inherited background routing cannot be re Layer.provide( Layer.mergeAll( Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({}), + Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }), IdAllocator.layer, Path.layer, @@ -140,7 +142,7 @@ it("does not commit running state when inherited background routing cannot be re tryHandlePromptCommand: () => Effect.succeed(false), }), Layer.mock(RunExecutionService.RunExecutionServiceV2)({ startRootRun }), - Layer.mock(RuntimePolicy.RuntimePolicyV2)({}), + RuntimePolicy.layer, ), ), ); @@ -173,6 +175,7 @@ function makeLocalCommandHarness(input: { readonly openFailure?: unknown; readonly ensureThreadFailure?: unknown; readonly historyReadFailureAfterFallback?: unknown; + readonly baseline?: "failure" | "defect" | "interrupt" | "stop"; readonly interruptOpen?: boolean; readonly interruptRunBeforeOpenFailure?: boolean; readonly writeFailure?: unknown; @@ -400,8 +403,27 @@ function makeLocalCommandHarness(input: { ), ensureThread: () => Effect.succeed(providerThread), }; - const open = vi.fn(() => - input.interruptOpen === true + let baselineSettled = false; + const captureBaseline = vi.fn(() => + Effect.gen(function* () { + if (input.baseline === "interrupt") return yield* Effect.interrupt; + baselineSettled = true; + if (input.baseline === "stop") interruptRun(); + if (input.baseline === "failure" || input.baseline === "defect") { + if (input.baseline === "defect") return yield* Effect.die("disk full"); + return yield* Effect.fail( + new CheckpointService.CheckpointBaselineCaptureError({ + scopeId: checkpointScopeId, + ordinalWithinScope: 1, + cause: "disk full", + }), + ); + } + }), + ); + const open = vi.fn(() => { + expect(baselineSettled).toBe(true); + return input.interruptOpen === true ? Effect.interrupt : "historyReadFailureAfterFallback" in input ? Effect.succeed(resumeFallbackSession as never) @@ -438,8 +460,8 @@ function makeLocalCommandHarness(input: { }, ensureThread: () => Effect.succeed(providerThread), } as never) - : Effect.die("A local command must not open a native session."), - ); + : Effect.die("A local command must not open a native session."); + }); const startRootRun = vi.fn< (input: RunExecutionService.RunExecutionServiceV2StartRootRunInput) => Effect.Effect >(() => @@ -514,6 +536,7 @@ function makeLocalCommandHarness(input: { Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({ prepareProviderHandoff: () => Effect.die("history read must fail first"), }), + Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline }), Layer.mock(EventSink.EventSinkV2)({ writeIfRunCurrent }), IdAllocator.layer, Path.layer, @@ -558,6 +581,7 @@ function makeLocalCommandHarness(input: { ), ); return { + captureBaseline, getThreadRecords, getTurnStartContext, setProjection: (next: OrchestrationV2ThreadProjection) => { @@ -1042,3 +1066,27 @@ for (const previousMessages of [[], ["/compact", " /COMPACT "]]) { }), ); } + +for (const baseline of ["failure", "defect", "interrupt", "stop"] as const) { + effectIt.effect(`settles baseline capture before starting a current provider (${baseline})`, () => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ + text: "Continue", + baseline, + failReadsAfterRunning: true, + }); + const result = yield* harness.startWithRetry.pipe(Effect.exit); + expect(harness.captureBaseline).toHaveBeenCalledOnce(); + if (baseline === "failure" || baseline === "defect") { + expect(harness.open).toHaveBeenCalledOnce(); + expect(harness.startRootRun).toHaveBeenCalledOnce(); + } else { + expect(harness.open).not.toHaveBeenCalled(); + if (baseline === "interrupt") { + expect(result._tag).toBe("Failure"); + expect(harness.writeIfRunCurrent).not.toHaveBeenCalled(); + } else expect(harness.projection().runs.at(-1)).toMatchObject({ status: "interrupted" }); + } + }), + ); +} diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 36354f8c1f4ef..5d075f2f4779f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -27,6 +27,7 @@ import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; +import * as CheckpointService from "./CheckpointService.ts"; import * as EventSink from "./EventSink.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import { @@ -84,6 +85,7 @@ export const layer: Layer.Layer< ProviderTurnStartServiceV2, never, | EventSink.EventSinkV2 + | CheckpointService.CheckpointServiceV2 | ContextHandoffService.ContextHandoffServiceV2 | IdAllocator.IdAllocatorV2 | FileSystem.FileSystem @@ -99,6 +101,7 @@ export const layer: Layer.Layer< ProviderTurnStartServiceV2, Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; + const checkpoints = yield* CheckpointService.CheckpointServiceV2; const contextHandoffService = yield* ContextHandoffService.ContextHandoffServiceV2; const idAllocator = yield* IdAllocator.IdAllocatorV2; const fileSystem = yield* FileSystem.FileSystem; @@ -490,9 +493,39 @@ export const layer: Layer.Layer< runs: current.runs, turnItems: current.turnItems, }); - const inheritedBackgroundTurnItems = yield* projectionStore - .getRuntimeRecoveryProjection(projection.thread.id) - .pipe(Effect.map(selectInheritedBackgroundItems)); + // These reads cannot start a workspace-capable provider. Do them while + // Git captures the baseline, then gate every session operation below. + const [inheritedBackgroundTurnItems, resolvedRuntimePolicy] = yield* Effect.all( + [ + projectionStore + .getRuntimeRecoveryProjection(projection.thread.id) + .pipe(Effect.map(selectInheritedBackgroundItems)), + runtimePolicy.resolve({ + thread: projection.thread, + modelSelection: run.modelSelection, + }), + // Session open and native thread setup can run hooks and MCP servers + // before a prompt. Capture their edits in the turn, never in its + // baseline. A capture that fails or dies still starts the turn, + // without a baseline. + checkpoints + .captureBaseline({ + scope: checkpointScope, + ordinalWithinScope: Math.max(0, run.ordinal - 1), + }) + .pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning( + "orchestration V2 checkpoint baseline capture failed; starting provider without a baseline", + { runId, cause: Cause.pretty(cause) }, + ), + ), + ), + ], + { concurrency: "unbounded" }, + ); const providerSessionId = providerThread.providerSessionId; const runControls = makeRunControls({ threadId: projection.thread.id, @@ -504,33 +537,13 @@ export const layer: Layer.Layer< }); const { isCurrentAttemptInStatus } = runControls; - const resolvedRuntimePolicy = yield* runtimePolicy.resolve({ - thread: projection.thread, - modelSelection: run.modelSelection, - }); const existingSessionProjection = projection.providerSessions.find( (candidate) => candidate.id === providerSessionId, ); - const sessionResult = yield* Effect.result( - providerSessions.open({ - threadId: projection.thread.id, - providerSessionId, - modelSelection: run.modelSelection, - runtimePolicy: resolvedRuntimePolicy, - ...(existingSessionProjection === undefined - ? {} - : { resumeFromSession: existingSessionProjection }), - ...(providerThread.nativeThreadRef?.nativeId == null - ? {} - : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }), - ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined - ? {} - : { - initialProviderItemIdentityVersion: - providerThread.nativeMetadata.itemIdentityVersion, - }), - }), - ); + // The last start attempt fails the run with the provider's own reason + // instead of leaving it `starting` after the effect gives up. A run that + // already left `starting` is not overwritten, and a failed write returns + // its error to the effect worker. const settleStartFailure = (failed: { readonly signal: string; readonly title: string; @@ -560,6 +573,27 @@ export const layer: Layer.Layer< }, }); }); + if (!(yield* isCurrentAttemptInStatus("starting"))) return; + const sessionResult = yield* Effect.result( + providerSessions.open({ + threadId: projection.thread.id, + providerSessionId, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + ...(existingSessionProjection === undefined + ? {} + : { resumeFromSession: existingSessionProjection }), + ...(providerThread.nativeThreadRef?.nativeId == null + ? {} + : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }), + ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined + ? {} + : { + initialProviderItemIdentityVersion: + providerThread.nativeMetadata.itemIdentityVersion, + }), + }), + ); if (sessionResult._tag === "Failure") { if (input.willRetry === true) return yield* sessionResult.failure; yield* settleStartFailure({ @@ -1201,6 +1235,7 @@ export const layer: Layer.Layer< run: runningRun, rootNode: runningRootNode, checkpointScope, + baselineCaptured: true, providerThread: runningProviderThread, attempt: runningAttempt, attemptId: attempt.id, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 1034f82f3cea9..f17fde771c832 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -525,6 +525,8 @@ export interface RunExecutionServiceV2StartRootRunInput { readonly run: OrchestrationV2Run; readonly rootNode: OrchestrationV2ExecutionNode; readonly checkpointScope: OrchestrationV2CheckpointScope; + /** Session startup already awaited the baseline, before opening the provider. */ + readonly baselineCaptured?: true; readonly providerThread: OrchestrationV2ProviderThread; readonly attempt: OrchestrationV2RunAttempt; readonly attemptId: RunAttemptId; @@ -854,21 +856,23 @@ export const layer: Layer.Layer< // Nothing provider-side may start before the baseline exists: // opening a Claude process already runs the user's SessionStart // hooks and MCP servers, which can edit the workspace. - yield* checkpointService - .captureBaseline({ - scope: input.checkpointScope, - ordinalWithinScope: Math.max(0, input.run.ordinal - 1), - }) - .pipe( - Effect.catchCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.failCause(cause) - : Effect.logWarning( - "orchestration V2 checkpoint baseline capture failed; starting provider without a baseline", - { runId: input.run.id }, - ), - ), - ); + if (input.baselineCaptured !== true) { + yield* checkpointService + .captureBaseline({ + scope: input.checkpointScope, + ordinalWithinScope: Math.max(0, input.run.ordinal - 1), + }) + .pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning( + "orchestration V2 checkpoint baseline capture failed; starting provider without a baseline", + { runId: input.run.id }, + ), + ), + ); + } if ( input.shouldStartProviderTurn !== undefined && !(yield* input.shouldStartProviderTurn()) diff --git a/apps/server/src/orchestration-v2/StreamSerialization.ts b/apps/server/src/orchestration-v2/StreamSerialization.ts new file mode 100644 index 0000000000000..7f80c94b7c450 --- /dev/null +++ b/apps/server/src/orchestration-v2/StreamSerialization.ts @@ -0,0 +1,92 @@ +import { OrchestrationV2ThreadStreamItem } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Predicate from "effect/Predicate"; +import * as Schema from "effect/Schema"; +import * as SchemaParser from "effect/SchemaParser"; +import * as RpcSerialization from "effect/rpc/RpcSerialization"; + +// Committed events are immutable and shared by subscriptions. The first subscriber +// validates and encodes an event; the others reuse its encoded stream item. +const cachedThreadItem = Schema.declareConstructor()( + [Schema.toCodecJson(OrchestrationV2ThreadStreamItem)], + ([codec]) => { + const parse = SchemaParser.decodeUnknownEffect(codec); + const cache = new WeakMap< + object, + { readonly sequence: number; readonly value: OrchestrationV2ThreadStreamItem } + >(); + return (input, _ast, options) => { + if ( + !Predicate.isObject(input) || + input.kind !== "event" || + typeof input.sequence !== "number" || + !Predicate.isObject(input.event) + ) + return parse(input, options); + const key = input.event; + const sequence = input.sequence; + const previous = cache.get(key); + if (previous?.sequence === sequence) return Effect.succeed(previous.value); + return parse(input, options).pipe( + Effect.tap((value) => + Effect.sync(() => { + cache.set(key, { sequence, value }); + }), + ), + ); + }; + }, +); +const threadChunkCodec = Schema.NonEmptyArray(cachedThreadItem); +const strings = new WeakMap(); + +const codecFor: RpcSerialization.CodecFor = (schema) => { + const ast = schema.ast; + if ( + ast._tag === "Arrays" && + ast.elements.length === 1 && + ast.rest.length === 1 && + ast.elements[0] === OrchestrationV2ThreadStreamItem.ast && + ast.rest[0] === OrchestrationV2ThreadStreamItem.ast + ) { + // The identity check above selects exactly NonEmptyArray(ThreadStreamItem). + return threadChunkCodec as Schema.Codec< + typeof schema.Type, + unknown, + typeof schema.DecodingServices, + typeof schema.EncodingServices + >; + } + return RpcSerialization.json.codecFor(schema); +}; + +export const layer = Layer.succeed(RpcSerialization.RpcSerialization, { + ...RpcSerialization.json, + codecFor, + makeUnsafe: () => { + const parser = RpcSerialization.json.makeUnsafe(); + const stringifyItem = (item: unknown) => { + if (!Predicate.isObject(item) || item.kind !== "event") return JSON.stringify(item); + const cached = strings.get(item); + if (cached !== undefined) return cached; + const encoded = JSON.stringify(item); + strings.set(item, encoded); + return encoded; + }; + return { + ...parser, + encode: (response: unknown) => { + if ( + !Predicate.isObject(response) || + response._tag !== "Chunk" || + !Array.isArray(response.values) || + !response.values.every((item) => Predicate.isObject(item) && item.kind === "event") + ) { + return parser.encode(response); + } + return `{"_tag":"Chunk","requestId":${JSON.stringify(response.requestId)},"values":[${response.values.map(stringifyItem).join(",")}]}`; + }, + }; + }, +}); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index f23771e4ab738..18b9891582861 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1132,10 +1132,16 @@ const make = Effect.gen(function* () { Effect.andThen(orchestrator.searchThread(input)), ), dispatch, - getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly) => + getThreadHistoryPage: (threadId, cursor, throughEntryId, conversationOnly, includeSupport) => ensureProjectionTranscript(threadId).pipe( Effect.andThen( - orchestrator.getThreadHistoryPage(threadId, cursor, throughEntryId, conversationOnly), + orchestrator.getThreadHistoryPage( + threadId, + cursor, + throughEntryId, + conversationOnly, + includeSupport, + ), ), ), getTimelinePage: (threadId, options) => diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index bad0156b3ea58..a159097ffeb6e 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -11,6 +11,7 @@ import { import { buildBoundedThreadProjection, type BoundedProjectionResult, + THREAD_HISTORY_PAGE_POLICY, } from "./threadHistoryPaging.ts"; import { projectThreadProjectionForWire } from "./WireProjection.ts"; @@ -74,10 +75,20 @@ export function buildBoundedThreadStreamSnapshot(input: { readonly projection: OrchestrationV2ThreadProjection; readonly compactTurnItems?: boolean; readonly compactCheckpointItems?: boolean; + readonly viewportWindow?: boolean; }): ThreadSnapshotStreamItem { const bounded = buildBoundedThreadProjection({ snapshotSequence: input.snapshotSequence, projection: projectThreadProjectionForWire(input.projection), + viewportWindow: input.viewportWindow === true, + ...(input.viewportWindow + ? { + policy: { + maxItems: THREAD_HISTORY_PAGE_POLICY.maxItems, + maxEncodedBytes: THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes, + }, + } + : {}), }); return { kind: "snapshot", diff --git a/apps/server/src/orchestration-v2/WireProjection.test.ts b/apps/server/src/orchestration-v2/WireProjection.test.ts index 21f8cfb058a56..3cc251f41b788 100644 --- a/apps/server/src/orchestration-v2/WireProjection.test.ts +++ b/apps/server/src/orchestration-v2/WireProjection.test.ts @@ -12,10 +12,15 @@ import { type OrchestrationV2ThreadProjection, OrchestrationV2TurnItem, OrchestrationV2TurnItemJson, + OrchestrationV2ThreadStreamItem, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Schema from "effect/Schema"; +import * as RpcSerialization from "effect/rpc/RpcSerialization"; +import * as StreamSerialization from "./StreamSerialization.ts"; import { projectTurnItemForWire, @@ -50,6 +55,31 @@ const base = { }; describe("orchestration V2 wire projection", () => { + it.effect("keeps streamed JSON compatible across subscribers and rejects invalid events", () => + Effect.gen(function* () { + const serialization = yield* RpcSerialization.RpcSerialization; + const schema = serialization.codecFor(Schema.NonEmptyArray(OrchestrationV2ThreadStreamItem)); + const event = projectDomainEventForWire({ + id: EventId.make("event:shared-wire"), + type: "turn-item.updated", + threadId: base.threadId, + occurredAt: base.updatedAt, + payload: { ...base, output: { text: "PRIVATE_TOOL_OUTPUT" } }, + }); + const encode = Schema.encodeEffect(schema); + for (const requestId of ["first", "second"]) { + const values = yield* encode([{ kind: "event", sequence: 1, event }]); + const response = { _tag: "Chunk", requestId, values }; + const json = serialization.makeUnsafe().encode(response); + expect(json).toBe(RpcSerialization.json.makeUnsafe().encode(response)); + expect(json).not.toContain("PRIVATE_TOOL_OUTPUT"); + } + expect( + Exit.isFailure(yield* Effect.exit(encode([{ kind: "event", sequence: -1, event }]))), + ).toBe(true); + }).pipe(Effect.provide(StreamSerialization.layer)), + ); + it("keeps copied handoff transcripts out of activity items and live events", () => { const item = { ...base, diff --git a/apps/server/src/orchestration-v2/WireProjection.ts b/apps/server/src/orchestration-v2/WireProjection.ts index 043fbd3435acb..3be743d303d46 100644 --- a/apps/server/src/orchestration-v2/WireProjection.ts +++ b/apps/server/src/orchestration-v2/WireProjection.ts @@ -236,12 +236,19 @@ export function projectThreadProjectionForWire( }; } +const wireEvents = new WeakMap(); + export function projectDomainEventForWire( event: OrchestrationV2DomainEvent, ): OrchestrationV2DomainEvent { - return event.type === "turn-item.updated" - ? { ...event, payload: projectTurnItemForWire(event.payload) } - : event.type === "context-handoff.updated" - ? { ...event, payload: projectContextHandoffForWire(event.payload) } - : event; + const cached = wireEvents.get(event); + if (cached !== undefined) return cached; + const projected = + event.type === "turn-item.updated" + ? { ...event, payload: projectTurnItemForWire(event.payload) } + : event.type === "context-handoff.updated" + ? { ...event, payload: projectContextHandoffForWire(event.payload) } + : event; + wireEvents.set(event, projected); + return projected; } diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index 545ca5ef98430..3cd569c53ed22 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -125,12 +125,15 @@ export const layer = HttpApiBuilder.group( anchorThreadId?: Parameters< typeof threadManagement.getThreadSnapshotWindow >[1]["anchorThreadId"], + viewportWindow = false, ) { return yield* threadManagement .getThreadSnapshotWindow(threadId, { rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, - userTurnLimit: - anchorItemId === undefined + deferHistoricalSupport: viewportWindow, + userTurnLimit: viewportWindow + ? undefined + : anchorItemId === undefined ? THREAD_HISTORY_PAGE_POLICY.maxUserTurns : OLDER_THREAD_USER_TURN_LIMIT, ...(anchorItemId === undefined ? {} : { anchorItemId }), @@ -212,12 +215,25 @@ export const layer = HttpApiBuilder.group( Effect.fn("environment.orchestration.threadBoundedSnapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); - const snapshot = yield* loadThreadSnapshotWindow(args.params.threadId).pipe( - traceLocalHandlerWork, - ); + const viewportWindow = args.query.viewportWindow === "1"; + const snapshot = yield* loadThreadSnapshotWindow( + args.params.threadId, + undefined, + undefined, + viewportWindow, + ).pipe(traceLocalHandlerWork); const bounded = buildBoundedThreadProjection({ projection: snapshot.projection, snapshotSequence: snapshot.snapshotSequence, + viewportWindow, + ...(viewportWindow + ? { + policy: { + maxItems: THREAD_HISTORY_PAGE_POLICY.maxItems, + maxEncodedBytes: THREAD_HISTORY_PAGE_POLICY.maxEncodedBytes, + }, + } + : {}), }); return { snapshotSequence: snapshot.snapshotSequence, @@ -247,6 +263,7 @@ export const layer = HttpApiBuilder.group( args.query.cursor, args.query.throughEntryId, args.query.view === "conversation", + args.query.includeSupport === "1", ) .pipe( traceLocalHandlerWork, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 52e9adf187064..e121b9152e100 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -162,6 +162,7 @@ const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( const layerProviderTurnStartServiceProvided = ProviderTurnStartService.layer.pipe( Layer.provide( Layer.mergeAll( + layerCheckpointServiceProvided, layerContextHandoffServiceProvided, layerEventSinkProvided, IdAllocator.layer, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 14c08d4922146..d4e4606752bac 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -391,6 +391,7 @@ export function layerWithRegistry( const layerProviderTurnStartServiceProvided = ProviderTurnStartService.layer.pipe( Layer.provide( Layer.mergeAll( + layerCheckpointServiceProvided, layerContextHandoffServiceProvided, layerEventSinkProvided, IdAllocator.layer, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs b/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs index 0fafaf063762d..a9e701739635c 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs +++ b/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs @@ -30,6 +30,7 @@ const [Start, Projection, Run, Sessions, Policy, Id, Sink, Handoff, Git, Project app("project/ProjectService"), app("provider/ProviderAuthService"), ]); +const Checkpoint = await app("orchestration-v2/CheckpointService"); let current; let fullReads = 0; let threadReads = 0; @@ -48,6 +49,7 @@ const session = { }; const dependencies = Layer.mergeAll( Layer.mock(Handoff.ContextHandoffServiceV2)({}), + Layer.mock(Checkpoint.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), Id.layer, FileSystem.layerNoop({}), Path.layer, diff --git a/apps/server/src/orchestration-v2/threadHistoryPaging.ts b/apps/server/src/orchestration-v2/threadHistoryPaging.ts index d65bc8f07a7f4..419540fad0b83 100644 --- a/apps/server/src/orchestration-v2/threadHistoryPaging.ts +++ b/apps/server/src/orchestration-v2/threadHistoryPaging.ts @@ -302,7 +302,9 @@ export function selectHistoryPageFromCursor(input: { }; if (input.throughEntryId !== undefined) { const targetIndex = input.items.findIndex( - (row) => (row.item.messageId ?? row.item.id) === input.throughEntryId, + (row) => + (row.item.messageId ?? row.item.id) === input.throughEntryId || + row.item.runId === input.throughEntryId, ); if (targetIndex >= 0 && targetIndex < anchorIndex) { const turns = input.items @@ -455,11 +457,38 @@ export function buildBoundedThreadProjection(input: { readonly projection: OrchestrationV2ThreadProjection; readonly snapshotSequence: number; readonly policy?: ThreadHistoryPagePolicy | undefined; + readonly viewportWindow?: boolean; }): BoundedProjectionResult { const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY; const threadId = input.projection.thread.id; + const recentRows = input.projection.visibleTurnItems.slice(-policy.maxItems); + const retainedRunIds = new Set(recentRows.map(({ item }) => item.runId)); + for (const run of input.projection.runs) { + if (["queued", "preparing", "starting", "running", "waiting"].includes(run.status)) { + retainedRunIds.add(run.id); + } + } + const retainedNodeIds = new Set(recentRows.map(({ item }) => item.nodeId)); + const retainedCheckpointIds = new Set( + recentRows.flatMap(({ item }) => ("checkpointId" in item ? [item.checkpointId] : [])), + ); const controlProjection = { ...input.projection, + ...(input.viewportWindow + ? { + // The store already bounds which turns are listed. Turns outside the + // window, including runs that only reach the store's surplus rows, + // stay listed without their files until their history page loads. + checkpoints: input.projection.checkpoints.map((checkpoint) => + (checkpoint.runId !== null && retainedRunIds.has(checkpoint.runId)) || + retainedNodeIds.has(checkpoint.nodeId) || + retainedCheckpointIds.has(checkpoint.id) || + checkpoint.deferredFileCount !== undefined + ? checkpoint + : { ...checkpoint, files: [], deferredFileCount: checkpoint.files.length }, + ), + } + : {}), plans: input.projection.plans.filter((plan) => plan.status === "active"), contextHandoffs: input.projection.contextHandoffs.filter( (handoff) => handoff.status === "pending" || handoff.status === "ready", @@ -498,7 +527,22 @@ export function buildBoundedThreadProjection(input: { policy: windowPolicy, rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId), }); - const visibleTurnItems = renumberPositions(window.items); + // A viewport that starts inside a long turn keeps that turn's prompt above + // the gap. The cursor still points at the window, so paging fills the gap. + const oldestWindowRow = window.items[0]; + const truncatedTurnStart = + input.viewportWindow === true && + oldestWindowRow !== undefined && + !isThreadHistoryTurnStart(oldestWindowRow.item) && + oldestWindowRow.item.runId !== null + ? controlProjection.visibleTurnItems.find( + (row) => + row.item.runId === oldestWindowRow.item.runId && isThreadHistoryTurnStart(row.item), + ) + : undefined; + const visibleTurnItems = renumberPositions( + truncatedTurnStart === undefined ? window.items : [truncatedTurnStart, ...window.items], + ); const windowTurnItems = localTurnItemsForVisibleWindow(controlProjection, visibleTurnItems); const dependencyTurnItems = retainedInterruptRequestTurnItems( controlProjection, diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index a794074611673..f7a0a67c2a165 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -180,6 +180,12 @@ interface ApplicationEventRow { const decodeV2EventJson = Schema.decodeUnknownEffect(OrchestrationV2DomainEventJson); const encodeV2EventJson = Schema.encodeEffect(OrchestrationV2DomainEventJson); const decodeV2StoredEvent = Schema.decodeUnknownEffect(OrchestrationV2StoredEvent); +const decodeV2StoredHeader = Schema.decodeUnknownEffect( + Schema.Struct({ + sequence: OrchestrationV2StoredEvent.fields.sequence, + commandId: OrchestrationV2StoredEvent.fields.commandId, + }), +); const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)); const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); @@ -448,6 +454,7 @@ const makeEventStore = Effect.gen(function* () { (event) => Effect.gen(function* () { const encoded = yield* encodeV2EventJson(event); + const normalized = yield* decodeV2EventJson(encoded); const subagent = encoded.type === "subagent.updated" ? encoded.payload @@ -529,11 +536,11 @@ const makeEventStore = Effect.gen(function* () { if (oldest !== undefined) promptHints.delete(oldest); } } - return yield* decodeV2StoredEvent({ + const header = yield* decodeV2StoredHeader({ sequence: rows[0]?.sequence, commandId: input.commandId ?? null, - event, }); + return { ...header, event: normalized }; }), { concurrency: 1 }, ).pipe( diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 703190e494bae..55cb450f9574e 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -70,6 +70,78 @@ const baseStatus: VcsStatusResult = { ...baseRemoteStatus, }; +it.effect("summarizes passive VCS subscriptions and sends a fresh snapshot on remount", () => { + const state = { + currentLocalStatus: { + ...baseLocalStatus, + hasWorkingTreeChanges: true, + workingTree: { + insertions: 1, + deletions: 0, + files: [{ path: "changed.ts", insertions: 1, deletions: 0 }], + }, + }, + currentRemoteStatus: baseRemoteStatus, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const read = (options: { compact?: boolean; includeFiles?: boolean }) => + broadcaster + .streamStatus({ cwd: "/repo", includeRemote: false, ...options }) + .pipe(Stream.runHead); + const initial = Option.getOrThrow(yield* read({ compact: true })); + assert.strictEqual(initial._tag, "snapshot"); + if (initial._tag !== "snapshot") return; + assert.deepStrictEqual( + initial.local.workingTree.files, + state.currentLocalStatus.workingTree.files, + ); + assert.deepStrictEqual(Option.getOrThrow(yield* read({ compact: true })), initial); + for (const options of [{}, { compact: true, includeFiles: true }]) { + const full = Option.getOrThrow(yield* read(options)); + assert.strictEqual(full._tag, "snapshot"); + if (full._tag === "snapshot") + assert.deepStrictEqual( + full.local.workingTree.files, + state.currentLocalStatus.workingTree.files, + ); + } + const activeSnapshot = yield* Deferred.make(); + const activeStream = yield* broadcaster + .streamStatus({ cwd: "/repo", includeRemote: false }) + .pipe( + Stream.runForEach(() => Deferred.succeed(activeSnapshot, undefined)), + Effect.forkScoped, + ); + yield* Deferred.await(activeSnapshot); + yield* broadcaster.getStatus({ cwd: "/repo" }); + for (let index = 0; index < 300; index++) { + yield* broadcaster.getStatus({ cwd: `/repo-${index}` }); + } + const reads = state.localStatusCalls; + state.currentLocalStatus = { ...state.currentLocalStatus, refName: "fresh-branch" }; + const reloaded = yield* broadcaster.getStatus({ cwd: "/repo-0" }); + assert.equal(reloaded.refName, "fresh-branch"); + assert.equal(state.localStatusCalls, reads + 1); + state.currentRemoteStatus = remoteStatusWithPr; + assert.deepStrictEqual( + yield* broadcaster.refreshPullRequestStatus("/repo"), + remoteStatusWithPr, + ); + yield* Fiber.interrupt(activeStream); + for (let index = 0; index < 300; index++) { + yield* broadcaster.getStatus({ cwd: `/later-${index}` }); + } + const remoteReads = state.remoteStatusCalls; + yield* broadcaster.getStatus({ cwd: "/repo" }); + assert.equal(state.remoteStatusCalls, remoteReads + 1); + }).pipe(Effect.provide(layerTestFor(state))); +}); + function layerTestFor(state: { currentLocalStatus: VcsStatusLocalResult; currentRemoteStatus: VcsStatusRemoteResult | null; diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 7c1f7f364c25e..9a2f3687acb9e 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -33,6 +33,7 @@ import * as ServerSettings from "../serverSettings.ts"; const DEFAULT_VCS_STATUS_REFRESH_INTERVAL = Duration.seconds(30); const VCS_STATUS_REFRESH_FAILURE_BASE_DELAY = Duration.seconds(30); const VCS_STATUS_REFRESH_FAILURE_MAX_DELAY = Duration.minutes(15); +const MAX_CACHED_WORKSPACES = 256; const MAX_FAILURE_DIAGNOSTIC_VALUES = 8; const MAX_FAILURE_DIAGNOSTIC_VALUE_LENGTH = 128; @@ -227,6 +228,14 @@ export const make = Effect.gen(function* () { Scope.close(scope, Exit.void), ); const cacheRef = yield* Ref.make(new Map()); + const activeCwds = new Map(); + const trimCache = (cache: Map) => { + for (const cwd of cache.keys()) { + if (cache.size <= MAX_CACHED_WORKSPACES) break; + if (!activeCwds.has(cwd)) cache.delete(cwd); + } + return cache; + }; // One permit per cwd for remote reads that write the cache. Without it a // periodic poll that started before `gh pr create` can finish after the // turn-end refresh and overwrite the fresh PR with its stale `pr: null`. @@ -254,7 +263,10 @@ export const make = Effect.gen(function* () { ...previous, local: nextLocal, }); - return [previous.local?.fingerprint !== nextLocal.fingerprint, nextCache] as const; + return [ + previous.local?.fingerprint !== nextLocal.fingerprint, + trimCache(nextCache), + ] as const; }); if (options?.publish && shouldPublish) { @@ -284,7 +296,10 @@ export const make = Effect.gen(function* () { ...previous, remote: nextRemote, }); - return [previous.remote?.fingerprint !== nextRemote.fingerprint, nextCache] as const; + return [ + previous.remote?.fingerprint !== nextRemote.fingerprint, + trimCache(nextCache), + ] as const; }); if (options?.publish && shouldPublish) { @@ -325,7 +340,7 @@ export const make = Effect.gen(function* () { return [ previous.local?.fingerprint !== nextLocal.fingerprint || previous.remote?.fingerprint !== nextRemote.fingerprint, - nextCache, + trimCache(nextCache), ] as const; }); @@ -697,6 +712,16 @@ export const make = Effect.gen(function* () { Stream.unwrap( Effect.gen(function* () { const cwd = yield* withFileSystem(normalizeCwd(input.cwd)); + // Active subscriptions need their remote cache for turn-end PR refreshes. + yield* Effect.acquireRelease( + Effect.sync(() => activeCwds.set(cwd, (activeCwds.get(cwd) ?? 0) + 1)), + () => { + const count = activeCwds.get(cwd) ?? 0; + if (count <= 1) activeCwds.delete(cwd); + else activeCwds.set(cwd, count - 1); + return Ref.update(cacheRef, (cache) => trimCache(new Map(cache))); + }, + ); const subscription = yield* PubSub.subscribe(changesPubSub); const initialLocal = yield* getOrLoadLocalStatus(cwd); const cachedStatus = yield* getCachedStatus(cwd); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 8cda52521d581..63a080d1548bd 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -109,7 +109,7 @@ import { HttpServerRespondable, HttpServerResponse, } from "effect/http"; -import { RpcSerialization, RpcServer } from "effect/rpc"; +import { RpcServer } from "effect/rpc"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as ServerConfig from "./config.ts"; @@ -117,6 +117,7 @@ import * as EnvironmentTheme from "./environmentTheme.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as ThreadManagementService from "./orchestration-v2/ThreadManagementService.ts"; +import * as StreamSerialization from "./orchestration-v2/StreamSerialization.ts"; import * as CommandOutputHub from "./orchestration-v2/CommandOutputHub.ts"; import * as McpAppRequests from "./mcpApps/McpAppRequests.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; @@ -741,6 +742,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs readonly acceptBoundedSnapshot?: boolean; readonly acceptCompactTurnItems?: boolean; readonly acceptCompactCheckpointItems?: boolean; + readonly acceptViewportWindow?: boolean; readonly acceptThreadFieldEvents?: boolean; }) { const threadManagement = yield* ThreadManagementService.ThreadManagementService; @@ -850,7 +852,10 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs useBoundedSnapshot ? threadManagement.getThreadSnapshotWindow(input.threadId, { rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, - userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + deferHistoricalSupport: input.acceptViewportWindow === true, + userTurnLimit: input.acceptViewportWindow + ? undefined + : THREAD_HISTORY_PAGE_POLICY.maxUserTurns, }) : threadManagement.getThreadSnapshot(input.threadId) ).pipe( @@ -869,6 +874,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs ...snapshot, compactTurnItems: input.acceptCompactTurnItems === true, compactCheckpointItems: input.acceptCompactCheckpointItems === true, + viewportWindow: input.acceptViewportWindow === true, }) : { kind: "snapshot" as const, @@ -1728,17 +1734,28 @@ const layerWsRpc = ( const environment = yield* serverEnvironment.getDescriptor; const auth = yield* serverAuth.getDescriptor(); const scratchWorkspaceRoot = yield* managedFolders.scratchRoot; - const editorConfig = - options.deferEditors === true - ? { availableEditors: [] } - : yield* resolveEditorConfig( - yield* resolveAvailableEditorsForConfig( - externalLauncher.resolveAvailableEditors(), - ), - resolveFileManagerRevealKindForConfig( - externalLauncher.resolveFileManagerRevealKind(), - ), - ); + const [editorConfig, availableRemoteOpenTargets, availableDirectEndpoints] = + yield* Effect.all( + [ + options.deferEditors === true + ? Effect.succeed({ availableEditors: [] }) + : resolveAvailableEditorsForConfig( + externalLauncher.resolveAvailableEditors(), + ).pipe( + Effect.flatMap((editors) => + resolveEditorConfig( + editors, + resolveFileManagerRevealKindForConfig( + externalLauncher.resolveFileManagerRevealKind(), + ), + ), + ), + ), + resolveAvailableEditorsForConfig(remoteOpenTargets.resolveTargets()), + resolveAvailableEditorsForConfig(directEndpoints.resolve()), + ], + { concurrency: "unbounded" }, + ); return { environment, @@ -1749,10 +1766,10 @@ const layerWsRpc = ( issues: keybindingsConfig.issues, providers, ...editorConfig, - remoteOpenTargets: yield* resolveAvailableEditorsForConfig( - remoteOpenTargets.resolveTargets(), - ), - directEndpoints: yield* resolveAvailableEditorsForConfig(directEndpoints.resolve()), + // Same discovery-with-timeout treatment as editors: a slow probe + // must not stall server.getConfig, so it degrades to no targets. + remoteOpenTargets: availableRemoteOpenTargets, + directEndpoints: availableDirectEndpoints, observability: { logsDirectoryPath: config.logsDir, localTracingEnabled: true, @@ -3241,7 +3258,7 @@ export const layer = Layer.unwrap( previewAutomationBroker, serverBrowser, ).pipe( - Layer.provideMerge(RpcSerialization.layerJson), + Layer.provideMerge(StreamSerialization.layer), // Request fibers run in the handlers' context, so this reporter sees // their defects, not the rest of the server's. Layer.provide(DefectReporter.layer), diff --git a/apps/web/package.json b/apps/web/package.json index 8b8aec8ffa577..1b6d2c7f819cc 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -37,7 +37,6 @@ "@t3tools/provider-opencode": "workspace:*", "@t3tools/provider-pi": "workspace:*", "@t3tools/shared": "workspace:*", - "@tanstack/react-pacer": "^0.19.4", "@tanstack/react-router": "^1.160.2", "@tiptap/core": "^3.31.3", "@tiptap/extension-blockquote": "^3.31.3", diff --git a/apps/web/src/components/BranchToolbar.tsx b/apps/web/src/components/BranchToolbar.tsx index 9d1ffe2c5e9e8..82d2f42a3ce8d 100644 --- a/apps/web/src/components/BranchToolbar.tsx +++ b/apps/web/src/components/BranchToolbar.tsx @@ -23,6 +23,7 @@ import { useRef, useState, } from "react"; +import { flushSync } from "react-dom"; import { useComposerDraftStore, type DraftId } from "../composerDraftStore"; import { EnvironmentMachineIcon } from "./EnvironmentMachineIcon"; @@ -549,14 +550,21 @@ function useLabelsOverflow(element: HTMLDivElement | null): boolean { useLayoutEffect(() => { measure(); - }); + }, [element, measure]); useEffect(() => { if (!element) return; const stopObserving = observeResize(element, measure); + const mutations = new MutationObserver(() => flushSync(measure)); + mutations.observe(element, { subtree: true, childList: true, characterData: true }); + mutations.observe(element.ownerDocument.documentElement, { + attributes: true, + attributeFilter: ["style", "class"], + }); document.fonts.addEventListener("loadingdone", measure); return () => { stopObserving(); + mutations.disconnect(); document.fonts.removeEventListener("loadingdone", measure); }; }, [element, measure]); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 18b6b7012367b..ea2110b17f763 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -137,7 +137,7 @@ import { sourceControlRepositorySelector } from "@t3tools/shared/sourceControl"; import { truncate } from "@t3tools/shared/String"; import { resolveThreadReferenceCopyTarget } from "@t3tools/shared/threadReference"; import { resolveTerminalSessionLabel } from "@t3tools/shared/terminalLabels"; -import { Debouncer } from "@tanstack/react-pacer"; +import { createDebouncer } from "~/lib/debounce"; import { useAtomValue } from "@effect/atom-react"; import { Atom } from "effect/reactivity"; import { @@ -393,18 +393,17 @@ import { useEnvironments, usePrimaryEnvironment } from "../state/environments"; import { resolveThreadDetailRef, useEnvironmentSupportsServerBrowser, + useEnvironmentShellBootstrapped, useProject, useProjects, - useThreadProjection, useThreadStatus, useThreadHistory, useThreadShell, useChildThreadInputs, useThreadRefs, - useThreadVisibleTurnItems, waitForThreadShell, } from "../state/entities"; -import { environmentShell } from "../state/shell"; +import { useChatThreadPresentation } from "../hooks/useChatThreadPresentation"; import { ChatComposer, type ChatComposerHandle } from "./chat/ChatComposer"; import { createPageScrollController, type PageScrollKey } from "./chat/pageScrollController"; import { isTimelineScrollTarget } from "./chat/timelineScrollTarget"; @@ -1228,8 +1227,9 @@ export default function ChatView(props: ChatViewProps) { shellExists: serverThread !== null, waitForShell: draftThread !== null, }); - const serverThreadProjection = useThreadProjection(routeThreadDetailRef); - const serverProjection = serverThreadProjection?.projection ?? null; + const [isThreadFindActive, setIsThreadFindActive] = useState(false); + const { projection: serverProjection, visibleTurnItems: serverVisibleTurnItems } = + useChatThreadPresentation(routeThreadDetailRef, isThreadFindActive); const promptRef = useRef(""); // Each view owns a composer; only the shown view publishes the shared ref. const composerRef = useRef(null); @@ -1273,7 +1273,6 @@ export default function ChatView(props: ChatViewProps) { () => latestProviderTurnTokenUsage(serverProjection?.providerTurns ?? []), [serverProjection?.providerTurns], ); - const serverVisibleTurnItems = useThreadVisibleTurnItems(routeThreadDetailRef); const serverThreadHistory = useThreadHistory(routeThreadDetailRef); const threadHistoryControls = useMemo(() => { if (routeThreadDetailRef === null || !shouldShowLoadEarlierControl(serverThreadHistory)) { @@ -1283,6 +1282,7 @@ export default function ChatView(props: ChatViewProps) { hasMoreHistory: serverThreadHistory.hasMoreHistory, loading: serverThreadHistory.loading, error: serverThreadHistory.error, + cursor: serverThreadHistory.historyCursor, onLoadEarlier: (throughEntryId) => { void loadEarlierThreadHistory({ environmentId: routeThreadDetailRef.environmentId, @@ -1294,6 +1294,13 @@ export default function ChatView(props: ChatViewProps) { }, }; }, [loadEarlierThreadHistory, routeThreadDetailRef, serverThreadHistory]); + // Prompt recall past the loaded window pages older turns in. True while a + // page is on its way, false once there is nothing older to load. + const loadOlderPromptHistory = useCallback(() => { + if (!threadHistoryControls?.hasMoreHistory || threadHistoryControls.error) return false; + if (!threadHistoryControls.loading) threadHistoryControls.onLoadEarlier(); + return true; + }, [threadHistoryControls]); const committedServerMessageIds = useMemo( () => deriveCommittedServerUserMessageIds(serverVisibleTurnItems), [serverVisibleTurnItems], @@ -2093,10 +2100,9 @@ export default function ChatView(props: ChatViewProps) { }); }; const handleOpenDraftProjectSettingsStable = useStableCallback(handleOpenDraftProjectSettings); - const activeEnvironmentShell = useEnvironmentQuery( - activeThread ? environmentShell.stateAtom(activeThread.environmentId) : null, + const activeEnvironmentBootstrapComplete = useEnvironmentShellBootstrapped( + activeThread?.environmentId ?? null, ); - const activeEnvironmentBootstrapComplete = activeEnvironmentShell.data?.snapshot._tag === "Some"; const activeProjectKey = activeProject ? `${activeProject.environmentId}:${activeProject.workspaceRoot}` : null; @@ -4355,9 +4361,7 @@ export default function ChatView(props: ChatViewProps) { // Debounce *showing* the scroll-to-bottom pill so it doesn't flash during // thread switches. LegendList fires scroll events with isAtEnd=false while // initialScrollAtEnd is settling; hiding is always immediate. - const showScrollDebouncer = useRef( - new Debouncer(() => setShowScrollToBottom(true), { wait: 150 }), - ); + const showScrollDebouncer = useRef(createDebouncer(() => setShowScrollToBottom(true), 150)); const timelineScrollIntentRef = useRef<"toward-end" | "away-from-end" | null>(null); const timelineScrollModeRef = useRef("following-end"); // State mirror of the follow mode refs. LegendList's maintainScrollAtEnd @@ -4811,7 +4815,7 @@ export default function ChatView(props: ChatViewProps) { } else { timelineScrollModeRef.current = "free-scrolling"; liveFollowUserScrollGenerationRef.current = null; - showScrollDebouncer.current.maybeExecute(); + showScrollDebouncer.current.schedule(); } }; const onIsAtEndChangeStable = useStableCallback(onIsAtEndChange); @@ -5907,7 +5911,6 @@ export default function ChatView(props: ChatViewProps) { displayedThreadRef?.environmentId ?? activeThread?.environmentId ?? environmentId, )?.serverConfig?.providers ?? EMPTY_PROVIDERS; const threadFindControlsRef = useRef(null); - const [isThreadFindActive, setIsThreadFindActive] = useState(false); const openThreadFind = () => threadFindControlsRef.current?.open(); const openThreadFindStable = useStableCallback(openThreadFind); const closeThreadFind = () => threadFindControlsRef.current?.close(); @@ -8849,6 +8852,8 @@ export default function ChatView(props: ChatViewProps) { activeThread={activeThread} activeThreadShell={activeThreadShell} promptHistoryMessages={timelineMessages} + onLoadOlderPromptHistory={loadOlderPromptHistory} + olderPromptHistoryFailed={Boolean(threadHistoryControls?.error)} isServerThread={isServerThread} isLocalDraftThread={isLocalDraftThread} forceExpandedOnMobile={ diff --git a/apps/web/src/components/CommandPalette.tsx b/apps/web/src/components/CommandPalette.tsx index 4d6de58529310..f2e3a1df68ffb 100644 --- a/apps/web/src/components/CommandPalette.tsx +++ b/apps/web/src/components/CommandPalette.tsx @@ -54,7 +54,7 @@ import { PRIMARY_LOCAL_ENVIRONMENT_ID, resolveEnvironmentMachineKind, } from "@t3tools/contracts"; -import { useLocation, useNavigate, useParams } from "@tanstack/react-router"; +import { useLocation, useNavigate, useParams, useRouter } from "@tanstack/react-router"; import * as Option from "effect/Option"; import { ArrowLeftIcon, @@ -599,6 +599,7 @@ function OpenCommandPaletteDialog(props: { readonly clearOpenIntent: () => void; }) { const navigate = useNavigate(); + const router = useRouter(); const pathname = useLocation({ select: (location) => location.pathname }); const { clearOpenIntent, openIntent, openOverlayMode, setOpen } = props; const [query, setQuery] = useState(openIntent?.kind === "search" ? openIntent.query : ""); @@ -642,6 +643,14 @@ function OpenCommandPaletteDialog(props: { const desktopLocalBootstraps = useDesktopLocalBootstraps(); const primaryEnvironmentId = usePrimaryEnvironmentId(); const availableSettingsSearchItems = useAvailableSettingsSearchItems(); + const settingsIntent = + highlightedItemValue === "action:settings" + ? "/settings/general" + : availableSettingsSearchItems.find((item) => `setting:${item.id}` === highlightedItemValue) + ?.to; + useEffect(() => { + if (settingsIntent) void router.preloadRoute({ to: settingsIntent }).catch(() => undefined); + }, [router, settingsIntent]); const { activeDraftThread, activeThread, defaultProjectRef, handleNewThread } = useHandleNewThread(); const paletteRouteTarget = useParams({ @@ -763,13 +772,7 @@ function OpenCommandPaletteDialog(props: { [threadSearch.matches], ); const [browseGeneration, setBrowseGeneration] = useState(0); - const browseNavigationRef = useRef | null>( - null, - ); - if (browseNavigationRef.current === null) { - browseNavigationRef.current = createBrowseNavigationCoordinator(); - } - const browseNavigation = browseNavigationRef.current; + const [browseNavigation] = useState(createBrowseNavigationCoordinator); const [addProjectEnvironmentId, setAddProjectEnvironmentId] = useState( null, ); @@ -1744,7 +1747,9 @@ function OpenCommandPaletteDialog(props: { setViewStack([]); setLinkedThreadSearch(openIntent); setQuery(openIntent.query); - clearOpenIntent(); + if (document.activeElement?.matches("#t3-command-palette input[role=combobox]")) { + clearOpenIntent(); + } }, [browseNavigation, clearOpenIntent, openIntent]); useLayoutEffect(() => { @@ -3301,6 +3306,7 @@ function OpenCommandPaletteDialog(props: { handleAddProjectForEnvironment, isPickingProjectFolder, primaryEnvironmentId, + setIsPickingProjectFolder, ]); const inputAccessory = @@ -3468,6 +3474,9 @@ function OpenCommandPaletteDialog(props: { ? { startAddon: } : {}), onKeyDown: handleKeyDown, + onFocus: () => { + if (openIntent?.kind === "search") clearOpenIntent(); + }, }} mode="none" items={resultRows.itemValues} diff --git a/apps/web/src/components/CommandPaletteContent.tsx b/apps/web/src/components/CommandPaletteContent.tsx index 8592df68ab570..e341f4cc7a7d0 100644 --- a/apps/web/src/components/CommandPaletteContent.tsx +++ b/apps/web/src/components/CommandPaletteContent.tsx @@ -33,8 +33,10 @@ export function CommandPaletteContent({ // Direct-open flows replace the initial palette view after the dialog has // already moved focus. Reclaim it when the replacement input mounts so // typing cannot continue in the composer behind the modal. + // Focus in the mount commit, inside the opening gesture: mobile browsers + // raise the keyboard only then, and keys typed meanwhile reach the input. useLayoutEffect(() => { - inputRef.current?.focus(); + inputRef.current?.focus({ preventScroll: true }); }, []); return ( diff --git a/apps/web/src/components/CommandPaletteResults.tsx b/apps/web/src/components/CommandPaletteResults.tsx index fe480e2a0d1ec..03e53da630ae0 100644 --- a/apps/web/src/components/CommandPaletteResults.tsx +++ b/apps/web/src/components/CommandPaletteResults.tsx @@ -1,7 +1,7 @@ import { type ResolvedKeybindingsConfig } from "@t3tools/contracts"; import { LegendList, type LegendListRef } from "@legendapp/list/react"; import { ChevronRightIcon } from "lucide-react"; -import { type RefObject, useCallback, useState } from "react"; +import { type RefObject, memo, useCallback, useState } from "react"; import { cn } from "~/lib/utils"; import { shortcutLabelForCommand } from "../keybindings"; import { @@ -23,6 +23,9 @@ import { import { getVirtualizedScrollFadeClassName } from "./ui/scroll-area"; import { ThreadSearchMatchExcerpt } from "./ThreadSearchMatch"; +const paletteRowKey = (row: CommandPaletteRow) => row.key; +const paletteRowType = (row: CommandPaletteRow) => row.kind; + interface CommandPaletteResultsProps { emptyStateMessage?: string; groups: ReadonlyArray; @@ -107,6 +110,26 @@ export function CommandPaletteVirtualizedResults(props: CommandPaletteVirtualize ); }, [listRef]); + const renderItem = useCallback( + ({ item: row }: { item: CommandPaletteRow }) => + row.kind === "label" ? ( +
+ {row.label} +
+ ) : row.itemIndex === null ? ( + + ) : ( + + ), + [props.highlightedItemValue, props.keybindings, props.onExecuteItem], + ); + if (props.rows.length === 0) { return ; } @@ -116,28 +139,12 @@ export function CommandPaletteVirtualizedResults(props: CommandPaletteVirtualize ref={listRef} data={props.rows} - keyExtractor={(row) => row.key} - getItemType={(row) => row.kind} + keyExtractor={paletteRowKey} + getItemType={paletteRowType} extraData={props.highlightedItemValue} - renderItem={({ item: row }) => - row.kind === "label" ? ( -
- {row.label} -
- ) : row.itemIndex === null ? ( - - ) : ( - - ) - } + renderItem={renderItem} estimatedItemSize={40} - drawDistance={400} + drawDistance={120} onLayout={updateScrollFade} onScroll={updateScrollFade} contentContainerClassName="px-2" @@ -182,7 +189,7 @@ function DisabledCommandPaletteResultRow(props: { ); } -function CommandPaletteResultRow(props: { +const CommandPaletteResultRow = memo(function CommandPaletteResultRow(props: { index?: number; item: CommandPaletteActionItem | CommandPaletteSubmenuItem; isActive: boolean; @@ -239,4 +246,4 @@ function CommandPaletteResultRow(props: { ) : null} ); -} +}); diff --git a/apps/web/src/components/ComposerPromptEditorTiptap.tsx b/apps/web/src/components/ComposerPromptEditorTiptap.tsx index 1dc8037453943..cb4cc31e26514 100644 --- a/apps/web/src/components/ComposerPromptEditorTiptap.tsx +++ b/apps/web/src/components/ComposerPromptEditorTiptap.tsx @@ -47,7 +47,6 @@ import { clampCollapsedComposerCursor, collapseExpandedComposerCursor, expandCollapsedComposerCursor, - isCollapsedCursorAdjacentToInlineToken, pastedPathQueryLength, } from "~/composer-logic"; import { @@ -134,7 +133,21 @@ export interface ComposerPromptEditorHandle { export interface ComposerPromptEditorProps { value: string; cursor: number; - /** Resets undo history when a kept editor switches threads. */ + /** + * Changes when the value was rewritten outside the editor. A rewrite can equal the last + * rendered value while the editor holds newer typing, so the new version still applies it. + */ + valueVersion?: number; + /** + * Reads the prompt's current source of truth. A concurrent render can commit + * a value read before the latest keystroke; while the source already equals + * the editor, that older value is not applied. + */ + readValue?: (() => string) | undefined; + /** + * The thread the prompt belongs to. A kept chat view that switches threads + * keeps its editor, and a new scope starts a new undo history. + */ historyScope?: string; /** * Render Markdown styling (bold, italic, code, strike, task checkboxes). @@ -798,6 +811,8 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { suggestionListId, activeSuggestionId, isPathQueryActive, + valueVersion, + readValue, containerClassName, className, placeholderClassName, @@ -886,6 +901,7 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { contextIds: literalText ? [] : collectInlineContextIds(value), }); const selectionRangeRef = useRef({ start: initialExpandedCursor, end: initialExpandedCursor }); + const initialContentRef = useRef | null>(null); const isApplyingControlledUpdateRef = useRef(false); const hasAppliedControlledSelectionRef = useRef(false); const citationRequestRef = useRef(null); @@ -922,7 +938,10 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { const fromFlat = pmToFlat(map, from); const toFlat = pmToFlat(map, to); const nextValue = map.value; - const nextCursor = clampEditorCursor(map.value, flatToCollapsed(map, fromFlat)); + // The doc cursor counts typed `@name` text by its characters. The prompt counts it as one + // token, as the controlled effect below does, so report the clamped cursor. + const docCursor = flatToCollapsed(map, fromFlat); + const nextCursor = clampEditorCursor(nextValue, docCursor); const nextExpandedCursor = Math.max( 0, Math.min(map.value.length, flatToMarkdown(map, fromFlat)), @@ -955,6 +974,7 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { if (previousSnapshot.value === nextValue && nextValue !== latestValueRef.current) { return; } + latestValueRef.current = nextValue; snapshotRef.current = { value: nextValue, cursor: nextCursor, @@ -965,8 +985,11 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { const suppressTrigger = literalText || inCodeBlock || - isCollapsedCursorAdjacentToInlineToken(nextValue, nextCursor, "left") || - isCollapsedCursorAdjacentToInlineToken(nextValue, nextCursor, "right"); + map.runs.some((run) => { + if (run.kind !== "token") return false; + const start = run.collapsedStart + run.openLen; + return docCursor === start || docCursor === start + run.docLen; + }); onChangeRef.current(nextValue, nextCursor, nextExpandedCursor, suppressTrigger, contextIds); }, [clampEditorCursor, literalText], @@ -1108,7 +1131,7 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { ] : []), ], - content: buildDocJson( + content: (initialContentRef.current ??= buildDocJson( value, (name) => { const normalized = name.startsWith("$") ? name.slice(1) : name; @@ -1126,7 +1149,7 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { }; }, { styling: richText, literalText }, - ), + )), editable: !disabled, editorProps: { attributes: editorAttributes, @@ -1432,11 +1455,10 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { return true; }, }, - onUpdate: ({ editor: updated }) => { - handleEditorChange(updated); - }, - onSelectionUpdate: ({ editor: updated }) => { - handleEditorChange(updated); + onTransaction: ({ editor: updated, transaction, appendedTransactions }) => { + if ([transaction, ...appendedTransactions].some((tr) => tr.docChanged || tr.selectionSet)) { + handleEditorChange(updated); + } }, }, [], @@ -1528,6 +1550,16 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { ) { return; } + if ( + !initialSelection && + previousSnapshot.value !== value && + readValue?.() === previousSnapshot.value + ) { + // The editor's text is the latest, so selection updates and focusAt + // must not wait for the older value to be applied. + latestValueRef.current = previousSnapshot.value; + return; + } const normalizedExpandedCursor = expandEditorCursor(value, normalizedCursor); snapshotRef.current = { value, @@ -1589,9 +1621,12 @@ function ComposerPromptEditorTiptapInner(props: ComposerPromptEditorProps) { editor, expandEditorCursor, literalText, + readValue, richText, skillLabelFor, value, + // oxlint-disable-next-line react/exhaustive-effect-dependencies -- An outside rewrite can equal the last rendered value. + valueVersion, ]); // Runs after the effect above has loaded the new thread's prompt, so undo diff --git a/apps/web/src/components/DiffPanel.tsx b/apps/web/src/components/DiffPanel.tsx index f2952e471af82..4a6195233fc7e 100644 --- a/apps/web/src/components/DiffPanel.tsx +++ b/apps/web/src/components/DiffPanel.tsx @@ -44,7 +44,6 @@ import { import { PREFERRED_HIGHLIGHTER } from "../lib/syntaxHighlighting"; import { areAllDiffFilesCollapsed, toggleAllDiffFiles } from "../lib/diffCollapse"; import { useTurnDiffSummaries } from "../hooks/useTurnDiffSummaries"; -import { useWorkspaceMutationRefresh } from "../hooks/useWorkspaceMutationRefresh"; import { useProject, useThreadProjection, useThreadShell } from "../state/entities"; import { useClientSettings, useUpdateClientSettings } from "../hooks/useSettings"; import { formatShortTimestamp } from "../timestampFormat"; @@ -81,7 +80,7 @@ import { Tooltip, TooltipPopup, TooltipTrigger } from "./ui/tooltip"; import { useEnvironmentQuery } from "../state/query"; import { useAtomCommand } from "../state/use-atom-command"; import { serverEnvironment } from "../state/server"; -import { reviewEnvironment } from "../state/review"; +import { reviewEnvironment, reviewWorkspaceRevision } from "../state/review"; import { vcsEnvironment } from "../state/vcs"; import { buildBaseRefChoices, filterBaseRefChoices } from "../lib/baseRefChoices"; import { createGitDiffFileContentsLoader } from "../lib/diffFileContents"; @@ -183,6 +182,12 @@ function DiffFileHeaderSuffix({ ); } +// Preserve explicit refresh epochs across short panel remounts. +const diffRefreshVersions = new Map(); +// The cache key each review scope's preview last loaded, so a workspace +// change while the panel was closed still refreshes it on reopen. +const loadedPreviewCacheKeys = new Map(); + interface DiffPanelProps { mode?: DiffPanelMode; composerDraftTarget: ScopedThreadRef | DraftId; @@ -334,8 +339,36 @@ export default function DiffPanel({ }, { enabled: isGitRepo && selectedTurn !== undefined }, ); - const branchDiffPreview = useEnvironmentQuery( - canReadFiles && selectedRunId === null && activeThread && activeCwd + const reviewScope = JSON.stringify([ + activeThread?.environmentId, + activeCwd, + selectedBaseRef, + diffIgnoreWhitespace, + ]); + const [refreshVersion, setRefreshVersion] = useState({ + scope: reviewScope, + version: diffRefreshVersions.get(reviewScope) ?? 0, + }); + const version = + refreshVersion.scope === reviewScope + ? refreshVersion.version + : (diffRefreshVersions.get(reviewScope) ?? 0); + const savedFileRevision = useAtomValue( + reviewWorkspaceRevision(activeThread?.environmentId, activeCwd ?? undefined), + ); + const reviewCacheKey = JSON.stringify([workspaceMutationId, savedFileRevision, version]); + const refreshBranchDiffPreview = useCallback(() => { + const next = Math.max(Date.now(), (diffRefreshVersions.get(reviewScope) ?? 0) + 1); + if (diffRefreshVersions.size >= 64 && !diffRefreshVersions.has(reviewScope)) + diffRefreshVersions.delete(diffRefreshVersions.keys().next().value!); + diffRefreshVersions.set(reviewScope, next); + setRefreshVersion({ scope: reviewScope, version: next }); + }, [reviewScope, setRefreshVersion]); + const previewEnabled = canReadFiles && selectedRunId === null && activeThread && activeCwd; + // One preview per scope: a new cache key refreshes it in place, so a + // superseded preview is never kept beside the current one. + const requestedBranchDiffPreview = useEnvironmentQuery( + previewEnabled ? reviewEnvironment.diffPreview({ environmentId: activeThread.environmentId, input: { @@ -346,17 +379,40 @@ export default function DiffPanel({ }) : null, ); + const refreshPreviewQuery = requestedBranchDiffPreview.refresh; + useEffect(() => { + if (!previewEnabled) return; + const loaded = loadedPreviewCacheKeys.get(reviewScope); + if (loaded === reviewCacheKey) return; + if (loadedPreviewCacheKeys.size >= 64 && loaded === undefined) + loadedPreviewCacheKeys.delete(loadedPreviewCacheKeys.keys().next().value!); + loadedPreviewCacheKeys.set(reviewScope, reviewCacheKey); + if (loaded !== undefined) refreshPreviewQuery(); + }, [previewEnabled, refreshPreviewQuery, reviewCacheKey, reviewScope]); + const [previousPreview, setPreviousPreview] = useState({ + scope: reviewScope, + data: requestedBranchDiffPreview.data, + }); + if ( + requestedBranchDiffPreview.data !== null && + (previousPreview.scope !== reviewScope || + previousPreview.data !== requestedBranchDiffPreview.data) + ) { + setPreviousPreview({ scope: reviewScope, data: requestedBranchDiffPreview.data }); + } + const branchDiffPreview = { + ...requestedBranchDiffPreview, + data: + requestedBranchDiffPreview.data ?? + (previousPreview.scope === reviewScope ? previousPreview.data : null), + }; const canRefreshGitDiff = isGitRepo && selectedRunId === null && activeThread != null && activeCwd != null; - const activeThreadRefreshKey = routeThreadRef - ? `${routeThreadRef.environmentId}:${routeThreadRef.threadId}` - : null; const selectedGitSource = branchDiffPreview.data?.sources.find( (source) => source.kind === (selectedGitScope === "unstaged" ? "working-tree" : "branch-range"), ); - const refreshPreviewQuery = branchDiffPreview.refresh; - const refreshDiffFromUserAction = refreshPreviewQuery; + const refreshDiffFromUserAction = refreshBranchDiffPreview; const currentLoadDiffFiles = useMemo(() => { const preview = branchDiffPreview.data; @@ -485,7 +541,6 @@ export default function DiffPanel({ : undefined, preview: renderablePatch, }); - const refreshBranchDiffPreview = refreshPreviewQuery; useEffect(() => { if (!canRefreshGitDiff) return; @@ -494,13 +549,6 @@ export default function DiffPanel({ return () => window.removeEventListener("focus", refreshOnFocus); }, [canRefreshGitDiff, refreshBranchDiffPreview]); - useWorkspaceMutationRefresh({ - enabled: canRefreshGitDiff, - mutationId: workspaceMutationId, - refresh: refreshBranchDiffPreview, - resourceKey: `diff:${activeThreadRefreshKey ?? ""}`, - }); - const isRefreshingDiff = branchDiffPreview.isPending || areFilePatchesPending; const renderableFileEntries = useMemo( () => renderableFiles.map(getCachedFileEntry), diff --git a/apps/web/src/components/DiffWorkerPoolProvider.tsx b/apps/web/src/components/DiffWorkerPoolProvider.tsx index cf83d33d62dce..5632824d37584 100644 --- a/apps/web/src/components/DiffWorkerPoolProvider.tsx +++ b/apps/web/src/components/DiffWorkerPoolProvider.tsx @@ -2,14 +2,7 @@ import { WorkerPoolContext, useWorkerPool } from "@pierre/diffs/react"; import { WorkerPoolManager } from "@pierre/diffs/worker"; import DiffsWorker from "@pierre/diffs/worker/worker.js?worker"; import * as Schema from "effect/Schema"; -import { - useCallback, - useEffect, - useMemo, - useState, - useSyncExternalStore, - type ReactNode, -} from "react"; +import { useCallback, useEffect, useState, useSyncExternalStore, type ReactNode } from "react"; import { useTheme } from "../hooks/useTheme"; import { resolveDiffThemeName, type DiffThemeName } from "../lib/diffRendering"; import { PREFERRED_HIGHLIGHTER } from "../lib/syntaxHighlighting"; @@ -33,6 +26,13 @@ let sharedWorkerPool: } | undefined; +function diffWorkerPoolSize() { + const cores = + typeof navigator === "undefined" ? 4 : Math.max(1, navigator.hardwareConcurrency || 4); + return Math.max(2, Math.min(6, Math.floor(cores / 2))); +} + +/** Reuse workers across viewer intent and short panel closures. */ function acquireDiffWorkerPool(themeName: DiffThemeName, poolSize: number) { const entry = (sharedWorkerPool ??= { pool: new WorkerPoolManager( @@ -63,6 +63,26 @@ function acquireDiffWorkerPool(themeName: DiffThemeName, poolSize: number) { return entry; } +function releaseDiffWorkerPool(entry: NonNullable) { + entry.consumers -= 1; + if (entry.consumers !== 0) return; + entry.idleTimer = setTimeout(() => { + entry.idleTimer = undefined; + entry.pool.terminate(); + if (sharedWorkerPool === entry) sharedWorkerPool = undefined; + }, DIFF_WORKER_IDLE_TTL_MS); +} + +/** File picking is viewer intent; overlap worker assets with the file search. */ +export async function preloadDiffWorkerPool(themeName: DiffThemeName) { + const entry = acquireDiffWorkerPool(themeName, diffWorkerPoolSize()); + try { + await entry.pool.initialize(); + } finally { + releaseDiffWorkerPool(entry); + } +} + function DiffWorkerThemeSync({ themeName }: { themeName: DiffThemeName }) { const workerPool = useWorkerPool(); @@ -129,26 +149,14 @@ function DiffWorkerReady({ children }: { children?: ReactNode }) { export function DiffWorkerPoolProvider({ children }: { children?: ReactNode }) { const { resolvedTheme } = useTheme(); const diffThemeName = resolveDiffThemeName(resolvedTheme); - const workerPoolSize = useMemo(() => { - const cores = - typeof navigator === "undefined" ? 4 : Math.max(1, navigator.hardwareConcurrency || 4); - return Math.max(2, Math.min(6, Math.floor(cores / 2))); - }, []); + const workerPoolSize = diffWorkerPoolSize(); const workerPool = useSyncExternalStore( useCallback( (onStoreChange) => { if (typeof window === "undefined") return () => {}; const entry = acquireDiffWorkerPool(diffThemeName, workerPoolSize); onStoreChange(); - return () => { - entry.consumers -= 1; - if (entry.consumers !== 0) return; - entry.idleTimer = setTimeout(() => { - entry.idleTimer = undefined; - entry.pool.terminate(); - if (sharedWorkerPool === entry) sharedWorkerPool = undefined; - }, DIFF_WORKER_IDLE_TTL_MS); - }; + return () => releaseDiffWorkerPool(entry); }, [diffThemeName, workerPoolSize], ), diff --git a/apps/web/src/components/PullRequestThreadDialog.test.ts b/apps/web/src/components/PullRequestThreadDialog.test.ts index c852a853da39c..8349c02478c51 100644 --- a/apps/web/src/components/PullRequestThreadDialog.test.ts +++ b/apps/web/src/components/PullRequestThreadDialog.test.ts @@ -37,9 +37,6 @@ vi.mock("react", async (importOriginal) => ({ useRef: (current: unknown) => ({ current }), useEffect: () => {}, })); -vi.mock("@tanstack/react-pacer", () => ({ - useDebouncedValue: (value: unknown) => [value, { state: { isPending: false } }], -})); vi.mock("~/state/session", () => ({ useEnvironmentScope: (_environmentId: unknown, scope: string) => scope === AuthOrchestrationOperateScope diff --git a/apps/web/src/components/PullRequestThreadDialog.tsx b/apps/web/src/components/PullRequestThreadDialog.tsx index 47e4b361aaeeb..3f3bb7ed652fc 100644 --- a/apps/web/src/components/PullRequestThreadDialog.tsx +++ b/apps/web/src/components/PullRequestThreadDialog.tsx @@ -7,7 +7,6 @@ import { isAtomCommandInterrupted, squashAtomCommandFailure, } from "@t3tools/client-runtime/state/runtime"; -import { useDebouncedValue } from "@tanstack/react-pacer"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { @@ -58,11 +57,13 @@ export function PullRequestThreadDialog({ const [referenceDirty, setReferenceDirty] = useState(false); const [preparingMode, setPreparingMode] = useState<"local" | "worktree" | null>(null); const [prepareErrorMessage, setPrepareErrorMessage] = useState(null); - const [debouncedReference, referenceDebouncer] = useDebouncedValue( - reference, - { wait: 450 }, - (debouncerState) => ({ isPending: debouncerState.isPending }), - ); + const [debouncedReference, setDebouncedReference] = useState(reference); + useEffect(() => { + const timer = window.setTimeout(() => { + setDebouncedReference(reference); + }, 450); + return () => window.clearTimeout(timer); + }, [reference]); const { data: gitStatus } = useEnvironmentQuery( cwd === null ? null @@ -122,7 +123,7 @@ export function PullRequestThreadDialog({ open && parsedReference !== null && resolvedPullRequest === null && - (referenceDebouncer.state.isPending || + (reference !== debouncedReference || parsedReference !== parsedDebouncedReference || pullRequestResolution.isPending || pullRequestResolution.isFetching); diff --git a/apps/web/src/components/RightPanelTabs.tsx b/apps/web/src/components/RightPanelTabs.tsx index 59a3dc6532090..0d933160e5ba2 100644 --- a/apps/web/src/components/RightPanelTabs.tsx +++ b/apps/web/src/components/RightPanelTabs.tsx @@ -73,7 +73,6 @@ import { MenuTrigger, } from "~/components/ui/menu"; import { useBrowserDefaults } from "~/browser/browserDefaults"; -import { ScrollArea } from "~/components/ui/scroll-area"; import { PanelTabCloseButton } from "~/components/ui/panel-tab-close-button"; import { faviconUrlForOrigin } from "~/lib/favicon"; import { useTheme } from "~/hooks/useTheme"; @@ -211,7 +210,7 @@ const TAB_SCROLL_EDGE_TOLERANCE = 1; const TAB_DRAG_MODIFIERS = [restrictToHorizontalAxis, restrictToFirstScrollableAncestor]; function tabScrollViewport(root: HTMLDivElement | null): HTMLDivElement | null { - return root?.querySelector('[data-slot="scroll-area-viewport"]') ?? null; + return root; } function previewTabIdOf( @@ -1121,12 +1120,15 @@ export function RightPanelTabs(props: RightPanelTabsProps) { )} data-right-panel-tabbar > -
@@ -1338,7 +1340,7 @@ export function RightPanelTabs(props: RightPanelTabsProps) { ) : null}
- + {tabScrollState.hasOverflow ? (
| null>(null); + +/** One floating popup serves every row and its quick actions. */ +function TooltipProvider({ children, ...props }: ComponentProps) { + const [handle] = useState(() => TooltipPrimitive.createHandle()); + useEffect(() => { + // Only a scroll that moves the open trigger dismisses it; the chat + // timeline scrolls on its own while a reply streams. + const close = (event: Event) => { + if (!handle.isOpen) return; + const trigger = document.querySelector('[data-slot="tooltip-trigger"][data-popup-open]'); + if (!(event.target instanceof Node) || !trigger || !event.target.contains(trigger)) return; + if (trigger === document.activeElement) return; + handle.close(); + }; + document.addEventListener("scroll", close, true); + return () => document.removeEventListener("scroll", close, true); + }, [handle]); + return ( + + + {children} + {({ payload }) => payload} + + + ); +} + +function Tooltip({ + children, + disabled = false, +}: { + children: ReactNode; + disabled?: boolean | undefined; +}) { + const handle = use(SidebarTooltipHandle); + useEffect(() => { + if (disabled) handle?.close(); + }, [disabled, handle]); + const parts = Children.toArray(children); + const trigger = parts.find((child) => isValidElement(child) && child.type === TooltipTrigger); + if (handle === null || !isValidElement>(trigger)) + return null; + const payload = parts.filter((child) => child !== trigger); + return cloneElement(trigger, { handle, payload, disabled }); +} + function compactSidebarTimeLabel(label: string): string { if (label === "just now") return "now"; return label.endsWith(" ago") ? label.slice(0, -4) : label; @@ -1046,10 +1111,7 @@ function SnoozeMenuButton(props: { ); } -type SortableThreadRowBag = Pick< - ReturnType, - "listeners" | "setNodeRef" | "transform" | "transition" | "isDragging" ->; +type SortableThreadRowBag = { isDragging: boolean }; // useSortable memoizes on its `disabled` object, so a fresh one per render // re-renders every row and marker on each sidebar update. @@ -1058,6 +1120,10 @@ const SIDEBAR_SORTABLE_DRAG_ENABLED = { draggable: false } as const; function SortableThreadRow(props: { id: string; + title: string; + variant: "card" | "slim"; + keepMounted: boolean; + initiallyVisible: boolean; disabled: boolean; contextDrag: boolean; environmentId: EnvironmentId; @@ -1072,16 +1138,60 @@ function SortableThreadRow(props: { : SIDEBAR_SORTABLE_DRAG_ENABLED, animateLayoutChanges: animateSidebarLayoutChanges, }); - // dnd-kit memoizes each field but not the bag, so the memoized row would - // rerender on every shell update without this. - const bag = { - listeners, - setNodeRef, - transform: props.contextDrag ? null : transform, - transition: props.contextDrag && isDragging ? "transform 150ms ease-out" : transition, - isDragging: isDragging && !props.contextDrag, + const [node, setNode] = useState(null); + const [visible, setVisible] = useSidebarRowViewport(node, props.initiallyVisible); + const focusedRef = useRef(false); + const attach = (element: HTMLElement | null) => { + setNodeRef(element); + setNode(element); }; - return props.children(bag); + useLayoutEffect(() => { + // The swap commit still holds the detached placeholder; wait for the row. + if (!visible || !focusedRef.current || !node?.isConnected) return; + focusedRef.current = false; + node.querySelector('[tabindex="0"]')?.focus(); + }, [visible, node]); + const bag = { isDragging: isDragging && !props.contextDrag }; + const rowTransform = props.contextDrag ? null : transform; + const mounted = + visible || props.keepMounted || isDragging || node?.contains(document.activeElement); + return ( +
  • + {mounted ? ( + props.children(bag) + ) : ( +
    { + focusedRef.current = true; + setVisible(true); + }} + /> + )} +
    • + ); } const draftSurfaceClassName = "bg-warning/4 hover:bg-warning/8"; @@ -1818,22 +1928,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { "bg-sidebar bg-linear-to-b from-sidebar-row-active to-sidebar-row-active text-sidebar-foreground opacity-100 shadow-lg", ); const sortable = props.sortable; - const sortableRootProps = sortable - ? { - ref: sortable.setNodeRef, - style: { - transform: CSS.Translate.toString(sortable.transform), - transition: sortable.transition, - // A zero-height boundary also makes dnd-kit scale the source to - // zero. Only projected peers use scaleY as a visibility sentinel. - visibility: - !sortable.isDragging && sortable.transform?.scaleY === 0 - ? ("hidden" as const) - : undefined, - }, - ...sortable.listeners, - } - : {}; + // Sweeps reuse the corresponding row-drop action badge. const destinationVerb = sortable?.isDragging ? props.dropVerb : props.sweepAction === "unsnooze" @@ -1986,20 +2081,13 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { if (variant === "slim") { return ( -
  • + <> {detailsTooltip} -
    • + ); } const diff = latestRunDiff(thread); return ( -
  • + <> {detailsTooltip} -
    • + ); }); @@ -2406,7 +2487,7 @@ function latestRunDiff( return null; } -const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props: { +type SidebarSearchResultRowProps = { isSettled: boolean; thread: SidebarThreadSummary; project: EnvironmentProject | null; @@ -2419,10 +2500,49 @@ const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props: { resultId: string; searchMatch: EnvironmentThreadSearchMatch | null; searchQuery: string; - onHighlight: () => void; - onSelect: () => void; + index: number; + onHighlight: (index: number) => void; + onSelect: (thread: SidebarThreadSummary) => void; onFileDropThreads: (threadRef: ScopedThreadRef, files: File[]) => void; -}) { +}; + +const SidebarSearchResultRow = memo(function SidebarSearchResultRow( + props: SidebarSearchResultRowProps, +) { + const [node, setNode] = useState(null); + const [visible] = useSidebarRowViewport(node, props.index < 24); + const keepMounted = props.isHighlighted || props.isRouteActive; + const [height, setHeight] = useState(props.searchMatch ? "3.75rem" : "2.25rem"); + useEffect(() => { + if (!node || (!visible && !keepMounted)) return; + return observeResize(node, ([entry]) => { + const size = entry?.borderBoxSize[0]?.blockSize; + if (size !== undefined && size > 0) { + const rem = Number.parseFloat( + getComputedStyle(node.ownerDocument.documentElement).fontSize, + ); + setHeight(`${size / rem}rem`); + } + }); + }, [node, visible, keepMounted]); + return visible || keepMounted ? ( + + ) : ( +
  • +
    +
    • + ); +}); + +const SidebarSearchResultContent = memo(function SidebarSearchResultContent({ + setSlotNode, + ...props +}: SidebarSearchResultRowProps & { setSlotNode: (node: HTMLLIElement | null) => void }) { const { thread } = props; const accessibility = resolveSidebarRowAccessibility({ title: thread.title, @@ -2485,7 +2605,7 @@ const SidebarSearchResultRow = memo(function SidebarSearchResultRow(props: { return () => window.removeEventListener("dragend", clearFileDrag); }, [isFileDragOver]); return ( -
  • +
  • props.onHighlight(props.index)} + onClick={() => props.onSelect(thread)} className={cn( "group/sidebar-row group/v2-row flex min-h-9 w-full cursor-pointer items-center gap-2.5 rounded-md px-2.5 py-1 text-left text-sm outline-none", props.isHighlighted || props.isRouteActive @@ -4630,8 +4750,9 @@ export default function Sidebar() { ) ?? null } searchQuery={threadSearchQuery} - onHighlight={() => setActiveSearchResultIndex(index)} - onSelect={() => selectThreadSearchResult(thread)} + index={index} + onHighlight={setActiveSearchResultIndex} + onSelect={selectThreadSearchResult} onFileDropThreads={handleThreadFileDrop} /> ); @@ -4782,17 +4903,29 @@ export default function Sidebar() { /> ); }; + let rowIndex = 0; + let rowTopRem = 0; + const initialRowBudgetRem = sidebarInitialRowBudgetRem(); const renderThreadRow = ( thread: EnvironmentThreadShell, section: SidebarSection, + initiallyVisible: boolean, ) => { const threadKey = scopedThreadKey( scopeThreadRef(thread.environmentId, thread.id), ); + const variant = + section === "settled" || section === "snoozed" ? "slim" : "card"; return ( void>; + } +>(); +// Boot uses the budget; rows added to a painted list need their actual position. +const paintedRoots = new WeakSet(); + +/** Share viewport observation across row slots; expensive content mounts only nearby. */ +export function useSidebarRowViewport(node: HTMLElement | null, initiallyVisible: boolean) { + const [visible, setVisible] = useState( + initiallyVisible || typeof IntersectionObserver === "undefined", + ); + const placedRef = useRef(false); + useLayoutEffect(() => { + // An opened shelf can join a scrolled list; mount its visible rows before paint. + if (!node || placedRef.current) return; + placedRef.current = true; + const root = node.closest('[data-slot="scroll-area-viewport"]'); + if (visible || !root || !paintedRoots.has(root)) return; + const bounds = root.getBoundingClientRect(); + const box = node.getBoundingClientRect(); + if ( + bounds.height > 0 && + box.bottom > bounds.top - ROW_OBSERVER_MARGIN_PX && + box.top < bounds.bottom + ROW_OBSERVER_MARGIN_PX + ) { + // oxlint-disable-next-line react/set-state-in-effect -- The slot must render its row before the first paint. + setVisible(true); + } + }, [node, visible]); + useEffect(() => { + if (!node || typeof IntersectionObserver === "undefined") return; + const root = node.closest('[data-slot="scroll-area-viewport"]'); + let group = observers.get(root); + if (!group) { + const listeners = new Map void>(); + const observer = new IntersectionObserver( + (entries) => { + if (root) paintedRoots.add(root); + for (const entry of entries) listeners.get(entry.target)?.(entry.isIntersecting); + }, + { root, rootMargin: `${ROW_OBSERVER_MARGIN_PX}px 0px` }, + ); + group = { observer, listeners }; + observers.set(root, group); + } + group.listeners.set(node, setVisible); + group.observer.observe(node); + return () => { + group.listeners.delete(node); + group.observer.unobserve(node); + if (group.listeners.size === 0) { + group.observer.disconnect(); + observers.delete(root); + } + }; + }, [node]); + return [visible, setVisible] as const; +} diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 4622b146e3b83..bfc13c9f09e4d 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -1,3 +1,4 @@ +import { useShallow } from "zustand/react/shallow"; import { formatProviderSkillDisplayName } from "@t3tools/shared/inlineSkills"; import { useComposerTypingGuard } from "./useComposerTypingGuard"; import { DESKTOP_PASTE_AS_TEXT_EVENT } from "../../lib/desktopPasteAsText"; @@ -123,7 +124,8 @@ import { composerTargetKey, hydrateImagesFromPersisted, useComposerDraftStore, - useComposerThreadDraft, + useComposerThreadDraftSelector, + type ComposerThreadTarget, useEffectiveComposerModelState, } from "../../composerDraftStore"; import { @@ -213,6 +215,7 @@ import { measureRestingComposerControls } from "./restingComposerControlsMeasure import { type ComposerCitationCommentRequest, type ComposerPromptEditorHandle, + type ComposerPromptEditorProps, ComposerPromptEditor, } from "../ComposerPromptEditor"; import { @@ -1420,6 +1423,73 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions( ); }); +function readComposerDraftPrompt(target: ComposerThreadTarget) { + return useComposerDraftStore.getState().getComposerDraft(target)?.prompt ?? ""; +} + +// Local transactions have already painted in ProseMirror, so typing does not render this. +// `promptVersion` moves on every rewrite from outside the editor (recall, chips, stash, send, +// reload). It re-renders this memo, and the editor applies the value even when it equals the +// last rendered one. +const ComposerDraftEditor = memo(function ComposerDraftEditor({ + target, + readCursor, + promptVersion, + value: overriddenValue, + ...props +}: Omit & { + target: ComposerThreadTarget; + readCursor: () => number; + promptVersion: number; + value: string | undefined; +}) { + "use no memo"; // The caret is imperative; caching readCursor() would rewind typed text. + const readValue = () => useComposerDraftStore.getState().getComposerDraft(target)?.prompt ?? ""; + return ( + + ); +}); + +const ComposerDraftPromptText = memo(function ComposerDraftPromptText({ + target, + placeholder, +}: { + target: ComposerThreadTarget; + placeholder: string; +}) { + const prompt = useComposerThreadDraftSelector(target, (draft) => draft.prompt); + return prompt.trim() || placeholder; +}); + +const ComposerDraftTraits = memo(function ComposerDraftTraits({ + target, + menu = false, + ...input +}: Omit[0], "prompt"> & { + target: ComposerThreadTarget; + menu?: boolean; +}) { + "use no memo"; + useComposerThreadDraftSelector( + target, + useShallow((draft) => [ + getComposerPromptInjectionState(draft.prompt), + getComposerPromptInjectionState(draft.prompt.replace(/^Ultrathink:\s*/i, "")), + ]), + ); + const getPrompt = () => useComposerDraftStore.getState().getComposerDraft(target)?.prompt ?? ""; + const prompt = getPrompt(); + return menu + ? renderProviderTraitsMenuContent({ ...input, prompt, getPrompt }) + : renderProviderTraitsPicker({ ...input, prompt, getPrompt }); +}); + export interface ChatComposerHandle { focusAtEnd: () => void; focusAt: (cursor: number) => void; @@ -1501,6 +1571,10 @@ export interface ChatComposerProps { activeThread: Thread | undefined; activeThreadShell: ThreadShell | null; promptHistoryMessages: ReadonlyArray; + /** Loads an older history page; false when there is none to load right now. */ + onLoadOlderPromptHistory?: (() => boolean) | undefined; + /** True once an older history page failed to load. */ + olderPromptHistoryFailed?: boolean | undefined; isServerThread: boolean; isLocalDraftThread: boolean; forceExpandedOnMobile: boolean; @@ -1870,6 +1944,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) activeThreadEnvironmentId: _activeThreadEnvironmentId, activeThread, promptHistoryMessages, + onLoadOlderPromptHistory, + olderPromptHistoryFailed = false, isServerThread: _isServerThread, isLocalDraftThread: _isLocalDraftThread, forceExpandedOnMobile, @@ -1984,7 +2060,25 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const activeTasksProgress = shownSyncPhase === null ? props.activeTasksProgress : null; const activeTaskSteps = shownSyncPhase === null ? props.activeTaskSteps : null; const isEditingQueuedMessage = editingQueuedAttachments !== null; - const composerDraft = useComposerThreadDraft(composerDraftTarget); + // Text belongs to the editor. The frame observes only changes to its presentation. + const composerDraft = useComposerThreadDraftSelector( + composerDraftTarget, + useShallow(({ prompt, ...draft }) => ({ + ...draft, + hasText: prompt.trim().length > 0, + hasMultilineText: prompt.includes("\n"), + promptInjection: getComposerPromptInjectionState(prompt), + inlineContextKey: collectInlineContextIds(prompt).join("\n"), + promptValidation: getComposerPromptLengthValidationMessage(prompt), + hasSendableText: deriveComposerSendState({ + prompt, + imageCount: 0, + terminalContexts: [], + }).hasSendableContent, + })), + ); + // Live target key, for async flows that must notice a thread switch that + // happened while they awaited. const composerDraftTargetKeyRef = useRef(""); useLayoutEffect(() => { composerDraftTargetKeyRef.current = composerDraftTargetKey; @@ -1999,7 +2093,15 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ) : null; const attachmentDraftTarget = questionAttachmentTarget ?? composerDraftTarget; - const attachmentDraft = useComposerThreadDraft(attachmentDraftTarget); + const attachmentDraft = useComposerThreadDraftSelector( + attachmentDraftTarget, + useShallow((draft) => ({ + images: draft.images, + files: draft.files, + threadContexts: draft.threadContexts, + nonPersistedImageIds: draft.nonPersistedImageIds, + })), + ); const attachmentTargetKey = composerTargetKey(attachmentDraftTarget); // An import that finishes after a draft change must compare against the draft open *now*, not // the one captured in the closure that started it. @@ -2008,7 +2110,18 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) attachmentTargetKeyRef.current = attachmentTargetKey; }); const questionPreparations = useQuestionAttachmentPreparation((state) => state.counts); - const prompt = composerDraft.prompt; + const prompt = readComposerDraftPrompt(composerDraftTarget); + // The prompt the editor shows. onPromptChange records a local edit before it reaches the + // store, so the version moves only for rewrites from outside the editor. + const shownPromptRef = useRef(prompt); + const promptVersionRef = useRef(0); + const promptVersion = useComposerThreadDraftSelector(composerDraftTarget, (draft) => { + if (draft.prompt !== shownPromptRef.current) { + shownPromptRef.current = draft.prompt; + promptVersionRef.current += 1; + } + return promptVersionRef.current; + }); const composerImages = attachmentDraft.images; const composerFiles = attachmentDraft.files; const inlineFileIdSet = (() => { @@ -2378,10 +2491,20 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) : null, }); - const [composerCursor, setComposerCursor] = useState(() => { + const [composerCursor, updateComposerCursor] = useState(() => { const value = isLiteralPendingAnswer ? (activePendingProgress?.customAnswer ?? prompt) : prompt; return collapseExpandedComposerCursor(value, value.length, isLiteralPendingAnswer); }); + const composerCursorRef = useRef(composerCursor); + useLayoutEffect(() => { + composerCursorRef.current = composerCursor; + }, [composerCursor]); + const setComposerCursor = useStableCallback((cursor: number | ((current: number) => number)) => { + const next = typeof cursor === "function" ? cursor(composerCursorRef.current) : cursor; + composerCursorRef.current = next; + updateComposerCursor(next); + }); + const readComposerCursor = () => composerCursorRef.current; const { trigger: composerTrigger, setTrigger: setComposerTrigger, @@ -2398,6 +2521,10 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) targetKey: string; position: ComposerPromptHistoryPosition; } | null>(null); + const olderPromptRecallPendingRef = useRef<{ + readonly targetKey: string; + readonly oldestMessageId: string | undefined; + } | null>(null); const [composerHighlightedSearchKey, setComposerHighlightedSearchKey] = useState( null, ); @@ -2419,7 +2546,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ); const [composerMenuAnchor, setComposerMenuAnchor] = useElementState(); const hasWrappedPrompt = useComposerMultilinePrompt(composerMenuAnchor); - const hasMultilinePrompt = prompt.includes("\n") || hasWrappedPrompt; + const hasMultilinePrompt = composerDraft.hasMultilineText || hasWrappedPrompt; const [isStashMenuOpen, setIsStashMenuOpen] = useState(false); const [isTasksDrawerOpen, setIsTasksDrawerOpen] = useState(false); const [stashPulse, setStashPulse] = useState<{ key: number; active: boolean }>({ @@ -2840,19 +2967,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) }; const setPromptFromTraitsStable = useStableCallback(setPromptFromTraits); - const providerTraitsMenuContent = renderProviderTraitsMenuContent({ - provider: selectedProvider, - instanceId: selectedInstanceId, - ...(routeKind === "server" ? { threadRef: routeThreadRef } : {}), - ...(routeKind === "draft" && draftId ? { draftId } : {}), - model: selectedModel, - models: selectedProviderModels, - modelOptions: composerModelOptions?.[selectedInstanceId], - reportedModelSelection, - prompt, - onPromptChange: setPromptFromTraitsStable, - planModeEnabled: settings.planModeEnabled, - }); const providerTraitsPickerInput = { provider: selectedProvider, instanceId: selectedInstanceId, @@ -2862,12 +2976,17 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) models: selectedProviderModels, modelOptions: composerModelOptions?.[selectedInstanceId], reportedModelSelection, - prompt, onPromptChange: setPromptFromTraitsStable, planModeEnabled: settings.planModeEnabled, isComposerOwned: true, - } satisfies Parameters[0]; - const providerTraitsPicker = renderProviderTraitsPicker(providerTraitsPickerInput); + } satisfies Omit[0], "prompt">; + const providerTraitsMenuContent = ( + + ); + const providerTraitsPicker = renderProviderTraitsPicker({ + ...providerTraitsPickerInput, + prompt, + }); const [inlineRestingControlsHost, setInlineRestingControlsHost] = useElementState(); const { @@ -2895,7 +3014,9 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) } : null; const showResumeAction = - canResume && !composerDraftHasUserContent(composerDraft) && !isEditingQueuedMessage; + canResume && + !composerDraftHasUserContent({ ...composerDraft, prompt }) && + !isEditingQueuedMessage; const collapsedComposerPrimaryActionDisabled = phase === "running" || isSendBusy || @@ -3263,7 +3384,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ) { setCursorClampedFor({ prompt, isLiteralPendingAnswer }); if (!isLiteralPendingAnswer) { - setComposerCursor((existing) => clampCollapsedComposerCursor(prompt, existing)); + updateComposerCursor((existing) => clampCollapsedComposerCursor(prompt, existing)); } } @@ -3279,7 +3400,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) composerPreviewAnnotations, composerReviewComments, composerTerminalContexts, - prompt, + promptVersion, selectedModel, selectedPromptEffort, selectedProvider, @@ -3353,6 +3474,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) resetComposerTrigger(trigger); setComposerHighlightedItemId(null); }, [ + setComposerCursor, activePendingProgress?.customAnswer, activePendingProgress?.activeQuestion?.id, isLiteralPendingAnswer, @@ -3375,7 +3497,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) typeof pendingAnswer === "string" ? pendingAnswer : prompt, isLiteralPendingAnswer, ); - setComposerCursor(cursor); + updateComposerCursor(cursor); resetComposerTrigger(trigger); setIsDragOverComposer(false); setIsComposerScrollCollapsed(false); @@ -3536,10 +3658,16 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) } onTypingGuardDraftChange(); promptRef.current = nextPrompt; + shownPromptRef.current = nextPrompt; + composerCursorRef.current = nextCursor; setPrompt(nextPrompt); + const recalled = promptHistoryPositionRef.current?.position.recalled; if (promptHistoryPositionRef.current?.position.recalled !== nextPrompt) { promptHistoryPositionRef.current = null; } + if ((recalled ?? "") !== nextPrompt) { + olderPromptRecallPendingRef.current = null; + } const referenced = new Set(contextIds); const retained = removedContextPayloadsRef.current; @@ -3622,10 +3750,14 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) if (attachmentChanges.filesToRestore.length > 0) { addComposerDraftFiles(attachmentDraftTarget, attachmentChanges.filesToRestore); } - setComposerCursor(nextCursor); setComposerTrigger( cursorAdjacentToMention ? null : detectComposerTrigger(nextPrompt, expandedCursor), ); + setIsStashMenuOpen(false); + setProviderInputSubmissionError(null); + setComposerSubmissionError((current) => + current === null ? null : getComposerPromptLengthValidationMessage(nextPrompt), + ); }; const onPromptChangeStable = useStableCallback(onPromptChange); @@ -3719,10 +3851,10 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) } return { value: promptRef.current, - cursor: composerCursor, + cursor: composerCursorRef.current, expandedCursor: expandCollapsedComposerCursor( promptRef.current, - composerCursor, + composerCursorRef.current, isLiteralPendingAnswer, ), contextIds: isLiteralPendingAnswer ? [] : collectInlineContextIds(promptRef.current), @@ -3988,6 +4120,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) }); setComposerSubmissionError(submission.validationMessage); if (!submission.didDispatch) return; + olderPromptRecallPendingRef.current = null; onTypingGuardSend(); if (shouldBlurMobileComposerOnSubmit()) { blurMobileComposerAfterSend(); @@ -4088,6 +4221,15 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) setComposerHighlightedItemId(null); }; + const promptHistoryBlocked = + isComposerApprovalState || + pendingUserInputs.length > 0 || + composerImages.length > 0 || + composerFiles.length > 0 || + composerTerminalContexts.length > 0 || + composerPreviewAnnotations.length > 0 || + composerReviewComments.length > 0; + const navigatePromptHistory = ( direction: "backward" | "forward", event: KeyboardEvent, @@ -4095,16 +4237,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) if (event.shiftKey || event.altKey || event.metaKey || event.ctrlKey || event.isComposing) { return false; } - if (isComposerApprovalState || pendingUserInputs.length > 0) return false; - if ( - composerImagesRef.current.length > 0 || - composerFilesRef.current.length > 0 || - composerTerminalContextsRef.current.length > 0 || - composerPreviewAnnotations.length > 0 || - composerReviewComments.length > 0 - ) { - return false; - } + if (promptHistoryBlocked) return false; const recalled = promptHistoryPositionRef.current; const position = recalled?.targetKey === promptHistoryTargetKey ? recalled.position : null; if (position === null && promptRef.current.length > 0) { @@ -4120,7 +4253,21 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) position, currentPrompt: promptRef.current, }); - if (!step) return false; + if (!step) { + // Past the oldest loaded prompt, or with none loaded, a long thread + // still has older ones: load one page per key press and step once it + // arrives. + if (direction !== "backward" || !onLoadOlderPromptHistory?.()) { + olderPromptRecallPendingRef.current = null; + return false; + } + olderPromptRecallPendingRef.current = { + targetKey: promptHistoryTargetKey, + oldestMessageId: promptHistoryMessagesRef.current[0]?.id, + }; + return true; + } + olderPromptRecallPendingRef.current = null; promptHistoryPositionRef.current = step.position ? { targetKey: promptHistoryTargetKey, position: step.position } : null; @@ -4128,6 +4275,46 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) return true; }; + useEffect(() => { + const pending = olderPromptRecallPendingRef.current; + if (pending === null) return; + const recalled = promptHistoryPositionRef.current; + const position = recalled?.targetKey === promptHistoryTargetKey ? recalled.position : null; + // An edit since the key press, a failed page, or context added while the + // page loaded ends browsing. + if ( + pending.targetKey !== promptHistoryTargetKey || + promptRef.current !== (position?.recalled ?? "") || + olderPromptHistoryFailed || + promptHistoryBlocked + ) { + olderPromptRecallPendingRef.current = null; + return; + } + if (promptHistoryMessages[0]?.id === pending.oldestMessageId) return; + olderPromptRecallPendingRef.current = null; + const step = stepComposerPromptHistory({ + direction: "backward", + entries: buildComposerPromptHistoryEntries(promptHistoryMessages), + position, + currentPrompt: promptRef.current, + }); + // A page of only agent work brings no prompt. The next ArrowUp loads + // another page instead of chaining through a long run unasked. + if (!step) return; + promptHistoryPositionRef.current = step.position + ? { targetKey: promptHistoryTargetKey, position: step.position } + : null; + replacePromptFromHistory(step.prompt); + }, [ + promptHistoryTargetKey, + promptHistoryBlocked, + olderPromptHistoryFailed, + promptHistoryMessages, + promptRef, + replacePromptFromHistory, + ]); + const onComposerCommandKey = (key: string, event: KeyboardEvent, isTaskItem = false) => { const submissionIntent = composerSubmissionIntentForKey({ event, @@ -5019,11 +5206,16 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const iconOnlyBlockCount = composerControlsCollapsed ? restingControlsIconOnlyBlockCount : expandedControlsIconOnlyBlockCount; - const restingProviderTraitsPicker = renderProviderTraitsPicker({ - ...providerTraitsPickerInput, - size: composerControlsCollapsed ? "xs" : "sm", - hidden: composerControlsHidden || restingHiddenBlockCount > 1, - }); + const restingProviderTraitsPicker = ( + 1, + }} + /> + ); const restingBlockDefs = [ ...(providerTraitsPicker ? [ @@ -5265,13 +5457,14 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) } if (tasksDrawerThreadId !== activeThreadId) setTasksDrawerThreadId(activeThreadId); - const [stashMenuClosesOn, setStashMenuClosesOn] = useState({ prompt, composerMenuOpen }); + const [stashMenuClosesOn, setStashMenuClosesOn] = useState({ promptVersion, composerMenuOpen }); if ( - stashMenuClosesOn.prompt !== prompt || + stashMenuClosesOn.promptVersion !== promptVersion || stashMenuClosesOn.composerMenuOpen !== composerMenuOpen ) { - setStashMenuClosesOn({ prompt, composerMenuOpen }); - if (stashMenuClosesOn.prompt !== prompt || composerMenuOpen) setIsStashMenuOpen(false); + setStashMenuClosesOn({ promptVersion, composerMenuOpen }); + if (stashMenuClosesOn.promptVersion !== promptVersion || composerMenuOpen) + setIsStashMenuOpen(false); } useEffect(() => { @@ -6547,15 +6740,23 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) disabled={isChoiceOnlyPendingQuestion} aria-label="Expand composer" > - {activePendingProgress - ? isChoiceOnlyPendingQuestion - ? "Choose an option above" - : activePendingProgress.customAnswer || - "Type your own answer, or leave this blank to use the selected option" - : prompt.trim() || - (showProviderUnavailable - ? "Enable a provider in Settings" - : "Ask anything...")} + {activePendingProgress ? ( + isChoiceOnlyPendingQuestion ? ( + "Choose an option above" + ) : ( + activePendingProgress.customAnswer || + "Type your own answer, or leave this blank to use the selected option" + ) + ) : ( + + )} {collapsedComposerImagePreviews}