fix(server): mark unvisited thread completions unread

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

@@ -21,6 +21,7 @@ import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
+import * as TestClock from "effect/testing/TestClock";
 import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
 import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
 import * as Orchestrator from "./Orchestrator.ts";
@@ -49,6 +50,117 @@ const testLayer = Layer.mergeAll(
   ),
 );
 
+it.effect("establishes the read watermark on fresh dispatch and preserves it across retries", () =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    for (const outcome of ["failed", "interrupted"] as const) {
+      const threadId = ThreadId.make(`thread:completion-watermark:${outcome}`);
+      yield* orchestrator.dispatch({
+        type: "thread.create",
+        commandId: CommandId.make(`create-completion-watermark-${outcome}`),
+        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 requestedAt = yield* DateTime.now;
+      const firstDispatch = {
+        type: "message.dispatch" as const,
+        commandId: CommandId.make(`first-completion-watermark-${outcome}`),
+        threadId,
+        messageId: MessageId.make(`first-completion-watermark-${outcome}`),
+        text: "Run in the background",
+        attachments: [],
+        dispatchMode: { type: "defer_start" as const },
+        createdBy: "agent" as const,
+        creationSource: "mcp" as const,
+      };
+      const rejected = yield* Effect.exit(
+        orchestrator.dispatch({
+          ...firstDispatch,
+          commandId: CommandId.make(`rejected-completion-watermark-${outcome}`),
+          modelSelection: { ...modelSelection, instanceId: ProviderInstanceId.make("missing") },
+        }),
+      );
+      assert.equal(rejected._tag, "Failure");
+      assert.isNull((yield* projections.getThreadShell(threadId))?.lastVisitedAt);
+      const accepted = yield* orchestrator.dispatch(firstDispatch);
+      const first = yield* projections.getThreadProjection(threadId);
+      assert.deepEqual(first.thread.lastVisitedAt, requestedAt);
+      assert.deepEqual((yield* projections.getThreadShell(threadId))?.lastVisitedAt, requestedAt);
+      yield* TestClock.adjust("1 second");
+      const retried = yield* orchestrator.dispatch(firstDispatch);
+      assert.equal(retried.sequence, accepted.sequence);
+      assert.deepEqual((yield* projections.getThread(threadId)).lastVisitedAt, requestedAt);
+      assert.lengthOf((yield* projections.getThreadProjection(threadId)).runs, 1);
+
+      const runId = first.runs[0]!.id;
+      if (outcome === "failed") {
+        yield* orchestrator.dispatch({
+          type: "prepared-run.fail",
+          commandId: CommandId.make(`fail-completion-watermark-${outcome}`),
+          threadId,
+          runId,
+          failure: { class: "unknown", message: "Preparation failed", code: null, retryable: true },
+        });
+      } else {
+        yield* orchestrator.dispatch({
+          type: "run.interrupt",
+          commandId: CommandId.make(`interrupt-completion-watermark-${outcome}`),
+          threadId,
+          runId,
+        });
+      }
+      const ended = yield* projections.getThreadShell(threadId);
+      assert.ok(ended?.latestRunCompletedAt);
+      assert.isAbove(
+        DateTime.toEpochMillis(ended.latestRunCompletedAt),
+        DateTime.toEpochMillis(ended.lastVisitedAt!),
+      );
+      yield* orchestrator.dispatch({
+        type: "thread.mark-unread",
+        commandId: CommandId.make(`unread-completion-watermark-${outcome}`),
+        threadId,
+      });
+      const markedUnreadAt = (yield* projections.getThread(threadId)).lastVisitedAt;
+      yield* orchestrator.dispatch({
+        ...firstDispatch,
+        commandId: CommandId.make(`second-completion-watermark-${outcome}`),
+        messageId: MessageId.make(`second-completion-watermark-${outcome}`),
+      });
+      assert.deepEqual((yield* projections.getThread(threadId)).lastVisitedAt, markedUnreadAt);
+
+      const visitedAt = DateTime.formatIso(yield* DateTime.now);
+      yield* orchestrator.dispatch({
+        type: "thread.visit",
+        commandId: CommandId.make(`visit-completion-watermark-${outcome}`),
+        threadId,
+        visitedAt,
+      });
+      yield* TestClock.adjust("1 second");
+      yield* orchestrator.dispatch({
+        ...firstDispatch,
+        commandId: CommandId.make(`queued-completion-watermark-${outcome}`),
+        messageId: MessageId.make(`queued-completion-watermark-${outcome}`),
+        dispatchMode: { type: "queue_after_active" },
+      });
+      assert.equal(
+        DateTime.formatIso((yield* projections.getThread(threadId)).lastVisitedAt!),
+        visitedAt,
+      );
+    }
+  }).pipe(Effect.provide(testLayer)),
+);
+
 it.effect(
   "dispatches metadata, queue resume and request controls without hydrating unrelated history",
   () =>

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

@@ -4407,6 +4407,23 @@ 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.
+        const event = yield* emit(
+          events,
+          command,
+        )({
+          type: "thread.metadata-updated",
+          threadId: command.threadId,
+          providerInstanceId: projection.thread.providerInstanceId,
+          occurredAt: now,
+          payload: { ...projection.thread, lastVisitedAt: now },
+        });
+        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

@@ -629,6 +629,7 @@
         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",
@@ -652,6 +653,9 @@
             ?.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")
@@ -2154,7 +2158,7 @@
       assert.equal((yield* tracker.get(launched.threadId))?.phase, "cancelled");
       const projection = yield* threads.getThreadProjection(launched.threadId);
       assert.equal(projection.runs[0]?.status, "failed");
       assert.isNull(projection.thread.worktreePath);
       assert.isEmpty(yield* outbox.listByCommandId(CommandId.make(`${input.commandId}:release`)));
     }).pipe(Effect.provide(harness.layer));
   }),