From 1d975ad4310389b8e22bbce3d02a746854a11471 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 26 Feb 2026 10:12:42 -0800 Subject: [PATCH] migrate to single turn store --- .../Layers/ProjectionPipeline.test.ts | 148 +++++++- .../Layers/ProjectionPipeline.ts | 253 ++++++------- .../Layers/ProjectionSnapshotQuery.test.ts | 25 +- .../Layers/ProjectionSnapshotQuery.ts | 7 +- .../Layers/ProviderCommandReactor.ts | 42 ++- .../src/orchestration/projector.test.ts | 93 +++++ apps/server/src/orchestration/projector.ts | 14 +- .../Layers/ProjectionCheckpoints.ts | 113 ++++-- .../Layers/ProjectionPendingTurnStarts.ts | 92 ----- .../Layers/ProjectionThreadTurns.ts | 135 ------- .../src/persistence/Layers/ProjectionTurns.ts | 336 ++++++++++++++++++ apps/server/src/persistence/Migrations.ts | 2 - .../persistence/Migrations/005_Projections.ts | 40 +-- .../006_ProjectionPendingTurnStarts.ts | 14 - .../Services/ProjectionPendingTurnStarts.ts | 66 ---- .../Services/ProjectionThreadTurns.ts | 89 ----- .../persistence/Services/ProjectionTurns.ts | 161 +++++++++ apps/server/src/wsServer.test.ts | 2 +- 18 files changed, 1031 insertions(+), 601 deletions(-) delete mode 100644 apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts delete mode 100644 apps/server/src/persistence/Layers/ProjectionThreadTurns.ts create mode 100644 apps/server/src/persistence/Layers/ProjectionTurns.ts delete mode 100644 apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts delete mode 100644 apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts delete mode 100644 apps/server/src/persistence/Services/ProjectionThreadTurns.ts create mode 100644 apps/server/src/persistence/Services/ProjectionTurns.ts diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index a082100602670738dbcb22feb8eebb56a1c3c5c1..bede591860f06979d17620ec7a374e3453c32c2c 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -412,6 +412,146 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("resolves turn-count conflicts when checkpoint completion rewrites provisional turns", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.makeUnsafe("evt-conflict-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-conflict"), + occurredAt: "2026-02-26T13:00:00.000Z", + commandId: CommandId.makeUnsafe("cmd-conflict-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-conflict-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-conflict"), + title: "Project Conflict", + workspaceRoot: "/tmp/project-conflict", + defaultModel: null, + scripts: [], + createdAt: "2026-02-26T13:00:00.000Z", + updatedAt: "2026-02-26T13:00:00.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-conflict-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-conflict"), + occurredAt: "2026-02-26T13:00:01.000Z", + commandId: CommandId.makeUnsafe("cmd-conflict-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-conflict-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-conflict"), + projectId: ProjectId.makeUnsafe("project-conflict"), + title: "Thread Conflict", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: "2026-02-26T13:00:01.000Z", + updatedAt: "2026-02-26T13:00:01.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.turn-interrupt-requested", + eventId: EventId.makeUnsafe("evt-conflict-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-conflict"), + occurredAt: "2026-02-26T13:00:02.000Z", + commandId: CommandId.makeUnsafe("cmd-conflict-3"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-conflict-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-conflict"), + turnId: TurnId.makeUnsafe("turn-interrupted"), + createdAt: "2026-02-26T13:00:02.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-conflict-4"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-conflict"), + occurredAt: "2026-02-26T13:00:03.000Z", + commandId: CommandId.makeUnsafe("cmd-conflict-4"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-conflict-4"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-conflict"), + messageId: MessageId.makeUnsafe("assistant-conflict"), + role: "assistant", + text: "done", + turnId: TurnId.makeUnsafe("turn-completed"), + streaming: false, + createdAt: "2026-02-26T13:00:03.000Z", + updatedAt: "2026-02-26T13:00:03.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.makeUnsafe("evt-conflict-5"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-conflict"), + occurredAt: "2026-02-26T13:00:04.000Z", + commandId: CommandId.makeUnsafe("cmd-conflict-5"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-conflict-5"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-conflict"), + turnId: TurnId.makeUnsafe("turn-completed"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-conflict/turn/1"), + status: "ready", + files: [], + assistantMessageId: MessageId.makeUnsafe("assistant-conflict"), + completedAt: "2026-02-26T13:00:04.000Z", + }, + }); + + const turnRows = yield* sql<{ + readonly turnId: string; + readonly checkpointTurnCount: number | null; + readonly status: string; + }>` + SELECT + turn_id AS "turnId", + checkpoint_turn_count AS "checkpointTurnCount", + state AS "status" + FROM projection_turns + WHERE thread_id = 'thread-conflict' + ORDER BY + CASE + WHEN checkpoint_turn_count IS NULL THEN 1 + ELSE 0 + END ASC, + checkpoint_turn_count ASC, + requested_at ASC + `; + assert.deepEqual(turnRows, [ + { turnId: "turn-completed", checkpointTurnCount: 1, status: "completed" }, + { turnId: "turn-interrupted", checkpointTurnCount: null, status: "interrupted" }, + ]); + }), + ); + it.effect("does not fallback-retain messages whose turnId is removed by revert", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; @@ -697,8 +837,10 @@ it.effect("restores pending turn-start metadata across projection pipeline resta const pendingRows = yield* sql<{ readonly threadId: string }>` SELECT thread_id AS "threadId" - FROM projection_pending_turn_starts + FROM projection_turns WHERE thread_id = ${threadId} + AND turn_id IS NULL + AND state = 'pending' `; assert.deepEqual(pendingRows, []); @@ -709,9 +851,9 @@ it.effect("restores pending turn-start metadata across projection pipeline resta }>` SELECT turn_id AS "turnId", - user_message_id AS "userMessageId", + pending_message_id AS "userMessageId", started_at AS "startedAt" - FROM projection_thread_turns + FROM projection_turns WHERE turn_id = ${turnId} `; }).pipe(Effect.provide(secondProjectionLayer)); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 81d0eb95d93bea950391d731a77b9eff1c20c820..562190d14368e4270898443c12e9f043a32505f3 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -4,9 +4,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; -import { ProjectionCheckpointRepository } from "../../persistence/Services/ProjectionCheckpoints.ts"; import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts"; -import { ProjectionPendingTurnStartRepository } from "../../persistence/Services/ProjectionPendingTurnStarts.ts"; import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts"; import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts"; import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts"; @@ -16,18 +14,15 @@ import { ProjectionThreadMessageRepository, } from "../../persistence/Services/ProjectionThreadMessages.ts"; import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts"; -import { type ProjectionThreadTurn } from "../../persistence/Services/ProjectionThreadTurns.ts"; -import { ProjectionThreadTurnRepository } from "../../persistence/Services/ProjectionThreadTurns.ts"; +import { type ProjectionTurn, ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts"; import { ProjectionThreadRepository } from "../../persistence/Services/ProjectionThreads.ts"; -import { ProjectionCheckpointRepositoryLive } from "../../persistence/Layers/ProjectionCheckpoints.ts"; import { ProjectionPendingApprovalRepositoryLive } from "../../persistence/Layers/ProjectionPendingApprovals.ts"; -import { ProjectionPendingTurnStartRepositoryLive } from "../../persistence/Layers/ProjectionPendingTurnStarts.ts"; import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/ProjectionProjects.ts"; import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts"; import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts"; import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts"; import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts"; -import { ProjectionThreadTurnRepositoryLive } from "../../persistence/Layers/ProjectionThreadTurns.ts"; +import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts"; import { OrchestrationProjectionPipeline, @@ -61,22 +56,22 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null { return typeof requestId === "string" ? ApprovalRequestId.makeUnsafe(requestId) : null; } -function nextTurnCount(turns: ReadonlyArray): number { - return turns.reduce((maxTurnCount, turn) => Math.max(maxTurnCount, turn.turnCount), 0) + 1; -} - function retainProjectionMessagesAfterRevert( messages: ReadonlyArray, - turns: ReadonlyArray, + turns: ReadonlyArray, turnCount: number, ): ReadonlyArray { const retainedMessageIds = new Set(); const retainedTurnIds = new Set(); - const keptTurns = turns.filter((turn) => turn.turnCount <= turnCount); + const keptTurns = turns.filter( + (turn) => turn.turnId !== null && turn.checkpointTurnCount !== null && turn.checkpointTurnCount <= turnCount, + ); for (const turn of keptTurns) { - retainedTurnIds.add(turn.turnId); - if (turn.userMessageId !== null) { - retainedMessageIds.add(turn.userMessageId); + if (turn.turnId !== null) { + retainedTurnIds.add(turn.turnId); + } + if (turn.pendingMessageId !== null) { + retainedMessageIds.add(turn.pendingMessageId); } if (turn.assistantMessageId !== null) { retainedMessageIds.add(turn.assistantMessageId); @@ -144,11 +139,18 @@ function retainProjectionMessagesAfterRevert( function retainProjectionActivitiesAfterRevert( activities: ReadonlyArray, - turns: ReadonlyArray, + turns: ReadonlyArray, turnCount: number, ): ReadonlyArray { - const retainedTurnIds = new Set( - turns.filter((turn) => turn.turnCount <= turnCount).map((turn) => turn.turnId), + const retainedTurnIds = new Set( + turns + .filter( + (turn) => + turn.turnId !== null && + turn.checkpointTurnCount !== null && + turn.checkpointTurnCount <= turnCount, + ) + .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])), ); return activities.filter( (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId), @@ -164,10 +166,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository; const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository; const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository; - const projectionThreadTurnRepository = yield* ProjectionThreadTurnRepository; - const projectionCheckpointRepository = yield* ProjectionCheckpointRepository; + const projectionTurnRepository = yield* ProjectionTurnRepository; const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository; - const projectionPendingTurnStartRepository = yield* ProjectionPendingTurnStartRepository; const applyProjectsProjection: ProjectorDefinition["apply"] = (event) => Effect.gen(function* () { @@ -382,7 +382,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { return; } - const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({ + const existingTurns = yield* projectionTurnRepository.listByThreadId({ threadId: event.payload.threadId, }); const keptRows = retainProjectionMessagesAfterRevert( @@ -431,7 +431,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { if (existingRows.length === 0) { return; } - const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({ + const existingTurns = yield* projectionTurnRepository.listByThreadId({ threadId: event.payload.threadId, }); const keptRows = retainProjectionActivitiesAfterRevert( @@ -477,10 +477,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { Effect.gen(function* () { switch (event.type) { case "thread.turn-start-requested": { - yield* projectionPendingTurnStartRepository.upsert({ + yield* projectionTurnRepository.replacePendingTurnStart({ threadId: event.payload.threadId, messageId: event.payload.messageId, - createdAt: event.payload.createdAt, + requestedAt: event.payload.createdAt, }); return; } @@ -491,38 +491,53 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { return; } - const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({ turnId }); - const pendingTurnStart = yield* projectionPendingTurnStartRepository.getByThreadId({ + const existingTurn = yield* projectionTurnRepository.getByTurnId({ + threadId: event.payload.threadId, + turnId, + }); + const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({ threadId: event.payload.threadId, }); if (Option.isSome(existingTurn)) { - yield* projectionThreadTurnRepository.upsert({ + const nextState = + existingTurn.value.state === "completed" || existingTurn.value.state === "error" + ? existingTurn.value.state + : "running"; + yield* projectionTurnRepository.upsertByTurnId({ ...existingTurn.value, - status: existingTurn.value.status === "completed" ? "completed" : "running", - userMessageId: - existingTurn.value.userMessageId ?? + state: nextState, + pendingMessageId: + existingTurn.value.pendingMessageId ?? (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null), - startedAt: existingTurn.value.startedAt, + startedAt: + existingTurn.value.startedAt ?? + (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.requestedAt : event.occurredAt), + requestedAt: + existingTurn.value.requestedAt ?? + (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.requestedAt : event.occurredAt), }); } else { - const existingThreadTurns = yield* projectionThreadTurnRepository.listByThreadId({ - threadId: event.payload.threadId, - }); - yield* projectionThreadTurnRepository.upsert({ + yield* projectionTurnRepository.upsertByTurnId({ turnId, threadId: event.payload.threadId, - turnCount: nextTurnCount(existingThreadTurns), - status: "running", - userMessageId: Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null, + pendingMessageId: Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null, assistantMessageId: null, + state: "running", + requestedAt: Option.isSome(pendingTurnStart) + ? pendingTurnStart.value.requestedAt + : event.occurredAt, startedAt: Option.isSome(pendingTurnStart) - ? pendingTurnStart.value.createdAt + ? pendingTurnStart.value.requestedAt : event.occurredAt, completedAt: null, + checkpointTurnCount: null, + checkpointRef: null, + checkpointStatus: null, + checkpointFiles: [], }); } - yield* projectionPendingTurnStartRepository.deleteByThreadId({ + yield* projectionTurnRepository.deletePendingTurnStartByThreadId({ threadId: event.payload.threadId, }); return; @@ -532,38 +547,42 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { if (event.payload.turnId === null || event.payload.role !== "assistant") { return; } - const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({ + const existingTurn = yield* projectionTurnRepository.getByTurnId({ + threadId: event.payload.threadId, turnId: event.payload.turnId, }); if (Option.isSome(existingTurn)) { - yield* projectionThreadTurnRepository.upsert({ + yield* projectionTurnRepository.upsertByTurnId({ ...existingTurn.value, assistantMessageId: event.payload.messageId, - status: event.payload.streaming - ? existingTurn.value.status - : existingTurn.value.status === "interrupted" + state: event.payload.streaming + ? existingTurn.value.state + : existingTurn.value.state === "interrupted" ? "interrupted" - : existingTurn.value.status === "error" + : existingTurn.value.state === "error" ? "error" : "completed", completedAt: event.payload.streaming ? existingTurn.value.completedAt : (existingTurn.value.completedAt ?? event.payload.updatedAt), + startedAt: existingTurn.value.startedAt ?? event.payload.createdAt, + requestedAt: existingTurn.value.requestedAt ?? event.payload.createdAt, }); return; } - const existingThreadTurns = yield* projectionThreadTurnRepository.listByThreadId({ - threadId: event.payload.threadId, - }); - yield* projectionThreadTurnRepository.upsert({ + yield* projectionTurnRepository.upsertByTurnId({ turnId: event.payload.turnId, threadId: event.payload.threadId, - turnCount: nextTurnCount(existingThreadTurns), - status: event.payload.streaming ? "running" : "completed", - userMessageId: null, + pendingMessageId: null, assistantMessageId: event.payload.messageId, + state: event.payload.streaming ? "running" : "completed", + requestedAt: event.payload.createdAt, startedAt: event.payload.createdAt, completedAt: event.payload.streaming ? null : event.payload.updatedAt, + checkpointTurnCount: null, + checkpointRef: null, + checkpointStatus: null, + checkpointFiles: [], }); return; } @@ -572,111 +591,105 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { if (event.payload.turnId === undefined) { return; } - const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({ + const existingTurn = yield* projectionTurnRepository.getByTurnId({ + threadId: event.payload.threadId, turnId: event.payload.turnId, }); if (Option.isSome(existingTurn)) { - yield* projectionThreadTurnRepository.upsert({ + yield* projectionTurnRepository.upsertByTurnId({ ...existingTurn.value, - status: "interrupted", + state: "interrupted", completedAt: existingTurn.value.completedAt ?? event.payload.createdAt, + startedAt: existingTurn.value.startedAt ?? event.payload.createdAt, + requestedAt: existingTurn.value.requestedAt ?? event.payload.createdAt, }); return; } - const existingThreadTurns = yield* projectionThreadTurnRepository.listByThreadId({ - threadId: event.payload.threadId, - }); - yield* projectionThreadTurnRepository.upsert({ + yield* projectionTurnRepository.upsertByTurnId({ turnId: event.payload.turnId, threadId: event.payload.threadId, - turnCount: nextTurnCount(existingThreadTurns), - status: "interrupted", - userMessageId: null, + pendingMessageId: null, assistantMessageId: null, + state: "interrupted", + requestedAt: event.payload.createdAt, startedAt: event.payload.createdAt, completedAt: event.payload.createdAt, + checkpointTurnCount: null, + checkpointRef: null, + checkpointStatus: null, + checkpointFiles: [], }); return; } case "thread.turn-diff-completed": { - const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({ + const existingTurn = yield* projectionTurnRepository.getByTurnId({ + threadId: event.payload.threadId, turnId: event.payload.turnId, }); - const nextStatus = event.payload.status === "error" ? "error" : "completed"; + const nextState = event.payload.status === "error" ? "error" : "completed"; + yield* projectionTurnRepository.clearCheckpointTurnConflict({ + threadId: event.payload.threadId, + turnId: event.payload.turnId, + checkpointTurnCount: event.payload.checkpointTurnCount, + }); + if (Option.isSome(existingTurn)) { - yield* projectionThreadTurnRepository.upsert({ + yield* projectionTurnRepository.upsertByTurnId({ ...existingTurn.value, - turnCount: event.payload.checkpointTurnCount, - status: nextStatus, assistantMessageId: event.payload.assistantMessageId, + state: nextState, + checkpointTurnCount: event.payload.checkpointTurnCount, + checkpointRef: event.payload.checkpointRef, + checkpointStatus: event.payload.status, + checkpointFiles: event.payload.files, + startedAt: existingTurn.value.startedAt ?? event.payload.completedAt, + requestedAt: existingTurn.value.requestedAt ?? event.payload.completedAt, completedAt: event.payload.completedAt, }); return; } - yield* projectionThreadTurnRepository.upsert({ + yield* projectionTurnRepository.upsertByTurnId({ turnId: event.payload.turnId, threadId: event.payload.threadId, - turnCount: event.payload.checkpointTurnCount, - status: nextStatus, - userMessageId: null, + pendingMessageId: null, assistantMessageId: event.payload.assistantMessageId, + state: nextState, + requestedAt: event.payload.completedAt, startedAt: event.payload.completedAt, completedAt: event.payload.completedAt, - }); - return; - } - - case "thread.reverted": { - const existingTurns = yield* projectionThreadTurnRepository.listByThreadId({ - threadId: event.payload.threadId, - }); - const keptTurns = existingTurns.filter( - (turn) => turn.turnCount <= event.payload.turnCount, - ); - yield* projectionThreadTurnRepository.deleteByThreadId({ - threadId: event.payload.threadId, - }); - yield* Effect.forEach(keptTurns, projectionThreadTurnRepository.upsert, { - concurrency: 1, - }).pipe(Effect.asVoid); - return; - } - - default: - return; - } - }); - - const applyCheckpointsProjection: ProjectorDefinition["apply"] = (event) => - Effect.gen(function* () { - switch (event.type) { - case "thread.turn-diff-completed": - yield* projectionCheckpointRepository.upsert({ - threadId: event.payload.threadId, - turnId: event.payload.turnId, checkpointTurnCount: event.payload.checkpointTurnCount, checkpointRef: event.payload.checkpointRef, - status: event.payload.status, - files: event.payload.files, - assistantMessageId: event.payload.assistantMessageId, - completedAt: event.payload.completedAt, + checkpointStatus: event.payload.status, + checkpointFiles: event.payload.files, }); return; + } case "thread.reverted": { - const existingCheckpoints = yield* projectionCheckpointRepository.listByThreadId({ + const existingTurns = yield* projectionTurnRepository.listByThreadId({ threadId: event.payload.threadId, }); - const keptCheckpoints = existingCheckpoints.filter( - (checkpoint) => checkpoint.checkpointTurnCount <= event.payload.turnCount, + const keptTurns = existingTurns.filter( + (turn) => + turn.turnId !== null && + turn.checkpointTurnCount !== null && + turn.checkpointTurnCount <= event.payload.turnCount, ); - yield* projectionCheckpointRepository.deleteByThreadId({ + yield* projectionTurnRepository.deleteByThreadId({ threadId: event.payload.threadId, }); - yield* Effect.forEach(keptCheckpoints, projectionCheckpointRepository.upsert, { - concurrency: 1, - }).pipe(Effect.asVoid); + yield* Effect.forEach( + keptTurns, + (turn) => + turn.turnId === null + ? Effect.void + : projectionTurnRepository.upsertByTurnId({ + ...turn, + turnId: turn.turnId, + }), + { concurrency: 1 }, + ).pipe(Effect.asVoid); return; } @@ -685,6 +698,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } }); + const applyCheckpointsProjection: ProjectorDefinition["apply"] = () => Effect.void; + const applyPendingApprovalsProjection: ProjectorDefinition["apply"] = (event) => Effect.gen(function* () { switch (event.type) { @@ -877,9 +892,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect( Layer.provideMerge(ProjectionThreadMessageRepositoryLive), Layer.provideMerge(ProjectionThreadActivityRepositoryLive), Layer.provideMerge(ProjectionThreadSessionRepositoryLive), - Layer.provideMerge(ProjectionThreadTurnRepositoryLive), - Layer.provideMerge(ProjectionCheckpointRepositoryLive), + Layer.provideMerge(ProjectionTurnRepositoryLive), Layer.provideMerge(ProjectionPendingApprovalRepositoryLive), - Layer.provideMerge(ProjectionPendingTurnStartRepositoryLive), Layer.provideMerge(ProjectionStateRepositoryLive), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index f19b2891cce1d94214dd0bce15f8641cc18f1147..824f9a7cfd67f272eec0a6d100c866cf1fc94042 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -38,6 +38,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { yield* sql`DELETE FROM projection_projects`; yield* sql`DELETE FROM projection_state`; + yield* sql`DELETE FROM projection_turns`; yield* sql` INSERT INTO projection_projects ( @@ -159,25 +160,33 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { `; yield* sql` - INSERT INTO projection_checkpoints ( + INSERT INTO projection_turns ( thread_id, turn_id, + pending_message_id, + assistant_message_id, + state, + requested_at, + started_at, + completed_at, checkpoint_turn_count, checkpoint_ref, - status, - files_json, - assistant_message_id, - completed_at + checkpoint_status, + checkpoint_files_json ) VALUES ( 'thread-1', 'turn-1', + NULL, + 'message-1', + 'completed', + '2026-02-24T00:00:08.000Z', + '2026-02-24T00:00:08.000Z', + '2026-02-24T00:00:08.000Z', 1, 'checkpoint-1', 'ready', - '[{"path":"README.md","kind":"modified","additions":2,"deletions":1}]', - 'message-1', - '2026-02-24T00:00:08.000Z' + '[{"path":"README.md","kind":"modified","additions":2,"deletions":1}]' ) `; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 2c2b77001bb490fb067dd5d11018ffd9701dd199..babf83171a63d85cc8a9e38ca8de0fc91e19ed1e 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -214,11 +214,12 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { turn_id AS "turnId", checkpoint_turn_count AS "checkpointTurnCount", checkpoint_ref AS "checkpointRef", - status, - files_json AS "files", + checkpoint_status AS "status", + checkpoint_files_json AS "files", assistant_message_id AS "assistantMessageId", completed_at AS "completedAt" - FROM projection_checkpoints + FROM projection_turns + WHERE checkpoint_turn_count IS NOT NULL ORDER BY thread_id ASC, checkpoint_turn_count ASC `, }); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 1b09cc0a409606fc666f90173f767c6e434f1ffc..6be2c8af31d3c9a83d04b13cb523c922403da3e4 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -75,10 +75,47 @@ const turnStartKeyForEvent = (event: ProviderIntentEvent): string => const serverCommandId = (tag: string): CommandId => CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`); +const HANDLED_TURN_START_KEY_MAX = 10_000; +const HANDLED_TURN_START_KEY_TTL_MS = 30 * 60 * 1_000; + +function pruneHandledTurnStartKeys(handledKeys: Map, nowMs: number): void { + const staleBefore = nowMs - HANDLED_TURN_START_KEY_TTL_MS; + for (const [key, seenAt] of handledKeys) { + if (seenAt >= staleBefore) { + break; + } + handledKeys.delete(key); + } + + while (handledKeys.size > HANDLED_TURN_START_KEY_MAX) { + const oldestKey = handledKeys.keys().next().value; + if (!oldestKey) { + break; + } + handledKeys.delete(oldestKey); + } +} + +function hasHandledTurnStartRecently( + handledKeys: Map, + key: string, + nowMs = Date.now(), +): boolean { + const seenAt = handledKeys.get(key); + if (seenAt !== undefined && nowMs - seenAt <= HANDLED_TURN_START_KEY_TTL_MS) { + handledKeys.set(key, nowMs); + return true; + } + + handledKeys.set(key, nowMs); + pruneHandledTurnStartKeys(handledKeys, nowMs); + return false; +} + const make = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const providerService = yield* ProviderService; - const handledTurnStartKeys = new Set(); + const handledTurnStartKeys = new Map(); const appendProviderFailureActivity = (input: { readonly threadId: ThreadId; @@ -209,10 +246,9 @@ const make = Effect.gen(function* () { event: Extract, ) { const key = turnStartKeyForEvent(event); - if (handledTurnStartKeys.has(key)) { + if (hasHandledTurnStartRecently(handledTurnStartKeys, key)) { return; } - handledTurnStartKeys.add(key); const thread = yield* resolveThread(event.payload.threadId); if (!thread) { diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 770744bc9f91852c5291727cbfa32731e5f75722..4f415e31a3b7713f4906505e08053ae1d0821c5d 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -651,4 +651,97 @@ describe("orchestration projector", () => { })), ).toEqual([{ id: "assistant-keep", role: "assistant", turnId: "turn-1" }]); }); + + it("caps message and checkpoint retention for long-lived threads", async () => { + const createdAt = "2026-03-01T10:00:00.000Z"; + const model = createEmptyReadModel(createdAt); + + const afterCreate = await Effect.runPromise( + projectEvent( + model, + makeEvent({ + sequence: 1, + type: "thread.created", + aggregateKind: "thread", + aggregateId: "thread-capped", + occurredAt: createdAt, + commandId: "cmd-create-capped", + payload: { + threadId: "thread-capped", + projectId: "project-1", + title: "capped", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }), + ), + ); + + const messageEvents: ReadonlyArray = Array.from( + { length: 2_100 }, + (_, index) => + makeEvent({ + sequence: index + 2, + type: "thread.message-sent", + aggregateKind: "thread", + aggregateId: "thread-capped", + occurredAt: `2026-03-01T10:00:${String(index % 60).padStart(2, "0")}.000Z`, + commandId: `cmd-message-${index}`, + payload: { + threadId: "thread-capped", + messageId: `msg-${index}`, + role: "assistant", + text: `message-${index}`, + turnId: `turn-${index}`, + streaming: false, + createdAt: `2026-03-01T10:00:${String(index % 60).padStart(2, "0")}.000Z`, + updatedAt: `2026-03-01T10:00:${String(index % 60).padStart(2, "0")}.000Z`, + }, + }), + ); + const afterMessages = await messageEvents.reduce>>( + (statePromise, event) => + statePromise.then((state) => Effect.runPromise(projectEvent(state, event))), + Promise.resolve(afterCreate), + ); + + const checkpointEvents: ReadonlyArray = Array.from( + { length: 600 }, + (_, index) => + makeEvent({ + sequence: index + 2_102, + type: "thread.turn-diff-completed", + aggregateKind: "thread", + aggregateId: "thread-capped", + occurredAt: `2026-03-01T10:30:${String(index % 60).padStart(2, "0")}.000Z`, + commandId: `cmd-checkpoint-${index}`, + payload: { + threadId: "thread-capped", + turnId: `turn-${index}`, + checkpointTurnCount: index + 1, + checkpointRef: `refs/t3/checkpoints/thread-capped/turn/${index + 1}`, + status: "ready", + files: [], + assistantMessageId: `msg-${index}`, + completedAt: `2026-03-01T10:30:${String(index % 60).padStart(2, "0")}.000Z`, + }, + }), + ); + const finalState = await checkpointEvents.reduce>>( + (statePromise, event) => + statePromise.then((state) => Effect.runPromise(projectEvent(state, event))), + Promise.resolve(afterMessages), + ); + + const thread = finalState.threads[0]; + expect(thread?.messages).toHaveLength(2_000); + expect(thread?.messages[0]?.id).toBe("msg-100"); + expect(thread?.messages.at(-1)?.id).toBe("msg-2099"); + expect(thread?.checkpoints).toHaveLength(500); + expect(thread?.checkpoints[0]?.turnId).toBe("turn-100"); + expect(thread?.checkpoints.at(-1)?.turnId).toBe("turn-599"); + }); }); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index d17722d8c93dbea42b9187d01d4bcab5e98187c4..296cac2cbeaefab204237d696fca7acf3c189bd0 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -23,6 +23,8 @@ import { } from "./Schemas.ts"; type ThreadPatch = Partial>; +const MAX_THREAD_MESSAGES = 2_000; +const MAX_THREAD_CHECKPOINTS = 500; function updateThread( threads: ReadonlyArray, @@ -313,11 +315,12 @@ export function projectEvent( : entry, ) : [...thread.messages, message]; + const cappedMessages = messages.slice(-MAX_THREAD_MESSAGES); return { ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { - messages, + messages: cappedMessages, updatedAt: event.occurredAt, }), }; @@ -384,7 +387,9 @@ export function projectEvent( const checkpoints = [ ...thread.checkpoints.filter((entry) => entry.turnId !== checkpoint.turnId), checkpoint, - ].toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount); + ] + .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount) + .slice(-MAX_THREAD_CHECKPOINTS); return { ...nextBase, @@ -406,13 +411,14 @@ export function projectEvent( const checkpoints = thread.checkpoints .filter((entry) => entry.checkpointTurnCount <= payload.turnCount) - .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount); + .toSorted((left, right) => left.checkpointTurnCount - right.checkpointTurnCount) + .slice(-MAX_THREAD_CHECKPOINTS); const retainedTurnIds = new Set(checkpoints.map((checkpoint) => checkpoint.turnId)); const messages = retainThreadMessagesAfterRevert( thread.messages, retainedTurnIds, payload.turnCount, - ); + ).slice(-MAX_THREAD_MESSAGES); const activities = retainThreadActivitiesAfterRevert(thread.activities, retainedTurnIds); const latestTurnId = diff --git a/apps/server/src/persistence/Layers/ProjectionCheckpoints.ts b/apps/server/src/persistence/Layers/ProjectionCheckpoints.ts index 0a36aba76caf552d57a4ee5dbd2419b01df580aa..26ef8fed1b9e681a8bed1375ec60edade2e59385 100644 --- a/apps/server/src/persistence/Layers/ProjectionCheckpoints.ts +++ b/apps/server/src/persistence/Layers/ProjectionCheckpoints.ts @@ -29,39 +29,63 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st const makeProjectionCheckpointRepository = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; + const clearCheckpointConflict = SqlSchema.void({ + Request: GetByThreadAndTurnCountInput, + execute: ({ threadId, checkpointTurnCount }) => + sql` + UPDATE projection_turns + SET + checkpoint_turn_count = NULL, + checkpoint_ref = NULL, + checkpoint_status = NULL, + checkpoint_files_json = '[]' + WHERE thread_id = ${threadId} + AND checkpoint_turn_count = ${checkpointTurnCount} + `, + }); + const upsertProjectionCheckpointRow = SqlSchema.void({ Request: ProjectionCheckpointDbRowSchema, execute: (row) => sql` - INSERT INTO projection_checkpoints ( - thread_id, - turn_id, - checkpoint_turn_count, - checkpoint_ref, - status, - files_json, - assistant_message_id, - completed_at - ) - VALUES ( - ${row.threadId}, - ${row.turnId}, - ${row.checkpointTurnCount}, - ${row.checkpointRef}, - ${row.status}, - ${row.files}, - ${row.assistantMessageId}, - ${row.completedAt} - ) - ON CONFLICT (thread_id, checkpoint_turn_count) - DO UPDATE SET - turn_id = excluded.turn_id, - checkpoint_ref = excluded.checkpoint_ref, - status = excluded.status, - files_json = excluded.files_json, - assistant_message_id = excluded.assistant_message_id, - completed_at = excluded.completed_at - `, + INSERT INTO projection_turns ( + thread_id, + turn_id, + pending_message_id, + assistant_message_id, + state, + requested_at, + started_at, + completed_at, + checkpoint_turn_count, + checkpoint_ref, + checkpoint_status, + checkpoint_files_json + ) + VALUES ( + ${row.threadId}, + ${row.turnId}, + NULL, + ${row.assistantMessageId}, + ${row.status === "error" ? "error" : "completed"}, + ${row.completedAt}, + ${row.completedAt}, + ${row.completedAt}, + ${row.checkpointTurnCount}, + ${row.checkpointRef}, + ${row.status}, + ${row.files} + ) + ON CONFLICT (thread_id, turn_id) + DO UPDATE SET + assistant_message_id = excluded.assistant_message_id, + state = excluded.state, + completed_at = excluded.completed_at, + checkpoint_turn_count = excluded.checkpoint_turn_count, + checkpoint_ref = excluded.checkpoint_ref, + checkpoint_status = excluded.checkpoint_status, + checkpoint_files_json = excluded.checkpoint_files_json + `, }); const listProjectionCheckpointRows = SqlSchema.findAll({ @@ -74,12 +98,13 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () { turn_id AS "turnId", checkpoint_turn_count AS "checkpointTurnCount", checkpoint_ref AS "checkpointRef", - status, - files_json AS "files", + checkpoint_status AS "status", + checkpoint_files_json AS "files", assistant_message_id AS "assistantMessageId", completed_at AS "completedAt" - FROM projection_checkpoints + FROM projection_turns WHERE thread_id = ${threadId} + AND checkpoint_turn_count IS NOT NULL ORDER BY checkpoint_turn_count ASC `, }); @@ -94,11 +119,11 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () { turn_id AS "turnId", checkpoint_turn_count AS "checkpointTurnCount", checkpoint_ref AS "checkpointRef", - status, - files_json AS "files", + checkpoint_status AS "status", + checkpoint_files_json AS "files", assistant_message_id AS "assistantMessageId", completed_at AS "completedAt" - FROM projection_checkpoints + FROM projection_turns WHERE thread_id = ${threadId} AND checkpoint_turn_count = ${checkpointTurnCount} `, @@ -108,13 +133,27 @@ const makeProjectionCheckpointRepository = Effect.gen(function* () { Request: DeleteByThreadIdInput, execute: ({ threadId }) => sql` - DELETE FROM projection_checkpoints + UPDATE projection_turns + SET + checkpoint_turn_count = NULL, + checkpoint_ref = NULL, + checkpoint_status = NULL, + checkpoint_files_json = '[]' WHERE thread_id = ${threadId} + AND checkpoint_turn_count IS NOT NULL `, }); + const upsertCheckpointRow = (row: Schema.Schema.Type) => + sql.withTransaction( + clearCheckpointConflict({ + threadId: row.threadId, + checkpointTurnCount: row.checkpointTurnCount, + }).pipe(Effect.flatMap(() => upsertProjectionCheckpointRow(row))), + ); + const upsert: ProjectionCheckpointRepositoryShape["upsert"] = (row) => - upsertProjectionCheckpointRow(row).pipe( + upsertCheckpointRow(row).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionCheckpointRepository.upsert:query", diff --git a/apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts b/apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts deleted file mode 100644 index 7d46dfdf5daac2f0b0d07c50e99fbb0c0b8182ba..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts +++ /dev/null @@ -1,92 +0,0 @@ -import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqlSchema from "effect/unstable/sql/SqlSchema"; -import { Effect, Layer } from "effect"; - -import { toPersistenceSqlError } from "../Errors.ts"; -import { - DeleteProjectionPendingTurnStartInput, - GetProjectionPendingTurnStartInput, - ProjectionPendingTurnStart, - ProjectionPendingTurnStartRepository, - type ProjectionPendingTurnStartRepositoryShape, -} from "../Services/ProjectionPendingTurnStarts.ts"; - -const makeProjectionPendingTurnStartRepository = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - const upsertProjectionPendingTurnStartRow = SqlSchema.void({ - Request: ProjectionPendingTurnStart, - execute: (row) => - sql` - INSERT INTO projection_pending_turn_starts ( - thread_id, - message_id, - created_at - ) - VALUES ( - ${row.threadId}, - ${row.messageId}, - ${row.createdAt} - ) - ON CONFLICT (thread_id) - DO UPDATE SET - message_id = excluded.message_id, - created_at = excluded.created_at - `, - }); - - const getProjectionPendingTurnStartRow = SqlSchema.findOneOption({ - Request: GetProjectionPendingTurnStartInput, - Result: ProjectionPendingTurnStart, - execute: ({ threadId }) => - sql` - SELECT - thread_id AS "threadId", - message_id AS "messageId", - created_at AS "createdAt" - FROM projection_pending_turn_starts - WHERE thread_id = ${threadId} - `, - }); - - const deleteProjectionPendingTurnStartRow = SqlSchema.void({ - Request: DeleteProjectionPendingTurnStartInput, - execute: ({ threadId }) => - sql` - DELETE FROM projection_pending_turn_starts - WHERE thread_id = ${threadId} - `, - }); - - const upsert: ProjectionPendingTurnStartRepositoryShape["upsert"] = (row) => - upsertProjectionPendingTurnStartRow(row).pipe( - Effect.mapError(toPersistenceSqlError("ProjectionPendingTurnStartRepository.upsert:query")), - ); - - const getByThreadId: ProjectionPendingTurnStartRepositoryShape["getByThreadId"] = (input) => - getProjectionPendingTurnStartRow(input).pipe( - Effect.mapError( - toPersistenceSqlError("ProjectionPendingTurnStartRepository.getByThreadId:query"), - ), - ); - - const deleteByThreadId: ProjectionPendingTurnStartRepositoryShape["deleteByThreadId"] = ( - input, - ) => - deleteProjectionPendingTurnStartRow(input).pipe( - Effect.mapError( - toPersistenceSqlError("ProjectionPendingTurnStartRepository.deleteByThreadId:query"), - ), - ); - - return { - upsert, - getByThreadId, - deleteByThreadId, - } satisfies ProjectionPendingTurnStartRepositoryShape; -}); - -export const ProjectionPendingTurnStartRepositoryLive = Layer.effect( - ProjectionPendingTurnStartRepository, - makeProjectionPendingTurnStartRepository, -); diff --git a/apps/server/src/persistence/Layers/ProjectionThreadTurns.ts b/apps/server/src/persistence/Layers/ProjectionThreadTurns.ts deleted file mode 100644 index f402ea93c64fc82ecc7d8b8e7c7bfbf711bc04fa..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Layers/ProjectionThreadTurns.ts +++ /dev/null @@ -1,135 +0,0 @@ -import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqlSchema from "effect/unstable/sql/SqlSchema"; -import { Effect, Layer } from "effect"; - -import { toPersistenceSqlError } from "../Errors.ts"; -import { - ProjectionThreadTurn, - ProjectionThreadTurnRepository, - type ProjectionThreadTurnRepositoryShape, - DeleteProjectionThreadTurnsInput, - GetProjectionThreadTurnInput, - ListProjectionThreadTurnsInput, -} from "../Services/ProjectionThreadTurns.ts"; - -const makeProjectionThreadTurnRepository = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - const upsertProjectionThreadTurnRow = SqlSchema.void({ - Request: ProjectionThreadTurn, - execute: (row) => - sql` - INSERT INTO projection_thread_turns ( - turn_id, - thread_id, - turn_count, - status, - user_message_id, - assistant_message_id, - started_at, - completed_at - ) - VALUES ( - ${row.turnId}, - ${row.threadId}, - ${row.turnCount}, - ${row.status}, - ${row.userMessageId}, - ${row.assistantMessageId}, - ${row.startedAt}, - ${row.completedAt} - ) - ON CONFLICT (turn_id) - DO UPDATE SET - thread_id = excluded.thread_id, - turn_count = excluded.turn_count, - status = excluded.status, - user_message_id = excluded.user_message_id, - assistant_message_id = excluded.assistant_message_id, - started_at = excluded.started_at, - completed_at = excluded.completed_at - `, - }); - - const listProjectionThreadTurnRows = SqlSchema.findAll({ - Request: ListProjectionThreadTurnsInput, - Result: ProjectionThreadTurn, - execute: ({ threadId }) => - sql` - SELECT - turn_id AS "turnId", - thread_id AS "threadId", - turn_count AS "turnCount", - status, - user_message_id AS "userMessageId", - assistant_message_id AS "assistantMessageId", - started_at AS "startedAt", - completed_at AS "completedAt" - FROM projection_thread_turns - WHERE thread_id = ${threadId} - ORDER BY turn_count ASC - `, - }); - - const getProjectionThreadTurnRow = SqlSchema.findOneOption({ - Request: GetProjectionThreadTurnInput, - Result: ProjectionThreadTurn, - execute: ({ turnId }) => - sql` - SELECT - turn_id AS "turnId", - thread_id AS "threadId", - turn_count AS "turnCount", - status, - user_message_id AS "userMessageId", - assistant_message_id AS "assistantMessageId", - started_at AS "startedAt", - completed_at AS "completedAt" - FROM projection_thread_turns - WHERE turn_id = ${turnId} - `, - }); - - const deleteProjectionThreadTurnRows = SqlSchema.void({ - Request: DeleteProjectionThreadTurnsInput, - execute: ({ threadId }) => - sql` - DELETE FROM projection_thread_turns - WHERE thread_id = ${threadId} - `, - }); - - const upsert: ProjectionThreadTurnRepositoryShape["upsert"] = (row) => - upsertProjectionThreadTurnRow(row).pipe( - Effect.mapError(toPersistenceSqlError("ProjectionThreadTurnRepository.upsert:query")), - ); - - const listByThreadId: ProjectionThreadTurnRepositoryShape["listByThreadId"] = (input) => - listProjectionThreadTurnRows(input).pipe( - Effect.mapError(toPersistenceSqlError("ProjectionThreadTurnRepository.listByThreadId:query")), - ); - - const getByTurnId: ProjectionThreadTurnRepositoryShape["getByTurnId"] = (input) => - getProjectionThreadTurnRow(input).pipe( - Effect.mapError(toPersistenceSqlError("ProjectionThreadTurnRepository.getByTurnId:query")), - ); - - const deleteByThreadId: ProjectionThreadTurnRepositoryShape["deleteByThreadId"] = (input) => - deleteProjectionThreadTurnRows(input).pipe( - Effect.mapError( - toPersistenceSqlError("ProjectionThreadTurnRepository.deleteByThreadId:query"), - ), - ); - - return { - upsert, - listByThreadId, - getByTurnId, - deleteByThreadId, - } satisfies ProjectionThreadTurnRepositoryShape; -}); - -export const ProjectionThreadTurnRepositoryLive = Layer.effect( - ProjectionThreadTurnRepository, - makeProjectionThreadTurnRepository, -); diff --git a/apps/server/src/persistence/Layers/ProjectionTurns.ts b/apps/server/src/persistence/Layers/ProjectionTurns.ts new file mode 100644 index 0000000000000000000000000000000000000000..4d25fd5c21f21e50c871ecca387f6891f4483bc5 --- /dev/null +++ b/apps/server/src/persistence/Layers/ProjectionTurns.ts @@ -0,0 +1,336 @@ +import { OrchestrationCheckpointFile } from "@t3tools/contracts"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import { Effect, Layer, Option, Schema, Struct } from "effect"; + +import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts"; +import { + ClearCheckpointTurnConflictInput, + DeleteProjectionTurnsByThreadInput, + GetProjectionPendingTurnStartInput, + GetProjectionTurnByTurnIdInput, + ListProjectionTurnsByThreadInput, + ProjectionPendingTurnStart, + ProjectionTurn, + ProjectionTurnById, + ProjectionTurnRepository, + type ProjectionTurnRepositoryShape, +} from "../Services/ProjectionTurns.ts"; + +const ProjectionTurnDbRowSchema = ProjectionTurn.mapFields( + Struct.assign({ + checkpointFiles: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)), + }), +); + +const ProjectionTurnByIdDbRowSchema = ProjectionTurnById.mapFields( + Struct.assign({ + checkpointFiles: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)), + }), +); + +function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { + return (cause: unknown) => + Schema.isSchemaError(cause) + ? toPersistenceDecodeError(decodeOperation)(cause) + : toPersistenceSqlError(sqlOperation)(cause); +} + +const makeProjectionTurnRepository = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const upsertProjectionTurnById = SqlSchema.void({ + Request: ProjectionTurnByIdDbRowSchema, + execute: (row) => + sql` + INSERT INTO projection_turns ( + thread_id, + turn_id, + pending_message_id, + assistant_message_id, + state, + requested_at, + started_at, + completed_at, + checkpoint_turn_count, + checkpoint_ref, + checkpoint_status, + checkpoint_files_json + ) + VALUES ( + ${row.threadId}, + ${row.turnId}, + ${row.pendingMessageId}, + ${row.assistantMessageId}, + ${row.state}, + ${row.requestedAt}, + ${row.startedAt}, + ${row.completedAt}, + ${row.checkpointTurnCount}, + ${row.checkpointRef}, + ${row.checkpointStatus}, + ${row.checkpointFiles} + ) + ON CONFLICT (thread_id, turn_id) + DO UPDATE SET + pending_message_id = excluded.pending_message_id, + assistant_message_id = excluded.assistant_message_id, + state = excluded.state, + requested_at = excluded.requested_at, + started_at = excluded.started_at, + completed_at = excluded.completed_at, + checkpoint_turn_count = excluded.checkpoint_turn_count, + checkpoint_ref = excluded.checkpoint_ref, + checkpoint_status = excluded.checkpoint_status, + checkpoint_files_json = excluded.checkpoint_files_json + `, + }); + + const clearPendingProjectionTurnsByThread = SqlSchema.void({ + Request: DeleteProjectionTurnsByThreadInput, + execute: ({ threadId }) => + sql` + DELETE FROM projection_turns + WHERE thread_id = ${threadId} + AND turn_id IS NULL + AND state = 'pending' + AND checkpoint_turn_count IS NULL + `, + }); + + const insertPendingProjectionTurn = SqlSchema.void({ + Request: ProjectionPendingTurnStart, + execute: (row) => + sql` + INSERT INTO projection_turns ( + thread_id, + turn_id, + pending_message_id, + assistant_message_id, + state, + requested_at, + started_at, + completed_at, + checkpoint_turn_count, + checkpoint_ref, + checkpoint_status, + checkpoint_files_json + ) + VALUES ( + ${row.threadId}, + NULL, + ${row.messageId}, + NULL, + 'pending', + ${row.requestedAt}, + NULL, + NULL, + NULL, + NULL, + NULL, + '[]' + ) + `, + }); + + const getPendingProjectionTurn = SqlSchema.findOneOption({ + Request: GetProjectionPendingTurnStartInput, + Result: ProjectionPendingTurnStart, + execute: ({ threadId }) => + sql` + SELECT + thread_id AS "threadId", + pending_message_id AS "messageId", + requested_at AS "requestedAt" + FROM projection_turns + WHERE thread_id = ${threadId} + AND turn_id IS NULL + AND state = 'pending' + AND pending_message_id IS NOT NULL + AND checkpoint_turn_count IS NULL + ORDER BY requested_at DESC + LIMIT 1 + `, + }); + + const listProjectionTurnsByThread = SqlSchema.findAll({ + Request: ListProjectionTurnsByThreadInput, + Result: ProjectionTurnDbRowSchema, + execute: ({ threadId }) => + sql` + SELECT + thread_id AS "threadId", + turn_id AS "turnId", + pending_message_id AS "pendingMessageId", + assistant_message_id AS "assistantMessageId", + state, + requested_at AS "requestedAt", + started_at AS "startedAt", + completed_at AS "completedAt", + checkpoint_turn_count AS "checkpointTurnCount", + checkpoint_ref AS "checkpointRef", + checkpoint_status AS "checkpointStatus", + checkpoint_files_json AS "checkpointFiles" + FROM projection_turns + WHERE thread_id = ${threadId} + ORDER BY + CASE + WHEN checkpoint_turn_count IS NULL THEN 1 + ELSE 0 + END ASC, + checkpoint_turn_count ASC, + requested_at ASC, + turn_id ASC + `, + }); + + const getProjectionTurnByTurnId = SqlSchema.findOneOption({ + Request: GetProjectionTurnByTurnIdInput, + Result: ProjectionTurnByIdDbRowSchema, + execute: ({ threadId, turnId }) => + sql` + SELECT + thread_id AS "threadId", + turn_id AS "turnId", + pending_message_id AS "pendingMessageId", + assistant_message_id AS "assistantMessageId", + state, + requested_at AS "requestedAt", + started_at AS "startedAt", + completed_at AS "completedAt", + checkpoint_turn_count AS "checkpointTurnCount", + checkpoint_ref AS "checkpointRef", + checkpoint_status AS "checkpointStatus", + checkpoint_files_json AS "checkpointFiles" + FROM projection_turns + WHERE thread_id = ${threadId} + AND turn_id = ${turnId} + LIMIT 1 + `, + }); + + const clearCheckpointTurnConflictRow = SqlSchema.void({ + Request: ClearCheckpointTurnConflictInput, + execute: ({ threadId, turnId, checkpointTurnCount }) => + sql` + UPDATE projection_turns + SET + checkpoint_turn_count = NULL, + checkpoint_ref = NULL, + checkpoint_status = NULL, + checkpoint_files_json = '[]' + WHERE thread_id = ${threadId} + AND checkpoint_turn_count = ${checkpointTurnCount} + AND (turn_id IS NULL OR turn_id <> ${turnId}) + `, + }); + + const deleteProjectionTurnsByThread = SqlSchema.void({ + Request: DeleteProjectionTurnsByThreadInput, + execute: ({ threadId }) => + sql` + DELETE FROM projection_turns + WHERE thread_id = ${threadId} + `, + }); + + const upsertByTurnId: ProjectionTurnRepositoryShape["upsertByTurnId"] = (row) => + upsertProjectionTurnById(row).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionTurnRepository.upsertByTurnId:query", + "ProjectionTurnRepository.upsertByTurnId:encodeRequest", + ), + ), + ); + + const replacePendingTurnStart: ProjectionTurnRepositoryShape["replacePendingTurnStart"] = (row) => + sql + .withTransaction( + clearPendingProjectionTurnsByThread({ threadId: row.threadId }).pipe( + Effect.flatMap(() => insertPendingProjectionTurn(row)), + ), + ) + .pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionTurnRepository.replacePendingTurnStart:query", + "ProjectionTurnRepository.replacePendingTurnStart:encodeRequest", + ), + ), + ); + + const getPendingTurnStartByThreadId: ProjectionTurnRepositoryShape["getPendingTurnStartByThreadId"] = ( + input, + ) => + getPendingProjectionTurn(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionTurnRepository.getPendingTurnStartByThreadId:query"), + ), + ); + + const deletePendingTurnStartByThreadId: ProjectionTurnRepositoryShape["deletePendingTurnStartByThreadId"] = ( + input, + ) => + clearPendingProjectionTurnsByThread(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionTurnRepository.deletePendingTurnStartByThreadId:query"), + ), + ); + + const listByThreadId: ProjectionTurnRepositoryShape["listByThreadId"] = (input) => + listProjectionTurnsByThread(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionTurnRepository.listByThreadId:query", + "ProjectionTurnRepository.listByThreadId:decodeRows", + ), + ), + Effect.map((rows) => rows as ReadonlyArray>), + ); + + const getByTurnId: ProjectionTurnRepositoryShape["getByTurnId"] = (input) => + getProjectionTurnByTurnId(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionTurnRepository.getByTurnId:query", + "ProjectionTurnRepository.getByTurnId:decodeRow", + ), + ), + Effect.flatMap((rowOption) => + Option.match(rowOption, { + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => + Effect.succeed(Option.some(row as Schema.Schema.Type)), + }), + ), + ); + + const clearCheckpointTurnConflict: ProjectionTurnRepositoryShape["clearCheckpointTurnConflict"] = ( + input, + ) => + clearCheckpointTurnConflictRow(input).pipe( + Effect.mapError(toPersistenceSqlError("ProjectionTurnRepository.clearCheckpointTurnConflict:query")), + ); + + const deleteByThreadId: ProjectionTurnRepositoryShape["deleteByThreadId"] = (input) => + deleteProjectionTurnsByThread(input).pipe( + Effect.mapError(toPersistenceSqlError("ProjectionTurnRepository.deleteByThreadId:query")), + ); + + return { + upsertByTurnId, + replacePendingTurnStart, + getPendingTurnStartByThreadId, + deletePendingTurnStartByThreadId, + listByThreadId, + getByTurnId, + clearCheckpointTurnConflict, + deleteByThreadId, + } satisfies ProjectionTurnRepositoryShape; +}); + +export const ProjectionTurnRepositoryLive = Layer.effect( + ProjectionTurnRepository, + makeProjectionTurnRepository, +); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 7d73258519201d668c5ec54e099706ce305ea26a..3779c16d4f130bc3b3c62e8330b9b4b943e40a2d 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -17,7 +17,6 @@ import Migration0002 from "./Migrations/002_OrchestrationCommandReceipts.ts"; import Migration0003 from "./Migrations/003_CheckpointDiffBlobs.ts"; import Migration0004 from "./Migrations/004_ProviderSessionRuntime.ts"; import Migration0005 from "./Migrations/005_Projections.ts"; -import Migration0006 from "./Migrations/006_ProjectionPendingTurnStarts.ts"; /** * Migration loader with all migrations defined inline. @@ -35,7 +34,6 @@ const loader = Migrator.fromRecord({ "3_CheckpointDiffBlobs": Migration0003, "4_ProviderSessionRuntime": Migration0004, "5_Projections": Migration0005, - "6_ProjectionPendingTurnStarts": Migration0006, }); /** diff --git a/apps/server/src/persistence/Migrations/005_Projections.ts b/apps/server/src/persistence/Migrations/005_Projections.ts index 3231b3eafc5b024e5dfa276f3960d8e775522ae2..c950da76a18e2fd25609109764831ded8c84fbfe 100644 --- a/apps/server/src/persistence/Migrations/005_Projections.ts +++ b/apps/server/src/persistence/Migrations/005_Projections.ts @@ -72,29 +72,21 @@ export default Effect.gen(function* () { `; yield* sql` - CREATE TABLE IF NOT EXISTS projection_thread_turns ( - turn_id TEXT PRIMARY KEY, + CREATE TABLE IF NOT EXISTS projection_turns ( + row_id INTEGER PRIMARY KEY AUTOINCREMENT, thread_id TEXT NOT NULL, - turn_count INTEGER NOT NULL, - status TEXT NOT NULL, - user_message_id TEXT, + turn_id TEXT, + pending_message_id TEXT, assistant_message_id TEXT, - started_at TEXT NOT NULL, + state TEXT NOT NULL, + requested_at TEXT NOT NULL, + started_at TEXT, completed_at TEXT, - UNIQUE (thread_id, turn_count) - ) - `; - - yield* sql` - CREATE TABLE IF NOT EXISTS projection_checkpoints ( - thread_id TEXT NOT NULL, - turn_id TEXT NOT NULL, - checkpoint_turn_count INTEGER NOT NULL, - checkpoint_ref TEXT NOT NULL, - status TEXT NOT NULL, - files_json TEXT NOT NULL, - assistant_message_id TEXT, - completed_at TEXT NOT NULL, + checkpoint_turn_count INTEGER, + checkpoint_ref TEXT, + checkpoint_status TEXT, + checkpoint_files_json TEXT NOT NULL, + UNIQUE (thread_id, turn_id), UNIQUE (thread_id, checkpoint_turn_count) ) `; @@ -145,13 +137,13 @@ export default Effect.gen(function* () { `; yield* sql` - CREATE INDEX IF NOT EXISTS idx_projection_thread_turns_thread_turn_count - ON projection_thread_turns(thread_id, turn_count) + CREATE INDEX IF NOT EXISTS idx_projection_turns_thread_requested + ON projection_turns(thread_id, requested_at) `; yield* sql` - CREATE INDEX IF NOT EXISTS idx_projection_checkpoints_thread_completed - ON projection_checkpoints(thread_id, completed_at) + CREATE INDEX IF NOT EXISTS idx_projection_turns_thread_checkpoint_completed + ON projection_turns(thread_id, checkpoint_turn_count, completed_at) `; yield* sql` diff --git a/apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts b/apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts deleted file mode 100644 index 5e000a73b33727d14fceabf1312f9fbc2b8abd61..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts +++ /dev/null @@ -1,14 +0,0 @@ -import { Effect } from "effect"; -import * as SqlClient from "effect/unstable/sql/SqlClient"; - -export default Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - yield* sql` - CREATE TABLE IF NOT EXISTS projection_pending_turn_starts ( - thread_id TEXT PRIMARY KEY, - message_id TEXT NOT NULL, - created_at TEXT NOT NULL - ) - `; -}); diff --git a/apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts b/apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts deleted file mode 100644 index 6b6f9a065897aed76239ee8feee7e835701d4ccc..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts +++ /dev/null @@ -1,66 +0,0 @@ -/** - * ProjectionPendingTurnStartRepository - Repository interface for pending turn starts. - * - * Owns persistence operations for projected "pending turn start" markers used - * while creating new turns from queued messages. - * - * @module ProjectionPendingTurnStartRepository - */ -import { IsoDateTime, MessageId, ThreadId } from "@t3tools/contracts"; -import { Option, Schema, ServiceMap } from "effect"; -import type { Effect } from "effect"; - -import type { ProjectionRepositoryError } from "../Errors.ts"; - -export const ProjectionPendingTurnStart = Schema.Struct({ - threadId: ThreadId, - messageId: MessageId, - createdAt: IsoDateTime, -}); -export type ProjectionPendingTurnStart = typeof ProjectionPendingTurnStart.Type; - -export const GetProjectionPendingTurnStartInput = Schema.Struct({ - threadId: ThreadId, -}); -export type GetProjectionPendingTurnStartInput = typeof GetProjectionPendingTurnStartInput.Type; - -export const DeleteProjectionPendingTurnStartInput = Schema.Struct({ - threadId: ThreadId, -}); -export type DeleteProjectionPendingTurnStartInput = typeof DeleteProjectionPendingTurnStartInput.Type; - -/** - * ProjectionPendingTurnStartRepositoryShape - Service API for pending turn-start markers. - */ -export interface ProjectionPendingTurnStartRepositoryShape { - /** - * Insert or replace a pending turn-start row. - * - * Maintains at most one pending marker per thread. - */ - readonly upsert: ( - row: ProjectionPendingTurnStart, - ) => Effect.Effect; - - /** - * Read a pending turn-start marker by thread id. - */ - readonly getByThreadId: ( - input: GetProjectionPendingTurnStartInput, - ) => Effect.Effect, ProjectionRepositoryError>; - - /** - * Delete a pending turn-start marker by thread id. - */ - readonly deleteByThreadId: ( - input: DeleteProjectionPendingTurnStartInput, - ) => Effect.Effect; -} - -/** - * ProjectionPendingTurnStartRepository - Service tag for pending turn-start persistence. - */ -export class ProjectionPendingTurnStartRepository extends ServiceMap.Service< - ProjectionPendingTurnStartRepository, - ProjectionPendingTurnStartRepositoryShape ->()("persistence/ProjectionPendingTurnStartRepository") {} diff --git a/apps/server/src/persistence/Services/ProjectionThreadTurns.ts b/apps/server/src/persistence/Services/ProjectionThreadTurns.ts deleted file mode 100644 index d332717931707eb4cafaf5d543564c6630c70795..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Services/ProjectionThreadTurns.ts +++ /dev/null @@ -1,89 +0,0 @@ -/** - * ProjectionThreadTurnRepository - Projection repository interface for turns. - * - * Owns persistence operations for projected turn records in thread timelines. - * - * @module ProjectionThreadTurnRepository - */ -import { - IsoDateTime, - MessageId, - NonNegativeInt, - ProjectionThreadTurnStatus, - ThreadId, - TurnId, -} from "@t3tools/contracts"; -import { Option, Schema, ServiceMap } from "effect"; -import type { Effect } from "effect"; - -import type { ProjectionRepositoryError } from "../Errors.ts"; - -export const ProjectionThreadTurn = Schema.Struct({ - turnId: TurnId, - threadId: ThreadId, - turnCount: NonNegativeInt, - status: ProjectionThreadTurnStatus, - userMessageId: Schema.NullOr(MessageId), - assistantMessageId: Schema.NullOr(MessageId), - startedAt: IsoDateTime, - completedAt: Schema.NullOr(IsoDateTime), -}); -export type ProjectionThreadTurn = typeof ProjectionThreadTurn.Type; - -export const ListProjectionThreadTurnsInput = Schema.Struct({ - threadId: ThreadId, -}); -export type ListProjectionThreadTurnsInput = typeof ListProjectionThreadTurnsInput.Type; - -export const GetProjectionThreadTurnInput = Schema.Struct({ - turnId: TurnId, -}); -export type GetProjectionThreadTurnInput = typeof GetProjectionThreadTurnInput.Type; - -export const DeleteProjectionThreadTurnsInput = Schema.Struct({ - threadId: ThreadId, -}); -export type DeleteProjectionThreadTurnsInput = typeof DeleteProjectionThreadTurnsInput.Type; - -/** - * ProjectionThreadTurnRepositoryShape - Service API for projected thread turns. - */ -export interface ProjectionThreadTurnRepositoryShape { - /** - * Insert or replace a projected turn row. - * - * Upserts by `turnId`. - */ - readonly upsert: (turn: ProjectionThreadTurn) => Effect.Effect; - - /** - * List projected turns for a thread. - * - * Returned in ascending `turnCount` order. - */ - readonly listByThreadId: ( - input: ListProjectionThreadTurnsInput, - ) => Effect.Effect, ProjectionRepositoryError>; - - /** - * Read a projected turn row by turn id. - */ - readonly getByTurnId: ( - input: GetProjectionThreadTurnInput, - ) => Effect.Effect, ProjectionRepositoryError>; - - /** - * Delete projected turn rows by thread. - */ - readonly deleteByThreadId: ( - input: DeleteProjectionThreadTurnsInput, - ) => Effect.Effect; -} - -/** - * ProjectionThreadTurnRepository - Service tag for turn projection persistence. - */ -export class ProjectionThreadTurnRepository extends ServiceMap.Service< - ProjectionThreadTurnRepository, - ProjectionThreadTurnRepositoryShape ->()("persistence/ProjectionThreadTurnRepository") {} diff --git a/apps/server/src/persistence/Services/ProjectionTurns.ts b/apps/server/src/persistence/Services/ProjectionTurns.ts new file mode 100644 index 0000000000000000000000000000000000000000..8933c2c7cd7fc6c91c65aa875f920120ff76cb83 --- /dev/null +++ b/apps/server/src/persistence/Services/ProjectionTurns.ts @@ -0,0 +1,161 @@ +/** + * ProjectionTurnRepository - Projection repository interface for unified turn state. + * + * Owns persistence operations for pending starts, running/completed turn lifecycle, + * and checkpoint metadata in a single projection table. + * + * @module ProjectionTurnRepository + */ +import { + CheckpointRef, + IsoDateTime, + MessageId, + NonNegativeInt, + OrchestrationCheckpointFile, + OrchestrationCheckpointStatus, + ThreadId, + TurnId, +} from "@t3tools/contracts"; +import { Option, Schema, ServiceMap } from "effect"; +import type { Effect } from "effect"; + +import type { ProjectionRepositoryError } from "../Errors.ts"; + +export const ProjectionTurnState = Schema.Literals([ + "pending", + "running", + "interrupted", + "completed", + "error", +]); +export type ProjectionTurnState = typeof ProjectionTurnState.Type; + +export const ProjectionTurn = Schema.Struct({ + threadId: ThreadId, + turnId: Schema.NullOr(TurnId), + pendingMessageId: Schema.NullOr(MessageId), + assistantMessageId: Schema.NullOr(MessageId), + state: ProjectionTurnState, + requestedAt: IsoDateTime, + startedAt: Schema.NullOr(IsoDateTime), + completedAt: Schema.NullOr(IsoDateTime), + checkpointTurnCount: Schema.NullOr(NonNegativeInt), + checkpointRef: Schema.NullOr(CheckpointRef), + checkpointStatus: Schema.NullOr(OrchestrationCheckpointStatus), + checkpointFiles: Schema.Array(OrchestrationCheckpointFile), +}); +export type ProjectionTurn = typeof ProjectionTurn.Type; + +export const ProjectionTurnById = Schema.Struct({ + threadId: ThreadId, + turnId: TurnId, + pendingMessageId: Schema.NullOr(MessageId), + assistantMessageId: Schema.NullOr(MessageId), + state: ProjectionTurnState, + requestedAt: IsoDateTime, + startedAt: Schema.NullOr(IsoDateTime), + completedAt: Schema.NullOr(IsoDateTime), + checkpointTurnCount: Schema.NullOr(NonNegativeInt), + checkpointRef: Schema.NullOr(CheckpointRef), + checkpointStatus: Schema.NullOr(OrchestrationCheckpointStatus), + checkpointFiles: Schema.Array(OrchestrationCheckpointFile), +}); +export type ProjectionTurnById = typeof ProjectionTurnById.Type; + +export const ProjectionPendingTurnStart = Schema.Struct({ + threadId: ThreadId, + messageId: MessageId, + requestedAt: IsoDateTime, +}); +export type ProjectionPendingTurnStart = typeof ProjectionPendingTurnStart.Type; + +export const ListProjectionTurnsByThreadInput = Schema.Struct({ + threadId: ThreadId, +}); +export type ListProjectionTurnsByThreadInput = typeof ListProjectionTurnsByThreadInput.Type; + +export const GetProjectionTurnByTurnIdInput = Schema.Struct({ + threadId: ThreadId, + turnId: TurnId, +}); +export type GetProjectionTurnByTurnIdInput = typeof GetProjectionTurnByTurnIdInput.Type; + +export const GetProjectionPendingTurnStartInput = Schema.Struct({ + threadId: ThreadId, +}); +export type GetProjectionPendingTurnStartInput = typeof GetProjectionPendingTurnStartInput.Type; + +export const DeleteProjectionTurnsByThreadInput = Schema.Struct({ + threadId: ThreadId, +}); +export type DeleteProjectionTurnsByThreadInput = typeof DeleteProjectionTurnsByThreadInput.Type; + +export const ClearCheckpointTurnConflictInput = Schema.Struct({ + threadId: ThreadId, + turnId: TurnId, + checkpointTurnCount: NonNegativeInt, +}); +export type ClearCheckpointTurnConflictInput = typeof ClearCheckpointTurnConflictInput.Type; + +export interface ProjectionTurnRepositoryShape { + /** + * Inserts or updates the canonical row for a concrete `{threadId, turnId}` turn lifecycle state. + */ + readonly upsertByTurnId: ( + row: ProjectionTurnById, + ) => Effect.Effect; + + /** + * Replaces any existing pending-start placeholder rows for a thread with exactly one latest pending-start row. + */ + readonly replacePendingTurnStart: ( + row: ProjectionPendingTurnStart, + ) => Effect.Effect; + + /** + * Returns the newest pending-start placeholder for a thread; this is expected to be at most one row after replacement writes. + */ + readonly getPendingTurnStartByThreadId: ( + input: GetProjectionPendingTurnStartInput, + ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * Deletes only pending-start placeholder rows (`turnId = null`) for a thread and leaves concrete turn rows untouched. + */ + readonly deletePendingTurnStartByThreadId: ( + input: GetProjectionPendingTurnStartInput, + ) => Effect.Effect; + + /** + * Lists all projection rows for a thread, including pending placeholders, with checkpoint rows ordered before non-checkpoint rows. + */ + readonly listByThreadId: ( + input: ListProjectionTurnsByThreadInput, + ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * Looks up a concrete turn row by `{threadId, turnId}` and never returns pending placeholder rows. + */ + readonly getByTurnId: ( + input: GetProjectionTurnByTurnIdInput, + ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * Clears checkpoint fields on conflicting rows that reuse the same checkpoint turn count in a thread, excluding the provided turn. + */ + readonly clearCheckpointTurnConflict: ( + input: ClearCheckpointTurnConflictInput, + ) => Effect.Effect; + + /** + * Hard-deletes all projection rows for a thread, including pending-start placeholders and checkpoint metadata rows. + */ + readonly deleteByThreadId: ( + input: DeleteProjectionTurnsByThreadInput, + ) => Effect.Effect; +} + +export class ProjectionTurnRepository extends ServiceMap.Service< + ProjectionTurnRepository, + ProjectionTurnRepositoryShape +>()("persistence/ProjectionTurnRepository") {} diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index ab67bdbd7c0f3a899623760fceeb6ba276ce76a0..85cf527edb2867a7e83ac7ba3434dc3f3316b208 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -194,7 +194,7 @@ class MockTerminalManager implements TerminalManagerShape { }; }); - readonly dispose: TerminalManagerShape["dispose"] = () => Effect.void; + readonly dispose: TerminalManagerShape["dispose"] = Effect.void; } function connectWs(port: number, token?: string): Promise { -- 2.54.0