From baf26e182a251d229fe5007db223aae96c86b871 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:08:55 +0000 Subject: [PATCH 1/2] feat(compaction): Prepare a summary in a background job When a completed Turn's last model request is at or above the mark (70% of the compaction threshold, provider-reported input), the Turn writes compaction_requested and sends a signed queue job. The job repeats the request prefix of the last agent request and writes compaction_prepared with a summary or a fixed failure label. --- TELEMETRY.md | 28 +- packages/junior-evals/src/fixture/agent.ts | 4 + packages/junior/src/app.ts | 33 ++ .../junior/src/chat/agent/context-summary.ts | 2 +- packages/junior/src/chat/agent/index.ts | 9 + packages/junior/src/chat/agent/types.ts | 3 + .../junior/src/chat/runtime/agent-runner.ts | 6 + .../background-compaction-callback.ts | 60 +++ .../services/background-compaction-queue.ts | 81 ++++ .../chat/services/background-compaction.ts | 434 ++++++++++++++++++ .../src/chat/services/context-budget.ts | 19 + .../services/context-compaction-telemetry.ts | 7 +- packages/junior/src/deployment.ts | 2 + packages/junior/src/nitro.ts | 28 ++ .../background-compaction-events.test.ts | 220 +++++++++ .../unit/build/nitro-plugin-module.test.ts | 13 + 16 files changed, 944 insertions(+), 5 deletions(-) create mode 100644 packages/junior/src/chat/services/background-compaction-callback.ts create mode 100644 packages/junior/src/chat/services/background-compaction-queue.ts create mode 100644 packages/junior/src/chat/services/background-compaction.ts diff --git a/TELEMETRY.md b/TELEMETRY.md index 1ace1dd345..66c5eb41a9 100644 --- a/TELEMETRY.md +++ b/TELEMETRY.md @@ -314,7 +314,7 @@ Attributes on the `agent.context_compaction` span: | Attribute | Type | Meaning | | ------------------------------------- | ------ | --------------------------------------------- | -| `app.compaction.mode` | string | `blocking`, `active`, or `handoff` | +| `app.compaction.mode` | string | `blocking`, `active`, `handoff`, `background` | | `app.compaction.outcome` | string | `compacted` or `summary_failed` | | `app.compaction.input.estimate` | int | estimated history tokens before compaction | | `app.compaction.replacement.estimate` | int | estimated history tokens after compaction | @@ -356,6 +356,32 @@ result. - A `handoff` span has only `app.compaction.mode`. A failed handoff or an input above the limit sets `error.type`. +### 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`. + +| Attribute | Type | Meaning | +| -------------------------------- | ------ | ------------------------------------------------ | +| `app.compaction.outcome` | string | `ready`, `failed`, `stale`, or `duplicate` | +| `app.compaction.mark_seq` | int | last event seq that the summary covers | +| `app.compaction.history_version` | int | History version of the request | +| `app.compaction.input.size` | int | provider-reported input of the marked request | +| `app.compaction.queue_wait_ms` | int | time from queue send to job start | +| `app.compaction.ready_ms` | int | time from request to ready summary | +| `app.compaction.summary_chars` | int | summary length in characters | +| `error.type` | string | failure label, also stored on the `failed` event | + +- `stale`: the History was replaced before the job ran. `duplicate`: the + request already has a result. Both write nothing. +- Failure labels: `timeout`, `snapshot_missing`, `provider_`, + `summary_error`, and the summary failures `empty`, `max_tokens`, + `missing_summary`, `tool_call`. + Summary call latency and cache reads by mode. In queries, a numeric `app.*` attribute needs the `tags[,number]` form. diff --git a/packages/junior-evals/src/fixture/agent.ts b/packages/junior-evals/src/fixture/agent.ts index 9078c6060c..7a76d5f989 100644 --- a/packages/junior-evals/src/fixture/agent.ts +++ b/packages/junior-evals/src/fixture/agent.ts @@ -225,6 +225,10 @@ export async function createFixtureAgent( pluginTaskQueue: (consume) => ({ send: async (message) => track(consume(message)), }), + // Background compaction jobs also run in process. + backgroundCompactionQueue: (consume) => ({ + send: async (message) => track(consume(message)), + }), waitUntil: (task) => track(typeof task === "function" ? task() : task), }); // The dashboard mounts the same API after sign-in. Each request signs in diff --git a/packages/junior/src/app.ts b/packages/junior/src/app.ts index 7dcaf477a6..2f203876e9 100644 --- a/packages/junior/src/app.ts +++ b/packages/junior/src/app.ts @@ -72,6 +72,7 @@ import { } from "@/handlers/sandbox-egress-signals"; import { POST as slackWebhookPOST } from "@/handlers/slack-webhook"; import { + JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE, JUNIOR_PLUGIN_TASK_CALLBACK_ROUTE, JUNIOR_WORKSPACE_SNAPSHOT_JOB_CALLBACK_ROUTE, } from "@/deployment"; @@ -100,6 +101,17 @@ import { registerVercelWorkspaceSnapshotJobDevConsumer, workspaceSnapshotJobQueueConsumer, } from "@/chat/sandbox/snapshot/job-callback"; +import { + backgroundCompactionQueueConsumer, + createVercelBackgroundCompactionCallback, + registerVercelBackgroundCompactionDevConsumer, +} from "@/chat/services/background-compaction-callback"; +import { processBackgroundCompaction } from "@/chat/services/background-compaction"; +import { + vercelBackgroundCompactionQueue, + type BackgroundCompactionMessage, + type BackgroundCompactionQueue, +} from "@/chat/services/background-compaction-queue"; import { getQueueProvider, type QueueConsumer, @@ -235,6 +247,13 @@ export interface JuniorAppOptions extends BotModelConfig { pluginTaskQueue?: ( consume: (message: PluginTaskQueueMessage) => Promise, ) => { send(message: PluginTaskQueueMessage): Promise }; + /** + * Replace the Vercel Queue transport for background compaction jobs, for + * example with an in-process queue. `consume` runs one job. + */ + backgroundCompactionQueue?: ( + consume: (message: BackgroundCompactionMessage) => Promise, + ) => BackgroundCompactionQueue; /** Direct plugin set override. Usually omitted when `juniorNitro()` uses a plugin module. */ plugins?: JuniorPluginSet; /** Sandbox execution options. */ @@ -588,9 +607,14 @@ export async function createApp( ) ?? getVercelConversationWorkQueue(); const pluginTaskQueue = options?.pluginTaskQueue?.(processPluginTask); + const backgroundCompactionQueue = + options?.backgroundCompactionQueue?.((message) => + processBackgroundCompaction(message), + ) ?? vercelBackgroundCompactionQueue; const attachmentStorage = getAttachmentStorage(); const agentRunner = createAgentRunner(executeAgentRun, { attachmentStorage, + backgroundCompactionQueue, bindMessageAgent: (request) => bindMessageAgent(request, { queue: conversationWorkQueue }), bindSpawnAgent: (request) => @@ -808,6 +832,7 @@ export async function createApp( registerVercelConversationWorkDevConsumer(getConversationWorkOptions()); registerVercelPluginTaskDevConsumer(); registerVercelWorkspaceSnapshotJobDevConsumer(); + registerVercelBackgroundCompactionDevConsumer(); } app.post("/api/internal/agent/continue", (c) => { agentContinuePOST ??= createVercelConversationWorkCallback( @@ -826,6 +851,13 @@ export async function createApp( workspaceSnapshotJobPOST ??= createVercelWorkspaceSnapshotJobCallback(); return workspaceSnapshotJobPOST(c.req.raw); }); + let backgroundCompactionPOST: + | ReturnType + | undefined; + app.post(JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE, (c) => { + backgroundCompactionPOST ??= createVercelBackgroundCompactionCallback(); + return backgroundCompactionPOST(c.req.raw); + }); app.get("/api/internal/heartbeat", (c) => { return heartbeatGET(c.req.raw, waitUntil, { conversationWorkQueue }); @@ -852,6 +884,7 @@ export async function createApp( conversationWorkQueueConsumer(getConversationWorkOptions()), pluginTaskQueueConsumer(), workspaceSnapshotJobQueueConsumer(), + backgroundCompactionQueueConsumer(), ], backgroundWork, ), diff --git a/packages/junior/src/chat/agent/context-summary.ts b/packages/junior/src/chat/agent/context-summary.ts index 6cf1d2df2f..4dfc64e2c8 100644 --- a/packages/junior/src/chat/agent/context-summary.ts +++ b/packages/junior/src/chat/agent/context-summary.ts @@ -64,7 +64,7 @@ export interface AgentModelRequest { type SummaryFailure = "empty" | "max_tokens" | "missing_summary" | "tool_call"; /** Raised when the model returns no usable summary after one nudge. */ -class ContextSummaryError extends Error { +export class ContextSummaryError extends Error { readonly failure: SummaryFailure; constructor(failure: SummaryFailure) { diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index eac1b6cc49..6f0ea5cc03 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -162,6 +162,7 @@ import { type ModelProfile, } from "@/chat/model-profile"; import { compactWithPreparedSummary } from "@/chat/services/context-compaction-swap"; +import { requestBackgroundCompaction } from "@/chat/services/background-compaction"; import { buildAgentModelRequest, CONTEXT_COMPACTION_PROMPT_NAME, @@ -1789,6 +1790,14 @@ async function executeAgentRunInPrivacyContext( } await recordActiveMcpProviders(); + await requestBackgroundCompaction({ + agent, + conversationId, + conversationPrivacy, + modelId: activeModelId, + modelProfile: activeModelProfile, + queue: run.environment?.backgroundCompactionQueue, + }); // 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/agent/types.ts b/packages/junior/src/chat/agent/types.ts index 6de08198f6..f417f261e8 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -34,6 +34,7 @@ import type { import type { AssistantMessage } from "@earendil-works/pi-ai"; import type { AttachmentStorage } from "@/chat/attachments/storage"; import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import type { BackgroundCompactionQueue } from "@/chat/services/background-compaction-queue"; /** One attachment the model may see for the current instruction. */ export type AgentAttachment = { @@ -225,6 +226,8 @@ export type AgentEvent = /** Resolved environment and optional per-run tool test overrides. */ export type AgentEnvironment = { attachmentStorage?: AttachmentStorage; + /** Queue for background compaction jobs. Without it, Turns request none. */ + backgroundCompactionQueue?: BackgroundCompactionQueue; configuration?: Record; /** Queue of the app, for tools that wake other Conversations. */ conversationWorkQueue?: ConversationWorkQueue; diff --git a/packages/junior/src/chat/runtime/agent-runner.ts b/packages/junior/src/chat/runtime/agent-runner.ts index 85cd015bca..be7e4fb675 100644 --- a/packages/junior/src/chat/runtime/agent-runner.ts +++ b/packages/junior/src/chat/runtime/agent-runner.ts @@ -11,6 +11,7 @@ import type { AgentRunOutcome } from "@/chat/runtime/agent-run-outcome"; import type { SandboxEgressTracePropagationConfig } from "@/chat/sandbox/egress/tracing"; import type { AttachmentStorage } from "@/chat/attachments/storage"; import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import type { BackgroundCompactionQueue } from "@/chat/services/background-compaction-queue"; /** Run one agent-run slice behind runtime-owned orchestration boundaries. */ export interface AgentRunner { @@ -22,6 +23,7 @@ export function createAgentRunner( execute: (run: AgentRun, streamFn?: StreamFn) => Promise, options?: { attachmentStorage?: AttachmentStorage; + backgroundCompactionQueue?: BackgroundCompactionQueue; bindMessageAgent?: (run: AgentRun) => MessageAgent | undefined; bindSpawnAgent?: (run: AgentRun) => SpawnAgent | undefined; conversationWorkQueue?: ConversationWorkQueue; @@ -30,6 +32,7 @@ export function createAgentRunner( }, ): AgentRunner { const attachmentStorage = options?.attachmentStorage; + const backgroundCompactionQueue = options?.backgroundCompactionQueue; const conversationWorkQueue = options?.conversationWorkQueue; const streamFn = options?.streamFn; const tracePropagation = options?.tracePropagation; @@ -56,6 +59,9 @@ export function createAgentRunner( ...run.environment, attachmentStorage: run.environment?.attachmentStorage ?? attachmentStorage, + backgroundCompactionQueue: + run.environment?.backgroundCompactionQueue ?? + backgroundCompactionQueue, conversationWorkQueue: run.environment?.conversationWorkQueue ?? conversationWorkQueue, sandboxTracePropagation: diff --git a/packages/junior/src/chat/services/background-compaction-callback.ts b/packages/junior/src/chat/services/background-compaction-callback.ts new file mode 100644 index 0000000000..fd92dadab4 --- /dev/null +++ b/packages/junior/src/chat/services/background-compaction-callback.ts @@ -0,0 +1,60 @@ +/** Receive background compaction jobs from the queue. */ +import { logWarn } from "@/chat/logging"; +import { queueCallback } from "@/chat/queue/callback"; +import type { + QueueConsumer, + QueueMessageMetadata, +} from "@/chat/queue/provider"; +import { processBackgroundCompaction } from "./background-compaction"; +import { + BACKGROUND_COMPACTION_QUEUE_TOPIC, + verifyBackgroundCompactionMessage, +} from "./background-compaction-queue"; + +const BACKGROUND_COMPACTION_DEV_CONSUMER_GROUP = + "junior_context_compaction_dev"; +// The job records model failures itself. Retries cover storage errors only. +const BACKGROUND_COMPACTION_MAX_DELIVERIES = 3; + +function logRejected(reason: string, metadata: QueueMessageMetadata): void { + logWarn("context_compaction.queue_message.rejected", { + "app.queue.consumer_group": metadata.consumerGroup, + "app.queue.delivery_count": metadata.deliveryCount, + "app.queue.message_id": metadata.messageId, + "app.queue.reject_reason": reason, + "app.queue.topic_name": metadata.topicName, + }); +} + +function backgroundCompactionCallback() { + return queueCallback({ + consumerGroup: BACKGROUND_COMPACTION_DEV_CONSUMER_GROUP, + maxDeliveries: BACKGROUND_COMPACTION_MAX_DELIVERIES, + onRejected: logRejected, + run: async (message, metadata) => + await processBackgroundCompaction(message, { + enqueuedAtMs: metadata.createdAt.getTime(), + }), + topic: BACKGROUND_COMPACTION_QUEUE_TOPIC, + verify: verifyBackgroundCompactionMessage, + }); +} + +/** Create the HTTP route for background compaction jobs. */ +export function createVercelBackgroundCompactionCallback(): ( + request: Request, +) => Promise { + return backgroundCompactionCallback().create(); +} + +/** Return the queue consumer for background compaction, for a pulling worker. */ +export function backgroundCompactionQueueConsumer(): QueueConsumer { + return backgroundCompactionCallback().consumer; +} + +/** Receive background compaction jobs during local development. */ +export function registerVercelBackgroundCompactionDevConsumer(): + | (() => void) + | undefined { + return backgroundCompactionCallback().registerDev(); +} diff --git a/packages/junior/src/chat/services/background-compaction-queue.ts b/packages/junior/src/chat/services/background-compaction-queue.ts new file mode 100644 index 0000000000..5b75387328 --- /dev/null +++ b/packages/junior/src/chat/services/background-compaction-queue.ts @@ -0,0 +1,81 @@ +/** + * Messages of background compaction jobs. + * + * A message names one `compaction_requested` event. The event log holds the + * request, and the State adapter holds the request snapshot, so the message + * stays small. + */ +import { createHash } from "node:crypto"; +import { z } from "zod"; +import { getQueueProvider } from "@/chat/queue/provider"; +import { + QUEUE_SIGNATURE_MAX_AGE_MS, + signQueueMessage, + verifyQueueMessage, +} from "@/chat/queue/sign"; + +/** Deploy topic name. Keep stable for Vercel queue wiring. */ +export const BACKGROUND_COMPACTION_QUEUE_TOPIC = "junior_context_compaction"; + +const backgroundCompactionMessageSchema = z + .object({ + conversationId: z.string().min(1), + historyVersion: z.number().int().nonnegative(), + markSeq: z.number().int().nonnegative(), + }) + .strict(); + +/** One background compaction job: the request that it prepares a summary for. */ +export type BackgroundCompactionMessage = z.output< + typeof backgroundCompactionMessageSchema +>; + +/** Sends background compaction jobs. Tests and evals can run them in process. */ +export interface BackgroundCompactionQueue { + send(message: BackgroundCompactionMessage): Promise; +} + +const backgroundCompactionSign = { + context: "junior.context_compaction_queue.v1", + schema: backgroundCompactionMessageSchema, + signatureVersion: "v1" as const, + parts: (message: BackgroundCompactionMessage) => [ + message.conversationId, + String(message.historyVersion), + String(message.markSeq), + ], +}; + +/** Stable id of one request, for event, snapshot, and delivery dedupe. */ +export function backgroundCompactionId( + message: BackgroundCompactionMessage, +): string { + const digest = createHash("sha256") + .update(message.conversationId) + .update("\0") + .update(String(message.historyVersion)) + .update("\0") + .update(String(message.markSeq)) + .digest("hex") + .slice(0, 32); + return `context-compaction_${digest}`; +} + +/** Check that a recent queue message came from Junior. */ +export function verifyBackgroundCompactionMessage(value: unknown) { + return verifyQueueMessage(backgroundCompactionSign, value); +} + +/** The production queue for background compaction jobs. */ +export const vercelBackgroundCompactionQueue: BackgroundCompactionQueue = { + send: async (message) => { + await getQueueProvider().send( + BACKGROUND_COMPACTION_QUEUE_TOPIC, + signQueueMessage(backgroundCompactionSign, message), + { + idempotencyKey: backgroundCompactionId(message), + retentionSeconds: QUEUE_SIGNATURE_MAX_AGE_MS / 1000, + }, + ); + }, +}; diff --git a/packages/junior/src/chat/services/background-compaction.ts b/packages/junior/src/chat/services/background-compaction.ts new file mode 100644 index 0000000000..906a83f7cd --- /dev/null +++ b/packages/junior/src/chat/services/background-compaction.ts @@ -0,0 +1,434 @@ +/** + * 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 + * change swaps the summary in. Blocking compaction stays the fallback. + * + * The summary request repeats the system prompt and tools of the last agent + * request, so the provider can read its cached prompt prefix. The job runs in + * another request, so the Turn stores them in the State adapter. + */ +import { + Agent, + type AgentMessage, + type AgentTool, + type StreamFn, + type ThinkingLevel, +} from "@earendil-works/pi-agent-core"; +import type { Tool } from "@earendil-works/pi-ai"; +import { + buildAgentModelRequest, + CONTEXT_COMPACTION_PROMPT_NAME, + ContextSummaryError, + requestContextSummary, +} from "@/chat/agent/context-summary"; +import { botConfig } from "@/chat/config"; +import { + runWithConversationPrivacy, + type ConversationPrivacy, +} from "@/chat/conversation-privacy"; +import type { ConversationEvent } from "@/chat/conversations/history"; +import { getConversationEventStore } from "@/chat/db"; +import { + logInfo, + logWarn, + setSpanAttributes, + withLogContext, +} from "@/chat/logging"; +import type { ModelProfile } from "@/chat/model-profile"; +import { getGatewayApiKey, resolveGatewayModel } from "@/chat/pi/client"; +import { projectConversationEvents } from "@/chat/pi/conversation-events"; +import { createTracedStreamFn } from "@/chat/pi/traced-stream"; +import { QUEUE_SIGNATURE_MAX_AGE_MS } from "@/chat/queue/sign"; +import { getAgentBackgroundCompactionMarkTokens } from "@/chat/services/context-budget"; +import { withCompactionSpan } from "@/chat/services/context-compaction-telemetry"; +import { findProviderError } from "@/chat/services/provider-error"; +import { getStateAdapter } from "@/chat/state/adapter"; +import { + backgroundCompactionId, + type BackgroundCompactionMessage, + type BackgroundCompactionQueue, +} from "./background-compaction-queue"; + +/** Failed attempts for one History version before background tries stop. */ +const MAX_FAILED_ATTEMPTS = 2; +/** + * A request without a prepared result after this long counts as a failed + * attempt, so a later Turn can request again. This covers a lost job. + */ +const REQUEST_TIMEOUT_MS = 10 * 60 * 1000; +/** The queue drops a message after this long, so the snapshot can go too. */ +const SNAPSHOT_TTL_MS = QUEUE_SIGNATURE_MAX_AGE_MS; + +/** What the job needs from the Turn to repeat its request prefix. */ +interface RequestSnapshot { + conversationPrivacy: ConversationPrivacy; + /** Provider-reported input of the last model request. */ + inputSize: number; + modelId: string; + systemPrompt: string; + thinkingLevel: ThinkingLevel; + tools: Array< + Pick & + Pick + >; +} + +function snapshotKey(message: BackgroundCompactionMessage): string { + return `${backgroundCompactionId(message)}:snapshot`; +} + +/** Provider-reported input of the last model request, cache included. */ +function lastRequestInputSize(messages: AgentMessage[]): number | undefined { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]!; + if (message.role === "assistant") { + const { input, cacheRead, cacheWrite } = message.usage; + return input + cacheRead + cacheWrite; + } + } + return undefined; +} + +/** + * Return whether a Turn may request a background summary for this History + * version. It may not while a request is pending, when a summary is ready, or + * after the failure limit. A ready summary stays usable for the version, so + * one is enough. + */ +function mayRequest( + events: ConversationEvent[], + historyVersion: number, + nowMs: number, +): boolean { + const outcomes = new Map(); + for (const event of events) { + if ( + event.data.type === "compaction_prepared" && + event.data.historyVersion === historyVersion + ) { + outcomes.set(event.data.markSeq, event.data.outcome); + } + } + let failed = 0; + for (const event of events) { + if ( + event.data.type !== "compaction_requested" || + event.data.historyVersion !== historyVersion + ) { + continue; + } + const outcome = outcomes.get(event.data.markSeq); + if (outcome === "ready") { + return false; + } + if ( + outcome === "failed" || + nowMs - event.createdAtMs >= REQUEST_TIMEOUT_MS + ) { + failed += 1; + } else { + return false; + } + } + return failed < MAX_FAILED_ATTEMPTS; +} + +/** + * Request a background summary when the last model request of a completed + * Turn was at or above the mark. + * + * 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; + conversationId: string; + conversationPrivacy: ConversationPrivacy | undefined; + modelId: string; + modelProfile: ModelProfile; + queue: BackgroundCompactionQueue | undefined; +}): Promise { + if (!args.queue) { + return; + } + const inputSize = lastRequestInputSize(args.agent.state.messages); + if ( + inputSize === undefined || + inputSize < getAgentBackgroundCompactionMarkTokens(args.modelId) + ) { + return; + } + try { + await requestAboveMark({ ...args, inputSize, queue: args.queue }); + } catch (error) { + logWarn("context_compaction.background.request_failed", { + "error.type": error instanceof Error ? error.name : typeof error, + }); + } +} + +async function requestAboveMark(args: { + agent: Agent; + conversationId: string; + conversationPrivacy: ConversationPrivacy | undefined; + inputSize: number; + modelId: string; + modelProfile: ModelProfile; + queue: BackgroundCompactionQueue; +}): Promise { + const nowMs = Date.now(); + const eventStore = getConversationEventStore(); + const events = await eventStore.loadCurrentHistory(args.conversationId); + const historyVersion = events[0]?.historyVersion; + if ( + historyVersion === undefined || + !mayRequest(events, historyVersion, nowMs) + ) { + return; + } + const projection = projectConversationEvents(events, { + defaultProfile: botConfig.defaultProfile, + }); + const markSeq = projection.seqs.at(-1); + // After a handoff, the history belongs to another Model profile. + if (markSeq === undefined || projection.modelProfile !== args.modelProfile) { + return; + } + + const message = { + conversationId: args.conversationId, + historyVersion, + markSeq, + }; + const { systemPrompt, thinkingLevel, tools } = args.agent.state; + const snapshot: RequestSnapshot = { + // Unknown privacy is private, as in the Turn. + conversationPrivacy: args.conversationPrivacy ?? "private", + inputSize: args.inputSize, + modelId: args.modelId, + systemPrompt, + thinkingLevel, + // The provider request reads only these fields of a tool. + tools: tools.map((tool) => ({ + name: tool.name, + label: tool.label, + description: tool.description, + parameters: tool.parameters, + ...(tool.constrainedSampling === undefined + ? undefined + : { constrainedSampling: tool.constrainedSampling }), + })), + }; + // Write the snapshot first: the job can run as soon as the request exists. + await getStateAdapter().set( + snapshotKey(message), + JSON.parse(JSON.stringify(snapshot)), + SNAPSHOT_TTL_MS, + ); + const appended = await eventStore.append( + args.conversationId, + [ + { + idempotencyKey: `${backgroundCompactionId(message)}:requested`, + createdAtMs: nowMs, + data: { + type: "compaction_requested", + markSeq, + historyVersion, + modelProfile: args.modelProfile, + modelId: args.modelId, + }, + }, + ], + { activity: "preserve" }, + ); + // Another Turn wrote the same request and sent its job. + if (appended.length === 0) { + return; + } + await args.queue.send(message); + logInfo("context_compaction.background.requested", { + "app.compaction.mark_seq": markSeq, + "app.compaction.history_version": historyVersion, + "app.compaction.input.size": args.inputSize, + }); +} + +/** A fixed label for a failed summary. Never raw exception text. */ +function failureLabel(error: unknown, signal: AbortSignal): string { + if (signal.aborted) { + return "timeout"; + } + if (error instanceof ContextSummaryError) { + return error.failure; + } + const providerError = findProviderError(error); + return providerError ? `provider_${providerError.kind}` : "summary_error"; +} + +type JobOutcome = "ready" | "failed" | "stale" | "duplicate"; + +/** + * Run one background compaction job. + * + * It writes `compaction_prepared` once for the request, with a summary or a + * fixed failure label. It writes nothing when the request is no longer in + * the current History version, because no Turn could use the summary. + * `streamFn` replaces the model stream, for tests. + */ +export async function processBackgroundCompaction( + message: BackgroundCompactionMessage, + options: { enqueuedAtMs?: number; streamFn?: StreamFn } = {}, +): Promise { + const startedAtMs = Date.now(); + await withLogContext({ conversationId: message.conversationId }, () => + withCompactionSpan("background", async () => { + setSpanAttributes({ + "app.compaction.mark_seq": message.markSeq, + "app.compaction.history_version": message.historyVersion, + ...(options.enqueuedAtMs === undefined + ? undefined + : { + "app.compaction.queue_wait_ms": Math.max( + 0, + startedAtMs - options.enqueuedAtMs, + ), + }), + }); + const outcome = await prepareSummary(message, options.streamFn); + setSpanAttributes({ "app.compaction.outcome": outcome }); + }), + ); +} + +async function prepareSummary( + message: BackgroundCompactionMessage, + streamFn: StreamFn | undefined, +): Promise { + const eventStore = getConversationEventStore(); + const state = getStateAdapter(); + const events = await eventStore.loadCurrentHistory(message.conversationId); + const isThisRequest = (event: ConversationEvent) => + (event.data.type === "compaction_requested" || + event.data.type === "compaction_prepared") && + event.data.historyVersion === message.historyVersion && + event.data.markSeq === message.markSeq; + const request = events.find( + (event) => + isThisRequest(event) && event.data.type === "compaction_requested", + ); + // The history was replaced since the request. + if (request?.data.type !== "compaction_requested") { + await state.delete(snapshotKey(message)); + return "stale"; + } + const requested = request.data; + if ( + events.some( + (event) => + isThisRequest(event) && event.data.type === "compaction_prepared", + ) + ) { + return "duplicate"; + } + + const writePrepared = async ( + result: + | { outcome: "ready"; summary: string } + | { outcome: "failed"; errorType: string }, + ): Promise => { + await eventStore.append( + message.conversationId, + [ + { + idempotencyKey: `${backgroundCompactionId(message)}:prepared`, + createdAtMs: Date.now(), + data: { + type: "compaction_prepared", + markSeq: message.markSeq, + historyVersion: message.historyVersion, + modelProfile: requested.modelProfile, + modelId: requested.modelId, + ...result, + }, + }, + ], + { activity: "preserve" }, + ); + await state.delete(snapshotKey(message)); + if (result.outcome === "failed") { + setSpanAttributes({ "error.type": result.errorType }); + } + }; + + const remainingMs = request.createdAtMs + REQUEST_TIMEOUT_MS - Date.now(); + if (remainingMs <= 0) { + await writePrepared({ outcome: "failed", errorType: "timeout" }); + return "failed"; + } + const snapshot = await state.get(snapshotKey(message)); + if (!snapshot) { + await writePrepared({ outcome: "failed", errorType: "snapshot_missing" }); + return "failed"; + } + setSpanAttributes({ "app.compaction.input.size": snapshot.inputSize }); + const projection = projectConversationEvents(events, { + defaultProfile: botConfig.defaultProfile, + maxSeq: message.markSeq, + }); + + const signal = AbortSignal.timeout(remainingMs); + let summary: string; + try { + summary = await runWithConversationPrivacy( + snapshot.conversationPrivacy, + async () => { + const summaryStreamFn = createTracedStreamFn({ + conversationPrivacy: snapshot.conversationPrivacy, + promptName: CONTEXT_COMPACTION_PROMPT_NAME, + ...(streamFn ? { base: streamFn } : undefined), + }); + // The live agent sets only these request options. Keep them equal, + // or the request prefix differs and the provider cache misses. + const agent = new Agent({ + getApiKey: getGatewayApiKey, + sessionId: message.conversationId, + streamFn: summaryStreamFn, + initialState: { + systemPrompt: snapshot.systemPrompt, + model: resolveGatewayModel(snapshot.modelId), + thinkingLevel: snapshot.thinkingLevel, + tools: snapshot.tools.map((tool) => ({ + ...tool, + execute: async () => { + throw new Error("Background compaction does not run tools"); + }, + })), + }, + }); + return await requestContextSummary( + await buildAgentModelRequest(agent, projection.messages, signal), + summaryStreamFn, + ); + }, + ); + } catch (error) { + await writePrepared({ + outcome: "failed", + errorType: failureLabel(error, signal), + }); + return "failed"; + } + await writePrepared({ outcome: "ready", summary }); + setSpanAttributes({ + "app.compaction.summary_chars": summary.length, + "app.compaction.ready_ms": Date.now() - request.createdAtMs, + }); + return "ready"; +} diff --git a/packages/junior/src/chat/services/context-budget.ts b/packages/junior/src/chat/services/context-budget.ts index 5ceba96489..cd7d772c7c 100644 --- a/packages/junior/src/chat/services/context-budget.ts +++ b/packages/junior/src/chat/services/context-budget.ts @@ -4,6 +4,9 @@ import { resolveGatewayModel } from "@/chat/pi/client"; const COMPACTION_TRIGGER_RATIO = 0.9; const CONTEXT_INPUT_LIMIT_RATIO = 0.95; const COMPACTION_TARGET_RATIO = 0.8; +// Share of the compaction threshold where a background summary starts, so it +// is ready before blocking compaction would run. +const BACKGROUND_COMPACTION_MARK_RATIO = 0.7; const FALLBACK_CONTEXT_WINDOW_TOKENS = 400_000; export interface ModelContextBudget { @@ -72,6 +75,22 @@ export function getAgentContextCompactionTriggerTokens( ); } +/** + * Resolve the input size where a Turn requests a background summary for the + * active agent model. It is below the compaction threshold of the same budget. + */ +export function getAgentBackgroundCompactionMarkTokens( + modelId: string, +): number { + return Math.max( + 1, + Math.floor( + getAgentContextCompactionTriggerTokens(modelId) * + BACKGROUND_COMPACTION_MARK_RATIO, + ), + ); +} + /** Resolve the hard input ceiling for the active agent model. */ export function getAgentContextInputLimitTokens(modelId: string): number { return calculateContextInputLimitTokens(getModelContextBudget(modelId)); diff --git a/packages/junior/src/chat/services/context-compaction-telemetry.ts b/packages/junior/src/chat/services/context-compaction-telemetry.ts index 95d84aa085..e4397f0c31 100644 --- a/packages/junior/src/chat/services/context-compaction-telemetry.ts +++ b/packages/junior/src/chat/services/context-compaction-telemetry.ts @@ -15,9 +15,10 @@ import { withSpan, type SetSpanAttributes } from "@/chat/logging"; /** * Where a compaction runs: `blocking` at Turn start, `active` between model - * calls in a Turn, `handoff` for a model handoff. + * calls in a Turn, `handoff` for a model handoff, `background` in a job after + * a Turn. */ -export type CompactionMode = "blocking" | "active" | "handoff"; +export type CompactionMode = "blocking" | "active" | "handoff" | "background"; interface TurnCompactionWait { ms: number; @@ -41,7 +42,7 @@ export async function withCompactionSpan( ), ); } finally { - // Every mode runs on the critical path of the Turn today. + // A background job runs outside a Turn, so it has no Turn wait. const wait = turnCompactionWait.getStore(); if (wait) { wait.ms += performance.now() - startedAtMs; diff --git a/packages/junior/src/deployment.ts b/packages/junior/src/deployment.ts index 07c9bc3ed5..511b535cac 100644 --- a/packages/junior/src/deployment.ts +++ b/packages/junior/src/deployment.ts @@ -12,6 +12,8 @@ export const JUNIOR_CONVERSATION_WORK_CALLBACK_ROUTE = export const JUNIOR_PLUGIN_TASK_CALLBACK_ROUTE = "/api/internal/plugin/tasks"; export const JUNIOR_WORKSPACE_SNAPSHOT_JOB_CALLBACK_ROUTE = "/api/internal/workspace/snapshots"; +export const JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE = + "/api/internal/context/compactions"; export const LEGACY_JUNIOR_CONVERSATION_WORK_FUNCTION = "api/internal/agent/continue.ts"; diff --git a/packages/junior/src/nitro.ts b/packages/junior/src/nitro.ts index 357a1de1bb..2b6f5f50dc 100644 --- a/packages/junior/src/nitro.ts +++ b/packages/junior/src/nitro.ts @@ -14,6 +14,7 @@ import { } from "@/build/virtual-config"; import { PLUGIN_TASK_QUEUE_TOPIC } from "@/chat/plugins/task-queue"; import { WORKSPACE_SNAPSHOT_JOB_QUEUE_TOPIC } from "@/chat/sandbox/snapshot/job-queue"; +import { BACKGROUND_COMPACTION_QUEUE_TOPIC } from "@/chat/services/background-compaction-queue"; import { resolveConversationWorkQueueTopic } from "@/chat/task-execution/vercel-queue"; import { JUNIOR_CONVERSATION_WORK_CALLBACK_ROUTE, @@ -23,6 +24,7 @@ import { JUNIOR_RETENTION_CRON_SCHEDULE, JUNIOR_RETENTION_ROUTE, JUNIOR_WORKSPACE_SNAPSHOT_JOB_CALLBACK_ROUTE, + JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE, getDeploymentServiceVersion, } from "@/deployment"; import { @@ -260,6 +262,32 @@ function configureVercelDeployment( ], }; + const existingCompactionRule = + nitro.options.vercel.functionRules[ + JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE + ] ?? {}; + const existingCompactionTriggers = Array.isArray( + existingCompactionRule.experimentalTriggers, + ) + ? existingCompactionRule.experimentalTriggers + : []; + const otherCompactionTriggers = existingCompactionTriggers.filter( + (trigger) => trigger.type !== VERCEL_QUEUE_TRIGGER_TYPE, + ); + + nitro.options.vercel.functionRules[JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE] = + { + ...existingCompactionRule, + maxDuration: defaultMaxDuration, + experimentalTriggers: [ + ...otherCompactionTriggers, + { + type: VERCEL_QUEUE_TRIGGER_TYPE, + topic: BACKGROUND_COMPACTION_QUEUE_TOPIC, + }, + ], + }; + return defaultMaxDuration; } diff --git a/packages/junior/tests/integration/background-compaction-events.test.ts b/packages/junior/tests/integration/background-compaction-events.test.ts index c6c238b775..7ddf03720d 100644 --- a/packages/junior/tests/integration/background-compaction-events.test.ts +++ b/packages/junior/tests/integration/background-compaction-events.test.ts @@ -1,5 +1,22 @@ +import { Agent, type StreamFn } from "@earendil-works/pi-agent-core"; +import { + createAssistantMessageEventStream, + type AssistantMessage, + type Context, +} from "@earendil-works/pi-ai"; +import { Type } from "@sinclair/typebox"; import { afterEach, describe, expect, it } from "vitest"; +import { buildAgentModelRequest } from "@/chat/agent/context-summary"; +import { botConfig } from "@/chat/config"; import { getConversationEventStore } from "@/chat/db"; +import { defaultModelId } from "@/chat/model-profile"; +import { getGatewayApiKey, resolveGatewayModel } from "@/chat/pi/client"; +import { + processBackgroundCompaction, + requestBackgroundCompaction, +} from "@/chat/services/background-compaction"; +import type { BackgroundCompactionMessage } from "@/chat/services/background-compaction-queue"; +import { getAgentBackgroundCompactionMarkTokens } from "@/chat/services/context-budget"; import type { HistoryReplacement } from "@/chat/conversations/history"; import { openConversationProjection } from "@/chat/conversations/projection"; import { contextProvenance } from "@/chat/conversations/provenance"; @@ -143,3 +160,206 @@ describe("background compaction events", () => { ]); }); }); + +const MODEL_ID = defaultModelId(botConfig); + +function assistant(text: string, input = 0): AssistantMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "anthropic-messages", + provider: "vercel-ai-gateway", + model: MODEL_ID, + usage: { + input, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: input, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: 0, + }; +} + +/** A model stream that answers each call with the next text. */ +function replies(texts: string[]): { calls: Context[]; streamFn: StreamFn } { + const calls: Context[] = []; + const streamFn: StreamFn = (_model, context) => { + const message = assistant(texts[calls.length]!); + calls.push({ ...context, messages: [...context.messages] }); + const stream = createAssistantMessageEventStream(); + queueMicrotask(() => { + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + }); + return stream; + }; + return { calls, streamFn }; +} + +/** The live agent at the end of a Turn whose last request had this input. */ +function liveAgent(inputSize: number): Agent { + return new Agent({ + getApiKey: getGatewayApiKey, + sessionId: CONVERSATION_ID, + streamFn: () => { + throw new Error("The Turn is complete"); + }, + initialState: { + model: resolveGatewayModel(MODEL_ID), + systemPrompt: "You are Junior.", + thinkingLevel: "low", + tools: [ + { + name: "lookup", + label: "Lookup", + description: "Look up a record.", + parameters: Type.Object({ id: Type.String() }), + execute: async () => ({ content: [], details: undefined }), + }, + ], + messages: [assistant("The blue option is ready.", inputSize)], + }, + }); +} + +async function completeTurn( + agent: Agent, +): Promise { + const sent: BackgroundCompactionMessage[] = []; + await requestBackgroundCompaction({ + agent, + conversationId: CONVERSATION_ID, + conversationPrivacy: "private", + modelId: MODEL_ID, + modelProfile: botConfig.defaultProfile, + queue: { send: async (message) => void sent.push(message) }, + }); + return sent; +} + +async function bookkeeping() { + const history = + await getConversationEventStore().loadCurrentHistory(CONVERSATION_ID); + return history.flatMap((event) => + event.data.type === "compaction_requested" || + event.data.type === "compaction_prepared" + ? [event.data] + : [], + ); +} + +describe("background compaction job", () => { + afterEach(closeConversationFixture); + + it("prepares a summary with the request prefix of the last agent request", async () => { + const [first] = await getConversationEventStore().append(CONVERSATION_ID, [ + { createdAtMs: 1, data: userMessage("Remember the blue option.", 1) }, + ]); + const mark = getAgentBackgroundCompactionMarkTokens(MODEL_ID); + + await expect(completeTurn(liveAgent(mark - 1))).resolves.toEqual([]); + const agent = liveAgent(mark); + const sent = await completeTurn(agent); + expect(sent).toEqual([ + { + conversationId: CONVERSATION_ID, + historyVersion: 0, + markSeq: first!.seq, + }, + ]); + + const model = replies(["The person picked blue."]); + await processBackgroundCompaction(sent[0]!, { streamFn: model.streamFn }); + + // The provider reads the same system prompt and tools, so it can read + // the cached prefix. Only the summary instruction is added. + const expected = await buildAgentModelRequest(agent, [ + { + role: "user", + content: [{ type: "text", text: "Remember the blue option." }], + timestamp: 1, + }, + ]); + expect(model.calls).toHaveLength(1); + expect(model.calls[0]!.systemPrompt).toBe(expected.context.systemPrompt); + expect(JSON.stringify(model.calls[0]!.tools)).toBe( + JSON.stringify(expected.context.tools), + ); + expect(model.calls[0]!.messages.slice(0, -1)).toEqual( + expected.context.messages, + ); + const request = { + markSeq: first!.seq, + historyVersion: 0, + modelProfile: botConfig.defaultProfile, + modelId: MODEL_ID, + }; + await expect(bookkeeping()).resolves.toEqual([ + { type: "compaction_requested", ...request }, + { + type: "compaction_prepared", + ...request, + outcome: "ready", + summary: "The person picked blue.", + }, + ]); + + // One ready summary is enough for the History version. + await expect(completeTurn(liveAgent(mark))).resolves.toEqual([]); + // A repeated delivery writes nothing. + await processBackgroundCompaction(sent[0]!, { streamFn: model.streamFn }); + expect(model.calls).toHaveLength(1); + }); + + it("records a fixed failure label and stops after two failed attempts", async () => { + const events = getConversationEventStore(); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 1, data: userMessage("First question.", 1) }, + ]); + const mark = getAgentBackgroundCompactionMarkTokens(MODEL_ID); + + const [firstJob] = await completeTurn(liveAgent(mark)); + await processBackgroundCompaction(firstJob!, { + streamFn: replies(["No tags.", "Still no tags."]).streamFn, + }); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 2, data: userMessage("Second question.", 2) }, + ]); + const [secondJob] = await completeTurn(liveAgent(mark)); + await processBackgroundCompaction(secondJob!, { + streamFn: replies(["No tags.", "Still no tags."]).streamFn, + }); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 3, data: userMessage("Third question.", 3) }, + ]); + + await expect(completeTurn(liveAgent(mark))).resolves.toEqual([]); + const prepared = (await bookkeeping()).filter( + (event) => event.type === "compaction_prepared", + ); + expect(prepared).toMatchObject([ + { outcome: "failed", errorType: "missing_summary" }, + { outcome: "failed", errorType: "missing_summary" }, + ]); + }); + + it("writes nothing when the history was replaced before the job ran", async () => { + const events = getConversationEventStore(); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 1, data: userMessage("First question.", 1) }, + ]); + const [job] = await completeTurn( + liveAgent(getAgentBackgroundCompactionMarkTokens(MODEL_ID)), + ); + await events.replaceHistory(CONVERSATION_ID, compaction("Foreground.")); + + const model = replies(["Late."]); + await processBackgroundCompaction(job!, { streamFn: model.streamFn }); + + expect(model.calls).toHaveLength(0); + await expect(bookkeeping()).resolves.toEqual([]); + }); +}); diff --git a/packages/junior/tests/unit/build/nitro-plugin-module.test.ts b/packages/junior/tests/unit/build/nitro-plugin-module.test.ts index 97fd82aaa8..1bdde29624 100644 --- a/packages/junior/tests/unit/build/nitro-plugin-module.test.ts +++ b/packages/junior/tests/unit/build/nitro-plugin-module.test.ts @@ -4,8 +4,10 @@ import path from "node:path"; import { defineJuniorPlugin } from "@sentry/junior-plugin-api"; import { afterEach, describe, expect, it } from "vitest"; import { PLUGIN_TASK_QUEUE_TOPIC } from "@/chat/plugins/task-queue"; +import { BACKGROUND_COMPACTION_QUEUE_TOPIC } from "@/chat/services/background-compaction-queue"; import { DEFAULT_CONVERSATION_WORK_QUEUE_TOPIC } from "@/chat/task-execution/vercel-queue"; import { + JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE, JUNIOR_CONVERSATION_WORK_CALLBACK_ROUTE, JUNIOR_HEARTBEAT_CRON_SCHEDULE, JUNIOR_HEARTBEAT_ROUTE, @@ -127,6 +129,17 @@ describe("juniorNitro plugin modules", () => { }, ], }); + expect( + vercel.functionRules?.[JUNIOR_CONTEXT_COMPACTION_CALLBACK_ROUTE], + ).toEqual({ + maxDuration: 300, + experimentalTriggers: [ + { + type: "queue/v2beta", + topic: BACKGROUND_COMPACTION_QUEUE_TOPIC, + }, + ], + }); }); it("preserves existing Vercel route function settings", () => { From 28dc7cbd3e3b5a527720d48fcd4ee3025dc1c794 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:15:53 +0000 Subject: [PATCH 2/2] fix(compaction): Request again after a Turn passes over a ready summary A Turn swaps a ready summary in before its first model request, which makes a new History version. A ready summary of the current version that predates the Turn was discarded (over target, split tool call, or another Model profile), so it no longer blocks a new request. It counts toward the attempt limit. Drop the Model profile guard on requests: a handoff already makes a new History version, and the request records the Turn's profile. --- packages/junior/src/chat/agent/index.ts | 1 + .../chat/services/background-compaction.ts | 54 ++++++++++++------- .../background-compaction-events.test.ts | 43 ++++++++++++++- 3 files changed, 78 insertions(+), 20 deletions(-) diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index 6f0ea5cc03..6ab632dd09 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -1797,6 +1797,7 @@ async function executeAgentRunInPrivacyContext( modelId: activeModelId, modelProfile: activeModelProfile, 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 diff --git a/packages/junior/src/chat/services/background-compaction.ts b/packages/junior/src/chat/services/background-compaction.ts index 906a83f7cd..342a8fbf4f 100644 --- a/packages/junior/src/chat/services/background-compaction.ts +++ b/packages/junior/src/chat/services/background-compaction.ts @@ -53,7 +53,10 @@ import { type BackgroundCompactionQueue, } from "./background-compaction-queue"; -/** Failed attempts for one History version before background tries stop. */ +/** + * Failed or discarded attempts for one History version before background + * tries stop. + */ const MAX_FAILED_ATTEMPTS = 2; /** * A request without a prepared result after this long counts as a failed @@ -95,22 +98,34 @@ function lastRequestInputSize(messages: AgentMessage[]): number | undefined { /** * Return whether a Turn may request a background summary for this History - * version. It may not while a request is pending, when a summary is ready, or - * after the failure limit. A ready summary stays usable for the version, so - * one is enough. + * version. It may not while a request is pending, while a ready summary is + * still unused, or after the attempt limit. + * + * A Turn swaps a ready summary in before its first model request. A swap + * makes a new History version. So a ready summary of this version that was + * prepared before the Turn started was discarded, for example because the + * history after its mark is too large now. It counts as a failed attempt, so + * a new request can replace it. */ function mayRequest( events: ConversationEvent[], historyVersion: number, nowMs: number, + turnStartedAtMs: number, ): boolean { - const outcomes = new Map(); + const outcomes = new Map< + number, + { outcome: "ready" | "failed"; createdAtMs: number } + >(); for (const event of events) { if ( event.data.type === "compaction_prepared" && event.data.historyVersion === historyVersion ) { - outcomes.set(event.data.markSeq, event.data.outcome); + outcomes.set(event.data.markSeq, { + outcome: event.data.outcome, + createdAtMs: event.createdAtMs, + }); } } let failed = 0; @@ -121,14 +136,14 @@ function mayRequest( ) { continue; } - const outcome = outcomes.get(event.data.markSeq); - if (outcome === "ready") { - return false; - } + const prepared = outcomes.get(event.data.markSeq); if ( - outcome === "failed" || - nowMs - event.createdAtMs >= REQUEST_TIMEOUT_MS + prepared?.outcome === "ready" && + prepared.createdAtMs >= turnStartedAtMs ) { + return false; + } + if (prepared || nowMs - event.createdAtMs >= REQUEST_TIMEOUT_MS) { failed += 1; } else { return false; @@ -141,7 +156,8 @@ function mayRequest( * Request a background summary when the last model request of a completed * Turn was at or above the mark. * - * Without a queue or below the mark, it does nothing. A failure here leaves + * `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. @@ -153,6 +169,7 @@ export async function requestBackgroundCompaction(args: { modelId: string; modelProfile: ModelProfile; queue: BackgroundCompactionQueue | undefined; + turnStartedAtMs: number; }): Promise { if (!args.queue) { return; @@ -181,6 +198,7 @@ async function requestAboveMark(args: { modelId: string; modelProfile: ModelProfile; queue: BackgroundCompactionQueue; + turnStartedAtMs: number; }): Promise { const nowMs = Date.now(); const eventStore = getConversationEventStore(); @@ -188,16 +206,14 @@ async function requestAboveMark(args: { const historyVersion = events[0]?.historyVersion; if ( historyVersion === undefined || - !mayRequest(events, historyVersion, nowMs) + !mayRequest(events, historyVersion, nowMs, args.turnStartedAtMs) ) { return; } - const projection = projectConversationEvents(events, { + const markSeq = projectConversationEvents(events, { defaultProfile: botConfig.defaultProfile, - }); - const markSeq = projection.seqs.at(-1); - // After a handoff, the history belongs to another Model profile. - if (markSeq === undefined || projection.modelProfile !== args.modelProfile) { + }).seqs.at(-1); + if (markSeq === undefined) { return; } diff --git a/packages/junior/tests/integration/background-compaction-events.test.ts b/packages/junior/tests/integration/background-compaction-events.test.ts index 7ddf03720d..d9ecda66b0 100644 --- a/packages/junior/tests/integration/background-compaction-events.test.ts +++ b/packages/junior/tests/integration/background-compaction-events.test.ts @@ -225,8 +225,13 @@ function liveAgent(inputSize: number): Agent { }); } +/** + * Complete a Turn. The default start is before every event, so no ready + * summary was passed over at its start. + */ async function completeTurn( agent: Agent, + turnStartedAtMs = 0, ): Promise { const sent: BackgroundCompactionMessage[] = []; await requestBackgroundCompaction({ @@ -236,6 +241,7 @@ async function completeTurn( modelId: MODEL_ID, modelProfile: botConfig.defaultProfile, queue: { send: async (message) => void sent.push(message) }, + turnStartedAtMs, }); return sent; } @@ -307,7 +313,7 @@ describe("background compaction job", () => { }, ]); - // One ready summary is enough for the History version. + // A summary that became ready during the Turn waits for the next Turn. await expect(completeTurn(liveAgent(mark))).resolves.toEqual([]); // A repeated delivery writes nothing. await processBackgroundCompaction(sent[0]!, { streamFn: model.streamFn }); @@ -346,6 +352,41 @@ describe("background compaction job", () => { ]); }); + it("requests again when a Turn started after a ready summary and did not swap it in", async () => { + const events = getConversationEventStore(); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 1, data: userMessage("First question.", 1) }, + ]); + const mark = getAgentBackgroundCompactionMarkTokens(MODEL_ID); + const ready = async (job: BackgroundCompactionMessage | undefined) => + await processBackgroundCompaction(job!, { + streamFn: replies(["Earlier work."]).streamFn, + }); + + await ready((await completeTurn(liveAgent(mark)))[0]); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 2, data: userMessage("Second question.", 2) }, + ]); + // The History version is the same, so this Turn discarded the summary, + // for example because the history after its mark is too large now. + const [second] = await completeTurn(liveAgent(mark), Date.now()); + expect(second).toBeDefined(); + await ready(second); + await events.append(CONVERSATION_ID, [ + { createdAtMs: 3, data: userMessage("Third question.", 3) }, + ]); + + // Two passed-over summaries reach the attempt limit. + await expect(completeTurn(liveAgent(mark), Date.now())).resolves.toEqual( + [], + ); + expect( + (await bookkeeping()).filter( + (event) => event.type === "compaction_requested", + ), + ).toHaveLength(2); + }); + it("writes nothing when the history was replaced before the job ran", async () => { const events = getConversationEventStore(); await events.append(CONVERSATION_ID, [