feat(mcp): defer thread organization safely#8683

@@ -7052,6 +7238,41 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   // The high-water subscription deliberately skips history, so recover the
   // two terminal side effects from current projections instead: one queued
   // run per idle thread, plus any app-owned child result not yet transferred.
+  yield* projectionStore.getShellSnapshot().pipe(
+    Effect.flatMap((shell) =>
+      Effect.forEach(
+        [...shell.threads, ...shell.archivedThreads].filter(
+          (thread) => thread.deferredOrganization != null,
+        ),
+        (thread) =>
+          threadDispatch.withLock(
+            thread.id,
+            Effect.gen(function* () {
+              const projection = yield* projectionStore.getThreadProjection(thread.id);
+              const intent = projection.thread.deferredOrganization;
+              if (intent == null) return;
+              const boundRun = projection.runs.find((run) => run.id === intent.runId);
+              const hasNewerRun =
+                boundRun !== undefined &&
+                projection.runs.some((run) => run.ordinal > boundRun.ordinal);
+              if (
+                boundRun === undefined ||
+                hasNewerRun ||
+                ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes(
+                  boundRun.status,
+                )
+              ) {
+                yield* applyDeferredOrganization(thread.id, intent.runId);
+              }
+            }),
+          ),
+        { concurrency: 8, discard: true },
+      ),
+    ),
+    Effect.catchCause((cause) =>
+      Effect.logWarning("Failed to recover deferred thread organization", { cause }),
+    ),

@@ -1983,6 +1986,168 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     }
   });
 
+  const dispatchDeferredOrganization = Effect.fn("orchestrationV2.dispatch.deferredOrganization")(
+    function* (
+      command: Extract<
+        OrchestrationV2Command,
+        {
+          readonly type:
+            | "thread.organization.defer"
+            | "thread.organization.defer.cancel"
+            | "thread.organization.defer.apply";
+        }
+      >,
+      events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
+    ) {
+      const projection = yield* projectionStore.getThreadProjection(command.threadId).pipe(
+        Effect.mapError(
+          (cause) =>
+            new OrchestratorProjectionError({
+              threadId: command.threadId,
+              cause,
+            }),
+        ),
+      );
+      const thread = projection.thread;
+      if (thread.deletedAt !== null) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: `Thread ${command.threadId} is deleted.`,
+        });
+      }
+      const now = yield* DateTime.now;
+      if (command.type === "thread.organization.defer") {
+        if (thread.archivedAt !== null) {
+          return yield* new OrchestratorDispatchError({
+            commandId: command.commandId,
+            commandType: command.type,
+            cause: `Thread ${command.threadId} is archived.`,
+          });
+        }
+        const activeRuns = projection.runs
+          .filter((run) => ["preparing", "starting", "running", "waiting"].includes(run.status))
+          .toSorted((left, right) => right.ordinal - left.ordinal);
+        if (activeRuns[0]?.id !== command.runId) {
+          return yield* new OrchestratorDispatchError({
+            commandId: command.commandId,
+            commandType: command.type,
+            cause: `Run ${command.runId} is not the active run for thread ${command.threadId}.`,
+          });
+        }
+        yield* emit(
+          events,
+          command,
+        )({
+          type: "thread.metadata-updated",
+          threadId: command.threadId,
+          providerInstanceId: thread.providerInstanceId,
+          occurredAt: now,
+          payload: {
+            ...thread,
+            deferredOrganization: {
+              runId: command.runId,
+              action: command.action,
+              requestedAt: now,
+            },
+            updatedAt: now,
+          },
+        });
+        return;
+      }
+
+      if (command.type === "thread.organization.defer.cancel") {
+        yield* emit(
+          events,
+          command,
+        )({
+          type: "thread.metadata-updated",
+          threadId: command.threadId,
+          providerInstanceId: thread.providerInstanceId,
+          occurredAt: now,
+          payload: {
+            ...thread,
+            deferredOrganization: null,
+            updatedAt: thread.deferredOrganization == null ? thread.updatedAt : now,
+          },
+        });
+        return;
+      }
+
+      const intent = thread.deferredOrganization;
+      if (intent == null || intent.runId !== command.runId) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: `Thread ${command.threadId} has no deferred organization intent for run ${command.runId}.`,
+        });
+      }
+      const boundRun = projection.runs.find((run) => run.id === command.runId);
+      const hasNewerRun =
+        boundRun !== undefined && projection.runs.some((run) => run.ordinal > boundRun.ordinal);
+      const hasPendingWork =
+        projection.runs.some((run) =>
+          ["preparing", "queued", "starting", "running", "waiting"].includes(run.status),
+        ) || projection.runtimeRequests.some((request) => request.status === "pending");
+      const canApply =
+        boundRun?.status === "completed" &&
+        !hasNewerRun &&
+        !hasPendingWork &&
+        thread.titleRegeneration == null &&
+        thread.archivedAt === null;
+      if (!canApply) {
+        yield* emit(
+          events,
+          command,
+        )({
+          type: "thread.metadata-updated",
+          threadId: command.threadId,
+          providerInstanceId: thread.providerInstanceId,
+          occurredAt: now,
+          payload: { ...thread, deferredOrganization: null, updatedAt: now },
+        });
+        return;
+      }
+      if (intent.action === "settle") {

@@ -7212,6 +7416,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           threadId,
           finalizeDelegatedCompletionDelivery(threadId, stored.event.payload.id),
         );
+        yield* threadDispatch.withLock(
+          threadId,
+          applyDeferredOrganization(threadId, stored.event.payload.id),
+        );

@@ -92,6 +94,29 @@ export class OrchestratorProjectionError extends Schema.TaggedErrorClass<Orchest
   }
 }
 
+const isProjectionStoreReadError = Schema.is(ProjectionStoreReadError);
+const isOrchestratorDispatchError = Schema.is(OrchestratorDispatchError);
+
+export const runDeferredOrganizationRepair = <A, E, R>(
+  threadId: ThreadId,
+  repair: Effect.Effect<A, E, R>,
+) =>
+  repair.pipe(
+    Effect.retry({
+      times: 1,
+      while: (error) => isProjectionStoreReadError(error) || isOrchestratorDispatchError(error),

@@ -25,6 +25,11 @@ Agents connected through T3 Code can organize threads too. They can pin, snooze,
 or change read state in the current project. T3 Code applies the same rules as the sidebar. For
 example, an agent cannot settle a thread while work or a user request is still pending.
 
+An agent can also ask T3 Code to settle or archive its own thread after the current run finishes.
+The request is saved by the server, but it is applied only when that run completes successfully
+and no newer, queued, or approval-blocked work remains. A new message or unresolved request keeps
+the thread visible and discards the stale organization request.

@@ -7052,6 +7238,41 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   // The high-water subscription deliberately skips history, so recover the
   // two terminal side effects from current projections instead: one queued
   // run per idle thread, plus any app-owned child result not yet transferred.
+  yield* projectionStore.getShellSnapshot().pipe(
+    Effect.flatMap((shell) =>
+      Effect.forEach(
+        [...shell.threads, ...shell.archivedThreads].filter(
+          (thread) => thread.deferredOrganization != null,
+        ),
+        (thread) =>
+          threadDispatch.withLock(
+            thread.id,
+            Effect.gen(function* () {
+              const projection = yield* projectionStore.getThreadProjection(thread.id);
+              const intent = projection.thread.deferredOrganization;
+              if (intent == null) return;
+              const boundRun = projection.runs.find((run) => run.id === intent.runId);
+              const hasNewerRun =
+                boundRun !== undefined &&
+                projection.runs.some((run) => run.ordinal > boundRun.ordinal);
+              if (
+                boundRun === undefined ||
+                hasNewerRun ||
+                ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes(
+                  boundRun.status,
+                )
+              ) {
+                yield* applyDeferredOrganization(thread.id, intent.runId);
+              }
+            }),
+          ),

@@ -7069,6 +7242,54 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   // The high-water subscription deliberately skips history, so recover the
   // two terminal side effects from current projections instead: one queued
   // run per idle thread, plus any app-owned child result not yet transferred.
+  const recoverDeferredOrganization = projectionStore.getShellSnapshot().pipe(
+    Effect.flatMap((shell) =>
+      Effect.forEach(
+        [...shell.threads, ...shell.archivedThreads].filter(
+          (thread) => thread.deferredOrganization != null,
+        ),
+        (thread) =>
+          threadDispatch
+            .withLock(
+              thread.id,
+              Effect.gen(function* () {
+                const projection = yield* projectionStore.getThreadProjection(thread.id);
+                const intent = projection.thread.deferredOrganization;
+                if (intent == null) return;
+                const boundRun = projection.runs.find((run) => run.id === intent.runId);
+                const hasNewerRun =
+                  boundRun !== undefined &&
+                  projection.runs.some((run) => run.ordinal > boundRun.ordinal);
+                if (
+                  boundRun === undefined ||
+                  hasNewerRun ||
+                  ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes(
+                    boundRun.status,
+                  )
+                ) {
+                  yield* applyDeferredOrganization(thread.id, intent.runId);
+                }
+              }),
+            )
+            .pipe(
+              Effect.catchCause((cause) =>
+                Effect.logWarning("Failed to recover deferred thread organization", {
+                  threadId: thread.id,
+                  cause,
+                }),
+              ),
+            ),

@@ -2131,6 +2156,24 @@ export const OrchestrationV2Command = Schema.Union([
     commandId: CommandId,
     threadId: ThreadId,
   }),
+  Schema.Struct({
+    type: Schema.Literal("thread.organization.defer"),
+    commandId: CommandId,
+    threadId: ThreadId,
+    runId: RunId,
+    action: Schema.Literals(["settle", "archive"]),
+  }),
+  Schema.Struct({
+    type: Schema.Literal("thread.organization.defer.cancel"),
+    commandId: CommandId,
+    threadId: ThreadId,
+  }),
+  Schema.Struct({
+    type: Schema.Literal("thread.organization.defer.apply"),