fix(server): idle compaction survives a late update for an earlier run

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

@@ -308,4 +308,89 @@ describe("IdleCompactionService", () => {
       );
     },
   );
+
+  it.effect("keeps the latest run's compaction when an earlier run reports completed again", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        yield* TestClock.setTime(NOW_MS);
+        const base = records({ cacheExpiresInMs: 10 * MINUTE_MS });
+        const laterRunId = RunId.make("run-2");
+        const laterMessageId = MessageId.make("message-2");
+        const earlierRun = base.runs[0]!;
+        const laterRun = {
+          ...earlierRun,
+          id: laterRunId,
+          ordinal: 2,
+          userMessageId: laterMessageId,
+        };
+        const data: OrchestrationV2ThreadProjection = {
+          ...base,
+          runs: [earlierRun, laterRun],
+          messages: [
+            ...base.messages,
+            { ...base.messages[0]!, id: laterMessageId, runId: laterRunId, text: "Keep going" },
+          ],
+          providerThreads: [{ ...base.providerThreads[0]!, lastRunOrdinal: 2 }],
+          attempts: [],
+          nodes: [],
+          subagents: [],
+          providerSessions: [],
+          plans: [],
+          checkpointScopes: [],
+          checkpoints: [],
+          contextHandoffs: [],
+          visibleTurnItems: [],
+          updatedAt: at(),
+        };
+        const completed = (run: typeof earlierRun) => ({
+          id: EventId.make(`completed-${run.id}`),
+          type: "run.updated" as const,
+          threadId,
+          runId: run.id,
+          occurredAt: at(),
+          payload: run,
+        });
+        const dispatched: Array<string> = [];
+        const streamDone = yield* Deferred.make<void>();
+        const dependencies = Layer.mergeAll(
+          Layer.mock(Orchestrator.OrchestratorV2)({
+            // A delegated completion on the earlier run lands after the later run completed.
+            streamDomainEvents: Stream.make(completed(laterRun), completed(earlierRun)).pipe(
+              Stream.ensuring(Deferred.succeed(streamDone, undefined)),
+            ),
+            dispatch: (command) =>
+              Effect.sync(() => {
+                dispatched.push(command.commandId);
+                return { sequence: 1, storedEvents: [] };
+              }),
+          }),
+          Layer.mock(ProjectionStore.ProjectionStoreV2)({
+            getThread: () => Effect.succeed(data.thread),
+            getThreadRecords: (_threadId, _keys, options) =>
+              Effect.succeed({
+                ...data,
+                messages: data.messages.filter(
+                  (message) =>
+                    options?.messageRunIds === undefined ||
+                    (message.runId !== null && options.messageRunIds.includes(message.runId)),
+                ),
+              }),
+          }),
+          Layer.mock(ServerSettings.ServerSettingsService)({
+            getSettings: Effect.succeed({
+              ...DEFAULT_SERVER_SETTINGS,
+              idleCompaction: "beforeCacheExpires",
+            }),
+          }),
+        );
+        yield* Effect.gen(function* () {
+          const service = yield* IdleCompactionService.make;
+          yield* service.start();
+          yield* Deferred.await(streamDone);
+          yield* TestClock.adjust(9 * MINUTE_MS);
+        }).pipe(Effect.provide(dependencies));
+        expect(dispatched).toEqual(["server:idle-compact:idle-thread:run-2"]);
+      }),
+    ),
+  );
 });

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

@@ -31,8 +31,9 @@ export const make = Effect.gen(function* () {
   const settings = yield* ServerSettings.ServerSettingsService;
 
   const compactWhenDue = Effect.fn("IdleCompactionService.compactWhenDue")(
-    function* (threadId: ThreadId, runId: RunId) {
-      // Loops once after the wait: the thread may have moved on.
+    function* (threadId: ThreadId, completedRunId: RunId) {
+      let runId = completedRunId;
+      // Loops after the wait or a retarget: the thread may have moved on.
       while (true) {
         const current = yield* settings.getSettings;
         const thread = yield* projections.getThread(threadId);
@@ -72,6 +73,15 @@ export const make = Effect.gen(function* () {
           )
         )
           return;
+        const latestRun = records.runs.reduce<(typeof records.runs)[number] | undefined>(
+          (latest, run) => (latest === undefined || run.ordinal > latest.ordinal ? run : latest),
+          undefined,
+        );
+        // A late update for an earlier run replaced the latest run's timer, so take it over.
+        if (latestRun?.status === "completed" && latestRun.id !== runId) {
+          runId = latestRun.id;
+          continue;
+        }
         const nowMs = DateTime.toEpochMillis(yield* DateTime.now);
         const window = idleCompactionWindow(records, nowMs);
         if (window?.runId !== runId) return;