fix(client-runtime): subagent threads load over T3 Connect without two 401s

packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts

@@ -7,8 +7,8 @@ import { HttpClient } from "effect/http";
 
 import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
 import type { PreparedConnection } from "../connection/model.ts";
-import { environmentEndpointUrl } from "../environment/endpoint.ts";
 import * as ManagedRelay from "../relay/managedRelay.ts";
+import { makeEnvironmentHttpApiUrlBuilder } from "../rpc/http.ts";
 import {
   executeAuthenticatedEnvironmentHttpRequest,
   withOrchestrationProtocolHeader,
@@ -35,7 +35,9 @@ export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn(
     group: "orchestration",
     method: "GET",
     url: (httpBaseUrl) =>
-      environmentEndpointUrl(httpBaseUrl, `/api/orchestration/threads/${input.threadId}/bounded`),
+      makeEnvironmentHttpApiUrlBuilder(httpBaseUrl).orchestration.threadBoundedSnapshot({
+        params: { threadId: input.threadId },
+      }),
     timeoutMs: input.timeoutMs ?? DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS,
     request: ({ client, headers }) =>
       client.threadBoundedSnapshot({

packages/client-runtime/src/state/environmentHttpAuth.test.ts

@@ -4,15 +4,19 @@ import {
   ORCHESTRATION_PROTOCOL_HEADER,
   ORCHESTRATION_PROTOCOL_VERSION_TEXT,
   ProjectId,
+  ThreadId,
   type AuthSessionState,
   type OrchestrationV2ShellSnapshot,
   OrchestrationV2ThreadDetailSnapshot,
   OrchestrationV2ThreadBoundedSnapshot,
   type OrchestrationV2ThreadHistoryPage,
 } from "@t3tools/contracts";
+import { verifyDpopProof } from "@t3tools/shared/dpop";
 import { RelayClientTracer } from "@t3tools/shared/relayTracing";
+import * as Clock from "effect/Clock";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Base64Url from "effect/encoding/Base64Url";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
@@ -278,7 +282,7 @@ describe("authenticated environment HTTP requests", () => {
       expect(harness.proofs).toEqual([
         {
           method: loader.method,
-          url: `${CURRENT_ORIGIN}${loader.path}`,
+          url: call.url,
           accessToken: "current-token",
         },
       ]);
@@ -345,7 +349,7 @@ describe("authenticated environment HTTP requests", () => {
         );
         expect(harness.proofs[1]).toEqual({
           method: "GET",
-          url: `${RENEWED_ORIGIN}${loader.path}`,
+          url: retried.url,
           accessToken: "renewed-token",
         });
         if (loader.name === "older thread history") {
@@ -655,6 +659,111 @@ describe("authenticated environment HTTP requests", () => {
   );
 });
 
+// Delegated threads join percent-encoded parts with ":" (see IdAllocator.joinId).
+const DELEGATED_THREAD_ID = ThreadId.make("thread:delegated-task:command%3Amcp%3Adelegate-1");
+
+/** Signs real ES256 proofs the way the web and mobile signers do. */
+const makeEs256Signer = Effect.gen(function* () {
+  const keyPair = yield* Effect.promise(() =>
+    crypto.subtle.generateKey({ name: "ECDSA", namedCurve: "P-256" }, false, ["sign", "verify"]),
+  );
+  const { kty, crv, x, y } = yield* Effect.promise(() =>
+    crypto.subtle.exportKey("jwk", keyPair.publicKey),
+  );
+  const encodeJson = (value: unknown) => Base64Url.encode(JSON.stringify(value));
+  let proofCount = 0;
+  return ManagedRelay.ManagedRelayDpopSigner.of({
+    thumbprint: Effect.succeed("test-thumbprint"),
+    createProof: (input) =>
+      Effect.gen(function* () {
+        const htu = new URL(input.url);
+        htu.search = "";
+        htu.hash = "";
+        const header = encodeJson({ typ: "dpop+jwt", alg: "ES256", jwk: { kty, crv, x, y } });
+        const payload = encodeJson({
+          htm: input.method,
+          htu: htu.toString(),
+          jti: `proof-${++proofCount}`,
+          iat: Math.floor((yield* Clock.currentTimeMillis) / 1_000),
+        });
+        const signature = yield* Effect.promise(() =>
+          crypto.subtle.sign(
+            { name: "ECDSA", hash: "SHA-256" },
+            keyPair.privateKey,
+            new TextEncoder().encode(`${header}.${payload}`),
+          ),
+        );
+        return `${header}.${payload}.${Base64Url.encode(new Uint8Array(signature))}`;
+      }),
+  });
+});
+
+const DELEGATED_THREAD_LOADERS: ReadonlyArray<{
+  readonly name: string;
+  readonly response: unknown;
+  readonly load: (
+    input: HttpInput,
+  ) => Effect.Effect<unknown, RemoteEnvironmentRequestError, HttpClient.HttpClient>;
+}> = [
+  {
+    name: "thread snapshot",
+    response: encodeThreadSnapshot(THREAD),
+    load: (input) =>
+      ThreadSnapshotLoader.fetchEnvironmentThreadSnapshot({
+        ...input,
+        threadId: DELEGATED_THREAD_ID,
+      }),
+  },
+  {
+    name: "bounded thread snapshot",
+    response: encodeBoundedSnapshot(BOUNDED_THREAD),
+    load: (input) =>
+      fetchEnvironmentBoundedThreadSnapshot({ ...input, threadId: DELEGATED_THREAD_ID }),
+  },
+  {
+    name: "older thread history",
+    response: THREAD_HISTORY,
+    load: (input) =>
+      fetchEnvironmentThreadHistoryPage({
+        ...input,
+        threadId: DELEGATED_THREAD_ID,
+        cursor: "older-page",
+      }),
+  },
+];
+
+describe("DPoP proofs for delegated thread IDs", () => {
+  it.effect.each(DELEGATED_THREAD_LOADERS)(
+    "binds the $name proof to the URL the server receives",
+    (loader) =>
+      Effect.gen(function* () {
+        const harness = makeHarness(() => Response.json(loader.response));
+        const signer = yield* makeEs256Signer;
+        yield* loader
+          .load({ ...harness.input, signer: Option.some(signer) })
+          .pipe(Effect.provide(harness.httpLayer));
+
+        expect(harness.calls).toHaveLength(1);
+        const call = harness.calls[0]!;
+        expect(new URL(call.url).pathname).toContain(
+          "/threads/thread%3Adelegated-task%3Acommand%253Amcp%253Adelegate-1",
+        );
+        const result = verifyDpopProof({
+          proof: new Headers(call.init.headers).get("dpop"),
+          method: call.init.method ?? "GET",
+          url: call.url,
+          nowEpochSeconds: Math.floor((yield* Clock.currentTimeMillis) / 1_000),
+        });
+        expect(result).toEqual({
+          ok: true,
+          thumbprint: expect.any(String),
+          jti: expect.any(String),
+          iat: expect.any(Number),
+        });
+      }),
+  );
+});
+
 describe("relay request tracing", () => {
   it.effect("starts an exported trace for a T3 Connect request", () =>
     Effect.gen(function* () {

packages/client-runtime/src/state/threadHistoryHttp.ts

@@ -4,8 +4,8 @@ import * as Option from "effect/Option";
 
 import type { RemoteEnvironmentAuthorization } from "../authorization/service.ts";
 import type { PreparedConnection } from "../connection/model.ts";
-import { environmentEndpointUrl } from "../environment/endpoint.ts";
 import * as ManagedRelay from "../relay/managedRelay.ts";
+import { makeEnvironmentHttpApiUrlBuilder } from "../rpc/http.ts";
 import {
   executeAuthenticatedEnvironmentHttpRequest,
   withOrchestrationProtocolHeader,
@@ -28,7 +28,10 @@ export const fetchEnvironmentThreadHistoryPage = Effect.fn(
     group: "orchestration",
     method: "GET",
     url: (httpBaseUrl) =>
-      environmentEndpointUrl(httpBaseUrl, `/api/orchestration/threads/${input.threadId}/history`),
+      makeEnvironmentHttpApiUrlBuilder(httpBaseUrl).orchestration.threadHistoryPage({
+        params: { threadId: input.threadId },
+        query: { cursor: input.cursor },
+      }),
     timeoutMs: input.timeoutMs ?? DEFAULT_THREAD_HISTORY_TIMEOUT_MS,
     request: ({ client, headers }) =>
       client.threadHistoryPage({

packages/client-runtime/src/state/threadSnapshotHttp.ts

@@ -5,9 +5,11 @@ import * as Option from "effect/Option";
 
 import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
 import type { PreparedConnection } from "../connection/model.ts";
-import { environmentEndpointUrl } from "../environment/endpoint.ts";
 import * as ManagedRelay from "../relay/managedRelay.ts";
-import type { RemoteEnvironmentRequestError } from "../rpc/http.ts";
+import {
+  makeEnvironmentHttpApiUrlBuilder,
+  type RemoteEnvironmentRequestError,
+} from "../rpc/http.ts";
 import {
   executeAuthenticatedEnvironmentHttpRequest,
   withOrchestrationProtocolHeader,
@@ -66,7 +68,9 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn(
     group: "orchestration",
     method: "GET",
     url: (httpBaseUrl) =>
-      environmentEndpointUrl(httpBaseUrl, `/api/orchestration/threads/${input.threadId}`),
+      makeEnvironmentHttpApiUrlBuilder(httpBaseUrl).orchestration.threadSnapshot({
+        params: { threadId: input.threadId },
+      }),
     timeoutMs: input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS,
     request: ({ client, headers }) =>
       client.threadSnapshot({