perf(server): idle compaction reads only the latest run, live work and newest usage

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

@@ -50,6 +50,8 @@ export const make = Effect.gen(function* () {
           )
         )
           return;
+        const activeProviderThreadIds =
+          thread.activeProviderThreadId === null ? [] : [thread.activeProviderThreadId];
         const records = yield* projections.getThreadRecords(
           threadId,
           [
@@ -61,7 +63,20 @@ export const make = Effect.gen(function* () {
             "turnItems",
             "contextTransfers",
           ],
-          { messageRunIds: [runId], turnItemStatuses: ["pending", "running", "waiting"] },
+          {
+            messageRunIds: [runId],
+            turnItemStatuses: ["pending", "running", "waiting"],
+            // The latest run, active runs that hold background work, and rolled-back runs whose
+            // items it ignores.
+            runStatuses: ["preparing", "starting", "running", "rolled_back"],
+            includeLatestRun: true,
+            // Without an active provider thread there is no session to compact.
+            providerThreadIds: activeProviderThreadIds,
+            providerTurnThreadIds: activeProviderThreadIds,
+            latestProviderTurnUsageOnly: true,
+            runtimeRequestStatuses: ["pending"],
+            contextTransferStatuses: ["pending"],
+          },
         );
         if (
           resolveIdleCompaction(current, records.thread.providerInstanceId) !==

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

@@ -4,6 +4,7 @@ import {
   EventId,
   CommandId,
   CheckpointId,
+  ContextTransferId,
   CheckpointRef,
   CheckpointScopeId,
   MessageId,
@@ -307,6 +308,169 @@ it.effect("record reads restrict provider histories to delivery identities", ()
   }).pipe(Effect.provide(layerTest)),
 );
 
+it.effect("idle compaction filters keep the latest run, live work and the newest usage", () =>
+  Effect.gen(function* () {
+    const store = yield* ProjectionStore.ProjectionStoreV2;
+    // Its rolled-back run 1 still has a running item, which background work must ignore.
+    const threadId = yield* addRolledBackRecoveryCandidate("idle-compaction");
+    const now = yield* DateTime.now;
+    const minutes = (count: number) => DateTime.add(now, { minutes: count });
+    const selected = ProviderThreadId.make("provider-thread:idle-selected");
+    const nodeId = NodeId.make("node:idle-compaction");
+    for (const [ordinal, status] of [
+      [2, "completed"],
+      [3, "completed"],
+      [4, "failed"],
+    ] as const) {
+      const runId = RunId.make(`run:idle-compaction:${ordinal}`);
+      yield* store.apply({
+        id: EventId.make(`event:idle-compaction:run-${ordinal}`),
+        type: "run.created",
+        threadId,
+        runId,
+        nodeId,
+        driver,
+        providerInstanceId,
+        occurredAt: now,
+        payload: {
+          id: runId,
+          threadId,
+          ordinal,
+          providerInstanceId,
+          modelSelection,
+          providerThreadId: selected,
+          userMessageId: MessageId.make(`message:idle-compaction:${ordinal}`),
+          rootNodeId: nodeId,
+          activeAttemptId: null,
+          status,
+          requestedAt: now,
+          startedAt: now,
+          completedAt: now,
+          checkpointId: null,
+          contextHandoffId: null,
+        },
+      });
+    }
+    for (const [id, providerThreadId, ordinal, usageAt] of [
+      ["turn:idle-old", selected, 1, minutes(1)],
+      ["turn:idle-newest", selected, 2, minutes(3)],
+      ["turn:idle-later-ordinal", selected, 3, minutes(2)],
+      ["turn:idle-no-usage", selected, 4, null],
+      [
+        "turn:idle-other-thread",
+        ProviderThreadId.make("provider-thread:idle-other"),
+        1,
+        minutes(9),
+      ],
+    ] as const) {
+      yield* store.apply({
+        id: EventId.make(`event:${id}`),
+        type: "provider-turn.updated",
+        threadId,
+        nodeId,
+        driver,
+        occurredAt: now,
+        payload: {
+          id: ProviderTurnId.make(id),
+          providerThreadId,
+          nodeId,
+          runAttemptId: null,
+          nativeTurnRef: null,
+          ordinal,
+          status: "completed",
+          startedAt: now,
+          completedAt: now,
+          ...(usageAt === null
+            ? {}
+            : { tokenUsage: { usedTokens: 150_000, updatedAt: DateTime.formatIso(usageAt) } }),
+        },
+      });
+    }
+    for (const status of ["pending", "resolved"] as const) {
+      yield* store.apply({
+        id: EventId.make(`event:idle-compaction:request-${status}`),
+        type: "runtime-request.updated",
+        threadId,
+        nodeId,
+        occurredAt: now,
+        payload: {
+          id: RuntimeRequestId.make(`request:idle-compaction:${status}`),
+          nodeId,
+          providerTurnId: null,
+          nativeRequestRef: null,
+          kind: "user_input",
+          status,
+          responseCapability: { type: "message" },
+          createdAt: now,
+          resolvedAt: status === "pending" ? null : now,
+        },
+      });
+    }
+    for (const status of ["pending", "consumed"] as const) {
+      yield* store.apply({
+        id: EventId.make(`event:idle-compaction:transfer-${status}`),
+        type: "context-transfer.created",
+        threadId,
+        occurredAt: now,
+        payload: {
+          id: ContextTransferId.make(`transfer:idle-compaction:${status}`),
+          type: "merge_back",
+          sourceThreadId: ThreadId.make("thread:idle-compaction:side"),
+          targetThreadId: threadId,
+          sourcePoint: { threadId: ThreadId.make("thread:idle-compaction:side") },
+          basePoint: null,
+          sourceProviderInstanceId: providerInstanceId,
+          targetProviderInstanceId: providerInstanceId,
+          targetRunId: null,
+          status,
+          resolution: null,
+          createdBy: "user",
+          error: null,
+          createdAt: now,
+          updatedAt: now,
+          consumedAt: status === "pending" ? null : now,
+        },
+      });
+    }
+
+    const records = yield* store.getThreadRecords(
+      threadId,
+      ["runs", "providerTurns", "runtimeRequests", "contextTransfers"],
+      {
+        runStatuses: ["preparing", "starting", "running", "rolled_back"],
+        includeLatestRun: true,
+        providerTurnThreadIds: [selected],
+        latestProviderTurnUsageOnly: true,
+        runtimeRequestStatuses: ["pending"],
+        contextTransferStatuses: ["pending"],
+      },
+    );
+    // The failed latest run stays, so a completed earlier run is not mistaken for the latest.
+    assert.deepEqual(
+      records.runs.map((run) => run.id),
+      ["run:idle-compaction:rolled-back", "run:idle-compaction:4"],
+    );
+    assert.deepEqual(
+      records.providerTurns.map((turn) => turn.id),
+      ["turn:idle-newest"],
+    );
+    assert.deepEqual(
+      records.runtimeRequests.map((request) => request.id),
+      ["request:idle-compaction:pending"],
+    );
+    assert.deepEqual(
+      records.contextTransfers.map((transfer) => transfer.id),
+      ["transfer:idle-compaction:pending"],
+    );
+    const empty = yield* store.getThreadRecords(threadId, ["runs", "providerTurns"], {
+      runStatuses: [],
+      providerTurnThreadIds: [],
+      latestProviderTurnUsageOnly: true,
+    });
+    assert.deepEqual([empty.runs, empty.providerTurns], [[], []]);
+  }).pipe(Effect.provide(layerTest)),
+);
+
 it.effect("record reads filter notification sources without dropping matching message bodies", () =>
   Effect.gen(function* () {
     const store = yield* ProjectionStore.ProjectionStoreV2;

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

@@ -25,13 +25,15 @@ import type {
   OrchestrationV2PendingRuntimeRequestSummary,
   PlanId,
   OrchestrationV2ConversationMessage,
+  OrchestrationV2ContextTransfer,
   OrchestrationV2DomainEvent,
   OrchestrationV2OlderSettledCount,
   OrchestrationV2OlderSettledCursor,
   OrchestrationV2ProjectedTurnItem,
   OrchestrationV2ProviderThread,
   OrchestrationV2ProviderTurn,
   OrchestrationV2Run,
+  OrchestrationV2RuntimeRequest,
   OrchestrationV2Subagent,
   OrchestrationV2ThreadShellSnapshot,
   OrchestrationV2ShellThreadStatus,
@@ -339,6 +341,14 @@ export interface ProjectionRecordFilter {
   readonly turnItemRunIds?: ReadonlyArray<RunId | null>;
   readonly turnItemNodeIds?: ReadonlyArray<NodeId>;
   readonly runIds?: ReadonlyArray<RunId>;
+  readonly runStatuses?: ReadonlyArray<OrchestrationV2Run["status"]>;
+  /** Keeps the thread's highest-ordinal run even when `runStatuses` excludes it. */
+  readonly includeLatestRun?: true;
+  readonly providerTurnThreadIds?: ReadonlyArray<ProviderThreadId>;
+  /** Only the provider turn whose token usage was updated last. */
+  readonly latestProviderTurnUsageOnly?: true;
+  readonly runtimeRequestStatuses?: ReadonlyArray<OrchestrationV2RuntimeRequest["status"]>;
+  readonly contextTransferStatuses?: ReadonlyArray<OrchestrationV2ContextTransfer["status"]>;
   readonly subagentIds?: ReadonlyArray<NodeId>;
   readonly subagentRunIds?: ReadonlyArray<RunId>;
   readonly nodeIds?: ReadonlyArray<NodeId>;
@@ -3244,14 +3254,27 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         ] = yield* Effect.all([
           decodeThreadPayload(threadRow.payload_json),
           (fields !== undefined && !fields.includes("runs")) ||
-          (window === undefined && matchesNothing(filter?.runIds))
+          (window === undefined &&
+            (matchesNothing(filter?.runIds) ||
+              (matchesNothing(filter?.runStatuses) && filter?.includeLatestRun !== true)))
             ? Effect.succeed([])
             : window === undefined
               ? sql<PayloadRow>`
             SELECT payload_json
             FROM orchestration_v2_projection_runs
             WHERE thread_id = ${threadId}
               ${filter?.runIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.runIds)}))`}
+              ${
+                filter?.runStatuses === undefined
+                  ? sql``
+                  : sql`AND (status IN (SELECT value FROM json_each(${encodeIdList(filter.runStatuses)}))
+                    OR (${filter.includeLatestRun === true ? 1 : 0} = 1 AND run_id = (
+                      SELECT run_id FROM orchestration_v2_projection_runs
+                      WHERE thread_id = ${threadId}
+                      ORDER BY ordinal DESC
+                      LIMIT 1
+                    )))`
+              }
             ORDER BY ordinal ASC
           `
               : sql<PayloadRow>`
@@ -3421,15 +3444,24 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
             ORDER BY COALESCE(first_run_ordinal, 0), provider_thread_id ASC
           `,
           (fields !== undefined && !fields.includes("providerTurns")) ||
-          (window === undefined && matchesNothing(filter?.providerTurnAttemptIds))
+          (window === undefined &&
+            matchesNothing(filter?.providerTurnAttemptIds, filter?.providerTurnThreadIds))
             ? Effect.succeed([])
             : window === undefined
               ? sql<PayloadRow>`
             SELECT payload_json
             FROM orchestration_v2_projection_provider_turns
             WHERE thread_id = ${threadId}
               ${filter?.providerTurnAttemptIds === undefined ? sql`` : sql`AND run_attempt_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnAttemptIds)}))`}
-            ORDER BY provider_thread_id ASC, ordinal ASC
+              ${filter?.providerTurnThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnThreadIds)}))`}
+              ${
+                filter?.latestProviderTurnUsageOnly === true
+                  ? sql`AND json_extract(payload_json, '$.tokenUsage') IS NOT NULL
+            ORDER BY julianday(json_extract(payload_json, '$.tokenUsage.updatedAt')) DESC,
+              provider_thread_id ASC, ordinal ASC
+            LIMIT 1`
+                  : sql`ORDER BY provider_thread_id ASC, ordinal ASC`
+              }
           `
               : sql<PayloadRow>`
             SELECT payload_json FROM orchestration_v2_projection_provider_turns
@@ -3439,13 +3471,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 OR node_id IN (SELECT value FROM json_each(${cohortNodeIds})))
             ORDER BY provider_thread_id ASC, ordinal ASC
           `,
-          fields !== undefined && !fields.includes("runtimeRequests")
+          (fields !== undefined && !fields.includes("runtimeRequests")) ||
+          (window === undefined && matchesNothing(filter?.runtimeRequestStatuses))
             ? Effect.succeed([])
             : window === undefined
               ? sql<PayloadRow>`
             SELECT payload_json
             FROM orchestration_v2_projection_runtime_requests
             WHERE thread_id = ${threadId}
+              ${filter?.runtimeRequestStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.runtimeRequestStatuses)}))`}
             ORDER BY created_at ASC, runtime_request_id ASC
           `
               : sql<PayloadRow>`
@@ -3560,13 +3594,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                   )
                 ORDER BY rowid ASC
               `,
-          fields !== undefined && !fields.includes("contextTransfers")
+          (fields !== undefined && !fields.includes("contextTransfers")) ||
+          (window === undefined && matchesNothing(filter?.contextTransferStatuses))
             ? Effect.succeed([])
             : window === undefined
               ? sql<PayloadRow>`
             SELECT payload_json
             FROM orchestration_v2_projection_context_transfers
-            WHERE source_thread_id = ${threadId} OR target_thread_id = ${threadId}
+            WHERE (source_thread_id = ${threadId} OR target_thread_id = ${threadId})
+              ${filter?.contextTransferStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)}))`}
             ORDER BY rowid ASC
           `
               : sql<PayloadRow>`