diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index 7f91fe9e8f50..623c0ec2c434 100644 --- a/packages/core/src/session/compaction.ts +++ b/packages/core/src/session/compaction.ts @@ -454,12 +454,13 @@ export const layer = Layer.effect( const prompt = buildPrompt(previous !== undefined, previous?.summary.includes(LEGACY_HEADING) ?? false) const headings = SUMMARY_TEMPLATE.split("\n").filter((line) => line.startsWith("##")) const filled = (text: string) => text.split("\n").some((line) => headings.includes(line.trim())) - const prepared = yield* prepare(context, messages, budget) + const build = (room?: number) => prepare(context, messages, budget, undefined, room, keepID) // Hooks saw the request without the summary prompt, so it is appended here. A reply that ignores the // template gets one reminder before it counts as a failure. - const send = (request: LLMRequest) => + const send = (prepared: Prepared) => Effect.gen(function* () { + const request = prepared.request const prompted = fitOutput( LLMRequest.update(request, { messages: [...request.messages, Message.user(prompt)] }), context.model.limit, @@ -492,10 +493,11 @@ export const layer = Layer.effect( const fit = trigger.reason === "overflow" ? budget : Math.max(budget, summaryRoom(context.model.limit)) const result = yield* deliver( trigger, - prepared, + () => prepare(context, messages, budget), + build, budget - overhead, send, - { fit: fit - overhead, history: withoutTail(prepared.request, keepID) }, + fit - overhead, background, ) if (!keepID) return result @@ -524,7 +526,7 @@ export const layer = Layer.effect( if (!context.messages.some(SessionRecentText.messageToText)) return yield* Effect.fail(NOTHING_TO_COMPACT) const unsupported = (message: string) => Effect.fail({ error: { type: "provider.unsupported-operation", message } }) - const prepared = yield* prepare(context, context.messages, budget, "session") + const build = (room?: number) => prepare(context, context.messages, budget, "session", room) const toResult = (window: ReadonlyArray, usage: Usage | undefined) => spend(context.session.id, usage && SessionUsage.record(usage, context.model.cost)).pipe( @@ -535,7 +537,8 @@ export const layer = Layer.effect( }), ) - const send = (request: LLMRequest) => { + const send = (prepared: Prepared) => { + const request = prepared.request if (LLMClient.canCompact(request, { mechanism: "trigger" })) { return Effect.gen(function* () { const history = yield* SessionHistory.load(db, context.session.id, "local").pipe(Effect.orDie) @@ -552,12 +555,13 @@ export const layer = Layer.effect( return unsupported(`Native compaction is not supported for ${request.model.provider}/${request.model.route.id}`) } - return yield* deliver(trigger, prepared, budget, send) + return yield* deliver(trigger, build, build, budget, send) }) /** - * Sends the request as-is if it is estimated to fit `shrink.fit`, else `shrink.history` as-is if it fits `target`, - * and otherwise `shrink.history` as text aimed at `target` (see `flattenAndDropOldest`). + * Sends the request as-is if it is estimated to fit `fit`, else rebuilds the history without the retained tail. + * That history is sent unchanged if it fits `target`, otherwise as text (see `flattenAndDropOldest`). + * Every changed draft passes through request hooks again; edits from earlier attempts do not accumulate. * Limits and estimates can be wrong, so until the provider rejects something as too long, a request that cannot * be made to fit is sent whole: unchanged, or as text once its media has to go. A "too long" rejection means the * estimate ran low, so the next attempts aim at 85%, 70%, 50%, then 35% of the first rejected request's estimate, @@ -569,56 +573,61 @@ export const layer = Layer.effect( */ const deliver = Effect.fnUntraced(function* ( trigger: Trigger, - prepared: Prepared, + initial: () => Effect.Effect, + build: (room?: number) => Effect.Effect, target: number, - send: (request: LLMRequest) => Effect.Effect, - shrink: { readonly fit: number; readonly history: LLMRequest } = { fit: target, history: prepared.request }, + send: (prepared: Prepared) => Effect.Effect, + fit = target, background?: Background, ): Effect.fn.Return { const context = trigger.context + let prepared = yield* initial() + if (!prepared) return yield* Effect.fail(NOTHING_TO_COMPACT) if (prepared.event.result) return yield* fromHook(context, prepared.event.result, background) const policy = yield* SessionRunnerRetry.policy(context.session.id) let rejections = 0 let rejected: number | undefined - let asText = false - let fit = shrink.fit + let forceText = false + let flattened = false + let rebuild = estimateRequest(prepared.request) > fit while (true) { - const unchanged = asText - ? undefined - : estimateRequest(prepared.request) <= fit - ? prepared.request - : estimateRequest(shrink.history) <= target - ? shrink.history - : undefined - const request = - unchanged ?? - flattenAndDropOldest(shrink.history, context, target) ?? - (rejections > 0 - ? undefined - : asText - ? flattenAndDropOldest(shrink.history, context, Number.POSITIVE_INFINITY) - : shrink.history) - if (!request) { - return yield* Effect.fail({ - error: { - type: "compaction.failed", - message: - "The compaction request cannot be reduced further without losing the latest exchange or checkpoint", - }, - }) + if (rebuild) { + const history = forceText ? undefined : yield* build() + if (history?.event.result) return yield* fromHook(context, history.event.result, background) + const unchanged = history && estimateRequest(history.request) <= target + // Reserve the latest hooked system overhead; flattened drafts omit tool definitions. + const overhead = estimateRequest({ ...(history ?? prepared).request, tools: [], messages: [] }) + const next = unchanged ? history : yield* build(target - overhead) + const candidate = + next ?? (rejections > 0 ? undefined : forceText ? yield* build(Number.POSITIVE_INFINITY) : history) + if (!candidate) { + return yield* Effect.fail({ + error: { + type: "compaction.failed", + message: + "The compaction request cannot be reduced further without losing the latest exchange or checkpoint", + }, + }) + } + prepared = candidate + flattened = (!unchanged && next !== undefined) || forceText + rebuild = false } + if (prepared.event.result) return yield* fromHook(context, prepared.event.result, background) + const request = prepared.request - const attempt = yield* Effect.result(send(request)) + const attempt = yield* Effect.result(send(prepared)) if (Result.isSuccess(attempt)) return attempt.success const cause = attempt.failure if (!(cause instanceof AIError)) return yield* Effect.fail(cause) const error = toSessionError(cause) const tooLarge = cause.reason._tag === "InvalidRequest" && cause.reason.classification === "payload-too-large" - if (tooLarge && (request === prepared.request || request === shrink.history)) { - asText = true + if (tooLarge && !flattened) { + forceText = true + rebuild = true continue } @@ -627,7 +636,7 @@ export const layer = Layer.effect( if (step === undefined) return yield* Effect.fail({ error }) rejected ??= estimateRequest(request) target = Math.floor(rejected * step) - fit = target + rebuild = true rejections++ continue } @@ -657,20 +666,19 @@ export const layer = Layer.effect( * detail is lost from the oldest history first and whole exchanges are omitted only as a last resort. The newest * exchange is never omitted; undefined when the transcript cannot fit even so. */ - const flattenAndDropOldest = (request: LLMRequest, context: SessionContext.Loaded, target: number) => { + const flattenAndDropOldest = (messages: ReadonlyArray, context: SessionContext.Loaded, budget: number) => { const previous = previousCompaction(context.messages) const end = - request.messages.findLastIndex( + messages.findLastIndex( (message) => (previous !== undefined && message.id === previous.id) || message.content.some((part) => part.type === "compaction"), ) + 1 const summary = previous !== undefined && !SessionHistory.isNativeCompaction(previous) ? previous : undefined - // TODO: Read the previous summary from the hooked request once it no longer merges summary and recent into one - // message. Until then it is rebuilt from storage, which drops a compaction hook's edits to that message. + // This is a fresh draft from history; request hooks run after it has been reduced. const lead = summary ? toLLMMessages([{ ...summary, metadata: { ...summary.metadata, recent: "" } }], context.model.ref) - : request.messages.slice(0, end) + : messages.slice(0, end) const recent = summary ? SessionRecentText.kept(summary) : "" // A user message and everything after it, up to the next one, is one exchange. Each is rendered at every level @@ -678,7 +686,7 @@ export const layer = Layer.effect( // shortened text: whole until it is omitted. const exchanges = [ ...(recent ? [TRANSCRIPT_LEVELS.map((level) => (level === OMITTED ? "" : recent))] : []), - ...request.messages + ...messages .slice(end) .reduce((groups, message) => { if (message.role === "user" || groups.length === 0) groups.push([message]) @@ -701,8 +709,8 @@ export const layer = Layer.effect( const size = (text: string) => (text ? Token.estimate(text) + 1 : 0) // Room for the longest possible note is set aside before choosing what to keep. const room = - target - - estimateRequest({ system: request.system, tools: [], messages: lead }) - + budget - + lead.reduce((sum, message) => sum + estimateMessage(message), 0) - Token.estimate(note(exchanges.length)) const levels = exchanges.map(() => 0) @@ -724,7 +732,7 @@ export const layer = Layer.effect( .map((renders, index) => renders[levels[index]]) .filter(Boolean) .join("\n\n") - return LLMRequest.update(request, { tools: [], messages: [...lead, Message.user(note(omitted) + text)] }) + return [...lead, Message.user(note(omitted) + text)] } const stream = ( @@ -793,11 +801,11 @@ export const layer = Layer.effect( } /** - * The conversation as the runner would send it, after request hooks. + * The conversation as the runner would send it, optionally reduced, then passed through request hooks. * * The output limit leaves room for `budget`, the most `deliver` sends. The request prepared here can be larger - * when the conversation overshot the threshold, and is only shrunk to fit after hooks have seen it, so sizing the - * output to it would leave next to no room. A prompt the estimate undersells is rejected and shrunk like any + * when the conversation overshot the threshold. Reduced attempts are rebuilt from history before hooks run, so + * sizing the output to the full conversation would leave next to no room. An underestimated prompt is shrunk like any * other. */ const prepare = ( @@ -805,13 +813,21 @@ export const layer = Layer.effect( messages: ReadonlyArray, budget: number, webSocket?: "session", + room?: number, + keepID?: SessionMessage.ID, ) => { const base = transcript(context, messages) + base.messages = withoutTail(base.messages, keepID) + if (room !== undefined) { + const reduced = flattenAndDropOldest(base.messages, context, room) + if (!reduced) return Effect.succeed(undefined) + base.messages = reduced + } return requests.compaction({ session: context.session, agent: context.agent.id, model: context.model, - tools: context.tools, + tools: room === undefined ? context.tools : undefined, system: base.system, messages: base.messages, webSocket, @@ -965,10 +981,9 @@ const splitConversation = (messages: ReadonlyArray, keep: n } /** The request without the retained tail, which starts at the message with ID `from`. */ -const withoutTail = (request: LLMRequest, from: SessionMessage.ID | undefined) => { - const index = from === undefined ? -1 : request.messages.findIndex((message) => message.id === from) - if (index < 0) return request - return LLMRequest.update(request, { messages: request.messages.slice(0, index) }) +const withoutTail = (messages: Message[], from: SessionMessage.ID | undefined) => { + const index = from === undefined ? -1 : messages.findIndex((message) => message.id === from) + return index < 0 ? messages : messages.slice(0, index) } const recentStart = ( diff --git a/packages/core/test/session-compaction.test.ts b/packages/core/test/session-compaction.test.ts index 50f0e1883f02..273881a4fdf8 100644 --- a/packages/core/test/session-compaction.test.ts +++ b/packages/core/test/session-compaction.test.ts @@ -1,5 +1,15 @@ import { expect, test } from "bun:test" -import { GenerationOptions, LLMClient, LLMEvent, LanguageModel, ToolDefinition, type LLMRequest } from "@opencode/ai/effect" +import { + AIError, + InvalidRequestError, + GenerationOptions, + LLMClient, + LLMEvent, + LanguageModel, + ToolDefinition, + type LLMRequest, + Message, +} from "@opencode/ai/effect" import { OpenAIChat } from "@opencode/ai/protocols" import { Database } from "@opencode/core/database/database" import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" @@ -85,24 +95,27 @@ const resolved = SessionRunnerModel.resolved(model, { cost, limit: { context: 200_000, output: 32_000 }, }) -const it = testEffect( - AppNodeBuilder.build( - LayerNode.group([ - Database.node, - Bus.node, - SessionProjector.node, - SessionStore.node, - SessionCompaction.node, - SessionModelRequest.node, - PluginHooks.node, - ]), - [ - Bus.node.replace(Bus.configured({ persist: true })), - llmClient.replace(client), - Location.node.replace(Location.boundNode({ directory: AbsolutePath.make("/project") })), - ], - ), -) +const suite = (replacement = client) => + testEffect( + AppNodeBuilder.build( + LayerNode.group([ + Database.node, + Bus.node, + SessionProjector.node, + SessionStore.node, + SessionCompaction.node, + SessionModelRequest.node, + PluginHooks.node, + ]), + [ + Bus.node.replace(Bus.configured({ persist: true })), + llmClient.replace(replacement), + Location.node.replace(Location.boundNode({ directory: AbsolutePath.make("/project") })), + ], + ), + ) + +const it = suite() test("compaction prompt preserves detailed work state and relevant files", () => { const prompt = SessionCompaction.buildPrompt(false) @@ -719,11 +732,22 @@ it.effect( }), ) + const hooks = yield* PluginHooks.Service + const drafts: number[] = [] + yield* hooks.register("session", "compaction", (event) => + Effect.sync(() => { + drafts.push(event.messages.length) + event.messages = [...event.messages, Message.user("Hook instruction")] + }), + ) + expect(yield* compactManually(session, messages)).toEqual({ status: "completed" }) expect(requests).toHaveLength(1) - expect(requests[0].messages).toHaveLength(2) + expect(drafts).toEqual([2, 1]) + expect(requests[0].messages).toHaveLength(3) expect(requests[0].messages[0].content).toEqual([{ type: "text", text: messages[0].text }]) - expect(requests[0].messages[1].content).toEqual([{ type: "text", text: SessionCompaction.buildPrompt(false) }]) + expect(requests[0].messages[1].content).toEqual([{ type: "text", text: "Hook instruction" }]) + expect(requests[0].messages[2].content).toEqual([{ type: "text", text: SessionCompaction.buildPrompt(false) }]) }), ) @@ -877,6 +901,189 @@ it.effect("forked session compaction reuses the fork root prompt cache key", () }), ) +const checkpointFixture = Effect.gen(function* () { + requests = [] + const session = yield* insertSession(Session.ID.create()) + const hooks = yield* PluginHooks.Service + const compaction = yield* SessionCompaction.Service + yield* compaction.transform((editor) => editor.configure({ keep: 0 })) + const canary = `COMPACTION-TEST-CANARY-${crypto.randomUUID()}` + const checkpoint = Schema.decodeUnknownSync(SessionMessage.CompactionCompleted)({ + id: SessionMessage.ID.create(), + type: "compaction", + status: "completed", + reason: "auto", + summary: `SUMMARY ${canary}`, + metadata: { recent: `RECENT ${canary}` }, + time: { created: 0, completed: 0 }, + }) + const user = (text: string) => + SessionMessage.User.make({ + id: SessionMessage.ID.create(), + type: "user", + text, + time: { created: DateTime.makeUnsafe(1) }, + }) + const messages = [ + checkpoint, + user("Exchange being summarized"), + Schema.decodeUnknownSync(SessionMessage.Assistant)({ + id: SessionMessage.ID.create(), + type: "assistant", + agent: Agent.defaultID, + model: { id: "summary-model", providerID: "test" }, + content: [{ type: "reasoning", text: "reasoning ".repeat(30_000) }], + time: { created: 1, completed: 2 }, + }), + user("Tail retained outside the summary request"), + ] + const context = loaded(session, messages) + return { + session, + hooks, + checkpoint, + canary, + redact: () => + hooks.register("session", "compaction", (event) => + Effect.sync(() => { + event.messages = JSON.parse(JSON.stringify(event.messages).replaceAll(canary, "[redacted]")) + }), + ), + compact: (window: number) => + compaction.compact({ + reason: "manual", + inputID: SessionMessage.ID.create(), + context: { ...context, model: { ...context.model, limit: { context: window, output: 2000 } } }, + }), + } +}) + +it.effect("shrinking preserves hook redaction of both summary and recent", () => + Effect.gen(function* () { + const fixture = yield* checkpointFixture + yield* fixture.redact() + expect(yield* fixture.compact(10_000)).toEqual({ status: "completed" }) + expect(requests).toHaveLength(1) + const sent = JSON.stringify(requests[0].messages) + expect(sent.includes(fixture.canary)).toBe(false) + expect(sent).toContain("SUMMARY [redacted]") + expect(sent).toContain("RECENT [redacted]") + expect(sent).not.toContain("reasoning reasoning") + expect(fixture.checkpoint.summary.includes(fixture.canary)).toBe(true) + expect(fixture.checkpoint.metadata?.recent).toContain(fixture.canary) + }), +) + +const overflowing = suite( + Layer.effect( + LLMClient.Service, + Effect.gen(function* () { + const llm = yield* LLMClient.Service + let rejected = false + return { + ...llm, + stream: (request, options) => { + if (rejected) return llm.stream(request, options) + rejected = true + requests.push(request) + return Stream.fail( + new AIError({ + reason: new InvalidRequestError({ + message: "Too large", + classification: "context-overflow", + }), + }), + ) + }, + } + }), + ).pipe(Layer.provide(client)), +) + +overflowing.effect("context overflow preserves checkpoint hook redaction", () => + Effect.gen(function* () { + const fixture = yield* checkpointFixture + yield* fixture.redact() + expect(yield* fixture.compact(200_000)).toEqual({ status: "completed" }) + expect(requests).toHaveLength(2) + expect(JSON.stringify(requests).includes(fixture.canary)).toBe(false) + expect(JSON.stringify(requests[0].messages)).toContain("reasoning reasoning") + const retried = JSON.stringify(requests[1].messages) + expect(retried).toContain("SUMMARY [redacted]") + expect(retried).toContain("RECENT [redacted]") + expect(retried).not.toContain("reasoning reasoning") + }), +) + +overflowing.effect("overflow reduction excludes tools removed by a hook", () => + Effect.gen(function* () { + requests = [] + const session = yield* insertSession(Session.ID.create()) + const context = { + ...loaded(session, [ + textUser("Keep this request in the summary."), + Schema.decodeUnknownSync(SessionMessage.Assistant)({ + id: SessionMessage.ID.create(), + type: "assistant", + agent: Agent.defaultID, + model: { id: "summary-model", providerID: "test" }, + content: [{ type: "reasoning", text: "reasoning ".repeat(30_000) }], + time: { created: 1, completed: 2 }, + }), + ]), + tools: { + // About 90k tokens: larger than the retry budget, but absent from every outgoing request. + definitions: [ + ToolDefinition.make({ + name: "unused", + description: "x".repeat(360_000), + inputSchema: { type: "object" }, + }), + ], + execute: () => Effect.die("Compaction must not execute tools"), + }, + } + const hooks = yield* PluginHooks.Service + yield* hooks.register("session", "compaction", (event) => + Effect.sync(() => { + event.tools = {} + }), + ) + + const compaction = yield* SessionCompaction.Service + const result = yield* compaction.compact({ reason: "manual", inputID: SessionMessage.ID.create(), context }) + + expect(result).toEqual({ status: "completed" }) + expect(requests).toHaveLength(2) + expect(requests.map((request) => request.tools)).toEqual([[], []]) + expect(JSON.stringify(requests[0].messages)).toContain("reasoning reasoning") + expect(JSON.stringify(requests[1].messages)).not.toContain("reasoning reasoning") + expect(JSON.stringify(requests[1].messages)).toContain("Keep this request in the summary.") + }), +) + +it.effect("a result supplied on the flattened draft skips the model call", () => + Effect.gen(function* () { + const fixture = yield* checkpointFixture + yield* fixture.hooks.register("session", "compaction", (event) => + Effect.sync(() => { + const flattened = event.messages.some((message) => + message.content.some( + (part) => part.type === "text" && part.text.includes("[User]: Exchange being summarized"), + ), + ) + if (flattened) event.result = { summary: "## Objective\n- supplied on flattened draft" } + }), + ) + expect(yield* fixture.compact(10_000)).toEqual({ status: "completed" }) + expect(requests).toHaveLength(0) + const store = yield* SessionStore.Service + expect(yield* store.context(fixture.session.id)).toMatchObject([ + { type: "compaction", status: "completed", summary: "## Objective\n- supplied on flattened draft" }, + ]) + }), +) + const measured = (text: string, input: number) => Schema.decodeUnknownSync(SessionMessage.Assistant)({ id: SessionMessage.ID.create(), @@ -930,6 +1137,11 @@ it.effect("automatic compaction swaps in the background summary without another const sessionID = Session.ID.make("ses_precompute_swap") const session = yield* insertSession(sessionID) const history = precomputeHistory() + // Force reduction while the latest response remains outside the summarized prefix. + history.near[1] = SessionMessage.Assistant.make({ + ...measured("First answer", 80_000), + content: [{ type: "reasoning", text: "reasoning ".repeat(80_000) }], + }) yield* storeMessages(sessionID, history.due) yield* compaction.precompute(Effect.succeed(loaded(session, history.near))) @@ -939,7 +1151,7 @@ it.effect("automatic compaction swaps in the background summary without another // The one request summarized the prefix the latest response was generated from. expect(requests).toHaveLength(1) - expect(JSON.stringify(requests[0].messages)).toContain("Second question") + expect(JSON.stringify(requests[0].messages)).toContain("[User]: Second question") expect(JSON.stringify(requests[0].messages)).not.toContain("Second answer") expect(yield* store.context(sessionID)).toMatchObject([ { diff --git a/services/www/src/docs/content/build/plugins/index.mdx b/services/www/src/docs/content/build/plugins/index.mdx index 102b4a97448c..c187434e52a7 100644 --- a/services/www/src/docs/content/build/plugins/index.mdx +++ b/services/www/src/docs/content/build/plugins/index.mdx @@ -1400,7 +1400,9 @@ differently: - `context` runs for the agent loop, including tool-driven continuations. - `compaction` runs for checkpoint summaries. `messages` is the transcript being summarized; OpenCode appends its summary prompt after hooks run. Set `result` to record the compaction yourself and skip the model call; it takes the - same fields as a completed compaction message. + same fields as a completed compaction message. If the transcript must be reduced to fit, `compaction` and + `model.request` run again on a fresh reduced draft, without edits from earlier attempts. A supplied `result` + completes compaction immediately; it is not reduced or carried into another attempt. - `generate` runs for transient `ctx.session.generate` calls. - `title` runs for title generation. It has no `agent` or `tools`. Set `result` to supply the title yourself.