diff --git a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts b/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts index ffad41874c94..da449d6414bf 100644 --- a/packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts +++ b/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({ diff --git a/packages/client-runtime/src/state/environmentHttpAuth.test.ts b/packages/client-runtime/src/state/environmentHttpAuth.test.ts index c7c7fc53197a..19bea516faff 100644 --- a/packages/client-runtime/src/state/environmentHttpAuth.test.ts +++ b/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; +}> = [ + { + 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* () { diff --git a/packages/client-runtime/src/state/threadHistoryHttp.ts b/packages/client-runtime/src/state/threadHistoryHttp.ts index a2416bda26f9..1d26a6c845d9 100644 --- a/packages/client-runtime/src/state/threadHistoryHttp.ts +++ b/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({ diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index 8b31e4c60687..2bae0464d779 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/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({