authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-26 10:12:42-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log1d975ad4310389b8e22bbce3d02a746854a11471
treeeab564bfb5a130cbe536667918d7229191ca4976
parent8ab72e0964d95ca0ef139fba9994e3fb925343c9

migrate to single turn store

18 files changed, 1030 insertions(+), 600 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+145-3
...@@ -412,6 +412,146 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -412,6 +412,146 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
412 }),412 }),
413 );413 );
414414
415 it.effect("resolves turn-count conflicts when checkpoint completion rewrites provisional turns", () =>
416 Effect.gen(function* () {
417 const projectionPipeline = yield* OrchestrationProjectionPipeline;
418 const eventStore = yield* OrchestrationEventStore;
419 const sql = yield* SqlClient.SqlClient;
420 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
421 eventStore
422 .append(event)
423 .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
424
425 yield* appendAndProject({
426 type: "project.created",
427 eventId: EventId.makeUnsafe("evt-conflict-1"),
428 aggregateKind: "project",
429 aggregateId: ProjectId.makeUnsafe("project-conflict"),
430 occurredAt: "2026-02-26T13:00:00.000Z",
431 commandId: CommandId.makeUnsafe("cmd-conflict-1"),
432 causationEventId: null,
433 correlationId: CorrelationId.makeUnsafe("cmd-conflict-1"),
434 metadata: {},
435 payload: {
436 projectId: ProjectId.makeUnsafe("project-conflict"),
437 title: "Project Conflict",
438 workspaceRoot: "/tmp/project-conflict",
439 defaultModel: null,
440 scripts: [],
441 createdAt: "2026-02-26T13:00:00.000Z",
442 updatedAt: "2026-02-26T13:00:00.000Z",
443 },
444 });
445
446 yield* appendAndProject({
447 type: "thread.created",
448 eventId: EventId.makeUnsafe("evt-conflict-2"),
449 aggregateKind: "thread",
450 aggregateId: ThreadId.makeUnsafe("thread-conflict"),
451 occurredAt: "2026-02-26T13:00:01.000Z",
452 commandId: CommandId.makeUnsafe("cmd-conflict-2"),
453 causationEventId: null,
454 correlationId: CorrelationId.makeUnsafe("cmd-conflict-2"),
455 metadata: {},
456 payload: {
457 threadId: ThreadId.makeUnsafe("thread-conflict"),
458 projectId: ProjectId.makeUnsafe("project-conflict"),
459 title: "Thread Conflict",
460 model: "gpt-5-codex",
461 branch: null,
462 worktreePath: null,
463 createdAt: "2026-02-26T13:00:01.000Z",
464 updatedAt: "2026-02-26T13:00:01.000Z",
465 },
466 });
467
468 yield* appendAndProject({
469 type: "thread.turn-interrupt-requested",
470 eventId: EventId.makeUnsafe("evt-conflict-3"),
471 aggregateKind: "thread",
472 aggregateId: ThreadId.makeUnsafe("thread-conflict"),
473 occurredAt: "2026-02-26T13:00:02.000Z",
474 commandId: CommandId.makeUnsafe("cmd-conflict-3"),
475 causationEventId: null,
476 correlationId: CorrelationId.makeUnsafe("cmd-conflict-3"),
477 metadata: {},
478 payload: {
479 threadId: ThreadId.makeUnsafe("thread-conflict"),
480 turnId: TurnId.makeUnsafe("turn-interrupted"),
481 createdAt: "2026-02-26T13:00:02.000Z",
482 },
483 });
484
485 yield* appendAndProject({
486 type: "thread.message-sent",
487 eventId: EventId.makeUnsafe("evt-conflict-4"),
488 aggregateKind: "thread",
489 aggregateId: ThreadId.makeUnsafe("thread-conflict"),
490 occurredAt: "2026-02-26T13:00:03.000Z",
491 commandId: CommandId.makeUnsafe("cmd-conflict-4"),
492 causationEventId: null,
493 correlationId: CorrelationId.makeUnsafe("cmd-conflict-4"),
494 metadata: {},
495 payload: {
496 threadId: ThreadId.makeUnsafe("thread-conflict"),
497 messageId: MessageId.makeUnsafe("assistant-conflict"),
498 role: "assistant",
499 text: "done",
500 turnId: TurnId.makeUnsafe("turn-completed"),
501 streaming: false,
502 createdAt: "2026-02-26T13:00:03.000Z",
503 updatedAt: "2026-02-26T13:00:03.000Z",
504 },
505 });
506
507 yield* appendAndProject({
508 type: "thread.turn-diff-completed",
509 eventId: EventId.makeUnsafe("evt-conflict-5"),
510 aggregateKind: "thread",
511 aggregateId: ThreadId.makeUnsafe("thread-conflict"),
512 occurredAt: "2026-02-26T13:00:04.000Z",
513 commandId: CommandId.makeUnsafe("cmd-conflict-5"),
514 causationEventId: null,
515 correlationId: CorrelationId.makeUnsafe("cmd-conflict-5"),
516 metadata: {},
517 payload: {
518 threadId: ThreadId.makeUnsafe("thread-conflict"),
519 turnId: TurnId.makeUnsafe("turn-completed"),
520 checkpointTurnCount: 1,
521 checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-conflict/turn/1"),
522 status: "ready",
523 files: [],
524 assistantMessageId: MessageId.makeUnsafe("assistant-conflict"),
525 completedAt: "2026-02-26T13:00:04.000Z",
526 },
527 });
528
529 const turnRows = yield* sql<{
530 readonly turnId: string;
531 readonly checkpointTurnCount: number | null;
532 readonly status: string;
533 }>`
534 SELECT
535 turn_id AS "turnId",
536 checkpoint_turn_count AS "checkpointTurnCount",
537 state AS "status"
538 FROM projection_turns
539 WHERE thread_id = 'thread-conflict'
540 ORDER BY
541 CASE
542 WHEN checkpoint_turn_count IS NULL THEN 1
543 ELSE 0
544 END ASC,
545 checkpoint_turn_count ASC,
546 requested_at ASC
547 `;
548 assert.deepEqual(turnRows, [
549 { turnId: "turn-completed", checkpointTurnCount: 1, status: "completed" },
550 { turnId: "turn-interrupted", checkpointTurnCount: null, status: "interrupted" },
551 ]);
552 }),
553 );
554
415 it.effect("does not fallback-retain messages whose turnId is removed by revert", () =>555 it.effect("does not fallback-retain messages whose turnId is removed by revert", () =>
416 Effect.gen(function* () {556 Effect.gen(function* () {
417 const projectionPipeline = yield* OrchestrationProjectionPipeline;557 const projectionPipeline = yield* OrchestrationProjectionPipeline;
...@@ -697,8 +837,10 @@ it.effect("restores pending turn-start metadata across projection pipeline resta...@@ -697,8 +837,10 @@ it.effect("restores pending turn-start metadata across projection pipeline resta
697837
698 const pendingRows = yield* sql<{ readonly threadId: string }>`838 const pendingRows = yield* sql<{ readonly threadId: string }>`
699 SELECT thread_id AS "threadId"839 SELECT thread_id AS "threadId"
700 FROM projection_pending_turn_starts840 FROM projection_turns
701 WHERE thread_id = ${threadId}841 WHERE thread_id = ${threadId}
842 AND turn_id IS NULL
843 AND state = 'pending'
702 `;844 `;
703 assert.deepEqual(pendingRows, []);845 assert.deepEqual(pendingRows, []);
704846
...@@ -709,9 +851,9 @@ it.effect("restores pending turn-start metadata across projection pipeline resta...@@ -709,9 +851,9 @@ it.effect("restores pending turn-start metadata across projection pipeline resta
709 }>`851 }>`
710 SELECT852 SELECT
711 turn_id AS "turnId",853 turn_id AS "turnId",
712 user_message_id AS "userMessageId",854 pending_message_id AS "userMessageId",
713 started_at AS "startedAt"855 started_at AS "startedAt"
714 FROM projection_thread_turns856 FROM projection_turns
715 WHERE turn_id = ${turnId}857 WHERE turn_id = ${turnId}
716 `;858 `;
717 }).pipe(Effect.provide(secondProjectionLayer));859 }).pipe(Effect.provide(secondProjectionLayer));
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+132-119
...@@ -4,9 +4,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";...@@ -4,9 +4,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";
44
5import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts";5import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts";
6import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";6import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
7import { ProjectionCheckpointRepository } from "../../persistence/Services/ProjectionCheckpoints.ts";
8import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts";7import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts";
9import { ProjectionPendingTurnStartRepository } from "../../persistence/Services/ProjectionPendingTurnStarts.ts";
10import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts";8import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts";
11import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";9import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";
12import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";10import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";
...@@ -16,18 +14,15 @@ import {...@@ -16,18 +14,15 @@ import {
16 ProjectionThreadMessageRepository,14 ProjectionThreadMessageRepository,
17} from "../../persistence/Services/ProjectionThreadMessages.ts";15} from "../../persistence/Services/ProjectionThreadMessages.ts";
18import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts";16import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts";
19import { type ProjectionThreadTurn } from "../../persistence/Services/ProjectionThreadTurns.ts";17import { type ProjectionTurn, ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts";
20import { ProjectionThreadTurnRepository } from "../../persistence/Services/ProjectionThreadTurns.ts";
21import { ProjectionThreadRepository } from "../../persistence/Services/ProjectionThreads.ts";18import { ProjectionThreadRepository } from "../../persistence/Services/ProjectionThreads.ts";
22import { ProjectionCheckpointRepositoryLive } from "../../persistence/Layers/ProjectionCheckpoints.ts";
23import { ProjectionPendingApprovalRepositoryLive } from "../../persistence/Layers/ProjectionPendingApprovals.ts";19import { ProjectionPendingApprovalRepositoryLive } from "../../persistence/Layers/ProjectionPendingApprovals.ts";
24import { ProjectionPendingTurnStartRepositoryLive } from "../../persistence/Layers/ProjectionPendingTurnStarts.ts";
25import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/ProjectionProjects.ts";20import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/ProjectionProjects.ts";
26import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts";21import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts";
27import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";22import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";
28import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts";23import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts";
29import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts";24import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts";
30import { ProjectionThreadTurnRepositoryLive } from "../../persistence/Layers/ProjectionThreadTurns.ts";25import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";
31import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts";26import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts";
32import {27import {
33 OrchestrationProjectionPipeline,28 OrchestrationProjectionPipeline,
...@@ -61,22 +56,22 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {...@@ -61,22 +56,22 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
61 return typeof requestId === "string" ? ApprovalRequestId.makeUnsafe(requestId) : null;56 return typeof requestId === "string" ? ApprovalRequestId.makeUnsafe(requestId) : null;
62}57}
6358
64function nextTurnCount(turns: ReadonlyArray<ProjectionThreadTurn>): number {
65 return turns.reduce((maxTurnCount, turn) => Math.max(maxTurnCount, turn.turnCount), 0) + 1;
66}
67
68function retainProjectionMessagesAfterRevert(59function retainProjectionMessagesAfterRevert(
69 messages: ReadonlyArray<ProjectionThreadMessage>,60 messages: ReadonlyArray<ProjectionThreadMessage>,
70 turns: ReadonlyArray<ProjectionThreadTurn>,61 turns: ReadonlyArray<ProjectionTurn>,
71 turnCount: number,62 turnCount: number,
72): ReadonlyArray<ProjectionThreadMessage> {63): ReadonlyArray<ProjectionThreadMessage> {
73 const retainedMessageIds = new Set<string>();64 const retainedMessageIds = new Set<string>();
74 const retainedTurnIds = new Set<string>();65 const retainedTurnIds = new Set<string>();
75 const keptTurns = turns.filter((turn) => turn.turnCount <= turnCount);66 const keptTurns = turns.filter(
67 (turn) => turn.turnId !== null && turn.checkpointTurnCount !== null && turn.checkpointTurnCount <= turnCount,
68 );
76 for (const turn of keptTurns) {69 for (const turn of keptTurns) {
77 retainedTurnIds.add(turn.turnId);70 if (turn.turnId !== null) {
78 if (turn.userMessageId !== null) {71 retainedTurnIds.add(turn.turnId);
79 retainedMessageIds.add(turn.userMessageId);72 }
73 if (turn.pendingMessageId !== null) {
74 retainedMessageIds.add(turn.pendingMessageId);
80 }75 }
81 if (turn.assistantMessageId !== null) {76 if (turn.assistantMessageId !== null) {
82 retainedMessageIds.add(turn.assistantMessageId);77 retainedMessageIds.add(turn.assistantMessageId);
...@@ -144,11 +139,18 @@ function retainProjectionMessagesAfterRevert(...@@ -144,11 +139,18 @@ function retainProjectionMessagesAfterRevert(
144139
145function retainProjectionActivitiesAfterRevert(140function retainProjectionActivitiesAfterRevert(
146 activities: ReadonlyArray<ProjectionThreadActivity>,141 activities: ReadonlyArray<ProjectionThreadActivity>,
147 turns: ReadonlyArray<ProjectionThreadTurn>,142 turns: ReadonlyArray<ProjectionTurn>,
148 turnCount: number,143 turnCount: number,
149): ReadonlyArray<ProjectionThreadActivity> {144): ReadonlyArray<ProjectionThreadActivity> {
150 const retainedTurnIds = new Set(145 const retainedTurnIds = new Set<string>(
151 turns.filter((turn) => turn.turnCount <= turnCount).map((turn) => turn.turnId),146 turns
147 .filter(
148 (turn) =>
149 turn.turnId !== null &&
150 turn.checkpointTurnCount !== null &&
151 turn.checkpointTurnCount <= turnCount,
152 )
153 .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])),
152 );154 );
153 return activities.filter(155 return activities.filter(
154 (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId),156 (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId),
...@@ -164,10 +166,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -164,10 +166,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
164 const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository;166 const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository;
165 const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;167 const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;
166 const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository;168 const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository;
167 const projectionThreadTurnRepository = yield* ProjectionThreadTurnRepository;169 const projectionTurnRepository = yield* ProjectionTurnRepository;
168 const projectionCheckpointRepository = yield* ProjectionCheckpointRepository;
169 const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository;170 const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository;
170 const projectionPendingTurnStartRepository = yield* ProjectionPendingTurnStartRepository;
171171
172 const applyProjectsProjection: ProjectorDefinition["apply"] = (event) =>172 const applyProjectsProjection: ProjectorDefinition["apply"] = (event) =>
173 Effect.gen(function* () {173 Effect.gen(function* () {
...@@ -382,7 +382,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -382,7 +382,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
382 return;382 return;
383 }383 }
384384
385 const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({385 const existingTurns = yield* projectionTurnRepository.listByThreadId({
386 threadId: event.payload.threadId,386 threadId: event.payload.threadId,
387 });387 });
388 const keptRows = retainProjectionMessagesAfterRevert(388 const keptRows = retainProjectionMessagesAfterRevert(
...@@ -431,7 +431,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -431,7 +431,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
431 if (existingRows.length === 0) {431 if (existingRows.length === 0) {
432 return;432 return;
433 }433 }
434 const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({434 const existingTurns = yield* projectionTurnRepository.listByThreadId({
435 threadId: event.payload.threadId,435 threadId: event.payload.threadId,
436 });436 });
437 const keptRows = retainProjectionActivitiesAfterRevert(437 const keptRows = retainProjectionActivitiesAfterRevert(
...@@ -477,10 +477,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -477,10 +477,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
477 Effect.gen(function* () {477 Effect.gen(function* () {
478 switch (event.type) {478 switch (event.type) {
479 case "thread.turn-start-requested": {479 case "thread.turn-start-requested": {
480 yield* projectionPendingTurnStartRepository.upsert({480 yield* projectionTurnRepository.replacePendingTurnStart({
481 threadId: event.payload.threadId,481 threadId: event.payload.threadId,
482 messageId: event.payload.messageId,482 messageId: event.payload.messageId,
483 createdAt: event.payload.createdAt,483 requestedAt: event.payload.createdAt,
484 });484 });
485 return;485 return;
486 }486 }
...@@ -491,38 +491,53 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -491,38 +491,53 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
491 return;491 return;
492 }492 }
493493
494 const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({ turnId });494 const existingTurn = yield* projectionTurnRepository.getByTurnId({
495 const pendingTurnStart = yield* projectionPendingTurnStartRepository.getByThreadId({495 threadId: event.payload.threadId,
496 turnId,
497 });
498 const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({
496 threadId: event.payload.threadId,499 threadId: event.payload.threadId,
497 });500 });
498 if (Option.isSome(existingTurn)) {501 if (Option.isSome(existingTurn)) {
499 yield* projectionThreadTurnRepository.upsert({502 const nextState =
503 existingTurn.value.state === "completed" || existingTurn.value.state === "error"
504 ? existingTurn.value.state
505 : "running";
506 yield* projectionTurnRepository.upsertByTurnId({
500 ...existingTurn.value,507 ...existingTurn.value,
501 status: existingTurn.value.status === "completed" ? "completed" : "running",508 state: nextState,
502 userMessageId:509 pendingMessageId:
503 existingTurn.value.userMessageId ??510 existingTurn.value.pendingMessageId ??
504 (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null),511 (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null),
505 startedAt: existingTurn.value.startedAt,512 startedAt:
513 existingTurn.value.startedAt ??
514 (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.requestedAt : event.occurredAt),
515 requestedAt:
516 existingTurn.value.requestedAt ??
517 (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.requestedAt : event.occurredAt),
506 });518 });
507 } else {519 } else {
508 const existingThreadTurns = yield* projectionThreadTurnRepository.listByThreadId({520 yield* projectionTurnRepository.upsertByTurnId({
509 threadId: event.payload.threadId,
510 });
511 yield* projectionThreadTurnRepository.upsert({
512 turnId,521 turnId,
513 threadId: event.payload.threadId,522 threadId: event.payload.threadId,
514 turnCount: nextTurnCount(existingThreadTurns),523 pendingMessageId: Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null,
515 status: "running",
516 userMessageId: Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null,
517 assistantMessageId: null,524 assistantMessageId: null,
525 state: "running",
526 requestedAt: Option.isSome(pendingTurnStart)
527 ? pendingTurnStart.value.requestedAt
528 : event.occurredAt,
518 startedAt: Option.isSome(pendingTurnStart)529 startedAt: Option.isSome(pendingTurnStart)
519 ? pendingTurnStart.value.createdAt530 ? pendingTurnStart.value.requestedAt
520 : event.occurredAt,531 : event.occurredAt,
521 completedAt: null,532 completedAt: null,
533 checkpointTurnCount: null,
534 checkpointRef: null,
535 checkpointStatus: null,
536 checkpointFiles: [],
522 });537 });
523 }538 }
524539
525 yield* projectionPendingTurnStartRepository.deleteByThreadId({540 yield* projectionTurnRepository.deletePendingTurnStartByThreadId({
526 threadId: event.payload.threadId,541 threadId: event.payload.threadId,
527 });542 });
528 return;543 return;
...@@ -532,38 +547,42 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -532,38 +547,42 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
532 if (event.payload.turnId === null || event.payload.role !== "assistant") {547 if (event.payload.turnId === null || event.payload.role !== "assistant") {
533 return;548 return;
534 }549 }
535 const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({550 const existingTurn = yield* projectionTurnRepository.getByTurnId({
551 threadId: event.payload.threadId,
536 turnId: event.payload.turnId,552 turnId: event.payload.turnId,
537 });553 });
538 if (Option.isSome(existingTurn)) {554 if (Option.isSome(existingTurn)) {
539 yield* projectionThreadTurnRepository.upsert({555 yield* projectionTurnRepository.upsertByTurnId({
540 ...existingTurn.value,556 ...existingTurn.value,
541 assistantMessageId: event.payload.messageId,557 assistantMessageId: event.payload.messageId,
542 status: event.payload.streaming558 state: event.payload.streaming
543 ? existingTurn.value.status559 ? existingTurn.value.state
544 : existingTurn.value.status === "interrupted"560 : existingTurn.value.state === "interrupted"
545 ? "interrupted"561 ? "interrupted"
546 : existingTurn.value.status === "error"562 : existingTurn.value.state === "error"
547 ? "error"563 ? "error"
548 : "completed",564 : "completed",
549 completedAt: event.payload.streaming565 completedAt: event.payload.streaming
550 ? existingTurn.value.completedAt566 ? existingTurn.value.completedAt
551 : (existingTurn.value.completedAt ?? event.payload.updatedAt),567 : (existingTurn.value.completedAt ?? event.payload.updatedAt),
568 startedAt: existingTurn.value.startedAt ?? event.payload.createdAt,
569 requestedAt: existingTurn.value.requestedAt ?? event.payload.createdAt,
552 });570 });
553 return;571 return;
554 }572 }
555 const existingThreadTurns = yield* projectionThreadTurnRepository.listByThreadId({573 yield* projectionTurnRepository.upsertByTurnId({
556 threadId: event.payload.threadId,
557 });
558 yield* projectionThreadTurnRepository.upsert({
559 turnId: event.payload.turnId,574 turnId: event.payload.turnId,
560 threadId: event.payload.threadId,575 threadId: event.payload.threadId,
561 turnCount: nextTurnCount(existingThreadTurns),576 pendingMessageId: null,
562 status: event.payload.streaming ? "running" : "completed",
563 userMessageId: null,
564 assistantMessageId: event.payload.messageId,577 assistantMessageId: event.payload.messageId,
578 state: event.payload.streaming ? "running" : "completed",
579 requestedAt: event.payload.createdAt,
565 startedAt: event.payload.createdAt,580 startedAt: event.payload.createdAt,
566 completedAt: event.payload.streaming ? null : event.payload.updatedAt,581 completedAt: event.payload.streaming ? null : event.payload.updatedAt,
582 checkpointTurnCount: null,
583 checkpointRef: null,
584 checkpointStatus: null,
585 checkpointFiles: [],
567 });586 });
568 return;587 return;
569 }588 }
...@@ -572,74 +591,105 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -572,74 +591,105 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
572 if (event.payload.turnId === undefined) {591 if (event.payload.turnId === undefined) {
573 return;592 return;
574 }593 }
575 const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({594 const existingTurn = yield* projectionTurnRepository.getByTurnId({
595 threadId: event.payload.threadId,
576 turnId: event.payload.turnId,596 turnId: event.payload.turnId,
577 });597 });
578 if (Option.isSome(existingTurn)) {598 if (Option.isSome(existingTurn)) {
579 yield* projectionThreadTurnRepository.upsert({599 yield* projectionTurnRepository.upsertByTurnId({
580 ...existingTurn.value,600 ...existingTurn.value,
581 status: "interrupted",601 state: "interrupted",
582 completedAt: existingTurn.value.completedAt ?? event.payload.createdAt,602 completedAt: existingTurn.value.completedAt ?? event.payload.createdAt,
603 startedAt: existingTurn.value.startedAt ?? event.payload.createdAt,
604 requestedAt: existingTurn.value.requestedAt ?? event.payload.createdAt,
583 });605 });
584 return;606 return;
585 }607 }
586 const existingThreadTurns = yield* projectionThreadTurnRepository.listByThreadId({608 yield* projectionTurnRepository.upsertByTurnId({
587 threadId: event.payload.threadId,
588 });
589 yield* projectionThreadTurnRepository.upsert({
590 turnId: event.payload.turnId,609 turnId: event.payload.turnId,
591 threadId: event.payload.threadId,610 threadId: event.payload.threadId,
592 turnCount: nextTurnCount(existingThreadTurns),611 pendingMessageId: null,
593 status: "interrupted",
594 userMessageId: null,
595 assistantMessageId: null,612 assistantMessageId: null,
613 state: "interrupted",
614 requestedAt: event.payload.createdAt,
596 startedAt: event.payload.createdAt,615 startedAt: event.payload.createdAt,
597 completedAt: event.payload.createdAt,616 completedAt: event.payload.createdAt,
617 checkpointTurnCount: null,
618 checkpointRef: null,
619 checkpointStatus: null,
620 checkpointFiles: [],
598 });621 });
599 return;622 return;
600 }623 }
601624
602 case "thread.turn-diff-completed": {625 case "thread.turn-diff-completed": {
603 const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({626 const existingTurn = yield* projectionTurnRepository.getByTurnId({
627 threadId: event.payload.threadId,
604 turnId: event.payload.turnId,628 turnId: event.payload.turnId,
605 });629 });
606 const nextStatus = event.payload.status === "error" ? "error" : "completed";630 const nextState = event.payload.status === "error" ? "error" : "completed";
631 yield* projectionTurnRepository.clearCheckpointTurnConflict({
632 threadId: event.payload.threadId,
633 turnId: event.payload.turnId,
634 checkpointTurnCount: event.payload.checkpointTurnCount,
635 });
636
607 if (Option.isSome(existingTurn)) {637 if (Option.isSome(existingTurn)) {
608 yield* projectionThreadTurnRepository.upsert({638 yield* projectionTurnRepository.upsertByTurnId({
609 ...existingTurn.value,639 ...existingTurn.value,
610 turnCount: event.payload.checkpointTurnCount,
611 status: nextStatus,
612 assistantMessageId: event.payload.assistantMessageId,640 assistantMessageId: event.payload.assistantMessageId,
641 state: nextState,
642 checkpointTurnCount: event.payload.checkpointTurnCount,
643 checkpointRef: event.payload.checkpointRef,
644 checkpointStatus: event.payload.status,
645 checkpointFiles: event.payload.files,
646 startedAt: existingTurn.value.startedAt ?? event.payload.completedAt,
647 requestedAt: existingTurn.value.requestedAt ?? event.payload.completedAt,
613 completedAt: event.payload.completedAt,648 completedAt: event.payload.completedAt,
614 });649 });
615 return;650 return;
616 }651 }
617 yield* projectionThreadTurnRepository.upsert({652 yield* projectionTurnRepository.upsertByTurnId({
618 turnId: event.payload.turnId,653 turnId: event.payload.turnId,
619 threadId: event.payload.threadId,654 threadId: event.payload.threadId,
620 turnCount: event.payload.checkpointTurnCount,655 pendingMessageId: null,
621 status: nextStatus,
622 userMessageId: null,
623 assistantMessageId: event.payload.assistantMessageId,656 assistantMessageId: event.payload.assistantMessageId,
657 state: nextState,
658 requestedAt: event.payload.completedAt,
624 startedAt: event.payload.completedAt,659 startedAt: event.payload.completedAt,
625 completedAt: event.payload.completedAt,660 completedAt: event.payload.completedAt,
661 checkpointTurnCount: event.payload.checkpointTurnCount,
662 checkpointRef: event.payload.checkpointRef,
663 checkpointStatus: event.payload.status,
664 checkpointFiles: event.payload.files,
626 });665 });
627 return;666 return;
628 }667 }
629668
630 case "thread.reverted": {669 case "thread.reverted": {
631 const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({670 const existingTurns = yield* projectionTurnRepository.listByThreadId({
632 threadId: event.payload.threadId,671 threadId: event.payload.threadId,
633 });672 });
634 const keptTurns = existingTurns.filter(673 const keptTurns = existingTurns.filter(
635 (turn) => turn.turnCount <= event.payload.turnCount,674 (turn) =>
675 turn.turnId !== null &&
676 turn.checkpointTurnCount !== null &&
677 turn.checkpointTurnCount <= event.payload.turnCount,
636 );678 );
637 yield* projectionThreadTurnRepository.deleteByThreadId({679 yield* projectionTurnRepository.deleteByThreadId({
638 threadId: event.payload.threadId,680 threadId: event.payload.threadId,
639 });681 });
640 yield* Effect.forEach(keptTurns, projectionThreadTurnRepository.upsert, {682 yield* Effect.forEach(
641 concurrency: 1,683 keptTurns,
642 }).pipe(Effect.asVoid);684 (turn) =>
685 turn.turnId === null
686 ? Effect.void
687 : projectionTurnRepository.upsertByTurnId({
688 ...turn,
689 turnId: turn.turnId,
690 }),
691 { concurrency: 1 },
692 ).pipe(Effect.asVoid);
643 return;693 return;
644 }694 }
645695
...@@ -648,42 +698,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -648,42 +698,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
648 }698 }
649 });699 });
650700
651 const applyCheckpointsProjection: ProjectorDefinition["apply"] = (event) =>701 const applyCheckpointsProjection: ProjectorDefinition["apply"] = () => Effect.void;
652 Effect.gen(function* () {
653 switch (event.type) {
654 case "thread.turn-diff-completed":
655 yield* projectionCheckpointRepository.upsert({
656 threadId: event.payload.threadId,
657 turnId: event.payload.turnId,
658 checkpointTurnCount: event.payload.checkpointTurnCount,
659 checkpointRef: event.payload.checkpointRef,
660 status: event.payload.status,
661 files: event.payload.files,
662 assistantMessageId: event.payload.assistantMessageId,
663 completedAt: event.payload.completedAt,
664 });
665 return;
666
667 case "thread.reverted": {
668 const existingCheckpoints = yield* projectionCheckpointRepository.listByThreadId({
669 threadId: event.payload.threadId,
670 });
671 const keptCheckpoints = existingCheckpoints.filter(
672 (checkpoint) => checkpoint.checkpointTurnCount <= event.payload.turnCount,
673 );
674 yield* projectionCheckpointRepository.deleteByThreadId({
675 threadId: event.payload.threadId,
676 });
677 yield* Effect.forEach(keptCheckpoints, projectionCheckpointRepository.upsert, {
678 concurrency: 1,
679 }).pipe(Effect.asVoid);
680 return;
681 }
682
683 default:
684 return;
685 }
686 });
687702
688 const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = (event) =>703 const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = (event) =>
689 Effect.gen(function* () {704 Effect.gen(function* () {
...@@ -877,9 +892,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect(...@@ -877,9 +892,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect(
877 Layer.provideMerge(ProjectionThreadMessageRepositoryLive),892 Layer.provideMerge(ProjectionThreadMessageRepositoryLive),
878 Layer.provideMerge(ProjectionThreadActivityRepositoryLive),893 Layer.provideMerge(ProjectionThreadActivityRepositoryLive),
879 Layer.provideMerge(ProjectionThreadSessionRepositoryLive),894 Layer.provideMerge(ProjectionThreadSessionRepositoryLive),
880 Layer.provideMerge(ProjectionThreadTurnRepositoryLive),895 Layer.provideMerge(ProjectionTurnRepositoryLive),
881 Layer.provideMerge(ProjectionCheckpointRepositoryLive),
882 Layer.provideMerge(ProjectionPendingApprovalRepositoryLive),896 Layer.provideMerge(ProjectionPendingApprovalRepositoryLive),
883 Layer.provideMerge(ProjectionPendingTurnStartRepositoryLive),
884 Layer.provideMerge(ProjectionStateRepositoryLive),897 Layer.provideMerge(ProjectionStateRepositoryLive),
885);898);
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+17-8
...@@ -38,6 +38,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {...@@ -38,6 +38,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
3838
39 yield* sql`DELETE FROM projection_projects`;39 yield* sql`DELETE FROM projection_projects`;
40 yield* sql`DELETE FROM projection_state`;40 yield* sql`DELETE FROM projection_state`;
41 yield* sql`DELETE FROM projection_turns`;
4142
42 yield* sql`43 yield* sql`
43 INSERT INTO projection_projects (44 INSERT INTO projection_projects (
...@@ -159,25 +160,33 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {...@@ -159,25 +160,33 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
159 `;160 `;
160161
161 yield* sql`162 yield* sql`
162 INSERT INTO projection_checkpoints (163 INSERT INTO projection_turns (
163 thread_id,164 thread_id,
164 turn_id,165 turn_id,
166 pending_message_id,
167 assistant_message_id,
168 state,
169 requested_at,
170 started_at,
171 completed_at,
165 checkpoint_turn_count,172 checkpoint_turn_count,
166 checkpoint_ref,173 checkpoint_ref,
167 status,174 checkpoint_status,
168 files_json,175 checkpoint_files_json
169 assistant_message_id,
170 completed_at
171 )176 )
172 VALUES (177 VALUES (
173 'thread-1',178 'thread-1',
174 'turn-1',179 'turn-1',
180 NULL,
181 'message-1',
182 'completed',
183 '2026-02-24T00:00:08.000Z',
184 '2026-02-24T00:00:08.000Z',
185 '2026-02-24T00:00:08.000Z',
175 1,186 1,
176 'checkpoint-1',187 'checkpoint-1',
177 'ready',188 'ready',
178 '[{"path":"README.md","kind":"modified","additions":2,"deletions":1}]',189 '[{"path":"README.md","kind":"modified","additions":2,"deletions":1}]'
179 'message-1',
180 '2026-02-24T00:00:08.000Z'
181 )190 )
182 `;191 `;
183192
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+4-3
...@@ -214,11 +214,12 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -214,11 +214,12 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
214 turn_id AS "turnId",214 turn_id AS "turnId",
215 checkpoint_turn_count AS "checkpointTurnCount",215 checkpoint_turn_count AS "checkpointTurnCount",
216 checkpoint_ref AS "checkpointRef",216 checkpoint_ref AS "checkpointRef",
217 status,217 checkpoint_status AS "status",
218 files_json AS "files",218 checkpoint_files_json AS "files",
219 assistant_message_id AS "assistantMessageId",219 assistant_message_id AS "assistantMessageId",
220 completed_at AS "completedAt"220 completed_at AS "completedAt"
221 FROM projection_checkpoints221 FROM projection_turns
222 WHERE checkpoint_turn_count IS NOT NULL
222 ORDER BY thread_id ASC, checkpoint_turn_count ASC223 ORDER BY thread_id ASC, checkpoint_turn_count ASC
223 `,224 `,
224 });225 });
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+39-3
...@@ -75,10 +75,47 @@ const turnStartKeyForEvent = (event: ProviderIntentEvent): string =>...@@ -75,10 +75,47 @@ const turnStartKeyForEvent = (event: ProviderIntentEvent): string =>
75const serverCommandId = (tag: string): CommandId =>75const serverCommandId = (tag: string): CommandId =>
76 CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`);76 CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`);
7777
78const HANDLED_TURN_START_KEY_MAX = 10_000;
79const HANDLED_TURN_START_KEY_TTL_MS = 30 * 60 * 1_000;
80
81function pruneHandledTurnStartKeys(handledKeys: Map<string, number>, nowMs: number): void {
82 const staleBefore = nowMs - HANDLED_TURN_START_KEY_TTL_MS;
83 for (const [key, seenAt] of handledKeys) {
84 if (seenAt >= staleBefore) {
85 break;
86 }
87 handledKeys.delete(key);
88 }
89
90 while (handledKeys.size > HANDLED_TURN_START_KEY_MAX) {
91 const oldestKey = handledKeys.keys().next().value;
92 if (!oldestKey) {
93 break;
94 }
95 handledKeys.delete(oldestKey);
96 }
97}
98
99function hasHandledTurnStartRecently(
100 handledKeys: Map<string, number>,
101 key: string,
102 nowMs = Date.now(),
103): boolean {
104 const seenAt = handledKeys.get(key);
105 if (seenAt !== undefined && nowMs - seenAt <= HANDLED_TURN_START_KEY_TTL_MS) {
106 handledKeys.set(key, nowMs);
107 return true;
108 }
109
110 handledKeys.set(key, nowMs);
111 pruneHandledTurnStartKeys(handledKeys, nowMs);
112 return false;
113}
114
78const make = Effect.gen(function* () {115const make = Effect.gen(function* () {
79 const orchestrationEngine = yield* OrchestrationEngineService;116 const orchestrationEngine = yield* OrchestrationEngineService;
80 const providerService = yield* ProviderService;117 const providerService = yield* ProviderService;
81 const handledTurnStartKeys = new Set<string>();118 const handledTurnStartKeys = new Map<string, number>();
82119
83 const appendProviderFailureActivity = (input: {120 const appendProviderFailureActivity = (input: {
84 readonly threadId: ThreadId;121 readonly threadId: ThreadId;
...@@ -209,10 +246,9 @@ const make = Effect.gen(function* () {...@@ -209,10 +246,9 @@ const make = Effect.gen(function* () {
209 event: Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>,246 event: Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>,
210 ) {247 ) {
211 const key = turnStartKeyForEvent(event);248 const key = turnStartKeyForEvent(event);
212 if (handledTurnStartKeys.has(key)) {249 if (hasHandledTurnStartRecently(handledTurnStartKeys, key)) {
213 return;250 return;
214 }251 }
215 handledTurnStartKeys.add(key);
216252
217 const thread = yield* resolveThread(event.payload.threadId);253 const thread = yield* resolveThread(event.payload.threadId);
218 if (!thread) {254 if (!thread) {
apps/server/src/orchestration/projector.test.ts+93
...@@ -651,4 +651,97 @@ describe("orchestration projector", () => {...@@ -651,4 +651,97 @@ describe("orchestration projector", () => {
651 })),651 })),
652 ).toEqual([{ id: "assistant-keep", role: "assistant", turnId: "turn-1" }]);652 ).toEqual([{ id: "assistant-keep", role: "assistant", turnId: "turn-1" }]);
653 });653 });
654
655 it("caps message and checkpoint retention for long-lived threads", async () => {
656 const createdAt = "2026-03-01T10:00:00.000Z";
657 const model = createEmptyReadModel(createdAt);
658
659 const afterCreate = await Effect.runPromise(
660 projectEvent(
661 model,
662 makeEvent({
663 sequence: 1,
664 type: "thread.created",
665 aggregateKind: "thread",
666 aggregateId: "thread-capped",
667 occurredAt: createdAt,
668 commandId: "cmd-create-capped",
669 payload: {
670 threadId: "thread-capped",
671 projectId: "project-1",
672 title: "capped",
673 model: "gpt-5-codex",
674 branch: null,
675 worktreePath: null,
676 createdAt,
677 updatedAt: createdAt,
678 },
679 }),
680 ),
681 );
682
683 const messageEvents: ReadonlyArray<OrchestrationEvent> = Array.from(
684 { length: 2_100 },
685 (_, index) =>
686 makeEvent({
687 sequence: index + 2,
688 type: "thread.message-sent",
689 aggregateKind: "thread",
690 aggregateId: "thread-capped",
691 occurredAt: `2026-03-01T10:00:${String(index % 60).padStart(2, "0")}.000Z`,
692 commandId: `cmd-message-${index}`,
693 payload: {
694 threadId: "thread-capped",
695 messageId: `msg-${index}`,
696 role: "assistant",
697 text: `message-${index}`,
698 turnId: `turn-${index}`,
699 streaming: false,
700 createdAt: `2026-03-01T10:00:${String(index % 60).padStart(2, "0")}.000Z`,
701 updatedAt: `2026-03-01T10:00:${String(index % 60).padStart(2, "0")}.000Z`,
702 },
703 }),
704 );
705 const afterMessages = await messageEvents.reduce<Promise<ReturnType<typeof createEmptyReadModel>>>(
706 (statePromise, event) =>
707 statePromise.then((state) => Effect.runPromise(projectEvent(state, event))),
708 Promise.resolve(afterCreate),
709 );
710
711 const checkpointEvents: ReadonlyArray<OrchestrationEvent> = Array.from(
712 { length: 600 },
713 (_, index) =>
714 makeEvent({
715 sequence: index + 2_102,
716 type: "thread.turn-diff-completed",
717 aggregateKind: "thread",
718 aggregateId: "thread-capped",
719 occurredAt: `2026-03-01T10:30:${String(index % 60).padStart(2, "0")}.000Z`,
720 commandId: `cmd-checkpoint-${index}`,
721 payload: {
722 threadId: "thread-capped",
723 turnId: `turn-${index}`,
724 checkpointTurnCount: index + 1,
725 checkpointRef: `refs/t3/checkpoints/thread-capped/turn/${index + 1}`,
726 status: "ready",
727 files: [],
728 assistantMessageId: `msg-${index}`,
729 completedAt: `2026-03-01T10:30:${String(index % 60).padStart(2, "0")}.000Z`,
730 },
731 }),
732 );
733 const finalState = await checkpointEvents.reduce<Promise<ReturnType<typeof createEmptyReadModel>>>(
734 (statePromise, event) =>
735 statePromise.then((state) => Effect.runPromise(projectEvent(state, event))),
736 Promise.resolve(afterMessages),
737 );
738
739 const thread = finalState.threads[0];
740 expect(thread?.messages).toHaveLength(2_000);
741 expect(thread?.messages[0]?.id).toBe("msg-100");
742 expect(thread?.messages.at(-1)?.id).toBe("msg-2099");
743 expect(thread?.checkpoints).toHaveLength(500);
744 expect(thread?.checkpoints[0]?.turnId).toBe("turn-100");
745 expect(thread?.checkpoints.at(-1)?.turnId).toBe("turn-599");
746 });
654});747});
apps/server/src/orchestration/projector.ts+10-4
...@@ -23,6 +23,8 @@ import {...@@ -23,6 +23,8 @@ import {
23} from "./Schemas.ts";23} from "./Schemas.ts";
2424
25type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;25type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;
26const MAX_THREAD_MESSAGES = 2_000;
27const MAX_THREAD_CHECKPOINTS = 500;
2628
27function updateThread(29function updateThread(
28 threads: ReadonlyArray<OrchestrationThread>,30 threads: ReadonlyArray<OrchestrationThread>,
...@@ -313,11 +315,12 @@ export function projectEvent(...@@ -313,11 +315,12 @@ export function projectEvent(
313 : entry,315 : entry,
314 )316 )
315 : [...thread.messages, message];317 : [...thread.messages, message];
318 const cappedMessages = messages.slice(-MAX_THREAD_MESSAGES);
316319
317 return {320 return {
318 ...nextBase,321 ...nextBase,
319 threads: updateThread(nextBase.threads, payload.threadId, {322 threads: updateThread(nextBase.threads, payload.threadId, {
320 messages,323 messages: cappedMessages,
321 updatedAt: event.occurredAt,324 updatedAt: event.occurredAt,
322 }),325 }),
323 };326 };
...@@ -384,7 +387,9 @@ export function projectEvent(...@@ -384,7 +387,9 @@ export function projectEvent(
384 const checkpoints = [387 const checkpoints = [
385 ...thread.checkpoints.filter((entry) => entry.turnId !== checkpoint.turnId),388 ...thread.checkpoints.filter((entry) => entry.turnId !== checkpoint.turnId),
386 checkpoint,389 checkpoint,
387 ].toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount);390 ]
391 .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount)
392 .slice(-MAX_THREAD_CHECKPOINTS);
388393
389 return {394 return {
390 ...nextBase,395 ...nextBase,
...@@ -406,13 +411,14 @@ export function projectEvent(...@@ -406,13 +411,14 @@ export function projectEvent(
406411
407 const checkpoints = thread.checkpoints412 const checkpoints = thread.checkpoints
408 .filter((entry) => entry.checkpointTurnCount <= payload.turnCount)413 .filter((entry) => entry.checkpointTurnCount <= payload.turnCount)
409 .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount);414 .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount)
415 .slice(-MAX_THREAD_CHECKPOINTS);
410 const retainedTurnIds = new Set(checkpoints.map((checkpoint) => checkpoint.turnId));416 const retainedTurnIds = new Set(checkpoints.map((checkpoint) => checkpoint.turnId));
411 const messages = retainThreadMessagesAfterRevert(417 const messages = retainThreadMessagesAfterRevert(
412 thread.messages,418 thread.messages,
413 retainedTurnIds,419 retainedTurnIds,
414 payload.turnCount,420 payload.turnCount,
415 );421 ).slice(-MAX_THREAD_MESSAGES);
416 const activities = retainThreadActivitiesAfterRevert(thread.activities, retainedTurnIds);422 const activities = retainThreadActivitiesAfterRevert(thread.activities, retainedTurnIds);
417423
418 const latestTurnId =424 const latestTurnId =
apps/server/src/persistence/Layers/ProjectionCheckpoints.ts+76-37
...@@ -29,39 +29,63 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st...@@ -29,39 +29,63 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st
29const makeProjectionCheckpointRepository = Effect.gen(function* () {29const makeProjectionCheckpointRepository = Effect.gen(function* () {
30 const sql = yield* SqlClient.SqlClient;30 const sql = yield* SqlClient.SqlClient;
3131
32 const clearCheckpointConflict = SqlSchema.void({
33 Request: GetByThreadAndTurnCountInput,
34 execute: ({ threadId, checkpointTurnCount }) =>
35 sql`
36 UPDATE projection_turns
37 SET
38 checkpoint_turn_count = NULL,
39 checkpoint_ref = NULL,
40 checkpoint_status = NULL,
41 checkpoint_files_json = '[]'
42 WHERE thread_id = ${threadId}
43 AND checkpoint_turn_count = ${checkpointTurnCount}
44 `,
45 });
46
32 const upsertProjectionCheckpointRow = SqlSchema.void({47 const upsertProjectionCheckpointRow = SqlSchema.void({
33 Request: ProjectionCheckpointDbRowSchema,48 Request: ProjectionCheckpointDbRowSchema,
34 execute: (row) =>49 execute: (row) =>
35 sql`50 sql`
36 INSERT INTO projection_checkpoints (51 INSERT INTO projection_turns (
37 thread_id,52 thread_id,
38 turn_id,53 turn_id,
39 checkpoint_turn_count,54 pending_message_id,
40 checkpoint_ref,55 assistant_message_id,
41 status,56 state,
42 files_json,57 requested_at,
43 assistant_message_id,58 started_at,
44 completed_at59 completed_at,
45 )60 checkpoint_turn_count,
46 VALUES (61 checkpoint_ref,
47 ${row.threadId},62 checkpoint_status,
48 ${row.turnId},63 checkpoint_files_json
49 ${row.checkpointTurnCount},64 )
50 ${row.checkpointRef},65 VALUES (
51 ${row.status},66 ${row.threadId},
52 ${row.files},67 ${row.turnId},
53 ${row.assistantMessageId},68 NULL,
54 ${row.completedAt}69 ${row.assistantMessageId},
55 )70 ${row.status === "error" ? "error" : "completed"},
56 ON CONFLICT (thread_id, checkpoint_turn_count)71 ${row.completedAt},
57 DO UPDATE SET72 ${row.completedAt},
58 turn_id = excluded.turn_id,73 ${row.completedAt},
59 checkpoint_ref = excluded.checkpoint_ref,74 ${row.checkpointTurnCount},
60 status = excluded.status,75 ${row.checkpointRef},
61 files_json = excluded.files_json,76 ${row.status},
62 assistant_message_id = excluded.assistant_message_id,77 ${row.files}
63 completed_at = excluded.completed_at78 )
64 `,79 ON CONFLICT (thread_id, turn_id)
80 DO UPDATE SET
81 assistant_message_id = excluded.assistant_message_id,
82 state = excluded.state,
83 completed_at = excluded.completed_at,
84 checkpoint_turn_count = excluded.checkpoint_turn_count,
85 checkpoint_ref = excluded.checkpoint_ref,
86 checkpoint_status = excluded.checkpoint_status,
87 checkpoint_files_json = excluded.checkpoint_files_json
88 `,
65 });89 });
6690
67 const listProjectionCheckpointRows = SqlSchema.findAll({91 const listProjectionCheckpointRows = SqlSchema.findAll({
...@@ -74,12 +98,13 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () {...@@ -74,12 +98,13 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () {
74 turn_id AS "turnId",98 turn_id AS "turnId",
75 checkpoint_turn_count AS "checkpointTurnCount",99 checkpoint_turn_count AS "checkpointTurnCount",
76 checkpoint_ref AS "checkpointRef",100 checkpoint_ref AS "checkpointRef",
77 status,101 checkpoint_status AS "status",
78 files_json AS "files",102 checkpoint_files_json AS "files",
79 assistant_message_id AS "assistantMessageId",103 assistant_message_id AS "assistantMessageId",
80 completed_at AS "completedAt"104 completed_at AS "completedAt"
81 FROM projection_checkpoints105 FROM projection_turns
82 WHERE thread_id = ${threadId}106 WHERE thread_id = ${threadId}
107 AND checkpoint_turn_count IS NOT NULL
83 ORDER BY checkpoint_turn_count ASC108 ORDER BY checkpoint_turn_count ASC
84 `,109 `,
85 });110 });
...@@ -94,11 +119,11 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () {...@@ -94,11 +119,11 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () {
94 turn_id AS "turnId",119 turn_id AS "turnId",
95 checkpoint_turn_count AS "checkpointTurnCount",120 checkpoint_turn_count AS "checkpointTurnCount",
96 checkpoint_ref AS "checkpointRef",121 checkpoint_ref AS "checkpointRef",
97 status,122 checkpoint_status AS "status",
98 files_json AS "files",123 checkpoint_files_json AS "files",
99 assistant_message_id AS "assistantMessageId",124 assistant_message_id AS "assistantMessageId",
100 completed_at AS "completedAt"125 completed_at AS "completedAt"
101 FROM projection_checkpoints126 FROM projection_turns
102 WHERE thread_id = ${threadId}127 WHERE thread_id = ${threadId}
103 AND checkpoint_turn_count = ${checkpointTurnCount}128 AND checkpoint_turn_count = ${checkpointTurnCount}
104 `,129 `,
...@@ -108,13 +133,27 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () {...@@ -108,13 +133,27 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () {
108 Request: DeleteByThreadIdInput,133 Request: DeleteByThreadIdInput,
109 execute: ({ threadId }) =>134 execute: ({ threadId }) =>
110 sql`135 sql`
111 DELETE FROM projection_checkpoints136 UPDATE projection_turns
137 SET
138 checkpoint_turn_count = NULL,
139 checkpoint_ref = NULL,
140 checkpoint_status = NULL,
141 checkpoint_files_json = '[]'
112 WHERE thread_id = ${threadId}142 WHERE thread_id = ${threadId}
143 AND checkpoint_turn_count IS NOT NULL
113 `,144 `,
114 });145 });
115146
147 const upsertCheckpointRow = (row: Schema.Schema.Type<typeof ProjectionCheckpointDbRowSchema>) =>
148 sql.withTransaction(
149 clearCheckpointConflict({
150 threadId: row.threadId,
151 checkpointTurnCount: row.checkpointTurnCount,
152 }).pipe(Effect.flatMap(() => upsertProjectionCheckpointRow(row))),
153 );
154
116 const upsert: ProjectionCheckpointRepositoryShape["upsert"] = (row) =>155 const upsert: ProjectionCheckpointRepositoryShape["upsert"] = (row) =>
117 upsertProjectionCheckpointRow(row).pipe(156 upsertCheckpointRow(row).pipe(
118 Effect.mapError(157 Effect.mapError(
119 toPersistenceSqlOrDecodeError(158 toPersistenceSqlOrDecodeError(
120 "ProjectionCheckpointRepository.upsert:query",159 "ProjectionCheckpointRepository.upsert:query",
apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts deleted-92
...@@ -1,92 +0,0 @@
1import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as SqlSchema from "effect/unstable/sql/SqlSchema";
3import { Effect, Layer } from "effect";
4
5import { toPersistenceSqlError } from "../Errors.ts";
6import {
7 DeleteProjectionPendingTurnStartInput,
8 GetProjectionPendingTurnStartInput,
9 ProjectionPendingTurnStart,
10 ProjectionPendingTurnStartRepository,
11 type ProjectionPendingTurnStartRepositoryShape,
12} from "../Services/ProjectionPendingTurnStarts.ts";
13
14const makeProjectionPendingTurnStartRepository = Effect.gen(function* () {
15 const sql = yield* SqlClient.SqlClient;
16
17 const upsertProjectionPendingTurnStartRow = SqlSchema.void({
18 Request: ProjectionPendingTurnStart,
19 execute: (row) =>
20 sql`
21 INSERT INTO projection_pending_turn_starts (
22 thread_id,
23 message_id,
24 created_at
25 )
26 VALUES (
27 ${row.threadId},
28 ${row.messageId},
29 ${row.createdAt}
30 )
31 ON CONFLICT (thread_id)
32 DO UPDATE SET
33 message_id = excluded.message_id,
34 created_at = excluded.created_at
35 `,
36 });
37
38 const getProjectionPendingTurnStartRow = SqlSchema.findOneOption({
39 Request: GetProjectionPendingTurnStartInput,
40 Result: ProjectionPendingTurnStart,
41 execute: ({ threadId }) =>
42 sql`
43 SELECT
44 thread_id AS "threadId",
45 message_id AS "messageId",
46 created_at AS "createdAt"
47 FROM projection_pending_turn_starts
48 WHERE thread_id = ${threadId}
49 `,
50 });
51
52 const deleteProjectionPendingTurnStartRow = SqlSchema.void({
53 Request: DeleteProjectionPendingTurnStartInput,
54 execute: ({ threadId }) =>
55 sql`
56 DELETE FROM projection_pending_turn_starts
57 WHERE thread_id = ${threadId}
58 `,
59 });
60
61 const upsert: ProjectionPendingTurnStartRepositoryShape["upsert"] = (row) =>
62 upsertProjectionPendingTurnStartRow(row).pipe(
63 Effect.mapError(toPersistenceSqlError("ProjectionPendingTurnStartRepository.upsert:query")),
64 );
65
66 const getByThreadId: ProjectionPendingTurnStartRepositoryShape["getByThreadId"] = (input) =>
67 getProjectionPendingTurnStartRow(input).pipe(
68 Effect.mapError(
69 toPersistenceSqlError("ProjectionPendingTurnStartRepository.getByThreadId:query"),
70 ),
71 );
72
73 const deleteByThreadId: ProjectionPendingTurnStartRepositoryShape["deleteByThreadId"] = (
74 input,
75 ) =>
76 deleteProjectionPendingTurnStartRow(input).pipe(
77 Effect.mapError(
78 toPersistenceSqlError("ProjectionPendingTurnStartRepository.deleteByThreadId:query"),
79 ),
80 );
81
82 return {
83 upsert,
84 getByThreadId,
85 deleteByThreadId,
86 } satisfies ProjectionPendingTurnStartRepositoryShape;
87});
88
89export const ProjectionPendingTurnStartRepositoryLive = Layer.effect(
90 ProjectionPendingTurnStartRepository,
91 makeProjectionPendingTurnStartRepository,
92);
apps/server/src/persistence/Layers/ProjectionThreadTurns.ts deleted-135
...@@ -1,135 +0,0 @@
1import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as SqlSchema from "effect/unstable/sql/SqlSchema";
3import { Effect, Layer } from "effect";
4
5import { toPersistenceSqlError } from "../Errors.ts";
6import {
7 ProjectionThreadTurn,
8 ProjectionThreadTurnRepository,
9 type ProjectionThreadTurnRepositoryShape,
10 DeleteProjectionThreadTurnsInput,
11 GetProjectionThreadTurnInput,
12 ListProjectionThreadTurnsInput,
13} from "../Services/ProjectionThreadTurns.ts";
14
15const makeProjectionThreadTurnRepository = Effect.gen(function* () {
16 const sql = yield* SqlClient.SqlClient;
17
18 const upsertProjectionThreadTurnRow = SqlSchema.void({
19 Request: ProjectionThreadTurn,
20 execute: (row) =>
21 sql`
22 INSERT INTO projection_thread_turns (
23 turn_id,
24 thread_id,
25 turn_count,
26 status,
27 user_message_id,
28 assistant_message_id,
29 started_at,
30 completed_at
31 )
32 VALUES (
33 ${row.turnId},
34 ${row.threadId},
35 ${row.turnCount},
36 ${row.status},
37 ${row.userMessageId},
38 ${row.assistantMessageId},
39 ${row.startedAt},
40 ${row.completedAt}
41 )
42 ON CONFLICT (turn_id)
43 DO UPDATE SET
44 thread_id = excluded.thread_id,
45 turn_count = excluded.turn_count,
46 status = excluded.status,
47 user_message_id = excluded.user_message_id,
48 assistant_message_id = excluded.assistant_message_id,
49 started_at = excluded.started_at,
50 completed_at = excluded.completed_at
51 `,
52 });
53
54 const listProjectionThreadTurnRows = SqlSchema.findAll({
55 Request: ListProjectionThreadTurnsInput,
56 Result: ProjectionThreadTurn,
57 execute: ({ threadId }) =>
58 sql`
59 SELECT
60 turn_id AS "turnId",
61 thread_id AS "threadId",
62 turn_count AS "turnCount",
63 status,
64 user_message_id AS "userMessageId",
65 assistant_message_id AS "assistantMessageId",
66 started_at AS "startedAt",
67 completed_at AS "completedAt"
68 FROM projection_thread_turns
69 WHERE thread_id = ${threadId}
70 ORDER BY turn_count ASC
71 `,
72 });
73
74 const getProjectionThreadTurnRow = SqlSchema.findOneOption({
75 Request: GetProjectionThreadTurnInput,
76 Result: ProjectionThreadTurn,
77 execute: ({ turnId }) =>
78 sql`
79 SELECT
80 turn_id AS "turnId",
81 thread_id AS "threadId",
82 turn_count AS "turnCount",
83 status,
84 user_message_id AS "userMessageId",
85 assistant_message_id AS "assistantMessageId",
86 started_at AS "startedAt",
87 completed_at AS "completedAt"
88 FROM projection_thread_turns
89 WHERE turn_id = ${turnId}
90 `,
91 });
92
93 const deleteProjectionThreadTurnRows = SqlSchema.void({
94 Request: DeleteProjectionThreadTurnsInput,
95 execute: ({ threadId }) =>
96 sql`
97 DELETE FROM projection_thread_turns
98 WHERE thread_id = ${threadId}
99 `,
100 });
101
102 const upsert: ProjectionThreadTurnRepositoryShape["upsert"] = (row) =>
103 upsertProjectionThreadTurnRow(row).pipe(
104 Effect.mapError(toPersistenceSqlError("ProjectionThreadTurnRepository.upsert:query")),
105 );
106
107 const listByThreadId: ProjectionThreadTurnRepositoryShape["listByThreadId"] = (input) =>
108 listProjectionThreadTurnRows(input).pipe(
109 Effect.mapError(toPersistenceSqlError("ProjectionThreadTurnRepository.listByThreadId:query")),
110 );
111
112 const getByTurnId: ProjectionThreadTurnRepositoryShape["getByTurnId"] = (input) =>
113 getProjectionThreadTurnRow(input).pipe(
114 Effect.mapError(toPersistenceSqlError("ProjectionThreadTurnRepository.getByTurnId:query")),
115 );
116
117 const deleteByThreadId: ProjectionThreadTurnRepositoryShape["deleteByThreadId"] = (input) =>
118 deleteProjectionThreadTurnRows(input).pipe(
119 Effect.mapError(
120 toPersistenceSqlError("ProjectionThreadTurnRepository.deleteByThreadId:query"),
121 ),
122 );
123
124 return {
125 upsert,
126 listByThreadId,
127 getByTurnId,
128 deleteByThreadId,
129 } satisfies ProjectionThreadTurnRepositoryShape;
130});
131
132export const ProjectionThreadTurnRepositoryLive = Layer.effect(
133 ProjectionThreadTurnRepository,
134 makeProjectionThreadTurnRepository,
135);
apps/server/src/persistence/Layers/ProjectionTurns.ts created+336
...@@ -0,0 +1,336 @@
1import { OrchestrationCheckpointFile } from "@t3tools/contracts";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
3import * as SqlSchema from "effect/unstable/sql/SqlSchema";
4import { Effect, Layer, Option, Schema, Struct } from "effect";
5
6import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts";
7import {
8 ClearCheckpointTurnConflictInput,
9 DeleteProjectionTurnsByThreadInput,
10 GetProjectionPendingTurnStartInput,
11 GetProjectionTurnByTurnIdInput,
12 ListProjectionTurnsByThreadInput,
13 ProjectionPendingTurnStart,
14 ProjectionTurn,
15 ProjectionTurnById,
16 ProjectionTurnRepository,
17 type ProjectionTurnRepositoryShape,
18} from "../Services/ProjectionTurns.ts";
19
20const ProjectionTurnDbRowSchema = ProjectionTurn.mapFields(
21 Struct.assign({
22 checkpointFiles: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)),
23 }),
24);
25
26const ProjectionTurnByIdDbRowSchema = ProjectionTurnById.mapFields(
27 Struct.assign({
28 checkpointFiles: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)),
29 }),
30);
31
32function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
33 return (cause: unknown) =>
34 Schema.isSchemaError(cause)
35 ? toPersistenceDecodeError(decodeOperation)(cause)
36 : toPersistenceSqlError(sqlOperation)(cause);
37}
38
39const makeProjectionTurnRepository = Effect.gen(function* () {
40 const sql = yield* SqlClient.SqlClient;
41
42 const upsertProjectionTurnById = SqlSchema.void({
43 Request: ProjectionTurnByIdDbRowSchema,
44 execute: (row) =>
45 sql`
46 INSERT INTO projection_turns (
47 thread_id,
48 turn_id,
49 pending_message_id,
50 assistant_message_id,
51 state,
52 requested_at,
53 started_at,
54 completed_at,
55 checkpoint_turn_count,
56 checkpoint_ref,
57 checkpoint_status,
58 checkpoint_files_json
59 )
60 VALUES (
61 ${row.threadId},
62 ${row.turnId},
63 ${row.pendingMessageId},
64 ${row.assistantMessageId},
65 ${row.state},
66 ${row.requestedAt},
67 ${row.startedAt},
68 ${row.completedAt},
69 ${row.checkpointTurnCount},
70 ${row.checkpointRef},
71 ${row.checkpointStatus},
72 ${row.checkpointFiles}
73 )
74 ON CONFLICT (thread_id, turn_id)
75 DO UPDATE SET
76 pending_message_id = excluded.pending_message_id,
77 assistant_message_id = excluded.assistant_message_id,
78 state = excluded.state,
79 requested_at = excluded.requested_at,
80 started_at = excluded.started_at,
81 completed_at = excluded.completed_at,
82 checkpoint_turn_count = excluded.checkpoint_turn_count,
83 checkpoint_ref = excluded.checkpoint_ref,
84 checkpoint_status = excluded.checkpoint_status,
85 checkpoint_files_json = excluded.checkpoint_files_json
86 `,
87 });
88
89 const clearPendingProjectionTurnsByThread = SqlSchema.void({
90 Request: DeleteProjectionTurnsByThreadInput,
91 execute: ({ threadId }) =>
92 sql`
93 DELETE FROM projection_turns
94 WHERE thread_id = ${threadId}
95 AND turn_id IS NULL
96 AND state = 'pending'
97 AND checkpoint_turn_count IS NULL
98 `,
99 });
100
101 const insertPendingProjectionTurn = SqlSchema.void({
102 Request: ProjectionPendingTurnStart,
103 execute: (row) =>
104 sql`
105 INSERT INTO projection_turns (
106 thread_id,
107 turn_id,
108 pending_message_id,
109 assistant_message_id,
110 state,
111 requested_at,
112 started_at,
113 completed_at,
114 checkpoint_turn_count,
115 checkpoint_ref,
116 checkpoint_status,
117 checkpoint_files_json
118 )
119 VALUES (
120 ${row.threadId},
121 NULL,
122 ${row.messageId},
123 NULL,
124 'pending',
125 ${row.requestedAt},
126 NULL,
127 NULL,
128 NULL,
129 NULL,
130 NULL,
131 '[]'
132 )
133 `,
134 });
135
136 const getPendingProjectionTurn = SqlSchema.findOneOption({
137 Request: GetProjectionPendingTurnStartInput,
138 Result: ProjectionPendingTurnStart,
139 execute: ({ threadId }) =>
140 sql`
141 SELECT
142 thread_id AS "threadId",
143 pending_message_id AS "messageId",
144 requested_at AS "requestedAt"
145 FROM projection_turns
146 WHERE thread_id = ${threadId}
147 AND turn_id IS NULL
148 AND state = 'pending'
149 AND pending_message_id IS NOT NULL
150 AND checkpoint_turn_count IS NULL
151 ORDER BY requested_at DESC
152 LIMIT 1
153 `,
154 });
155
156 const listProjectionTurnsByThread = SqlSchema.findAll({
157 Request: ListProjectionTurnsByThreadInput,
158 Result: ProjectionTurnDbRowSchema,
159 execute: ({ threadId }) =>
160 sql`
161 SELECT
162 thread_id AS "threadId",
163 turn_id AS "turnId",
164 pending_message_id AS "pendingMessageId",
165 assistant_message_id AS "assistantMessageId",
166 state,
167 requested_at AS "requestedAt",
168 started_at AS "startedAt",
169 completed_at AS "completedAt",
170 checkpoint_turn_count AS "checkpointTurnCount",
171 checkpoint_ref AS "checkpointRef",
172 checkpoint_status AS "checkpointStatus",
173 checkpoint_files_json AS "checkpointFiles"
174 FROM projection_turns
175 WHERE thread_id = ${threadId}
176 ORDER BY
177 CASE
178 WHEN checkpoint_turn_count IS NULL THEN 1
179 ELSE 0
180 END ASC,
181 checkpoint_turn_count ASC,
182 requested_at ASC,
183 turn_id ASC
184 `,
185 });
186
187 const getProjectionTurnByTurnId = SqlSchema.findOneOption({
188 Request: GetProjectionTurnByTurnIdInput,
189 Result: ProjectionTurnByIdDbRowSchema,
190 execute: ({ threadId, turnId }) =>
191 sql`
192 SELECT
193 thread_id AS "threadId",
194 turn_id AS "turnId",
195 pending_message_id AS "pendingMessageId",
196 assistant_message_id AS "assistantMessageId",
197 state,
198 requested_at AS "requestedAt",
199 started_at AS "startedAt",
200 completed_at AS "completedAt",
201 checkpoint_turn_count AS "checkpointTurnCount",
202 checkpoint_ref AS "checkpointRef",
203 checkpoint_status AS "checkpointStatus",
204 checkpoint_files_json AS "checkpointFiles"
205 FROM projection_turns
206 WHERE thread_id = ${threadId}
207 AND turn_id = ${turnId}
208 LIMIT 1
209 `,
210 });
211
212 const clearCheckpointTurnConflictRow = SqlSchema.void({
213 Request: ClearCheckpointTurnConflictInput,
214 execute: ({ threadId, turnId, checkpointTurnCount }) =>
215 sql`
216 UPDATE projection_turns
217 SET
218 checkpoint_turn_count = NULL,
219 checkpoint_ref = NULL,
220 checkpoint_status = NULL,
221 checkpoint_files_json = '[]'
222 WHERE thread_id = ${threadId}
223 AND checkpoint_turn_count = ${checkpointTurnCount}
224 AND (turn_id IS NULL OR turn_id <> ${turnId})
225 `,
226 });
227
228 const deleteProjectionTurnsByThread = SqlSchema.void({
229 Request: DeleteProjectionTurnsByThreadInput,
230 execute: ({ threadId }) =>
231 sql`
232 DELETE FROM projection_turns
233 WHERE thread_id = ${threadId}
234 `,
235 });
236
237 const upsertByTurnId: ProjectionTurnRepositoryShape["upsertByTurnId"] = (row) =>
238 upsertProjectionTurnById(row).pipe(
239 Effect.mapError(
240 toPersistenceSqlOrDecodeError(
241 "ProjectionTurnRepository.upsertByTurnId:query",
242 "ProjectionTurnRepository.upsertByTurnId:encodeRequest",
243 ),
244 ),
245 );
246
247 const replacePendingTurnStart: ProjectionTurnRepositoryShape["replacePendingTurnStart"] = (row) =>
248 sql
249 .withTransaction(
250 clearPendingProjectionTurnsByThread({ threadId: row.threadId }).pipe(
251 Effect.flatMap(() => insertPendingProjectionTurn(row)),
252 ),
253 )
254 .pipe(
255 Effect.mapError(
256 toPersistenceSqlOrDecodeError(
257 "ProjectionTurnRepository.replacePendingTurnStart:query",
258 "ProjectionTurnRepository.replacePendingTurnStart:encodeRequest",
259 ),
260 ),
261 );
262
263 const getPendingTurnStartByThreadId: ProjectionTurnRepositoryShape["getPendingTurnStartByThreadId"] = (
264 input,
265 ) =>
266 getPendingProjectionTurn(input).pipe(
267 Effect.mapError(
268 toPersistenceSqlError("ProjectionTurnRepository.getPendingTurnStartByThreadId:query"),
269 ),
270 );
271
272 const deletePendingTurnStartByThreadId: ProjectionTurnRepositoryShape["deletePendingTurnStartByThreadId"] = (
273 input,
274 ) =>
275 clearPendingProjectionTurnsByThread(input).pipe(
276 Effect.mapError(
277 toPersistenceSqlError("ProjectionTurnRepository.deletePendingTurnStartByThreadId:query"),
278 ),
279 );
280
281 const listByThreadId: ProjectionTurnRepositoryShape["listByThreadId"] = (input) =>
282 listProjectionTurnsByThread(input).pipe(
283 Effect.mapError(
284 toPersistenceSqlOrDecodeError(
285 "ProjectionTurnRepository.listByThreadId:query",
286 "ProjectionTurnRepository.listByThreadId:decodeRows",
287 ),
288 ),
289 Effect.map((rows) => rows as ReadonlyArray<Schema.Schema.Type<typeof ProjectionTurn>>),
290 );
291
292 const getByTurnId: ProjectionTurnRepositoryShape["getByTurnId"] = (input) =>
293 getProjectionTurnByTurnId(input).pipe(
294 Effect.mapError(
295 toPersistenceSqlOrDecodeError(
296 "ProjectionTurnRepository.getByTurnId:query",
297 "ProjectionTurnRepository.getByTurnId:decodeRow",
298 ),
299 ),
300 Effect.flatMap((rowOption) =>
301 Option.match(rowOption, {
302 onNone: () => Effect.succeed(Option.none()),
303 onSome: (row) =>
304 Effect.succeed(Option.some(row as Schema.Schema.Type<typeof ProjectionTurnById>)),
305 }),
306 ),
307 );
308
309 const clearCheckpointTurnConflict: ProjectionTurnRepositoryShape["clearCheckpointTurnConflict"] = (
310 input,
311 ) =>
312 clearCheckpointTurnConflictRow(input).pipe(
313 Effect.mapError(toPersistenceSqlError("ProjectionTurnRepository.clearCheckpointTurnConflict:query")),
314 );
315
316 const deleteByThreadId: ProjectionTurnRepositoryShape["deleteByThreadId"] = (input) =>
317 deleteProjectionTurnsByThread(input).pipe(
318 Effect.mapError(toPersistenceSqlError("ProjectionTurnRepository.deleteByThreadId:query")),
319 );
320
321 return {
322 upsertByTurnId,
323 replacePendingTurnStart,
324 getPendingTurnStartByThreadId,
325 deletePendingTurnStartByThreadId,
326 listByThreadId,
327 getByTurnId,
328 clearCheckpointTurnConflict,
329 deleteByThreadId,
330 } satisfies ProjectionTurnRepositoryShape;
331});
332
333export const ProjectionTurnRepositoryLive = Layer.effect(
334 ProjectionTurnRepository,
335 makeProjectionTurnRepository,
336);
apps/server/src/persistence/Migrations.ts-2
...@@ -17,7 +17,6 @@ import Migration0002 from "./Migrations/002_OrchestrationCommandReceipts.ts";...@@ -17,7 +17,6 @@ import Migration0002 from "./Migrations/002_OrchestrationCommandReceipts.ts";
17import Migration0003 from "./Migrations/003_CheckpointDiffBlobs.ts";17import Migration0003 from "./Migrations/003_CheckpointDiffBlobs.ts";
18import Migration0004 from "./Migrations/004_ProviderSessionRuntime.ts";18import Migration0004 from "./Migrations/004_ProviderSessionRuntime.ts";
19import Migration0005 from "./Migrations/005_Projections.ts";19import Migration0005 from "./Migrations/005_Projections.ts";
20import Migration0006 from "./Migrations/006_ProjectionPendingTurnStarts.ts";
2120
22/**21/**
23 * Migration loader with all migrations defined inline.22 * Migration loader with all migrations defined inline.
...@@ -35,7 +34,6 @@ const loader = Migrator.fromRecord({...@@ -35,7 +34,6 @@ const loader = Migrator.fromRecord({
35 "3_CheckpointDiffBlobs": Migration0003,34 "3_CheckpointDiffBlobs": Migration0003,
36 "4_ProviderSessionRuntime": Migration0004,35 "4_ProviderSessionRuntime": Migration0004,
37 "5_Projections": Migration0005,36 "5_Projections": Migration0005,
38 "6_ProjectionPendingTurnStarts": Migration0006,
39});37});
4038
41/**39/**
apps/server/src/persistence/Migrations/005_Projections.ts+16-24
...@@ -72,29 +72,21 @@ export default Effect.gen(function* () {...@@ -72,29 +72,21 @@ export default Effect.gen(function* () {
72 `;72 `;
7373
74 yield* sql`74 yield* sql`
75 CREATE TABLE IF NOT EXISTS projection_thread_turns (75 CREATE TABLE IF NOT EXISTS projection_turns (
76 turn_id TEXT PRIMARY KEY,76 row_id INTEGER PRIMARY KEY AUTOINCREMENT,
77 thread_id TEXT NOT NULL,77 thread_id TEXT NOT NULL,
78 turn_count INTEGER NOT NULL,78 turn_id TEXT,
79 status TEXT NOT NULL,79 pending_message_id TEXT,
80 user_message_id TEXT,
81 assistant_message_id TEXT,80 assistant_message_id TEXT,
82 started_at TEXT NOT NULL,81 state TEXT NOT NULL,
82 requested_at TEXT NOT NULL,
83 started_at TEXT,
83 completed_at TEXT,84 completed_at TEXT,
84 UNIQUE (thread_id, turn_count)85 checkpoint_turn_count INTEGER,
85 )86 checkpoint_ref TEXT,
86 `;87 checkpoint_status TEXT,
8788 checkpoint_files_json TEXT NOT NULL,
88 yield* sql`89 UNIQUE (thread_id, turn_id),
89 CREATE TABLE IF NOT EXISTS projection_checkpoints (
90 thread_id TEXT NOT NULL,
91 turn_id TEXT NOT NULL,
92 checkpoint_turn_count INTEGER NOT NULL,
93 checkpoint_ref TEXT NOT NULL,
94 status TEXT NOT NULL,
95 files_json TEXT NOT NULL,
96 assistant_message_id TEXT,
97 completed_at TEXT NOT NULL,
98 UNIQUE (thread_id, checkpoint_turn_count)90 UNIQUE (thread_id, checkpoint_turn_count)
99 )91 )
100 `;92 `;
...@@ -145,13 +137,13 @@ export default Effect.gen(function* () {...@@ -145,13 +137,13 @@ export default Effect.gen(function* () {
145 `;137 `;
146138
147 yield* sql`139 yield* sql`
148 CREATE INDEX IF NOT EXISTS idx_projection_thread_turns_thread_turn_count140 CREATE INDEX IF NOT EXISTS idx_projection_turns_thread_requested
149 ON projection_thread_turns(thread_id, turn_count)141 ON projection_turns(thread_id, requested_at)
150 `;142 `;
151143
152 yield* sql`144 yield* sql`
153 CREATE INDEX IF NOT EXISTS idx_projection_checkpoints_thread_completed145 CREATE INDEX IF NOT EXISTS idx_projection_turns_thread_checkpoint_completed
154 ON projection_checkpoints(thread_id, completed_at)146 ON projection_turns(thread_id, checkpoint_turn_count, completed_at)
155 `;147 `;
156148
157 yield* sql`149 yield* sql`
apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts deleted-14
...@@ -1,14 +0,0 @@
1import { Effect } from "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 TABLE IF NOT EXISTS projection_pending_turn_starts (
9 thread_id TEXT PRIMARY KEY,
10 message_id TEXT NOT NULL,
11 created_at TEXT NOT NULL
12 )
13 `;
14});
apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts deleted-66
...@@ -1,66 +0,0 @@
1/**
2 * ProjectionPendingTurnStartRepository - Repository interface for pending turn starts.
3 *
4 * Owns persistence operations for projected "pending turn start" markers used
5 * while creating new turns from queued messages.
6 *
7 * @module ProjectionPendingTurnStartRepository
8 */
9import { IsoDateTime, MessageId, ThreadId } from "@t3tools/contracts";
10import { Option, Schema, ServiceMap } from "effect";
11import type { Effect } from "effect";
12
13import type { ProjectionRepositoryError } from "../Errors.ts";
14
15export const ProjectionPendingTurnStart = Schema.Struct({
16 threadId: ThreadId,
17 messageId: MessageId,
18 createdAt: IsoDateTime,
19});
20export type ProjectionPendingTurnStart = typeof ProjectionPendingTurnStart.Type;
21
22export const GetProjectionPendingTurnStartInput = Schema.Struct({
23 threadId: ThreadId,
24});
25export type GetProjectionPendingTurnStartInput = typeof GetProjectionPendingTurnStartInput.Type;
26
27export const DeleteProjectionPendingTurnStartInput = Schema.Struct({
28 threadId: ThreadId,
29});
30export type DeleteProjectionPendingTurnStartInput = typeof DeleteProjectionPendingTurnStartInput.Type;
31
32/**
33 * ProjectionPendingTurnStartRepositoryShape - Service API for pending turn-start markers.
34 */
35export interface ProjectionPendingTurnStartRepositoryShape {
36 /**
37 * Insert or replace a pending turn-start row.
38 *
39 * Maintains at most one pending marker per thread.
40 */
41 readonly upsert: (
42 row: ProjectionPendingTurnStart,
43 ) => Effect.Effect<void, ProjectionRepositoryError>;
44
45 /**
46 * Read a pending turn-start marker by thread id.
47 */
48 readonly getByThreadId: (
49 input: GetProjectionPendingTurnStartInput,
50 ) => Effect.Effect<Option.Option<ProjectionPendingTurnStart>, ProjectionRepositoryError>;
51
52 /**
53 * Delete a pending turn-start marker by thread id.
54 */
55 readonly deleteByThreadId: (
56 input: DeleteProjectionPendingTurnStartInput,
57 ) => Effect.Effect<void, ProjectionRepositoryError>;
58}
59
60/**
61 * ProjectionPendingTurnStartRepository - Service tag for pending turn-start persistence.
62 */
63export class ProjectionPendingTurnStartRepository extends ServiceMap.Service<
64 ProjectionPendingTurnStartRepository,
65 ProjectionPendingTurnStartRepositoryShape
66>()("persistence/ProjectionPendingTurnStartRepository") {}
apps/server/src/persistence/Services/ProjectionThreadTurns.ts deleted-89
...@@ -1,89 +0,0 @@
1/**
2 * ProjectionThreadTurnRepository - Projection repository interface for turns.
3 *
4 * Owns persistence operations for projected turn records in thread timelines.
5 *
6 * @module ProjectionThreadTurnRepository
7 */
8import {
9 IsoDateTime,
10 MessageId,
11 NonNegativeInt,
12 ProjectionThreadTurnStatus,
13 ThreadId,
14 TurnId,
15} from "@t3tools/contracts";
16import { Option, Schema, ServiceMap } from "effect";
17import type { Effect } from "effect";
18
19import type { ProjectionRepositoryError } from "../Errors.ts";
20
21export const ProjectionThreadTurn = Schema.Struct({
22 turnId: TurnId,
23 threadId: ThreadId,
24 turnCount: NonNegativeInt,
25 status: ProjectionThreadTurnStatus,
26 userMessageId: Schema.NullOr(MessageId),
27 assistantMessageId: Schema.NullOr(MessageId),
28 startedAt: IsoDateTime,
29 completedAt: Schema.NullOr(IsoDateTime),
30});
31export type ProjectionThreadTurn = typeof ProjectionThreadTurn.Type;
32
33export const ListProjectionThreadTurnsInput = Schema.Struct({
34 threadId: ThreadId,
35});
36export type ListProjectionThreadTurnsInput = typeof ListProjectionThreadTurnsInput.Type;
37
38export const GetProjectionThreadTurnInput = Schema.Struct({
39 turnId: TurnId,
40});
41export type GetProjectionThreadTurnInput = typeof GetProjectionThreadTurnInput.Type;
42
43export const DeleteProjectionThreadTurnsInput = Schema.Struct({
44 threadId: ThreadId,
45});
46export type DeleteProjectionThreadTurnsInput = typeof DeleteProjectionThreadTurnsInput.Type;
47
48/**
49 * ProjectionThreadTurnRepositoryShape - Service API for projected thread turns.
50 */
51export interface ProjectionThreadTurnRepositoryShape {
52 /**
53 * Insert or replace a projected turn row.
54 *
55 * Upserts by `turnId`.
56 */
57 readonly upsert: (turn: ProjectionThreadTurn) => Effect.Effect<void, ProjectionRepositoryError>;
58
59 /**
60 * List projected turns for a thread.
61 *
62 * Returned in ascending `turnCount` order.
63 */
64 readonly listByThreadId: (
65 input: ListProjectionThreadTurnsInput,
66 ) => Effect.Effect<ReadonlyArray<ProjectionThreadTurn>, ProjectionRepositoryError>;
67
68 /**
69 * Read a projected turn row by turn id.
70 */
71 readonly getByTurnId: (
72 input: GetProjectionThreadTurnInput,
73 ) => Effect.Effect<Option.Option<ProjectionThreadTurn>, ProjectionRepositoryError>;
74
75 /**
76 * Delete projected turn rows by thread.
77 */
78 readonly deleteByThreadId: (
79 input: DeleteProjectionThreadTurnsInput,
80 ) => Effect.Effect<void, ProjectionRepositoryError>;
81}
82
83/**
84 * ProjectionThreadTurnRepository - Service tag for turn projection persistence.
85 */
86export class ProjectionThreadTurnRepository extends ServiceMap.Service<
87 ProjectionThreadTurnRepository,
88 ProjectionThreadTurnRepositoryShape
89>()("persistence/ProjectionThreadTurnRepository") {}
apps/server/src/persistence/Services/ProjectionTurns.ts created+161
...@@ -0,0 +1,161 @@
1/**
2 * ProjectionTurnRepository - Projection repository interface for unified turn state.
3 *
4 * Owns persistence operations for pending starts, running/completed turn lifecycle,
5 * and checkpoint metadata in a single projection table.
6 *
7 * @module ProjectionTurnRepository
8 */
9import {
10 CheckpointRef,
11 IsoDateTime,
12 MessageId,
13 NonNegativeInt,
14 OrchestrationCheckpointFile,
15 OrchestrationCheckpointStatus,
16 ThreadId,
17 TurnId,
18} from "@t3tools/contracts";
19import { Option, Schema, ServiceMap } from "effect";
20import type { Effect } from "effect";
21
22import type { ProjectionRepositoryError } from "../Errors.ts";
23
24export const ProjectionTurnState = Schema.Literals([
25 "pending",
26 "running",
27 "interrupted",
28 "completed",
29 "error",
30]);
31export type ProjectionTurnState = typeof ProjectionTurnState.Type;
32
33export const ProjectionTurn = Schema.Struct({
34 threadId: ThreadId,
35 turnId: Schema.NullOr(TurnId),
36 pendingMessageId: Schema.NullOr(MessageId),
37 assistantMessageId: Schema.NullOr(MessageId),
38 state: ProjectionTurnState,
39 requestedAt: IsoDateTime,
40 startedAt: Schema.NullOr(IsoDateTime),
41 completedAt: Schema.NullOr(IsoDateTime),
42 checkpointTurnCount: Schema.NullOr(NonNegativeInt),
43 checkpointRef: Schema.NullOr(CheckpointRef),
44 checkpointStatus: Schema.NullOr(OrchestrationCheckpointStatus),
45 checkpointFiles: Schema.Array(OrchestrationCheckpointFile),
46});
47export type ProjectionTurn = typeof ProjectionTurn.Type;
48
49export const ProjectionTurnById = Schema.Struct({
50 threadId: ThreadId,
51 turnId: TurnId,
52 pendingMessageId: Schema.NullOr(MessageId),
53 assistantMessageId: Schema.NullOr(MessageId),
54 state: ProjectionTurnState,
55 requestedAt: IsoDateTime,
56 startedAt: Schema.NullOr(IsoDateTime),
57 completedAt: Schema.NullOr(IsoDateTime),
58 checkpointTurnCount: Schema.NullOr(NonNegativeInt),
59 checkpointRef: Schema.NullOr(CheckpointRef),
60 checkpointStatus: Schema.NullOr(OrchestrationCheckpointStatus),
61 checkpointFiles: Schema.Array(OrchestrationCheckpointFile),
62});
63export type ProjectionTurnById = typeof ProjectionTurnById.Type;
64
65export const ProjectionPendingTurnStart = Schema.Struct({
66 threadId: ThreadId,
67 messageId: MessageId,
68 requestedAt: IsoDateTime,
69});
70export type ProjectionPendingTurnStart = typeof ProjectionPendingTurnStart.Type;
71
72export const ListProjectionTurnsByThreadInput = Schema.Struct({
73 threadId: ThreadId,
74});
75export type ListProjectionTurnsByThreadInput = typeof ListProjectionTurnsByThreadInput.Type;
76
77export const GetProjectionTurnByTurnIdInput = Schema.Struct({
78 threadId: ThreadId,
79 turnId: TurnId,
80});
81export type GetProjectionTurnByTurnIdInput = typeof GetProjectionTurnByTurnIdInput.Type;
82
83export const GetProjectionPendingTurnStartInput = Schema.Struct({
84 threadId: ThreadId,
85});
86export type GetProjectionPendingTurnStartInput = typeof GetProjectionPendingTurnStartInput.Type;
87
88export const DeleteProjectionTurnsByThreadInput = Schema.Struct({
89 threadId: ThreadId,
90});
91export type DeleteProjectionTurnsByThreadInput = typeof DeleteProjectionTurnsByThreadInput.Type;
92
93export const ClearCheckpointTurnConflictInput = Schema.Struct({
94 threadId: ThreadId,
95 turnId: TurnId,
96 checkpointTurnCount: NonNegativeInt,
97});
98export type ClearCheckpointTurnConflictInput = typeof ClearCheckpointTurnConflictInput.Type;
99
100export interface ProjectionTurnRepositoryShape {
101 /**
102 * Inserts or updates the canonical row for a concrete `{threadId, turnId}` turn lifecycle state.
103 */
104 readonly upsertByTurnId: (
105 row: ProjectionTurnById,
106 ) => Effect.Effect<void, ProjectionRepositoryError>;
107
108 /**
109 * Replaces any existing pending-start placeholder rows for a thread with exactly one latest pending-start row.
110 */
111 readonly replacePendingTurnStart: (
112 row: ProjectionPendingTurnStart,
113 ) => Effect.Effect<void, ProjectionRepositoryError>;
114
115 /**
116 * Returns the newest pending-start placeholder for a thread; this is expected to be at most one row after replacement writes.
117 */
118 readonly getPendingTurnStartByThreadId: (
119 input: GetProjectionPendingTurnStartInput,
120 ) => Effect.Effect<Option.Option<ProjectionPendingTurnStart>, ProjectionRepositoryError>;
121
122 /**
123 * Deletes only pending-start placeholder rows (`turnId = null`) for a thread and leaves concrete turn rows untouched.
124 */
125 readonly deletePendingTurnStartByThreadId: (
126 input: GetProjectionPendingTurnStartInput,
127 ) => Effect.Effect<void, ProjectionRepositoryError>;
128
129 /**
130 * Lists all projection rows for a thread, including pending placeholders, with checkpoint rows ordered before non-checkpoint rows.
131 */
132 readonly listByThreadId: (
133 input: ListProjectionTurnsByThreadInput,
134 ) => Effect.Effect<ReadonlyArray<ProjectionTurn>, ProjectionRepositoryError>;
135
136 /**
137 * Looks up a concrete turn row by `{threadId, turnId}` and never returns pending placeholder rows.
138 */
139 readonly getByTurnId: (
140 input: GetProjectionTurnByTurnIdInput,
141 ) => Effect.Effect<Option.Option<ProjectionTurnById>, ProjectionRepositoryError>;
142
143 /**
144 * Clears checkpoint fields on conflicting rows that reuse the same checkpoint turn count in a thread, excluding the provided turn.
145 */
146 readonly clearCheckpointTurnConflict: (
147 input: ClearCheckpointTurnConflictInput,
148 ) => Effect.Effect<void, ProjectionRepositoryError>;
149
150 /**
151 * Hard-deletes all projection rows for a thread, including pending-start placeholders and checkpoint metadata rows.
152 */
153 readonly deleteByThreadId: (
154 input: DeleteProjectionTurnsByThreadInput,
155 ) => Effect.Effect<void, ProjectionRepositoryError>;
156}
157
158export class ProjectionTurnRepository extends ServiceMap.Service<
159 ProjectionTurnRepository,
160 ProjectionTurnRepositoryShape
161>()("persistence/ProjectionTurnRepository") {}
apps/server/src/wsServer.test.ts+1-1
...@@ -194,7 +194,7 @@ class MockTerminalManager implements TerminalManagerShape {...@@ -194,7 +194,7 @@ class MockTerminalManager implements TerminalManagerShape {
194 };194 };
195 });195 });
196196
197 readonly dispose: TerminalManagerShape["dispose"] = () => Effect.void;197 readonly dispose: TerminalManagerShape["dispose"] = Effect.void;
198}198}
199199
200function connectWs(port: number, token?: string): Promise<WebSocket> {200function connectWs(port: number, token?: string): Promise<WebSocket> {