fix(server): mark unvisited thread completions unread

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -4400,15 +4400,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
                 node: artifacts.node,
               });
               yield* emitProviderEvent({
-                type: "runtime_request.updated",
+                type: "turn_item.updated",
                 driver: CODEX_PROVIDER,
-                threadId: artifacts.node.threadId,
-                runtimeRequest: artifacts.request,
+                turnItem: artifacts.turnItem,
               });
               yield* emitProviderEvent({
-                type: "turn_item.updated",
+                type: "runtime_request.updated",
                 driver: CODEX_PROVIDER,
-                turnItem: artifacts.turnItem,
+                threadId: artifacts.node.threadId,
+                runtimeRequest: artifacts.request,
               });
               yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
                 const ids = current.get(payload.turnId);
@@ -4507,15 +4507,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               node: artifacts.node,
             });
             yield* emitProviderEvent({
-              type: "runtime_request.updated",
+              type: "turn_item.updated",
               driver: CODEX_PROVIDER,
-              threadId: artifacts.node.threadId,
-              runtimeRequest: artifacts.request,
+              turnItem: artifacts.turnItem,
             });
             yield* emitProviderEvent({
-              type: "turn_item.updated",
+              type: "runtime_request.updated",
               driver: CODEX_PROVIDER,
-              turnItem: artifacts.turnItem,
+              threadId: artifacts.node.threadId,
+              runtimeRequest: artifacts.request,
             });
 
             const resolved = yield* Deferred.await(decision).pipe(
@@ -4567,15 +4567,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               node: artifacts.node,
             });
             yield* emitProviderEvent({
-              type: "runtime_request.updated",
+              type: "turn_item.updated",
               driver: CODEX_PROVIDER,
-              threadId: artifacts.node.threadId,
-              runtimeRequest: artifacts.request,
+              turnItem: artifacts.turnItem,
             });
             yield* emitProviderEvent({
-              type: "turn_item.updated",
+              type: "runtime_request.updated",
               driver: CODEX_PROVIDER,
-              turnItem: artifacts.turnItem,
+              threadId: artifacts.node.threadId,
+              runtimeRequest: artifacts.request,
             });
 
             const resolved = yield* Deferred.await(decision).pipe(
@@ -4628,15 +4628,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               node: artifacts.node,
             });
             yield* emitProviderEvent({
-              type: "runtime_request.updated",
+              type: "turn_item.updated",
               driver: CODEX_PROVIDER,
-              threadId: artifacts.node.threadId,
-              runtimeRequest: artifacts.request,
+              turnItem: artifacts.turnItem,
             });
             yield* emitProviderEvent({
-              type: "turn_item.updated",
+              type: "runtime_request.updated",
               driver: CODEX_PROVIDER,
-              turnItem: artifacts.turnItem,
+              threadId: artifacts.node.threadId,
+              runtimeRequest: artifacts.request,
             });
 
             const resolved = yield* Deferred.await(decision).pipe(
@@ -4713,15 +4713,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               node: artifacts.node,
             });
             yield* emitProviderEvent({
-              type: "runtime_request.updated",
+              type: "turn_item.updated",
               driver: CODEX_PROVIDER,
-              threadId: artifacts.node.threadId,
-              runtimeRequest: artifacts.request,
+              turnItem: artifacts.turnItem,
             });
             yield* emitProviderEvent({
-              type: "turn_item.updated",
+              type: "runtime_request.updated",
               driver: CODEX_PROVIDER,
-              turnItem: artifacts.turnItem,
+              threadId: artifacts.node.threadId,
+              runtimeRequest: artifacts.request,
             });
 
             const resolved = yield* Deferred.await(decision).pipe(
@@ -4772,15 +4772,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               node: artifacts.node,
             });
             yield* emitProviderEvent({
-              type: "runtime_request.updated",
+              type: "turn_item.updated",
               driver: CODEX_PROVIDER,
-              threadId: artifacts.node.threadId,
-              runtimeRequest: artifacts.request,
+              turnItem: artifacts.turnItem,
             });
             yield* emitProviderEvent({
-              type: "turn_item.updated",
+              type: "runtime_request.updated",
               driver: CODEX_PROVIDER,
-              turnItem: artifacts.turnItem,
+              threadId: artifacts.node.threadId,
+              runtimeRequest: artifacts.request,
             });
 
             const resolved = yield* Deferred.await(decision).pipe(
@@ -4832,15 +4832,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               node: artifacts.node,
             });
             yield* emitProviderEvent({
-              type: "runtime_request.updated",
+              type: "turn_item.updated",
               driver: CODEX_PROVIDER,
-              threadId: artifacts.node.threadId,
-              runtimeRequest: artifacts.request,
+              turnItem: artifacts.turnItem,
             });
             yield* emitProviderEvent({
-              type: "turn_item.updated",
+              type: "runtime_request.updated",
               driver: CODEX_PROVIDER,
-              turnItem: artifacts.turnItem,
+              threadId: artifacts.node.threadId,
+              runtimeRequest: artifacts.request,
             });
 
             const resolved = yield* Deferred.await(decision).pipe(
@@ -4890,15 +4890,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               node: artifacts.node,
             });
             yield* emitProviderEvent({
-              type: "runtime_request.updated",
+              type: "turn_item.updated",
               driver: CODEX_PROVIDER,
-              threadId: artifacts.node.threadId,
-              runtimeRequest: artifacts.request,
+              turnItem: artifacts.turnItem,
             });
             yield* emitProviderEvent({
-              type: "turn_item.updated",
+              type: "runtime_request.updated",
               driver: CODEX_PROVIDER,
-              turnItem: artifacts.turnItem,
+              threadId: artifacts.node.threadId,
+              runtimeRequest: artifacts.request,
             });
 
             const resolved = yield* Deferred.await(answers).pipe(

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

@@ -33,6 +33,7 @@ import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Layer from "effect/Layer";
+import * as Logger from "effect/Logger";
 import * as Option from "effect/Option";
 import * as Ref from "effect/Ref";
 import * as Stream from "effect/Stream";
@@ -629,6 +630,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",
@@ -652,6 +654,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")
@@ -2133,7 +2138,16 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch
 it.effect("cancels tracked setup before provider work is released", () =>
   Effect.gen(function* () {
     const entered = yield* Deferred.make<void>();
+    const allowRename = yield* Deferred.make<void>();
+    const renameRejected = yield* Deferred.make<void>();
+    const logger = Logger.make(({ message }) => {
+      if (String(message).includes("Thread worktree branch rename failed")) {
+        Deferred.doneUnsafe(renameRejected, Effect.void);
+      }
+    });
     const harness = makeHarness({
+      generateBranchName: () =>
+        Deferred.await(allowRename).pipe(Effect.as({ branch: "generated-branch" })),
       runSetup: () => Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)),
     });
     yield* Effect.gen(function* () {
@@ -2152,11 +2166,25 @@ it.effect("cancels tracked setup before provider work is released", () =>
       assert.equal((yield* tracker.get(launched.threadId))?.phase, "running");
       assert.isTrue(yield* tracker.cancel(launched.threadId));
       assert.equal((yield* tracker.get(launched.threadId))?.phase, "cancelled");
+      yield* Deferred.succeed(allowRename, undefined);
+      yield* Effect.raceFirst(
+        Deferred.await(renameRejected),
+        threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe(
+          Stream.filter(
+            (stored) => stored.commandId === CommandId.make(`${input.commandId}:branch-rename`),
+          ),
+          Stream.runHead,
+        ),
+      );
       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));
+    }).pipe(
+      Effect.provide(
+        Layer.merge(harness.layer, Logger.layer([logger], { mergeWithExisting: false })),
+      ),
+    );
   }),
 );
 

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

@@ -436,6 +436,7 @@ const make = Effect.gen(function* () {
               threadId,
               branch: renamed.branch,
               worktreePath: worktreeCwd,
+              expectedWorktreePath: worktreeCwd,
             }),
           ),
           Effect.catchCause((cause) =>