fix(server): automatic compactions no longer mark read threads unread

apps/server/src/orchestration-v2/IdleCompactionService.test.ts

@@ -0,0 +1,87 @@
+import { expect, it } from "@effect/vitest";
+import type {
+  OrchestrationV2DomainEvent,
+  OrchestrationV2Run,
+  OrchestrationV2ServerCommand,
+  OrchestrationV2ThreadProjection,
+} from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Queue from "effect/Queue";
+import * as Stream from "effect/Stream";
+
+import * as ServerSettings from "../serverSettings.ts";
+import * as IdleCompactionService from "./IdleCompactionService.ts";
+import * as Orchestrator from "./Orchestrator.ts";
+import * as ProjectionStore from "./ProjectionStore.ts";
+
+const at = (minute: number) => DateTime.makeUnsafe(Date.UTC(2026, 9, 19, 12, minute));
+
+function compactedThread(threadId: string, lastVisitedAt: DateTime.Utc | null) {
+  const run = (ordinal: number, userMessageId: string, completedAt: DateTime.Utc) =>
+    ({
+      id: `${threadId}:run-${ordinal}`,
+      threadId,
+      ordinal,
+      userMessageId,
+      status: "completed",
+      requestedAt: DateTime.subtract(completedAt, { minutes: 1 }),
+      completedAt,
+    }) as unknown as OrchestrationV2Run;
+  const previous = run(1, "message", at(0));
+  const compaction = run(2, `idle-compact:${threadId}:${previous.id}`, at(58));
+  return {
+    records: {
+      thread: { id: threadId, lastVisitedAt },
+      runs: [previous, compaction],
+    } as unknown as OrchestrationV2ThreadProjection,
+    completed: {
+      type: "run.updated",
+      threadId,
+      payload: compaction,
+    } as unknown as OrchestrationV2DomainEvent,
+  };
+}
+
+it.effect("marks a caught-up thread read when its automatic compaction completes", () =>
+  Effect.gen(function* () {
+    const threads = [
+      compactedThread("unseen", null),
+      compactedThread("behind", at(-1)),
+      compactedThread("marked-unread", DateTime.subtract(at(58), { milliseconds: 1 })),
+      compactedThread("caught-up", at(1)),
+    ];
+    const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
+    const commands = yield* Queue.unbounded<OrchestrationV2ServerCommand>();
+    const service = yield* IdleCompactionService.make.pipe(
+      Effect.provide(
+        Layer.mergeAll(
+          Layer.mock(Orchestrator.OrchestratorV2)({
+            streamDomainEvents: Stream.fromQueue(events),
+            dispatch: (command) =>
+              Queue.offer(commands, command).pipe(Effect.as({ sequence: 1, storedEvents: [] })),
+          }),
+          Layer.mock(ProjectionStore.ProjectionStoreV2)({
+            getThreadRecords: (threadId) =>
+              Effect.succeed(threads.find((t) => t.records.thread.id === threadId)!.records),
+          }),
+          ServerSettings.layerTest(),
+        ),
+      ),
+    );
+    yield* service.start();
+
+    yield* Queue.offerAll(
+      events,
+      threads.map((thread) => thread.completed),
+    );
+
+    expect(yield* Queue.take(commands)).toMatchObject({
+      type: "thread.visit",
+      threadId: "caught-up",
+      visitedAt: DateTime.formatIso(at(58)),
+      opened: false,
+    });
+  }).pipe(Effect.scoped),
+);

apps/server/src/orchestration-v2/IdleCompactionService.ts

@@ -2,6 +2,7 @@ import {
   CommandId,
   IDLE_COMPACTION_TEXT,
   MessageId,
+  type OrchestrationV2Run,
   resolveIdleCompaction,
   resolveProjectProviderInstanceEnabled,
   type RunId,
@@ -19,6 +20,8 @@ import { idleCompactionWindow } from "./IdleCompaction.ts";
 import * as Orchestrator from "./Orchestrator.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 
+const IDLE_COMPACTION_MESSAGE_PREFIX = "idle-compact:";
+
 /**
  * Beta: when a run completes on a Claude instance set to compact before the
  * cache expires, waits until shortly before its prompt cache expires and sends
@@ -69,7 +72,7 @@ export const make = Effect.gen(function* () {
           type: "message.dispatch",
           // One compaction per completed run, even across duplicate events.
           commandId: CommandId.make(`server:idle-compact:${threadId}:${runId}`),
-          messageId: MessageId.make(`idle-compact:${threadId}:${runId}`),
+          messageId: MessageId.make(`${IDLE_COMPACTION_MESSAGE_PREFIX}${threadId}:${runId}`),
           threadId,
           text: IDLE_COMPACTION_TEXT,
           attachments: [],
@@ -87,13 +90,56 @@ export const make = Effect.gen(function* () {
       ),
   );
 
+  const markSeenIfCaughtUp = Effect.fn("IdleCompactionService.markSeenIfCaughtUp")(
+    function* (run: OrchestrationV2Run) {
+      if (run.completedAt === null || !run.userMessageId.startsWith(IDLE_COMPACTION_MESSAGE_PREFIX))
+        return;
+      const { thread, runs } = yield* projections.getThreadRecords(run.threadId, ["runs"]);
+      const previous = runs.reduce<OrchestrationV2Run | null>(
+        (latest, candidate) =>
+          candidate.ordinal < run.ordinal && (latest === null || candidate.ordinal > latest.ordinal)
+            ? candidate
+            : latest,
+        null,
+      );
+      if (
+        thread.lastVisitedAt === null ||
+        previous?.completedAt == null ||
+        DateTime.isGreaterThan(previous.completedAt, thread.lastVisitedAt) ||
+        !DateTime.isLessThan(thread.lastVisitedAt, run.requestedAt)
+      )
+        return;
+      yield* orchestrator.dispatch({
+        type: "thread.visit",
+        commandId: CommandId.make(`server:idle-compact-visit:${run.threadId}:${run.id}`),
+        threadId: run.threadId,
+        visitedAt: DateTime.formatIso(run.completedAt),
+        opened: false,
+      });
+    },
+    (effect, run) =>
+      effect.pipe(
+        Effect.catch((cause) =>
+          Effect.logWarning("idle compaction visit skipped", { threadId: run.threadId, cause }),
+        ),
+      ),
+  );
+
   const start = Effect.fn("IdleCompactionService.start")(function* () {
     // Keyed by thread: a newer completed run replaces the waiting timer.
     const timers = yield* FiberMap.make<ThreadId>();
     yield* forkParked(
       Stream.runForEach(orchestrator.streamDomainEvents, (event) =>
         event.type === "run.updated" && event.payload.status === "completed"
-          ? FiberMap.run(timers, event.threadId, compactWhenDue(event.threadId, event.payload.id))
+          ? markSeenIfCaughtUp(event.payload).pipe(
+              Effect.andThen(
+                FiberMap.run(
+                  timers,
+                  event.threadId,
+                  compactWhenDue(event.threadId, event.payload.id),
+                ),
+              ),
+            )
           : Effect.void,
       ),
     );