feat(lastcode): add thread annotations

apps/server/src/environment/ServerEnvironment.test.ts

@@ -92,6 +92,7 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => {
       expect(second.capabilities.connectionProbe).toBe(true);
       expect(second.capabilities.pullRequests).toBe(true);
       expect(second.capabilities.threadTitleRegeneration).toBe(true);
+      expect(second.capabilities.threadAnnotations).toBe(true);
       expect(second.capabilities.agentActivityPublishing).toBe(false);
     }),
   );

apps/server/src/environment/ServerEnvironment.ts

@@ -152,6 +152,7 @@ export const make = Effect.gen(function* () {
       threadPinning: true,
       threadPinReorder: true,
       threadTitleRegeneration: true,
+      threadAnnotations: true,
       ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
       ...(serverSelfUpdate === "boot-service" ? { serverSelfUpdateProgress: true } : {}),
     },

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -6,6 +6,7 @@ import {
   MessageId,
   ProjectId,
   ThreadId,
+  ThreadAnnotation,
   TurnId,
   ProviderInstanceId,
 } from "@t3tools/contracts";
@@ -15,6 +16,7 @@ import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
 import * as Path from "effect/Path";
+import * as Schema from "effect/Schema";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
 
 import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
@@ -37,6 +39,10 @@ import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
 import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
 import { ServerConfig } from "../../config.ts";
 
+const decodeThreadAnnotationJson = Schema.decodeUnknownEffect(
+  Schema.fromJsonString(ThreadAnnotation),
+);
+
 const makeProjectionPipelinePrefixedTestLayer = (prefix: string) =>
   OrchestrationProjectionPipelineLive.pipe(
     Layer.provideMerge(OrchestrationEventStoreLive),
@@ -122,7 +128,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
         payload: {
           threadId: ThreadId.make("thread-1"),
           messageId: MessageId.make("message-1"),
-          role: "assistant",
+          role: "user",
           text: "hello",
           turnId: null,
           streaming: false,
@@ -131,6 +137,28 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
         },
       });
 
+      yield* eventStore.append({
+        type: "thread.annotation-upserted",
+        eventId: EventId.make("evt-annotation"),
+        aggregateKind: "thread",
+        aggregateId: ThreadId.make("thread-1"),
+        occurredAt: "2026-01-01T00:01:00.000Z",
+        commandId: CommandId.make("cmd-annotation"),
+        causationEventId: null,
+        correlationId: CommandId.make("cmd-annotation"),
+        metadata: {},
+        payload: {
+          threadId: ThreadId.make("thread-1"),
+          annotation: {
+            body: "# Follow up",
+            anchorMessageId: MessageId.make("message-1"),
+            createdAt: "2026-01-01T00:01:00.000Z",
+            updatedAt: "2026-01-01T00:01:00.000Z",
+            resolvedAt: null,
+          },
+        },
+      });
+
       yield* projectionPipeline.bootstrap;
 
       const projectRows = yield* sql<{
@@ -159,6 +187,20 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
       `;
       assert.deepEqual(messageRows, [{ messageId: "message-1", text: "hello" }]);
 
+      const annotationRows = yield* sql<{
+        readonly annotation: string | null;
+        readonly latestUserMessageId: string | null;
+      }>`
+        SELECT
+          annotation_json AS "annotation",
+          latest_user_message_id AS "latestUserMessageId"
+        FROM projection_threads
+        WHERE thread_id = 'thread-1'
+      `;
+      const annotation = yield* decodeThreadAnnotationJson(annotationRows[0]?.annotation);
+      assert.equal(annotation.body, "# Follow up");
+      assert.equal(annotationRows[0]?.latestUserMessageId, "message-1");
+
       const stateRows = yield* sql<{
         readonly projector: string;
         readonly lastAppliedSequence: number;
@@ -171,7 +213,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
       `;
       assert.equal(stateRows.length, Object.keys(ORCHESTRATION_PROJECTOR_NAMES).length);
       for (const row of stateRows) {
-        assert.equal(row.lastAppliedSequence, 3);
+        assert.equal(row.lastAppliedSequence, 4);
       }
 
       // Settled lifecycle through the DB pipeline: thread.settled writes the
@@ -2469,6 +2511,12 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
           role: "assistant",
         },
       ]);
+      const threadRows = yield* sql<{ readonly latestUserMessageId: string | null }>`
+        SELECT latest_user_message_id AS "latestUserMessageId"
+        FROM projection_threads
+        WHERE thread_id = 'thread-revert'
+      `;
+      assert.equal(threadRows[0]?.latestUserMessageId, null);
     }),
   );
 });

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -3,6 +3,7 @@ import {
   type ChatAttachment,
   type OrchestrationEvent,
   type OrchestrationSessionStatus,
+  type MessageId,
   ThreadId,
 } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
@@ -570,12 +571,17 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
       ]);
 
       let latestUserMessageAt: string | null = null;
+      let latestUserMessageId: MessageId | null = null;
       for (const message of messages) {
         if (
           message.role === "user" &&
-          (latestUserMessageAt === null || message.createdAt > latestUserMessageAt)
+          (latestUserMessageAt === null ||
+            message.createdAt > latestUserMessageAt ||
+            (message.createdAt === latestUserMessageAt &&
+              (latestUserMessageId === null || message.messageId > latestUserMessageId)))
         ) {
           latestUserMessageAt = message.createdAt;
+          latestUserMessageId = message.messageId;
         }
       }
 
@@ -590,6 +596,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
       yield* projectionThreadRepository.upsert({
         ...existingRow.value,
+        latestUserMessageId,
         latestUserMessageAt,
         pendingApprovalCount,
         pendingUserInputCount,
@@ -623,6 +630,8 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             pinOrderKey: null,
             titleRegenerationRequestId: null,
             titleRegenerationStartedAt: null,
+            annotation: null,
+            latestUserMessageId: null,
             latestUserMessageAt: null,
             pendingApprovalCount: 0,
             pendingUserInputCount: 0,
@@ -776,6 +785,22 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           return;
         }
 
+        case "thread.annotation-upserted":
+        case "thread.annotation-resolved":
+        case "thread.annotation-reopened": {
+          const existingRow = yield* projectionThreadRepository.getById({
+            threadId: event.payload.threadId,
+          });
+          if (Option.isNone(existingRow)) {
+            return;
+          }
+          yield* projectionThreadRepository.upsert({
+            ...existingRow.value,
+            annotation: event.payload.annotation,
+          });
+          return;
+        }
+
         case "thread.meta-updated": {
           const existingRow = yield* projectionThreadRepository.getById({
             threadId: event.payload.threadId,

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts

@@ -85,12 +85,14 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           branch,
           worktree_path,
           latest_turn_id,
+          latest_user_message_id,
           latest_user_message_at,
           pending_approval_count,
           pending_user_input_count,
           has_actionable_proposed_plan,
           pinned_at,
           pin_order_key,
+          annotation_json,
           created_at,
           updated_at,
           deleted_at
@@ -105,12 +107,14 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           NULL,
           NULL,
           'turn-1',
+          'message-1',
           '2026-02-24T00:00:04.000Z',
           1,
           0,
           0,
           '2026-02-24T00:00:01.000Z',
           'gm',
+          '{"body":"# Follow up","anchorMessageId":"message-1","createdAt":"2026-02-24T00:00:02.500Z","updatedAt":"2026-02-24T00:00:02.500Z","resolvedAt":null}',
           '2026-02-24T00:00:02.000Z',
           '2026-02-24T00:00:03.000Z',
           NULL
@@ -328,6 +332,13 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           pinnedAt: "2026-02-24T00:00:01.000Z",
           pinOrderKey: "gm",
           titleRegeneration: null,
+          annotation: {
+            body: "# Follow up",
+            anchorMessageId: asMessageId("message-1"),
+            createdAt: "2026-02-24T00:00:02.500Z",
+            updatedAt: "2026-02-24T00:00:02.500Z",
+            resolvedAt: null,
+          },
           deletedAt: null,
           messages: [
             {
@@ -447,6 +458,13 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           pinnedAt: "2026-02-24T00:00:01.000Z",
           pinOrderKey: "gm",
           titleRegeneration: null,
+          annotation: {
+            body: "# Follow up",
+            anchorMessageId: asMessageId("message-1"),
+            createdAt: "2026-02-24T00:00:02.500Z",
+            updatedAt: "2026-02-24T00:00:02.500Z",
+            resolvedAt: null,
+          },
           session: {
             threadId: ThreadId.make("thread-1"),
             status: "running",
@@ -516,6 +534,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           branch,
           worktree_path,
           latest_turn_id,
+          latest_user_message_id,
           latest_user_message_at,
           pending_approval_count,
           pending_user_input_count,
@@ -537,6 +556,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
             NULL,
             NULL,
             NULL,
+            NULL,
             0,
             0,
             0,
@@ -556,6 +576,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
             NULL,
             NULL,
             NULL,
+            NULL,
             0,
             0,
             0,
@@ -636,6 +657,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           branch,
           worktree_path,
           latest_turn_id,
+          latest_user_message_id,
           latest_user_message_at,
           pending_approval_count,
           pending_user_input_count,
@@ -658,6 +680,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           NULL,
           NULL,
           NULL,
+          NULL,
           0,
           0,
           0,
@@ -1293,7 +1316,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
     }),
   );
 
-  it.effect("uses projection_threads.latest_turn_id for bulk command and shell snapshots", () =>
+  it.effect("uses projection_threads latest markers for bulk command and shell snapshots", () =>
     Effect.gen(function* () {
       const snapshotQuery = yield* ProjectionSnapshotQuery;
       const sql = yield* SqlClient.SqlClient;
@@ -1337,6 +1360,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           branch,
           worktree_path,
           latest_turn_id,
+          latest_user_message_id,
           latest_user_message_at,
           pending_approval_count,
           pending_user_input_count,
@@ -1356,6 +1380,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           NULL,
           NULL,
           'turn-running',
+          'message-user-2',
           '2026-04-03T00:00:04.000Z',
           0,
           0,
@@ -1367,6 +1392,29 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
         )
       `;
 
+      yield* sql`
+        INSERT INTO projection_thread_messages (
+          message_id,
+          thread_id,
+          turn_id,
+          role,
+          text,
+          is_streaming,
+          created_at,
+          updated_at
+        )
+        VALUES (
+          'message-user-2',
+          'thread-1',
+          NULL,
+          'user',
+          'Latest prompt',
+          0,
+          '2026-04-03T00:00:30.000Z',
+          '2026-04-03T00:00:30.000Z'
+        )
+      `;
+
       yield* sql`
         INSERT INTO projection_turns (
           thread_id,
@@ -1434,6 +1482,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
       const commandReadModel = yield* snapshotQuery.getCommandReadModel();
       assert.equal(commandReadModel.threads[0]?.latestTurn?.turnId, asTurnId("turn-running"));
       assert.equal(commandReadModel.threads[0]?.latestTurn?.state, "running");
+      assert.equal(commandReadModel.threads[0]?.messages.length, 0);
+      assert.equal(commandReadModel.threads[0]?.latestUserMessageId, asMessageId("message-user-2"));
 
       const shellSnapshot = yield* snapshotQuery.getShellSnapshot();
       assert.equal(shellSnapshot.threads[0]?.latestTurn?.turnId, asTurnId("turn-running"));