[pull] main from pingdotgg:main

apps/server/src/cloud/pinnedRuntime.test.ts

@@ -1,3 +1,4 @@
+import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it } from "@effect/vitest";
 import * as Deferred from "effect/Deferred";
@@ -94,6 +95,19 @@ it.layer(NodeServices.layer)("ensurePinnedRuntimeInstalled", (it) => {
       assert.deepEqual(commands, ["tar"]);
       assert.equal(yield* fs.readFileString(paths.sentinelPath), `${version}\n`);
       assert.isFalse(yield* fs.exists(path.join(paths.versionDir, "t3-runtime-archive")));
+      if ((yield* HostProcessPlatform) !== "win32") {
+        // The old launcher must still be able to start this archive after the
+        // first npm-to-executable update, including from the final directory.
+        yield* fs.writeFileString(paths.entryPath, '#!/bin/sh\nprintf "%s\\n" "$@"\n');
+        yield* fs.chmod(paths.entryPath, 0o755);
+        const runner = yield* ProcessRunner.make();
+        const legacyStart = yield* runner.run({
+          command: process.execPath,
+          args: [path.join(paths.versionDir, "node_modules/t3/dist/bin.mjs"), "serve"],
+        });
+        assert.equal(Number(legacyStart.code), 0, legacyStart.stderr);
+        assert.equal(legacyStart.stdout.trim(), "serve");
+      }
     }),
   );
 
@@ -209,6 +223,40 @@ it.layer(NodeServices.layer)("ensurePinnedRuntimeInstalled", (it) => {
     }),
   );
 
+  it.effect("backfills a cached archive without downloading or replacing it", () =>
+    Effect.gen(function* () {
+      const fs = yield* FileSystem.FileSystem;
+      const path = yield* Path.Path;
+      const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pinned-legacy-cache-" });
+      const cached = pinnedRuntimePaths(path, baseDir, version, "linux");
+      const legacyEntry = path.join(cached.versionDir, "node_modules/t3/dist/bin.mjs");
+      yield* fs.makeDirectory(cached.versionDir, { recursive: true });
+      yield* fs.writeFileString(cached.entryPath, "cached executable\n");
+      yield* fs.writeFileString(cached.sentinelPath, `${version}\n`);
+      const requests: string[] = [];
+      const commands: string[] = [];
+      yield* ensurePinnedRuntimeInstalled({
+        baseDir,
+        version,
+        fs,
+        path,
+        platform: "linux",
+        arch: "x64",
+        httpClient: releaseHttpClient(yield* validChecksums, requests),
+        runner: extractingRunner(fs, path, commands),
+        validate: () =>
+          fs.exists(legacyEntry).pipe(
+            Effect.flatMap((exists) => (exists ? Effect.void : Effect.die("missing legacy entry"))),
+            Effect.orDie,
+          ),
+      });
+      assert.deepEqual(requests, []);
+      assert.deepEqual(commands, []);
+      assert.equal(yield* fs.readFileString(cached.entryPath), "cached executable\n");
+      assert.equal(yield* fs.readFileString(cached.sentinelPath), `${version}\n`);
+    }),
+  );
+
   it.effect("preserves a completed runtime when validation fails", () =>
     Effect.gen(function* () {
       const fs = yield* FileSystem.FileSystem;

apps/server/src/cloud/pinnedRuntime.ts

@@ -8,6 +8,7 @@ import * as Option from "effect/Option";
 import * as Semaphore from "effect/Semaphore";
 import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http";
 
+import { legacyCliLauncherScript } from "@t3tools/shared/legacyCliLauncher";
 import {
   CLI_RELEASE_CHECKSUMS_FILE,
   cliArchiveFileName,
@@ -229,6 +230,24 @@ const installPinnedRuntime = Effect.fn("cloud.pinned_runtime.ensure_installed")(
   input: PinnedRuntimeInstallInput,
 ) {
   const { fs } = input;
+  // Old service launchers still use the npm entry point, including when an
+  // archive was cached before this compatibility wrapper existed.
+  const ensureLegacyEntry = Effect.fn("cloud.pinned_runtime.ensure_legacy_entry")(
+    function* (versionDir: string) {
+      const legacyDir = input.path.join(versionDir, "node_modules", "t3", "dist");
+      const entryPath = input.path.join(legacyDir, "bin.mjs");
+      if (yield* fs.exists(entryPath)) return;
+      yield* fs.makeDirectory(legacyDir, { recursive: true });
+      yield* fs.writeFileString(entryPath, legacyCliLauncherScript("archive"));
+    },
+    Effect.mapError(
+      (cause) =>
+        new PinnedRuntimeInstallError({
+          step: "writing the legacy service entry point",
+          cause,
+        }),
+    ),
+  );
   const paths = pinnedRuntimePaths(input.path, input.baseDir, input.version, input.platform);
   const [versionDirExists, entryExists, sentinel] = yield* Effect.all([
     fs.exists(paths.versionDir),
@@ -242,6 +261,7 @@ const installPinnedRuntime = Effect.fn("cloud.pinned_runtime.ensure_installed")(
   const alreadyPinned =
     entryExists && Option.isSome(sentinel) && sentinel.value.trim() === input.version;
   if (alreadyPinned) {
+    yield* ensureLegacyEntry(paths.versionDir);
     yield* input.validate(paths);
     return paths;
   }
@@ -290,6 +310,8 @@ const installPinnedRuntime = Effect.fn("cloud.pinned_runtime.ensure_installed")(
   return yield* Effect.gen(function* () {
     yield* installFromArchive(input, stagingDir);
 
+    yield* ensureLegacyEntry(stagingDir);
+
     yield* input.validate(stagingPaths);
     yield* fs
       .writeFileString(stagingPaths.sentinelPath, `${input.version}\n`)
@@ -328,7 +350,10 @@ const installPinnedRuntime = Effect.fn("cloud.pinned_runtime.ensure_installed")(
         ),
       ),
     );
-    if (!published) yield* input.validate(paths);
+    if (!published) {
+      yield* ensureLegacyEntry(paths.versionDir);
+      yield* input.validate(paths);
+    }
     return paths;
   }).pipe(
     Effect.ensuring(fs.remove(stagingDir, { recursive: true, force: true }).pipe(Effect.ignore)),

apps/server/src/terminal/OutputProtocol.test.ts

@@ -0,0 +1,79 @@
+import { assert, describe, it } from "@effect/vitest";
+import { WS_METHODS } from "@t3tools/contracts";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Queue from "effect/Queue";
+import * as Schema from "effect/Schema";
+import * as Stream from "effect/Stream";
+import * as TestClock from "effect/testing/TestClock";
+import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/unstable/rpc";
+
+import { withTerminalOutputWindow } from "./OutputProtocol.ts";
+
+describe("terminal output window", () => {
+  for (const { tag, size, limit } of [
+    { tag: WS_METHODS.terminalAttach, size: 1, limit: 8 },
+    { tag: WS_METHODS.subscribeTerminalEvents, size: 1, limit: 8 },
+    { tag: WS_METHODS.terminalAttach, size: 64 * 1024, limit: 1 },
+    { tag: WS_METHODS.subscribeTerminalMetadata, size: 1, limit: 1 },
+  ]) {
+    it.effect(`limits ${tag} with ${size}-byte values to ${limit} pending chunks`, () =>
+      Effect.gen(function* () {
+        const group = RpcGroup.make(Rpc.make(tag, { success: Schema.String, stream: true }));
+        const output = yield* Queue.unbounded<string>();
+        const responses = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
+        const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
+        const protocol = yield* RpcServer.Protocol.make((write) =>
+          Effect.gen(function* () {
+            yield* Deferred.succeed(receive, write);
+            const serialization = yield* RpcSerialization.RpcSerialization;
+            return {
+              disconnects: yield* Queue.unbounded<number>(),
+              send: (_clientId, response) => Queue.offer(responses, response),
+              end: () => Effect.void,
+              clientIds: Effect.succeed(new Set([0])),
+              initialMessage: Effect.succeedNone,
+              supportsAck: true,
+              supportsTransferables: false,
+              supportsSpanPropagation: false,
+              supportsNotifications: true,
+              codecFor: serialization.codecFor,
+            };
+          }),
+        );
+        yield* RpcServer.make(group).pipe(
+          Effect.provide(group.toLayerHandler(tag, () => Stream.fromQueue(output))),
+          Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
+          Effect.forkScoped,
+        );
+        const write = yield* Deferred.await(receive);
+        yield* write(0, { _tag: "Request", id: "1", tag, payload: null, headers: [] });
+
+        for (let index = 0; index < limit; index++) {
+          const value = String(index).repeat(size);
+          yield* Queue.offer(output, value);
+          yield* TestClock.adjust(0);
+          assert.equal(yield* Queue.size(responses), 1);
+          assert.deepEqual(yield* Queue.take(responses), {
+            _tag: "Chunk",
+            requestId: "1",
+            values: [value],
+          });
+        }
+        yield* Queue.offer(output, "i".repeat(size));
+        yield* TestClock.adjust(0);
+        assert.equal(yield* Queue.size(responses), 0);
+        yield* write(0, { _tag: "Ack", requestId: "1" });
+        const resumed = yield* Queue.take(responses);
+        assert.equal(resumed._tag, "Chunk");
+        if (resumed._tag === "Chunk") assert.deepEqual(resumed.values, ["i".repeat(size)]);
+
+        yield* Queue.offer(output, "j");
+        yield* TestClock.adjust(0);
+        assert.equal(yield* Queue.size(responses), 0);
+        yield* write(0, { _tag: "Interrupt", requestId: "1" });
+        assert.equal((yield* Queue.take(responses))._tag, "Exit");
+      }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
+    );
+  }
+});

apps/server/src/terminal/OutputProtocol.ts

@@ -0,0 +1,67 @@
+import { WS_METHODS } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import type { RpcServer } from "effect/unstable/rpc";
+
+const MAX_PENDING_CHUNKS = 8;
+const MAX_PENDING_BYTES = 64 * 1024;
+const isFull = (sizes: number[]) =>
+  sizes.length >= MAX_PENDING_CHUNKS ||
+  sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES;
+
+export function withTerminalOutputWindow(
+  protocol: RpcServer.Protocol["Service"],
+): RpcServer.Protocol["Service"] {
+  const windows = new Map<string, number[]>();
+  let receive: Parameters<RpcServer.Protocol["Service"]["run"]>[0];
+  return {
+    ...protocol,
+    run: (write) => {
+      receive = write;
+      return protocol.run((clientId, message) =>
+        Effect.suspend(() => {
+          if (
+            message._tag === "Request" &&
+            (message.tag === WS_METHODS.terminalAttach ||
+              message.tag === WS_METHODS.subscribeTerminalEvents)
+          ) {
+            const key = `${clientId}:${message.id}`;
+            if (!windows.has(key)) windows.set(key, []);
+          } else if (message._tag === "Ack") {
+            const window = windows.get(`${clientId}:${message.requestId}`);
+            if (window) {
+              const wasFull = isFull(window);
+              window.shift();
+              if (!wasFull || isFull(window)) return Effect.void;
+            }
+          }
+          return write(clientId, message);
+        }),
+      );
+    },
+    send: (clientId, response, transferables) =>
+      Effect.suspend(() => {
+        const send = protocol.send(clientId, response, transferables);
+        if (response._tag === "Exit") {
+          windows.delete(`${clientId}:${response.requestId}`);
+        } else if (response._tag === "Chunk") {
+          const window = windows.get(`${clientId}:${response.requestId}`);
+          if (window) {
+            // @effect-diagnostics-next-line preferSchemaOverJson:off
+            const size = Buffer.byteLength(JSON.stringify(response));
+            window.push(size);
+            if (!isFull(window)) {
+              return send.pipe(
+                Effect.andThen(() =>
+                  receive(clientId, {
+                    _tag: "Ack",
+                    requestId: response.requestId,
+                  }),
+                ),
+              );
+            }
+          }
+        }
+        return send;
+      }),
+  };
+}

apps/server/src/ws.ts

@@ -115,6 +115,7 @@ import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
 import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
 import * as ServerSettings from "./serverSettings.ts";
 import * as TerminalManager from "./terminal/Manager.ts";
+import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts";
 import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts";
 import * as DeviceService from "./device/DeviceService.ts";
 import { remoteSshDeviceHosts } from "./device/localSshDeviceHost.ts";
@@ -3446,8 +3447,14 @@ export const websocketRpcRouteLayer = Layer.unwrap(
         const clientAnalyticsProps = readClientAnalyticsProps(request);
         yield* sessions.recordClientConnection(session.sessionId, clientOrigin);
         yield* analytics.record("client.connected", clientAnalyticsProps);
-        const rpcWebSocketHttpEffect = yield* RpcServer.toHttpEffectWebsocket(WsRpcGroup, {
-          disableTracing: true,
+        const rpcWebSocketHttpEffect = yield* Effect.gen(function* () {
+          const { protocol, httpEffect } = yield* RpcServer.makeProtocolWithHttpEffectWebsocket;
+          yield* RpcServer.make(WsRpcGroup, { disableTracing: true }).pipe(
+            Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
+            Effect.forkScoped,
+          );
+          // @effect-diagnostics-next-line returnEffectInGen:off
+          return httpEffect;
         }).pipe(
           Effect.provide(
             makeWsRpcLayer(

apps/web/src/components/ChatView.tsx

@@ -9364,6 +9364,7 @@ export default function ChatView(props: ChatViewProps) {
       {rightPanelPresent && !shouldUseRightPanelSheet && activeThreadRef ? (
         <RightPanelTabs
           mode="inline"
+          widthStorageKey={`t3code:preview-panel-width:${activeThreadKey}`}
           open={rightPanelOpen}
           maximized={rightPanelMaximized}
           surfaces={renderedRightPanelSurfaces}

apps/web/src/components/chat/ChatComposer.tsx

@@ -242,6 +242,7 @@ import { ProviderModelPicker } from "./ProviderModelPicker";
 import { type ComposerCommandItem, ComposerCommandMenu } from "./ComposerCommandMenu";
 import { ComposerPendingApprovalActions } from "./ComposerPendingApprovalActions";
 import { CompactComposerControlsMenu } from "./CompactComposerControlsMenu";
+import { ComposerImageThumbnail } from "./ComposerImageThumbnail";
 import { ComposerPrimaryActions } from "./ComposerPrimaryActions";
 import { ComposerPendingApprovalPanel } from "./ComposerPendingApprovalPanel";
 import { ComposerPendingUserInputPanel } from "./ComposerPendingUserInputPanel";
@@ -4673,7 +4674,19 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
             }}
           >
             {image.previewUrl ? (
-              <img src={image.previewUrl} alt="" className="size-full object-cover" />
+              <ComposerImageThumbnail
+                file={image.file}
+                alt=""
+                className="size-full object-cover"
+                fallback={
+                  <PierreEntryIcon
+                    pathValue={image.name}
+                    kind="file"
+                    theme={resolvedTheme}
+                    className="m-auto size-3.5"
+                  />
+                }
+              />
             ) : (
               <PierreEntryIcon
                 pathValue={image.name}
@@ -6287,10 +6300,15 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
                                   onExpandImage(preview);
                                 }}
                               >
-                                <img
-                                  src={image.previewUrl}
+                                <ComposerImageThumbnail
+                                  file={image.file}
                                   alt={image.name}
                                   className="h-full w-full object-cover"
+                                  fallback={
+                                    <span className="flex h-full items-center justify-center px-1 text-[10px] text-secondary-label">
+                                      {image.name}
+                                    </span>
+                                  }
                                 />
                               </button>
                             ) : (

apps/web/src/components/chat/ComposerImageThumbnail.tsx

@@ -0,0 +1,29 @@
+import { memo, useEffect, useState, type ReactNode } from "react";
+
+import { createComposerImageThumbnail } from "../../lib/imageCompression";
+
+/** Keep full-resolution image decoding out of composer rerenders. */
+export const ComposerImageThumbnail = memo(function ComposerImageThumbnail({
+  file,
+  alt,
+  className,
+  fallback,
+}: {
+  file: File;
+  alt: string;
+  className: string;
+  fallback: ReactNode;
+}) {
+  const [preview, setPreview] = useState<{ file: File; src: string | null } | null>(null);
+  useEffect(() => {
+    let active = true;
+    void createComposerImageThumbnail(file).then((src) => {
+      if (active) setPreview({ file, src });
+    });
+    return () => {
+      active = false;
+    };
+  }, [file]);
+  const src = preview?.file === file ? preview.src : null;
+  return src ? <img src={src} alt={alt} className={className} /> : fallback;
+});