From a9d2a8009d189a52e144abf044020e5bfdd8f8cc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 09:13:24 +0000 Subject: [PATCH] perf(client): resume completed shells and deduplicate repeated links Co-authored-by: maria-rcks --- .../src/orchestration-v2/ShellStream.ts | 19 + apps/server/src/orchestration-v2/http.test.ts | 191 +++++++--- apps/server/src/orchestration-v2/http.ts | 15 +- .../src/state/shell-sync.test.ts | 337 ++++++++++++++---- packages/client-runtime/src/state/shell.ts | 55 +-- .../src/state/shellPullRequests.ts | 30 +- .../src/state/shellSnapshotHttp.ts | 3 +- packages/contracts/src/environmentHttp.ts | 1 + packages/contracts/src/orchestrationV2.ts | 2 + 9 files changed, 499 insertions(+), 154 deletions(-) diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index c5a80e70281b1..fcdfc9be14856 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -13,6 +13,7 @@ import { OrchestrationV2ThreadShell as ThreadShellSchema, } from "@t3tools/contracts"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import { stableStringify } from "@t3tools/shared/relaySigning"; import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; @@ -33,6 +34,24 @@ export function buildActiveShellSnapshot(input: { }; } +export function deduplicateShellPullRequests( + snapshot: OrchestrationV2ShellSnapshot, +): OrchestrationV2ShellSnapshot { + const firstRowByLinks = new Map(); + const threads = snapshot.threads.map((thread, index) => { + if (thread.pullRequests === undefined || thread.pullRequests.length === 0) return thread; + const key = stableStringify(thread.pullRequests); + const reference = firstRowByLinks.get(key); + if (reference === undefined) { + firstRowByLinks.set(key, index); + return thread; + } + const { pullRequests: _links, ...row } = thread; + return { ...row, pullRequestReference: reference }; + }); + return { ...snapshot, threads }; +} + /** * One transaction covers the reads so they agree; the threads are decoded after it commits, * because the server shares one SQLite connection and decoding a large shell takes longer than diff --git a/apps/server/src/orchestration-v2/http.test.ts b/apps/server/src/orchestration-v2/http.test.ts index 74476d658ed0c..27c7a218a5384 100644 --- a/apps/server/src/orchestration-v2/http.test.ts +++ b/apps/server/src/orchestration-v2/http.test.ts @@ -9,6 +9,9 @@ import { ORCHESTRATION_PROTOCOL_HEADER, ORCHESTRATION_PROTOCOL_VERSION_TEXT, OrchestrationV2AppThread, + OrchestrationV2ShellSnapshot, + OrchestrationV2ShellSnapshotJson, + type ThreadPullRequestLink, OrchestrationV2TurnItem, OrchestrationV2ThreadTranscript, PROVIDER_SEND_TURN_MAX_FILE_BYTES, @@ -41,13 +44,21 @@ import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; -class TranscriptTestApi extends HttpApi.make("environment").add( +class OrchestrationTestApi extends HttpApi.make("environment").add( EnvironmentHttpApi.groups.orchestration, ) {} const decodeTranscript = Schema.decodeUnknownEffect( Schema.toCodecJson(OrchestrationV2ThreadTranscript), ); +const decodeShellSnapshot = Schema.decodeUnknownEffect( + Schema.toCodecJson(OrchestrationV2ShellSnapshot), +); + +const decodeShellSnapshotWire = Schema.decodeUnknownEffect( + Schema.toEncoded(OrchestrationV2ShellSnapshotJson), +); + const now = DateTime.makeUnsafe("2026-09-29T00:00:00.000Z"); const environmentId = EnvironmentId.make("source-بيئة"); const output = "Full tool output. ".repeat(4_000); @@ -96,27 +107,69 @@ const items = Schema.decodeUnknownSync(Schema.Array(OrchestrationV2TurnItem))( })), ); -it.effect("exports full history with size, scope, and missing-thread checks", () => - Effect.gen(function* () { - let allowed = true; - const runtime = ProviderReplayHarness.layerWithRegistry( - { name: "thread-transcript" }, - ProviderAdapterRegistry.layerFromAdapters([]), - { runEffectWorker: false }, - ); - const services = yield* Layer.build( - Layer.mergeAll( - runtime, - ThreadManagementService.layer.pipe( - Layer.provide( - Layer.mergeAll(runtime, CommandReceiptStore.layer, ServerSettings.layerTest({})), +const makeHttpHarness = Effect.fnUntraced(function* (allowed = () => true) { + const runtime = ProviderReplayHarness.layerWithRegistry( + { name: "orchestration-http" }, + ProviderAdapterRegistry.layerFromAdapters([]), + { runEffectWorker: false }, + ); + const services = yield* Layer.build( + Layer.mergeAll( + runtime, + ThreadManagementService.layer.pipe( + Layer.provide( + Layer.mergeAll(runtime, CommandReceiptStore.layer, ServerSettings.layerTest({})), + ), + ), + OrchestrationEventStore.layer, + ProjectStore.layer, + ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), + ); + const eventSink = Context.get(services, EventSinkV2); + const routes = HttpApiBuilder.layer(OrchestrationTestApi).pipe( + Layer.provide(OrchestrationHttp.layer), + Layer.provide( + Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed(environmentId), + }), + ), + Layer.provide( + Layer.succeed(EnvironmentAuthenticatedAuth, (effect) => + Effect.suspend(() => + effect.pipe( + Effect.provideService(EnvironmentAuthenticatedPrincipal, { + sessionId: AuthSessionId.make("test-session"), + subject: "test", + method: "bearer-access-token", + scopes: new Set(allowed() ? ["orchestration:read"] : []), + }), ), ), - OrchestrationEventStore.layer, - ProjectStore.layer, - ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), + ), + ), + Layer.provide(Layer.succeedContext(services)), + Layer.provide(Layer.mock(ProjectEnrichmentService)({})), + Layer.provide(HttpPlatform.layer), + Layer.provide(Etag.layerWeak), + Layer.provide(NodeServices.layer), + ); + const { handler, dispose } = HttpRouter.toWebHandler(routes, { disableLogger: true }); + yield* Effect.addFinalizer(() => Effect.promise(dispose)); + const request = (path: string) => + Effect.promise(() => + handler( + new Request(`http://localhost${path}`, { + headers: { [ORCHESTRATION_PROTOCOL_HEADER]: ORCHESTRATION_PROTOCOL_VERSION_TEXT }, + }), + ), ); - const eventSink = Context.get(services, EventSinkV2); + return { services, eventSink, request }; +}); + +it.effect("exports full history with size, scope, and missing-thread checks", () => + Effect.gen(function* () { + let allowed = true; + const { services, eventSink, request } = yield* makeHttpHarness(() => allowed); yield* eventSink.write({ events: [ { @@ -128,43 +181,8 @@ it.effect("exports full history with size, scope, and missing-thread checks", () }, ], }); - const routes = HttpApiBuilder.layer(TranscriptTestApi).pipe( - Layer.provide(OrchestrationHttp.layer), - Layer.provide( - Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, { - getEnvironmentId: Effect.succeed(environmentId), - }), - ), - Layer.provide( - Layer.succeed(EnvironmentAuthenticatedAuth, (effect) => - Effect.suspend(() => - effect.pipe( - Effect.provideService(EnvironmentAuthenticatedPrincipal, { - sessionId: AuthSessionId.make("test-session"), - subject: "test", - method: "bearer-access-token", - scopes: new Set(allowed ? ["orchestration:read"] : []), - }), - ), - ), - ), - ), - Layer.provide(Layer.succeedContext(services)), - Layer.provide(Layer.mock(ProjectEnrichmentService)({})), - Layer.provide(HttpPlatform.layer), - Layer.provide(Etag.layerWeak), - Layer.provide(NodeServices.layer), - ); - const { handler, dispose } = HttpRouter.toWebHandler(routes, { disableLogger: true }); - yield* Effect.addFinalizer(() => Effect.promise(dispose)); const read = (threadId: ThreadId) => - Effect.promise(() => - handler( - new Request(`http://localhost/api/orchestration/threads/${threadId}/transcript`, { - headers: { [ORCHESTRATION_PROTOCOL_HEADER]: ORCHESTRATION_PROTOCOL_VERSION_TEXT }, - }), - ), - ); + request(`/api/orchestration/threads/${threadId}/transcript`); const emptyResponse = yield* read(thread.id); expect(emptyResponse.status).toBe(200); @@ -242,3 +260,66 @@ it.effect("exports full history with size, scope, and missing-thread checks", () expect((yield* read(thread.id)).status).toBe(403); }), ); + +it.effect("deduplicates equal pull request lists only in opted-in HTTP shell snapshots", () => + Effect.gen(function* () { + const { eventSink, request } = yield* makeHttpHarness(); + const link = (number: number): ThreadPullRequestLink => ({ + host: "github.com", + repository: "pingdotgg/t3code", + number, + url: `https://github.com/pingdotgg/t3code/pull/${number}`, + source: "agent", + linkedAt: "2026-09-29T00:00:00.000Z", + snapshot: null, + stack: null, + }); + const lists = [[link(1), link(2)], [link(1), link(2)], [link(1)]]; + yield* eventSink.write({ + events: lists.map((pullRequests, index) => { + const id = ThreadId.make(`shell-${index}`); + return { + id: EventId.make(`create-${id}`), + type: "thread.created" as const, + threadId: id, + occurredAt: now, + payload: { ...thread, id, pullRequests }, + }; + }), + }); + + const defaultResponse = yield* request("/api/orchestration/shell"); + expect(defaultResponse.status).toBe(200); + const json = yield* Effect.promise(() => defaultResponse.json()); + const full = yield* decodeShellSnapshotWire(json); + const decoded = yield* decodeShellSnapshot(json); + expect(decoded.threads).toHaveLength(3); + for (const [index, links] of lists.entries()) { + expect(decoded.threads.find((row) => row.id === `shell-${index}`)?.pullRequests).toEqual( + links, + ); + } + expect(full.threads.every((row) => !("pullRequestReference" in row))).toBe(true); + + const response = yield* request("/api/orchestration/shell?deduplicatePullRequests=true"); + expect(response.status).toBe(200); + const compact = yield* Effect.promise(() => response.json()); + const duplicateIndices = decoded.threads.flatMap((row, index) => + row.id === "shell-0" || row.id === "shell-1" ? [index] : [], + ); + const [first, duplicate] = duplicateIndices; + expect(first).toBeDefined(); + expect(duplicate).toBeDefined(); + expect(compact).toEqual({ + ...full, + threads: full.threads.map((row, index) => { + if (index !== duplicate) return row; + const { pullRequests: _links, ...rest } = row; + return { ...rest, pullRequestReference: first }; + }), + }); + const disabled = yield* request("/api/orchestration/shell?deduplicatePullRequests=false"); + expect(disabled.status).toBe(200); + expect(yield* Effect.promise(() => disabled.json())).toEqual(full); + }), +); diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index cc8e987880df8..5586021d079dd 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -30,7 +30,11 @@ import { } from "./threadHistoryPaging.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ProjectStore from "./ProjectStore.ts"; -import { buildActiveShellSnapshot, loadShellSnapshotParts } from "./ShellStream.ts"; +import { + buildActiveShellSnapshot, + deduplicateShellPullRequests, + loadShellSnapshotParts, +} from "./ShellStream.ts"; import { boundedSnapshotResponseFields } from "./ThreadStream.ts"; import { projectThreadProjectionForWire } from "./WireProjection.ts"; @@ -75,7 +79,9 @@ export const layer = HttpApiBuilder.group( ), ); - const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* () { + const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* ( + deduplicatePullRequests: boolean, + ) { const base = buildActiveShellSnapshot( yield* loadShellSnapshotParts({ sql, @@ -85,7 +91,8 @@ export const layer = HttpApiBuilder.group( }), ); const projects = yield* enrichProjectShells(base.projects); - return { ...base, projects }; + const snapshot = { ...base, projects }; + return deduplicatePullRequests ? deduplicateShellPullRequests(snapshot) : snapshot; }); const loadThreadSnapshot = Effect.fn("http.orchestration.loadThreadSnapshot")(function* ( @@ -154,7 +161,7 @@ export const layer = HttpApiBuilder.group( Effect.fn("environment.orchestration.shellSnapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); - return yield* loadShellSnapshot().pipe( + return yield* loadShellSnapshot(args.query.deduplicatePullRequests === "true").pipe( traceLocalHandlerWork, Effect.catch((cause) => failEnvironmentInternal("orchestration_snapshot_failed", cause), diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 5f783854ee559..ad6c1368d23e3 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1,7 +1,8 @@ import { EnvironmentId, ORCHESTRATION_V2_WS_METHODS, - type OrchestrationV2ShellSnapshot, + OrchestrationV2GetShellSnapshotError, + OrchestrationV2ShellSnapshot, type OrchestrationV2ShellStreamItem, ThreadId, type ThreadPullRequestLink, @@ -15,6 +16,7 @@ import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; @@ -28,6 +30,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; +import * as RpcHttp from "../rpc/http.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { makeEnvironmentShellState } from "./shell.ts"; import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts"; @@ -49,15 +52,24 @@ const PREPARED: PreparedConnection = { target: TARGET, }; +const encodeShellSnapshot = Schema.encodeEffect(OrchestrationV2ShellSnapshot); + const LIVE_SHELL_SNAPSHOT: OrchestrationV2ShellSnapshot = { ...v2ShellSnapshot, snapshotSequence: 1, }; -function session(client: WsRpcProtocolClient): RpcSession.RpcSession { +function session( + client: WsRpcProtocolClient, + completionMarker = true, + serverVersion = "1.0.0", +): RpcSession.RpcSession { return { client, - initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never), + initialConfig: Effect.succeed({ + environment: { serverVersion }, + ...(completionMarker ? { shellResumeCompletionMarker: true } : {}), + } as never), subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, @@ -66,6 +78,78 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession { } describe("environment shell synchronization", () => { + it.effect( + "opts into HTTP link deduplication and restores shared lists while accepting old servers", + () => + Effect.gen(function* () { + const link: ThreadPullRequestLink = { + host: "github.com", + repository: "pingdotgg/t3code", + number: 7, + url: "https://github.com/pingdotgg/t3code/pull/7", + source: "agent", + linkedAt: "2026-06-20T00:00:00.000Z", + snapshot: null, + stack: null, + }; + for (const references of [false, true]) { + const duplicateId = ThreadId.make("duplicate"); + const encoded = yield* encodeShellSnapshot({ + ...v2ShellSnapshot, + threads: [ + { ...v2ThreadShell, pullRequests: [link] }, + { ...v2ThreadShell, id: duplicateId, pullRequests: [link] }, + ], + }); + const invalidRows = [8, 99, 4, -1, 0.5].map((reference, index) => ({ + ...encoded.threads[0], + id: `invalid-${index}`, + pullRequests: undefined, + pullRequestReference: reference, + })); + const response = { + ...encoded, + threads: [ + encoded.threads[0], + references + ? { ...encoded.threads[1], pullRequests: undefined, pullRequestReference: 0 } + : encoded.threads[1], + ...invalidRows, + { ...encoded.threads[0], id: "unreadable", pullRequests: [{ number: "invalid" }] }, + { ...encoded.threads[0], id: "future", pullRequests: [link] }, + ], + }; + const requests: Array = []; + const fetchFn: typeof fetch = async (url) => { + requests.push(String(url)); + return Response.json(response); + }; + const decoded = yield* ShellSnapshotLoader.fetchEnvironmentShellSnapshot({ + prepared: PREPARED, + signer: Option.none(), + }).pipe(Effect.provide(RpcHttp.layerRemoteHttpClient(fetchFn))); + expect(requests).toHaveLength(1); + const url = new URL(requests[0]!); + expect(url.pathname).toBe("/api/orchestration/shell"); + expect(url.searchParams.get("deduplicatePullRequests")).toBe("true"); + expect(decoded.threads).toHaveLength(response.threads.length); + expect(decoded.threads.every((row) => row.pullRequests === undefined)).toBe(true); + const loading = yield* Effect.forkScoped(decoded.loadPullRequests!); + yield* TestClock.adjust("1 millis"); + const restored = yield* Fiber.join(loading); + expect(restored.get(v2ThreadShell.id)).toEqual([link]); + expect(restored.get(duplicateId)).toEqual([link]); + if (references) { + expect(restored.get(duplicateId)).toBe(restored.get(v2ThreadShell.id)); + } + expect(restored.get(ThreadId.make("future"))).toEqual([link]); + expect([...restored.keys()].sort()).toEqual( + [duplicateId, "future", v2ThreadShell.id].sort(), + ); + } + }), + ); + it.effect("publishes live state before persistence and preserves it when ready", () => Effect.gen(function* () { const events = yield* Queue.unbounded(); @@ -245,15 +329,19 @@ describe("environment shell synchronization", () => { }).pipe(Effect.scoped), ); - it.effect("refreshes a warm shell cache from HTTP before resuming", () => + it.effect("refreshes a warm shell cache then retries stream errors from the live cursor", () => Effect.gen(function* () { const cachedSnapshot: OrchestrationV2ShellSnapshot = { ...v2ShellSnapshot, snapshotSequence: 5, }; - const events = yield* Queue.unbounded(); - const capturedAfterSequence = yield* SubscriptionRef.make(undefined); - const capturedCompletionMarker = yield* Ref.make(false); + const events = yield* Queue.unbounded< + OrchestrationV2ShellStreamItem | OrchestrationV2GetShellSnapshotError + >(); + const subscriptions = yield* Queue.unbounded<{ + readonly afterSequence?: number; + readonly requestCompletionMarker?: true; + }>(); const loaderCalls = yield* SubscriptionRef.make(0); const httpSnapshot: OrchestrationV2ShellSnapshot = { ...v2ShellSnapshot, @@ -265,10 +353,15 @@ describe("environment shell synchronization", () => { readonly requestCompletionMarker?: true; }) => Stream.unwrap( - Effect.all([ - Ref.set(capturedCompletionMarker, input.requestCompletionMarker === true), - SubscriptionRef.set(capturedAfterSequence, input.afterSequence), - ]).pipe(Effect.as(Stream.fromQueue(events))), + Queue.offer(subscriptions, input).pipe( + Effect.as( + Stream.fromQueue(events).pipe( + Stream.mapEffect((item) => + "_tag" in item ? Effect.fail(item) : Effect.succeed(item), + ), + ), + ), + ), ), } as unknown as WsRpcProtocolClient; const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); @@ -310,50 +403,126 @@ describe("environment shell synchronization", () => { Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), ); - yield* SubscriptionRef.changes(capturedAfterSequence).pipe( - Stream.filter((value) => value !== undefined), - Stream.runHead, - ); - - expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(9); - expect(yield* Ref.get(capturedCompletionMarker)).toBe(true); + expect(yield* Queue.take(subscriptions)).toEqual({ + afterSequence: 9, + requestCompletionMarker: true, + }); expect(yield* SubscriptionRef.get(loaderCalls)).toBe(1); const synchronizing = yield* SubscriptionRef.get(shellState); expect(synchronizing.status).toBe("synchronizing"); expect(Option.getOrThrow(synchronizing.snapshot)).toEqual(httpSnapshot); + const updatedRow = { ...v2ThreadShell, title: "Updated before stream failure" }; + yield* Queue.offer(events, { + kind: "thread.updated", + sequence: 10, + location: "active", + thread: updatedRow, + }); yield* Queue.offer(events, { kind: "synchronized" }); yield* SubscriptionRef.changes(shellState).pipe( Stream.filter((value) => value.status === "live"), Stream.runHead, ); + + yield* Queue.offer( + events, + new OrchestrationV2GetShellSnapshotError({ + message: "Failed while streaming the application shell", + }), + ); + const failed = yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((value) => Option.isSome(value.error)), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + expect(Option.getOrThrow(failed.snapshot).snapshotSequence).toBe(10); + yield* TestClock.adjust("250 millis"); + const retry = yield* Queue.take(subscriptions); + expect({ input: retry, loads: yield* SubscriptionRef.get(loaderCalls) }).toEqual({ + input: { afterSequence: 10, requestCompletionMarker: true }, + loads: 1, + }); + yield* Queue.offer(events, { kind: "synchronized" }); + const recovered = yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((value) => value.status === "live"), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + expect(recovered.error).toEqual(Option.none()); + expect(Option.getOrThrow(recovered.snapshot).threads[0]).toEqual(updatedRow); }), ); - it.effect("resubscribes from the in-memory shell cursor when the app becomes active", () => + it.effect.each([ + { + name: "completed in-memory sync", + completionMarker: true, + synchronized: true, + resume: true, + serverVersion: "1.0.0", + refreshSequence: 20, + }, + { + name: "server version change without new events", + completionMarker: true, + synchronized: true, + resume: false, + serverVersion: "1.0.1", + refreshSequence: 11, + }, + { + name: "partial initial sync", + completionMarker: true, + synchronized: false, + resume: false, + serverVersion: "1.0.0", + refreshSequence: 20, + }, + { + name: "legacy server", + completionMarker: false, + synchronized: true, + resume: false, + serverVersion: "1.0.0", + refreshSequence: 20, + }, + ])("resubscribes on foreground and reconnect after $name", (scenario) => Effect.gen(function* () { const events = yield* Queue.unbounded(); const wakeups = yield* Queue.unbounded(); + const subscriptions = yield* Queue.unbounded<{ + readonly afterSequence?: number; + readonly requestCompletionMarker?: true; + }>(); + const linksReady = yield* Deferred.make(); const loaderCalls = yield* Ref.make(0); - const capturedAfterSequences = yield* Ref.make>([]); + const otherThreadId = ThreadId.make("other-thread"); + const link = (number: number): ThreadPullRequestLink => ({ + host: "github.com", + repository: "pingdotgg/t3code", + number, + url: `https://github.com/pingdotgg/t3code/pull/${number}`, + source: "agent", + linkedAt: "2026-06-20T00:00:00.000Z", + snapshot: null, + stack: null, + }); const client = { [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number; + readonly requestCompletionMarker?: true; }) => Stream.unwrap( - Ref.update(capturedAfterSequences, (captured) => [ - ...captured, - input.afterSequence, - ]).pipe(Effect.as(Stream.fromQueue(events))), + Queue.offer(subscriptions, input).pipe(Effect.as(Stream.fromQueue(events))), ), } as unknown as WsRpcProtocolClient; - const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); const activeSession = yield* SubscriptionRef.make>( Option.some(session(client)), ); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, - state: supervisorState, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), session: activeSession, prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), connect: Effect.void, @@ -378,7 +547,19 @@ describe("environment shell synchronization", () => { load: () => Ref.updateAndGet(loaderCalls, (count) => count + 1).pipe( Effect.map((count) => - Option.some({ ...LIVE_SHELL_SNAPSHOT, snapshotSequence: count * 10 }), + Option.some({ + ...LIVE_SHELL_SNAPSHOT, + snapshotSequence: count === 1 ? 10 : scenario.refreshSequence, + threads: [v2ThreadShell, { ...v2ThreadShell, id: otherThreadId }], + loadPullRequests: Deferred.await(linksReady).pipe( + Effect.as( + new Map([ + [v2ThreadShell.id, [link(1)]], + [otherThreadId, [link(2)]], + ]), + ), + ), + }), ), ), }); @@ -392,57 +573,75 @@ describe("environment shell synchronization", () => { ), ); - for (let attempt = 0; attempt < 100; attempt += 1) { - if ((yield* Ref.get(capturedAfterSequences)).length >= 1) break; - yield* Effect.yieldNow; - } - expect(yield* Ref.get(capturedAfterSequences)).toEqual([10]); - yield* Queue.offer(events, { kind: "synchronized" }); - yield* SubscriptionRef.changes(shellState).pipe( - Stream.filter((value) => value.status === "live"), - Stream.runHead, - ); - + const marker = { requestCompletionMarker: true }; + expect(yield* Queue.take(subscriptions)).toEqual({ afterSequence: 10, ...marker }); + expect(yield* Ref.get(loaderCalls)).toBe(1); + const updatedRow = { + ...v2ThreadShell, + title: "Changed while connected", + pullRequests: [link(9)], + }; yield* Queue.offer(events, { - kind: "snapshot", - snapshot: { ...LIVE_SHELL_SNAPSHOT, snapshotSequence: 40 }, + kind: "thread.updated", + sequence: 11, + location: "active", + thread: updatedRow, }); yield* SubscriptionRef.changes(shellState).pipe( - Stream.filter( - (value) => Option.isSome(value.snapshot) && value.snapshot.value.snapshotSequence === 40, - ), + Stream.filter((value) => Option.getOrNull(value.snapshot)?.snapshotSequence === 11), Stream.runHead, ); - - yield* Queue.offer(wakeups, "application-active"); - for (let attempt = 0; attempt < 100; attempt += 1) { - if ((yield* Ref.get(capturedAfterSequences)).length >= 2) break; - yield* Effect.yieldNow; + const completeSync = Effect.gen(function* () { + yield* Queue.offer(events, { kind: "synchronized" }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((value) => value.status === "live"), + Stream.runHead, + ); + }); + if (scenario.synchronized) yield* completeSync; + for (const wakeup of ["application-active", "application-active-probe"] as const) { + yield* Queue.offer(wakeups, wakeup); + expect(yield* Queue.take(subscriptions)).toEqual({ afterSequence: 11, ...marker }); + if (scenario.synchronized) yield* completeSync; } - expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40]); - yield* Queue.offer(events, { kind: "synchronized" }); + expect(yield* Ref.get(loaderCalls)).toBe(1); - yield* Queue.offer(wakeups, "application-active-probe"); - for (let attempt = 0; attempt < 100; attempt += 1) { - if ((yield* Ref.get(capturedAfterSequences)).length >= 3) break; - yield* Effect.yieldNow; - } - expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40, 40]); + yield* SubscriptionRef.set( + activeSession, + Option.some(session(client, scenario.completionMarker, scenario.serverVersion)), + ); + expect(yield* Queue.take(subscriptions)).toEqual({ + afterSequence: scenario.resume ? 11 : scenario.refreshSequence, + ...(scenario.completionMarker ? marker : {}), + }); + expect(yield* Ref.get(loaderCalls)).toBe(scenario.resume ? 1 : 2); + if (scenario.completionMarker) yield* completeSync; - yield* Queue.offer(wakeups, "application-active-reconnect"); - for (let attempt = 0; attempt < 10; attempt += 1) { - yield* Effect.yieldNow; - } - expect((yield* Ref.get(capturedAfterSequences)).length).toBe(3); - expect(yield* Ref.get(loaderCalls)).toBe(1); + yield* Deferred.succeed(linksReady, undefined); + const filled = yield* SubscriptionRef.changes(shellState).pipe( + Stream.map((state) => Option.getOrThrow(state.snapshot).threads), + Stream.filter((rows) => rows[1]?.pullRequests !== undefined), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + expect(filled[1]?.pullRequests).toEqual([link(2)]); + expect(filled[0]?.pullRequests).toEqual([link(scenario.resume ? 9 : 1)]); + if (scenario.resume) expect(filled[0]).toBe(updatedRow); - yield* SubscriptionRef.set(activeSession, Option.some(session(client))); - for (let attempt = 0; attempt < 100; attempt += 1) { - if ((yield* Ref.get(capturedAfterSequences)).length >= 4) break; - yield* Effect.yieldNow; - } - expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40, 40, 20]); - expect(yield* Ref.get(loaderCalls)).toBe(2); + yield* Queue.offer(events, { + kind: "thread.updated", + sequence: 21, + location: "active", + thread: { ...updatedRow, title: "Changed while disconnected" }, + }); + const replayed = yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((value) => Option.getOrNull(value.snapshot)?.snapshotSequence === 21), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + expect(Option.getOrThrow(replayed.snapshot).threads[0]?.title).toBe( + "Changed while disconnected", + ); }), ); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 12e5f19e8997b..9124f1a56997d 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -26,7 +26,7 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; -import { subscribeDynamic } from "../rpc/client.ts"; +import { subscribeDynamicWithSession } from "../rpc/client.ts"; import type { RpcSession } from "../rpc/session.ts"; import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts"; import type { DeferredShellSnapshot } from "./shellPullRequests.ts"; @@ -91,6 +91,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") error: Option.none(), }); const awaitingCompletion = yield* Ref.make(false); + const synchronizedSnapshot = yield* Ref.make(false); + const synchronizedServerVersion = yield* Ref.make(null); const lastAuthoritativeSession = yield* Ref.make(null); const activeSubscriptionSession = yield* Ref.make(null); const latestLiveSnapshot = yield* Ref.make>( @@ -160,6 +162,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); const setStreamError = (error: unknown) => Ref.set(awaitingCompletion, false).pipe( + Effect.andThen(Ref.set(synchronizedSnapshot, false)), Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")), Effect.annotateLogs({ environmentId, @@ -179,21 +182,32 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const snapshotWriteLock = yield* Semaphore.make(1); const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( - items: ReadonlyArray, + items: ReadonlyArray, pendingLinkIds?: ReadonlySet, ) { const initial = yield* SubscriptionRef.get(state); + const activeSession = yield* Ref.get(activeSubscriptionSession); let waiting = yield* Ref.get(awaitingCompletion); let next = initial; - let receivedSnapshot = false; - for (const item of items) { + let snapshotSession: RpcSession | null = null; + for (const [session, item] of items) { + if (session !== activeSession) continue; if (item.kind === "synchronized") { waiting = false; if (Option.isSome(next.snapshot)) { + const serverVersion = yield* session.initialConfig.pipe( + Effect.map((config) => config.environment.serverVersion), + Effect.orElseSucceed(() => null), + ); + yield* Ref.set(synchronizedServerVersion, serverVersion); + yield* Ref.set(synchronizedSnapshot, true); next = { ...next, status: "live", error: Option.none() }; } continue; } + if (item.kind === "snapshot" && item.resolvedRepositoryIdentityRoots === undefined) { + yield* Ref.set(synchronizedSnapshot, false); + } const nextSnapshot = item.kind === "snapshot" ? mergeShellSnapshotProjects( @@ -213,7 +227,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") : snapshot, }); if (nextSnapshot === null) continue; - receivedSnapshot ||= item.kind === "snapshot"; + if (item.kind === "snapshot") snapshotSession = session; next = { snapshot: Option.some(nextSnapshot), status: waiting ? "synchronizing" : "live", @@ -231,11 +245,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* Ref.set(latestLiveSnapshot, next.snapshot); } yield* SubscriptionRef.set(state, next); - if (receivedSnapshot) { - const session = yield* Ref.get(activeSubscriptionSession); - if (session !== null) { - yield* Ref.set(lastAuthoritativeSession, session); - } + if (snapshotSession !== null) { + yield* Ref.set(lastAuthoritativeSession, snapshotSession); } if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) { yield* Queue.offer(persistence, next.snapshot.value); @@ -317,21 +328,24 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* setSynchronizing; yield* Effect.forkScoped( - subscribeDynamic( + subscribeDynamicWithSession( ORCHESTRATION_V2_WS_METHODS.subscribeShell, Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) { yield* Ref.set(activeSubscriptionSession, session); - const supportsCompletionMarker = yield* session.initialConfig.pipe( - Effect.map((config) => config.shellResumeCompletionMarker === true), - Effect.orElseSucceed(() => false), - ); + const initialConfig = yield* session.initialConfig.pipe(Effect.orElseSucceed(() => null)); + const supportsCompletionMarker = initialConfig?.shellResumeCompletionMarker === true; yield* Ref.set(awaitingCompletion, supportsCompletionMarker); yield* setSynchronizing; - // Foreground resubscriptions on the same live session can resume from - // the in-memory cursor. A new session reloads the authoritative HTTP - // snapshot so a valid cursor cannot preserve incomplete cached data. - const hasAuthoritativeSnapshot = (yield* Ref.get(lastAuthoritativeSession)) === session; + // Only completed in-memory syncs survive a session change. Disk caches and partial + // loads need a refresh; upgrades can rewrite projections without adding events. + const hasAuthoritativeSnapshot = + (yield* Ref.get(lastAuthoritativeSession)) === session || + (supportsCompletionMarker && + (yield* Ref.get(synchronizedSnapshot)) && + (yield* Ref.get(synchronizedServerVersion)) === + initialConfig.environment.serverVersion); + yield* Ref.set(synchronizedSnapshot, false); let canResume = hasAuthoritativeSnapshot; let current = yield* SubscriptionRef.get(state); if (!hasAuthoritativeSnapshot || Option.isNone(current.snapshot)) { @@ -362,7 +376,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ? [] : httpSnapshot.value.threads.map((thread) => thread.id), ); - yield* applyItems([{ kind: "snapshot", snapshot }], pendingLinkIds); + yield* applyItems([[session, { kind: "snapshot", snapshot }]], pendingLinkIds); canResume = true; current = yield* SubscriptionRef.get(state); if (Option.isSome(current.snapshot)) { @@ -374,6 +388,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (!canResume || Option.isNone(current.snapshot)) { return supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}; } + yield* Ref.set(lastAuthoritativeSession, session); if (!supportsCompletionMarker) { yield* SubscriptionRef.update(state, (value) => ({ ...value, diff --git a/packages/client-runtime/src/state/shellPullRequests.ts b/packages/client-runtime/src/state/shellPullRequests.ts index 340b1403f2937..5170e35773102 100644 --- a/packages/client-runtime/src/state/shellPullRequests.ts +++ b/packages/client-runtime/src/state/shellPullRequests.ts @@ -30,12 +30,25 @@ export function detachPullRequests(snapshot: unknown): ReadonlyArray { if (!Predicate.isObject(snapshot)) return []; const rows: unknown = snapshot.threads; if (!Array.isArray(rows)) return []; - return rows.map((row: unknown) => { - if (!Predicate.isObject(row)) return undefined; - const links = row.pullRequests; + const detached: Array = []; + for (const [index, row] of rows.entries()) { + if (!Predicate.isObject(row)) { + detached.push(undefined); + continue; + } + const reference = row.pullRequestReference; + const links = + typeof reference === "number" && + Number.isInteger(reference) && + reference >= 0 && + reference < index + ? detached[reference] + : row.pullRequests; delete row.pullRequests; - return links; - }); + delete row.pullRequestReference; + detached.push(links); + } + return detached; } export function deferPullRequests( @@ -46,10 +59,16 @@ export function deferPullRequests( // Schema decoding runs when called, so each thread's decode is built after a yield. const loadPullRequests = Effect.gen(function* () { const linksByThreadId = new Map>(); + const decodedLists = new Map | undefined>(); let batchSize = LINKS_PER_BATCH; for (const [index, threadId] of threadIds.entries()) { const raw = rawLinks[index]; if (raw === undefined) continue; + if (decodedLists.has(raw)) { + const links = decodedLists.get(raw); + if (links !== undefined) linksByThreadId.set(threadId, links); + continue; + } const size = Array.isArray(raw) ? raw.length : 1; if (batchSize + size > LINKS_PER_BATCH) { yield* yieldToHost; @@ -64,6 +83,7 @@ export function deferPullRequests( }).pipe(Effect.as(undefined)), ), ); + decodedLists.set(raw, links); if (links !== undefined) linksByThreadId.set(threadId, links); } return linksByThreadId as ReadonlyMap>; diff --git a/packages/client-runtime/src/state/shellSnapshotHttp.ts b/packages/client-runtime/src/state/shellSnapshotHttp.ts index cdb5ca77db56f..ee6813c9685e0 100644 --- a/packages/client-runtime/src/state/shellSnapshotHttp.ts +++ b/packages/client-runtime/src/state/shellSnapshotHttp.ts @@ -38,12 +38,13 @@ export const fetchEnvironmentShellSnapshot = Effect.fn( ...input, group: "orchestration", method: "GET", - url: (urls) => urls.shellSnapshot(), + url: (urls) => urls.shellSnapshot({ query: { deduplicatePullRequests: "true" } }), timeoutMs: input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS, request: ({ client, headers }) => client .shellSnapshot({ headers: withOrchestrationProtocolHeader(headers), + query: { deduplicatePullRequests: "true" }, responseMode: "response-only", }) .pipe(Effect.flatMap(decodeShellSnapshotResponse)), diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 31b69997f43cf..ca4613c99951f 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -594,6 +594,7 @@ class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration") .add( HttpApiEndpoint.get("shellSnapshot", "/api/orchestration/shell", { headers: OrchestrationProtocolHeaders, + query: Schema.Struct({ deduplicatePullRequests: Schema.optional(Schema.String) }), success: OrchestrationV2ShellSnapshot, error: EnvironmentOrchestrationSnapshotErrors, }).middleware(EnvironmentAuthenticatedAuth), diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..2ce4a3e7a89e2 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1808,6 +1808,8 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ worktreePath: Schema.NullOr(TrimmedNonEmptyString), linkedPullRequest: Schema.optional(Schema.NullOr(ThreadLinkedPullRequest)), pullRequests: Schema.optional(Schema.Array(ThreadPullRequestLink)), + /** Opted-in HTTP snapshots reuse an earlier thread row's identical link list. */ + pullRequestReference: Schema.optionalKey(NonNegativeInt), branchPullRequest: Schema.optional(Schema.NullOr(ThreadLinkedPullRequest)), lineage: OrchestrationV2AppThreadLineage, forkedFrom: Schema.NullOr(OrchestrationV2AppThread.fields.forkedFrom),