authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-25 21:28:51-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log76208f79de4cb79146f1501feed6162eef71e24b
tree4cfe207a23788cbdaf4fc0886a06e16b27da93b1
parent9e2324c2e567d75f37b305fbaf53b545178db5b0

C015 & C016

9 files changed, 678 insertions(+), 35 deletions(-)

.plans/16c-pr89-remediation-checklist.md+12-14
......@@ -11,7 +11,7 @@ Status values:
1111- `DONE`: Implemented and verified
1212- `CLOSED_INVALID`: Stale/invalid review finding
1313
14Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5`
14Counts: active `51` (`valid=33`, `partially-valid=18`), closed-invalid `6`
1515
1616## Active Checklist
1717
......@@ -172,17 +172,8 @@ Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5`
172172 - Threads: PRRT_kwDORLtfbc5whtrM
173173 - Audit note: Persisted event can outpace in-memory projection on mid-flight failure.
174174
175- [ ] `C014` Engine error handler catches all errors including non-invariant ones
176 - Status: `TODO`
177 - Verdict: `partially-valid`
178 - Severity: `Medium`
179 - Area: `Event ordering and state consistency`
180 - File: `apps/server/src/orchestration/Layers/OrchestrationEngine.ts:144`
181 - Threads: PRRT_kwDORLtfbc5wkPaJ
182 - Audit note: Broad catch may be intentional for liveness; correctness concern is contextual.
183
184- [ ] `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`. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>
185 - Status: `TODO`
175- [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`. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>
176 - Status: `DONE`
186177 - Verdict: `partially-valid`
187178 - Severity: `Medium`
188179 - Area: `Event ordering and state consistency`
......@@ -190,8 +181,8 @@ Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5`
190181 - Threads: PRRT_kwDORLtfbc5whxJO
191182 - Audit note: Message fallback retention issue is real, but prior FK-violation claim is overstated.
192183
193- [ ] `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. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>
194 - Status: `TODO`
184- [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. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>
185 - Status: `DONE`
195186 - Verdict: `valid`
196187 - Severity: `Medium`
197188 - Area: `Event ordering and state consistency`
......@@ -506,6 +497,13 @@ Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5`
506497
507498## Closed Invalid Items
508499
500- [x] `C014` Engine error handler catches all errors including non-invariant ones
501 - Status: `CLOSED_INVALID`
502 - Severity: `Medium`
503 - File: `apps/server/src/orchestration/Layers/OrchestrationEngine.ts:144`
504 - Threads: PRRT_kwDORLtfbc5wkPaJ
505 - Rationale: Broad catch is intentional for worker liveness; transactional dispatch path prevents the claimed non-invariant idempotency break in current design.
506
509507- [x] `C021` Shared mutable default metadata object causes stale eventId
510508 - Status: `CLOSED_INVALID`
511509 - Severity: `Medium`
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+327-1
......@@ -1,18 +1,29 @@
11import {
2 CheckpointRef,
23 CommandId,
34 CorrelationId,
45 EventId,
56 MessageId,
7 ProviderSessionId,
8 ProviderThreadId,
69 ProjectId,
710 ThreadId,
11 TurnId,
812} from "@t3tools/contracts";
13import { NodeServices } from "@effect/platform-node";
914import { assert, it } from "@effect/vitest";
1015import { Effect, Layer } from "effect";
16import fs from "node:fs";
17import os from "node:os";
18import path from "node:path";
1119import * as SqlClient from "effect/unstable/sql/SqlClient";
1220
1321import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
1422import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
15import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
23import {
24 makeSqlitePersistenceLive,
25 SqlitePersistenceMemory,
26} from "../../persistence/Layers/Sqlite.ts";
1627import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
1728import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
1829import {
......@@ -400,8 +411,323 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
400411 assert.isFalse(Boolean(messageRows[0]?.isStreaming));
401412 }),
402413 );
414
415 it.effect("does not fallback-retain messages whose turnId is removed by revert", () =>
416 Effect.gen(function* () {
417 const projectionPipeline = yield* OrchestrationProjectionPipeline;
418 const eventStore = yield* OrchestrationEventStore;
419 const sql = yield* SqlClient.SqlClient;
420 const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
421 eventStore
422 .append(event)
423 .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
424
425 yield* appendAndProject({
426 type: "project.created",
427 eventId: EventId.makeUnsafe("evt-revert-1"),
428 aggregateKind: "project",
429 aggregateId: ProjectId.makeUnsafe("project-revert"),
430 occurredAt: "2026-02-26T12:00:00.000Z",
431 commandId: CommandId.makeUnsafe("cmd-revert-1"),
432 causationEventId: null,
433 correlationId: CorrelationId.makeUnsafe("cmd-revert-1"),
434 metadata: {},
435 payload: {
436 projectId: ProjectId.makeUnsafe("project-revert"),
437 title: "Project Revert",
438 workspaceRoot: "/tmp/project-revert",
439 defaultModel: null,
440 scripts: [],
441 createdAt: "2026-02-26T12:00:00.000Z",
442 updatedAt: "2026-02-26T12:00:00.000Z",
443 },
444 });
445
446 yield* appendAndProject({
447 type: "thread.created",
448 eventId: EventId.makeUnsafe("evt-revert-2"),
449 aggregateKind: "thread",
450 aggregateId: ThreadId.makeUnsafe("thread-revert"),
451 occurredAt: "2026-02-26T12:00:01.000Z",
452 commandId: CommandId.makeUnsafe("cmd-revert-2"),
453 causationEventId: null,
454 correlationId: CorrelationId.makeUnsafe("cmd-revert-2"),
455 metadata: {},
456 payload: {
457 threadId: ThreadId.makeUnsafe("thread-revert"),
458 projectId: ProjectId.makeUnsafe("project-revert"),
459 title: "Thread Revert",
460 model: "gpt-5-codex",
461 branch: null,
462 worktreePath: null,
463 createdAt: "2026-02-26T12:00:01.000Z",
464 updatedAt: "2026-02-26T12:00:01.000Z",
465 },
466 });
467
468 yield* appendAndProject({
469 type: "thread.turn-diff-completed",
470 eventId: EventId.makeUnsafe("evt-revert-3"),
471 aggregateKind: "thread",
472 aggregateId: ThreadId.makeUnsafe("thread-revert"),
473 occurredAt: "2026-02-26T12:00:02.000Z",
474 commandId: CommandId.makeUnsafe("cmd-revert-3"),
475 causationEventId: null,
476 correlationId: CorrelationId.makeUnsafe("cmd-revert-3"),
477 metadata: {},
478 payload: {
479 threadId: ThreadId.makeUnsafe("thread-revert"),
480 turnId: TurnId.makeUnsafe("turn-1"),
481 checkpointTurnCount: 1,
482 checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert/turn/1"),
483 status: "ready",
484 files: [],
485 assistantMessageId: MessageId.makeUnsafe("assistant-keep"),
486 completedAt: "2026-02-26T12:00:02.000Z",
487 },
488 });
489
490 yield* appendAndProject({
491 type: "thread.message-sent",
492 eventId: EventId.makeUnsafe("evt-revert-4"),
493 aggregateKind: "thread",
494 aggregateId: ThreadId.makeUnsafe("thread-revert"),
495 occurredAt: "2026-02-26T12:00:02.100Z",
496 commandId: CommandId.makeUnsafe("cmd-revert-4"),
497 causationEventId: null,
498 correlationId: CorrelationId.makeUnsafe("cmd-revert-4"),
499 metadata: {},
500 payload: {
501 threadId: ThreadId.makeUnsafe("thread-revert"),
502 messageId: MessageId.makeUnsafe("assistant-keep"),
503 role: "assistant",
504 text: "kept",
505 turnId: TurnId.makeUnsafe("turn-1"),
506 streaming: false,
507 createdAt: "2026-02-26T12:00:02.100Z",
508 updatedAt: "2026-02-26T12:00:02.100Z",
509 },
510 });
511
512 yield* appendAndProject({
513 type: "thread.turn-diff-completed",
514 eventId: EventId.makeUnsafe("evt-revert-5"),
515 aggregateKind: "thread",
516 aggregateId: ThreadId.makeUnsafe("thread-revert"),
517 occurredAt: "2026-02-26T12:00:03.000Z",
518 commandId: CommandId.makeUnsafe("cmd-revert-5"),
519 causationEventId: null,
520 correlationId: CorrelationId.makeUnsafe("cmd-revert-5"),
521 metadata: {},
522 payload: {
523 threadId: ThreadId.makeUnsafe("thread-revert"),
524 turnId: TurnId.makeUnsafe("turn-2"),
525 checkpointTurnCount: 2,
526 checkpointRef: CheckpointRef.makeUnsafe("refs/t3/checkpoints/thread-revert/turn/2"),
527 status: "ready",
528 files: [],
529 assistantMessageId: MessageId.makeUnsafe("assistant-remove"),
530 completedAt: "2026-02-26T12:00:03.000Z",
531 },
532 });
533
534 yield* appendAndProject({
535 type: "thread.message-sent",
536 eventId: EventId.makeUnsafe("evt-revert-6"),
537 aggregateKind: "thread",
538 aggregateId: ThreadId.makeUnsafe("thread-revert"),
539 occurredAt: "2026-02-26T12:00:03.050Z",
540 commandId: CommandId.makeUnsafe("cmd-revert-6"),
541 causationEventId: null,
542 correlationId: CorrelationId.makeUnsafe("cmd-revert-6"),
543 metadata: {},
544 payload: {
545 threadId: ThreadId.makeUnsafe("thread-revert"),
546 messageId: MessageId.makeUnsafe("user-remove"),
547 role: "user",
548 text: "removed",
549 turnId: TurnId.makeUnsafe("turn-2"),
550 streaming: false,
551 createdAt: "2026-02-26T12:00:03.050Z",
552 updatedAt: "2026-02-26T12:00:03.050Z",
553 },
554 });
555
556 yield* appendAndProject({
557 type: "thread.message-sent",
558 eventId: EventId.makeUnsafe("evt-revert-7"),
559 aggregateKind: "thread",
560 aggregateId: ThreadId.makeUnsafe("thread-revert"),
561 occurredAt: "2026-02-26T12:00:03.100Z",
562 commandId: CommandId.makeUnsafe("cmd-revert-7"),
563 causationEventId: null,
564 correlationId: CorrelationId.makeUnsafe("cmd-revert-7"),
565 metadata: {},
566 payload: {
567 threadId: ThreadId.makeUnsafe("thread-revert"),
568 messageId: MessageId.makeUnsafe("assistant-remove"),
569 role: "assistant",
570 text: "removed",
571 turnId: TurnId.makeUnsafe("turn-2"),
572 streaming: false,
573 createdAt: "2026-02-26T12:00:03.100Z",
574 updatedAt: "2026-02-26T12:00:03.100Z",
575 },
576 });
577
578 yield* appendAndProject({
579 type: "thread.reverted",
580 eventId: EventId.makeUnsafe("evt-revert-8"),
581 aggregateKind: "thread",
582 aggregateId: ThreadId.makeUnsafe("thread-revert"),
583 occurredAt: "2026-02-26T12:00:04.000Z",
584 commandId: CommandId.makeUnsafe("cmd-revert-8"),
585 causationEventId: null,
586 correlationId: CorrelationId.makeUnsafe("cmd-revert-8"),
587 metadata: {},
588 payload: {
589 threadId: ThreadId.makeUnsafe("thread-revert"),
590 turnCount: 1,
591 },
592 });
593
594 const messageRows = yield* sql<{
595 readonly messageId: string;
596 readonly turnId: string | null;
597 readonly role: string;
598 }>`
599 SELECT
600 message_id AS "messageId",
601 turn_id AS "turnId",
602 role
603 FROM projection_thread_messages
604 WHERE thread_id = 'thread-revert'
605 ORDER BY created_at ASC, message_id ASC
606 `;
607 assert.deepEqual(messageRows, [
608 {
609 messageId: "assistant-keep",
610 turnId: "turn-1",
611 role: "assistant",
612 },
613 ]);
614 }),
615 );
616
403617});
404618
619it.effect("restores pending turn-start metadata across projection pipeline restart", () =>
620 Effect.gen(function* () {
621 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-pipeline-restart-"));
622 const dbPath = path.join(tempDir, "orchestration.sqlite");
623 const persistenceLayer = makeSqlitePersistenceLive(dbPath);
624 const firstProjectionLayer = OrchestrationProjectionPipelineLive.pipe(
625 Layer.provideMerge(OrchestrationEventStoreLive),
626 Layer.provideMerge(persistenceLayer),
627 );
628 const secondProjectionLayer = OrchestrationProjectionPipelineLive.pipe(
629 Layer.provideMerge(OrchestrationEventStoreLive),
630 Layer.provideMerge(persistenceLayer),
631 );
632
633 const threadId = ThreadId.makeUnsafe("thread-restart");
634 const turnId = TurnId.makeUnsafe("turn-restart");
635 const messageId = MessageId.makeUnsafe("message-restart");
636 const turnStartedAt = "2026-02-26T14:00:00.000Z";
637 const sessionSetAt = "2026-02-26T14:00:05.000Z";
638
639 yield* Effect.gen(function* () {
640 const eventStore = yield* OrchestrationEventStore;
641 const projectionPipeline = yield* OrchestrationProjectionPipeline;
642
643 yield* eventStore.append({
644 type: "thread.turn-start-requested",
645 eventId: EventId.makeUnsafe("evt-restart-1"),
646 aggregateKind: "thread",
647 aggregateId: threadId,
648 occurredAt: turnStartedAt,
649 commandId: CommandId.makeUnsafe("cmd-restart-1"),
650 causationEventId: null,
651 correlationId: CorrelationId.makeUnsafe("cmd-restart-1"),
652 metadata: {},
653 payload: {
654 threadId,
655 messageId,
656 createdAt: turnStartedAt,
657 },
658 });
659
660 yield* projectionPipeline.bootstrap;
661 }).pipe(Effect.provide(firstProjectionLayer));
662
663 const turnRows = yield* Effect.gen(function* () {
664 const eventStore = yield* OrchestrationEventStore;
665 const projectionPipeline = yield* OrchestrationProjectionPipeline;
666 const sql = yield* SqlClient.SqlClient;
667
668 yield* eventStore.append({
669 type: "thread.session-set",
670 eventId: EventId.makeUnsafe("evt-restart-2"),
671 aggregateKind: "thread",
672 aggregateId: threadId,
673 occurredAt: sessionSetAt,
674 commandId: CommandId.makeUnsafe("cmd-restart-2"),
675 causationEventId: null,
676 correlationId: CorrelationId.makeUnsafe("cmd-restart-2"),
677 metadata: {
678 providerSessionId: ProviderSessionId.makeUnsafe("provider-session-restart"),
679 providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-restart"),
680 },
681 payload: {
682 threadId,
683 session: {
684 threadId,
685 status: "running",
686 providerName: "codex",
687 providerSessionId: ProviderSessionId.makeUnsafe("provider-session-restart"),
688 providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-restart"),
689 activeTurnId: turnId,
690 lastError: null,
691 updatedAt: sessionSetAt,
692 },
693 },
694 });
695
696 yield* projectionPipeline.bootstrap;
697
698 const pendingRows = yield* sql<{ readonly threadId: string }>`
699 SELECT thread_id AS "threadId"
700 FROM projection_pending_turn_starts
701 WHERE thread_id = ${threadId}
702 `;
703 assert.deepEqual(pendingRows, []);
704
705 return yield* sql<{
706 readonly turnId: string;
707 readonly userMessageId: string | null;
708 readonly startedAt: string;
709 }>`
710 SELECT
711 turn_id AS "turnId",
712 user_message_id AS "userMessageId",
713 started_at AS "startedAt"
714 FROM projection_thread_turns
715 WHERE turn_id = ${turnId}
716 `;
717 }).pipe(Effect.provide(secondProjectionLayer));
718
719 assert.deepEqual(turnRows, [
720 {
721 turnId: "turn-restart",
722 userMessageId: "message-restart",
723 startedAt: turnStartedAt,
724 },
725 ]);
726
727 fs.rmSync(tempDir, { recursive: true, force: true });
728 }).pipe(Effect.provide(NodeServices.layer)),
729);
730
405731const engineLayer = it.layer(
406732 OrchestrationEngineLive.pipe(
407733 Layer.provide(OrchestrationProjectionPipelineLive),
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+29-18
......@@ -6,6 +6,7 @@ import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../per
66import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
77import { ProjectionCheckpointRepository } from "../../persistence/Services/ProjectionCheckpoints.ts";
88import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts";
9import { ProjectionPendingTurnStartRepository } from "../../persistence/Services/ProjectionPendingTurnStarts.ts";
910import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts";
1011import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";
1112import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";
......@@ -20,6 +21,7 @@ import { ProjectionThreadTurnRepository } from "../../persistence/Services/Proje
2021import { ProjectionThreadRepository } from "../../persistence/Services/ProjectionThreads.ts";
2122import { ProjectionCheckpointRepositoryLive } from "../../persistence/Layers/ProjectionCheckpoints.ts";
2223import { ProjectionPendingApprovalRepositoryLive } from "../../persistence/Layers/ProjectionPendingApprovals.ts";
24import { ProjectionPendingTurnStartRepositoryLive } from "../../persistence/Layers/ProjectionPendingTurnStarts.ts";
2325import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/ProjectionProjects.ts";
2426import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts";
2527import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";
......@@ -97,7 +99,12 @@ function retainProjectionMessagesAfterRevert(
9799 const missingUserCount = Math.max(0, turnCount - retainedUserCount);
98100 if (missingUserCount > 0) {
99101 const fallbackUserMessages = messages
100 .filter((message) => message.role === "user" && !retainedMessageIds.has(message.messageId))
102 .filter(
103 (message) =>
104 message.role === "user" &&
105 !retainedMessageIds.has(message.messageId) &&
106 (message.turnId === null || retainedTurnIds.has(message.turnId)),
107 )
101108 .toSorted(
102109 (left, right) =>
103110 left.createdAt.localeCompare(right.createdAt) ||
......@@ -116,7 +123,10 @@ function retainProjectionMessagesAfterRevert(
116123 if (missingAssistantCount > 0) {
117124 const fallbackAssistantMessages = messages
118125 .filter(
119 (message) => message.role === "assistant" && !retainedMessageIds.has(message.messageId),
126 (message) =>
127 message.role === "assistant" &&
128 !retainedMessageIds.has(message.messageId) &&
129 (message.turnId === null || retainedTurnIds.has(message.turnId)),
120130 )
121131 .toSorted(
122132 (left, right) =>
......@@ -157,13 +167,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
157167 const projectionThreadTurnRepository = yield* ProjectionThreadTurnRepository;
158168 const projectionCheckpointRepository = yield* ProjectionCheckpointRepository;
159169 const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository;
160 const pendingTurnStartByThreadId = new Map<
161 string,
162 {
163 readonly messageId: NonNullable<ProjectionThreadTurn["userMessageId"]>;
164 readonly createdAt: string;
165 }
166 >();
170 const projectionPendingTurnStartRepository = yield* ProjectionPendingTurnStartRepository;
167171
168172 const applyProjectsProjection: ProjectorDefinition["apply"] = (event) =>
169173 Effect.gen(function* () {
......@@ -473,7 +477,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
473477 Effect.gen(function* () {
474478 switch (event.type) {
475479 case "thread.turn-start-requested": {
476 pendingTurnStartByThreadId.set(event.payload.threadId, {
480 yield* projectionPendingTurnStartRepository.upsert({
481 threadId: event.payload.threadId,
477482 messageId: event.payload.messageId,
478483 createdAt: event.payload.createdAt,
479484 });
......@@ -487,13 +492,16 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
487492 }
488493
489494 const existingTurn = yield* projectionThreadTurnRepository.getByTurnId({ turnId });
490 const pendingTurnStart = pendingTurnStartByThreadId.get(event.payload.threadId);
495 const pendingTurnStart = yield* projectionPendingTurnStartRepository.getByThreadId({
496 threadId: event.payload.threadId,
497 });
491498 if (Option.isSome(existingTurn)) {
492499 yield* projectionThreadTurnRepository.upsert({
493500 ...existingTurn.value,
494501 status: existingTurn.value.status === "completed" ? "completed" : "running",
495502 userMessageId:
496 existingTurn.value.userMessageId ?? pendingTurnStart?.messageId ?? null,
503 existingTurn.value.userMessageId ??
504 (Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null),
497505 startedAt: existingTurn.value.startedAt,
498506 });
499507 } else {
......@@ -505,14 +513,18 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
505513 threadId: event.payload.threadId,
506514 turnCount: nextTurnCount(existingThreadTurns),
507515 status: "running",
508 userMessageId: pendingTurnStart?.messageId ?? null,
516 userMessageId: Option.isSome(pendingTurnStart) ? pendingTurnStart.value.messageId : null,
509517 assistantMessageId: null,
510 startedAt: pendingTurnStart?.createdAt ?? event.occurredAt,
518 startedAt: Option.isSome(pendingTurnStart)
519 ? pendingTurnStart.value.createdAt
520 : event.occurredAt,
511521 completedAt: null,
512522 });
513523 }
514524
515 pendingTurnStartByThreadId.delete(event.payload.threadId);
525 yield* projectionPendingTurnStartRepository.deleteByThreadId({
526 threadId: event.payload.threadId,
527 });
516528 return;
517529 }
518530
......@@ -837,9 +849,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
837849 const bootstrap: OrchestrationProjectionPipelineShape["bootstrap"] = Effect.forEach(
838850 projectors,
839851 bootstrapProjector,
840 {
841 concurrency: 1,
842 },
852 { concurrency: 1 },
843853 ).pipe(
844854 Effect.asVoid,
845855 Effect.tap(() =>
......@@ -870,5 +880,6 @@ export const OrchestrationProjectionPipelineLive = Layer.effect(
870880 Layer.provideMerge(ProjectionThreadTurnRepositoryLive),
871881 Layer.provideMerge(ProjectionCheckpointRepositoryLive),
872882 Layer.provideMerge(ProjectionPendingApprovalRepositoryLive),
883 Layer.provideMerge(ProjectionPendingTurnStartRepositoryLive),
873884 Layer.provideMerge(ProjectionStateRepositoryLive),
874885);
apps/server/src/orchestration/projector.test.ts+149
......@@ -502,4 +502,153 @@ describe("orchestration projector", () => {
502502 expect(thread?.checkpoints.map((checkpoint) => checkpoint.checkpointTurnCount)).toEqual([1]);
503503 expect(thread?.latestTurnId).toBe("turn-1");
504504 });
505
506 it("does not fallback-retain messages tied to removed turn IDs", async () => {
507 const createdAt = "2026-02-26T12:00:00.000Z";
508 const model = createEmptyReadModel(createdAt);
509
510 const afterCreate = await Effect.runPromise(
511 projectEvent(
512 model,
513 makeEvent({
514 sequence: 1,
515 type: "thread.created",
516 aggregateKind: "thread",
517 aggregateId: "thread-revert",
518 occurredAt: createdAt,
519 commandId: "cmd-create-revert",
520 payload: {
521 threadId: "thread-revert",
522 projectId: "project-1",
523 title: "demo",
524 model: "gpt-5.3-codex",
525 branch: null,
526 worktreePath: null,
527 createdAt,
528 updatedAt: createdAt,
529 },
530 }),
531 ),
532 );
533
534 const events: ReadonlyArray<OrchestrationEvent> = [
535 makeEvent({
536 sequence: 2,
537 type: "thread.turn-diff-completed",
538 aggregateKind: "thread",
539 aggregateId: "thread-revert",
540 occurredAt: "2026-02-26T12:00:01.000Z",
541 commandId: "cmd-turn-1",
542 payload: {
543 threadId: "thread-revert",
544 turnId: "turn-1",
545 checkpointTurnCount: 1,
546 checkpointRef: "refs/t3/checkpoints/thread-revert/turn/1",
547 status: "ready",
548 files: [],
549 assistantMessageId: "assistant-keep",
550 completedAt: "2026-02-26T12:00:01.000Z",
551 },
552 }),
553 makeEvent({
554 sequence: 3,
555 type: "thread.message-sent",
556 aggregateKind: "thread",
557 aggregateId: "thread-revert",
558 occurredAt: "2026-02-26T12:00:01.100Z",
559 commandId: "cmd-assistant-keep",
560 payload: {
561 threadId: "thread-revert",
562 messageId: "assistant-keep",
563 role: "assistant",
564 text: "kept",
565 turnId: "turn-1",
566 streaming: false,
567 createdAt: "2026-02-26T12:00:01.100Z",
568 updatedAt: "2026-02-26T12:00:01.100Z",
569 },
570 }),
571 makeEvent({
572 sequence: 4,
573 type: "thread.turn-diff-completed",
574 aggregateKind: "thread",
575 aggregateId: "thread-revert",
576 occurredAt: "2026-02-26T12:00:02.000Z",
577 commandId: "cmd-turn-2",
578 payload: {
579 threadId: "thread-revert",
580 turnId: "turn-2",
581 checkpointTurnCount: 2,
582 checkpointRef: "refs/t3/checkpoints/thread-revert/turn/2",
583 status: "ready",
584 files: [],
585 assistantMessageId: "assistant-remove",
586 completedAt: "2026-02-26T12:00:02.000Z",
587 },
588 }),
589 makeEvent({
590 sequence: 5,
591 type: "thread.message-sent",
592 aggregateKind: "thread",
593 aggregateId: "thread-revert",
594 occurredAt: "2026-02-26T12:00:02.050Z",
595 commandId: "cmd-user-remove",
596 payload: {
597 threadId: "thread-revert",
598 messageId: "user-remove",
599 role: "user",
600 text: "removed",
601 turnId: "turn-2",
602 streaming: false,
603 createdAt: "2026-02-26T12:00:02.050Z",
604 updatedAt: "2026-02-26T12:00:02.050Z",
605 },
606 }),
607 makeEvent({
608 sequence: 6,
609 type: "thread.message-sent",
610 aggregateKind: "thread",
611 aggregateId: "thread-revert",
612 occurredAt: "2026-02-26T12:00:02.100Z",
613 commandId: "cmd-assistant-remove",
614 payload: {
615 threadId: "thread-revert",
616 messageId: "assistant-remove",
617 role: "assistant",
618 text: "removed",
619 turnId: "turn-2",
620 streaming: false,
621 createdAt: "2026-02-26T12:00:02.100Z",
622 updatedAt: "2026-02-26T12:00:02.100Z",
623 },
624 }),
625 makeEvent({
626 sequence: 7,
627 type: "thread.reverted",
628 aggregateKind: "thread",
629 aggregateId: "thread-revert",
630 occurredAt: "2026-02-26T12:00:03.000Z",
631 commandId: "cmd-revert",
632 payload: {
633 threadId: "thread-revert",
634 turnCount: 1,
635 },
636 }),
637 ];
638
639 const afterRevert = await events.reduce<Promise<ReturnType<typeof createEmptyReadModel>>>(
640 (statePromise, event) =>
641 statePromise.then((state) => Effect.runPromise(projectEvent(state, event))),
642 Promise.resolve(afterCreate),
643 );
644
645 const thread = afterRevert.threads[0];
646 expect(
647 thread?.messages.map((message) => ({
648 id: message.id,
649 role: message.role,
650 turnId: message.turnId,
651 })),
652 ).toEqual([{ id: "assistant-keep", role: "assistant", turnId: "turn-1" }]);
653 });
505654});
apps/server/src/orchestration/projector.ts+12-2
......@@ -65,7 +65,12 @@ function retainThreadMessagesAfterRevert(
6565 const missingUserCount = Math.max(0, turnCount - retainedUserCount);
6666 if (missingUserCount > 0) {
6767 const fallbackUserMessages = messages
68 .filter((message) => message.role === "user" && !retainedMessageIds.has(message.id))
68 .filter(
69 (message) =>
70 message.role === "user" &&
71 !retainedMessageIds.has(message.id) &&
72 (message.turnId === null || retainedTurnIds.has(message.turnId)),
73 )
6974 .toSorted(
7075 (left, right) =>
7176 left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id),
......@@ -82,7 +87,12 @@ function retainThreadMessagesAfterRevert(
8287 const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount);
8388 if (missingAssistantCount > 0) {
8489 const fallbackAssistantMessages = messages
85 .filter((message) => message.role === "assistant" && !retainedMessageIds.has(message.id))
90 .filter(
91 (message) =>
92 message.role === "assistant" &&
93 !retainedMessageIds.has(message.id) &&
94 (message.turnId === null || retainedTurnIds.has(message.turnId)),
95 )
8696 .toSorted(
8797 (left, right) =>
8898 left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id),
apps/server/src/persistence/Layers/ProjectionPendingTurnStarts.ts created+92
......@@ -0,0 +1,92 @@
1import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as SqlSchema from "effect/unstable/sql/SqlSchema";
3import { Effect, Layer } from "effect";
4
5import { toPersistenceSqlError } from "../Errors.ts";
6import {
7 DeleteProjectionPendingTurnStartInput,
8 GetProjectionPendingTurnStartInput,
9 ProjectionPendingTurnStart,
10 ProjectionPendingTurnStartRepository,
11 type ProjectionPendingTurnStartRepositoryShape,
12} from "../Services/ProjectionPendingTurnStarts.ts";
13
14const makeProjectionPendingTurnStartRepository = Effect.gen(function* () {
15 const sql = yield* SqlClient.SqlClient;
16
17 const upsertProjectionPendingTurnStartRow = SqlSchema.void({
18 Request: ProjectionPendingTurnStart,
19 execute: (row) =>
20 sql`
21 INSERT INTO projection_pending_turn_starts (
22 thread_id,
23 message_id,
24 created_at
25 )
26 VALUES (
27 ${row.threadId},
28 ${row.messageId},
29 ${row.createdAt}
30 )
31 ON CONFLICT (thread_id)
32 DO UPDATE SET
33 message_id = excluded.message_id,
34 created_at = excluded.created_at
35 `,
36 });
37
38 const getProjectionPendingTurnStartRow = SqlSchema.findOneOption({
39 Request: GetProjectionPendingTurnStartInput,
40 Result: ProjectionPendingTurnStart,
41 execute: ({ threadId }) =>
42 sql`
43 SELECT
44 thread_id AS "threadId",
45 message_id AS "messageId",
46 created_at AS "createdAt"
47 FROM projection_pending_turn_starts
48 WHERE thread_id = ${threadId}
49 `,
50 });
51
52 const deleteProjectionPendingTurnStartRow = SqlSchema.void({
53 Request: DeleteProjectionPendingTurnStartInput,
54 execute: ({ threadId }) =>
55 sql`
56 DELETE FROM projection_pending_turn_starts
57 WHERE thread_id = ${threadId}
58 `,
59 });
60
61 const upsert: ProjectionPendingTurnStartRepositoryShape["upsert"] = (row) =>
62 upsertProjectionPendingTurnStartRow(row).pipe(
63 Effect.mapError(toPersistenceSqlError("ProjectionPendingTurnStartRepository.upsert:query")),
64 );
65
66 const getByThreadId: ProjectionPendingTurnStartRepositoryShape["getByThreadId"] = (input) =>
67 getProjectionPendingTurnStartRow(input).pipe(
68 Effect.mapError(
69 toPersistenceSqlError("ProjectionPendingTurnStartRepository.getByThreadId:query"),
70 ),
71 );
72
73 const deleteByThreadId: ProjectionPendingTurnStartRepositoryShape["deleteByThreadId"] = (
74 input,
75 ) =>
76 deleteProjectionPendingTurnStartRow(input).pipe(
77 Effect.mapError(
78 toPersistenceSqlError("ProjectionPendingTurnStartRepository.deleteByThreadId:query"),
79 ),
80 );
81
82 return {
83 upsert,
84 getByThreadId,
85 deleteByThreadId,
86 } satisfies ProjectionPendingTurnStartRepositoryShape;
87});
88
89export const ProjectionPendingTurnStartRepositoryLive = Layer.effect(
90 ProjectionPendingTurnStartRepository,
91 makeProjectionPendingTurnStartRepository,
92);
apps/server/src/persistence/Migrations.ts+2
......@@ -17,6 +17,7 @@ import Migration0002 from "./Migrations/002_OrchestrationCommandReceipts.ts";
1717import Migration0003 from "./Migrations/003_CheckpointDiffBlobs.ts";
1818import Migration0004 from "./Migrations/004_ProviderSessionRuntime.ts";
1919import Migration0005 from "./Migrations/005_Projections.ts";
20import Migration0006 from "./Migrations/006_ProjectionPendingTurnStarts.ts";
2021
2122/**
2223 * Migration loader with all migrations defined inline.
......@@ -34,6 +35,7 @@ const loader = Migrator.fromRecord({
3435 "3_CheckpointDiffBlobs": Migration0003,
3536 "4_ProviderSessionRuntime": Migration0004,
3637 "5_Projections": Migration0005,
38 "6_ProjectionPendingTurnStarts": Migration0006,
3739});
3840
3941/**
apps/server/src/persistence/Migrations/006_ProjectionPendingTurnStarts.ts created+14
......@@ -0,0 +1,14 @@
1import { Effect } from "effect";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 yield* sql`
8 CREATE TABLE IF NOT EXISTS projection_pending_turn_starts (
9 thread_id TEXT PRIMARY KEY,
10 message_id TEXT NOT NULL,
11 created_at TEXT NOT NULL
12 )
13 `;
14});
apps/server/src/persistence/Services/ProjectionPendingTurnStarts.ts created+41
......@@ -0,0 +1,41 @@
1import { IsoDateTime, MessageId, ThreadId } from "@t3tools/contracts";
2import { Option, Schema, ServiceMap } from "effect";
3import type { Effect } from "effect";
4
5import type { ProjectionRepositoryError } from "../Errors.ts";
6
7export const ProjectionPendingTurnStart = Schema.Struct({
8 threadId: ThreadId,
9 messageId: MessageId,
10 createdAt: IsoDateTime,
11});
12export type ProjectionPendingTurnStart = typeof ProjectionPendingTurnStart.Type;
13
14export const GetProjectionPendingTurnStartInput = Schema.Struct({
15 threadId: ThreadId,
16});
17export type GetProjectionPendingTurnStartInput = typeof GetProjectionPendingTurnStartInput.Type;
18
19export const DeleteProjectionPendingTurnStartInput = Schema.Struct({
20 threadId: ThreadId,
21});
22export type DeleteProjectionPendingTurnStartInput = typeof DeleteProjectionPendingTurnStartInput.Type;
23
24export interface ProjectionPendingTurnStartRepositoryShape {
25 readonly upsert: (
26 row: ProjectionPendingTurnStart,
27 ) => Effect.Effect<void, ProjectionRepositoryError>;
28
29 readonly getByThreadId: (
30 input: GetProjectionPendingTurnStartInput,
31 ) => Effect.Effect<Option.Option<ProjectionPendingTurnStart>, ProjectionRepositoryError>;
32
33 readonly deleteByThreadId: (
34 input: DeleteProjectionPendingTurnStartInput,
35 ) => Effect.Effect<void, ProjectionRepositoryError>;
36}
37
38export class ProjectionPendingTurnStartRepository extends ServiceMap.Service<
39 ProjectionPendingTurnStartRepository,
40 ProjectionPendingTurnStartRepositoryShape
41>()("persistence/ProjectionPendingTurnStartRepository") {}