authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-07-28 15:50:43-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-28 15:50:43-07:00
log5fcdefd057751eb511de9e61142ef797f142059b
treedeae3a0c133c53cd313e884f15c50d5d484657c2
parent38a6e3ce65b1d4fa58788faa96ff9410a8c26ffe
signature Signed by PGP key B5690EEEBB952194

perf(server): trim stale context-window rows and drop dead replay RPC (#4791)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

6 files changed, 179 insertions(+), 184 deletions(-)

apps/server/src/orchestration/ActivityPayloadProjection.ts+49-1
......@@ -201,6 +201,52 @@ export function projectActivityPayload(
201201 };
202202}
203203
204/**
205 * Matches the validity rule in the web client's
206 * `deriveLatestContextWindowSnapshot`: rows without a finite, non-negative
207 * `usedTokens` are skipped during its backward walk, so they must not shadow
208 * an earlier resolvable row here.
209 */
210function isResolvableContextWindowActivity(activity: OrchestrationThreadActivity): boolean {
211 if (activity.kind !== "context-window.updated") {
212 return false;
213 }
214 const payload = asRecord(activity.payload);
215 const usedTokens = payload?.usedTokens;
216 return typeof usedTokens === "number" && Number.isFinite(usedTokens) && usedTokens >= 0;
217}
218
219/**
220 * Drops all but the last resolvable context-window activity per turn from a
221 * snapshot. Clients only ever read the latest usage value (walking the array
222 * backwards), so shipping the full history — often thousands of rows on long
223 * threads — buys nothing. Retention is per turn rather than per thread because
224 * a live `thread.reverted` makes the client discard whole turns; keeping each
225 * turn's latest row means the meter can still resolve a value from the turns
226 * that survive. Malformed rows pass through untouched rather than shadowing a
227 * valid earlier row. Live `thread.activity-appended` events are untouched:
228 * newer updates still stream through and supersede the retained rows on the
229 * client.
230 */
231function dropStaleContextWindowActivities(
232 activities: ReadonlyArray<OrchestrationThreadActivity>,
233): ReadonlyArray<OrchestrationThreadActivity> {
234 const latestIndexByTurn = new Map<string | null, number>();
235 for (let index = 0; index < activities.length; index += 1) {
236 if (isResolvableContextWindowActivity(activities[index]!)) {
237 latestIndexByTurn.set(activities[index]!.turnId, index);
238 }
239 }
240 if (latestIndexByTurn.size === 0) {
241 return activities;
242 }
243 return activities.filter(
244 (activity, index) =>
245 !isResolvableContextWindowActivity(activity) ||
246 latestIndexByTurn.get(activity.turnId) === index,
247 );
248}
249
204250export function projectThreadDetailSnapshot(
205251 snapshot: OrchestrationThreadDetailSnapshot,
206252): OrchestrationThreadDetailSnapshot {
......@@ -208,7 +254,9 @@ export function projectThreadDetailSnapshot(
208254 ...snapshot,
209255 thread: {
210256 ...snapshot.thread,
211 activities: snapshot.thread.activities.map(projectActivityPayload),
257 activities: dropStaleContextWindowActivities(snapshot.thread.activities).map(
258 projectActivityPayload,
259 ),
212260 },
213261 };
214262}
apps/server/src/server.test.ts-76
......@@ -5690,15 +5690,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
56905690 ),
56915691 );
56925692 assert.equal(fullDiffResult.diff, "full-diff");
5693
5694 const replayResult = yield* Effect.scoped(
5695 withWsRpcClient(wsUrl, (client) =>
5696 client[ORCHESTRATION_WS_METHODS.replayEvents]({
5697 fromSequenceExclusive: 0,
5698 }),
5699 ),
5700 );
5701 assert.deepEqual(replayResult, []);
57025693 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
57035694 );
57045695
......@@ -6361,73 +6352,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
63616352 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
63626353 );
63636354
6364 it.effect("enriches replayed project events with repository identity metadata", () =>
6365 Effect.gen(function* () {
6366 const repositoryIdentity = {
6367 canonicalKey: "github.com/t3tools/t3code",
6368 locator: {
6369 source: "git-remote" as const,
6370 remoteName: "origin",
6371 remoteUrl: "git@github.com:T3Tools/t3code.git",
6372 },
6373 displayName: "T3Tools/t3code",
6374 provider: "github",
6375 owner: "T3Tools",
6376 name: "t3code",
6377 };
6378
6379 yield* buildAppUnderTest({
6380 layers: {
6381 orchestrationEngine: {
6382 readEvents: (_fromSequenceExclusive) =>
6383 Stream.make({
6384 sequence: 1,
6385 eventId: EventId.make("event-1"),
6386 aggregateKind: "project",
6387 aggregateId: defaultProjectId,
6388 occurredAt: "2026-04-05T00:00:00.000Z",
6389 commandId: null,
6390 causationEventId: null,
6391 correlationId: null,
6392 metadata: {},
6393 type: "project.created",
6394 payload: {
6395 projectId: defaultProjectId,
6396 title: "Default Project",
6397 workspaceRoot: "/tmp/default-project",
6398 defaultModelSelection,
6399 scripts: [],
6400 createdAt: "2026-04-05T00:00:00.000Z",
6401 updatedAt: "2026-04-05T00:00:00.000Z",
6402 },
6403 } satisfies Extract<OrchestrationEvent, { type: "project.created" }>),
6404 },
6405 repositoryIdentityResolver: {
6406 resolve: () => Effect.succeed(repositoryIdentity),
6407 },
6408 },
6409 });
6410
6411 const wsUrl = yield* getWsServerUrl("/ws");
6412 const replayResult = yield* Effect.scoped(
6413 withWsRpcClient(wsUrl, (client) =>
6414 client[ORCHESTRATION_WS_METHODS.replayEvents]({
6415 fromSequenceExclusive: 0,
6416 }),
6417 ),
6418 );
6419
6420 const replayedEvent = replayResult[0];
6421 assert.equal(replayedEvent?.type, "project.created");
6422 assert.deepEqual(
6423 replayedEvent && replayedEvent.type === "project.created"
6424 ? replayedEvent.payload.repositoryIdentity
6425 : null,
6426 repositoryIdentity,
6427 );
6428 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
6429 );
6430
64316355 it.effect("stops the provider session and closes thread terminals after archive", () =>
64326356 Effect.gen(function* () {
64336357 const threadId = ThreadId.make("thread-archive");
apps/server/src/ws.ts-77
......@@ -46,7 +46,6 @@ import {
4646 ProjectWriteFileError,
4747 RelayClientInstallFailedError,
4848 type RelayClientInstallProgressEvent,
49 OrchestrationReplayEventsError,
5049 type FilesystemBrowseFailure,
5150 FilesystemBrowseError,
5251 AssetWorkspaceContextNotFoundError,
......@@ -60,7 +59,6 @@ import {
6059 WS_METHODS,
6160 WsRpcGroup,
6261} from "@t3tools/contracts";
63import { clamp } from "effect/Number";
6462import { HttpRouter, HttpServerRequest, HttpServerRespondable } from "effect/unstable/http";
6563import { RpcSerialization, RpcServer } from "effect/unstable/rpc";
6664
......@@ -99,7 +97,6 @@ import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts";
9997import * as GitWorkflowService from "./git/GitWorkflowService.ts";
10098import * as ReviewService from "./review/ReviewService.ts";
10199import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts";
102import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts";
103100import * as ServerEnvironment from "./environment/ServerEnvironment.ts";
104101import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
105102import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts";
......@@ -302,7 +299,6 @@ const RPC_REQUIRED_SCOPE = new Map<string, AuthEnvironmentScope>([
302299 [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope],
303300 [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope],
304301 [ORCHESTRATION_WS_METHODS.getFullThreadDiff, AuthOrchestrationReadScope],
305 [ORCHESTRATION_WS_METHODS.replayEvents, AuthOrchestrationReadScope],
306302 [ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope],
307303 [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, AuthOrchestrationReadScope],
308304 [ORCHESTRATION_WS_METHODS.subscribeThread, AuthOrchestrationReadScope],
......@@ -441,8 +437,6 @@ const makeWsRpcLayer = (
441437 const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
442438 const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem;
443439 const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;
444 const repositoryIdentityResolver =
445 yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
446440 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
447441 const serverAuth = yield* EnvironmentAuth.EnvironmentAuth;
448442 const sourceControlDiscovery = yield* SourceControlDiscovery.SourceControlDiscovery;
......@@ -592,53 +586,6 @@ const makeWsRpcLayer = (
592586 });
593587 };
594588
595 const enrichProjectEvent = (
596 event: OrchestrationEvent,
597 ): Effect.Effect<OrchestrationEvent, never, never> => {
598 switch (event.type) {
599 case "project.created":
600 return repositoryIdentityResolver.resolve(event.payload.workspaceRoot).pipe(
601 Effect.map((repositoryIdentity) => ({
602 ...event,
603 payload: {
604 ...event.payload,
605 repositoryIdentity,
606 },
607 })),
608 );
609 case "project.meta-updated":
610 return Effect.gen(function* () {
611 const workspaceRoot =
612 event.payload.workspaceRoot ??
613 Option.match(
614 yield* projectionSnapshotQuery.getProjectShellById(event.payload.projectId),
615 {
616 onNone: () => null,
617 onSome: (project) => project.workspaceRoot,
618 },
619 ) ??
620 null;
621 if (workspaceRoot === null) {
622 return event;
623 }
624
625 const repositoryIdentity = yield* repositoryIdentityResolver.resolve(workspaceRoot);
626 return {
627 ...event,
628 payload: {
629 ...event.payload,
630 repositoryIdentity,
631 },
632 } satisfies OrchestrationEvent;
633 }).pipe(Effect.orElseSucceed(() => event));
634 default:
635 return Effect.succeed(event);
636 }
637 };
638
639 const enrichOrchestrationEvents = (events: ReadonlyArray<OrchestrationEvent>) =>
640 Effect.forEach(events, enrichProjectEvent, { concurrency: 4 });
641
642589 const toShellStreamEvent = (
643590 event: OrchestrationEvent,
644591 ): Effect.Effect<Option.Option<OrchestrationShellStreamEvent>, never, never> => {
......@@ -1215,30 +1162,6 @@ const makeWsRpcLayer = (
12151162 ),
12161163 { "rpc.aggregate": "orchestration" },
12171164 ),
1218 [ORCHESTRATION_WS_METHODS.replayEvents]: (input) =>
1219 observeRpcEffect(
1220 ORCHESTRATION_WS_METHODS.replayEvents,
1221 Stream.runCollect(
1222 orchestrationEngine.readEvents(
1223 clamp(input.fromSequenceExclusive, {
1224 maximum: Number.MAX_SAFE_INTEGER,
1225 minimum: 0,
1226 }),
1227 ),
1228 ).pipe(
1229 Effect.map((events) => Array.from(events)),
1230 Effect.flatMap(enrichOrchestrationEvents),
1231 Effect.map((events) => events.map(projectActivityEvent)),
1232 Effect.mapError(
1233 (cause) =>
1234 new OrchestrationReplayEventsError({
1235 message: "Failed to replay orchestration events",
1236 cause,
1237 }),
1238 ),
1239 ),
1240 { "rpc.aggregate": "orchestration" },
1241 ),
12421165 [ORCHESTRATION_WS_METHODS.subscribeShell]: (input) =>
12431166 observeRpcStreamEffect(
12441167 ORCHESTRATION_WS_METHODS.subscribeShell,
apps/server/test/ActivityPayloadProjection.test.ts+130
......@@ -11,6 +11,7 @@ import {
1111import { describe, expect, it } from "vite-plus/test";
1212
1313import { buildThreadFeed, type ThreadFeedActivity } from "../../mobile/src/lib/threadActivity.ts";
14import { deriveLatestContextWindowSnapshot } from "../../web/src/lib/contextWindow.ts";
1415import { deriveWorkLogEntries } from "../../web/src/session-logic.ts";
1516import {
1617 projectActivityEvent,
......@@ -231,3 +232,132 @@ describe("projectActivityPayload", () => {
231232 expect(event.payload.activity).toBe(activity);
232233 });
233234});
235
236describe("context-window snapshot dedup", () => {
237 function makeContextWindowActivity(
238 id: string,
239 usedTokens: number,
240 turn = `turn-${id}`,
241 ): OrchestrationThreadActivity {
242 return {
243 id: EventId.make(id),
244 tone: "info",
245 kind: "context-window.updated",
246 summary: "Context window updated",
247 payload: { usedTokens, maxTokens: 200_000 },
248 turnId: TurnId.make(turn),
249 createdAt: "2026-07-27T00:00:00.000Z",
250 };
251 }
252
253 it("keeps only the latest context-window activity per turn in snapshots", () => {
254 const stale1 = makeContextWindowActivity("ctx-1", 1_000, "turn-a");
255 const latestA = makeContextWindowActivity("ctx-2", 2_000, "turn-a");
256 const latestB = makeContextWindowActivity("ctx-3", 3_000, "turn-b");
257 const tool = fixtures[0]!;
258
259 const projected = projectThreadDetailSnapshot({
260 snapshotSequence: 7,
261 thread: makeThread([stale1, tool, latestA, latestB]),
262 });
263
264 expect(projected.thread.activities.map((activity) => activity.id)).toEqual([
265 tool.id,
266 latestA.id,
267 latestB.id,
268 ]);
269 // The retained rows keep their payloads untouched — the tool-data
270 // projection only rewrites payloads with a `data` record.
271 expect(projected.thread.activities[2]?.payload).toEqual(latestB.payload);
272 });
273
274 it("still resolves a meter value after the client reverts the newest turn", () => {
275 // A live thread.reverted makes the client drop all activities from
276 // discarded turns; each surviving turn must keep a usable row.
277 const olderTurn = makeContextWindowActivity("ctx-old", 1_500, "turn-kept");
278 const revertedTurn = makeContextWindowActivity("ctx-new", 9_000, "turn-reverted");
279
280 const projected = projectThreadDetailSnapshot({
281 snapshotSequence: 7,
282 thread: makeThread([olderTurn, revertedTurn]),
283 });
284 const afterRevert = projected.thread.activities.filter(
285 (activity) => activity.turnId === TurnId.make("turn-kept"),
286 );
287
288 expect(deriveLatestContextWindowSnapshot(afterRevert)).toEqual(
289 deriveLatestContextWindowSnapshot([olderTurn]),
290 );
291 });
292
293 it("matches what the web client derives from the full history", () => {
294 const activities = [
295 makeContextWindowActivity("ctx-1", 1_000),
296 makeContextWindowActivity("ctx-2", 2_000),
297 ];
298 const projected = projectThreadDetailSnapshot({
299 snapshotSequence: 7,
300 thread: makeThread(activities),
301 });
302
303 expect(deriveLatestContextWindowSnapshot(projected.thread.activities)).toEqual(
304 deriveLatestContextWindowSnapshot(activities),
305 );
306 });
307
308 it("does not let a malformed row shadow an earlier valid row in the same turn", () => {
309 const valid = makeContextWindowActivity("ctx-valid", 5_000, "turn-a");
310 const malformed: OrchestrationThreadActivity = {
311 ...makeContextWindowActivity("ctx-broken", 0, "turn-a"),
312 payload: { usedTokens: null },
313 };
314
315 const projected = projectThreadDetailSnapshot({
316 snapshotSequence: 7,
317 thread: makeThread([valid, malformed]),
318 });
319
320 // The malformed row passes through, the valid row survives, and the
321 // client's backward walk resolves the same value as with full history.
322 expect(projected.thread.activities.map((activity) => activity.id)).toEqual([
323 valid.id,
324 malformed.id,
325 ]);
326 expect(deriveLatestContextWindowSnapshot(projected.thread.activities)).toEqual(
327 deriveLatestContextWindowSnapshot([valid, malformed]),
328 );
329 });
330
331 it("leaves snapshots without context-window activities untouched", () => {
332 const projected = projectThreadDetailSnapshot({
333 snapshotSequence: 7,
334 thread: makeThread([fixtures[4]!]),
335 });
336 expect(projected.thread.activities).toEqual([fixtures[4]]);
337 });
338
339 it("does not filter live activity-appended events", () => {
340 const activity = makeContextWindowActivity("ctx-live", 4_000);
341 const event = {
342 sequence: 9,
343 eventId: EventId.make("event-ctx"),
344 aggregateKind: "thread",
345 aggregateId: ThreadId.make("thread-projection"),
346 occurredAt: "2026-07-27T00:00:02.000Z",
347 commandId: null,
348 causationEventId: null,
349 correlationId: null,
350 metadata: {},
351 type: "thread.activity-appended",
352 payload: {
353 threadId: ThreadId.make("thread-projection"),
354 activity,
355 },
356 } satisfies Extract<OrchestrationEvent, { type: "thread.activity-appended" }>;
357
358 const projected = projectActivityEvent(event);
359 expect(
360 projected.type === "thread.activity-appended" ? projected.payload.activity : undefined,
361 ).toEqual(activity);
362 });
363});
packages/contracts/src/orchestration.ts-21
......@@ -26,7 +26,6 @@ export const ORCHESTRATION_WS_METHODS = {
2626 dispatchCommand: "orchestration.dispatchCommand",
2727 getTurnDiff: "orchestration.getTurnDiff",
2828 getFullThreadDiff: "orchestration.getFullThreadDiff",
29 replayEvents: "orchestration.replayEvents",
3029 getArchivedShellSnapshot: "orchestration.getArchivedShellSnapshot",
3130 subscribeShell: "orchestration.subscribeShell",
3231 subscribeThread: "orchestration.subscribeThread",
......@@ -1363,14 +1362,6 @@ export type OrchestrationGetFullThreadDiffInput = typeof OrchestrationGetFullThr
13631362export const OrchestrationGetFullThreadDiffResult = ThreadTurnDiff;
13641363export type OrchestrationGetFullThreadDiffResult = typeof OrchestrationGetFullThreadDiffResult.Type;
13651364
1366export const OrchestrationReplayEventsInput = Schema.Struct({
1367 fromSequenceExclusive: NonNegativeInt,
1368});
1369export type OrchestrationReplayEventsInput = typeof OrchestrationReplayEventsInput.Type;
1370
1371const OrchestrationReplayEventsResult = Schema.Array(OrchestrationEvent);
1372export type OrchestrationReplayEventsResult = typeof OrchestrationReplayEventsResult.Type;
1373
13741365export const OrchestrationRpcSchemas = {
13751366 dispatchCommand: {
13761367 input: ClientOrchestrationCommand,
......@@ -1384,10 +1375,6 @@ export const OrchestrationRpcSchemas = {
13841375 input: OrchestrationGetFullThreadDiffInput,
13851376 output: OrchestrationGetFullThreadDiffResult,
13861377 },
1387 replayEvents: {
1388 input: OrchestrationReplayEventsInput,
1389 output: OrchestrationReplayEventsResult,
1390 },
13911378 getArchivedShellSnapshot: {
13921379 input: Schema.Struct({}),
13931380 output: OrchestrationShellSnapshot,
......@@ -1433,11 +1420,3 @@ export class OrchestrationGetFullThreadDiffError extends Schema.TaggedErrorClass
14331420 cause: Schema.optional(Schema.Defect()),
14341421 },
14351422) {}
1436
1437export class OrchestrationReplayEventsError extends Schema.TaggedErrorClass<OrchestrationReplayEventsError>()(
1438 "OrchestrationReplayEventsError",
1439 {
1440 message: TrimmedNonEmptyString,
1441 cause: Schema.optional(Schema.Defect()),
1442 },
1443) {}
packages/contracts/src/rpc.ts-9
......@@ -54,8 +54,6 @@ import {
5454 OrchestrationGetSnapshotError,
5555 OrchestrationGetTurnDiffError,
5656 OrchestrationGetTurnDiffInput,
57 OrchestrationReplayEventsError,
58 OrchestrationReplayEventsInput,
5957 OrchestrationRpcSchemas,
6058} from "./orchestration.ts";
6159import { ProviderInstanceId } from "./providerInstance.ts";
......@@ -631,12 +629,6 @@ export const WsOrchestrationGetFullThreadDiffRpc = Rpc.make(
631629 },
632630);
633631
634export const WsOrchestrationReplayEventsRpc = Rpc.make(ORCHESTRATION_WS_METHODS.replayEvents, {
635 payload: OrchestrationReplayEventsInput,
636 success: OrchestrationRpcSchemas.replayEvents.output,
637 error: Schema.Union([OrchestrationReplayEventsError, EnvironmentAuthorizationError]),
638});
639
640632export const WsOrchestrationGetArchivedShellSnapshotRpc = Rpc.make(
641633 ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot,
642634 {
......@@ -765,7 +757,6 @@ export const WsRpcGroup = RpcGroup.make(
765757 WsOrchestrationDispatchCommandRpc,
766758 WsOrchestrationGetTurnDiffRpc,
767759 WsOrchestrationGetFullThreadDiffRpc,
768 WsOrchestrationReplayEventsRpc,
769760 WsOrchestrationGetArchivedShellSnapshotRpc,
770761 WsOrchestrationSubscribeShellRpc,
771762 WsOrchestrationSubscribeThreadRpc,