fix(server): idle compaction re-arms after a server restart

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

@@ -3,6 +3,7 @@ import {
   DEFAULT_SERVER_SETTINGS,
   EventId,
   MessageId,
+  type OrchestrationV2DomainEvent,
   type OrchestrationV2ThreadProjection,
   NodeId,
   ProjectId,
@@ -16,12 +17,15 @@ import {
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
 import * as Stream from "effect/Stream";
 import * as Tracer from "effect/Tracer";
 import * as TestClock from "effect/testing/TestClock";
 
 import * as ServerSettings from "../serverSettings.ts";
+import * as EventSink from "./EventSink.ts";
 import * as IdleCompactionService from "./IdleCompactionService.ts";
 import * as Orchestrator from "./Orchestrator.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
@@ -192,6 +196,12 @@ describe("idleCompactionWindow", () => {
   });
 });
 
+const eventSinkStreaming = (events: Stream.Stream<OrchestrationV2DomainEvent>) =>
+  Layer.mock(EventSink.EventSinkV2)({
+    latestSequence: () => Effect.succeed(0),
+    stream: () => events.pipe(Stream.map((event) => ({ sequence: 1, commandId: null, event }))),
+  });
+
 describe("IdleCompactionService", () => {
   it.effect.each([
     { mode: "onReturn", reads: 0 },
@@ -266,15 +276,17 @@ describe("IdleCompactionService", () => {
             },
           });
           const dependencies = Layer.mergeAll(
-            Layer.mock(Orchestrator.OrchestratorV2)({
-              streamDomainEvents: Stream.make({
+            eventSinkStreaming(
+              Stream.make({
                 id: EventId.make("idle-completed"),
                 type: "run.updated",
                 threadId,
                 runId,
                 occurredAt: at(),
                 payload: data.runs[0]!,
               }),
+            ),
+            Layer.mock(Orchestrator.OrchestratorV2)({
               dispatch: () =>
                 Effect.sync(() => {
                   dispatches++;
@@ -288,6 +300,7 @@ describe("IdleCompactionService", () => {
                   historyReads++;
                   return data;
                 }),
+              listIdleCompactionCandidates: () => Effect.succeed([]),
             }),
             Layer.mock(ServerSettings.ServerSettingsService)({
               getSettings: Effect.succeed({
@@ -377,11 +390,13 @@ describe("IdleCompactionService", () => {
         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(
+          // A delegated completion on the earlier run lands after the later run completed.
+          eventSinkStreaming(
+            Stream.make(completed(laterRun), completed(earlierRun)).pipe(
               Stream.ensuring(Deferred.succeed(streamDone, undefined)),
             ),
+          ),
+          Layer.mock(Orchestrator.OrchestratorV2)({
             dispatch: (command) =>
               Effect.sync(() => {
                 dispatched.push(command.commandId);
@@ -399,6 +414,7 @@ describe("IdleCompactionService", () => {
                     (message.runId !== null && options.messageRunIds.includes(message.runId)),
                 ),
               }),
+            listIdleCompactionCandidates: () => Effect.succeed([]),
           }),
           Layer.mock(ServerSettings.ServerSettingsService)({
             getSettings: Effect.succeed({
@@ -417,4 +433,64 @@ describe("IdleCompactionService", () => {
       }),
     ),
   );
+
+  it.effect("compacts a thread that went idle before the server restarted", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        yield* TestClock.setTime(NOW_MS);
+        const base = records({ cacheExpiresInMs: 30 * MINUTE_MS });
+        const data: OrchestrationV2ThreadProjection = {
+          ...base,
+          attempts: [],
+          nodes: [],
+          subagents: [],
+          providerSessions: [],
+          plans: [],
+          checkpointScopes: [],
+          checkpoints: [],
+          contextHandoffs: [],
+          visibleTurnItems: [],
+          updatedAt: at(),
+        };
+        const dispatched = yield* Deferred.make<ReadonlyArray<string>>();
+        const commandIds: Array<string> = [];
+        const dependencies = Layer.mergeAll(
+          // The run completed before the restart, so no run.updated event arrives.
+          eventSinkStreaming(Stream.never),
+          Layer.mock(Orchestrator.OrchestratorV2)({
+            dispatch: (command) =>
+              Effect.sync(() => {
+                commandIds.push(command.commandId);
+                Deferred.doneUnsafe(dispatched, Effect.succeed(commandIds));
+                return { sequence: 1, storedEvents: [] };
+              }),
+          }),
+          Layer.mock(ProjectionStore.ProjectionStoreV2)({
+            getThread: () => Effect.succeed(data.thread),
+            getThreadRecords: () => Effect.succeed(data),
+            listIdleCompactionCandidates: () => Effect.succeed([{ threadId, runId }]),
+          }),
+          Layer.mock(ServerSettings.ServerSettingsService)({
+            getSettings: Effect.succeed({
+              ...DEFAULT_SERVER_SETTINGS,
+              idleCompaction: "beforeCacheExpires",
+            }),
+          }),
+        );
+        const result = yield* Effect.gen(function* () {
+          const service = yield* IdleCompactionService.make;
+          yield* service.start();
+          const waiter = yield* Deferred.await(dispatched).pipe(
+            Effect.timeoutOption("1 hour"),
+            Effect.forkChild,
+          );
+          yield* TestClock.adjust("1 hour");
+          return yield* Fiber.join(waiter);
+        }).pipe(Effect.provide(dependencies));
+        expect(Option.getOrElse(result, () => [])).toEqual([
+          `server:idle-compact:${threadId}:${runId}`,
+        ]);
+      }),
+    ),
+  );
 });

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

@@ -2,6 +2,7 @@ import { isListedThreadLineage, type OrchestrationV2ThreadProjection } from "@t3
 import { COMPACTION_SUGGESTION_MIN_TOKENS } from "@t3tools/shared/claudeCompaction";
 import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
 import * as DateTime from "effect/DateTime";
+import * as Duration from "effect/Duration";
 
 // Claude Code's idleCompaction defaults to 200K and won't go below 100K. Below
 // that a cold re-read is cheap, so compacting mostly loses context for little
@@ -10,6 +11,8 @@ const MIN_CONTEXT_TOKENS = COMPACTION_SUGGESTION_MIN_TOKENS;
 // Start this long before the cache expires: enough for the session to resume
 // and send the compaction while the cache is still warm.
 const LEAD_MS = 2 * 60_000;
+// Claude's longest prompt cache TTL: a run that completed earlier has no warm cache to keep.
+export const MAX_PROMPT_CACHE_TTL = Duration.hours(1);
 
 export type IdleCompactionRecords = Pick<
   OrchestrationV2ThreadProjection,

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

@@ -15,19 +15,20 @@ import * as Stream from "effect/Stream";
 
 import { forkParked } from "../serverActivation.ts";
 import * as ServerSettings from "../serverSettings.ts";
-import { idleCompactionWindow } from "./IdleCompaction.ts";
+import { idleCompactionWindow, MAX_PROMPT_CACHE_TTL } from "./IdleCompaction.ts";
+import * as EventSink from "./EventSink.ts";
 import * as Orchestrator from "./Orchestrator.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 
 /**
  * Beta: when a run completes on a Claude instance set to compact before the
  * cache expires, waits until shortly before its prompt cache expires and sends
- * `/compact` as a visible T3 Code turn. Timers live in
- * memory, so a server restart skips compactions that were still waiting.
+ * `/compact` as a visible T3 Code turn. On start it re-arms runs that completed before a restart.
  */
 export const make = Effect.gen(function* () {
   const orchestrator = yield* Orchestrator.OrchestratorV2;
   const projections = yield* ProjectionStore.ProjectionStoreV2;
+  const eventSink = yield* EventSink.EventSinkV2;
   const settings = yield* ServerSettings.ServerSettingsService;
 
   const compactWhenDue = Effect.fn("IdleCompactionService.compactWhenDue")(
@@ -131,11 +132,41 @@ export const make = Effect.gen(function* () {
   const start = Effect.fn("IdleCompactionService.start")(function* () {
     // Keyed by thread: a newer completed run replaces the waiting timer.
     const timers = yield* FiberMap.make<ThreadId>();
+    // Events after this sequence come from the stream; earlier ones are already in the projections the scan reads.
+    const afterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie);
+    const rearm = Effect.fn("IdleCompactionService.rearm")(
+      function* () {
+        const completedAfter = DateTime.subtractDuration(yield* DateTime.now, MAX_PROMPT_CACHE_TTL);
+        const candidates = yield* projections.listIdleCompactionCandidates({ completedAfter });
+        yield* Effect.forEach(
+          candidates,
+          ({ threadId, runId }) =>
+            // A timer armed by a live event is newer than the scan.
+            FiberMap.run(timers, threadId, compactWhenDue(threadId, runId), {
+              onlyIfMissing: true,
+            }),
+          { discard: true },
+        );
+      },
+      Effect.catch((cause) => Effect.logWarning("idle compaction rearm skipped", { cause })),
+    );
     yield* forkParked(
-      Stream.runForEach(orchestrator.streamDomainEvents, (event) =>
-        event.type === "run.updated" && event.payload.status === "completed"
-          ? FiberMap.run(timers, event.threadId, compactWhenDue(event.threadId, event.payload.id))
-          : Effect.void,
+      Effect.all(
+        [
+          Stream.runForEach(
+            eventSink.stream({ afterSequence, eventType: "run.updated" }),
+            ({ event }) =>
+              event.type === "run.updated" && event.payload.status === "completed"
+                ? FiberMap.run(
+                    timers,
+                    event.threadId,
+                    compactWhenDue(event.threadId, event.payload.id),
+                  )
+                : Effect.void,
+          ),
+          rearm(),
+        ],
+        { concurrency: "unbounded", discard: true },
       ),
     );
   });

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

@@ -2519,6 +2519,69 @@ it.effect("lists only active subagent IDs without loading shell data", () =>
   }).pipe(Effect.provide(layerTest)),
 );
 
+it.effect("lists threads whose latest run completed after the cutoff", () =>
+  Effect.gen(function* () {
+    const store = yield* ProjectionStore.ProjectionStoreV2;
+    const sql = yield* SqlClient.SqlClient;
+    const recent = "2026-01-05T12:30:00.000Z";
+    const old = "2026-01-05T11:00:00.000Z";
+    for (const [id, archivedAt, runs] of [
+      ["recent", null, [["completed", recent]]],
+      ["old", null, [["completed", old]]],
+      [
+        "running-after-completed",
+        null,
+        [
+          ["completed", recent],
+          ["running", null],
+        ],
+      ],
+      [
+        "cancelled-before-start-after-completed",
+        null,
+        [
+          ["completed", recent],
+          ["cancelled", null],
+        ],
+      ],
+      ["archived", recent, [["completed", recent]]],
+    ] as const) {
+      yield* sql`
+        INSERT INTO orchestration_v2_projection_threads (
+          thread_id, project_id, title, default_provider, runtime_mode, interaction_mode,
+          created_at, updated_at, archived_at, deleted_at, payload_json
+        ) VALUES (
+          ${id}, 'project:test', ${id}, 'codex', 'full-access', 'default',
+          ${old}, ${old}, ${archivedAt}, NULL, '{}'
+        )
+      `;
+      for (const [index, [status, completedAt]] of runs.entries()) {
+        yield* sql`
+          INSERT INTO orchestration_v2_projection_runs (
+            run_id, thread_id, ordinal, provider, status, requested_at, completed_at, payload_json
+          ) VALUES (
+            ${`${id}:run-${index + 1}`}, ${id}, ${index + 1}, 'claudeAgent', ${status},
+            ${old}, ${completedAt}, '{}'
+          )
+        `;
+      }
+    }
+
+    assert.deepEqual(
+      yield* store.listIdleCompactionCandidates({
+        completedAfter: DateTime.makeUnsafe("2026-01-05T12:00:00.000Z"),
+      }),
+      [
+        { threadId: ThreadId.make("recent"), runId: RunId.make("recent:run-1") },
+        {
+          threadId: ThreadId.make("cancelled-before-start-after-completed"),
+          runId: RunId.make("cancelled-before-start-after-completed:run-1"),
+        },
+      ],
+    );
+  }).pipe(Effect.provide(layerTest)),
+);
+
 // regression: #25621 SQLite rejects statements binding more than 32,766 variables
 it.effect("lists every thread in the shell snapshot past SQLite's bound variable limit", () =>
   Effect.gen(function* () {

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

@@ -517,6 +517,13 @@ export interface ProjectionStoreV2Shape {
     ReadonlyArray<ThreadId>,
     ProjectionStoreV2Error
   >;
+  /** Active threads whose latest run completed at or after the cutoff. */
+  readonly listIdleCompactionCandidates: (query: {
+    readonly completedAfter: DateTime.Utc;
+  }) => Effect.Effect<
+    ReadonlyArray<{ readonly threadId: ThreadId; readonly runId: RunId }>,
+    ProjectionStoreV2Error
+  >;
   /** Threads that are not deleted, archived ones included. */
   readonly countThreads: () => Effect.Effect<number, ProjectionStoreV2Error>;
   /** Every active thread, in shell order. */
@@ -7333,6 +7340,36 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       Effect.mapError(shellSnapshotReadError),
     );
 
+    const listIdleCompactionCandidates = Effect.fn("ProjectionStore.listIdleCompactionCandidates")(
+      function* ({ completedAfter }: { readonly completedAfter: DateTime.Utc }) {
+        const rows = yield* sql<{ readonly thread_id: string; readonly run_id: string }>`
+        SELECT run.thread_id, run.run_id
+        FROM orchestration_v2_projection_threads thread
+        -- CROSS JOIN keeps threads outermost; the planner otherwise scans every run row.
+        CROSS JOIN orchestration_v2_projection_runs run
+        WHERE thread.deleted_at IS NULL
+          AND thread.archived_at IS NULL
+          AND run.thread_id = thread.thread_id
+          AND run.ordinal = (
+            SELECT MAX(latest.ordinal) FROM orchestration_v2_projection_runs latest
+            WHERE latest.thread_id = thread.thread_id
+              -- A removed or steered queued message leaves a cancelled run that never started.
+              AND (
+                latest.status <> 'cancelled'
+                OR json_extract(latest.payload_json, '$.startedAt') IS NOT NULL
+              )
+          )
+          AND run.status = 'completed'
+          AND run.completed_at >= ${DateTime.formatIso(completedAfter)}
+      `;
+        return rows.map((row) => ({
+          threadId: ThreadId.make(row.thread_id),
+          runId: RunId.make(row.run_id),
+        }));
+      },
+      Effect.mapError(shellSnapshotReadError),
+    );
+
     const countThreads = Effect.fn("ProjectionStore.countThreads")(function* () {
       const rows = yield* sql<{ readonly count: number }>`
         SELECT COUNT(*) AS count FROM orchestration_v2_projection_threads WHERE deleted_at IS NULL
@@ -7615,6 +7652,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       getShellSnapshot: readOnReader(getShellSnapshot),
       getPullRequestSnapshot: readOnReader(getPullRequestSnapshot),
       listActiveSubagentThreadIds: readOnReader(listActiveSubagentThreadIds),
+      listIdleCompactionCandidates: readOnReader(listIdleCompactionCandidates),
       countThreads: readOnReader(countThreads),
       listActiveShellIndex: readOnReader(listActiveShellIndex),
       readShellSnapshot: readOnReader(readShellSnapshot),

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

@@ -215,6 +215,7 @@ it.effect(
           getAgentAwarenessSnapshot: () => Effect.die("unused awareness snapshot"),
           getShellSnapshot: () => Effect.die("unused getShellSnapshot"),
           listActiveSubagentThreadIds: () => Effect.die("unused listActiveSubagentThreadIds"),
+          listIdleCompactionCandidates: () => Effect.die("unused listIdleCompactionCandidates"),
           countThreads: () => Effect.die("unused countThreads"),
           listActiveShellIndex: () => Effect.die("unused listActiveShellIndex"),
           listOlderSettledThreads: () => Effect.die("unused listOlderSettledThreads"),

apps/server/src/server.ts

@@ -589,7 +589,7 @@ const layerThreadSettlementWorker = Layer.effectDiscard(
 
 const layerIdleCompactionWorker = Layer.effectDiscard(
   IdleCompactionService.make.pipe(Effect.flatMap((service) => service.start())),
-).pipe(Layer.provide(ProjectionStoreV2.layer));
+).pipe(Layer.provide(Layer.merge(ProjectionStoreV2.layer, RuntimeLayer.layerEventSink)));
 
 const layerThreadPullRequestWorker = Layer.effectDiscard(
   ThreadPullRequestService.make.pipe(Effect.flatMap((service) => service.start())),