authorgravatar for 62337003+Bil0000@users.noreply.github.comBilal Bakr <62337003+Bil0000@users.noreply.github.com> 2026-09-11 23:20:54+03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-11 13:20:54-07:00
log6e8931d75615ec020e602e97df2694b0541a6272
tree447599d2ea45d9c4ad366076cae533a89bda3292
parent7bd7f99e6cab940892ff9240549c507bfd066aee
signature Signed by PGP key B5690EEEBB952194

perf(client): reduce remote request and message sync overhead (#11029)

12 files changed, 296 insertions(+), 47 deletions(-)

packages/client-runtime/src/authorization/remote.ts+13-13
...@@ -11,7 +11,7 @@ import * as Effect from "effect/Effect";...@@ -11,7 +11,7 @@ import * as Effect from "effect/Effect";
11import { environmentEndpointUrl } from "../environment/endpoint.ts";11import { environmentEndpointUrl } from "../environment/endpoint.ts";
12import {12import {
13 executeEnvironmentHttpRequest,13 executeEnvironmentHttpRequest,
14 makeEnvironmentHttpApiClient,14 makeEnvironmentHttpApiGroupClient,
15 type RemoteEnvironmentRequestError,15 type RemoteEnvironmentRequestError,
16} from "../rpc/http.ts";16} from "../rpc/http.ts";
1717
...@@ -93,11 +93,11 @@ export const exchangeRemoteDpopAccessToken = Effect.fn(...@@ -93,11 +93,11 @@ export const exchangeRemoteDpopAccessToken = Effect.fn(
93 readonly dpopProof: string;93 readonly dpopProof: string;
94 readonly timeoutMs?: number;94 readonly timeoutMs?: number;
95}) {95}) {
96 const client = yield* makeEnvironmentHttpApiClient(input.httpBaseUrl);96 const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth");
97 const response = yield* executeEnvironmentHttpRequest(97 const response = yield* executeEnvironmentHttpRequest(
98 environmentEndpointUrl(input.httpBaseUrl, "/oauth/token"),98 environmentEndpointUrl(input.httpBaseUrl, "/oauth/token"),
99 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,99 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,
100 client.auth.token({100 client.token({
101 headers: { dpop: input.dpopProof },101 headers: { dpop: input.dpopProof },
102 payload: {102 payload: {
103 grant_type: AuthTokenExchangeGrantType,103 grant_type: AuthTokenExchangeGrantType,
...@@ -121,11 +121,11 @@ export const bootstrapRemoteBearerSession = Effect.fn(...@@ -121,11 +121,11 @@ export const bootstrapRemoteBearerSession = Effect.fn(
121 readonly clientMetadata?: AuthClientPresentationMetadata;121 readonly clientMetadata?: AuthClientPresentationMetadata;
122 readonly timeoutMs?: number;122 readonly timeoutMs?: number;
123}) {123}) {
124 const client = yield* makeEnvironmentHttpApiClient(input.httpBaseUrl);124 const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth");
125 return yield* executeEnvironmentHttpRequest(125 return yield* executeEnvironmentHttpRequest(
126 environmentEndpointUrl(input.httpBaseUrl, "/oauth/token"),126 environmentEndpointUrl(input.httpBaseUrl, "/oauth/token"),
127 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,127 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,
128 client.auth.token({128 client.token({
129 headers: {},129 headers: {},
130 payload: {130 payload: {
131 grant_type: AuthTokenExchangeGrantType,131 grant_type: AuthTokenExchangeGrantType,
...@@ -146,11 +146,11 @@ export const fetchRemoteSessionState = Effect.fn(...@@ -146,11 +146,11 @@ export const fetchRemoteSessionState = Effect.fn(
146 readonly bearerToken: string;146 readonly bearerToken: string;
147 readonly timeoutMs?: number;147 readonly timeoutMs?: number;
148}) {148}) {
149 const client = yield* makeEnvironmentHttpApiClient(input.httpBaseUrl);149 const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth");
150 return yield* executeEnvironmentHttpRequest(150 return yield* executeEnvironmentHttpRequest(
151 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/session"),151 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/session"),
152 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,152 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,
153 client.auth.session({153 client.session({
154 headers: {154 headers: {
155 authorization: `Bearer ${input.bearerToken}`,155 authorization: `Bearer ${input.bearerToken}`,
156 },156 },
...@@ -166,11 +166,11 @@ export const fetchRemoteDpopSessionState = Effect.fn(...@@ -166,11 +166,11 @@ export const fetchRemoteDpopSessionState = Effect.fn(
166 readonly dpopProof: string;166 readonly dpopProof: string;
167 readonly timeoutMs?: number;167 readonly timeoutMs?: number;
168}) {168}) {
169 const client = yield* makeEnvironmentHttpApiClient(input.httpBaseUrl);169 const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth");
170 return yield* executeEnvironmentHttpRequest(170 return yield* executeEnvironmentHttpRequest(
171 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/session"),171 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/session"),
172 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,172 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,
173 client.auth.session({173 client.session({
174 headers: {174 headers: {
175 authorization: `DPoP ${input.accessToken}`,175 authorization: `DPoP ${input.accessToken}`,
176 dpop: input.dpopProof,176 dpop: input.dpopProof,
...@@ -186,11 +186,11 @@ export const issueRemoteWebSocketTicket = Effect.fn(...@@ -186,11 +186,11 @@ export const issueRemoteWebSocketTicket = Effect.fn(
186 readonly bearerToken: string;186 readonly bearerToken: string;
187 readonly timeoutMs?: number;187 readonly timeoutMs?: number;
188}) {188}) {
189 const client = yield* makeEnvironmentHttpApiClient(input.httpBaseUrl);189 const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth");
190 return yield* executeEnvironmentHttpRequest(190 return yield* executeEnvironmentHttpRequest(
191 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/websocket-ticket"),191 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/websocket-ticket"),
192 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,192 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,
193 client.auth.webSocketTicket({193 client.webSocketTicket({
194 headers: {194 headers: {
195 authorization: `Bearer ${input.bearerToken}`,195 authorization: `Bearer ${input.bearerToken}`,
196 },196 },
...@@ -206,11 +206,11 @@ export const issueRemoteDpopWebSocketTicket = Effect.fn(...@@ -206,11 +206,11 @@ export const issueRemoteDpopWebSocketTicket = Effect.fn(
206 readonly dpopProof: string;206 readonly dpopProof: string;
207 readonly timeoutMs?: number;207 readonly timeoutMs?: number;
208}) {208}) {
209 const client = yield* makeEnvironmentHttpApiClient(input.httpBaseUrl);209 const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth");
210 return yield* executeEnvironmentHttpRequest(210 return yield* executeEnvironmentHttpRequest(
211 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/websocket-ticket"),211 environmentEndpointUrl(input.httpBaseUrl, "/api/auth/websocket-ticket"),
212 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,212 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,
213 client.auth.webSocketTicket({213 client.webSocketTicket({
214 headers: {214 headers: {
215 authorization: `DPoP ${input.accessToken}`,215 authorization: `DPoP ${input.accessToken}`,
216 dpop: input.dpopProof,216 dpop: input.dpopProof,
packages/client-runtime/src/environment/descriptor.ts+3-3
...@@ -1,17 +1,17 @@...@@ -1,17 +1,17 @@
1import * as Effect from "effect/Effect";1import * as Effect from "effect/Effect";
22
3import { environmentEndpointUrl } from "./endpoint.ts";3import { environmentEndpointUrl } from "./endpoint.ts";
4import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient } from "../rpc/http.ts";4import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiGroupClient } from "../rpc/http.ts";
55
6const DEFAULT_REMOTE_REQUEST_TIMEOUT_MS = 10_000;6const DEFAULT_REMOTE_REQUEST_TIMEOUT_MS = 10_000;
77
8export const fetchRemoteEnvironmentDescriptor = Effect.fn(8export const fetchRemoteEnvironmentDescriptor = Effect.fn(
9 "clientRuntime.environment.fetchRemoteEnvironmentDescriptor",9 "clientRuntime.environment.fetchRemoteEnvironmentDescriptor",
10)(function* (input: { readonly httpBaseUrl: string; readonly timeoutMs?: number }) {10)(function* (input: { readonly httpBaseUrl: string; readonly timeoutMs?: number }) {
11 const client = yield* makeEnvironmentHttpApiClient(input.httpBaseUrl);11 const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "metadata");
12 return yield* executeEnvironmentHttpRequest(12 return yield* executeEnvironmentHttpRequest(
13 environmentEndpointUrl(input.httpBaseUrl, "/.well-known/t3/environment"),13 environmentEndpointUrl(input.httpBaseUrl, "/.well-known/t3/environment"),
14 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,14 input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS,
15 client.metadata.descriptor(),15 client.descriptor(),
16 );16 );
17});17});
packages/client-runtime/src/remotePerformance.bench.ts created+165
...@@ -0,0 +1,165 @@
1import {
2 EnvironmentId,
3 EventId,
4 MessageId,
5 ProjectId,
6 ProviderInstanceId,
7 ThreadId,
8 type OrchestrationEvent,
9 type OrchestrationThread,
10} from "@t3tools/contracts";
11import * as Effect from "effect/Effect";
12import * as Option from "effect/Option";
13import { HttpClient, HttpClientResponse } from "effect/unstable/http";
14import { bench, describe } from "vite-plus/test";
15
16import { issueRemoteWebSocketTicket } from "./authorization/remote.ts";
17import { PrimaryConnectionTarget } from "./connection/model.ts";
18import { fetchRemoteEnvironmentDescriptor } from "./environment/descriptor.ts";
19import type { RemoteEnvironmentRequestError } from "./rpc/http.ts";
20import { fetchEnvironmentThreadSnapshot } from "./state/threadSnapshotHttp.ts";
21import { applyThreadDetailEvent } from "./state/threadReducer.ts";
22
23const timestamp = "2026-09-01T00:00:00.000Z";
24const thread: OrchestrationThread = {
25 id: ThreadId.make("thread-1"),
26 projectId: ProjectId.make("project-1"),
27 title: "Remote thread",
28 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
29 runtimeMode: "full-access",
30 interactionMode: "default",
31 branch: null,
32 worktreePath: null,
33 latestTurn: null,
34 createdAt: timestamp,
35 updatedAt: timestamp,
36 archivedAt: null,
37 settledOverride: null,
38 settledAt: null,
39 deletedAt: null,
40 pullRequests: [],
41 messages: Array.from({ length: 100 }, (_, index) => ({
42 id: MessageId.make(`message-${index}`),
43 role: "assistant",
44 text: "Message text. ".repeat(40),
45 turnId: null,
46 streaming: false,
47 createdAt: timestamp,
48 updatedAt: timestamp,
49 })),
50 proposedPlans: [],
51 activities: [],
52 checkpoints: [],
53 session: null,
54};
55const target = new PrimaryConnectionTarget({
56 environmentId: EnvironmentId.make("remote-1"),
57 label: "Remote",
58 httpBaseUrl: "https://remote.example.test",
59 wsBaseUrl: "wss://remote.example.test/ws",
60});
61const responses = {
62 "/.well-known/t3/environment": {
63 environmentId: target.environmentId,
64 label: target.label,
65 platform: { os: "linux", arch: "x64" },
66 serverVersion: "0.0.0-test",
67 capabilities: { repositoryIdentity: true },
68 },
69 "/api/auth/websocket-ticket": { ticket: "test-ticket", expiresAt: timestamp },
70 "/api/orchestration/threads/thread-1": { snapshotSequence: 1, thread },
71};
72const httpClient = HttpClient.make((request) =>
73 Effect.sync(() => {
74 const path = new URL(request.url).pathname as keyof typeof responses;
75 return HttpClientResponse.fromWeb(request, Response.json(responses[path]));
76 }),
77);
78const requests: Record<
79 string,
80 Effect.Effect<unknown, RemoteEnvironmentRequestError, HttpClient.HttpClient>
81> = {
82 "read remote connection descriptor": fetchRemoteEnvironmentDescriptor({
83 httpBaseUrl: target.httpBaseUrl,
84 }),
85 "issue remote WebSocket ticket": issueRemoteWebSocketTicket({
86 httpBaseUrl: target.httpBaseUrl,
87 bearerToken: "test-token",
88 }),
89 "load remote snapshot with 100 messages": fetchEnvironmentThreadSnapshot({
90 prepared: {
91 environmentId: target.environmentId,
92 label: target.label,
93 httpBaseUrl: target.httpBaseUrl,
94 socketUrl: target.wsBaseUrl,
95 httpAuthorization: null,
96 target,
97 },
98 threadId: thread.id,
99 signer: Option.none(),
100 }),
101};
102
103describe("remote HTTP processing with an in-memory transport", () => {
104 for (const [name, request] of Object.entries(requests)) {
105 bench(
106 name,
107 async () => {
108 await Effect.runPromise(
109 request.pipe(Effect.provideService(HttpClient.HttpClient, httpClient)),
110 );
111 },
112 { warmupTime: 1_000, time: 1_500 },
113 );
114 }
115});
116
117const delta: OrchestrationEvent = {
118 eventId: EventId.make("delta"),
119 sequence: 2,
120 aggregateKind: "thread",
121 aggregateId: thread.id,
122 occurredAt: timestamp,
123 commandId: null,
124 causationEventId: null,
125 correlationId: null,
126 metadata: {},
127 type: "thread.message-sent",
128 payload: {
129 threadId: thread.id,
130 messageId: MessageId.make("message-99"),
131 role: "assistant",
132 text: " next",
133 turnId: null,
134 streaming: true,
135 createdAt: timestamp,
136 updatedAt: timestamp,
137 },
138};
139
140describe("remote message replay", () => {
141 for (const count of [100, 1_000]) {
142 const loaded = {
143 ...thread,
144 messages: Array.from({ length: count }, (_, index) => ({
145 ...thread.messages[0]!,
146 id: MessageId.make(`message-${index}`),
147 })),
148 };
149 const event = {
150 ...delta,
151 payload: { ...delta.payload, messageId: loaded.messages.at(-1)!.id },
152 };
153 bench(
154 `apply 200 text deltas to ${count} loaded messages`,
155 () => {
156 let current: OrchestrationThread = loaded;
157 for (let index = 0; index < 200; index += 1) {
158 const result = applyThreadDetailEvent(current, event);
159 if (result.kind === "updated") current = result.thread;
160 }
161 },
162 { warmupTime: 1_000, time: 1_500 },
163 );
164 }
165});
packages/client-runtime/src/rpc/http.ts+14
...@@ -99,6 +99,20 @@ export const makeEnvironmentHttpApiClient = (httpBaseUrl: string) =>...@@ -99,6 +99,20 @@ export const makeEnvironmentHttpApiClient = (httpBaseUrl: string) =>
99 baseUrl: remoteApiBaseUrl(httpBaseUrl),99 baseUrl: remoteApiBaseUrl(httpBaseUrl),
100 });100 });
101101
102export const makeEnvironmentHttpApiGroupClient = <
103 Group extends keyof typeof EnvironmentHttpApi.groups,
104>(
105 httpBaseUrl: string,
106 group: Group,
107) =>
108 Effect.flatMap(HttpClient.HttpClient, (httpClient) =>
109 HttpApiClient.group(EnvironmentHttpApi, {
110 httpClient,
111 group,
112 baseUrl: remoteApiBaseUrl(httpBaseUrl),
113 }),
114 );
115
102/** Contract-derived request URLs for authentication proofs, tracing, and structured errors. */116/** Contract-derived request URLs for authentication proofs, tracing, and structured errors. */
103export const makeEnvironmentHttpApiUrlBuilder = (httpBaseUrl: string) =>117export const makeEnvironmentHttpApiUrlBuilder = (httpBaseUrl: string) =>
104 HttpApiClient.urlBuilder(EnvironmentHttpApi, {118 HttpApiClient.urlBuilder(EnvironmentHttpApi, {
packages/client-runtime/src/state/environmentHttpAuth.test.ts+11
...@@ -215,6 +215,17 @@ const LOADERS: ReadonlyArray<{...@@ -215,6 +215,17 @@ const LOADERS: ReadonlyArray<{
215];215];
216216
217describe("authenticated environment HTTP requests", () => {217describe("authenticated environment HTTP requests", () => {
218 it.effect.each(LOADERS)("rejects an invalid $name response", (loader) =>
219 Effect.gen(function* () {
220 const harness = makeHarness(() => Response.json({}));
221 const result = yield* loader
222 .load(harness.input)
223 .pipe(Effect.provide(harness.httpLayer), Effect.asVoid, Effect.flip);
224 expect(result._tag).toBe("RemoteEnvironmentAuthInvalidJsonError");
225 expect(harness.calls).toHaveLength(1);
226 }),
227 );
228
218 it.effect.each(LOADERS)("uses current relay authorization and endpoint for $name", (loader) =>229 it.effect.each(LOADERS)("uses current relay authorization and endpoint for $name", (loader) =>
219 Effect.gen(function* () {230 Effect.gen(function* () {
220 const harness = makeHarness(() => Response.json(loader.response));231 const harness = makeHarness(() => Response.json(loader.response));
packages/client-runtime/src/state/environmentHttpAuth.ts+16-6
...@@ -1,14 +1,14 @@...@@ -1,14 +1,14 @@
1import * as Effect from "effect/Effect";1import * as Effect from "effect/Effect";
2import * as Option from "effect/Option";2import * as Option from "effect/Option";
3import * as Result from "effect/Result";3import * as Result from "effect/Result";
4import { FetchHttpClient, type HttpClient, type HttpMethod } from "effect/unstable/http";4import { FetchHttpClient, type HttpMethod } from "effect/unstable/http";
55
6import type { RemoteEnvironmentAuthorization } from "../authorization/service.ts";6import type { RemoteEnvironmentAuthorization } from "../authorization/service.ts";
7import type { PreparedConnection, PreparedHttpAuthorization } from "../connection/model.ts";7import type { PreparedConnection, PreparedHttpAuthorization } from "../connection/model.ts";
8import type { ManagedRelayDpopSigner } from "../relay/managedRelay.ts";8import type { ManagedRelayDpopSigner } from "../relay/managedRelay.ts";
9import {9import {
10 executeEnvironmentHttpRequest,10 executeEnvironmentHttpRequest,
11 makeEnvironmentHttpApiClient,11 makeEnvironmentHttpApiGroupClient,
12 RemoteEnvironmentAuthFetchError,12 RemoteEnvironmentAuthFetchError,
13 RemoteEnvironmentAuthTimeoutError,13 RemoteEnvironmentAuthTimeoutError,
14 type RemoteEnvironmentRequestError,14 type RemoteEnvironmentRequestError,
...@@ -86,20 +86,30 @@ const buildEnvironmentAuthHeaders = (...@@ -86,20 +86,30 @@ const buildEnvironmentAuthHeaders = (
86 */86 */
87export const executeAuthenticatedEnvironmentHttpRequest = Effect.fn(87export const executeAuthenticatedEnvironmentHttpRequest = Effect.fn(
88 "clientRuntime.state.executeAuthenticatedEnvironmentHttpRequest",88 "clientRuntime.state.executeAuthenticatedEnvironmentHttpRequest",
89)(function* <A, E, R>(input: {89)(function* <
90 Group extends Parameters<typeof makeEnvironmentHttpApiGroupClient>[1],
91 A,
92 E,
93 R,
94>(input: {
90 readonly prepared: PreparedConnection;95 readonly prepared: PreparedConnection;
91 readonly signer: Option.Option<ManagedRelayDpopSigner["Service"]>;96 readonly signer: Option.Option<ManagedRelayDpopSigner["Service"]>;
92 readonly remoteAuthorization?: Option.Option<RemoteEnvironmentAuthorization["Service"]>;97 readonly remoteAuthorization?: Option.Option<RemoteEnvironmentAuthorization["Service"]>;
93 readonly method: HttpMethod.HttpMethod;98 readonly method: HttpMethod.HttpMethod;
94 readonly url: (httpBaseUrl: string) => string;99 readonly url: (httpBaseUrl: string) => string;
95 readonly timeoutMs: number;100 readonly timeoutMs: number;
101 readonly group: Group;
96 readonly request: (input: {102 readonly request: (input: {
97 readonly client: Effect.Success<ReturnType<typeof makeEnvironmentHttpApiClient>>;103 readonly client: Effect.Success<ReturnType<typeof makeEnvironmentHttpApiGroupClient<Group>>>;
98 readonly headers: EnvironmentHttpAuthHeaders;104 readonly headers: EnvironmentHttpAuthHeaders;
99 }) => Effect.Effect<A, E, R>;105 }) => Effect.Effect<A, E, R>;
100 /** Some endpoints report rejected credentials in a successful response. */106 /** Some endpoints report rejected credentials in a successful response. */
101 readonly isUnauthorizedResponse?: (response: NoInfer<A>) => boolean;107 readonly isUnauthorizedResponse?: (response: NoInfer<A>) => boolean;
102}): Effect.fn.Return<A, RemoteEnvironmentRequestError, HttpClient.HttpClient | R> {108}): Effect.fn.Return<
109 A,
110 RemoteEnvironmentRequestError,
111 Effect.Services<ReturnType<typeof makeEnvironmentHttpApiGroupClient<Group>>> | R
112> {
103 let httpBaseUrl = input.prepared.httpBaseUrl;113 let httpBaseUrl = input.prepared.httpBaseUrl;
104 return yield* Effect.gen(function* () {114 return yield* Effect.gen(function* () {
105 let rejectedAccessToken: string | undefined;115 let rejectedAccessToken: string | undefined;
...@@ -132,7 +142,7 @@ export const executeAuthenticatedEnvironmentHttpRequest = Effect.fn(...@@ -132,7 +142,7 @@ export const executeAuthenticatedEnvironmentHttpRequest = Effect.fn(
132 }142 }
133143
134 const requestUrl = input.url(httpBaseUrl);144 const requestUrl = input.url(httpBaseUrl);
135 const client = yield* makeEnvironmentHttpApiClient(httpBaseUrl);145 const client = yield* makeEnvironmentHttpApiGroupClient(httpBaseUrl, input.group);
136 const headers = yield* buildEnvironmentAuthHeaders(146 const headers = yield* buildEnvironmentAuthHeaders(
137 authorization,147 authorization,
138 input.method,148 input.method,
packages/client-runtime/src/state/pullRequestDiffHttp.ts+2-1
...@@ -50,10 +50,11 @@ export const fetchEnvironmentPullRequestDiff = Effect.fn(...@@ -50,10 +50,11 @@ export const fetchEnvironmentPullRequestDiff = Effect.fn(
50}) {50}) {
51 return yield* executeAuthenticatedEnvironmentHttpRequest({51 return yield* executeAuthenticatedEnvironmentHttpRequest({
52 ...input,52 ...input,
53 group: "pullRequests",
53 method: "POST",54 method: "POST",
54 url: (httpBaseUrl) => makeEnvironmentHttpApiUrlBuilder(httpBaseUrl).pullRequests.diff(),55 url: (httpBaseUrl) => makeEnvironmentHttpApiUrlBuilder(httpBaseUrl).pullRequests.diff(),
55 timeoutMs: input.timeoutMs ?? DEFAULT_PULL_REQUEST_DIFF_TIMEOUT_MS,56 timeoutMs: input.timeoutMs ?? DEFAULT_PULL_REQUEST_DIFF_TIMEOUT_MS,
56 request: ({ client, headers }) => client.pullRequests.diff({ payload: input.diff, headers }),57 request: ({ client, headers }) => client.diff({ payload: input.diff, headers }),
57 }).pipe(58 }).pipe(
58 Effect.mapError((error) =>59 Effect.mapError((error) =>
59 error._tag === "EnvironmentAuthInvalidError" && error.reason === "invalid_credential"60 error._tag === "EnvironmentAuthInvalidError" && error.reason === "invalid_credential"
packages/client-runtime/src/state/session.ts+2-1
...@@ -49,10 +49,11 @@ export const fetchEnvironmentSessionState = Effect.fn(...@@ -49,10 +49,11 @@ export const fetchEnvironmentSessionState = Effect.fn(
49}) {49}) {
50 return yield* executeAuthenticatedEnvironmentHttpRequest({50 return yield* executeAuthenticatedEnvironmentHttpRequest({
51 ...input,51 ...input,
52 group: "auth",
52 method: "GET",53 method: "GET",
53 url: (httpBaseUrl) => environmentEndpointUrl(httpBaseUrl, "/api/auth/session"),54 url: (httpBaseUrl) => environmentEndpointUrl(httpBaseUrl, "/api/auth/session"),
54 timeoutMs: input.timeoutMs ?? DEFAULT_SESSION_STATE_TIMEOUT_MS,55 timeoutMs: input.timeoutMs ?? DEFAULT_SESSION_STATE_TIMEOUT_MS,
55 request: ({ client, headers }) => client.auth.session({ headers }),56 request: ({ client, headers }) => client.session({ headers }),
56 // This endpoint returns 200 with authenticated:false for expired credentials.57 // This endpoint returns 200 with authenticated:false for expired credentials.
57 isUnauthorizedResponse: (response) => !response.authenticated,58 isUnauthorizedResponse: (response) => !response.authenticated,
58 });59 });
packages/client-runtime/src/state/shellSnapshotHttp.ts+2-1
...@@ -32,10 +32,11 @@ export const fetchEnvironmentShellSnapshot = Effect.fn(...@@ -32,10 +32,11 @@ export const fetchEnvironmentShellSnapshot = Effect.fn(
32}) {32}) {
33 return yield* executeAuthenticatedEnvironmentHttpRequest({33 return yield* executeAuthenticatedEnvironmentHttpRequest({
34 ...input,34 ...input,
35 group: "orchestration",
35 method: "GET",36 method: "GET",
36 url: (httpBaseUrl) => environmentEndpointUrl(httpBaseUrl, "/api/orchestration/shell"),37 url: (httpBaseUrl) => environmentEndpointUrl(httpBaseUrl, "/api/orchestration/shell"),
37 timeoutMs: input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS,38 timeoutMs: input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS,
38 request: ({ client, headers }) => client.orchestration.shellSnapshot({ headers }),39 request: ({ client, headers }) => client.shellSnapshot({ headers }),
39 });40 });
40});41});
4142
packages/client-runtime/src/state/threadReducer.test.ts+48
...@@ -526,6 +526,54 @@ describe("applyThreadDetailEvent", () => {...@@ -526,6 +526,54 @@ describe("applyThreadDetailEvent", () => {
526 });526 });
527527
528 describe("thread.message-sent", () => {528 describe("thread.message-sent", () => {
529 it.each([
530 ["first", ["first+", "middle", "last"]],
531 ["middle", ["first", "middle+", "last"]],
532 ["last", ["first", "middle", "last+"]],
533 ["new", ["first", "middle", "last", "+"]],
534 ] as const)("applies a delta to %s without changing other messages", (id, texts) => {
535 const messages = Object.freeze(
536 ["first", "middle", "last"].map((name) =>
537 Object.freeze({
538 id: MessageId.make(name),
539 role: "assistant" as const,
540 text: name,
541 turnId: null,
542 streaming: false,
543 createdAt: baseThread.createdAt,
544 updatedAt: baseThread.updatedAt,
545 }),
546 ),
547 );
548 const result = applyThreadDetailEvent(
549 { ...baseThread, messages },
550 {
551 ...baseEventFields,
552 sequence: 6,
553 occurredAt: baseThread.updatedAt,
554 aggregateKind: "thread",
555 aggregateId: baseThread.id,
556 type: "thread.message-sent",
557 payload: {
558 threadId: baseThread.id,
559 messageId: MessageId.make(id),
560 role: "assistant",
561 text: "+",
562 turnId: null,
563 streaming: true,
564 createdAt: baseThread.createdAt,
565 updatedAt: baseThread.updatedAt,
566 },
567 },
568 );
569 expect(result.kind).toBe("updated");
570 if (result.kind !== "updated") return;
571 expect(result.thread.messages.map((message) => message.text)).toEqual(texts);
572 for (const [index, message] of messages.entries()) {
573 if (message.id !== id) expect(result.thread.messages[index]).toBe(message);
574 }
575 });
576
529 it("appends a new message", () => {577 it("appends a new message", () => {
530 const result = applyThreadDetailEvent(baseThread, {578 const result = applyThreadDetailEvent(baseThread, {
531 ...baseEventFields,579 ...baseEventFields,
packages/client-runtime/src/state/threadReducer.ts+18-21
...@@ -386,27 +386,24 @@ export function applyThreadDetailEvent(...@@ -386,27 +386,24 @@ export function applyThreadDetailEvent(
386 updatedAt: event.payload.updatedAt,386 updatedAt: event.payload.updatedAt,
387 };387 };
388388
389 const existingMessage = thread.messages.find((entry) => entry.id === message.id);389 let found = false;
390 const messages = existingMessage390 const messages = thread.messages.map((entry) => {
391 ? Arr.map(thread.messages, (entry) =>391 if (entry.id !== message.id) return entry;
392 entry.id !== message.id392 found = true;
393 ? entry393 return {
394 : {394 ...entry,
395 ...entry,395 text: message.streaming
396 text: message.streaming396 ? `${entry.text}${message.text}`
397 ? `${entry.text}${message.text}`397 : message.text.length > 0
398 : message.text.length > 0398 ? message.text
399 ? message.text399 : entry.text,
400 : entry.text,400 streaming: message.streaming,
401 streaming: message.streaming,401 ...(message.turnId !== undefined ? { turnId: message.turnId } : {}),
402 ...(message.turnId !== undefined ? { turnId: message.turnId } : {}),402 ...(message.streaming ? {} : { updatedAt: message.updatedAt }),
403 ...(message.streaming ? {} : { updatedAt: message.updatedAt }),403 ...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
404 ...(message.attachments !== undefined404 };
405 ? { attachments: message.attachments }405 });
406 : {}),406 if (!found) messages.push(message);
407 },
408 )
409 : Arr.append(thread.messages, message);
410 // Update latestTurn for assistant messages bound to a turn. A completed407 // Update latestTurn for assistant messages bound to a turn. A completed
411 // assistant message only settles the turn once the session is no longer408 // assistant message only settles the turn once the session is no longer
412 // running it — providers may emit several assistant messages per turn409 // running it — providers may emit several assistant messages per turn
packages/client-runtime/src/state/threadSnapshotHttp.ts+2-1
...@@ -45,12 +45,13 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn(...@@ -45,12 +45,13 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn(
45}) {45}) {
46 return yield* executeAuthenticatedEnvironmentHttpRequest({46 return yield* executeAuthenticatedEnvironmentHttpRequest({
47 ...input,47 ...input,
48 group: "orchestration",
48 method: "GET",49 method: "GET",
49 url: (httpBaseUrl) =>50 url: (httpBaseUrl) =>
50 environmentEndpointUrl(httpBaseUrl, `/api/orchestration/threads/${input.threadId}`),51 environmentEndpointUrl(httpBaseUrl, `/api/orchestration/threads/${input.threadId}`),
51 timeoutMs: input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS,52 timeoutMs: input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS,
52 request: ({ client, headers }) =>53 request: ({ client, headers }) =>
53 client.orchestration.threadSnapshot({54 client.threadSnapshot({
54 params: { threadId: input.threadId },55 params: { threadId: input.threadId },
55 payload: {56 payload: {
56 ...(input.window !== undefined ? { turnLimit: input.window.turnLimit } : {}),57 ...(input.window !== undefined ? { turnLimit: input.window.turnLimit } : {}),