fix(server): an agent can mark its own thread unread when its turn ends

apps/server/src/mcp/toolkits/thread/handlers.ts

@@ -301,6 +301,11 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
           .settleThread({ ...common, byOwnAgent: caller?.id === projection.thread.id })
           .pipe(Effect.mapError(dispatchFailure));
       }
+      if (input.action === "mark_unread") {
+        return yield* threads
+          .markThreadUnread({ ...common, byOwnAgent: caller?.id === projection.thread.id })
+          .pipe(Effect.mapError(dispatchFailure));
+      }
       let command: OrchestrationV2Command;
       switch (input.action) {
         case "snooze":
@@ -316,9 +321,6 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
         case "unsettle":
           command = { ...common, type: `thread.${input.action}`, reason: "user" };
           break;
-        case "mark_unread":
-          command = { ...common, type: "thread.mark-unread" };
-          break;
         default:
           command = { ...common, type: `thread.${input.action}` };
       }

apps/server/src/mcp/toolkits/thread/tools.ts

@@ -30,7 +30,7 @@ import * as McpInvocationContext from "../../McpInvocationContext.ts";
 
 const ThreadOrganizeTool = Tool.make("t3_thread_organize", {
   description:
-    "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply. Settling this thread takes effect when your turn completes, returning settlesWhenTurnEnds=true; a turn that fails or is interrupted, or a queued message, leaves it active.",
+    "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply. Settling or marking this thread unread takes effect when your turn completes, returning settlesWhenTurnEnds=true or marksUnreadWhenTurnEnds=true; a turn that fails or is interrupted, or a queued message, leaves the thread unchanged.",
   parameters: Schema.Struct({
     threadId: Schema.optional(ThreadId),
     action: Schema.Literals([
@@ -49,6 +49,7 @@ const ThreadOrganizeTool = Tool.make("t3_thread_organize", {
   success: Schema.Union([
     OrchestrationV2DispatchCommandResult,
     Schema.Struct({ settlesWhenTurnEnds: Schema.Literal(true) }),
+    Schema.Struct({ marksUnreadWhenTurnEnds: Schema.Literal(true) }),
   ]),
   failure: OrchestratorMcpFailure,
   failureMode: "return" as const,

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

@@ -337,3 +337,44 @@ it.effect.each([
     expect(dispatched).toEqual(settles ? ["thread.settle"] : []);
   }),
 );
+
+it.effect.each([
+  { status: "completed" as const, marks: true },
+  { status: "failed" as const, marks: false },
+  { status: "interrupted" as const, marks: false },
+])("markUnreadAfterRun marks unread only when the run $status", ({ status, marks }) =>
+  Effect.gen(function* () {
+    const projectId = ProjectId.make("project:thread-management:mark-unread-after-run");
+    const threadId = ThreadId.make("thread:thread-management:mark-unread-after-run");
+    const runId = RunId.make("run:thread-management:mark-unread-after-run");
+    const dispatched: Array<string> = [];
+    const layerTest = ThreadManagementService.layer.pipe(
+      Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
+      Layer.provide(
+        Layer.mock(Orchestrator.OrchestratorV2)({
+          getThreadEventSequence: () => Effect.succeed(0),
+          restartContinuationPending: () => Effect.succeed(false),
+          getThreadRecords: () =>
+            Effect.succeed({
+              thread: { id: threadId, projectId, deletedAt: null },
+              runs: [{ id: runId, status }],
+            } as unknown as OrchestrationV2ThreadProjection),
+          dispatch: (command) =>
+            Effect.sync(() => {
+              dispatched.push(command.type);
+              return { sequence: 1, storedEvents: [] } as never;
+            }),
+        }),
+      ),
+      Layer.provide(ServerSettings.layerTest()),
+      Layer.provide(layerReceiptsTest),
+    );
+    const service = yield* ThreadManagementService.ThreadManagementService.pipe(
+      Effect.provide(layerTest),
+    );
+
+    yield* service.markUnreadAfterRun({ projectId, threadId, runId });
+
+    expect(dispatched).toEqual(marks ? ["thread.mark-unread"] : []);
+  }),
+);

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

@@ -415,6 +415,26 @@ export interface ThreadManagementServiceShape {
     { readonly sequence: number } | { readonly settlesWhenTurnEnds: true },
     Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
   >;
+  /**
+   * Marking unread rewinds the visit to just before the latest run completed,
+   * so an agent that marks its own thread unread uses this to wait for its
+   * turn to end. A run that ends any other way leaves the thread as it was,
+   * as does anything `thread.mark-unread` refuses then, such as a queued
+   * message.
+   */
+  readonly markUnreadAfterRun: (input: {
+    readonly projectId: ProjectId;
+    readonly threadId: ThreadId;
+    readonly runId: RunId;
+  }) => Effect.Effect<void, ThreadManagementFailure>;
+  readonly markThreadUnread: (input: {
+    readonly threadId: ThreadId;
+    readonly commandId: CommandId;
+    readonly byOwnAgent: boolean;
+  }) => Effect.Effect<
+    { readonly sequence: number } | { readonly marksUnreadWhenTurnEnds: true },
+    Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
+  >;
   readonly interruptThread: (
     input: ThreadManagementInterruptInput,
   ) => Effect.Effect<ThreadManagementInterruptResult, ThreadManagementFailure>;
@@ -505,7 +525,7 @@ function latestSteerableRun(
     .toSorted((left, right) => right.ordinal - left.ordinal)[0];
 }
 
-const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000;
+const AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000;
 
 const make = Effect.gen(function* () {
   const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor;
@@ -932,7 +952,7 @@ const make = Effect.gen(function* () {
   const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn(
     "orchestrationV2.threadManagement.settleAfterRun",
   )(function* (input) {
-    const { run } = yield* waitForThread({ ...input, timeoutMs: SETTLE_AFTER_RUN_WAIT_MS });
+    const { run } = yield* waitForThread({ ...input, timeoutMs: AFTER_RUN_WAIT_MS });
     if (run?.status !== "completed") return;
     yield* dispatch({
       type: "thread.settle",
@@ -961,6 +981,38 @@ const make = Effect.gen(function* () {
     return { sequence: result.sequence };
   });
 
+  const markUnreadAfterRun: ThreadManagementServiceShape["markUnreadAfterRun"] = Effect.fn(
+    "orchestrationV2.threadManagement.markUnreadAfterRun",
+  )(function* (input) {
+    const { run } = yield* waitForThread({ ...input, timeoutMs: AFTER_RUN_WAIT_MS });
+    if (run?.status !== "completed") return;
+    yield* dispatch({
+      type: "thread.mark-unread",
+      commandId: CommandId.make(`server:mark-unread-after-run:${input.runId}`),
+      threadId: input.threadId,
+    });
+  });
+
+  const markThreadUnread: ThreadManagementServiceShape["markThreadUnread"] = Effect.fn(
+    "orchestrationV2.threadManagement.markThreadUnread",
+  )(function* (input) {
+    const shell = input.byOwnAgent ? yield* orchestrator.getThreadShell(input.threadId) : null;
+    if (shell != null && shell.activeRunId !== null) {
+      yield* markUnreadAfterRun({
+        projectId: shell.projectId,
+        threadId: shell.id,
+        runId: shell.activeRunId,
+      }).pipe(Effect.ignoreCause({ log: true }), Effect.forkIn(layerScope));
+      return { marksUnreadWhenTurnEnds: true } as const;
+    }
+    const result = yield* dispatch({
+      type: "thread.mark-unread",
+      commandId: input.commandId,
+      threadId: input.threadId,
+    });
+    return { sequence: result.sequence };
+  });
+
   const interruptThread: ThreadManagementServiceShape["interruptThread"] = Effect.fn(
     "ThreadManagementService.interruptThread",
   )(function* (input) {
@@ -1186,6 +1238,8 @@ const make = Effect.gen(function* () {
     waitForThread,
     settleAfterRun,
     settleThread,
+    markUnreadAfterRun,
+    markThreadUnread,
     interruptThread,
     stopDelegatedTasks,
     stopWorkflow,

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

@@ -3700,6 +3700,82 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
     }),
   );
 
+  it.effect("marks a thread its own agent marked unread once the turn completes", () =>
+    Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const eventSink = yield* EventSink.EventSinkV2;
+      const threadManagement = yield* ThreadManagementService.ThreadManagementService;
+      const projectId = ProjectId.make("runtime-layer-mark-unread-after-run-project");
+      const threadId = ThreadId.make("runtime-layer-mark-unread-after-run-thread");
+
+      yield* orchestrator.dispatch({
+        type: "thread.create",
+        createdBy: "user",
+        creationSource: "web",
+        commandId: CommandId.make("runtime-layer-mark-unread-after-run-create"),
+        threadId,
+        projectId,
+        title: "Mark unread after run",
+        modelSelection,
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        branch: null,
+        worktreePath: "/tmp/runtime-layer-mark-unread-after-run",
+      });
+      yield* orchestrator.dispatch({
+        type: "message.dispatch",
+        createdBy: "user",
+        creationSource: "web",
+        commandId: CommandId.make("runtime-layer-mark-unread-after-run-message"),
+        threadId,
+        messageId: MessageId.make("runtime-layer-mark-unread-after-run-message"),
+        text: "Report back and then mark this thread unread.",
+        attachments: [],
+        modelSelection,
+        dispatchMode: { type: "start_immediately" },
+      });
+      const started = yield* orchestrator.getThreadProjection(threadId);
+      const run = started.runs[0];
+      if (run === undefined) return yield* Effect.die(new Error("Run missing."));
+
+      const markedUnread = yield* orchestrator
+        .streamStoredEventsFrom({ threadId, afterSequence: 0, eventType: "thread.marked-unread" })
+        .pipe(Stream.runHead, Effect.forkChild);
+      const result = yield* threadManagement.markThreadUnread({
+        threadId,
+        commandId: CommandId.make("runtime-layer-mark-unread-after-run-mark"),
+        byOwnAgent: true,
+      });
+      assert.deepEqual(result, { marksUnreadWhenTurnEnds: true });
+      assert.deepEqual(
+        (yield* orchestrator.getThreadProjection(threadId)).thread.lastVisitedAt,
+        started.thread.lastVisitedAt,
+      );
+      yield* TestClock.adjust("1 second");
+      const now = yield* DateTime.now;
+      yield* eventSink.write({
+        commandId: CommandId.make("runtime-layer-mark-unread-after-run-completed"),
+        events: [
+          {
+            id: EventId.make("runtime-layer-mark-unread-after-run-completed"),
+            type: "run.updated",
+            threadId,
+            runId: run.id,
+            occurredAt: now,
+            payload: { ...run, status: "completed", startedAt: now, completedAt: now },
+          },
+        ],
+      });
+      yield* Fiber.join(markedUnread);
+
+      const projection = yield* orchestrator.getThreadProjection(threadId);
+      assert.deepEqual(
+        projection.thread.lastVisitedAt,
+        DateTime.subtract(now, { milliseconds: 1 }),
+      );
+    }),
+  );
+
   it.effect("settles past held automatic runs but not held user messages", () =>
     Effect.gen(function* () {
       const orchestrator = yield* Orchestrator.OrchestratorV2;

apps/server/src/relay/AgentAwarenessRelay.test.ts

@@ -213,6 +213,8 @@ const makeTestRelay = Effect.fnUntraced(function* (
     waitForThread: unused,
     settleAfterRun: unused,
     settleThread: unused,
+    markUnreadAfterRun: unused,
+    markThreadUnread: unused,
     interruptThread: unused,
     stopWorkflow: () => Effect.die("unused stopWorkflow"),
     stopDelegatedTasks: unused,