fix(runtime): deduplicate in-flight context compaction

packages/runtime/src/__tests__/ai-sdk-backend.test.ts

@@ -4690,6 +4690,93 @@ describe('AiSdkBackend model history', () => {
     assert.equal(result.contextBudget?.compactionDecisions?.[0]?.decision, 'replaced');
   });
 
+  test('coalesces identical in-flight compactHistory requests', async () => {
+    const recorded: HistoryCompactCheckpoint[] = [];
+    let summarizeCalls = 0;
+    let releaseSummary!: () => void;
+    const summaryReady = new Promise<void>((resolve) => {
+      releaseSummary = resolve;
+    });
+    const backend = createTestAiSdkBackend({
+      sessionId: 'session-1',
+      header: header(),
+      appendMessage: async () => {},
+      connection: connection(),
+      apiKey: 'sk-test',
+      modelId: 'mock-model-id',
+      modelFactory: () => completionModel(),
+      tools: [],
+      newId: idGenerator(),
+      now: monotonicClock(),
+      contextBudget: {
+        name: 'in-flight-dedup-test',
+        maxHistoryEstimatedTokens: 10_000,
+        charsPerToken: 1,
+      },
+      summarizeHistoryCompact: async () => {
+        summarizeCalls += 1;
+        await summaryReady;
+        return structuredSummary('IN_FLIGHT_DEDUP_SUMMARY');
+      },
+      recordHistoryCompactCheckpoint: (checkpoint) => {
+        recorded.push(checkpoint);
+      },
+    });
+    const runtimeContext = [
+      runtimeTextEvent({
+        id: 'dedup-old-user',
+        turnId: 'dedup-old-turn',
+        role: 'user',
+        author: 'user',
+        text: 'old context '.repeat(100),
+      }),
+      runtimeTextEvent({
+        id: 'dedup-old-agent',
+        turnId: 'dedup-old-turn',
+        role: 'model',
+        author: 'agent',
+        text: 'old response '.repeat(100),
+      }),
+    ];
+    const first = backend.compactHistory({
+      turnId: 'dedup-compact-1',
+      runId: 'run-dedup-1',
+      runtimeContext,
+    });
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    const second = backend.compactHistory({
+      turnId: 'dedup-compact-2',
+      runId: 'run-dedup-2',
+      runtimeContext: [
+        ...runtimeContext,
+        runtimeTextEvent({
+          id: 'dedup-current-turn',
+          turnId: 'dedup-compact-2',
+          role: 'user',
+          author: 'user',
+          text: 'current turn content must not affect the fold key',
+        }),
+      ],
+    });
+
+    releaseSummary();
+    const [firstResult, secondResult] = await Promise.all([first, second]);
+    assert.equal(summarizeCalls, 1);
+    assert.equal(secondResult.outcome.kind, 'compacted');
+    assert.equal(firstResult.outcome.kind, 'compacted');
+    assert.equal(recorded.length, 2);
+    const firstCheckpoint = recorded[0];
+    const secondCheckpoint = recorded[1];
+    assert.ok(firstCheckpoint);
+    assert.ok(secondCheckpoint);
+    assert.equal(firstCheckpoint.version, 2);
+    assert.equal(secondCheckpoint.version, 2);
+    if (firstCheckpoint.version === 2 && secondCheckpoint.version === 2) {
+      assert.equal(secondCheckpoint.summary, firstCheckpoint.summary);
+      assert.deepEqual(secondCheckpoint.coverage, firstCheckpoint.coverage);
+    }
+  });
+
   test('manual compactHistory compacts one completed turn with multiple agent steps', async () => {
     const recorded: HistoryCompactCheckpoint[] = [];
     const backend = createTestAiSdkBackend({

packages/runtime/src/ai-sdk-compaction.ts

@@ -215,7 +215,17 @@ export class AiSdkCompaction {
     providerReasoningReplayEventIds: ReadonlySet<string>,
   ) => Promise<ModelMessage[]>;
   private readonly canReplayProviderNative: (plan: RuntimeEventModelReplayPlan) => boolean;
-  private historyCompactAbortController: AbortController | null = null;
+  private readonly historyCompactAbortControllers = new Set<AbortController>();
+  /**
+   * Exact duplicate summary inputs share one physical provider call. The map
+   * lives at the summarizer boundary, where the existing effective-history
+   * fingerprint already describes the request that spends provider budget.
+   * Each caller still completes its own checkpoint/turn bookkeeping.
+   */
+  private readonly inFlightHistorySummaries = new Map<
+    string,
+    Promise<string | HistoryCompactProviderState | undefined>
+  >();
   /**
    * Session-scoped circuit for exact malformed compaction inputs. A retry or
    * regeneration on the same backend must not dispatch the same doomed call;
@@ -287,15 +297,22 @@ export class AiSdkCompaction {
 
   /** Abort an in-flight manual history compaction (called by AiSdkBackend.stop). */
   public abortHistoryCompact(): void {
-    this.historyCompactAbortController?.abort();
+    for (const controller of this.historyCompactAbortControllers) controller.abort();
   }
 
   public async compactHistory(
     input: Omit<BackendCompactHistoryInput, 'runId'> & { runId: string | undefined },
     automaticMemoryBoundary?: HistoryCompactMemoryExtractionBoundary,
+  ): Promise<AiSdkCompactHistoryResult> {
+    return this.compactHistoryOnce(input, automaticMemoryBoundary);
+  }
+
+  private async compactHistoryOnce(
+    input: Omit<BackendCompactHistoryInput, 'runId'> & { runId: string | undefined },
+    automaticMemoryBoundary?: HistoryCompactMemoryExtractionBoundary,
   ): Promise<AiSdkCompactHistoryResult> {
     const historyCompactAbortController = new AbortController();
-    this.historyCompactAbortController = historyCompactAbortController;
+    this.historyCompactAbortControllers.add(historyCompactAbortController);
     try {
       const policy = this.input.contextBudget;
       const summarizer = this.input.summarizeHistoryCompact;
@@ -399,7 +416,11 @@ export class AiSdkCompaction {
             source: {
               foldedRuntimeEvents: [...coveredRuntimeEvents],
               ...(input.runtimeContextRunHeaders
-                ? { runHeaders: input.runtimeContextRunHeaders }
+                ? {
+                    runHeaders: input.runtimeContextRunHeaders.filter((run) =>
+                      coveredRuntimeEvents.some((event) => event.runId === run.runId),
+                    ),
+                  }
                 : {}),
             },
             newlyFoldedRuntimeEvents: [...newlyFoldedRuntimeEvents],
@@ -501,9 +522,7 @@ export class AiSdkCompaction {
         }),
       };
     } finally {
-      if (this.historyCompactAbortController === historyCompactAbortController) {
-        this.historyCompactAbortController = null;
-      }
+      this.historyCompactAbortControllers.delete(historyCompactAbortController);
     }
   }
 
@@ -546,8 +565,13 @@ export class AiSdkCompaction {
     const priorFailure = this.malformedSummaryFailures.get(fingerprint);
     if (priorFailure) throw new HistoryCompactSummarizerError(priorFailure);
 
+    const existing = this.inFlightHistorySummaries.get(fingerprint);
+    if (existing) return existing;
+
+    const pending = Promise.resolve().then(() => summarizer(input));
+    this.inFlightHistorySummaries.set(fingerprint, pending);
     try {
-      return await Promise.resolve(summarizer(input));
+      return await pending;
     } catch (error) {
       if (
         error instanceof HistoryCompactSummarizerError &&
@@ -562,6 +586,10 @@ export class AiSdkCompaction {
         }
       }
       throw error;
+    } finally {
+      if (this.inFlightHistorySummaries.get(fingerprint) === pending) {
+        this.inFlightHistorySummaries.delete(fingerprint);
+      }
     }
   }
 

packages/runtime/src/history-compact-checkpoint-coordinator.ts

@@ -86,9 +86,17 @@ export class HistoryCompactCheckpointCoordinator {
       .catch(() => {})
       .then(async () => {
         const durableCheckpoint = await this.load(sessionId);
-        if (!canReplaceHistoryCompactCheckpoint(durableCheckpoint, checkpoint)) {
+        const sameEffectiveCheckpoint = hasSameEffectiveCoverage(durableCheckpoint, checkpoint);
+        if (
+          !sameEffectiveCheckpoint &&
+          !canReplaceHistoryCompactCheckpoint(durableCheckpoint, checkpoint)
+        ) {
           throw new Error('History compact checkpoint was superseded before persistence');
         }
+        // A concurrent caller may have received the same effective checkpoint
+        // from the summarizer coalescer. Keep its run-local ledger event too so
+        // the rider Turn retains provenance even though the session checkpoint
+        // itself is already current.
         await run.recordHistoryCompactCheckpoint(checkpoint);
         this.checkpoints.set(sessionId, checkpoint);
         this.scheduleCleanup(sessionId, checkpoint);
@@ -145,3 +153,35 @@ export class HistoryCompactCheckpointCoordinator {
     this.cleanups.set(sessionId, tracked);
   }
 }
+
+function hasSameEffectiveCoverage(
+  current: HistoryCompactCheckpoint | undefined,
+  candidate: HistoryCompactCheckpoint,
+): boolean {
+  if (!current) return false;
+  const currentContent = checkpointContent(current);
+  const candidateContent = checkpointContent(candidate);
+  return (
+    current.sessionId === candidate.sessionId &&
+    current.version === candidate.version &&
+    current.highWaterName === candidate.highWaterName &&
+    current.phase === candidate.phase &&
+    current.coverage.eventCount === candidate.coverage.eventCount &&
+    current.coverage.turnCount === candidate.coverage.turnCount &&
+    current.coverage.sourceDigest === candidate.coverage.sourceDigest &&
+    current.coverage.through.runId === candidate.coverage.through.runId &&
+    current.coverage.through.turnId === candidate.coverage.through.turnId &&
+    current.coverage.through.runtimeEventId === candidate.coverage.through.runtimeEventId &&
+    JSON.stringify(current.source) === JSON.stringify(candidate.source) &&
+    JSON.stringify(current.headAnchor) === JSON.stringify(candidate.headAnchor) &&
+    JSON.stringify(current.memoryExtractionBoundary) ===
+      JSON.stringify(candidate.memoryExtractionBoundary) &&
+    JSON.stringify(currentContent) === JSON.stringify(candidateContent)
+  );
+}
+
+function checkpointContent(checkpoint: HistoryCompactCheckpoint): unknown {
+  return checkpoint.version === 2
+    ? { summary: checkpoint.summary, summaryFormat: checkpoint.summaryFormat }
+    : { providerState: checkpoint.providerState };
+}