authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-01 14:41:03-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-01 14:41:03-07:00
logdd89d5ce1a483f29e0d5429439a60da7a2296a3f
tree586fc32f13d3dc1331e85c962e855927722b1674
parent9132c37c86cf5c382391dc65f4703ad407e8717e
signature Signed by PGP key B5690EEEBB952194

perf(server): add targeted projection queries (#1646)

Co-authored-by: codex <codex@users.noreply.github.com>

10 files changed, 754 insertions(+), 112 deletions(-)

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts+32-73
......@@ -1,83 +1,38 @@
1import {
2 CheckpointRef,
3 DEFAULT_PROVIDER_INTERACTION_MODE,
4 ProjectId,
5 ThreadId,
6 TurnId,
7 type OrchestrationReadModel,
8} from "@t3tools/contracts";
9import { Effect, Layer } from "effect";
1import { CheckpointRef, ProjectId, ThreadId, TurnId } from "@t3tools/contracts";
2import { Effect, Layer, Option } from "effect";
103import { describe, expect, it } from "vitest";
114
12import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
5import {
6 ProjectionSnapshotQuery,
7 type ProjectionThreadCheckpointContext,
8} from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
139import { checkpointRefForThreadTurn } from "../Utils.ts";
1410import { CheckpointDiffQueryLive } from "./CheckpointDiffQuery.ts";
1511import { CheckpointStore, type CheckpointStoreShape } from "../Services/CheckpointStore.ts";
1612import { CheckpointDiffQuery } from "../Services/CheckpointDiffQuery.ts";
1713
18function makeSnapshot(input: {
14function 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", () => {
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", () => {
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", () => {
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 {
44 type OrchestrationGetFullThreadDiffResult,
55 type OrchestrationGetTurnDiffResult as OrchestrationGetTurnDiffResultType,
66} from "@t3tools/contracts";
7import { Effect, Layer, Schema } from "effect";
7import { Effect, Layer, Option, Schema } from "effect";
88
99import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
1010import { CheckpointInvariantError, CheckpointUnavailableError } from "../Errors.ts";
11import { checkpointRefForThreadTurn, resolveThreadWorkspaceCwd } from "../Utils.ts";
11import { checkpointRefForThreadTurn } from "../Utils.ts";
1212import { CheckpointStore } from "../Services/CheckpointStore.ts";
1313import {
1414 CheckpointDiffQuery,
......@@ -41,16 +41,17 @@ const make = Effect.gen(function* () {
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* () {
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* () {
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* () {
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
......@@ -338,4 +338,290 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
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 {
77 OrchestrationProposedPlanId,
88 OrchestrationReadModel,
99 ProjectScript,
10 ThreadId,
1110 TurnId,
1211 type OrchestrationCheckpointSummary,
1312 type OrchestrationLatestTurn,
......@@ -18,8 +17,10 @@ import {
1817 type OrchestrationThread,
1918 type OrchestrationThreadActivity,
2019 ModelSelection,
20 ProjectId,
21 ThreadId,
2122} from "@t3tools/contracts";
22import { Effect, Layer, Schema, Struct } from "effect";
23import { Effect, Layer, Option, Schema, Struct } from "effect";
2324import * as SqlClient from "effect/unstable/sql/SqlClient";
2425import * as SqlSchema from "effect/unstable/sql/SqlSchema";
2526
......@@ -40,6 +41,8 @@ import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.t
4041import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
4142import {
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({
8689 sourceProposedPlanId: Schema.NullOr(OrchestrationProposedPlanId),
8790});
8891const ProjectionStateDbRowSchema = ProjectionState;
92const ProjectionCountsRowSchema = Schema.Struct({
93 projectCount: Schema.Number,
94 threadCount: Schema.Number,
95});
96const WorkspaceRootLookupInput = Schema.Struct({
97 workspaceRoot: Schema.String,
98});
99const ProjectIdLookupInput = Schema.Struct({
100 projectId: ProjectId,
101});
102const ThreadIdLookupInput = Schema.Struct({
103 threadId: ThreadId,
104});
105const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema;
106const ProjectionThreadIdLookupRowSchema = Schema.Struct({
107 threadId: ThreadId,
108});
109const ProjectionThreadCheckpointContextThreadRowSchema = Schema.Struct({
110 threadId: ThreadId,
111 projectId: ProjectId,
112 workspaceRoot: Schema.String,
113 worktreePath: Schema.NullOr(Schema.String),
114});
89115
90116const REQUIRED_SNAPSHOT_PROJECTORS = [
91117 ORCHESTRATION_PROJECTOR_NAMES.projects,
......@@ -319,6 +345,94 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
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* () {
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 @@
66 *
77 * @module ProjectionSnapshotQuery
88 */
9import type { OrchestrationReadModel } from "@t3tools/contracts";
9import type {
10 OrchestrationCheckpointSummary,
11 OrchestrationProject,
12 OrchestrationReadModel,
13 ProjectId,
14 ThreadId,
15} from "@t3tools/contracts";
1016import { ServiceMap } from "effect";
17import type { Option } from "effect";
1118import type { Effect } from "effect";
1219
1320import type { ProjectionRepositoryError } from "../../persistence/Errors.ts";
1421
22export interface ProjectionSnapshotCounts {
23 readonly projectCount: number;
24 readonly threadCount: number;
25}
26
27export 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 {
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
......@@ -31,6 +31,7 @@ import Migration0015 from "./Migrations/015_ProjectionTurnsSourceProposedPlan.ts
3131import Migration0016 from "./Migrations/016_CanonicalizeModelSelections.ts";
3232import Migration0017 from "./Migrations/017_ProjectionThreadsArchivedAt.ts";
3333import Migration0018 from "./Migrations/018_ProjectionThreadsArchivedAtIndex.ts";
34import Migration0019 from "./Migrations/019_ProjectionSnapshotLookupIndexes.ts";
3435
3536/**
3637 * Migration loader with all migrations defined inline.
......@@ -61,6 +62,7 @@ export const migrationEntries = [
6162 [16, "CanonicalizeModelSelections", Migration0016],
6263 [17, "ProjectionThreadsArchivedAt", Migration0017],
6364 [18, "ProjectionThreadsArchivedAtIndex", Migration0018],
65 [19, "ProjectionSnapshotLookupIndexes", Migration0019],
6466] as const;
6567
6668export const makeMigrationLoader = (throughId?: number) =>
apps/server/src/persistence/Migrations/019_ProjectionSnapshotLookupIndexes.test.ts created+73
......@@ -0,0 +1,73 @@
1import { assert, it } from "@effect/vitest";
2import { Effect, Layer } from "effect";
3import * as SqlClient from "effect/unstable/sql/SqlClient";
4
5import { runMigrations } from "../Migrations.ts";
6import * as NodeSqliteClient from "../NodeSqliteClient.ts";
7
8const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory()));
9
10layer("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 created+16
......@@ -0,0 +1,16 @@
1import * as Effect from "effect/Effect";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
3
4export 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.test.ts+37-4
......@@ -1,8 +1,13 @@
11import { assert, it } from "@effect/vitest";
2import { Deferred, Effect, Fiber, Ref } from "effect";
3import { TestClock } from "effect/testing";
2import { Deferred, Effect, Fiber, Option, Ref } from "effect";
43
5import { makeCommandGate, ServerRuntimeStartupError } from "./serverRuntimeStartup.ts";
4import { AnalyticsService } from "./telemetry/Services/AnalyticsService.ts";
5import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts";
6import {
7 launchStartupHeartbeat,
8 makeCommandGate,
9 ServerRuntimeStartupError,
10} from "./serverRuntimeStartup.ts";
611
712it.effect("enqueueCommand waits for readiness and then drains queued work", () =>
813 Effect.scoped(
......@@ -14,7 +19,7 @@ it.effect("enqueueCommand waits for readiness and then drains queued work", () =
1419 .enqueueCommand(Ref.updateAndGet(executionCount, (count) => count + 1))
1520 .pipe(Effect.forkScoped);
1621
17 yield* TestClock.adjust("50 millis");
22 yield* Effect.yieldNow;
1823 assert.equal(yield* Ref.get(executionCount), 0);
1924
2025 yield* commandGate.signalCommandReady;
......@@ -47,3 +52,31 @@ it.effect("enqueueCommand fails queued work when readiness fails", () =>
4752 }),
4853 ),
4954);
55
56it.effect("launchStartupHeartbeat does not block the caller while counts are loading", () =>
57 Effect.scoped(
58 Effect.gen(function* () {
59 const releaseCounts = yield* Deferred.make<void, never>();
60
61 yield* launchStartupHeartbeat.pipe(
62 Effect.provideService(ProjectionSnapshotQuery, {
63 getSnapshot: () => Effect.die("unused"),
64 getCounts: () =>
65 Deferred.await(releaseCounts).pipe(
66 Effect.as({
67 projectCount: 2,
68 threadCount: 3,
69 }),
70 ),
71 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
72 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
73 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
74 }),
75 Effect.provideService(AnalyticsService, {
76 record: () => Effect.void,
77 flush: Effect.void,
78 }),
79 );
80 }),
81 ),
82);
apps/server/src/serverRuntimeStartup.ts+34-20
......@@ -5,7 +5,19 @@ import {
55 ProjectId,
66 ThreadId,
77} from "@t3tools/contracts";
8import { Data, Deferred, Effect, Exit, Layer, Path, Queue, Ref, Scope, ServiceMap } from "effect";
8import {
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
1022import { ServerConfig } from "./config";
1123import { Keybindings } from "./keybindings";
......@@ -105,17 +117,15 @@ export const makeCommandGate = Effect.gen(function* () {
105117 } satisfies CommandGate;
106118});
107119
108const recordStartupHeartbeat = Effect.gen(function* () {
120export const recordStartupHeartbeat = Effect.gen(function* () {
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,
......@@ -130,6 +140,12 @@ const recordStartupHeartbeat = Effect.gen(function* () {
130140 });
131141});
132142
143export const launchStartupHeartbeat = recordStartupHeartbeat.pipe(
144 Effect.ignoreCause({ log: true }),
145 Effect.forkScoped,
146 Effect.asVoid,
147);
148
133149const autoBootstrapWelcome = Effect.gen(function* () {
134150 const serverConfig = yield* ServerConfig;
135151 const projectionReadModelQuery = yield* ProjectionSnapshotQuery;
......@@ -141,14 +157,13 @@ const autoBootstrapWelcome = Effect.gen(function* () {
141157
142158 if (serverConfig.autoBootstrapProjectFromCwd) {
143159 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,
160 const existingProject = yield* projectionReadModelQuery.getActiveProjectByWorkspaceRoot(
161 serverConfig.cwd,
147162 );
148163 let nextProjectId: ProjectId;
149164 let nextProjectDefaultModelSelection: ModelSelection;
150165
151 if (!existingProject) {
166 if (Option.isNone(existingProject)) {
152167 const createdAt = new Date().toISOString();
153168 nextProjectId = ProjectId.makeUnsafe(crypto.randomUUID());
154169 const bootstrapProjectTitle = path.basename(serverConfig.cwd) || "project";
......@@ -166,17 +181,16 @@ const autoBootstrapWelcome = Effect.gen(function* () {
166181 createdAt,
167182 });
168183 } else {
169 nextProjectId = existingProject.id;
170 nextProjectDefaultModelSelection = existingProject.defaultModelSelection ?? {
184 nextProjectId = existingProject.value.id;
185 nextProjectDefaultModelSelection = existingProject.value.defaultModelSelection ?? {
171186 provider: "codex",
172187 model: "gpt-5-codex",
173188 };
174189 }
175190
176 const existingThread = snapshot.threads.find(
177 (thread) => thread.projectId === nextProjectId && thread.deletedAt === null,
178 );
179 if (!existingThread) {
191 const existingThreadId =
192 yield* projectionReadModelQuery.getFirstActiveThreadIdByProjectId(nextProjectId);
193 if (Option.isNone(existingThreadId)) {
180194 const createdAt = new Date().toISOString();
181195 const createdThreadId = ThreadId.makeUnsafe(crypto.randomUUID());
182196 yield* orchestrationEngine.dispatch({
......@@ -196,7 +210,7 @@ const autoBootstrapWelcome = Effect.gen(function* () {
196210 bootstrapThreadId = createdThreadId;
197211 } else {
198212 bootstrapProjectId = nextProjectId;
199 bootstrapThreadId = existingThread.id;
213 bootstrapThreadId = existingThreadId.value;
200214 }
201215 });
202216 }
......@@ -314,7 +328,7 @@ const makeServerRuntimeStartup = Effect.gen(function* () {
314328 });
315329
316330 yield* Effect.logDebug("startup phase: recording startup heartbeat");
317 yield* recordStartupHeartbeat;
331 yield* launchStartupHeartbeat;
318332 yield* Effect.logDebug("startup phase: browser open check");
319333 yield* maybeOpenBrowser;
320334 yield* Effect.logDebug("startup phase: complete");