fix(server): a compaction cut by a restart runs again and sends what was queued behind it

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

@@ -2705,15 +2705,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   // the server or scheduled tasks, so a later completed run releases those unless it was stopped.
   const releaseUnattendedHolds = Effect.fn("Orchestrator.releaseUnattendedHolds")(function* (
     threadId: ThreadId,
-    completedRunId: RunId,
+    completedRun: OrchestrationV2Run,
   ) {
+    const completedRunId = completedRun.id;
     const held = yield* projectionStore.getHeldQueuedRuns(threadId);
     if (held.length === 0) return;
     const projection = yield* projectionStore.getThreadRecords(
       threadId,
       ["runs", "messages", "turnItems"],
       {
-        messageIds: held.map((run) => run.userMessageId),
+        messageIds: [...held.map((run) => run.userMessageId), completedRun.userMessageId],
         turnItemTypes: ["run_interrupt_request"],
         turnItemRunIds: [completedRunId],
       },
@@ -2732,8 +2733,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         .filter((message) => message.createdBy !== "user" || message.scheduledTaskId !== undefined)
         .map((message) => message.id),
     );
+    // A restart that cut a compaction records the holds it added; redoing it releases those.
+    // Retried or re-cut continuations chain back to the run that recorded them.
+    const restartHeld = new Set<RunId>();
+    const isCompaction = messages.some(
+      (message) => message.id === completedRun.userMessageId && isCompactionCommandMessage(message),
+    );
+    let id = isCompaction ? completedRun.restartContinuationOfRunId : undefined;
+    for (let hops = 0; id !== undefined && hops < projection.runs.length; hops++) {
+      const source = projection.runs.find((run) => run.id === id);
+      for (const heldId of source?.restartHeldRunIds ?? []) restartHeld.add(heldId);
+      id = source?.restartContinuationOfRunId;
+    }
     const now = yield* DateTime.now;
-    const released = held.filter((run) => unattended.has(run.userMessageId));
+    const released = held.filter(
+      (run) => unattended.has(run.userMessageId) || restartHeld.has(run.id),
+    );
     if (released.length === 0) return;
     yield* writeSystemEvents(
       released.map((run) => ({
@@ -14237,10 +14252,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         yield* threadDispatch
           .withLock(
             threadId,
-            (run.status === "completed"
-              ? releaseUnattendedHolds(threadId, run.id)
-              : Effect.void
-            ).pipe(
+            (run.status === "completed" ? releaseUnattendedHolds(threadId, run) : Effect.void).pipe(
               Effect.andThen(
                 startNextQueuedRun(threadId, {
                   endedRunId: run.id,

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

@@ -710,6 +710,8 @@ it.effect(
           providerThreadId,
           providerInstanceId: ProviderInstanceId.make("codex"),
         },
+        { id: RunId.make("run_recovery_queued"), status: "queued", ordinal: 2 },
+        { id: RunId.make("run_recovery_held"), status: "queued", ordinal: 3, queueHeld: true },
       ],
       attempts: [
         {
@@ -746,6 +748,8 @@ it.effect(
           Layer.mock(ProjectionStore.ProjectionStoreV2)({
             getRecoveryThreadIds: () => Effect.succeed([threadId]),
             getRuntimeRecoveryProjection: () => Effect.succeed(projection),
+            getMessagesById: () => Effect.succeed(new Map()),
+            getThreadRecords: () => Effect.succeed(projection),
           }),
           Layer.mock(EventSink.EventSinkV2)({
             commitCommand: (input) => {
@@ -784,6 +788,7 @@ it.effect(
           "status" in event.payload ? event.payload.status : null,
         ]),
         [
+          ["run.updated", "queued"],
           ["run.updated", "cancelled"],
           ["run-attempt.updated", "cancelled"],
           ["node.updated", "cancelled"],
@@ -797,6 +802,13 @@ it.effect(
           ["provider-session.updated", "stopped"],
         ],
       );
+      // Only the hold this restart added is recorded on the run it cut.
+      const cutRun = events.find(
+        (event) => event.type === "run.updated" && event.payload.status === "cancelled",
+      );
+      assert.deepEqual(cutRun?.type === "run.updated" ? cutRun.payload.restartHeldRunIds : null, [
+        RunId.make("run_recovery_queued"),
+      ]);
       const messageEvent = events.find((event) => event.type === "message.updated");
       assert.isFalse(messageEvent?.type === "message.updated" && messageEvent.payload.streaming);
       const stoppedSession = events.find((event) => event.type === "provider-session.updated");

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

@@ -5,6 +5,7 @@ import {
   isProviderNativeSubagentThread,
   type OrchestrationV2DomainEvent,
   type ProviderThreadId,
+  type RunId,
   type OrchestrationV2RestartCancelledBackgroundWork,
   type OrchestrationV2Subagent,
   type OrchestrationV2ThreadProjection,
@@ -421,12 +422,14 @@ export const make = Effect.gen(function* () {
         ...retryableRuns,
         ...queuedRunsInDeliveryOrder({ runs: queuedRuns, messages: queuedMessages }),
       ];
+      const restartHeldRunIds: Array<RunId> = [];
       for (const [index, run] of heldRuns.entries()) {
         if (bufferedMessageIds.has(run.userMessageId)) {
           runs.push(run);
           continue;
         }
         if (run.queueHeld === true && retryableRuns.length === 0) continue;
+        if (run.queueHeld !== true) restartHeldRunIds.push(run.id);
         events.push({
           id: yield* allocateEventId(),
           type: "run.updated",
@@ -536,7 +539,13 @@ export const make = Effect.gen(function* () {
           runId: run.id,
           providerInstanceId: run.providerInstanceId,
           occurredAt: now,
-          payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now },
+          payload: {
+            ...run,
+            status: "cancelled",
+            queuePosition: null,
+            completedAt: now,
+            ...(restartHeldRunIds.length > 0 ? { restartHeldRunIds } : {}),
+          },
         });
         for (const attempt of projection.attempts.filter(
           (candidate) =>

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

@@ -1077,14 +1077,29 @@ it.effect("does not continue a run the user asked to stop before the restart", (
   }),
 );
 
-it.effect("does not continue a cut /compact or /logout turn", () =>
+it.effect("compacts again after a cut /compact turn, and does not continue a cut /logout", () =>
   Effect.gen(function* () {
-    for (const text of ["/compact", " /LOGOUT "]) {
-      const texts = yield* continuationTexts({
+    for (const [text, creationSource, expected] of [
+      [" /Compact ", "mcp", ["user:/compact"]],
+      // Servers before `idleCompaction` marked their own compaction this way.
+      ["/compact", "server", ["agent:/compact"]],
+      [" /LOGOUT ", "mcp", []],
+    ] as const) {
+      const sent = yield* continuationDispatches({
         ...cutMidTurn(),
-        messages: [{ id: MessageId.make("message:user"), text, attachments: [] }],
+        messages: [
+          {
+            id: MessageId.make("message:user"),
+            role: "user",
+            createdBy: "agent",
+            creationSource,
+            text,
+            attachments: [],
+          },
+        ],
       } as unknown as OrchestrationV2ThreadProjection);
-      assert.deepEqual(texts, [], text);
+      const texts = sent.map((command) => `${command.createdBy}:${command.text}`);
+      assert.deepEqual(texts, [...expected], text);
     }
     const texts = yield* continuationTexts({
       ...cutMidTurn(),

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

@@ -2,6 +2,9 @@ import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
 import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
 import {
   CommandId,
+  IDLE_COMPACTION_TEXT,
+  idleCompactionOf,
+  isCompactionCommandMessage,
   isOrchestrationV2WorkActive,
   latestProviderTurnForAttempt,
   MessageId,
@@ -290,7 +293,13 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta
     const sourceMessage = sourceRecords.messages.find(
       (message) => message.id === source.userMessageId,
     );
-    if (sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) return;
+    // Compaction changes nothing a repeat could double, so a cut one runs again.
+    const recompact =
+      sourceMessage !== undefined &&
+      isCompactionCommandMessage(sourceMessage) &&
+      !isRestartNoteSource(source, projection.providerTurns);
+    if (!recompact && sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage))
+      return;
     const note = restartContinuationNote(
       source,
       projection.runs,
@@ -311,19 +320,27 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta
         commandId: CommandId.make(`command:restart-continuation:${input.sourceRunId}`),
         threadId: input.threadId,
         messageId,
-        text,
-        notification: {
-          source: { kind: "system" },
-          outcome: "updated",
-          summary: note.settled
-            ? "T3 Code restarted and stopped background work"
-            : "T3 Code restarted and resumed this turn",
-          detail: text,
-        },
+        text: recompact ? IDLE_COMPACTION_TEXT : text,
+        idleCompaction: recompact ? sourceMessage.idleCompaction : undefined,
+        notification: recompact
+          ? undefined
+          : {
+              source: { kind: "system" },
+              outcome: "updated",
+              summary: note.settled
+                ? "T3 Code restarted and stopped background work"
+                : "T3 Code restarted and resumed this turn",
+              detail: text,
+            },
         attachments: [],
         modelSelection: source.modelSelection,
         dispatchMode: { type: "start_immediately" },
-        createdBy: "agent",
+        // Agent + server with no idleCompaction reads as an automatic idle compaction.
+        createdBy: !recompact
+          ? "agent"
+          : idleCompactionOf(sourceMessage) === undefined
+            ? "user"
+            : sourceMessage.createdBy,
         creationSource: "server",
         restartContinuationOfRunId: input.sourceRunId,
       })

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

@@ -5597,7 +5597,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
         });
       const end = (
         key: string,
-        status: "completed" | "failed",
+        status: "completed" | "failed" | "cancelled",
         revision = "",
         patch: Partial<OrchestrationV2Run> = {},
       ) =>
@@ -5747,6 +5747,31 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
     }),
   );
 
+  it.effect("a compaction resumed after a restart releases only what that restart held", () =>
+    Effect.gen(function* () {
+      const threadId = ThreadId.make("runtime-layer-held-queue-resumed-compaction");
+      const { send, runFor, end, startedRunIds, awaitHeld } = yield* queueTestThread(threadId);
+      yield* send("failed", "agent", true);
+      yield* send("earlier", "user", false);
+      yield* send("later", "user", false);
+      yield* end("failed", "failed");
+      yield* awaitHeld("later");
+      yield* send("cut", "agent", true);
+      yield* end("cut", "cancelled", "", { restartHeldRunIds: [(yield* runFor("later")).id] });
+      // A second restart cuts the first continuation, which adds no holds of its own.
+      yield* send(" /compact", "agent", true);
+      yield* end(" /compact", "cancelled", "", {
+        restartContinuationOfRunId: (yield* runFor("cut")).id,
+      });
+      yield* send("/compact", "agent", true);
+      yield* end("/compact", "completed", "", {
+        restartContinuationOfRunId: (yield* runFor(" /compact")).id,
+      });
+      assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("later")).id);
+      assert.isTrue((yield* runFor("earlier")).queueHeld);
+    }),
+  );
+
   it.effect("keeps held agent messages while the thread's worktree setup is unresolved", () =>
     Effect.gen(function* () {
       const orchestrator = yield* Orchestrator.OrchestratorV2;

docs/internals/server-updates.md

@@ -87,8 +87,9 @@ cannot be overwritten by a stale cancellation.
 
 The [continuation handler](../../apps/server/src/orchestration-v2/RestartContinuation.ts)
 rechecks the preference, provider selection, newer user work, a stop the user requested,
-and maintenance turns such as `/compact` before dispatching. Intent and delivery both
-refuse a thread that is archived, settled, or still snoozed, a cut turn whose provider
+and maintenance turns such as `/logout` before dispatching. A cut `/compact` runs again, and
+its completion releases the holds that restart added, since they waited on it, not a person.
+Intent and delivery both refuse a thread that is archived, settled, or still snoozed, a cut turn whose provider
 turn already ended, and a run waiting on the user: an approval, question, or secret that was pending,
 or that recovery closed on the cut turn. Message-capable questions stay answerable, so
 they do not block. Delivery also stops after three consecutive continuations in one

docs/user/composer.md

@@ -261,6 +261,7 @@ If you stop the agent or the server restarts, queued messages keep their order a
 Press **Resume** in an empty composer to continue. **Resume** also continues a stopped
 run or one that hit a usage limit, even with nothing queued. Sending a new message also resumes the queue, unless it steers a turn that is already
 running. You can edit or remove held messages without starting them.
+A `/compact` cut by a restart runs again, and messages queued behind it send when it finishes.
 
 The pencil on a queued message opens it in the composer for editing. The original message
 stays in the queue until you save, and it is highlighted while you edit. The message's

packages/contracts/src/orchestrationV2.ts

@@ -706,6 +706,8 @@ export const OrchestrationV2Run = Schema.Struct({
   restartCancelledBackgroundWork: Schema.optional(
     Schema.Array(OrchestrationV2RestartCancelledBackgroundWork),
   ),
+  /** Queued runs a restart held when it cut this run, as opposed to holds already in place. */
+  restartHeldRunIds: Schema.optional(Schema.Array(RunId)),
   sourcePlanRef: Schema.optional(
     Schema.Struct({
       threadId: ThreadId,