authorgravatar for thevictorvogel@gmail.comonly21mil <thevictorvogel@gmail.com> 2026-10-10 10:37:40-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 10:37:40-05:00
loga11f464133291122e0f6381b35e1b245d0aa9c73
treed0d4b918a694cc9359c03e6d49ba02d19e49349c
parentafd98d7d1840ed76f57d3ece556257aebc00f1bc
signature Signed by PGP key B5690EEEBB952194

perf(server): t3_thread_list reads only the listed project's threads (#17843)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

3 files changed, 197 insertions(+), 9 deletions(-)

apps/server/src/orchestration-v2/ProjectionStore.test.ts+180
......@@ -349,6 +349,157 @@ it.effect("memory recovery selection includes unfinished items from missing runs
349349 }).pipe(Effect.provide(ProjectionStore.layerMemory)),
350350);
351351
352// Project A holds an active thread, an archived thread and a fork of a run in
353// an archived project B thread, so a project-scoped snapshot still has to load
354// a fork source that is outside the project and archived.
355const assertProjectScopedShells = Effect.fn("assertProjectScopedShells")(function* (
356 suffix: string,
357) {
358 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
359 const now = yield* DateTime.now;
360 const projectId = ProjectId.make(`project:${suffix}:a`);
361 const otherProjectId = ProjectId.make(`project:${suffix}:b`);
362 const activeThreadId = ThreadId.make(`thread:${suffix}:a-active`);
363 const archivedThreadId = ThreadId.make(`thread:${suffix}:a-archived`);
364 const forkThreadId = ThreadId.make(`thread:${suffix}:a-fork`);
365 const sourceThreadId = ThreadId.make(`thread:${suffix}:b-source`);
366 const otherThreadId = ThreadId.make(`thread:${suffix}:b-other`);
367 const sourceRunId = RunId.make(`run:${suffix}:b-source`);
368 const sourceNodeId = NodeId.make(`node:${suffix}:b-source`);
369 const createThread = (input: {
370 readonly threadId: ThreadId;
371 readonly projectId: ProjectId;
372 readonly archived?: boolean;
373 readonly forkedFromRunOf?: ThreadId;
374 }) =>
375 projectionStore.apply({
376 id: EventId.make(`event:${input.threadId}:created`),
377 type: "thread.created",
378 threadId: input.threadId,
379 occurredAt: now,
380 payload: {
381 createdBy: "user",
382 creationSource: "web",
383 id: input.threadId,
384 projectId: input.projectId,
385 title: input.threadId,
386 providerInstanceId,
387 modelSelection,
388 runtimeMode: "full-access",
389 interactionMode: "default",
390 branch: null,
391 worktreePath: null,
392 activeProviderThreadId: null,
393 lineage:
394 input.forkedFromRunOf === undefined
395 ? { parentThreadId: null, relationshipToParent: null, rootThreadId: input.threadId }
396 : {
397 parentThreadId: input.forkedFromRunOf,
398 relationshipToParent: "fork",
399 rootThreadId: input.forkedFromRunOf,
400 },
401 forkedFrom:
402 input.forkedFromRunOf === undefined
403 ? null
404 : { type: "run", threadId: input.forkedFromRunOf, runId: sourceRunId },
405 createdAt: now,
406 updatedAt: now,
407 archivedAt: input.archived === true ? now : null,
408 settledOverride: null,
409 settledAt: null,
410 lastVisitedAt: null,
411 deletedAt: null,
412 },
413 });
414
415 yield* createThread({ threadId: sourceThreadId, projectId: otherProjectId, archived: true });
416 yield* createThread({ threadId: otherThreadId, projectId: otherProjectId });
417 yield* createThread({ threadId: activeThreadId, projectId });
418 yield* createThread({ threadId: archivedThreadId, projectId, archived: true });
419 yield* createThread({ threadId: forkThreadId, projectId, forkedFromRunOf: sourceThreadId });
420 yield* projectionStore.apply({
421 id: EventId.make(`event:${suffix}:source-run`),
422 type: "run.created",
423 threadId: sourceThreadId,
424 runId: sourceRunId,
425 nodeId: sourceNodeId,
426 driver,
427 occurredAt: now,
428 payload: {
429 id: sourceRunId,
430 threadId: sourceThreadId,
431 ordinal: 1,
432 providerInstanceId,
433 modelSelection,
434 providerThreadId: null,
435 userMessageId: MessageId.make(`message:${suffix}:source-user`),
436 rootNodeId: sourceNodeId,
437 activeAttemptId: null,
438 status: "completed",
439 requestedAt: now,
440 startedAt: now,
441 completedAt: now,
442 checkpointId: null,
443 contextHandoffId: null,
444 },
445 });
446 yield* projectionStore.apply({
447 id: EventId.make(`event:${suffix}:source-item`),
448 type: "turn-item.updated",
449 threadId: sourceThreadId,
450 runId: sourceRunId,
451 occurredAt: now,
452 payload: {
453 id: TurnItemId.make(`turn-item:${suffix}:source`),
454 threadId: sourceThreadId,
455 runId: sourceRunId,
456 nodeId: null,
457 providerThreadId: null,
458 providerTurnId: null,
459 nativeItemRef: null,
460 parentItemId: null,
461 ordinal: 1,
462 status: "completed",
463 title: null,
464 startedAt: now,
465 completedAt: now,
466 updatedAt: now,
467 type: "assistant_message",
468 messageId: MessageId.make(`message:${suffix}:source-assistant`),
469 text: "Source answer",
470 streaming: false,
471 },
472 });
473
474 const unscoped = yield* projectionStore.getShellSnapshot();
475 const inProject = (thread: { readonly projectId: ProjectId }) => thread.projectId === projectId;
476 const active = yield* projectionStore.getShellSnapshot({ projectId, location: "active" });
477 assert.deepEqual(active.threads, unscoped.threads.filter(inProject));
478 assert.deepEqual(active.archivedThreads, []);
479 assert.sameMembers(
480 active.threads.map((thread) => thread.id),
481 [activeThreadId, forkThreadId],
482 );
483 const fork = active.threads.find((thread) => thread.id === forkThreadId);
484 assert.isAbove(fork?.visibleItemCount ?? 0, 0);
485
486 const all = yield* projectionStore.getShellSnapshot({ projectId });
487 assert.deepEqual(all.threads, active.threads);
488 assert.deepEqual(all.archivedThreads, unscoped.archivedThreads.filter(inProject));
489 assert.sameMembers(
490 all.archivedThreads.map((thread) => thread.id),
491 [archivedThreadId],
492 );
493 return { projectId, activeThreadId, forkThreadId, otherThreadId };
494});
495
496it.effect("memory shell snapshots scope to one project", () =>
497 assertProjectScopedShells("memory-project-shell").pipe(
498 Effect.asVoid,
499 Effect.provide(ProjectionStore.layerMemory),
500 ),
501);
502
352503it.layer(layerTest)("ProjectionStoreV2", (it) => {
353504 it.effect(
354505 "keeps restart-cancelled work through a stale run.updated",
......@@ -1884,6 +2035,35 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
18842035 }),
18852036 );
18862037
2038 it.effect("reads only the requested project's threads into a scoped shell snapshot", () =>
2039 Effect.gen(function* () {
2040 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2041 const sql = yield* SqlClient.SqlClient;
2042 const { projectId, activeThreadId, forkThreadId, otherThreadId } =
2043 yield* assertProjectScopedShells("sql-project-shell");
2044
2045 // A thread in another project that cannot decode fails the unscoped
2046 // snapshot, so a scoped snapshot that succeeds never read it.
2047 const [stored] = yield* sql<{ readonly payload_json: string }>`
2048 SELECT payload_json FROM orchestration_v2_projection_threads
2049 WHERE thread_id = ${otherThreadId}
2050 `;
2051 const setPayload = (payload: string) =>
2052 sql`UPDATE orchestration_v2_projection_threads SET payload_json = ${payload}
2053 WHERE thread_id = ${otherThreadId}`;
2054 yield* setPayload("{}");
2055 const [unscoped, scoped] = yield* Effect.all([
2056 projectionStore.getShellSnapshot().pipe(Effect.flip),
2057 projectionStore.getShellSnapshot({ projectId, location: "active" }),
2058 ]).pipe(Effect.ensuring(Effect.orDie(setPayload(stored!.payload_json))));
2059 assert.strictEqual(unscoped._tag, "ProjectionStoreReadError");
2060 assert.sameMembers(
2061 scoped.threads.map((thread) => thread.id),
2062 [activeThreadId, forkThreadId],
2063 );
2064 }),
2065 );
2066
18872067 it.effect("does not treat visited or marked-unread state as thread activity", () =>
18882068 Effect.gen(function* () {
18892069 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
apps/server/src/orchestration-v2/ProjectionStore.ts+16-7
......@@ -31,6 +31,7 @@ import type {
3131 OrchestrationV2ThreadShell,
3232 OrchestrationV2ThreadProjection,
3333 OrchestrationV2TurnItem,
34 ProjectId,
3435 ProviderInstanceId,
3536 ProviderSessionId,
3637 ProviderThreadId,
......@@ -325,6 +326,11 @@ export interface ProjectionTimelinePage {
325326
326327export interface ShellSnapshotOptions {
327328 readonly location?: "active" | "archive";
329 /**
330 * Reads only this project's threads. Fork sources in other projects still
331 * load, so visible item counts match the unscoped snapshot.
332 */
333 readonly projectId?: ProjectId;
328334 /**
329335 * For background sweeps, not clients: skips settled threads before any of
330336 * their run, item or session rows are read.
......@@ -5148,8 +5154,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
51485154
51495155 const selectShellThreadRows = (
51505156 threadId?: ThreadId,
5151 location?: "active" | "archive",
5152 unsettledOnly = false,
5157 { location, projectId, unsettledOnly = false }: ShellSnapshotOptions = {},
51535158 ) =>
51545159 sql<ShellThreadRow>`
51555160 SELECT
......@@ -5325,6 +5330,8 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
53255330 LIMIT 1
53265331 ) AND blocked.status = 'failed'
53275332 WHERE t.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND t.thread_id = ${threadId}`}${
5333 projectId === undefined ? sql`` : sql` AND t.project_id = ${projectId}`
5334 }${
53285335 location === "active"
53295336 ? sql` AND json_extract(t.payload_json, '$.archivedAt') IS NULL`
53305337 : location === "archive"
......@@ -5748,11 +5755,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
57485755
57495756 const readShellSnapshotRows = (options: ShellSnapshotOptions | undefined) =>
57505757 Effect.gen(function* () {
5751 const targetThreadRows = yield* selectShellThreadRows(
5752 undefined,
5753 options?.location,
5754 options?.unsettledOnly ?? false,
5755 );
5758 const targetThreadRows = yield* selectShellThreadRows(undefined, options);
57565759 const targetThreadIds = new Set(
57575760 targetThreadRows.map((row) => ThreadId.make(row.thread_id)),
57585761 );
......@@ -6008,6 +6011,12 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
60086011 const existing = (yield* Ref.get(replayState)).projections;
60096012 const selectedThreadIds = [...existing.entries()]
60106013 .filter(([, projection]) => {
6014 if (
6015 options?.projectId !== undefined &&
6016 projection.thread.projectId !== options.projectId
6017 ) {
6018 return false;
6019 }
60116020 if (
60126021 options?.unsettledOnly &&
60136022 (projection.thread.settledAt !== null ||
apps/server/src/orchestration-v2/ThreadManagementService.ts+1-2
......@@ -563,7 +563,7 @@ const make = Effect.gen(function* () {
563563 );
564564
565565 const listProjectThreads: ThreadManagementServiceShape["listProjectThreads"] = (input) =>
566 orchestrator.getShellSnapshot().pipe(
566 orchestrator.getShellSnapshot({ projectId: input.projectId, location: "active" }).pipe(
567567 Effect.mapError(
568568 (cause) =>
569569 new ThreadManagementProjectThreadsListError({
......@@ -573,7 +573,6 @@ const make = Effect.gen(function* () {
573573 ),
574574 Effect.map((snapshot) =>
575575 snapshot.threads
576 .filter((thread) => thread.projectId === input.projectId)
577576 .filter(
578577 (thread) =>
579578 input.includeSubagents || thread.lineage.relationshipToParent !== "subagent",