feat(server): agents can settle their own thread after the run ends

apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts

@@ -781,10 +781,42 @@ describe("orchestrator MCP toolkit", () => {
               "running",
             );
 
+            const otherThreadId = ThreadId.make("thread:mcp-other");
+            yield* orchestrator.dispatch({
+              type: "thread.create",
+              createdBy: "user",
+              creationSource: "web",
+              commandId: CommandId.make("command:mcp-other:create"),
+              threadId: otherThreadId,
+              projectId,
+              title: "MCP other",
+              modelSelection: codexSelection,
+              runtimeMode: "full-access",
+              interactionMode: "default",
+              branch: null,
+              worktreePath: cwd,
+            });
+            const otherThreadSettle = yield* invoke("t3_thread_organize", {
+              action: "settle_after_run",
+              threadId: otherThreadId,
+            });
+            expect(declaredFailure(otherThreadSettle)).toMatchObject({ code: "invalid_request" });
+            const deferredSettle = yield* invoke("t3_thread_organize", {
+              action: "settle_after_run",
+            });
+            expect(deferredSettle.isError).toBe(false);
+            const afterDeferredSettle = yield* orchestrator.getThreadProjection(parentThreadId);
+            expect(afterDeferredSettle.thread.settleAfterRunId).toBe(parentRun?.id);
+            expect(afterDeferredSettle.thread.settledOverride).not.toBe("settled");
+
+            // Pinning withdraws the request, so the run's end leaves the thread active.
             const pinned = yield* invoke("t3_thread_organize", { action: "pin" });
             expect(pinned.isError).toBe(false);
             expect(pinned.structuredContent).toHaveProperty("sequence");
             expect((yield* orchestrator.getThreadShell(parentThreadId))?.pinnedAt).not.toBeNull();
+            expect(
+              (yield* orchestrator.getThreadProjection(parentThreadId)).thread.settleAfterRunId,
+            ).toBeNull();
             yield* invoke("t3_thread_organize", { action: "unpin" });
             expect((yield* orchestrator.getThreadShell(parentThreadId))?.pinnedAt).toBeNull();
 

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

@@ -292,8 +292,21 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
   ),
   t3_thread_organize: writesThread((input) =>
     Effect.gen(function* () {
-      const { threads, projection } = yield* readThread(input.threadId);
+      const { threads, projection, caller } = yield* readThread(input.threadId);
       const common = { commandId: yield* newCommandId(), threadId: projection.thread.id };
+      if (input.action === "settle_after_run") {
+        // Only the agent running a thread knows whether its run was worth the user's attention.
+        if (caller === undefined || caller.id !== projection.thread.id || !caller.activeRunId) {
+          return yield* new OrchestratorMcpFailure({
+            code: "invalid_request",
+            message: "settle_after_run applies only to the calling thread's own active run.",
+          });
+        }
+        const result = yield* threads
+          .dispatch({ ...common, type: "thread.settle-after-run", runId: caller.activeRunId })
+          .pipe(Effect.mapError(dispatchFailure));
+        return { sequence: result.sequence };
+      }
       let command: OrchestrationV2Command;
       switch (input.action) {
         case "snooze":

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; this does not schedule a future action.",
+    "Pin, snooze, settle, archive, or mark a thread unread. Omit threadId for this thread. snooze requires snoozedUntil. Existing thread lifecycle rules apply. settle fails on a thread with an active run, including this one; to settle this thread when your current run finishes, use settle_after_run with threadId omitted. It settles only if the run completes and no newer message, scheduled delivery, unsettle, or pin reaches the thread first; otherwise the thread stays as it is.",
   parameters: Schema.Struct({
     threadId: Schema.optional(ThreadId),
     action: Schema.Literals([
@@ -39,6 +39,7 @@ const ThreadOrganizeTool = Tool.make("t3_thread_organize", {
       "snooze",
       "unsnooze",
       "settle",
+      "settle_after_run",
       "unsettle",
       "archive",
       "unarchive",

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

@@ -449,6 +449,8 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId {
     case "checkpoint.rollback.fail":
     case "thread.background-work.settle":
     case "thread.stop":
+    case "thread.settle-after-run":
+    case "thread.settle-after-run.apply":
     case "provider.switch":
       return command.threadId;
     case "delegated_task.request":
@@ -2770,6 +2772,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             ...thread,
             settledOverride: "settled",
             settledAt: alreadySettled ? thread.settledAt : (command.settledAt ?? now),
+            settleAfterRunId: null,
             pullRequests: thread.pullRequests?.map((link) => withPullRequestWatch(link, undefined)),
             unsettledAt: null,
             pinnedAt: null,
@@ -2784,6 +2787,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             ...thread,
             settledOverride: "active",
             settledAt: null,
+            settleAfterRunId: null,
             unsettledAt: alreadyPinnedActive ? (thread.unsettledAt ?? null) : now,
             updatedAt: alreadyPinnedActive ? thread.updatedAt : now,
           };
@@ -2837,6 +2841,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             settledOverride:
               thread.settledOverride === "settled" ? "active" : thread.settledOverride,
             settledAt: thread.settledOverride === "settled" ? null : thread.settledAt,
+            settleAfterRunId: null,
             snoozedUntil: null,
             snoozedAt: null,
             updatedAt: alreadyPinned && !promotes ? thread.updatedAt : now,
@@ -4536,6 +4541,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         }
       }
 
+      // A message outranks its thread agent's request to settle after the run.
+      if (projection.thread.settleAfterRunId != null) {
+        yield* emit(
+          events,
+          command,
+        )({
+          type: "thread.metadata-updated",
+          threadId: command.threadId,
+          providerInstanceId: projection.thread.providerInstanceId,
+          occurredAt: yield* DateTime.now,
+          payload: { ...projection.thread, settleAfterRunId: null },
+        });
+        projection = yield* getProjectionWithPendingEvents(command.threadId, events);
+      }
       if (projection.thread.settledOverride !== null) {
         const now = yield* DateTime.now;
         const thread: OrchestrationV2AppThread = {
@@ -10097,6 +10116,100 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       }),
     );
 
+  const dispatchSettleAfterRunRequest = (
+    command: Extract<OrchestrationV2ServerCommand, { readonly type: "thread.settle-after-run" }>,
+    events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
+  ) =>
+    Effect.gen(function* () {
+      const projection = yield* projectionStore
+        .getThreadRecords(command.threadId, ["runs"])
+        .pipe(
+          Effect.mapError(() => new OrchestratorProjectionError({ threadId: command.threadId })),
+        );
+      const reject = (cause: string) =>
+        new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause,
+        });
+      const thread = projection.thread;
+      if (thread.archivedAt !== null || thread.deletedAt !== null) {
+        return yield* reject(`Thread ${command.threadId} is archived or deleted.`);
+      }
+      const run = projection.runs.find((candidate) => candidate.id === command.runId);
+      if (run === undefined || !["starting", "running", "waiting"].includes(run.status)) {
+        return yield* reject(`Run ${command.runId} is not active on thread ${command.threadId}.`);
+      }
+      if (projection.runs.some((candidate) => candidate.ordinal > run.ordinal)) {
+        return yield* reject(
+          `Thread ${command.threadId} already has newer work queued, so it stays unsettled.`,
+        );
+      }
+      yield* emit(
+        events,
+        command,
+      )({
+        type: "thread.metadata-updated",
+        threadId: command.threadId,
+        providerInstanceId: thread.providerInstanceId,
+        occurredAt: yield* DateTime.now,
+        payload: { ...thread, settleAfterRunId: run.id },
+      });
+    });
+
+  const dispatchSettleAfterRunApply = (
+    command: Extract<
+      OrchestrationV2ServerCommand,
+      { readonly type: "thread.settle-after-run.apply" }
+    >,
+    events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
+    effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
+  ) =>
+    Effect.gen(function* () {
+      const projection = yield* projectionStore
+        .getThreadRecords(command.threadId, ["runs"])
+        .pipe(
+          Effect.mapError(() => new OrchestratorProjectionError({ threadId: command.threadId })),
+        );
+      const requested = projection.thread.settleAfterRunId === command.runId;
+      const run = projection.runs.find((candidate) => candidate.id === command.runId);
+      // A message dispatched after the request already cleared it; a later
+      // run that skipped message dispatch still wins here.
+      const settles =
+        requested &&
+        run?.status === "completed" &&
+        !projection.runs.some((candidate) => candidate.ordinal > run.ordinal);
+      if (settles) {
+        const eventCountBeforeSettle = (yield* Ref.get(events)).length;
+        const settled = yield* dispatchThreadMutation(
+          { type: "thread.settle", commandId: command.commandId, threadId: command.threadId },
+          events,
+          effects,
+        ).pipe(
+          Effect.as(true),
+          // Blocked work, such as a pending approval, keeps the thread active.
+          Effect.catchTags({
+            OrchestratorDispatchError: () =>
+              Ref.update(events, (existing) => existing.slice(0, eventCountBeforeSettle)).pipe(
+                Effect.as(false),
+              ),
+          }),
+        );
+        if (settled) return;
+      }
+      // The request is spent either way; record the no-op so the receipt is accepted.
+      yield* emit(
+        events,
+        command,
+      )({
+        type: "thread.metadata-updated",
+        threadId: command.threadId,
+        providerInstanceId: projection.thread.providerInstanceId,
+        occurredAt: yield* DateTime.now,
+        payload: requested ? { ...projection.thread, settleAfterRunId: null } : projection.thread,
+      });
+    });
+
   const dispatchOnce = Effect.fn("orchestrationV2.dispatch.once")(function* (
     command: OrchestrationV2ServerCommand,
   ): Effect.fn.Return<
@@ -10363,6 +10476,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       case "thread.background-work.settle":
         yield* dispatchBackgroundWorkSettle(command, events, effects);
         break;
+      case "thread.settle-after-run":
+        yield* dispatchSettleAfterRunRequest(command, events);
+        break;
+      case "thread.settle-after-run.apply":
+        yield* dispatchSettleAfterRunApply(command, events, effects);
+        break;
       case "thread.fork":
         yield* dispatchThreadFork(command, events);
         break;
@@ -10622,6 +10741,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           threadId,
           finalizeDelegatedCompletionDelivery(threadId, stored.event.payload.id),
         );
+        // Before queue promotion, so a queued run is still visible to the request's guard.
+        // A failure leaves the thread unsettled and must not hold up the queue.
+        const runId = stored.event.payload.id;
+        yield* projectionStore.getThread(threadId).pipe(
+          Effect.flatMap((thread) =>
+            thread.settleAfterRunId === runId
+              ? dispatchWithReceipt({
+                  type: "thread.settle-after-run.apply",
+                  commandId: CommandId.make(`command:system:settle-after-run:${runId}`),
+                  threadId,
+                  runId,
+                })
+              : Effect.void,
+          ),
+          Effect.catchCause((cause) =>
+            Effect.logWarning("Failed to settle a thread after its run", {
+              threadId,
+              runId,
+              cause,
+            }),
+          ),
+        );
       }
       yield* threadDispatch.withLock(
         threadId,