feat(compaction): Request a background summary after each model call

TELEMETRY.md

@@ -368,12 +368,13 @@ result.
 
 ### Background Compaction
 
-A Turn whose last model request had a provider-reported input at or above
-the mark writes `compaction_requested` and logs
-`context_compaction.background.requested`. The mark is 70% of the compaction
-threshold. A queue job then writes `compaction_prepared`. The job has its own
-`agent.context_compaction` span with `app.compaction.mode:background`. It runs
-in a new trace and adds no `app.turn.compaction_wait_ms`.
+After each model request with a provider-reported input at or above the
+mark, also in the middle of a Turn, the Turn writes `compaction_requested`
+and logs `context_compaction.background.requested`. The mark is 70% of the
+compaction threshold. A queue job then writes `compaction_prepared`. The job
+has its own `agent.context_compaction` span with
+`app.compaction.mode:background`. It runs in a new trace and adds no
+`app.turn.compaction_wait_ms`.
 
 | Attribute                        | Type   | Meaning                                          |
 | -------------------------------- | ------ | ------------------------------------------------ |

packages/junior/src/chat/agent/index.ts

@@ -855,6 +855,20 @@ async function executeAgentRunInPrivacyContext(
         },
       };
     };
+    /**
+     * Request a background summary when the model call that just ended was at
+     * or above the mark. It runs after each model call, so a long Turn gets a
+     * summary before it reaches the compaction threshold.
+     */
+    const requestBackgroundCompactionAfterModelCall = async (): Promise<void> =>
+      await requestBackgroundCompaction({
+        agent: agent!,
+        conversationId,
+        conversationPrivacy,
+        modelId: activeModelId,
+        queue: run.environment?.backgroundCompactionQueue,
+        turnStartedAtMs: replyStartedAtMs,
+      });
     // ── Agent execution ──────────────────────────────────────────────
     let assistantMessageDeliveryError:
       | AssistantMessageDeliveryError
@@ -1062,11 +1076,13 @@ async function executeAgentRunInPrivacyContext(
               const pendingMessages = await drainSteeringMessages();
               // Pi also runs this hook after the final reply. Another model
               // call follows only after tool results or steering. Compaction
-              // here would drop the delivered reply and start a retry.
+              // here would drop the delivered reply and start a retry. A
+              // background request does not change history, so it still runs.
               if (
                 nextTurn.toolResults.length === 0 &&
                 pendingMessages.length === 0
               ) {
+                await requestBackgroundCompactionAfterModelCall();
                 return contextUpdate;
               }
               const capacityUpdate = await applyActiveContextCompaction(
@@ -1077,6 +1093,11 @@ async function executeAgentRunInPrivacyContext(
                 pendingMessages,
                 true,
               );
+              // After a replacement, the usage of the last model call counts
+              // the old history. The next model call reports the new size.
+              if (!capacityUpdate) {
+                await requestBackgroundCompactionAfterModelCall();
+              }
               return capacityUpdate
                 ? { ...contextUpdate, ...capacityUpdate }
                 : contextUpdate;
@@ -1489,14 +1510,6 @@ async function executeAgentRunInPrivacyContext(
     }
 
     await recordActiveMcpProviders();
-    await requestBackgroundCompaction({
-      agent,
-      conversationId,
-      conversationPrivacy,
-      modelId: activeModelId,
-      queue: run.environment?.backgroundCompactionQueue,
-      turnStartedAtMs: replyStartedAtMs,
-    });
     // Generation completing is not durable completion: the session record
     // stays at its latest running safe boundary here. The destination owns the
     // completed session record after assistant output handling settles, so a

packages/junior/src/chat/services/background-compaction.ts

@@ -1,10 +1,10 @@
 /**
  * Background compaction: prepare a summary before the history is full.
  *
- * When a Turn completes and its last model request was at or above the mark,
- * the Turn appends `compaction_requested` and sends a job. The job asks the
- * agent model for a summary of the history up to the marked event and
- * appends `compaction_prepared`. Turns keep the full history until a later
+ * After each model request at or above the mark, also in the middle of a
+ * Turn, the Turn appends `compaction_requested` and sends a job. The job
+ * asks the agent model for a summary of the history up to the marked event
+ * and appends `compaction_prepared`. Turns keep the full history until a later
  * change swaps the summary in. Blocking compaction stays the fallback.
  *
  * The summary request repeats the system prompt and tools of the last agent
@@ -151,14 +151,57 @@ function mayRequest(
 }
 
 /**
- * Request a background summary when the last model request of a completed
- * Turn was at or above the mark.
+ * Find the newest background compaction fact of the current History version.
+ * Facts of earlier versions belong to history that a replacement already
+ * removed.
+ */
+export async function loadLatestBackgroundCompaction(conversationId: string) {
+  const store = getConversationEventStore();
+  const [historyVersion, page] = await Promise.all([
+    store.loadCurrentHistoryVersion(conversationId),
+    store.query(conversationId, {
+      limit: 1,
+      types: ["compaction_requested", "compaction_prepared"],
+    }),
+  ]);
+  const event = page.events.at(-1);
+  return event && event.historyVersion === historyVersion
+    ? { event, historyVersion }
+    : undefined;
+}
+
+/**
+ * Return whether the newest fact is a pending request or a summary that
+ * became ready in this execution slice. Then `mayRequest` is false too, so a
+ * Turn can skip the full history load. Only one request is pending at a time,
+ * so the newest fact decides it.
+ */
+function waitsForUse(
+  event: ConversationEvent,
+  nowMs: number,
+  turnStartedAtMs: number,
+): boolean {
+  if (event.data.type === "compaction_requested") {
+    return nowMs - event.createdAtMs < REQUEST_TIMEOUT_MS;
+  }
+  return (
+    event.data.type === "compaction_prepared" &&
+    event.data.outcome === "ready" &&
+    event.createdAtMs >= turnStartedAtMs
+  );
+}
+
+/**
+ * Request a background summary when the model request that just ended was at
+ * or above the mark.
  *
- * `turnStartedAtMs` is when this execution slice started, before its first
- * model request. Without a queue or below the mark, it does nothing. A failure here leaves
- * blocking compaction as the fallback, so it is logged and does not fail the
- * Turn. When the send fails after the request is written, the request times
- * out and counts as a failed attempt.
+ * The agent runs it after each model request, so a long Turn gets a summary
+ * before the compaction threshold. `turnStartedAtMs` is when this execution
+ * slice started, before its first model request. Without a queue or below the
+ * mark, it does nothing. A failure here leaves blocking compaction as the
+ * fallback, so it is logged and does not fail the Turn. When the send fails
+ * after the request is written, the request times out and counts as a
+ * failed attempt.
  */
 export async function requestBackgroundCompaction(args: {
   agent: Agent;
@@ -197,6 +240,12 @@ async function requestAboveMark(args: {
   turnStartedAtMs: number;
 }): Promise<void> {
   const nowMs = Date.now();
+  // Each later model request of a large Turn checks again. Read only the
+  // newest fact while a request waits.
+  const latest = await loadLatestBackgroundCompaction(args.conversationId);
+  if (latest && waitsForUse(latest.event, nowMs, args.turnStartedAtMs)) {
+    return;
+  }
   const eventStore = getConversationEventStore();
   const events = await eventStore.loadCurrentHistory(args.conversationId);
   const historyVersion = events[0]?.historyVersion;

packages/junior/src/chat/services/context-compaction-swap.ts

@@ -18,6 +18,7 @@ import {
   type ConversationMessageProvenance,
 } from "@/chat/conversations/provenance";
 import { setSpanAttributes } from "@/chat/logging";
+import { loadLatestBackgroundCompaction } from "@/chat/services/background-compaction";
 import { historyItemFromPiMessage } from "@/chat/pi/conversation-events";
 import type { PiMessage } from "@/chat/pi/messages";
 import { trimTrailingAssistantMessages } from "@/chat/pi/transcript";
@@ -95,26 +96,6 @@ function discard(reason: DiscardReason): CompactContextResult {
   return { compacted: false };
 }
 
-/**
- * Find the newest background compaction fact of the current History version.
- * Facts of earlier versions belong to history that a replacement already
- * removed.
- */
-async function loadLatestBackgroundCompaction(conversationId: string) {
-  const store = getConversationEventStore();
-  const [historyVersion, page] = await Promise.all([
-    store.loadCurrentHistoryVersion(conversationId),
-    store.query(conversationId, {
-      limit: 1,
-      types: ["compaction_requested", "compaction_prepared"],
-    }),
-  ]);
-  const event = page.events.at(-1);
-  return event && event.historyVersion === historyVersion
-    ? { event, historyVersion }
-    : undefined;
-}
-
 /**
  * Swap a ready prepared summary into history, or run today's capacity
  * compaction when none can be used.

packages/junior/tests/integration/background-compaction-events.test.ts

@@ -267,6 +267,8 @@ describe("background compaction job", () => {
         markSeq: first!.seq,
       },
     ]);
+    // A later model request waits for the pending request.
+    await expect(completeTurn(liveAgent(mark))).resolves.toEqual([]);
 
     const model = replies(["<summary>The person picked blue.</summary>"]);
     await processBackgroundCompaction(sent[0]!, { streamFn: model.streamFn });