Merge upstream/main into main

Author: donnes

.github/workflows/release.yml

@@ -331,6 +331,9 @@ jobs:
       # No Electron setup here: `t3` (apps/server) has no Electron dependency
       # and none of its tests touch the runtime. Only the non-server `test`
       # job, which covers @t3tools/desktop, needs the download.
+      - name: Install Chromium for server browser integration tests
+        run: node apps/server/node_modules/playwright-core/cli.js install --with-deps --only-shell chromium
+
       - name: Test
         run: vp run --filter t3 test --shard ${{ matrix.shard }}/${{ strategy.job-total }}
 

apps/desktop/src/electron/ElectronProtocol.test.ts

@@ -1,9 +1,11 @@
 import { assert, describe, it } from "@effect/vitest";
 import * as Cause from "effect/Cause";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
+import * as TestClock from "effect/testing/TestClock";
 import { beforeEach, vi } from "vite-plus/test";
 
 const { handleMock, netFetchMock, unhandleMock } = vi.hoisted(() => ({
@@ -174,7 +176,11 @@ describe("ElectronProtocol", () => {
             targetOrigin: new URL("http://127.0.0.1:5733/"),
             clerkFrontendApiHostname: undefined,
           });
-          return yield* Effect.promise(() => handler!(new Request("t3code-dev://app/")));
+          const fiber = yield* Effect.forkChild(
+            Effect.promise(() => handler!(new Request("t3code-dev://app/"))),
+          );
+          yield* TestClock.adjust("50 millis");
+          return yield* Fiber.join(fiber);
         }),
       );
 
@@ -183,6 +189,50 @@ describe("ElectronProtocol", () => {
     }).pipe(Effect.provide(layerProtocol)),
   );
 
+  it.effect("rejects with the last renderer target failure after 50ms and 150ms retries", () =>
+    Effect.gen(function* () {
+      let handler: ((request: Request) => Promise<Response>) | undefined;
+      handleMock.mockImplementation((_scheme, nextHandler) => {
+        handler = nextHandler;
+      });
+      const lastFailure = new Error("connect ECONNREFUSED 127.0.0.1:5733 (3)");
+      netFetchMock
+        .mockRejectedValueOnce(new Error("connect ECONNREFUSED 127.0.0.1:5733 (1)"))
+        .mockRejectedValueOnce(new Error("connect ECONNREFUSED 127.0.0.1:5733 (2)"))
+        .mockRejectedValueOnce(lastFailure);
+
+      const rejection = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const protocol = yield* ElectronProtocol.ElectronProtocol;
+          yield* protocol.registerDesktopProtocol({
+            scheme: "t3code-dev",
+            targetOrigin: new URL("http://127.0.0.1:5733/"),
+            clerkFrontendApiHostname: undefined,
+          });
+          const fiber = yield* Effect.forkChild(
+            Effect.promise(() =>
+              handler!(new Request("t3code-dev://app/")).then(
+                () => null,
+                (error: unknown) => error,
+              ),
+            ),
+          );
+          yield* TestClock.adjust("49 millis");
+          assert.equal(netFetchMock.mock.calls.length, 1);
+          yield* TestClock.adjust("1 millis");
+          assert.equal(netFetchMock.mock.calls.length, 2);
+          yield* TestClock.adjust("149 millis");
+          assert.equal(netFetchMock.mock.calls.length, 2);
+          yield* TestClock.adjust("1 millis");
+          return yield* Fiber.join(fiber);
+        }),
+      );
+
+      assert.strictEqual(rejection, lastFailure);
+      assert.equal(netFetchMock.mock.calls.length, 3);
+    }).pipe(Effect.provide(layerProtocol)),
+  );
+
   it.effect("preserves protocol registration failures", () =>
     Effect.gen(function* () {
       const cause = new Error("protocol registration failed");

apps/desktop/src/electron/ElectronProtocol.ts

@@ -3,10 +3,10 @@ import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
-import * as NodeTimersPromises from "node:timers/promises";
 import * as Path from "effect/Path";
 import * as Mime from "effect/http/Mime";
 import * as Ref from "effect/Ref";
+import * as Schedule from "effect/Schedule";
 import * as Schema from "effect/Schema";
 import * as Scope from "effect/Scope";
 
@@ -150,11 +150,29 @@ const registerDesktopSchemePrivileges = Effect.sync(registerDesktopSchemePrivile
 
 export const layerSchemePrivileges = Layer.effectDiscard(registerDesktopSchemePrivileges);
 
-async function proxyRequest(
+class ElectronProtocolFetchError extends Schema.TaggedError<ElectronProtocolFetchError>()(
+  "ElectronProtocolFetchError",
+  { cause: Schema.Defect() },
+) {}
+
+const netFetch = (url: string, init: RequestInit) =>
+  Effect.tryPromise({
+    try: () => Electron.net.fetch(url, init),
+    catch: (cause) => new ElectronProtocolFetchError({ cause }),
+  });
+
+// The dev renderer target can briefly refuse connections while Vite restarts:
+// retry idempotent requests after 50ms, then 150ms, and keep the last failure.
+const fetchWithTransientRetry = (url: string, init: RequestInit) =>
+  netFetch(url, init).pipe(
+    Effect.retry({ schedule: Schedule.exponential("50 millis", 3), times: 2 }),
+  );
+
+const proxyRequest = Effect.fn("desktop.protocol.proxyRequest")(function* (
   request: Request,
   targetOrigin: URL,
   contentSecurityPolicy: string,
-): Promise<Response> {
+) {
   const requestUrl = new URL(request.url);
   if (requestUrl.host !== DESKTOP_HOST) {
     return new Response(null, { status: 404 });
@@ -190,12 +208,10 @@ async function proxyRequest(
   }
   const response =
     request.method === "GET" || request.method === "HEAD"
-      ? await fetchWithTransientRetry(targetUrl.toString(), init)
-      : await Electron.net.fetch(targetUrl.toString(), init);
+      ? yield* fetchWithTransientRetry(targetUrl.toString(), init)
+      : yield* netFetch(targetUrl.toString(), init);
   return withContentSecurityPolicy(response, contentSecurityPolicy);
-}
-
-const TRANSIENT_FETCH_RETRY_DELAYS_MS = [0, 50, 150] as const;
+});
 
 // Serves the packaged web client without a backend: files resolve within the
 // asset directory, and any other path falls back to index.html so the SPA
@@ -238,24 +254,6 @@ const serveDesktopAsset = Effect.fn("desktop.protocol.serveAsset")(function* (
   });
 });
 
-async function fetchWithTransientRetry(url: string, init: RequestInit): Promise<Response> {
-  let lastError: unknown;
-
-  for (const delayMs of TRANSIENT_FETCH_RETRY_DELAYS_MS) {
-    if (delayMs > 0) {
-      await NodeTimersPromises.setTimeout(delayMs);
-    }
-
-    try {
-      return await Electron.net.fetch(url, init);
-    } catch (error) {
-      lastError = error;
-    }
-  }
-
-  throw lastError;
-}
-
 /** @public Service construction is part of the canonical Effect module API. */
 export const make = Effect.gen(function* () {
   const registered = yield* Ref.make(false);
@@ -278,7 +276,14 @@ export const make = Effect.gen(function* () {
                   contentSecurityPolicy,
                 );
               }
-              return proxyRequest(request, input.targetOrigin, contentSecurityPolicy);
+              // Reject with net.fetch's own error, as an unproxied fetch would.
+              return runPromise(
+                proxyRequest(request, input.targetOrigin, contentSecurityPolicy).pipe(
+                  Effect.catchTags({
+                    ElectronProtocolFetchError: (error) => Effect.die(error.cause),
+                  }),
+                ),
+              );
             });
           },
           catch: (cause) => new ElectronProtocolRegistrationError({ scheme: input.scheme, cause }),

apps/desktop/src/ssh/DesktopSshEnvironment.ts

@@ -18,6 +18,7 @@ import {
 } from "@t3tools/ssh/errors";
 import * as SshTunnel from "@t3tools/ssh/tunnel";
 import * as Context from "effect/Context";
+import type * as Crypto from "effect/Crypto";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
@@ -29,6 +30,7 @@ import * as DesktopSshPasswordPrompts from "./DesktopSshPasswordPrompts.ts";
 
 export type DesktopSshEnvironmentRuntimeServices =
   | ChildProcessSpawner.ChildProcessSpawner
+  | Crypto.Crypto
   | FileSystem.FileSystem
   | Path.Path
   | HttpClient.HttpClient

apps/server/package.json

@@ -34,6 +34,7 @@
     "@effect/platform-node-shared": "catalog:",
     "@ff-labs/fff-node": "0.9.4",
     "@napi-rs/keyring": "^1.3.0",
+    "@noble/hashes": "catalog:",
     "@opencode-ai/sdk": "^1.3.15",
     "@opencode/client": "2.0.23",
     "@opencode/protocol": "2.0.23",

apps/server/src/auth/ReusableDevAuth.ts

@@ -1,3 +1,4 @@
+// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no timingSafeEqual.
 import * as NodeCrypto from "node:crypto";
 import { AuthSessionId } from "@t3tools/contracts";
 import * as DateTime from "effect/DateTime";

apps/server/src/auth/utils.ts

@@ -4,6 +4,7 @@ import type {
   AuthClientPresentationMetadata,
 } from "@t3tools/contracts";
 import type * as HttpServerRequest from "effect/http/HttpServerRequest";
+// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no createHmac or timingSafeEqual.
 import * as NodeCrypto from "node:crypto";
 import * as Base64Url from "effect/encoding/Base64Url";
 import * as Result from "effect/Result";

apps/server/src/automation/AutomationService.ts

@@ -10,8 +10,6 @@
  *
  * @module AutomationService
  */
-import * as NodeCrypto from "node:crypto";
-
 import {
   Automation,
   AUTOMATION_MAX_RUNS,
@@ -655,21 +653,22 @@ const make = Effect.gen(function* () {
         })
         .pipe(
           Effect.as(true),
-          Effect.catchTag("OrchestratorDispatchError", (error) =>
-            threads.getThreadEventSequence(threadId).pipe(
-              // Only a thread that moved past the read is decided again against a fresh one;
-              // any other rejection waits for the next event instead of retrying in a loop.
-              Effect.flatMap((sequence) =>
-                sequence > expectedSequence
-                  ? recheck(threadId)
-                  : Effect.logWarning("automation thread delete rejected", {
-                      threadId,
-                      cause: error,
-                    }),
+          Effect.catchTags({
+            OrchestratorDispatchError: (error) =>
+              threads.getThreadEventSequence(threadId).pipe(
+                // Only a thread that moved past the read is decided again against a fresh one;
+                // any other rejection waits for the next event instead of retrying in a loop.
+                Effect.flatMap((sequence) =>
+                  sequence > expectedSequence
+                    ? recheck(threadId)
+                    : Effect.logWarning("automation thread delete rejected", {
+                        threadId,
+                        cause: error,
+                      }),
+                ),
+                Effect.as(false),
               ),
-              Effect.as(false),
-            ),
-          ),
+          }),
         );
       if (!deleted) return;
       yield* recordPendingRuns(threadId, [], "deleted");
@@ -723,7 +722,7 @@ const make = Effect.gen(function* () {
         path: path.join(
           config.worktreesDir,
           path.basename(workspaceRoot),
-          `pr-${pullRequestNumber}-${NodeCrypto.randomBytes(4).toString("hex")}`,
+          `pr-${pullRequestNumber}-${Buffer.from(yield* crypto.randomBytes(4).pipe(Effect.orDie)).toString("hex")}`,
         ),
       });
       const worktreePath = created.worktree.path;

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts

@@ -1,3 +1,4 @@
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
 import { assert, it, vi } from "@effect/vitest";
 import { CheckpointRef, CheckpointScopeId, RunId, ThreadId } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
@@ -59,6 +60,7 @@ function layerFor(input: {
         }),
       ),
     ),
+    Layer.provideMerge(NodeCrypto.layer),
   );
 }
 
@@ -80,7 +82,7 @@ it.effect("computes V2 run diffs from projected checkpoint scopes", () => {
     });
     assert.deepEqual(diffCheckpoints.mock.calls[0]?.[0], {
       cwd: "/repo",
-      fromCheckpointRef: checkpointRefForScopeOrdinal({
+      fromCheckpointRef: yield* checkpointRefForScopeOrdinal({
         scopeId: firstScopeId,
         ordinalWithinScope: 0,
       }),

apps/server/src/checkpointing/CheckpointDiffQuery.ts

@@ -15,6 +15,7 @@ import {
   type ThreadId,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
+import * as Crypto from "effect/Crypto";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
@@ -77,6 +78,7 @@ function buildTurnDiffResult(
 export const make = Effect.gen(function* () {
   const threads = yield* ThreadManagement.ThreadManagementService;
   const checkpointStore = yield* CheckpointStore.CheckpointStore;
+  const crypto = yield* Crypto.Crypto;
 
   const getTurnDiff: CheckpointDiffQuery["Service"]["getTurnDiff"] = Effect.fn("getTurnDiff")(
     function* (input) {
@@ -155,21 +157,20 @@ export const make = Effect.gen(function* () {
         });
       }
 
+      // The root scope is shared by every run in this thread. Its runId
+      // tracks the latest owner, while ordinal zero stays the baseline.
+      const firstScope =
+        input.fromTurnCount === 0
+          ? projection.checkpointScopes.find((scope) => scope.kind === "root_run")
+          : undefined;
       const fromCheckpointRef =
         input.fromTurnCount === 0
-          ? (() => {
-              // The root scope is shared by every run in this thread. Its
-              // runId tracks the latest owner, while ordinal zero stays the baseline.
-              const firstScope = projection.checkpointScopes.find(
-                (scope) => scope.kind === "root_run",
-              );
-              return firstScope === undefined
-                ? undefined
-                : checkpointRefForScopeOrdinal({
-                    scopeId: firstScope.id,
-                    ordinalWithinScope: 0,
-                  });
-            })()
+          ? firstScope === undefined
+            ? undefined
+            : yield* checkpointRefForScopeOrdinal({
+                scopeId: firstScope.id,
+                ordinalWithinScope: 0,
+              }).pipe(Effect.provideService(Crypto.Crypto, crypto))
           : readyCheckpoints.find((checkpoint) => checkpoint.appRunOrdinal === input.fromTurnCount)
               ?.ref;
       if (fromCheckpointRef === undefined) {