From e56b3e1a3740b9388dbeb7d785202ce617fbc207 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Tue, 27 Oct 2026 04:57:04 +0100 Subject: [PATCH 1/6] perf(server): skip stale message updates during reconnect --- .../ThreadTransportPerformance.test.ts | 48 ++++++++++++- apps/server/src/ws.test.ts | 68 +++++++++++++++++++ apps/server/src/ws.ts | 9 ++- 3 files changed, 121 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts b/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts index 7eb434176a137..dba94bf87b45f 100644 --- a/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts +++ b/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts @@ -1,5 +1,6 @@ import { EventId, + MessageId, OrchestrationV2RpcSchemas, ThreadId, TurnItemId, @@ -13,6 +14,7 @@ import type * as RpcMessage from "effect/rpc/RpcMessage"; import { describe, expect, it } from "vite-plus/test"; import { projectDomainEventForWire } from "./WireProjection.ts"; +import { coalesceLiveUpdatedEvents } from "./ThreadLiveEventCoalescer.ts"; const OUTPUT_BYTES_PER_ROW = 8_192; const NOW = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"); @@ -29,11 +31,14 @@ function encodedBytes(value: unknown): number { return Buffer.byteLength(JSON.stringify(value), "utf8"); } -function encodedThreadStreamChunkBytes(value: OrchestrationV2ThreadStreamItem): number { +function encodedThreadStreamChunkBytes( + value: OrchestrationV2ThreadStreamItem, + ...values: ReadonlyArray +): number { const response = { _tag: "Chunk", requestId: 0, - values: [encodeThreadStreamItem(value)], + values: [encodeThreadStreamItem(value), ...values.map((item) => encodeThreadStreamItem(item))], } satisfies RpcMessage.ResponseChunkEncoded; return encodedBytes(response); } @@ -72,6 +77,45 @@ function makeTurnItemEvent(item: OrchestrationV2TurnItem): OrchestrationV2Domain } describe("thread transport payload budget", () => { + it("sends only the newest growing message when coalescing a reconnect replay", () => { + const replay = Array.from({ length: 64 }, (_, index) => { + const sequence = index + 1; + return { + kind: "event", + sequence, + event: { + id: EventId.make(`event-${sequence}`), + type: "message.updated", + threadId: THREAD_ID, + occurredAt: NOW, + payload: { + id: MessageId.make("message-perf"), + threadId: THREAD_ID, + runId: null, + nodeId: null, + createdBy: "agent", + creationSource: "provider", + role: "assistant", + text: "x".repeat(sequence * 256), + attachments: [], + streaming: true, + createdAt: NOW, + updatedAt: NOW, + }, + }, + } satisfies Extract; + }); + const coalesced = coalesceLiveUpdatedEvents(replay); + const beforeBytes = encodedThreadStreamChunkBytes(replay[0]!, ...replay.slice(1)); + const afterBytes = encodedThreadStreamChunkBytes(coalesced[0]!, ...coalesced.slice(1)); + + expect(coalesced).toEqual([replay.at(-1)]); + expect(coalesced[0]?.sequence).toBe(64); + expect(beforeBytes).toBe(559_383); + expect(afterBytes).toBe(16_845); + expect(afterBytes / beforeBytes).toBeLessThan(0.04); + }); + it("omits command output from a live event without changing persisted source data", () => { const sourceItem = makeCommandItem(0); const sourceEvent = makeTurnItemEvent(sourceItem); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index f0d2e104ca864..655ded0a64dca 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -299,6 +299,74 @@ it.live.each(["thread", "shell"] as const)( }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); +it.effect("coalesces streaming replay before synchronization and the live tail", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread:streaming-replay"); + const now = "2026-10-27T00:00:00.000Z"; + const insert = (version: number, text: string, streaming: boolean) => sql<{ + sequence: number; + }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`streaming-replay:${version}`}, 'thread', ${threadId}, ${version}, 'message.updated', + ${now}, 'server', ${JSON.stringify({ + createdBy: "agent", + creationSource: "web", + id: "message:streaming-replay", + threadId, + runId: null, + nodeId: null, + role: "assistant", + text, + attachments: [], + streaming, + createdAt: now, + updatedAt: now, + })}, '{}', 2 + ) RETURNING sequence + `; + yield* insert(0, "H", true); + yield* insert(1, "He", true); + const [latest] = yield* insert(2, "Hello", true); + const [completed] = yield* insert(3, "Hello!", false); + const items = yield* subscribeOrchestrationV2Thread({ + threadId, + afterSequence: 0, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap(Stream.runCollect), + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + streamStoredEventsFrom: () => + Stream.unwrap( + insert(4, "Next", false).pipe( + Effect.map(() => + store.readAgentEvents({ threadId, afterSequence: completed!.sequence }), + ), + ), + ), + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + ); + assert.deepEqual( + items.map((item) => + item.kind === "event" + ? [item.sequence, item.event.type === "message.updated" ? item.event.payload.text : null] + : [item.kind], + ), + [ + [latest!.sequence, "Hello"], + [completed!.sequence, "Hello!"], + ["synchronized"], + [completed!.sequence + 1, "Next"], + ], + ); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); + it.effect("shell snapshots omit idle children and older settled threads only on request", () => Effect.gen(function* () { const shell = ( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index e5696e6f5cd17..5ebb557c519e9 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -145,7 +145,10 @@ import { } from "./orchestration-v2/ShellStream.ts"; import { ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION } from "./orchestration-v2/ProjectionStore.ts"; import { bufferLiveStream } from "./orchestration-v2/LiveStreamBudget.ts"; -import { coalesceThreadLiveStream } from "./orchestration-v2/ThreadLiveEventCoalescer.ts"; +import { + coalesceLiveUpdatedEvents, + coalesceThreadLiveStream, +} from "./orchestration-v2/ThreadLiveEventCoalescer.ts"; import { buildBoundedThreadStreamSnapshot, decideThreadResume, @@ -966,7 +969,9 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs // Retained creation events may describe deleted threads; replace replay only if a snapshot exists. if (shell !== null) return yield* snapshotThenLive(); } - const replay = yield* loadReplayThrough(input.afterSequence, highWater); + const replay = coalesceLiveUpdatedEvents( + yield* loadReplayThrough(input.afterSequence, highWater), + ); const plan = decideThreadResume({ afterSequence: input.afterSequence, highWater, From 805adb5f58affbda8ded8e4d9373369ea934049a Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:56:12 +0100 Subject: [PATCH 2/6] perf(client): keep sends responsive during history loads --- .../src/state/threadCommands.test.ts | 61 ++++++++++++++++++- .../src/state/threadCommands.ts | 3 +- 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index d61213c243ea8..dfed5233d66f9 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -3,6 +3,7 @@ import { CommandId, RuntimeRequestId, EnvironmentId, + MessageId, RunId, ORCHESTRATION_V2_WS_METHODS, ProjectId, @@ -86,6 +87,7 @@ const SNAPSHOT: OrchestrationV2ShellSnapshot = { }; const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (options?: { + readonly loadEarlier?: () => Effect.Effect; readonly reconcilePendingWork?: (threadId: ThreadId) => Effect.Effect; }) { const requests = yield* Queue.unbounded<{ @@ -126,7 +128,7 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (optio ThreadHistoryController.ThreadHistoryController.of({ register: () => Effect.die("unused"), unregister: () => Effect.void, - loadEarlier: () => Effect.die("unused"), + loadEarlier: () => options?.loadEarlier?.() ?? Effect.die("unused"), reconcilePendingWork: (_environmentId, threadId) => options?.reconcilePendingWork?.(threadId) ?? Queue.offer(reconciled, threadId), }), @@ -142,6 +144,63 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (optio return { registry, commands, snapshotAtom, visibleAtom, requests, reconciled }; }); +describe("history loading", () => { + it.effect.each([ + ["startTurn", "message.dispatch"], + ["interruptTurn", "run.interrupt"], + ] as const)("dispatches %s while earlier history is pending", ([action, commandType]) => + Effect.gen(function* () { + const historyStarted = yield* Deferred.make(); + const historyReply = + yield* Deferred.make(); + const h = yield* makeHarness({ + loadEarlier: () => + Deferred.succeed(historyStarted, undefined).pipe( + Effect.andThen(Deferred.await(historyReply)), + ), + }); + yield* Effect.addFinalizer(() => Deferred.succeed(historyReply, { _tag: "loaded" })); + const history = h.commands.loadEarlierHistory.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID }, + }); + yield* Deferred.await(historyStarted); + const input = { + threadId: THREAD_ID, + runId: RunId.make("running-run"), + message: { + messageId: MessageId.make("message"), + role: "user", + text: "Send while history loads", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + dispatchMode: "start", + } as const; + const result = h.commands[action].run(h.registry, { + environmentId: ENVIRONMENT_ID, + input, + }); + const controlThreadId = ThreadId.make("control-thread"); + const control = h.commands.interruptTurn.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: controlThreadId, runId: RunId.make("control-run") }, + }); + const first = yield* Queue.take(h.requests); + yield* Deferred.succeed(first.reply, { sequence: 1 }); + yield* Deferred.succeed(historyReply, { _tag: "loaded" }); + const second = yield* Queue.take(h.requests); + yield* Deferred.succeed(second.reply, { sequence: 2 }); + const results = yield* Effect.promise(() => Promise.all([history, result, control])); + + expect(first.command).toMatchObject({ type: commandType, threadId: THREAD_ID }); + expect(second.command).toMatchObject({ type: "run.interrupt", threadId: controlThreadId }); + expect(results.map((value) => value._tag)).toEqual(["Success", "Success", "Success"]); + }), + ); +}); + describe("interrupt turn", () => { // A stale cached item has nothing left to interrupt, so the server can reject Stop. it.effect("reconciles pending work when the server rejects the interrupt", () => diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 84a290057f578..8c3bf3d45ea9a 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -434,7 +434,8 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency: { mode: "serial", - key: ({ environmentId, input }) => JSON.stringify([environmentId, input.threadId]), + key: ({ environmentId, input }) => + `history:${JSON.stringify([environmentId, input.threadId])}`, }, }), uploadFeedback: createEnvironmentRpcCommand(runtime, { From eeb09e80ea98c126c162c4f8ac151204ce5e05d5 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:59:31 +0100 Subject: [PATCH 3/6] perf(client): reuse fresh descriptors during connection setup --- .../src/authorization/layer.test.ts | 200 ++++++++++++++++-- .../src/authorization/service.ts | 54 ++--- .../client-runtime/src/connection/resolver.ts | 37 ++-- 3 files changed, 217 insertions(+), 74 deletions(-) diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index d46878dedcd83..952de03b5ebdf 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -1,4 +1,8 @@ -import { AuthStandardClientScopes, EnvironmentId } from "@t3tools/contracts"; +import { + AuthStandardClientScopes, + EnvironmentId, + ORCHESTRATION_PROTOCOL_VERSION, +} from "@t3tools/contracts"; import { type RelayEnvironmentConnectResponse } from "@t3tools/contracts/relay"; import { describe, expect, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; @@ -15,6 +19,11 @@ import * as RpcHttp from "../rpc/http.ts"; import * as ClientCapabilities from "../platform/capabilities.ts"; import * as RemoteEnvironmentAuthorization from "./service.ts"; import * as TokenStore from "./tokenStore.ts"; +import * as ConnectionResolver from "../connection/resolver.ts"; +import * as ConnectionProfileStore from "../connection/profileStore.ts"; +import * as ConnectionCredentialStore from "../connection/credentialStore.ts"; +import { BearerConnectionCredential, BearerConnectionProfile } from "../connection/catalog.ts"; +import { BearerConnectionTarget } from "../connection/model.ts"; const ENVIRONMENT_ID = EnvironmentId.make("environment-1"); const ENDPOINT = { @@ -43,12 +52,17 @@ const BOOTSTRAP: RelayEnvironmentConnectResponse = { type RecordedResponse = Response | ((init: RequestInit) => Response); -function recordedFetch(responses: ReadonlyArray) { +function recordedFetch( + responses: + | ReadonlyArray + | ((input: RequestInfo | URL, init: RequestInit) => Response), +) { const calls: Array = []; let responseIndex = 0; const fetchFn = ((input, init) => { calls.push([input, init ?? {}]); - const response = responses[responseIndex++]; + const response = + typeof responses === "function" ? responses(input, init ?? {}) : responses[responseIndex++]; return response === undefined ? Promise.reject(new Error(`Unexpected fetch call to ${String(input)}`)) : Promise.resolve(typeof response === "function" ? response(init ?? {}) : response); @@ -102,7 +116,7 @@ const persistedToken = ( const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* (input: { readonly initialToken?: TokenStore.RemoteDpopAccessToken; - readonly responses: ReadonlyArray; + readonly responses: Parameters[0]; readonly bootstrap?: RelayEnvironmentConnectResponse; readonly beforeBootstrap?: Effect.Effect; readonly beforePut?: Effect.Effect; @@ -186,7 +200,7 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* ( resetTokenCache: Effect.void, }); const layer = RemoteEnvironmentAuthorization.layer.pipe( - Layer.provide( + Layer.provideMerge( Layer.mergeAll( RpcHttp.layerRemoteHttpClient(fetch.fetchFn), Layer.succeed(ManagedRelay.ManagedRelayDpopSigner, signer), @@ -225,15 +239,123 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* ( }; }); +const makeResolverHarness = Effect.fn("TestRemoteAuthorization.makeResolverHarness")(function* ( + responses: Parameters[0], + options?: { + readonly authorization?: "t3-connect"; + readonly initialToken?: TokenStore.RemoteDpopAccessToken; + readonly bootstrap?: RelayEnvironmentConnectResponse; + }, +) { + const harness = yield* makeHarness({ ...options, responses }); + const target = new BearerConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: DESCRIPTOR.label, + connectionId: "direct-1", + }); + const profile = new BearerConnectionProfile({ + connectionId: target.connectionId, + environmentId: target.environmentId, + label: target.label, + httpBaseUrl: ENDPOINT.httpBaseUrl, + wsBaseUrl: ENDPOINT.wsBaseUrl, + ...(options?.authorization === undefined ? {} : { authorization: options.authorization }), + }); + const unused = () => Effect.die("Unexpected connection store or SSH operation"); + const layer = ConnectionResolver.layer.pipe( + Layer.provide( + Layer.mergeAll( + harness.layer, + Layer.succeed(ConnectionProfileStore.ConnectionProfileStore, { + get: unused, + put: unused, + remove: unused, + }), + Layer.succeed(ConnectionCredentialStore.ConnectionCredentialStore, { + get: () => Effect.succeedSome(new BearerConnectionCredential({ token: "bearer-token" })), + put: unused, + remove: unused, + }), + Layer.succeed(ClientCapabilities.PrimaryEnvironmentAuth, { + bearerToken: Effect.succeedNone, + }), + Layer.succeed(ClientCapabilities.SshEnvironmentGateway, { + provision: unused, + prepare: unused, + disconnect: unused, + }), + ), + ), + ); + return { ...harness, layer, entry: { target, profile: Option.some(profile), enabled: true } }; +}); + describe("RemoteEnvironmentAuthorization", () => { - it.effect("reuses a validated bearer descriptor while issuing fresh websocket tickets", () => + it.effect("uses one fresh descriptor GET for a cold direct bearer prepare", () => + Effect.gen(function* () { + const descriptor = { + ...DESCRIPTOR, + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, + }; + const harness = yield* makeResolverHarness((url) => + String(url).endsWith("/.well-known/t3/environment") + ? Response.json(descriptor) + : websocketTicket("cold-ticket"), + ); + const prepared = yield* Effect.gen(function* () { + const resolver = yield* ConnectionResolver.ConnectionResolver; + return yield* resolver.prepare(harness.entry); + }).pipe(Effect.provide(harness.layer)); + + expect(prepared.socketUrl).toContain("wsTicket=cold-ticket"); + expect(prepared).not.toHaveProperty("descriptor"); + expect( + harness.fetch.calls + .filter(([url]) => String(url).endsWith("/.well-known/t3/environment")) + .map(([, init]) => init.method ?? "GET"), + ).toEqual(["GET"]); + }), + ); + + it.effect.each([ + { + name: "identity", + environmentId: EnvironmentId.make("another-environment"), + reason: "configuration", + }, + { name: "protocol", orchestrationProtocolVersion: 0, reason: "unsupported" }, + ])("checks fresh $name data on a warm direct bearer prepare", (change) => Effect.gen(function* () { + let descriptor = { + ...DESCRIPTOR, + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, + }; + const harness = yield* makeResolverHarness((url) => + String(url).endsWith("/.well-known/t3/environment") + ? Response.json(descriptor) + : websocketTicket("warm-ticket"), + ); + yield* Effect.gen(function* () { + const resolver = yield* ConnectionResolver.ConnectionResolver; + yield* resolver.prepare(harness.entry); + descriptor = { ...descriptor, ...change }; + const failure = yield* resolver.prepare(harness.entry).pipe(Effect.flip); + expect(failure).toMatchObject({ + _tag: "ConnectionBlockedError", + reason: change.reason, + }); + }).pipe(Effect.provide(harness.layer)); + }), + ); + + it.effect("fetches a fresh bearer descriptor while issuing fresh websocket tickets", () => + Effect.gen(function* () { + let ticketCount = 0; const harness = yield* makeHarness({ - responses: [ - Response.json(DESCRIPTOR), - websocketTicket("first-ticket"), - websocketTicket("second-ticket"), - ], + responses: (url) => + String(url).endsWith("/.well-known/t3/environment") + ? Response.json(DESCRIPTOR) + : websocketTicket(++ticketCount === 1 ? "first-ticket" : "second-ticket"), }); const [first, second] = yield* Effect.gen(function* () { @@ -253,7 +375,7 @@ describe("RemoteEnvironmentAuthorization", () => { expect(second.socketUrl).toContain("wsTicket=second-ticket"); expect( harness.fetch.calls.filter(([url]) => String(url).endsWith("/.well-known/t3/environment")), - ).toHaveLength(1); + ).toHaveLength(2); expect( harness.fetch.calls.filter(([url]) => String(url).endsWith("/api/auth/websocket-ticket/rpc"), @@ -262,6 +384,60 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); + it.effect.each(["cached", "renewed"] as const)( + "uses one fresh direct descriptor with a %s DPoP token and permits old-host updates", + (tokenState) => + Effect.gen(function* () { + const directDescriptor = { ...DESCRIPTOR, orchestrationProtocolVersion: 0 }; + const relayHttpBaseUrl = "https://relay-route.example.test"; + const harness = yield* makeResolverHarness( + (url) => { + const endpoint = String(url); + if (endpoint.endsWith("/.well-known/t3/environment")) { + return Response.json( + endpoint.startsWith(ENDPOINT.httpBaseUrl) + ? directDescriptor + : { ...DESCRIPTOR, orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION }, + ); + } + return endpoint.endsWith("/oauth/token") + ? accessToken("renewed-token") + : websocketTicket("direct-ticket"); + }, + { + authorization: "t3-connect", + ...(tokenState === "cached" ? { initialToken: persistedToken() } : {}), + bootstrap: { + ...BOOTSTRAP, + endpoint: { + ...ENDPOINT, + httpBaseUrl: relayHttpBaseUrl, + wsBaseUrl: "wss://relay-route.example.test", + }, + }, + }, + ); + yield* Effect.gen(function* () { + const resolver = yield* ConnectionResolver.ConnectionResolver; + const { prepared, descriptor } = yield* resolver.prepareForUpdate(harness.entry); + expect(descriptor).toMatchObject(directDescriptor); + expect(prepared.socketUrl).toContain("wsTicket=direct-ticket"); + expect(prepared.httpBaseUrl).toBe(ENDPOINT.httpBaseUrl); + expect(prepared).not.toHaveProperty("descriptor"); + expect( + harness.fetch.calls.filter( + ([url]) => String(url) === `${ENDPOINT.httpBaseUrl}/.well-known/t3/environment`, + ), + ).toHaveLength(1); + expect(yield* resolver.prepare(harness.entry).pipe(Effect.flip)).toMatchObject({ + _tag: "ConnectionBlockedError", + reason: "unsupported", + }); + }).pipe(Effect.provide(harness.layer)); + expect(yield* Ref.get(harness.bootstrapCalls)).toBe(tokenState === "cached" ? 0 : 1); + }), + ); + it.effect("reuses a same-account persisted environment token without contacting the relay", () => Effect.gen(function* () { const cached = new TokenStore.RemoteDpopAccessToken({ diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index d4c5ff9369616..2962393d7b4a2 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -33,7 +33,6 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; -import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; @@ -50,6 +49,7 @@ export interface AuthorizedRemoteEnvironment { readonly httpBaseUrl: string; readonly socketUrl: string; readonly httpAuthorization: PreparedHttpAuthorization; + readonly descriptor?: ExecutionEnvironmentDescriptor; } export interface AuthorizedRemoteHttpEnvironment { @@ -86,7 +86,6 @@ export class RemoteEnvironmentAuthorization extends Context.Service< >()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {} const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000; -const BEARER_DESCRIPTOR_CACHE_TTL_MS = 10_000; const DPOP_AUTHORIZATION_TIMEOUT_MS = 30_000; function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) { @@ -127,16 +126,6 @@ export const make = Effect.gen(function* () { readonly token: Effect.Effect; } >(); - const bearerDescriptors = yield* Ref.make< - ReadonlyMap< - EnvironmentId, - { - readonly httpBaseUrl: string; - readonly descriptor: ExecutionEnvironmentDescriptor; - readonly validatedAtEpochMs: number; - } - > - >(new Map()); /** * A saved LAN address can belong to another machine on a different network. @@ -147,33 +136,15 @@ export const make = Effect.gen(function* () { readonly httpBaseUrl: string; readonly connectionMethod: ClientConnectionMethod; }) { - const now = yield* Clock.currentTimeMillis; - const cachedDescriptor = (yield* Ref.get(bearerDescriptors)).get(input.expectedEnvironmentId); - const canReuseDescriptor = - cachedDescriptor?.httpBaseUrl === input.httpBaseUrl && - cachedDescriptor.validatedAtEpochMs + BEARER_DESCRIPTOR_CACHE_TTL_MS > now; - const descriptor = canReuseDescriptor - ? cachedDescriptor.descriptor - : yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - ); + const descriptor = yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + ); if (descriptor.environmentId !== input.expectedEnvironmentId) { return yield* environmentMismatchError({ expected: input.expectedEnvironmentId, actual: descriptor.environmentId, }); } - if (!canReuseDescriptor) { - yield* Ref.update(bearerDescriptors, (current) => { - const next = new Map(current); - next.set(input.expectedEnvironmentId, { - httpBaseUrl: input.httpBaseUrl, - descriptor, - validatedAtEpochMs: now, - }); - return next; - }); - } return descriptor; }, ); @@ -204,6 +175,7 @@ export const make = Effect.gen(function* () { label: descriptor.label, httpBaseUrl: input.httpBaseUrl, socketUrl, + descriptor, httpAuthorization: { _tag: "Bearer" as const, token: input.bearerToken, @@ -492,16 +464,18 @@ export const make = Effect.gen(function* () { input: Parameters[0], ) { const endpoint = input.directEndpoint; - if (endpoint !== undefined) { - yield* verifyDirectEndpoint({ - expectedEnvironmentId: input.expectedEnvironmentId, - httpBaseUrl: endpoint.httpBaseUrl, - connectionMethod: "direct", - }); - } + const descriptor = + endpoint === undefined + ? undefined + : yield* verifyDirectEndpoint({ + expectedEnvironmentId: input.expectedEnvironmentId, + httpBaseUrl: endpoint.httpBaseUrl, + connectionMethod: "direct", + }); const authorized = (token: TokenStore.RemoteDpopAccessToken, socketUrl: string) => ({ ...httpAuthorization(token, endpoint?.httpBaseUrl), socketUrl, + ...(descriptor === undefined ? {} : { descriptor }), }); let selected = yield* getDpopToken(input); if (selected.fromCache) { diff --git a/packages/client-runtime/src/connection/resolver.ts b/packages/client-runtime/src/connection/resolver.ts index 08908a64fd641..d3b22143d327a 100644 --- a/packages/client-runtime/src/connection/resolver.ts +++ b/packages/client-runtime/src/connection/resolver.ts @@ -171,11 +171,7 @@ const makeBearerBroker = Effect.fn("clientRuntime.connection.broker.makeBearer") connectionMethod: "direct", }); return { - environmentId: authorized.environmentId, - label: authorized.label, - httpBaseUrl: authorized.httpBaseUrl, - socketUrl: authorized.socketUrl, - httpAuthorization: authorized.httpAuthorization, + ...authorized, target, } satisfies PreparedConnection; }); @@ -190,11 +186,7 @@ const makeRelayBroker = Effect.fn("clientRuntime.connection.broker.makeRelay")(f expectedEnvironmentId: target.environmentId, }); return { - environmentId: authorized.environmentId, - label: authorized.label, - httpBaseUrl: authorized.httpBaseUrl, - socketUrl: authorized.socketUrl, - httpAuthorization: authorized.httpAuthorization, + ...authorized, target, } satisfies PreparedConnection; }, @@ -265,11 +257,7 @@ const makeSshBroker = Effect.fn("clientRuntime.connection.broker.makeSsh")(funct connectionMethod: "ssh", }); return { - environmentId: authorized.environmentId, - label: authorized.label, - httpBaseUrl: authorized.httpBaseUrl, - socketUrl: authorized.socketUrl, - httpAuthorization: authorized.httpAuthorization, + ...authorized, target, } satisfies PreparedConnection; }); @@ -291,7 +279,9 @@ export const make = Effect.gen(function* () { "connection.environment.id": target.environmentId, "connection.target.kind": target._tag, }); - const prepared = yield* (() => { + const authorized: PreparedConnection & { + readonly descriptor?: ExecutionEnvironmentDescriptor; + } = yield* (() => { switch (target._tag) { case "PrimaryConnectionTarget": return primary(target); @@ -303,12 +293,15 @@ export const make = Effect.gen(function* () { return ssh({ ...entry, target }); } })(); - const descriptor = yield* fetchRemoteEnvironmentDescriptor({ - httpBaseUrl: prepared.httpBaseUrl, - }).pipe( - Effect.mapError(mapRemoteEnvironmentError), - Effect.provideService(HttpClient.HttpClient, httpClient), - ); + const { descriptor: authorizedDescriptor, ...prepared } = authorized; + const descriptor = + authorizedDescriptor ?? + (yield* fetchRemoteEnvironmentDescriptor({ + httpBaseUrl: prepared.httpBaseUrl, + }).pipe( + Effect.mapError(mapRemoteEnvironmentError), + Effect.provideService(HttpClient.HttpClient, httpClient), + )); if (descriptor.environmentId !== target.environmentId) { return yield* environmentMismatchError({ expected: target.environmentId, From ed63c24223e922c8185993d7094100a99cbe2044 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:02:54 +0100 Subject: [PATCH 4/6] test(client): benchmark remote setup and history contention --- .../src/remotePerformance.bench.ts | 250 +++++++++++++++++- 1 file changed, 249 insertions(+), 1 deletion(-) diff --git a/packages/client-runtime/src/remotePerformance.bench.ts b/packages/client-runtime/src/remotePerformance.bench.ts index e4452ffa04d14..c9ea8b0bd6804 100644 --- a/packages/client-runtime/src/remotePerformance.bench.ts +++ b/packages/client-runtime/src/remotePerformance.bench.ts @@ -2,13 +2,22 @@ import { EnvironmentId, EventId, MessageId, + ORCHESTRATION_PROTOCOL_VERSION, + ORCHESTRATION_V2_WS_METHODS, + RunId, + type OrchestrationV2Command, type OrchestrationV2DomainEvent, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Crypto from "effect/Crypto"; +import * as Deferred from "effect/Deferred"; +import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as SubscriptionRef from "effect/SubscriptionRef"; import { HttpClient, HttpClientResponse } from "effect/http"; -import { describe, test } from "vite-plus/test"; +import { Atom, AtomRegistry } from "effect/reactivity"; +import { describe, expect, test } from "vite-plus/test"; import { issueRemoteWebSocketTicket } from "./authorization/remote.ts"; import { PrimaryConnectionTarget } from "./connection/model.ts"; @@ -17,6 +26,18 @@ import type { RemoteEnvironmentRequestError } from "./rpc/http.ts"; import { fetchEnvironmentThreadSnapshot } from "./state/threadSnapshotHttp.ts"; import { applyOrchestrationV2ProjectionEvent } from "./state/orchestrationV2Projection.ts"; import { v2Projection, v2Now } from "./state/orchestrationV2TestFixtures.ts"; +import * as RemoteAuthorization from "./authorization/service.ts"; +import { RemoteDpopAccessTokenStore } from "./authorization/tokenStore.ts"; +import { ConnectionCredentialStore } from "./connection/credentialStore.ts"; +import { ConnectionProfileStore } from "./connection/profileStore.ts"; +import * as EnvironmentRegistry from "./connection/registry.ts"; +import * as ConnectionResolver from "./connection/resolver.ts"; +import * as EnvironmentSupervisor from "./connection/supervisor.ts"; +import * as Capabilities from "./platform/capabilities.ts"; +import { ManagedRelayClient, ManagedRelayDpopSigner } from "./relay/managedRelay.ts"; +import type { RpcSession } from "./rpc/session.ts"; +import { createThreadEnvironmentAtoms } from "./state/threadCommands.ts"; +import { ThreadHistoryController } from "./state/threadHistoryController.ts"; const timestamp = "2026-09-01T00:00:00.000Z"; const runOptions = { warmupTime: 1_000, time: 1_500 }; @@ -49,9 +70,11 @@ const responses = { label: target.label, platform: { os: "linux", arch: "x64" }, serverVersion: "0.0.0-test", + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, capabilities: { repositoryIdentity: true }, }, "/api/auth/websocket-ticket": { ticket: "test-ticket", expiresAt: timestamp }, + "/api/auth/websocket-ticket/rpc": { ticket: "test-ticket", expiresAt: timestamp }, "/api/orchestration/threads/thread-v2": { snapshotSequence: 1, projection: thread }, }; const httpClient = HttpClient.make((request) => @@ -97,6 +120,231 @@ describe("remote HTTP processing with an in-memory transport", () => { } }); +const controlledRunOptions = { warmup: false, time: 0, iterations: 3 }; +const unused = () => Effect.die("Unexpected benchmark dependency"); +const presentation = Layer.succeed(Capabilities.ClientPresentation, { + metadata: { label: "Benchmark", deviceType: "mobile", os: "test" }, +}); + +const prepareSample = (delayMs: number) => + Effect.scoped( + Effect.gen(function* () { + const paths: string[] = []; + let observedHttpWaitSeconds = 0; + const transport = HttpClient.make((request) => + Effect.gen(function* () { + const path = new URL(request.url).pathname as keyof typeof responses; + paths.push(path); + if (delayMs > 0) { + const started = performance.now(); + yield* Effect.sleep(delayMs); + observedHttpWaitSeconds += (performance.now() - started) / 1_000; + } + return HttpClientResponse.fromWeb(request, Response.json(responses[path])); + }), + ); + const dependencies = Layer.mergeAll( + Layer.succeed(HttpClient.HttpClient, transport), + presentation, + Layer.succeed(ManagedRelayDpopSigner, { thumbprint: unused(), createProof: unused }), + Layer.succeed(ManagedRelayClient, { + relayUrl: "https://relay.example.test", + listEnvironments: unused, + listDevices: unused, + createEnvironmentLinkChallenge: unused, + linkEnvironment: unused, + unlinkEnvironment: unused, + getEnvironmentStatus: unused, + connectEnvironment: unused, + registerDevice: unused, + unregisterDevice: unused, + registerLiveActivity: unused, + getAgentActivitySnapshot: unused, + resetTokenCache: Effect.void, + }), + Layer.succeed(Capabilities.CloudSession, { identity: unused(), clerkToken: unused() }), + Layer.succeed(Capabilities.RelayDeviceIdentity, { deviceId: unused() }), + Layer.succeed(RemoteDpopAccessTokenStore, { get: unused, put: unused, remove: unused }), + Layer.succeed(Capabilities.PrimaryEnvironmentAuth, { + bearerToken: Effect.succeedSome("test-token"), + }), + Layer.succeed(Capabilities.SshEnvironmentGateway, { + provision: unused, + prepare: unused, + disconnect: unused, + }), + Layer.succeed(ConnectionProfileStore, { get: unused, put: unused, remove: unused }), + Layer.succeed(ConnectionCredentialStore, { get: unused, put: unused, remove: unused }), + ); + const layer = ConnectionResolver.layer.pipe( + Layer.provide(RemoteAuthorization.layer.pipe(Layer.provideMerge(dependencies))), + ); + return yield* Effect.gen(function* () { + const resolver = yield* ConnectionResolver.ConnectionResolver; + const samples = []; + for (const phase of ["cold", "warm"] as const) { + paths.length = 0; + observedHttpWaitSeconds = 0; + const started = performance.now(); + const prepared = yield* resolver.prepare({ + target, + profile: Option.none(), + enabled: true, + }); + expect(prepared.socketUrl).toContain("test-ticket"); + samples.push({ + phase, + seconds: (performance.now() - started) / 1_000, + observedHttpWaitSeconds, + paths: [...paths], + }); + } + return samples; + }).pipe(Effect.provide(layer)); + }), + ); + +const commandSample = (action: "startTurn" | "interruptTurn", historyDelayMs: number) => + Effect.scoped( + Effect.gen(function* () { + const historyStarted = yield* Deferred.make(); + let dispatchedAt = 0; + let observedHistoryWaitSeconds = 0; + let observedReplyWaitSeconds = 0; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target, + session: yield* SubscriptionRef.make( + Option.some({ + client: { + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: Effect.fn( + "RemotePerformance.dispatch", + )(function* (command: OrchestrationV2Command) { + dispatchedAt = performance.now(); + expect(command.type).toBe( + action === "startTurn" ? "message.dispatch" : "run.interrupt", + ); + const started = performance.now(); + yield* Effect.sleep(50); + observedReplyWaitSeconds = (performance.now() - started) / 1_000; + return { sequence: 1 }; + }), + }, + } as unknown as RpcSession), + ), + } as EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const runtime = Atom.runtime( + Layer.mergeAll( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, { + run: (_id, effect) => + Effect.provideService( + effect, + EnvironmentSupervisor.EnvironmentSupervisor, + supervisor, + ), + } as EnvironmentRegistry.EnvironmentRegistry["Service"]), + Layer.succeed( + Crypto.Crypto, + Crypto.make({ + randomBytes: (size) => new Uint8Array(size), + digest: (_algorithm, data) => Effect.succeed(data), + }), + ), + Layer.succeed(ThreadHistoryController, { + register: unused, + unregister: Effect.fn("ThreadHistoryController.unregister")(() => Effect.void), + reconcilePendingWork: Effect.fn("ThreadHistoryController.reconcilePendingWork")( + () => Effect.void, + ), + loadEarlier: Effect.fn("RemotePerformance.loadEarlier")(function* () { + yield* Deferred.succeed(historyStarted, undefined); + if (historyDelayMs > 0) { + const started = performance.now(); + yield* Effect.sleep(historyDelayMs); + observedHistoryWaitSeconds = (performance.now() - started) / 1_000; + } + return { _tag: "loaded" } as const; + }), + }), + ), + ); + const registry = AtomRegistry.make(); + yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); + const commands = createThreadEnvironmentAtoms(runtime, () => Atom.make(null)); + const history = commands.loadEarlierHistory.run(registry, { + environmentId: target.environmentId, + input: { threadId: thread.thread.id }, + }); + yield* Deferred.await(historyStarted); + const started = performance.now(); + const result = yield* Effect.promise(() => + commands[action].run(registry, { + environmentId: target.environmentId, + input: { + threadId: thread.thread.id, + runId: RunId.make("running-run"), + message: { + messageId: MessageId.make("send"), + role: "user", + text: "Send", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + dispatchMode: "start", + }, + }), + ); + const completeSeconds = (performance.now() - started) / 1_000; + expect(result._tag).toBe("Success"); + expect(dispatchedAt).toBeGreaterThan(0); + expect((yield* Effect.promise(() => history))._tag).toBe("Success"); + return { + dispatchSeconds: (dispatchedAt - started) / 1_000, + completeSeconds, + observedHistoryWaitSeconds, + observedReplyWaitSeconds, + }; + }), + ); + +describe("controlled fixture latency in seconds, not phone or Wi-Fi measurements", () => { + for (const delayMs of [0, 100]) { + test(`prepare with ${delayMs}ms injected per HTTP response`, async ({ bench }) => { + const samples: Effect.Success>[] = []; + await bench("actual resolver cold and warm prepare", async () => { + samples.push(await Effect.runPromise(prepareSample(delayMs))); + }).run(controlledRunOptions); + Effect.runSync( + Effect.log( + JSON.stringify({ scenario: "prepare", injectedHttpSeconds: delayMs / 1_000, samples }), + ), + ); + }); + } + for (const action of ["startTurn", "interruptTurn"] as const) { + for (const historyDelayMs of [0, 250]) { + test(`${action} with ${historyDelayMs}ms injected history and 50ms command reply`, async ({ + bench, + }) => { + const samples: Effect.Success>[] = []; + await bench(`actual thread command scheduler ${action}`, async () => { + samples.push(await Effect.runPromise(commandSample(action, historyDelayMs))); + }).run(controlledRunOptions); + Effect.runSync( + Effect.log( + JSON.stringify({ + scenario: action, + injectedHistorySeconds: historyDelayMs / 1_000, + injectedCommandSeconds: 0.05, + samples, + }), + ), + ); + }); + } + } +}); + const delta: Extract = { id: EventId.make("delta"), type: "message.updated", From db41c18dc69938e7196e221cc7e79982d82e5815 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 22:22:18 +0000 Subject: [PATCH 5/6] test: takeover tests for #25754 message loading speedups Takeover's tests for the PR: one test per latency fix. - ws.test.ts: a resume replay sends only the newest streaming message.updated, in order, before the synchronized marker and live tail. - threadCommands.test.ts: Send dispatches while an earlier-history load on the same thread is pending. - authorization/layer.test.ts: a cold direct bearer prepare through the real resolver makes one descriptor GET and one ticket request; a warm prepare whose fresh descriptor changes identity is blocked. Main's TTL-cache test now expects a fresh descriptor per authorization. Drops the hardcoded byte-count transport test, the protocol, cached and renewed DPoP variants, and the remotePerformance.bench.ts additions. Co-Authored-By: Claude Opus 5.5 --- .../ThreadTransportPerformance.test.ts | 48 +--- .../src/authorization/layer.test.ts | 216 ++++++--------- .../src/remotePerformance.bench.ts | 250 +----------------- .../src/state/threadCommands.test.ts | 38 ++- 4 files changed, 98 insertions(+), 454 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts b/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts index dba94bf87b45f..7eb434176a137 100644 --- a/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts +++ b/apps/server/src/orchestration-v2/ThreadTransportPerformance.test.ts @@ -1,6 +1,5 @@ import { EventId, - MessageId, OrchestrationV2RpcSchemas, ThreadId, TurnItemId, @@ -14,7 +13,6 @@ import type * as RpcMessage from "effect/rpc/RpcMessage"; import { describe, expect, it } from "vite-plus/test"; import { projectDomainEventForWire } from "./WireProjection.ts"; -import { coalesceLiveUpdatedEvents } from "./ThreadLiveEventCoalescer.ts"; const OUTPUT_BYTES_PER_ROW = 8_192; const NOW = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"); @@ -31,14 +29,11 @@ function encodedBytes(value: unknown): number { return Buffer.byteLength(JSON.stringify(value), "utf8"); } -function encodedThreadStreamChunkBytes( - value: OrchestrationV2ThreadStreamItem, - ...values: ReadonlyArray -): number { +function encodedThreadStreamChunkBytes(value: OrchestrationV2ThreadStreamItem): number { const response = { _tag: "Chunk", requestId: 0, - values: [encodeThreadStreamItem(value), ...values.map((item) => encodeThreadStreamItem(item))], + values: [encodeThreadStreamItem(value)], } satisfies RpcMessage.ResponseChunkEncoded; return encodedBytes(response); } @@ -77,45 +72,6 @@ function makeTurnItemEvent(item: OrchestrationV2TurnItem): OrchestrationV2Domain } describe("thread transport payload budget", () => { - it("sends only the newest growing message when coalescing a reconnect replay", () => { - const replay = Array.from({ length: 64 }, (_, index) => { - const sequence = index + 1; - return { - kind: "event", - sequence, - event: { - id: EventId.make(`event-${sequence}`), - type: "message.updated", - threadId: THREAD_ID, - occurredAt: NOW, - payload: { - id: MessageId.make("message-perf"), - threadId: THREAD_ID, - runId: null, - nodeId: null, - createdBy: "agent", - creationSource: "provider", - role: "assistant", - text: "x".repeat(sequence * 256), - attachments: [], - streaming: true, - createdAt: NOW, - updatedAt: NOW, - }, - }, - } satisfies Extract; - }); - const coalesced = coalesceLiveUpdatedEvents(replay); - const beforeBytes = encodedThreadStreamChunkBytes(replay[0]!, ...replay.slice(1)); - const afterBytes = encodedThreadStreamChunkBytes(coalesced[0]!, ...coalesced.slice(1)); - - expect(coalesced).toEqual([replay.at(-1)]); - expect(coalesced[0]?.sequence).toBe(64); - expect(beforeBytes).toBe(559_383); - expect(afterBytes).toBe(16_845); - expect(afterBytes / beforeBytes).toBeLessThan(0.04); - }); - it("omits command output from a live event without changing persisted source data", () => { const sourceItem = makeCommandItem(0); const sourceEvent = makeTurnItemEvent(sourceItem); diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index 952de03b5ebdf..6237f38dbd739 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -19,11 +19,11 @@ import * as RpcHttp from "../rpc/http.ts"; import * as ClientCapabilities from "../platform/capabilities.ts"; import * as RemoteEnvironmentAuthorization from "./service.ts"; import * as TokenStore from "./tokenStore.ts"; -import * as ConnectionResolver from "../connection/resolver.ts"; -import * as ConnectionProfileStore from "../connection/profileStore.ts"; -import * as ConnectionCredentialStore from "../connection/credentialStore.ts"; import { BearerConnectionCredential, BearerConnectionProfile } from "../connection/catalog.ts"; +import * as ConnectionCredentialStore from "../connection/credentialStore.ts"; import { BearerConnectionTarget } from "../connection/model.ts"; +import * as ConnectionProfileStore from "../connection/profileStore.ts"; +import * as ConnectionResolver from "../connection/resolver.ts"; const ENVIRONMENT_ID = EnvironmentId.make("environment-1"); const ENDPOINT = { @@ -52,17 +52,12 @@ const BOOTSTRAP: RelayEnvironmentConnectResponse = { type RecordedResponse = Response | ((init: RequestInit) => Response); -function recordedFetch( - responses: - | ReadonlyArray - | ((input: RequestInfo | URL, init: RequestInit) => Response), -) { +function recordedFetch(responses: ReadonlyArray) { const calls: Array = []; let responseIndex = 0; const fetchFn = ((input, init) => { calls.push([input, init ?? {}]); - const response = - typeof responses === "function" ? responses(input, init ?? {}) : responses[responseIndex++]; + const response = responses[responseIndex++]; return response === undefined ? Promise.reject(new Error(`Unexpected fetch call to ${String(input)}`)) : Promise.resolve(typeof response === "function" ? response(init ?? {}) : response); @@ -116,7 +111,7 @@ const persistedToken = ( const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* (input: { readonly initialToken?: TokenStore.RemoteDpopAccessToken; - readonly responses: Parameters[0]; + readonly responses: ReadonlyArray; readonly bootstrap?: RelayEnvironmentConnectResponse; readonly beforeBootstrap?: Effect.Effect; readonly beforePut?: Effect.Effect; @@ -239,29 +234,29 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* ( }; }); -const makeResolverHarness = Effect.fn("TestRemoteAuthorization.makeResolverHarness")(function* ( - responses: Parameters[0], - options?: { - readonly authorization?: "t3-connect"; - readonly initialToken?: TokenStore.RemoteDpopAccessToken; - readonly bootstrap?: RelayEnvironmentConnectResponse; - }, +const prepareDirectBearer = Effect.fn("TestRemoteAuthorization.prepareDirectBearer")(function* ( + harness: Effect.Success>, + times: number, ) { - const harness = yield* makeHarness({ ...options, responses }); const target = new BearerConnectionTarget({ environmentId: ENVIRONMENT_ID, label: DESCRIPTOR.label, connectionId: "direct-1", }); - const profile = new BearerConnectionProfile({ - connectionId: target.connectionId, - environmentId: target.environmentId, - label: target.label, - httpBaseUrl: ENDPOINT.httpBaseUrl, - wsBaseUrl: ENDPOINT.wsBaseUrl, - ...(options?.authorization === undefined ? {} : { authorization: options.authorization }), - }); - const unused = () => Effect.die("Unexpected connection store or SSH operation"); + const entry = { + target, + profile: Option.some( + new BearerConnectionProfile({ + connectionId: target.connectionId, + environmentId: ENVIRONMENT_ID, + label: target.label, + httpBaseUrl: ENDPOINT.httpBaseUrl, + wsBaseUrl: ENDPOINT.wsBaseUrl, + }), + ), + enabled: true, + }; + const unused = () => Effect.die("unused"); const layer = ConnectionResolver.layer.pipe( Layer.provide( Layer.mergeAll( @@ -287,75 +282,74 @@ const makeResolverHarness = Effect.fn("TestRemoteAuthorization.makeResolverHarne ), ), ); - return { ...harness, layer, entry: { target, profile: Option.some(profile), enabled: true } }; + const resolver = yield* ConnectionResolver.ConnectionResolver.pipe(Effect.provide(layer)); + return yield* Effect.forEach(Array.from({ length: times }), () => + resolver.prepare(entry).pipe(Effect.result), + ); }); describe("RemoteEnvironmentAuthorization", () => { - it.effect("uses one fresh descriptor GET for a cold direct bearer prepare", () => + it.effect("prepares a cold direct bearer connection with one descriptor request", () => Effect.gen(function* () { - const descriptor = { - ...DESCRIPTOR, - orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, - }; - const harness = yield* makeResolverHarness((url) => - String(url).endsWith("/.well-known/t3/environment") - ? Response.json(descriptor) - : websocketTicket("cold-ticket"), - ); - const prepared = yield* Effect.gen(function* () { - const resolver = yield* ConnectionResolver.ConnectionResolver; - return yield* resolver.prepare(harness.entry); - }).pipe(Effect.provide(harness.layer)); + const harness = yield* makeHarness({ + responses: [ + Response.json({ + ...DESCRIPTOR, + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, + }), + websocketTicket("cold-ticket"), + ], + }); - expect(prepared.socketUrl).toContain("wsTicket=cold-ticket"); - expect(prepared).not.toHaveProperty("descriptor"); - expect( - harness.fetch.calls - .filter(([url]) => String(url).endsWith("/.well-known/t3/environment")) - .map(([, init]) => init.method ?? "GET"), - ).toEqual(["GET"]); + const [prepared] = yield* prepareDirectBearer(harness, 1); + + expect(prepared).toMatchObject({ + _tag: "Success", + success: { socketUrl: expect.stringContaining("wsTicket=cold-ticket") }, + }); + expect(harness.fetch.calls.map(([url]) => new URL(String(url)).pathname)).toEqual([ + "/.well-known/t3/environment", + "/api/auth/websocket-ticket/rpc", + ]); }), ); - it.effect.each([ - { - name: "identity", - environmentId: EnvironmentId.make("another-environment"), - reason: "configuration", - }, - { name: "protocol", orchestrationProtocolVersion: 0, reason: "unsupported" }, - ])("checks fresh $name data on a warm direct bearer prepare", (change) => + it.effect("blocks a warm direct bearer prepare when the environment changes identity", () => Effect.gen(function* () { - let descriptor = { - ...DESCRIPTOR, - orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, - }; - const harness = yield* makeResolverHarness((url) => - String(url).endsWith("/.well-known/t3/environment") - ? Response.json(descriptor) - : websocketTicket("warm-ticket"), - ); - yield* Effect.gen(function* () { - const resolver = yield* ConnectionResolver.ConnectionResolver; - yield* resolver.prepare(harness.entry); - descriptor = { ...descriptor, ...change }; - const failure = yield* resolver.prepare(harness.entry).pipe(Effect.flip); - expect(failure).toMatchObject({ - _tag: "ConnectionBlockedError", - reason: change.reason, - }); - }).pipe(Effect.provide(harness.layer)); + const harness = yield* makeHarness({ + responses: [ + Response.json({ + ...DESCRIPTOR, + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, + }), + websocketTicket("first-ticket"), + Response.json({ + ...DESCRIPTOR, + environmentId: "another-environment", + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, + }), + websocketTicket("second-ticket"), + ], + }); + + const [first, second] = yield* prepareDirectBearer(harness, 2); + + expect(first?._tag).toBe("Success"); + expect(second).toMatchObject({ + _tag: "Failure", + failure: { _tag: "ConnectionBlockedError", reason: "configuration" }, + }); }), ); - - it.effect("fetches a fresh bearer descriptor while issuing fresh websocket tickets", () => + it.effect("validates a fresh bearer descriptor while issuing fresh websocket tickets", () => Effect.gen(function* () { - let ticketCount = 0; const harness = yield* makeHarness({ - responses: (url) => - String(url).endsWith("/.well-known/t3/environment") - ? Response.json(DESCRIPTOR) - : websocketTicket(++ticketCount === 1 ? "first-ticket" : "second-ticket"), + responses: [ + Response.json(DESCRIPTOR), + websocketTicket("first-ticket"), + Response.json(DESCRIPTOR), + websocketTicket("second-ticket"), + ], }); const [first, second] = yield* Effect.gen(function* () { @@ -384,60 +378,6 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); - it.effect.each(["cached", "renewed"] as const)( - "uses one fresh direct descriptor with a %s DPoP token and permits old-host updates", - (tokenState) => - Effect.gen(function* () { - const directDescriptor = { ...DESCRIPTOR, orchestrationProtocolVersion: 0 }; - const relayHttpBaseUrl = "https://relay-route.example.test"; - const harness = yield* makeResolverHarness( - (url) => { - const endpoint = String(url); - if (endpoint.endsWith("/.well-known/t3/environment")) { - return Response.json( - endpoint.startsWith(ENDPOINT.httpBaseUrl) - ? directDescriptor - : { ...DESCRIPTOR, orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION }, - ); - } - return endpoint.endsWith("/oauth/token") - ? accessToken("renewed-token") - : websocketTicket("direct-ticket"); - }, - { - authorization: "t3-connect", - ...(tokenState === "cached" ? { initialToken: persistedToken() } : {}), - bootstrap: { - ...BOOTSTRAP, - endpoint: { - ...ENDPOINT, - httpBaseUrl: relayHttpBaseUrl, - wsBaseUrl: "wss://relay-route.example.test", - }, - }, - }, - ); - yield* Effect.gen(function* () { - const resolver = yield* ConnectionResolver.ConnectionResolver; - const { prepared, descriptor } = yield* resolver.prepareForUpdate(harness.entry); - expect(descriptor).toMatchObject(directDescriptor); - expect(prepared.socketUrl).toContain("wsTicket=direct-ticket"); - expect(prepared.httpBaseUrl).toBe(ENDPOINT.httpBaseUrl); - expect(prepared).not.toHaveProperty("descriptor"); - expect( - harness.fetch.calls.filter( - ([url]) => String(url) === `${ENDPOINT.httpBaseUrl}/.well-known/t3/environment`, - ), - ).toHaveLength(1); - expect(yield* resolver.prepare(harness.entry).pipe(Effect.flip)).toMatchObject({ - _tag: "ConnectionBlockedError", - reason: "unsupported", - }); - }).pipe(Effect.provide(harness.layer)); - expect(yield* Ref.get(harness.bootstrapCalls)).toBe(tokenState === "cached" ? 0 : 1); - }), - ); - it.effect("reuses a same-account persisted environment token without contacting the relay", () => Effect.gen(function* () { const cached = new TokenStore.RemoteDpopAccessToken({ diff --git a/packages/client-runtime/src/remotePerformance.bench.ts b/packages/client-runtime/src/remotePerformance.bench.ts index c9ea8b0bd6804..e4452ffa04d14 100644 --- a/packages/client-runtime/src/remotePerformance.bench.ts +++ b/packages/client-runtime/src/remotePerformance.bench.ts @@ -2,22 +2,13 @@ import { EnvironmentId, EventId, MessageId, - ORCHESTRATION_PROTOCOL_VERSION, - ORCHESTRATION_V2_WS_METHODS, - RunId, - type OrchestrationV2Command, type OrchestrationV2DomainEvent, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; -import * as Crypto from "effect/Crypto"; -import * as Deferred from "effect/Deferred"; -import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; -import * as SubscriptionRef from "effect/SubscriptionRef"; import { HttpClient, HttpClientResponse } from "effect/http"; -import { Atom, AtomRegistry } from "effect/reactivity"; -import { describe, expect, test } from "vite-plus/test"; +import { describe, test } from "vite-plus/test"; import { issueRemoteWebSocketTicket } from "./authorization/remote.ts"; import { PrimaryConnectionTarget } from "./connection/model.ts"; @@ -26,18 +17,6 @@ import type { RemoteEnvironmentRequestError } from "./rpc/http.ts"; import { fetchEnvironmentThreadSnapshot } from "./state/threadSnapshotHttp.ts"; import { applyOrchestrationV2ProjectionEvent } from "./state/orchestrationV2Projection.ts"; import { v2Projection, v2Now } from "./state/orchestrationV2TestFixtures.ts"; -import * as RemoteAuthorization from "./authorization/service.ts"; -import { RemoteDpopAccessTokenStore } from "./authorization/tokenStore.ts"; -import { ConnectionCredentialStore } from "./connection/credentialStore.ts"; -import { ConnectionProfileStore } from "./connection/profileStore.ts"; -import * as EnvironmentRegistry from "./connection/registry.ts"; -import * as ConnectionResolver from "./connection/resolver.ts"; -import * as EnvironmentSupervisor from "./connection/supervisor.ts"; -import * as Capabilities from "./platform/capabilities.ts"; -import { ManagedRelayClient, ManagedRelayDpopSigner } from "./relay/managedRelay.ts"; -import type { RpcSession } from "./rpc/session.ts"; -import { createThreadEnvironmentAtoms } from "./state/threadCommands.ts"; -import { ThreadHistoryController } from "./state/threadHistoryController.ts"; const timestamp = "2026-09-01T00:00:00.000Z"; const runOptions = { warmupTime: 1_000, time: 1_500 }; @@ -70,11 +49,9 @@ const responses = { label: target.label, platform: { os: "linux", arch: "x64" }, serverVersion: "0.0.0-test", - orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, capabilities: { repositoryIdentity: true }, }, "/api/auth/websocket-ticket": { ticket: "test-ticket", expiresAt: timestamp }, - "/api/auth/websocket-ticket/rpc": { ticket: "test-ticket", expiresAt: timestamp }, "/api/orchestration/threads/thread-v2": { snapshotSequence: 1, projection: thread }, }; const httpClient = HttpClient.make((request) => @@ -120,231 +97,6 @@ describe("remote HTTP processing with an in-memory transport", () => { } }); -const controlledRunOptions = { warmup: false, time: 0, iterations: 3 }; -const unused = () => Effect.die("Unexpected benchmark dependency"); -const presentation = Layer.succeed(Capabilities.ClientPresentation, { - metadata: { label: "Benchmark", deviceType: "mobile", os: "test" }, -}); - -const prepareSample = (delayMs: number) => - Effect.scoped( - Effect.gen(function* () { - const paths: string[] = []; - let observedHttpWaitSeconds = 0; - const transport = HttpClient.make((request) => - Effect.gen(function* () { - const path = new URL(request.url).pathname as keyof typeof responses; - paths.push(path); - if (delayMs > 0) { - const started = performance.now(); - yield* Effect.sleep(delayMs); - observedHttpWaitSeconds += (performance.now() - started) / 1_000; - } - return HttpClientResponse.fromWeb(request, Response.json(responses[path])); - }), - ); - const dependencies = Layer.mergeAll( - Layer.succeed(HttpClient.HttpClient, transport), - presentation, - Layer.succeed(ManagedRelayDpopSigner, { thumbprint: unused(), createProof: unused }), - Layer.succeed(ManagedRelayClient, { - relayUrl: "https://relay.example.test", - listEnvironments: unused, - listDevices: unused, - createEnvironmentLinkChallenge: unused, - linkEnvironment: unused, - unlinkEnvironment: unused, - getEnvironmentStatus: unused, - connectEnvironment: unused, - registerDevice: unused, - unregisterDevice: unused, - registerLiveActivity: unused, - getAgentActivitySnapshot: unused, - resetTokenCache: Effect.void, - }), - Layer.succeed(Capabilities.CloudSession, { identity: unused(), clerkToken: unused() }), - Layer.succeed(Capabilities.RelayDeviceIdentity, { deviceId: unused() }), - Layer.succeed(RemoteDpopAccessTokenStore, { get: unused, put: unused, remove: unused }), - Layer.succeed(Capabilities.PrimaryEnvironmentAuth, { - bearerToken: Effect.succeedSome("test-token"), - }), - Layer.succeed(Capabilities.SshEnvironmentGateway, { - provision: unused, - prepare: unused, - disconnect: unused, - }), - Layer.succeed(ConnectionProfileStore, { get: unused, put: unused, remove: unused }), - Layer.succeed(ConnectionCredentialStore, { get: unused, put: unused, remove: unused }), - ); - const layer = ConnectionResolver.layer.pipe( - Layer.provide(RemoteAuthorization.layer.pipe(Layer.provideMerge(dependencies))), - ); - return yield* Effect.gen(function* () { - const resolver = yield* ConnectionResolver.ConnectionResolver; - const samples = []; - for (const phase of ["cold", "warm"] as const) { - paths.length = 0; - observedHttpWaitSeconds = 0; - const started = performance.now(); - const prepared = yield* resolver.prepare({ - target, - profile: Option.none(), - enabled: true, - }); - expect(prepared.socketUrl).toContain("test-ticket"); - samples.push({ - phase, - seconds: (performance.now() - started) / 1_000, - observedHttpWaitSeconds, - paths: [...paths], - }); - } - return samples; - }).pipe(Effect.provide(layer)); - }), - ); - -const commandSample = (action: "startTurn" | "interruptTurn", historyDelayMs: number) => - Effect.scoped( - Effect.gen(function* () { - const historyStarted = yield* Deferred.make(); - let dispatchedAt = 0; - let observedHistoryWaitSeconds = 0; - let observedReplyWaitSeconds = 0; - const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ - target, - session: yield* SubscriptionRef.make( - Option.some({ - client: { - [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: Effect.fn( - "RemotePerformance.dispatch", - )(function* (command: OrchestrationV2Command) { - dispatchedAt = performance.now(); - expect(command.type).toBe( - action === "startTurn" ? "message.dispatch" : "run.interrupt", - ); - const started = performance.now(); - yield* Effect.sleep(50); - observedReplyWaitSeconds = (performance.now() - started) / 1_000; - return { sequence: 1 }; - }), - }, - } as unknown as RpcSession), - ), - } as EnvironmentSupervisor.EnvironmentSupervisor["Service"]); - const runtime = Atom.runtime( - Layer.mergeAll( - Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, { - run: (_id, effect) => - Effect.provideService( - effect, - EnvironmentSupervisor.EnvironmentSupervisor, - supervisor, - ), - } as EnvironmentRegistry.EnvironmentRegistry["Service"]), - Layer.succeed( - Crypto.Crypto, - Crypto.make({ - randomBytes: (size) => new Uint8Array(size), - digest: (_algorithm, data) => Effect.succeed(data), - }), - ), - Layer.succeed(ThreadHistoryController, { - register: unused, - unregister: Effect.fn("ThreadHistoryController.unregister")(() => Effect.void), - reconcilePendingWork: Effect.fn("ThreadHistoryController.reconcilePendingWork")( - () => Effect.void, - ), - loadEarlier: Effect.fn("RemotePerformance.loadEarlier")(function* () { - yield* Deferred.succeed(historyStarted, undefined); - if (historyDelayMs > 0) { - const started = performance.now(); - yield* Effect.sleep(historyDelayMs); - observedHistoryWaitSeconds = (performance.now() - started) / 1_000; - } - return { _tag: "loaded" } as const; - }), - }), - ), - ); - const registry = AtomRegistry.make(); - yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); - const commands = createThreadEnvironmentAtoms(runtime, () => Atom.make(null)); - const history = commands.loadEarlierHistory.run(registry, { - environmentId: target.environmentId, - input: { threadId: thread.thread.id }, - }); - yield* Deferred.await(historyStarted); - const started = performance.now(); - const result = yield* Effect.promise(() => - commands[action].run(registry, { - environmentId: target.environmentId, - input: { - threadId: thread.thread.id, - runId: RunId.make("running-run"), - message: { - messageId: MessageId.make("send"), - role: "user", - text: "Send", - attachments: [], - }, - runtimeMode: "full-access", - interactionMode: "default", - dispatchMode: "start", - }, - }), - ); - const completeSeconds = (performance.now() - started) / 1_000; - expect(result._tag).toBe("Success"); - expect(dispatchedAt).toBeGreaterThan(0); - expect((yield* Effect.promise(() => history))._tag).toBe("Success"); - return { - dispatchSeconds: (dispatchedAt - started) / 1_000, - completeSeconds, - observedHistoryWaitSeconds, - observedReplyWaitSeconds, - }; - }), - ); - -describe("controlled fixture latency in seconds, not phone or Wi-Fi measurements", () => { - for (const delayMs of [0, 100]) { - test(`prepare with ${delayMs}ms injected per HTTP response`, async ({ bench }) => { - const samples: Effect.Success>[] = []; - await bench("actual resolver cold and warm prepare", async () => { - samples.push(await Effect.runPromise(prepareSample(delayMs))); - }).run(controlledRunOptions); - Effect.runSync( - Effect.log( - JSON.stringify({ scenario: "prepare", injectedHttpSeconds: delayMs / 1_000, samples }), - ), - ); - }); - } - for (const action of ["startTurn", "interruptTurn"] as const) { - for (const historyDelayMs of [0, 250]) { - test(`${action} with ${historyDelayMs}ms injected history and 50ms command reply`, async ({ - bench, - }) => { - const samples: Effect.Success>[] = []; - await bench(`actual thread command scheduler ${action}`, async () => { - samples.push(await Effect.runPromise(commandSample(action, historyDelayMs))); - }).run(controlledRunOptions); - Effect.runSync( - Effect.log( - JSON.stringify({ - scenario: action, - injectedHistorySeconds: historyDelayMs / 1_000, - injectedCommandSeconds: 0.05, - samples, - }), - ), - ); - }); - } - } -}); - const delta: Extract = { id: EventId.make("delta"), type: "message.updated", diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index dfed5233d66f9..2b1b1162d6bc9 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -145,10 +145,7 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (optio }); describe("history loading", () => { - it.effect.each([ - ["startTurn", "message.dispatch"], - ["interruptTurn", "run.interrupt"], - ] as const)("dispatches %s while earlier history is pending", ([action, commandType]) => + it.effect("sends a message on the thread while earlier history is loading", () => Effect.gen(function* () { const historyStarted = yield* Deferred.make(); const historyReply = @@ -165,23 +162,22 @@ describe("history loading", () => { input: { threadId: THREAD_ID }, }); yield* Deferred.await(historyStarted); - const input = { - threadId: THREAD_ID, - runId: RunId.make("running-run"), - message: { - messageId: MessageId.make("message"), - role: "user", - text: "Send while history loads", - attachments: [], - }, - runtimeMode: "full-access", - interactionMode: "default", - dispatchMode: "start", - } as const; - const result = h.commands[action].run(h.registry, { + const send = h.commands.startTurn.run(h.registry, { environmentId: ENVIRONMENT_ID, - input, + input: { + threadId: THREAD_ID, + message: { + messageId: MessageId.make("message"), + role: "user", + text: "Send while history loads", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + dispatchMode: "start", + }, }); + // Dispatched after Send on another thread, so it arrives first only if Send is queued. const controlThreadId = ThreadId.make("control-thread"); const control = h.commands.interruptTurn.run(h.registry, { environmentId: ENVIRONMENT_ID, @@ -192,9 +188,9 @@ describe("history loading", () => { yield* Deferred.succeed(historyReply, { _tag: "loaded" }); const second = yield* Queue.take(h.requests); yield* Deferred.succeed(second.reply, { sequence: 2 }); - const results = yield* Effect.promise(() => Promise.all([history, result, control])); + const results = yield* Effect.promise(() => Promise.all([history, send, control])); - expect(first.command).toMatchObject({ type: commandType, threadId: THREAD_ID }); + expect(first.command).toMatchObject({ type: "message.dispatch", threadId: THREAD_ID }); expect(second.command).toMatchObject({ type: "run.interrupt", threadId: controlThreadId }); expect(results.map((value) => value._tag)).toEqual(["Success", "Success", "Success"]); }), From 6fb0a78318954c836fbd6c9e65931fda06d24cdc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 22:39:13 +0000 Subject: [PATCH 6/6] test(client): identity mismatch sends no websocket ticket Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/authorization/layer.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index 6237f38dbd739..f7d8665549e7f 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -328,7 +328,6 @@ describe("RemoteEnvironmentAuthorization", () => { environmentId: "another-environment", orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, }), - websocketTicket("second-ticket"), ], }); @@ -339,6 +338,11 @@ describe("RemoteEnvironmentAuthorization", () => { _tag: "Failure", failure: { _tag: "ConnectionBlockedError", reason: "configuration" }, }); + expect( + harness.fetch.calls.filter(([url]) => + String(url).endsWith("/api/auth/websocket-ticket/rpc"), + ), + ).toHaveLength(1); }), ); it.effect("validates a fresh bearer descriptor while issuing fresh websocket tickets", () =>