perf(client): resume completed shells and deduplicate repeated links

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<string, number>();
+  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

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<AuthEnvironmentScope>(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<AuthEnvironmentScope>(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);
+  }),
+);

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),