History

perf(server): add targeted projection queries

Julius Marminge committed 7 months ago

Commit c296cee1 parent 9132c379 files · +709 −106Browse files

Unified · Split

9 files
  1. apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts +32 −73
  2. apps/server/src/checkpointing/Layers/CheckpointDiffQuery.ts +10 −12
  3. apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +286 −0
  4. apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +217 −2
  5. apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +47 −1
  6. apps/server/src/persistence/Migrations.ts +2 −0
  7. apps/server/src/persistence/Migrations/019_ProjectionSnapshotLookupIndexes.test.ts +73 −0
  8. apps/server/src/persistence/Migrations/019_ProjectionSnapshotLookupIndexes.ts +16 −0
  9. apps/server/src/serverRuntimeStartup.ts +26 −18
apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts+32−73
@@ -1,83 +1,38 @@
1−import {
2− CheckpointRef,
3− DEFAULT_PROVIDER_INTERACTION_MODE,
4− ProjectId,
5− ThreadId,
6− TurnId,
7− type OrchestrationReadModel,
8−} from "@t3tools/contracts";
9−import { Effect, Layer } from "effect";
1+import { CheckpointRef, ProjectId, ThreadId, TurnId } from "@t3tools/contracts";
2+import { Effect, Layer, Option } from "effect";
103 import { describe, expect, it } from "vitest";
114
12−import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
5+import {
6+ ProjectionSnapshotQuery,
7+ type ProjectionThreadCheckpointContext,
8+} from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
139 import { checkpointRefForThreadTurn } from "../Utils.ts";
1410 import { CheckpointDiffQueryLive } from "./CheckpointDiffQuery.ts";
1511 import { CheckpointStore, type CheckpointStoreShape } from "../Services/CheckpointStore.ts";
1612 import { CheckpointDiffQuery } from "../Services/CheckpointDiffQuery.ts";
1713
18−function makeSnapshot(input: {
14+function makeThreadCheckpointContext(input: {
1915 readonly projectId: ProjectId;
2016 readonly threadId: ThreadId;
2117 readonly workspaceRoot: string;
2218 readonly worktreePath: string | null;
2319 readonly checkpointTurnCount: number;
2420 readonly checkpointRef: CheckpointRef;
25−}): OrchestrationReadModel {
21+}): ProjectionThreadCheckpointContext {
2622 return {
27− snapshotSequence: 0,
28− updatedAt: "2026-01-01T00:00:00.000Z",
29− projects: [
30− {
31− id: input.projectId,
32− title: "Project",
33− workspaceRoot: input.workspaceRoot,
34− defaultModelSelection: null,
35− scripts: [],
36− createdAt: "2026-01-01T00:00:00.000Z",
37− updatedAt: "2026-01-01T00:00:00.000Z",
38− deletedAt: null,
39− },
40− ],
41− threads: [
23+ threadId: input.threadId,
24+ projectId: input.projectId,
25+ workspaceRoot: input.workspaceRoot,
26+ worktreePath: input.worktreePath,
27+ checkpoints: [
4228 {
43− id: input.threadId,
44− projectId: input.projectId,
45− title: "Thread",
46− modelSelection: {
47− provider: "codex",
48− model: "gpt-5-codex",
49− },
50− interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
51− runtimeMode: "full-access",
52− branch: null,
53− worktreePath: input.worktreePath,
54− latestTurn: {
55− turnId: TurnId.makeUnsafe("turn-1"),
56− state: "completed",
57− requestedAt: "2026-01-01T00:00:00.000Z",
58− startedAt: "2026-01-01T00:00:00.000Z",
59− completedAt: "2026-01-01T00:00:00.000Z",
60− assistantMessageId: null,
61− },
62− createdAt: "2026-01-01T00:00:00.000Z",
63− updatedAt: "2026-01-01T00:00:00.000Z",
64− archivedAt: null,
65− deletedAt: null,
66− messages: [],
67− activities: [],
68− proposedPlans: [],
69− checkpoints: [
70− {
71− turnId: TurnId.makeUnsafe("turn-1"),
72− checkpointTurnCount: input.checkpointTurnCount,
73− checkpointRef: input.checkpointRef,
74− status: "ready",
75− files: [],
76− assistantMessageId: null,
77− completedAt: "2026-01-01T00:00:00.000Z",
78− },
79− ],
80− session: null,
29+ turnId: TurnId.makeUnsafe("turn-1"),
30+ checkpointTurnCount: input.checkpointTurnCount,
31+ checkpointRef: input.checkpointRef,
32+ status: "ready",
33+ files: [],
34+ assistantMessageId: null,
35+ completedAt: "2026-01-01T00:00:00.000Z",
8136 },
8237 ],
8338 };
@@ -95,7 +50,7 @@ describe("CheckpointDiffQueryLive", () => { · 11 unmodified lines
9550 readonly cwd: string;
9651 }> = [];
9752
98− const snapshot = makeSnapshot({
53+ const threadCheckpointContext = makeThreadCheckpointContext({
9954 projectId,
10055 threadId,
10156 workspaceRoot: "/tmp/workspace",
@@ -125,7 +80,12 @@ describe("CheckpointDiffQueryLive", () => { · 23 unmodified lines
12580 Layer.provideMerge(Layer.succeed(CheckpointStore, checkpointStore)),
12681 Layer.provideMerge(
12782 Layer.succeed(ProjectionSnapshotQuery, {
128− getSnapshot: () => Effect.succeed(snapshot),
83+ getSnapshot: () =>
84+ Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
85+ getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
86+ getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
87+ getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
88+ getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
12989 }),
13090 ),
13191 );
@@ -175,12 +135,11 @@ describe("CheckpointDiffQueryLive", () => { · 43 unmodified lines
175135 Layer.provideMerge(
176136 Layer.succeed(ProjectionSnapshotQuery, {
177137 getSnapshot: () =>
178− Effect.succeed({
179− snapshotSequence: 0,
180− projects: [],
181− threads: [],
182− updatedAt: "2026-01-01T00:00:00.000Z",
183− } satisfies OrchestrationReadModel),
138+ Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
139+ getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
140+ getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
141+ getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
142+ getThreadCheckpointContext: () => Effect.succeed(Option.none()),
184143 }),
185144 ),
186145 );
apps/server/src/checkpointing/Layers/CheckpointDiffQuery.ts+10−12
@@ -4,11 +4,11 @@ import { · 3 unmodified lines
44 type OrchestrationGetFullThreadDiffResult,
55 type OrchestrationGetTurnDiffResult as OrchestrationGetTurnDiffResultType,
66 } from "@t3tools/contracts";
7−import { Effect, Layer, Schema } from "effect";
7+import { Effect, Layer, Option, Schema } from "effect";
88
99 import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
1010 import { CheckpointInvariantError, CheckpointUnavailableError } from "../Errors.ts";
11−import { checkpointRefForThreadTurn, resolveThreadWorkspaceCwd } from "../Utils.ts";
11+import { checkpointRefForThreadTurn } from "../Utils.ts";
1212 import { CheckpointStore } from "../Services/CheckpointStore.ts";
1313 import {
1414 CheckpointDiffQuery,
@@ -41,16 +41,17 @@ const make = Effect.gen(function* () { · 26 unmodified lines
4141 return emptyDiff;
4242 }
4343
44− const snapshot = yield* projectionSnapshotQuery.getSnapshot();
45− const thread = snapshot.threads.find((entry) => entry.id === input.threadId);
46− if (!thread) {
44+ const threadContext = yield* projectionSnapshotQuery.getThreadCheckpointContext(
45+ input.threadId,
46+ );
47+ if (Option.isNone(threadContext)) {
4748 return yield* new CheckpointInvariantError({
4849 operation,
4950 detail: `Thread '${input.threadId}' not found.`,
5051 });
5152 }
5253
53− const maxTurnCount = thread.checkpoints.reduce(
54+ const maxTurnCount = threadContext.value.checkpoints.reduce(
5455 (max, checkpoint) => Math.max(max, checkpoint.checkpointTurnCount),
5556 0,
5657 );
@@ -62,10 +63,7 @@ const make = Effect.gen(function* () { · 5 unmodified lines
6263 });
6364 }
6465
65− const workspaceCwd = resolveThreadWorkspaceCwd({
66− thread,
67− projects: snapshot.projects,
68− });
66+ const workspaceCwd = threadContext.value.worktreePath ?? threadContext.value.workspaceRoot;
6967 if (!workspaceCwd) {
7068 return yield* new CheckpointInvariantError({
7169 operation,
@@ -76,7 +74,7 @@ const make = Effect.gen(function* () { · 4 unmodified lines
7674 const fromCheckpointRef =
7775 input.fromTurnCount === 0
7876 ? checkpointRefForThreadTurn(input.threadId, 0)
79− : thread.checkpoints.find(
77+ : threadContext.value.checkpoints.find(
8078 (checkpoint) => checkpoint.checkpointTurnCount === input.fromTurnCount,
8179 )?.checkpointRef;
8280 if (!fromCheckpointRef) {
@@ -87,7 +85,7 @@ const make = Effect.gen(function* () { · 4 unmodified lines
8785 });
8886 }
8987
90− const toCheckpointRef = thread.checkpoints.find(
88+ const toCheckpointRef = threadContext.value.checkpoints.find(
9189 (checkpoint) => checkpoint.checkpointTurnCount === input.toTurnCount,
9290 )?.checkpointRef;
9391 if (!toCheckpointRef) {
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+286−0
@@ -338,4 +338,290 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { · 337 unmodified lines
338338 ]);
339339 }),
340340 );
341+
342+ it.effect(
343+ "reads targeted project, thread, and count queries without hydrating the full snapshot",
344+ () =>
345+ Effect.gen(function* () {
346+ const snapshotQuery = yield* ProjectionSnapshotQuery;
347+ const sql = yield* SqlClient.SqlClient;
348+
349+ yield* sql`DELETE FROM projection_projects`;
350+ yield* sql`DELETE FROM projection_threads`;
351+ yield* sql`DELETE FROM projection_turns`;
352+
353+ yield* sql`
354+ INSERT INTO projection_projects (
355+ project_id,
356+ title,
357+ workspace_root,
358+ default_model_selection_json,
359+ scripts_json,
360+ created_at,
361+ updated_at,
362+ deleted_at
363+ )
364+ VALUES
365+ (
366+ 'project-active',
367+ 'Active Project',
368+ '/tmp/workspace',
369+ '{"provider":"codex","model":"gpt-5-codex"}',
370+ '[]',
371+ '2026-03-01T00:00:00.000Z',
372+ '2026-03-01T00:00:01.000Z',
373+ NULL
374+ ),
375+ (
376+ 'project-deleted',
377+ 'Deleted Project',
378+ '/tmp/deleted',
379+ NULL,
380+ '[]',
381+ '2026-03-01T00:00:02.000Z',
382+ '2026-03-01T00:00:03.000Z',
383+ '2026-03-01T00:00:04.000Z'
384+ )
385+ `;
386+
387+ yield* sql`
388+ INSERT INTO projection_threads (
389+ thread_id,
390+ project_id,
391+ title,
392+ model_selection_json,
393+ runtime_mode,
394+ interaction_mode,
395+ branch,
396+ worktree_path,
397+ latest_turn_id,
398+ created_at,
399+ updated_at,
400+ archived_at,
401+ deleted_at
402+ )
403+ VALUES
404+ (
405+ 'thread-first',
406+ 'project-active',
407+ 'First Thread',
408+ '{"provider":"codex","model":"gpt-5-codex"}',
409+ 'full-access',
410+ 'default',
411+ NULL,
412+ NULL,
413+ NULL,
414+ '2026-03-01T00:00:05.000Z',
415+ '2026-03-01T00:00:06.000Z',
416+ NULL,
417+ NULL
418+ ),
419+ (
420+ 'thread-second',
421+ 'project-active',
422+ 'Second Thread',
423+ '{"provider":"codex","model":"gpt-5-codex"}',
424+ 'full-access',
425+ 'default',
426+ NULL,
427+ NULL,
428+ NULL,
429+ '2026-03-01T00:00:07.000Z',
430+ '2026-03-01T00:00:08.000Z',
431+ NULL,
432+ NULL
433+ ),
434+ (
435+ 'thread-deleted',
436+ 'project-active',
437+ 'Deleted Thread',
438+ '{"provider":"codex","model":"gpt-5-codex"}',
439+ 'full-access',
440+ 'default',
441+ NULL,
442+ NULL,
443+ NULL,
444+ '2026-03-01T00:00:09.000Z',
445+ '2026-03-01T00:00:10.000Z',
446+ NULL,
447+ '2026-03-01T00:00:11.000Z'
448+ )
449+ `;
450+
451+ const counts = yield* snapshotQuery.getCounts();
452+ assert.deepEqual(counts, {
453+ projectCount: 2,
454+ threadCount: 3,
455+ });
456+
457+ const project = yield* snapshotQuery.getActiveProjectByWorkspaceRoot("/tmp/workspace");
458+ assert.equal(project._tag, "Some");
459+ if (project._tag === "Some") {
460+ assert.equal(project.value.id, asProjectId("project-active"));
461+ }
462+
463+ const missingProject = yield* snapshotQuery.getActiveProjectByWorkspaceRoot("/tmp/missing");
464+ assert.equal(missingProject._tag, "None");
465+
466+ const firstThreadId = yield* snapshotQuery.getFirstActiveThreadIdByProjectId(
467+ asProjectId("project-active"),
468+ );
469+ assert.equal(firstThreadId._tag, "Some");
470+ if (firstThreadId._tag === "Some") {
471+ assert.equal(firstThreadId.value, ThreadId.makeUnsafe("thread-first"));
472+ }
473+ }),
474+ );
475+
476+ it.effect("reads single-thread checkpoint context without hydrating unrelated threads", () =>
477+ Effect.gen(function* () {
478+ const snapshotQuery = yield* ProjectionSnapshotQuery;
479+ const sql = yield* SqlClient.SqlClient;
480+
481+ yield* sql`DELETE FROM projection_projects`;
482+ yield* sql`DELETE FROM projection_threads`;
483+ yield* sql`DELETE FROM projection_turns`;
484+
485+ yield* sql`
486+ INSERT INTO projection_projects (
487+ project_id,
488+ title,
489+ workspace_root,
490+ default_model_selection_json,
491+ scripts_json,
492+ created_at,
493+ updated_at,
494+ deleted_at
495+ )
496+ VALUES (
497+ 'project-context',
498+ 'Context Project',
499+ '/tmp/context-workspace',
500+ NULL,
501+ '[]',
502+ '2026-03-02T00:00:00.000Z',
503+ '2026-03-02T00:00:01.000Z',
504+ NULL
505+ )
506+ `;
507+
508+ yield* sql`
509+ INSERT INTO projection_threads (
510+ thread_id,
511+ project_id,
512+ title,
513+ model_selection_json,
514+ runtime_mode,
515+ interaction_mode,
516+ branch,
517+ worktree_path,
518+ latest_turn_id,
519+ created_at,
520+ updated_at,
521+ archived_at,
522+ deleted_at
523+ )
524+ VALUES (
525+ 'thread-context',
526+ 'project-context',
527+ 'Context Thread',
528+ '{"provider":"codex","model":"gpt-5-codex"}',
529+ 'full-access',
530+ 'default',
531+ 'feature/perf',
532+ '/tmp/context-worktree',
533+ NULL,
534+ '2026-03-02T00:00:02.000Z',
535+ '2026-03-02T00:00:03.000Z',
536+ NULL,
537+ NULL
538+ )
539+ `;
540+
541+ yield* sql`
542+ INSERT INTO projection_turns (
543+ thread_id,
544+ turn_id,
545+ pending_message_id,
546+ source_proposed_plan_thread_id,
547+ source_proposed_plan_id,
548+ assistant_message_id,
549+ state,
550+ requested_at,
551+ started_at,
552+ completed_at,
553+ checkpoint_turn_count,
554+ checkpoint_ref,
555+ checkpoint_status,
556+ checkpoint_files_json
557+ )
558+ VALUES
559+ (
560+ 'thread-context',
561+ 'turn-1',
562+ NULL,
563+ NULL,
564+ NULL,
565+ NULL,
566+ 'completed',
567+ '2026-03-02T00:00:04.000Z',
568+ '2026-03-02T00:00:04.000Z',
569+ '2026-03-02T00:00:04.000Z',
570+ 1,
571+ 'checkpoint-a',
572+ 'ready',
573+ '[]'
574+ ),
575+ (
576+ 'thread-context',
577+ 'turn-2',
578+ NULL,
579+ NULL,
580+ NULL,
581+ NULL,
582+ 'completed',
583+ '2026-03-02T00:00:05.000Z',
584+ '2026-03-02T00:00:05.000Z',
585+ '2026-03-02T00:00:05.000Z',
586+ 2,
587+ 'checkpoint-b',
588+ 'ready',
589+ '[]'
590+ )
591+ `;
592+
593+ const context = yield* snapshotQuery.getThreadCheckpointContext(
594+ ThreadId.makeUnsafe("thread-context"),
595+ );
596+ assert.equal(context._tag, "Some");
597+ if (context._tag === "Some") {
598+ assert.deepEqual(context.value, {
599+ threadId: ThreadId.makeUnsafe("thread-context"),
600+ projectId: asProjectId("project-context"),
601+ workspaceRoot: "/tmp/context-workspace",
602+ worktreePath: "/tmp/context-worktree",
603+ checkpoints: [
604+ {
605+ turnId: asTurnId("turn-1"),
606+ checkpointTurnCount: 1,
607+ checkpointRef: asCheckpointRef("checkpoint-a"),
608+ status: "ready",
609+ files: [],
610+ assistantMessageId: null,
611+ completedAt: "2026-03-02T00:00:04.000Z",
612+ },
613+ {
614+ turnId: asTurnId("turn-2"),
615+ checkpointTurnCount: 2,
616+ checkpointRef: asCheckpointRef("checkpoint-b"),
617+ status: "ready",
618+ files: [],
619+ assistantMessageId: null,
620+ completedAt: "2026-03-02T00:00:05.000Z",
621+ },
622+ ],
623+ });
624+ }
625+ }),
626+ );
341627 });
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+217−2
@@ -7,7 +7,6 @@ import { · 6 unmodified lines
77 OrchestrationProposedPlanId,
88 OrchestrationReadModel,
99 ProjectScript,
10− ThreadId,
1110 TurnId,
1211 type OrchestrationCheckpointSummary,
1312 type OrchestrationLatestTurn,
@@ -18,8 +17,10 @@ import { · 4 unmodified lines
1817 type OrchestrationThread,
1918 type OrchestrationThreadActivity,
2019 ModelSelection,
20+ ProjectId,
21+ ThreadId,
2122 } from "@t3tools/contracts";
22−import { Effect, Layer, Schema, Struct } from "effect";
23+import { Effect, Layer, Option, Schema, Struct } from "effect";
2324 import * as SqlClient from "effect/unstable/sql/SqlClient";
2425 import * as SqlSchema from "effect/unstable/sql/SqlSchema";
2526
@@ -40,6 +41,8 @@ import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.t · 14 unmodified lines
4041 import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
4142 import {
4243 ProjectionSnapshotQuery,
44+ type ProjectionSnapshotCounts,
45+ type ProjectionThreadCheckpointContext,
4346 type ProjectionSnapshotQueryShape,
4447 } from "../Services/ProjectionSnapshotQuery.ts";
4548
@@ -86,6 +89,29 @@ const ProjectionLatestTurnDbRowSchema = Schema.Struct({ · 40 unmodified lines
8689 sourceProposedPlanId: Schema.NullOr(OrchestrationProposedPlanId),
8790 });
8891 const ProjectionStateDbRowSchema = ProjectionState;
92+const ProjectionCountsRowSchema = Schema.Struct({
93+ projectCount: Schema.Number,
94+ threadCount: Schema.Number,
95+});
96+const WorkspaceRootLookupInput = Schema.Struct({
97+ workspaceRoot: Schema.String,
98+});
99+const ProjectIdLookupInput = Schema.Struct({
100+ projectId: ProjectId,
101+});
102+const ThreadIdLookupInput = Schema.Struct({
103+ threadId: ThreadId,
104+});
105+const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema;
106+const ProjectionThreadIdLookupRowSchema = Schema.Struct({
107+ threadId: ThreadId,
108+});
109+const ProjectionThreadCheckpointContextThreadRowSchema = Schema.Struct({
110+ threadId: ThreadId,
111+ projectId: ProjectId,
112+ workspaceRoot: Schema.String,
113+ worktreePath: Schema.NullOr(Schema.String),
114+});
89115
90116 const REQUIRED_SNAPSHOT_PROJECTORS = [
91117 ORCHESTRATION_PROJECTOR_NAMES.projects,
@@ -319,6 +345,94 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 227 unmodified lines
319345 `,
320346 });
321347
348+ const readProjectionCounts = SqlSchema.findOne({
349+ Request: Schema.Void,
350+ Result: ProjectionCountsRowSchema,
351+ execute: () =>
352+ sql`
353+ SELECT
354+ (SELECT COUNT(*) FROM projection_projects) AS "projectCount",
355+ (SELECT COUNT(*) FROM projection_threads) AS "threadCount"
356+ `,
357+ });
358+
359+ const getActiveProjectRowByWorkspaceRoot = SqlSchema.findOneOption({
360+ Request: WorkspaceRootLookupInput,
361+ Result: ProjectionProjectLookupRowSchema,
362+ execute: ({ workspaceRoot }) =>
363+ sql`
364+ SELECT
365+ project_id AS "projectId",
366+ title,
367+ workspace_root AS "workspaceRoot",
368+ default_model_selection_json AS "defaultModelSelection",
369+ scripts_json AS "scripts",
370+ created_at AS "createdAt",
371+ updated_at AS "updatedAt",
372+ deleted_at AS "deletedAt"
373+ FROM projection_projects
374+ WHERE workspace_root = ${workspaceRoot}
375+ AND deleted_at IS NULL
376+ ORDER BY created_at ASC, project_id ASC
377+ LIMIT 1
378+ `,
379+ });
380+
381+ const getFirstActiveThreadIdByProject = SqlSchema.findOneOption({
382+ Request: ProjectIdLookupInput,
383+ Result: ProjectionThreadIdLookupRowSchema,
384+ execute: ({ projectId }) =>
385+ sql`
386+ SELECT
387+ thread_id AS "threadId"
388+ FROM projection_threads
389+ WHERE project_id = ${projectId}
390+ AND deleted_at IS NULL
391+ ORDER BY created_at ASC, thread_id ASC
392+ LIMIT 1
393+ `,
394+ });
395+
396+ const getThreadCheckpointContextThreadRow = SqlSchema.findOneOption({
397+ Request: ThreadIdLookupInput,
398+ Result: ProjectionThreadCheckpointContextThreadRowSchema,
399+ execute: ({ threadId }) =>
400+ sql`
401+ SELECT
402+ threads.thread_id AS "threadId",
403+ threads.project_id AS "projectId",
404+ projects.workspace_root AS "workspaceRoot",
405+ threads.worktree_path AS "worktreePath"
406+ FROM projection_threads AS threads
407+ INNER JOIN projection_projects AS projects
408+ ON projects.project_id = threads.project_id
409+ WHERE threads.thread_id = ${threadId}
410+ AND threads.deleted_at IS NULL
411+ LIMIT 1
412+ `,
413+ });
414+
415+ const listCheckpointRowsByThread = SqlSchema.findAll({
416+ Request: ThreadIdLookupInput,
417+ Result: ProjectionCheckpointDbRowSchema,
418+ execute: ({ threadId }) =>
419+ sql`
420+ SELECT
421+ thread_id AS "threadId",
422+ turn_id AS "turnId",
423+ checkpoint_turn_count AS "checkpointTurnCount",
424+ checkpoint_ref AS "checkpointRef",
425+ checkpoint_status AS "status",
426+ checkpoint_files_json AS "files",
427+ assistant_message_id AS "assistantMessageId",
428+ completed_at AS "completedAt"
429+ FROM projection_turns
430+ WHERE thread_id = ${threadId}
431+ AND checkpoint_turn_count IS NOT NULL
432+ ORDER BY checkpoint_turn_count ASC
433+ `,
434+ });
435+
322436 const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () =>
323437 sql
324438 .withTransaction(
@@ -593,8 +707,109 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 268 unmodified lines
593707 }),
594708 );
595709
710+ const getCounts: ProjectionSnapshotQueryShape["getCounts"] = () =>
711+ readProjectionCounts(undefined).pipe(
712+ Effect.mapError(
713+ toPersistenceSqlOrDecodeError(
714+ "ProjectionSnapshotQuery.getCounts:query",
715+ "ProjectionSnapshotQuery.getCounts:decodeRow",
716+ ),
717+ ),
718+ Effect.map(
719+ (row): ProjectionSnapshotCounts => ({
720+ projectCount: row.projectCount,
721+ threadCount: row.threadCount,
722+ }),
723+ ),
724+ );
725+
726+ const getActiveProjectByWorkspaceRoot: ProjectionSnapshotQueryShape["getActiveProjectByWorkspaceRoot"] =
727+ (workspaceRoot) =>
728+ getActiveProjectRowByWorkspaceRoot({ workspaceRoot }).pipe(
729+ Effect.mapError(
730+ toPersistenceSqlOrDecodeError(
731+ "ProjectionSnapshotQuery.getActiveProjectByWorkspaceRoot:query",
732+ "ProjectionSnapshotQuery.getActiveProjectByWorkspaceRoot:decodeRow",
733+ ),
734+ ),
735+ Effect.map(
736+ Option.map(
737+ (row): OrchestrationProject => ({
738+ id: row.projectId,
739+ title: row.title,
740+ workspaceRoot: row.workspaceRoot,
741+ defaultModelSelection: row.defaultModelSelection,
742+ scripts: row.scripts,
743+ createdAt: row.createdAt,
744+ updatedAt: row.updatedAt,
745+ deletedAt: row.deletedAt,
746+ }),
747+ ),
748+ ),
749+ );
750+
751+ const getFirstActiveThreadIdByProjectId: ProjectionSnapshotQueryShape["getFirstActiveThreadIdByProjectId"] =
752+ (projectId) =>
753+ getFirstActiveThreadIdByProject({ projectId }).pipe(
754+ Effect.mapError(
755+ toPersistenceSqlOrDecodeError(
756+ "ProjectionSnapshotQuery.getFirstActiveThreadIdByProjectId:query",
757+ "ProjectionSnapshotQuery.getFirstActiveThreadIdByProjectId:decodeRow",
758+ ),
759+ ),
760+ Effect.map(Option.map((row) => row.threadId)),
761+ );
762+
763+ const getThreadCheckpointContext: ProjectionSnapshotQueryShape["getThreadCheckpointContext"] = (
764+ threadId,
765+ ) =>
766+ Effect.gen(function* () {
767+ const threadRow = yield* getThreadCheckpointContextThreadRow({ threadId }).pipe(
768+ Effect.mapError(
769+ toPersistenceSqlOrDecodeError(
770+ "ProjectionSnapshotQuery.getThreadCheckpointContext:getThread:query",
771+ "ProjectionSnapshotQuery.getThreadCheckpointContext:getThread:decodeRow",
772+ ),
773+ ),
774+ );
775+ if (Option.isNone(threadRow)) {
776+ return Option.none<ProjectionThreadCheckpointContext>();
777+ }
778+
779+ const checkpointRows = yield* listCheckpointRowsByThread({ threadId }).pipe(
780+ Effect.mapError(
781+ toPersistenceSqlOrDecodeError(
782+ "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:query",
783+ "ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:decodeRows",
784+ ),
785+ ),
786+ );
787+
788+ return Option.some({
789+ threadId: threadRow.value.threadId,
790+ projectId: threadRow.value.projectId,
791+ workspaceRoot: threadRow.value.workspaceRoot,
792+ worktreePath: threadRow.value.worktreePath,
793+ checkpoints: checkpointRows.map(
794+ (row): OrchestrationCheckpointSummary => ({
795+ turnId: row.turnId,
796+ checkpointTurnCount: row.checkpointTurnCount,
797+ checkpointRef: row.checkpointRef,
798+ status: row.status,
799+ files: row.files,
800+ assistantMessageId: row.assistantMessageId,
801+ completedAt: row.completedAt,
802+ }),
803+ ),
804+ });
805+ });
806+
596807 return {
597808 getSnapshot,
809+ getCounts,
810+ getActiveProjectByWorkspaceRoot,
811+ getFirstActiveThreadIdByProjectId,
812+ getThreadCheckpointContext,
598813 } satisfies ProjectionSnapshotQueryShape;
599814 });
600815
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+47−1
@@ -6,12 +6,32 @@ · 5 unmodified lines
66 *
77 * @module ProjectionSnapshotQuery
88 */
9−import type { OrchestrationReadModel } from "@t3tools/contracts";
9+import type {
10+ OrchestrationCheckpointSummary,
11+ OrchestrationProject,
12+ OrchestrationReadModel,
13+ ProjectId,
14+ ThreadId,
15+} from "@t3tools/contracts";
1016 import { ServiceMap } from "effect";
17+import type { Option } from "effect";
1118 import type { Effect } from "effect";
1219
1320 import type { ProjectionRepositoryError } from "../../persistence/Errors.ts";
1421
22+export interface ProjectionSnapshotCounts {
23+ readonly projectCount: number;
24+ readonly threadCount: number;
25+}
26+
27+export interface ProjectionThreadCheckpointContext {
28+ readonly threadId: ThreadId;
29+ readonly projectId: ProjectId;
30+ readonly workspaceRoot: string;
31+ readonly worktreePath: string | null;
32+ readonly checkpoints: ReadonlyArray<OrchestrationCheckpointSummary>;
33+}
34+
1535 /**
1636 * ProjectionSnapshotQueryShape - Service API for read-model snapshots.
1737 */
@@ -23,6 +43,32 @@ export interface ProjectionSnapshotQueryShape { · 5 unmodified lines
2343 * projector cursor state.
2444 */
2545 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel, ProjectionRepositoryError>;
46+
47+ /**
48+ * Read aggregate projection counts without hydrating the full read model.
49+ */
50+ readonly getCounts: () => Effect.Effect<ProjectionSnapshotCounts, ProjectionRepositoryError>;
51+
52+ /**
53+ * Read the active project for an exact workspace root match.
54+ */
55+ readonly getActiveProjectByWorkspaceRoot: (
56+ workspaceRoot: string,
57+ ) => Effect.Effect<Option.Option<OrchestrationProject>, ProjectionRepositoryError>;
58+
59+ /**
60+ * Read the earliest active thread for a project.
61+ */
62+ readonly getFirstActiveThreadIdByProjectId: (
63+ projectId: ProjectId,
64+ ) => Effect.Effect<Option.Option<ThreadId>, ProjectionRepositoryError>;
65+
66+ /**
67+ * Read the checkpoint context needed to resolve a single thread diff.
68+ */
69+ readonly getThreadCheckpointContext: (
70+ threadId: ThreadId,
71+ ) => Effect.Effect<Option.Option<ProjectionThreadCheckpointContext>, ProjectionRepositoryError>;
2672 }
2773
2874 /**
apps/server/src/persistence/Migrations.ts+2−0
@@ -31,6 +31,7 @@ import Migration0015 from "./Migrations/015_ProjectionTurnsSourceProposedPlan.ts · 30 unmodified lines
3131 import Migration0016 from "./Migrations/016_CanonicalizeModelSelections.ts";
3232 import Migration0017 from "./Migrations/017_ProjectionThreadsArchivedAt.ts";
3333 import Migration0018 from "./Migrations/018_ProjectionThreadsArchivedAtIndex.ts";
34+import Migration0019 from "./Migrations/019_ProjectionSnapshotLookupIndexes.ts";
3435
3536 /**
3637 * Migration loader with all migrations defined inline.
@@ -61,6 +62,7 @@ export const migrationEntries = [ · 24 unmodified lines
6162 [16, "CanonicalizeModelSelections", Migration0016],
6263 [17, "ProjectionThreadsArchivedAt", Migration0017],
6364 [18, "ProjectionThreadsArchivedAtIndex", Migration0018],
65+ [19, "ProjectionSnapshotLookupIndexes", Migration0019],
6466 ] as const;
6567
6668 export const makeMigrationLoader = (throughId?: number) =>
apps/server/src/persistence/Migrations/019_ProjectionSnapshotLookupIndexes.test.ts+73−0
@@ -0,0 +1,73 @@
1+import { assert, it } from "@effect/vitest";
2+import { Effect, Layer } from "effect";
3+import * as SqlClient from "effect/unstable/sql/SqlClient";
4+
5+import { runMigrations } from "../Migrations.ts";
6+import * as NodeSqliteClient from "../NodeSqliteClient.ts";
7+
8+const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory()));
9+
10+layer("019_ProjectionSnapshotLookupIndexes", (it) => {
11+ it.effect("creates indexes for targeted projection lookup filters", () =>
12+ Effect.gen(function* () {
13+ const sql = yield* SqlClient.SqlClient;
14+
15+ yield* runMigrations({ toMigrationInclusive: 18 });
16+ yield* runMigrations({ toMigrationInclusive: 19 });
17+
18+ const projectIndexes = yield* sql<{
19+ readonly seq: number;
20+ readonly name: string;
21+ readonly unique: number;
22+ readonly origin: string;
23+ readonly partial: number;
24+ }>`
25+ PRAGMA index_list(projection_projects)
26+ `;
27+ assert.ok(
28+ projectIndexes.some(
29+ (index) => index.name === "idx_projection_projects_workspace_root_deleted_at",
30+ ),
31+ );
32+
33+ const projectIndexColumns = yield* sql<{
34+ readonly seqno: number;
35+ readonly cid: number;
36+ readonly name: string;
37+ }>`
38+ PRAGMA index_info('idx_projection_projects_workspace_root_deleted_at')
39+ `;
40+ assert.deepStrictEqual(
41+ projectIndexColumns.map((column) => column.name),
42+ ["workspace_root", "deleted_at"],
43+ );
44+
45+ const threadIndexes = yield* sql<{
46+ readonly seq: number;
47+ readonly name: string;
48+ readonly unique: number;
49+ readonly origin: string;
50+ readonly partial: number;
51+ }>`
52+ PRAGMA index_list(projection_threads)
53+ `;
54+ assert.ok(
55+ threadIndexes.some(
56+ (index) => index.name === "idx_projection_threads_project_deleted_created",
57+ ),
58+ );
59+
60+ const threadIndexColumns = yield* sql<{
61+ readonly seqno: number;
62+ readonly cid: number;
63+ readonly name: string;
64+ }>`
65+ PRAGMA index_info('idx_projection_threads_project_deleted_created')
66+ `;
67+ assert.deepStrictEqual(
68+ threadIndexColumns.map((column) => column.name),
69+ ["project_id", "deleted_at", "created_at"],
70+ );
71+ }),
72+ );
73+});
apps/server/src/persistence/Migrations/019_ProjectionSnapshotLookupIndexes.ts+16−0
@@ -0,0 +1,16 @@
1+import * as Effect from "effect/Effect";
2+import * as SqlClient from "effect/unstable/sql/SqlClient";
3+
4+export default Effect.gen(function* () {
5+ const sql = yield* SqlClient.SqlClient;
6+
7+ yield* sql`
8+ CREATE INDEX IF NOT EXISTS idx_projection_projects_workspace_root_deleted_at
9+ ON projection_projects(workspace_root, deleted_at)
10+ `;
11+
12+ yield* sql`
13+ CREATE INDEX IF NOT EXISTS idx_projection_threads_project_deleted_created
14+ ON projection_threads(project_id, deleted_at, created_at)
15+ `;
16+});
apps/server/src/serverRuntimeStartup.ts+26−18
@@ -5,7 +5,19 @@ import { · 4 unmodified lines
55 ProjectId,
66 ThreadId,
77 } from "@t3tools/contracts";
8−import { Data, Deferred, Effect, Exit, Layer, Path, Queue, Ref, Scope, ServiceMap } from "effect";
8+import {
9+ Data,
10+ Deferred,
11+ Effect,
12+ Exit,
13+ Layer,
14+ Option,
15+ Path,
16+ Queue,
17+ Ref,
18+ Scope,
19+ ServiceMap,
20+} from "effect";
921
1022 import { ServerConfig } from "./config";
1123 import { Keybindings } from "./keybindings";
@@ -109,13 +121,11 @@ const recordStartupHeartbeat = Effect.gen(function* () { · 97 unmodified lines
109121 const analytics = yield* AnalyticsService;
110122 const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
111123
112− const { threadCount, projectCount } = yield* projectionSnapshotQuery.getSnapshot().pipe(
113− Effect.map((snapshot) => ({
114− threadCount: snapshot.threads.length,
115− projectCount: snapshot.projects.length,
116− })),
124+ const { threadCount, projectCount } = yield* projectionSnapshotQuery.getCounts().pipe(
117125 Effect.catch((cause) =>
118− Effect.logWarning("failed to gather startup snapshot for telemetry", { cause }).pipe(
126+ Effect.logWarning("failed to gather startup projection counts for telemetry", {
127+ cause,
128+ }).pipe(
119129 Effect.as({
120130 threadCount: 0,
121131 projectCount: 0,
@@ -141,14 +151,13 @@ const autoBootstrapWelcome = Effect.gen(function* () { · 19 unmodified lines
141151
142152 if (serverConfig.autoBootstrapProjectFromCwd) {
143153 yield* Effect.gen(function* () {
144− const snapshot = yield* projectionReadModelQuery.getSnapshot();
145− const existingProject = snapshot.projects.find(
146− (project) => project.workspaceRoot === serverConfig.cwd && project.deletedAt === null,
154+ const existingProject = yield* projectionReadModelQuery.getActiveProjectByWorkspaceRoot(
155+ serverConfig.cwd,
147156 );
148157 let nextProjectId: ProjectId;
149158 let nextProjectDefaultModelSelection: ModelSelection;
150159
151− if (!existingProject) {
160+ if (Option.isNone(existingProject)) {
152161 const createdAt = new Date().toISOString();
153162 nextProjectId = ProjectId.makeUnsafe(crypto.randomUUID());
154163 const bootstrapProjectTitle = path.basename(serverConfig.cwd) || "project";
@@ -166,17 +175,16 @@ const autoBootstrapWelcome = Effect.gen(function* () { · 11 unmodified lines
166175 createdAt,
167176 });
168177 } else {
169− nextProjectId = existingProject.id;
170− nextProjectDefaultModelSelection = existingProject.defaultModelSelection ?? {
178+ nextProjectId = existingProject.value.id;
179+ nextProjectDefaultModelSelection = existingProject.value.defaultModelSelection ?? {
171180 provider: "codex",
172181 model: "gpt-5-codex",
173182 };
174183 }
175184
176− const existingThread = snapshot.threads.find(
177− (thread) => thread.projectId === nextProjectId && thread.deletedAt === null,
178− );
179− if (!existingThread) {
185+ const existingThreadId =
186+ yield* projectionReadModelQuery.getFirstActiveThreadIdByProjectId(nextProjectId);
187+ if (Option.isNone(existingThreadId)) {
180188 const createdAt = new Date().toISOString();
181189 const createdThreadId = ThreadId.makeUnsafe(crypto.randomUUID());
182190 yield* orchestrationEngine.dispatch({
@@ -196,7 +204,7 @@ const autoBootstrapWelcome = Effect.gen(function* () { · 13 unmodified lines
196204 bootstrapThreadId = createdThreadId;
197205 } else {
198206 bootstrapProjectId = nextProjectId;
199− bootstrapThreadId = existingThread.id;
207+ bootstrapThreadId = existingThreadId.value;
200208 }
201209 });
202210 }