fix(core): compact before inference body limit

packages/core/src/session/compaction.ts

@@ -17,6 +17,9 @@ import { Token } from "../util/token"
 const DEFAULT_BUFFER = 20_000
 const DEFAULT_KEEP_TOKENS = 8_000
 const TOOL_OUTPUT_MAX_CHARS = 2_000
+const COMPACTION_CHUNK_TOKENS = 32_000
+const SUMMARY_OUTPUT_TOKENS = 4_096
+const REQUEST_BODY_COMPACTION_BYTES = 8 * 1024 * 1024
 const SUMMARY_TEMPLATE = `Output exactly the Markdown structure shown inside <template> and keep the section order unchanged. Do not include the <template> tags in your response.
 <template>
 ## Objective
@@ -68,6 +71,7 @@ export type AutoInput = {
   readonly sessionID: SessionSchema.ID
   readonly messages: readonly SessionMessage.Info[]
   readonly model: Model
+  readonly requestBytes: number
 }
 
 export type ManualInput = {
@@ -80,7 +84,8 @@ type Plan = {
   readonly sessionID: SessionSchema.ID
   readonly model: Model
   readonly reason: SessionMessage.Compaction["reason"]
-  readonly prompt: string
+  readonly previousSummary?: string
+  readonly context: readonly string[]
   readonly recent: string
   readonly inputID?: SessionMessage.ID
 }
@@ -199,6 +204,14 @@ export const buildPrompt = (input: { readonly previousSummary?: string; readonly
     ...input.context,
   ].join("\n\n")
 
+const chunkContext = (context: readonly string[]) => {
+  const value = context.filter(Boolean).join("\n\n")
+  const size = COMPACTION_CHUNK_TOKENS * 4
+  return Array.from({ length: Math.ceil(value.length / size) }, (_, index) =>
+    value.slice(index * size, (index + 1) * size),
+  )
+}
+
 const planContent = (messages: readonly SessionMessage.Info[], tokens: number) => {
   const selected = select(messages, tokens)
   if (!selected) return
@@ -208,10 +221,8 @@ const planContent = (messages: readonly SessionMessage.Info[], tokens: number) =
   const previousRecent = previousSummary?.type === "compaction" ? previousSummary.recent : ""
   const summarizeRecent = !previousRecent && !selected.head
   return {
-    prompt: buildPrompt({
+    previousSummary: previousSummary?.type === "compaction" ? previousSummary.summary : undefined,
-      previousSummary: previousSummary?.type === "compaction" ? previousSummary.summary : undefined,
+    context: chunkContext(summarizeRecent ? [selected.recent] : [previousRecent, selected.head]),
-      context: summarizeRecent ? [selected.recent] : [previousRecent, selected.head].filter(Boolean),
-    }),
     recent: summarizeRecent ? "" : selected.recent,
   }
 }
@@ -235,62 +246,67 @@ const make = (dependencies: Dependencies) => {
       inputID: plan.inputID,
     })
 
-    const chunks: string[] = []
+    let summary = plan.previousSummary
-    let failure: SessionError.Error | undefined
+    for (const [index, context] of plan.context.entries()) {
-    yield* dependencies.llm
+      const chunks: string[] = []
-      .stream(
+      let failure: SessionError.Error | undefined
-        LLM.request({
+      const publish = index === plan.context.length - 1
-          model: plan.model,
+      yield* dependencies.llm
-          messages: [Message.user(plan.prompt)],
+        .stream(
-          tools: [],
+          LLM.request({
-        }),
+            model: plan.model,
-      )
+            messages: [Message.user(buildPrompt({ previousSummary: summary, context: [context] }))],
-      .pipe(
+            tools: [],
-        Stream.runForEach((event) => {
+            generation: { maxTokens: SUMMARY_OUTPUT_TOKENS },
-          if (LLMEvent.is.providerError(event))
-            failure = {
-              type: event.classification === "context-overflow" ? "provider.invalid-request" : "provider.error",
-              message: event.message,
-            }
-          if (LLMEvent.is.textDelta(event)) {
-            chunks.push(event.text)
-            return dependencies.events.publish(SessionEvent.Compaction.Delta, {
-              sessionID: plan.sessionID,
-              text: event.text,
-            })
-          }
-          return Effect.void
-        }),
-        Effect.catchTag("LLM.Error", (error) =>
-          Effect.sync(() => {
-            failure = toSessionError(error)
           }),
-        ),
+        )
-        Effect.onInterrupt(() =>
+        .pipe(
-          plan.reason === "auto"
+          Stream.runForEach((event) => {
-            ? failed({
+            if (LLMEvent.is.providerError(event))
-                sessionID: plan.sessionID,
+              failure = {
-                reason: plan.reason,
+                type: event.classification === "context-overflow" ? "provider.invalid-request" : "provider.error",
-                error: { type: "compaction.interrupted", message: "Compaction was interrupted" },
+                message: event.message,
-                inputID: plan.inputID,
+              }
-              }).pipe(Effect.asVoid)
+            if (LLMEvent.is.textDelta(event)) {
-            : Effect.void,
+              chunks.push(event.text)
-        ),
+              if (publish)
-      )
+                return dependencies.events.publish(SessionEvent.Compaction.Delta, {
-    const summary = chunks.join("")
+                  sessionID: plan.sessionID,
-    if (failure || !summary.trim()) {
+                  text: event.text,
-      const error = failure ?? { type: "compaction.failed" as const, message: "Compaction produced no summary" }
+                })
-      return yield* failed({
+            }
-        sessionID: plan.sessionID,
+            return Effect.void
-        reason: plan.reason,
+          }),
-        error,
+          Effect.catchTag("LLM.Error", (error) =>
-        inputID: plan.inputID,
+            Effect.sync(() => {
-      })
+              failure = toSessionError(error)
+            }),
+          ),
+          Effect.onInterrupt(() =>
+            plan.reason === "auto"
+              ? failed({
+                  sessionID: plan.sessionID,
+                  reason: plan.reason,
+                  error: { type: "compaction.interrupted", message: "Compaction was interrupted" },
+                  inputID: plan.inputID,
+                }).pipe(Effect.asVoid)
+              : Effect.void,
+          ),
+        )
+      const next = chunks.join("")
+      if (failure || !next.trim())
+        return yield* failed({
+          sessionID: plan.sessionID,
+          reason: plan.reason,
+          error: failure ?? { type: "compaction.failed", message: "Compaction produced no summary" },
+          inputID: plan.inputID,
+        })
+      summary = next
     }
     yield* dependencies.events.publish(SessionEvent.Compaction.Ended, {
       sessionID: plan.sessionID,
       reason: plan.reason,
-      text: summary,
+      text: summary ?? "",
       recent: plan.recent,
     })
     return { status: "completed" as const }
@@ -313,6 +329,7 @@ const make = (dependencies: Dependencies) => {
   })
   const required = (input: AutoInput) => {
     if (!config.auto) return false
+    if (input.requestBytes >= REQUEST_BODY_COMPACTION_BYTES) return true
     const context = input.model.route.defaults.limits?.context
     if (context === undefined || context <= 0) return false
     const last = input.messages.findLast(

packages/core/src/session/runner/llm.ts

@@ -186,12 +186,6 @@ const layer = Layer.effect(
       const providerMetadataKey = model.route.providerMetadataKey ?? model.provider
       const history = yield* SessionHistory.entriesForRunner(db, session.id, instructions)
       const context = history.entries.map((entry) => entry.message)
-      const compactionInput = { sessionID: session.id, messages: context, model }
-      if (compaction.required(compactionInput) && !(yield* SessionPending.compaction(db, session.id))) {
-        const compacted = yield* compaction.compact(compactionInput)
-        if (compacted.status === "completed") return { _tag: "RestartAfterCompaction", step: currentStep } as const
-        return yield* new StepFailedError({ error: compacted.error })
-      }
       const isLastStep = agentInfo.steps !== undefined && currentStep >= agentInfo.steps
       const toolMaterialization = isLastStep ? undefined : yield* tools.materialize(agentInfo.permissions)
       const promptCacheKey = /^ses_[0-9a-f]{64}$/.test(session.id) ? session.id.slice(4) : session.id
@@ -208,6 +202,19 @@ const layer = Layer.effect(
         tools: toolMaterialization?.definitions ?? [],
         toolChoice: isLastStep ? "none" : undefined,
       })
+      const compactionInput = {
+        sessionID: session.id,
+        messages: context,
+        model,
+        requestBytes: new TextEncoder().encode(
+          JSON.stringify({ system: request.system, messages: request.messages, tools: request.tools }),
+        ).length,
+      }
+      if (compaction.required(compactionInput) && !(yield* SessionPending.compaction(db, session.id))) {
+        const compacted = yield* compaction.compact(compactionInput)
+        if (compacted.status === "completed") return { _tag: "RestartAfterCompaction", step: currentStep } as const
+        return yield* new StepFailedError({ error: compacted.error })
+      }
       const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>()
       const ownedToolFibers: Array<Fiber.Fiber<void, ToolOutputStore.Error>> = []
       let needsContinuation = false
@@ -323,8 +330,7 @@ const layer = Layer.effect(
             recoverOverflow &&
             !publisher.hasRetryEvidence() &&
             isContextOverflowFailure(overflowFailure ?? streamFailure) &&
-            (yield* restore(recoverOverflow({ sessionID: session.id, messages: context, model }))).status ===
+            (yield* restore(recoverOverflow(compactionInput))).status === "completed"
-              "completed"
           )
             return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
 

packages/core/test/session-compaction.test.ts

@@ -96,6 +96,20 @@ test("compaction prompt requires the checkpoint headings in order", () => {
   expect(prompt).toContain("Keep every section, even when empty.")
 })
 
+it.effect("requires compaction before a request reaches the inference body limit", () =>
+  Effect.gen(function* () {
+    const compaction = yield* SessionCompaction.Service
+    expect(
+      compaction.required({
+        sessionID: SessionV2.ID.make("ses_request_body_compaction"),
+        messages: [],
+        model,
+        requestBytes: 8 * 1024 * 1024,
+      }),
+    ).toBe(true)
+  }),
+)
+
 it.effect("manual compaction summarizes short context instead of no-op", () =>
   Effect.gen(function* () {
     requests = []
@@ -151,7 +165,7 @@ it.effect("manual compaction summarizes short context instead of no-op", () =>
     expect(Array.from(yield* Fiber.join(delta)).map((event) => event.data.text)).toEqual(["manual summary"])
 
     expect(requests).toHaveLength(1)
-    expect(requests[0]?.generation).toBeUndefined()
+    expect(requests[0]?.generation?.maxTokens).toBe(4_096)
     expect(JSON.stringify(requests[0]?.messages)).toContain("Manual compaction should include this short conversation.")
     expect(yield* store.context(sessionID)).toMatchObject([
       { type: "compaction", reason: "manual", summary: "manual summary", recent: "" },
@@ -170,3 +184,53 @@ it.effect("manual compaction summarizes short context instead of no-op", () =>
     ])
   }),
 )
+
+it.effect("manual compaction rolls large histories through bounded requests", () =>
+  Effect.gen(function* () {
+    requests = []
+    const db = (yield* Database.Service).db
+    const compaction = yield* SessionCompaction.Service
+    const store = yield* SessionStore.Service
+    const sessionID = SessionV2.ID.make("ses_bounded_compaction")
+    yield* db
+      .insert(ProjectTable)
+      .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
+      .onConflictDoNothing()
+      .run()
+      .pipe(Effect.orDie)
+    yield* db
+      .insert(SessionTable)
+      .values({
+        id: sessionID,
+        project_id: Project.ID.global,
+        slug: "bounded-compaction",
+        directory: "/project",
+        title: "Bounded compaction",
+        version: "test",
+      })
+      .run()
+      .pipe(Effect.orDie)
+    const session = yield* store
+      .get(sessionID)
+      .pipe(Effect.flatMap((value) => (value ? Effect.succeed(value) : Effect.die("test session missing"))))
+
+    expect(
+      yield* compaction.compactManual({
+        session,
+        messages: [
+          {
+            id: SessionMessage.ID.create(),
+            type: "user",
+            text: "a".repeat(300_000),
+            time: { created: DateTime.makeUnsafe(0) },
+          },
+        ],
+        inputID: SessionMessage.ID.make("msg_bounded_compaction"),
+      }),
+    ).toEqual({ status: "completed" })
+
+    expect(requests.length).toBeGreaterThan(1)
+    expect(requests.every((request) => JSON.stringify(request.messages).length < 150_000)).toBe(true)
+    expect(JSON.stringify(requests[1]?.messages)).toContain("manual summary")
+  }),
+)