feat(compaction): Prepare a summary in a background job

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_<kind>`,
+  `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[<key>,number]` form.
 

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

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<void>,
   ) => { send(message: PluginTaskQueueMessage): Promise<void> };
+  /**
+   * 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<void>,
+  ) => 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<typeof createVercelBackgroundCompactionCallback>
+    | 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,
       ),

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) {

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,15 @@ async function executeAgentRunInPrivacyContext(
     }
 
     await recordActiveMcpProviders();
+    await requestBackgroundCompaction({
+      agent,
+      conversationId,
+      conversationPrivacy,
+      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
     // completed session record after assistant output handling settles, so a

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<string, unknown>;
   /** Queue of the app, for tools that wake other Conversations. */
   conversationWorkQueue?: ConversationWorkQueue;

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<AgentRunOutcome>,
   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:

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<Response> {
+  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();
+}

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<void>;
+}
+
+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,
+      },
+    );
+  },
+};