From 76208f79de4cb79146f1501feed6162eef71e24b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 21:28:51 -0800 Subject: [PATCH] C015 & C016 --- .plans/16c-pr89-remediation-checklist.md | 26 +- .../Layers/ProjectionPipeline.test.ts | 328 +++++++++++++++++- .../Layers/ProjectionPipeline.ts | 47 ++- .../src/orchestration/projector.test.ts | 149 ++++++++ apps/server/src/orchestration/projector.ts | 14 +- .../Layers/ProjectionPendingTurnStarts.ts | 92 +++++ apps/server/src/persistence/Migrations.ts | 2 + .../006_ProjectionPendingTurnStarts.ts | 14 + .../Services/ProjectionPendingTurnStarts.ts | 41 +++ 9 files changed, 678 insertions(+), 35 deletions(-) create mode 100644 apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts create mode 100644 apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts create mode 100644 apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts diff --git a/.plans/16c-pr89-remediation-checklist.md b/.plans/16c-pr89-remediation-checklist.md index ba7686dfd603dd2fb4a986c67ac7c3cf8c69dd5c..8d30b2dd408d98e75c8e44c1b6bb9089ee93f7ff 100644 --- a/.plans/16c-pr89-remediation-checklist.md +++ b/.plans/16c-pr89-remediation-checklist.md @@ -11,7 +11,7 @@ Status values: - `DONE`: Implemented and verified - `CLOSED_INVALID`: Stale/invalid review finding -Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5` +Counts: active `51` (`valid=33`, `partially-valid=18`), closed-invalid `6` ## Active Checklist @@ -172,17 +172,8 @@ Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5` - Threads: PRRT_kwDORLtfbc5whtrM - Audit note: Persisted event can outpace in-memory projection on mid-flight failure. -- [ ] `C014` Engine error handler catches all errors including non-invariant ones - - Status: `TODO` - - Verdict: `partially-valid` - - Severity: `Medium` - - Area: `Event ordering and state consistency` - - File: `apps/server/src/orchestration/Layers/OrchestrationEngine.ts:144` - - Threads: PRRT_kwDORLtfbc5wkPaJ - - Audit note: Broad catch may be intentional for liveness; correctness concern is contextual. - -- [ ] `C015` The gap-filling fallback logic can retain messages from turns that are about to be deleted, causing foreign key violations. Consider removing the fallback logic entirely, or filtering `fallbackUserMessages` and `fallbackAssistantMessages` to only include messages whose `turnId` is in `retainedTurnIds`.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: - - Status: `TODO` +- [x] `C015` The gap-filling fallback logic can retain messages from turns that are about to be deleted, causing foreign key violations. Consider removing the fallback logic entirely, or filtering `fallbackUserMessages` and `fallbackAssistantMessages` to only include messages whose `turnId` is in `retainedTurnIds`.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: + - Status: `DONE` - Verdict: `partially-valid` - Severity: `Medium` - Area: `Event ordering and state consistency` @@ -190,8 +181,8 @@ Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5` - Threads: PRRT_kwDORLtfbc5whxJO - Audit note: Message fallback retention issue is real, but prior FK-violation claim is overstated. -- [ ] `C016` The in-memory `pendingTurnStartByThreadId` map isn't restored during bootstrap. If the service restarts after processing `thread.turn-start-requested` but before `thread.session-set`, the `userMessageId` and `startedAt` will be lost since bootstrap resumes *after* the committed sequence. Consider persisting this pending state or processing these two events atomically.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: - - Status: `TODO` +- [x] `C016` The in-memory `pendingTurnStartByThreadId` map isn't restored during bootstrap. If the service restarts after processing `thread.turn-start-requested` but before `thread.session-set`, the `userMessageId` and `startedAt` will be lost since bootstrap resumes *after* the committed sequence. Consider persisting this pending state or processing these two events atomically.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: + - Status: `DONE` - Verdict: `valid` - Severity: `Medium` - Area: `Event ordering and state consistency` @@ -506,6 +497,13 @@ Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5` ## Closed Invalid Items +- [x] `C014` Engine error handler catches all errors including non-invariant ones + - Status: `CLOSED_INVALID` + - Severity: `Medium` + - File: `apps/server/src/orchestration/Layers/OrchestrationEngine.ts:144` + - Threads: PRRT_kwDORLtfbc5wkPaJ + - Rationale: Broad catch is intentional for worker liveness; transactional dispatch path prevents the claimed non-invariant idempotency break in current design. + - [x] `C021` Shared mutable default metadata object causes stale eventId - Status: `CLOSED_INVALID` - Severity: `Medium` diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index ce775b32fc2d6b003a979a76e983fdb2751f2e86..a082100602670738dbcb22feb8eebb56a1c3c5c1 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1,18 +1,29 @@ import { + CheckpointRef, CommandId, CorrelationId, EventId, MessageId, + ProviderSessionId, + ProviderThreadId, ProjectId, ThreadId, + TurnId, } from "@t3tools/contracts"; +import { NodeServices } from "@effect/platform-node"; import { assert, it } from "@effect/vitest"; import { Effect, Layer } from "effect"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; -import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { + makeSqlitePersistenceLive, + SqlitePersistenceMemory, +} from "../../persistence/Layers/Sqlite.ts"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { @@ -400,8 +411,323 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { assert.isFalse(Boolean(messageRows[0]?.isStreaming)); }), ); + + it.effect("does not fallback-retain messages whose turnId is removed by revert", () => + 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-revert-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-revert"), + occurredAt: "2026-02-26T12:00:00.000Z", + commandId: CommandId.makeUnsafe("cmd-revert-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-revert"), + title: "Project Revert", + workspaceRoot: "/tmp/project-revert", + defaultModel: null, + scripts: [], + createdAt: "2026-02-26T12:00:00.000Z", + updatedAt: "2026-02-26T12:00:00.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-revert-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert"), + occurredAt: "2026-02-26T12:00:01.000Z", + commandId: CommandId.makeUnsafe("cmd-revert-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert"), + projectId: ProjectId.makeUnsafe("project-revert"), + title: "Thread Revert", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: "2026-02-26T12:00:01.000Z", + updatedAt: "2026-02-26T12:00:01.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.makeUnsafe("evt-revert-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert"), + occurredAt: "2026-02-26T12:00:02.000Z", + commandId: CommandId.makeUnsafe("cmd-revert-3"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert"), + turnId: TurnId.makeUnsafe("turn-1"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert/turn/1"), + status: "ready", + files: [], + assistantMessageId: MessageId.makeUnsafe("assistant-keep"), + completedAt: "2026-02-26T12:00:02.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-revert-4"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert"), + occurredAt: "2026-02-26T12:00:02.100Z", + commandId: CommandId.makeUnsafe("cmd-revert-4"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-4"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert"), + messageId: MessageId.makeUnsafe("assistant-keep"), + role: "assistant", + text: "kept", + turnId: TurnId.makeUnsafe("turn-1"), + streaming: false, + createdAt: "2026-02-26T12:00:02.100Z", + updatedAt: "2026-02-26T12:00:02.100Z", + }, + }); + + yield* appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.makeUnsafe("evt-revert-5"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert"), + occurredAt: "2026-02-26T12:00:03.000Z", + commandId: CommandId.makeUnsafe("cmd-revert-5"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-5"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert"), + turnId: TurnId.makeUnsafe("turn-2"), + checkpointTurnCount: 2, + checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert/turn/2"), + status: "ready", + files: [], + assistantMessageId: MessageId.makeUnsafe("assistant-remove"), + completedAt: "2026-02-26T12:00:03.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-revert-6"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert"), + occurredAt: "2026-02-26T12:00:03.050Z", + commandId: CommandId.makeUnsafe("cmd-revert-6"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-6"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert"), + messageId: MessageId.makeUnsafe("user-remove"), + role: "user", + text: "removed", + turnId: TurnId.makeUnsafe("turn-2"), + streaming: false, + createdAt: "2026-02-26T12:00:03.050Z", + updatedAt: "2026-02-26T12:00:03.050Z", + }, + }); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-revert-7"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert"), + occurredAt: "2026-02-26T12:00:03.100Z", + commandId: CommandId.makeUnsafe("cmd-revert-7"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-7"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert"), + messageId: MessageId.makeUnsafe("assistant-remove"), + role: "assistant", + text: "removed", + turnId: TurnId.makeUnsafe("turn-2"), + streaming: false, + createdAt: "2026-02-26T12:00:03.100Z", + updatedAt: "2026-02-26T12:00:03.100Z", + }, + }); + + yield* appendAndProject({ + type: "thread.reverted", + eventId: EventId.makeUnsafe("evt-revert-8"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-revert"), + occurredAt: "2026-02-26T12:00:04.000Z", + commandId: CommandId.makeUnsafe("cmd-revert-8"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-revert-8"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-revert"), + turnCount: 1, + }, + }); + + const messageRows = yield* sql<{ + readonly messageId: string; + readonly turnId: string | null; + readonly role: string; + }>` + SELECT + message_id AS "messageId", + turn_id AS "turnId", + role + FROM projection_thread_messages + WHERE thread_id = 'thread-revert' + ORDER BY created_at ASC, message_id ASC + `; + assert.deepEqual(messageRows, [ + { + messageId: "assistant-keep", + turnId: "turn-1", + role: "assistant", + }, + ]); + }), + ); + }); +it.effect("restores pending turn-start metadata across projection pipeline restart", () => + Effect.gen(function* () { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-pipeline-restart-")); + const dbPath = path.join(tempDir, "orchestration.sqlite"); + const persistenceLayer = makeSqlitePersistenceLive(dbPath); + const firstProjectionLayer = OrchestrationProjectionPipelineLive.pipe( + Layer.provideMerge(OrchestrationEventStoreLive), + Layer.provideMerge(persistenceLayer), + ); + const secondProjectionLayer = OrchestrationProjectionPipelineLive.pipe( + Layer.provideMerge(OrchestrationEventStoreLive), + Layer.provideMerge(persistenceLayer), + ); + + const threadId = ThreadId.makeUnsafe("thread-restart"); + const turnId = TurnId.makeUnsafe("turn-restart"); + const messageId = MessageId.makeUnsafe("message-restart"); + const turnStartedAt = "2026-02-26T14:00:00.000Z"; + const sessionSetAt = "2026-02-26T14:00:05.000Z"; + + yield* Effect.gen(function* () { + const eventStore = yield* OrchestrationEventStore; + const projectionPipeline = yield* OrchestrationProjectionPipeline; + + yield* eventStore.append({ + type: "thread.turn-start-requested", + eventId: EventId.makeUnsafe("evt-restart-1"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: turnStartedAt, + commandId: CommandId.makeUnsafe("cmd-restart-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-restart-1"), + metadata: {}, + payload: { + threadId, + messageId, + createdAt: turnStartedAt, + }, + }); + + yield* projectionPipeline.bootstrap; + }).pipe(Effect.provide(firstProjectionLayer)); + + const turnRows = yield* Effect.gen(function* () { + const eventStore = yield* OrchestrationEventStore; + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const sql = yield* SqlClient.SqlClient; + + yield* eventStore.append({ + type: "thread.session-set", + eventId: EventId.makeUnsafe("evt-restart-2"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: sessionSetAt, + commandId: CommandId.makeUnsafe("cmd-restart-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-restart-2"), + metadata: { + providerSessionId: ProviderSessionId.makeUnsafe("provider-session-restart"), + providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-restart"), + }, + payload: { + threadId, + session: { + threadId, + status: "running", + providerName: "codex", + providerSessionId: ProviderSessionId.makeUnsafe("provider-session-restart"), + providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-restart"), + activeTurnId: turnId, + lastError: null, + updatedAt: sessionSetAt, + }, + }, + }); + + yield* projectionPipeline.bootstrap; + + const pendingRows = yield* sql<{ readonly threadId: string }>` + SELECT thread_id AS "threadId" + FROM projection_pending_turn_starts + WHERE thread_id = ${threadId} + `; + assert.deepEqual(pendingRows, []); + + return yield* sql<{ + readonly turnId: string; + readonly userMessageId: string | null; + readonly startedAt: string; + }>` + SELECT + turn_id AS "turnId", + user_message_id AS "userMessageId", + started_at AS "startedAt" + FROM projection_thread_turns + WHERE turn_id = ${turnId} + `; + }).pipe(Effect.provide(secondProjectionLayer)); + + assert.deepEqual(turnRows, [ + { + turnId: "turn-restart", + userMessageId: "message-restart", + startedAt: turnStartedAt, + }, + ]); + + fs.rmSync(tempDir, { recursive: true, force: true }); + }).pipe(Effect.provide(NodeServices.layer)), +); + const engineLayer = it.layer( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionPipelineLive), diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index b890adfbabd7fad51ae2236976fa0e260af28558..81d0eb95d93bea950391d731a77b9eff1c20c820 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -6,6 +6,7 @@ import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../per 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"; @@ -20,6 +21,7 @@ import { ProjectionThreadTurnRepository } from "../../persistence/Services/Proje 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"; @@ -97,7 +99,12 @@ function retainProjectionMessagesAfterRevert( const missingUserCount = Math.max(0, turnCount - retainedUserCount); if (missingUserCount > 0) { const fallbackUserMessages = messages - .filter((message) => message.role === "user" && !retainedMessageIds.has(message.messageId)) + .filter( + (message) => + message.role === "user" && + !retainedMessageIds.has(message.messageId) && + (message.turnId === null || retainedTurnIds.has(message.turnId)), + ) .toSorted( (left, right) => left.createdAt.localeCompare(right.createdAt) || @@ -116,7 +123,10 @@ function retainProjectionMessagesAfterRevert( if (missingAssistantCount > 0) { const fallbackAssistantMessages = messages .filter( - (message) => message.role === "assistant" && !retainedMessageIds.has(message.messageId), + (message) => + message.role === "assistant" && + !retainedMessageIds.has(message.messageId) && + (message.turnId === null || retainedTurnIds.has(message.turnId)), ) .toSorted( (left, right) => @@ -157,13 +167,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const projectionThreadTurnRepository = yield* ProjectionThreadTurnRepository; const projectionCheckpointRepository = yield* ProjectionCheckpointRepository; const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository; - const pendingTurnStartByThreadId = new Map< - string, - { - readonly messageId: NonNullable; - readonly createdAt: string; - } - >(); + const projectionPendingTurnStartRepository = yield* ProjectionPendingTurnStartRepository; const applyProjectsProjection: ProjectorDefinition["apply"] = (event) => Effect.gen(function* () { @@ -473,7 +477,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { Effect.gen(function* () { switch (event.type) { case "thread.turn-start-requested": { - pendingTurnStartByThreadId.set(event.payload.threadId, { + yield* projectionPendingTurnStartRepository.upsert({ + threadId: event.payload.threadId, messageId: event.payload.messageId, createdAt: event.payload.createdAt, }); @@ -487,13 +492,16 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({ turnId }); - const pendingTurnStart = pendingTurnStartByThreadId.get(event.payload.threadId); + const pendingTurnStart = yield* projectionPendingTurnStartRepository.getByThreadId({ + threadId: event.payload.threadId, + }); if (Option.isSome(existingTurn)) { yield* projectionThreadTurnRepository.upsert({ ...existingTurn.value, status: existingTurn.value.status === "completed" ? "completed" : "running", userMessageId: - existingTurn.value.userMessageId ?? pendingTurnStart?.messageId ?? null, + existingTurn.value.userMessageId ?? + (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null), startedAt: existingTurn.value.startedAt, }); } else { @@ -505,14 +513,18 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { threadId: event.payload.threadId, turnCount: nextTurnCount(existingThreadTurns), status: "running", - userMessageId: pendingTurnStart?.messageId ?? null, + userMessageId: Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null, assistantMessageId: null, - startedAt: pendingTurnStart?.createdAt ?? event.occurredAt, + startedAt: Option.isSome(pendingTurnStart) + ? pendingTurnStart.value.createdAt + : event.occurredAt, completedAt: null, }); } - pendingTurnStartByThreadId.delete(event.payload.threadId); + yield* projectionPendingTurnStartRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); return; } @@ -837,9 +849,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const bootstrap: OrchestrationProjectionPipelineShape["bootstrap"] = Effect.forEach( projectors, bootstrapProjector, - { - concurrency: 1, - }, + { concurrency: 1 }, ).pipe( Effect.asVoid, Effect.tap(() => @@ -870,5 +880,6 @@ export const OrchestrationProjectionPipelineLive = Layer.effect( Layer.provideMerge(ProjectionThreadTurnRepositoryLive), Layer.provideMerge(ProjectionCheckpointRepositoryLive), Layer.provideMerge(ProjectionPendingApprovalRepositoryLive), + Layer.provideMerge(ProjectionPendingTurnStartRepositoryLive), Layer.provideMerge(ProjectionStateRepositoryLive), ); diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index e03c260db817aa84b05031ed145572a41c3116db..770744bc9f91852c5291727cbfa32731e5f75722 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -502,4 +502,153 @@ describe("orchestration projector", () => { expect(thread?.checkpoints.map((checkpoint) => checkpoint.checkpointTurnCount)).toEqual([1]); expect(thread?.latestTurnId).toBe("turn-1"); }); + + it("does not fallback-retain messages tied to removed turn IDs", async () => { + const createdAt = "2026-02-26T12:00:00.000Z"; + const model = createEmptyReadModel(createdAt); + + const afterCreate = await Effect.runPromise( + projectEvent( + model, + makeEvent({ + sequence: 1, + type: "thread.created", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: createdAt, + commandId: "cmd-create-revert", + payload: { + threadId: "thread-revert", + projectId: "project-1", + title: "demo", + model: "gpt-5.3-codex", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }), + ), + ); + + const events: ReadonlyArray = [ + makeEvent({ + sequence: 2, + type: "thread.turn-diff-completed", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:01.000Z", + commandId: "cmd-turn-1", + payload: { + threadId: "thread-revert", + turnId: "turn-1", + checkpointTurnCount: 1, + checkpointRef: "refs/t3/checkpoints/thread-revert/turn/1", + status: "ready", + files: [], + assistantMessageId: "assistant-keep", + completedAt: "2026-02-26T12:00:01.000Z", + }, + }), + makeEvent({ + sequence: 3, + type: "thread.message-sent", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:01.100Z", + commandId: "cmd-assistant-keep", + payload: { + threadId: "thread-revert", + messageId: "assistant-keep", + role: "assistant", + text: "kept", + turnId: "turn-1", + streaming: false, + createdAt: "2026-02-26T12:00:01.100Z", + updatedAt: "2026-02-26T12:00:01.100Z", + }, + }), + makeEvent({ + sequence: 4, + type: "thread.turn-diff-completed", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:02.000Z", + commandId: "cmd-turn-2", + payload: { + threadId: "thread-revert", + turnId: "turn-2", + checkpointTurnCount: 2, + checkpointRef: "refs/t3/checkpoints/thread-revert/turn/2", + status: "ready", + files: [], + assistantMessageId: "assistant-remove", + completedAt: "2026-02-26T12:00:02.000Z", + }, + }), + makeEvent({ + sequence: 5, + type: "thread.message-sent", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:02.050Z", + commandId: "cmd-user-remove", + payload: { + threadId: "thread-revert", + messageId: "user-remove", + role: "user", + text: "removed", + turnId: "turn-2", + streaming: false, + createdAt: "2026-02-26T12:00:02.050Z", + updatedAt: "2026-02-26T12:00:02.050Z", + }, + }), + makeEvent({ + sequence: 6, + type: "thread.message-sent", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:02.100Z", + commandId: "cmd-assistant-remove", + payload: { + threadId: "thread-revert", + messageId: "assistant-remove", + role: "assistant", + text: "removed", + turnId: "turn-2", + streaming: false, + createdAt: "2026-02-26T12:00:02.100Z", + updatedAt: "2026-02-26T12:00:02.100Z", + }, + }), + makeEvent({ + sequence: 7, + type: "thread.reverted", + aggregateKind: "thread", + aggregateId: "thread-revert", + occurredAt: "2026-02-26T12:00:03.000Z", + commandId: "cmd-revert", + payload: { + threadId: "thread-revert", + turnCount: 1, + }, + }), + ]; + + const afterRevert = await events.reduce>>( + (statePromise, event) => + statePromise.then((state) => Effect.runPromise(projectEvent(state, event))), + Promise.resolve(afterCreate), + ); + + const thread = afterRevert.threads[0]; + expect( + thread?.messages.map((message) => ({ + id: message.id, + role: message.role, + turnId: message.turnId, + })), + ).toEqual([{ id: "assistant-keep", role: "assistant", turnId: "turn-1" }]); + }); }); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 6a3f8fe9c12baf3b88f49e9b15bf0ba776f13885..d17722d8c93dbea42b9187d01d4bcab5e98187c4 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -65,7 +65,12 @@ function retainThreadMessagesAfterRevert( const missingUserCount = Math.max(0, turnCount - retainedUserCount); if (missingUserCount > 0) { const fallbackUserMessages = messages - .filter((message) => message.role === "user" && !retainedMessageIds.has(message.id)) + .filter( + (message) => + message.role === "user" && + !retainedMessageIds.has(message.id) && + (message.turnId === null || retainedTurnIds.has(message.turnId)), + ) .toSorted( (left, right) => left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id), @@ -82,7 +87,12 @@ function retainThreadMessagesAfterRevert( const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount); if (missingAssistantCount > 0) { const fallbackAssistantMessages = messages - .filter((message) => message.role === "assistant" && !retainedMessageIds.has(message.id)) + .filter( + (message) => + message.role === "assistant" && + !retainedMessageIds.has(message.id) && + (message.turnId === null || retainedTurnIds.has(message.turnId)), + ) .toSorted( (left, right) => left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id), diff --git a/apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts b/apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts new file mode 100644 index 0000000000000000000000000000000000000000..7d46dfdf5daac2f0b0d07c50e99fbb0c0b8182ba --- /dev/null +++ b/apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts @@ -0,0 +1,92 @@ +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/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 3779c16d4f130bc3b3c62e8330b9b4b943e40a2d..7d73258519201d668c5ec54e099706ce305ea26a 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -17,6 +17,7 @@ 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. @@ -34,6 +35,7 @@ const loader = Migrator.fromRecord({ "3_CheckpointDiffBlobs": Migration0003, "4_ProviderSessionRuntime": Migration0004, "5_Projections": Migration0005, + "6_ProjectionPendingTurnStarts": Migration0006, }); /** diff --git a/apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts b/apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts new file mode 100644 index 0000000000000000000000000000000000000000..5e000a73b33727d14fceabf1312f9fbc2b8abd61 --- /dev/null +++ b/apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts @@ -0,0 +1,14 @@ +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 new file mode 100644 index 0000000000000000000000000000000000000000..58c2572fe2cb0c37af10590938fe53b1ab06f1b7 --- /dev/null +++ b/apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts @@ -0,0 +1,41 @@ +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; + +export interface ProjectionPendingTurnStartRepositoryShape { + readonly upsert: ( + row: ProjectionPendingTurnStart, + ) => Effect.Effect; + + readonly getByThreadId: ( + input: GetProjectionPendingTurnStartInput, + ) => Effect.Effect, ProjectionRepositoryError>; + + readonly deleteByThreadId: ( + input: DeleteProjectionPendingTurnStartInput, + ) => Effect.Effect; +} + +export class ProjectionPendingTurnStartRepository extends ServiceMap.Service< + ProjectionPendingTurnStartRepository, + ProjectionPendingTurnStartRepositoryShape +>()("persistence/ProjectionPendingTurnStartRepository") {} -- 2.54.0