fix(server): mark unvisited thread completions unread

apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts

@@ -20,6 +20,7 @@ import {
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
+import * as TestClock from "effect/testing/TestClock";
 import * as SqlClient from "effect/sql/SqlClient";
 import * as SqlitePersistence from "../persistence/Sqlite.ts";
 import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
@@ -49,6 +50,77 @@ const layerTest = Layer.mergeAll(
   ),
 );
 
+it.effect("marks a never-visited thread's first completion unread", () =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = ThreadId.make("thread:completion-watermark");
+    yield* orchestrator.dispatch({
+      type: "thread.create",
+      commandId: CommandId.make("create-completion-watermark"),
+      threadId,
+      projectId: ProjectId.make("project:completion-watermark"),
+      title: "Background task",
+      modelSelection,
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      branch: null,
+      worktreePath: null,
+      createdBy: "user",
+      creationSource: "web",
+    });
+    yield* TestClock.adjust("1 second");
+    assert.isNull((yield* projections.getThreadShell(threadId))?.lastVisitedAt);
+    const firstDispatch = {
+      type: "message.dispatch" as const,
+      commandId: CommandId.make("first-completion-watermark"),
+      threadId,
+      messageId: MessageId.make("first-completion-watermark"),
+      text: "Run in the background",
+      attachments: [],
+      dispatchMode: { type: "defer_start" as const },
+      createdBy: "agent" as const,
+      creationSource: "mcp" as const,
+    };
+    yield* orchestrator.dispatch(firstDispatch);
+    const first = yield* projections.getThreadProjection(threadId);
+    assert.isNotNull(first.thread.lastVisitedAt);
+
+    // A completion in the dispatch's millisecond is still newer than the watermark.
+    yield* orchestrator.dispatch({
+      type: "run.interrupt",
+      commandId: CommandId.make("interrupt-completion-watermark"),
+      threadId,
+      runId: first.runs[0]!.id,
+    });
+    const ended = yield* projections.getThreadShell(threadId);
+    assert.ok(ended?.latestRunCompletedAt);
+    assert.isAbove(
+      DateTime.toEpochMillis(ended.latestRunCompletedAt),
+      DateTime.toEpochMillis(ended.lastVisitedAt!),
+    );
+
+    // Later dispatches leave an existing watermark alone.
+    const visitedAt = DateTime.formatIso(yield* DateTime.now);
+    yield* orchestrator.dispatch({
+      type: "thread.visit",
+      commandId: CommandId.make("visit-completion-watermark"),
+      threadId,
+      visitedAt,
+    });
+    yield* TestClock.adjust("1 second");
+    yield* orchestrator.dispatch({
+      ...firstDispatch,
+      commandId: CommandId.make("second-completion-watermark"),
+      messageId: MessageId.make("second-completion-watermark"),
+    });
+    assert.equal(
+      DateTime.formatIso((yield* projections.getThread(threadId)).lastVisitedAt!),
+      visitedAt,
+    );
+  }).pipe(Effect.provide(layerTest)),
+);
+
 it.effect(
   "dispatches metadata, queue resume and request controls without hydrating unrelated history",
   () =>

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

@@ -4561,6 +4561,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         }
       }
 
+      if (projection.thread.lastVisitedAt === null && !isNativeMaintenanceCommand(command)) {
+        const now = yield* DateTime.now;
+        // Null keeps imported history read. The first fresh dispatch establishes
+        // a read watermark so its completion can be unread on every client.
+        // Offset it so a completion in the same millisecond is still unread.
+        const event = yield* emit(
+          events,
+          command,
+        )({
+          type: "thread.metadata-updated",
+          threadId: command.threadId,
+          providerInstanceId: projection.thread.providerInstanceId,
+          occurredAt: now,
+          payload: {
+            ...projection.thread,
+            lastVisitedAt: DateTime.subtract(now, { milliseconds: 1 }),
+          },
+        });
+        projection = applyToProjection(projection, event);
+      }
+
       if (projection.thread.settledOverride !== null) {
         const now = yield* DateTime.now;
         const thread: OrchestrationV2AppThread = {

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

@@ -651,6 +651,7 @@ it.effect.each([" /COMPACT ", "/logout"])(
         assert.isUndefined(
           (yield* threads.getThreadProjection(launched.threadId)).thread.titleRegeneration,
         );
+        assert.isNull((yield* threads.getThreadProjection(launched.threadId)).thread.lastVisitedAt);
         assert.isFalse(
           (yield* outbox.listByCommandId(CommandId.make("compact-title:initial-message"))).some(
             (effect) => effect.request.type === "thread-title.generate",
@@ -674,6 +675,9 @@ it.effect.each([" /COMPACT ", "/logout"])(
             ?.requestId,
           commandId,
         );
+        assert.isNotNull(
+          (yield* threads.getThreadProjection(launched.threadId)).thread.lastVisitedAt,
+        );
         assert.deepEqual(
           (yield* outbox.listByCommandId(commandId))
             .filter((effect) => effect.request.type === "thread-title.generate")