From 4ec99b83ffcfed06300262a26afad259de65604d Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:09:21 +0000 Subject: [PATCH] feat(compaction): Request a background summary after each model call Co-Authored-By: Thomas Jiang <20501235+thomasjiangcy@users.noreply.github.com> Co-Authored-By: Thomas Jiang --- TELEMETRY.md | 13 ++-- packages/junior/src/chat/agent/index.ts | 31 +++++--- .../chat/services/background-compaction.ts | 71 ++++++++++++++++--- .../chat/services/context-compaction-swap.ts | 21 +----- .../background-compaction-events.test.ts | 2 + 5 files changed, 92 insertions(+), 46 deletions(-) diff --git a/TELEMETRY.md b/TELEMETRY.md index 5c63cf662a..0d5e01b26a 100644 --- a/TELEMETRY.md +++ b/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 | | -------------------------------- | ------ | ------------------------------------------------ | diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index 58b29c7cb0..e22dff1926 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/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 => + 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 diff --git a/packages/junior/src/chat/services/background-compaction.ts b/packages/junior/src/chat/services/background-compaction.ts index 0412915acb..2c445bade6 100644 --- a/packages/junior/src/chat/services/background-compaction.ts +++ b/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 { 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; diff --git a/packages/junior/src/chat/services/context-compaction-swap.ts b/packages/junior/src/chat/services/context-compaction-swap.ts index 01a82edd41..9b0df2c003 100644 --- a/packages/junior/src/chat/services/context-compaction-swap.ts +++ b/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. diff --git a/packages/junior/tests/integration/background-compaction-events.test.ts b/packages/junior/tests/integration/background-compaction-events.test.ts index fb9c786621..47dfd4f4b2 100644 --- a/packages/junior/tests/integration/background-compaction-events.test.ts +++ b/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(["The person picked blue."]); await processBackgroundCompaction(sent[0]!, { streamFn: model.streamFn });