Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions packages/client/src/promise/generated/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2349,6 +2349,7 @@ export type ConfigEntry =
experimental?: {
portable_shell_scanner?: boolean
todo?: boolean
precompute_compaction?: boolean
subagent_depth?: number
location_idle_timeout?: string | false
policies?: Array<{
Expand Down
9 changes: 9 additions & 0 deletions packages/core/src/config/normalize.ts
Original file line number Diff line number Diff line change
Expand Up @@ -476,6 +476,15 @@ function normalizeExperimental(
)
if (value !== undefined) result.todo = value
}
if (own(experimental, "precompute_compaction")) {
const value = decodeEncoded(
ConfigExperimental.Info.fields.precompute_compaction,
experimental.precompute_compaction,
["experimental", "precompute_compaction"],
diagnostics,
)
if (value !== undefined) result.precompute_compaction = value
}
if (own(experimental, "subagent_depth")) {
const value = decodeEncoded(
ConfigExperimental.Info.fields.subagent_depth,
Expand Down
1 change: 1 addition & 0 deletions packages/core/src/config/plugin/compaction.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ export const Plugin = define({
const compaction = yield* SessionCompaction.Service
const loaded = yield* ConfigEntryObserver.observe(config, ctx.event, compaction.reload())
yield* compaction.transform((editor) => {
editor.configure({ precompute: Config.latest(loaded.entries, "experimental")?.precompute_compaction === true })
for (const entry of loaded.entries) {
if (entry.type !== "document" || !entry.info.compaction) continue
editor.configure({
Expand Down
282 changes: 236 additions & 46 deletions packages/core/src/session/compaction.ts

Large diffs are not rendered by default.

10 changes: 9 additions & 1 deletion packages/core/src/session/runner/llm.ts
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,7 @@ const layer = Layer.effect(
assistantMessageID = SessionMessage.ID.create()
continue
}
yield* compaction.precompute(Effect.succeed(loaded))
const stepLimitReached = loaded.agent.info.steps !== undefined && step >= loaded.agent.info.steps
const transcript = SessionModelRequest.baseTranscript({
agent: loaded.agent.info,
Expand Down Expand Up @@ -265,7 +266,14 @@ const layer = Layer.effect(
),
})
const completed = yield* SessionStep.Outcome.$match(outcome, {
Completed: (outcome) => Effect.succeed(outcome.needsContinuation),
Completed: Effect.fnUntraced(function* (outcome) {
// The Session is about to go idle, which leaves the background summary time to finish.
if (!outcome.needsContinuation)
yield* compaction.precompute(
store.context(sessionID).pipe(Effect.map((messages) => ({ ...loaded, messages }))),
)
return outcome.needsContinuation
}),
Retry: (outcome) =>
retry.wait({
decision: outcome.decision,
Expand Down
2 changes: 2 additions & 0 deletions packages/core/test/config/normalization.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -525,6 +525,7 @@ describe("ConfigNormalize", () => {
experimental: {
portable_shell_scanner: true,
todo: true,
precompute_compaction: true,
subagent_depth: 0,
location_idle_timeout: "2 hours",
policies: [{ action: "provider.use", resource: "custom", effect: "allow" }],
Expand All @@ -534,6 +535,7 @@ describe("ConfigNormalize", () => {
experimental: {
portable_shell_scanner: true,
todo: true,
precompute_compaction: true,
subagent_depth: 0,
location_idle_timeout: "7200000 millis",
},
Expand Down
271 changes: 253 additions & 18 deletions packages/core/test/session-compaction.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,16 +23,19 @@ import { Project } from "@opencode/core/project"
import { ProjectTable } from "@opencode/core/project/sql"
import { App } from "@opencode/core/app"
import { Agent } from "@opencode/core/agent"
import { Model } from "@opencode/core/model"
import { Provider } from "@opencode/core/provider"
import { AbsolutePath } from "@opencode/core/schema"
import { Money } from "@opencode/schema/money"
import { Skill } from "@opencode/schema/skill"
import { Shell } from "@opencode/schema/shell"
import { DateTime, Effect, Fiber, Layer, Schema, Stream } from "effect"
import { DateTime, Deferred, Effect, Fiber, Layer, Schema, Stream } from "effect"
import { asc, eq } from "drizzle-orm"
import { testEffect } from "./lib/effect"

let requests: LLMRequest[] = []
/** Runs before a request's response streams; tests hold it back to keep a background summary in flight. */
let beforeResponse: Effect.Effect<void> = Effect.void
const model = LanguageModel.make({
id: "summary-model",
provider: "test",
Expand All @@ -51,23 +54,28 @@ const cost = [
const client = Layer.mock(LLMClient.Service)({
stream: (request: LLMRequest) => {
requests.push(request)
return Stream.make(
LLMEvent.textDelta({ id: "summary", text: "## Objective\n- manual summary" }),
LLMEvent.stepFinish({
index: 0,
reason: { normalized: "stop" },
usage: {
inputTokens: 15,
outputTokens: 6,
nonCachedInputTokens: 10,
cacheReadInputTokens: 3,
cacheWriteInputTokens: 2,
reasoningTokens: 2,
},
}),
LLMEvent.finish({
reason: { normalized: "stop" },
}),
return Stream.fromEffect(beforeResponse).pipe(
Stream.drain,
Stream.concat(
Stream.make(
LLMEvent.textDelta({ id: "summary", text: "## Objective\n- manual summary" }),
LLMEvent.stepFinish({
index: 0,
reason: { normalized: "stop" },
usage: {
inputTokens: 15,
outputTokens: 6,
nonCachedInputTokens: 10,
cacheReadInputTokens: 3,
cacheWriteInputTokens: 2,
reasoningTokens: 2,
},
}),
LLMEvent.finish({
reason: { normalized: "stop" },
}),
),
),
)
},
generate: () => Effect.die("unused"),
Expand Down Expand Up @@ -856,3 +864,230 @@ it.effect("forked session compaction reuses the fork root prompt cache key", ()
expect(requests[0]?.promptCacheKey).toBe(rootID)
}),
)

const measured = (text: string, input: number) =>
Schema.decodeUnknownSync(SessionMessage.Assistant)({
id: SessionMessage.ID.create(),
type: "assistant",
agent: Agent.defaultID,
model: { id: "summary-model", providerID: "test" },
content: [{ type: "text", text }],
tokens: { input, output: 100, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: 0, completed: 1 },
})

/**
* A 200k window compacts at 170k and starts a background summary at 85% of that, 144.5k. `near` ends past that
* point; `due` continues it past the threshold.
*/
const precomputeHistory = () => {
const near = [textUser("First question"), measured("First answer", 80_000), textUser("Second question")]
const anchor = measured("Second answer", 156_000)
const due = [...near, anchor, textUser("Third question"), measured("Third answer", 182_000)]
return { near: [...near, anchor], due, anchor }
}

const enablePrecompute = Effect.gen(function* () {
const compaction = yield* SessionCompaction.Service
yield* compaction.transform((editor) => editor.configure({ precompute: true }))
return compaction
})

it.effect("background summaries stay off unless enabled", () =>
Effect.gen(function* () {
requests = []
const compaction = yield* SessionCompaction.Service
const session = yield* insertSession(Session.ID.make("ses_precompute_off"))
const history = precomputeHistory()

yield* compaction.precompute(Effect.succeed(loaded(session, history.near)))
expect(yield* compaction.compact({ reason: "auto", context: loaded(session, history.due) })).toEqual({
status: "completed",
})
expect(requests).toHaveLength(1)
expect(JSON.stringify(requests[0].messages)).toContain("Third answer")
}),
)

it.effect("automatic compaction swaps in the background summary without another model call", () =>
Effect.gen(function* () {
requests = []
const db = (yield* Database.Service).db
const store = yield* SessionStore.Service
const compaction = yield* enablePrecompute
const sessionID = Session.ID.make("ses_precompute_swap")
const session = yield* insertSession(sessionID)
const history = precomputeHistory()
yield* storeMessages(sessionID, history.due)

yield* compaction.precompute(Effect.succeed(loaded(session, history.near)))
expect(yield* compaction.compact({ reason: "auto", context: loaded(session, history.due) })).toEqual({
status: "completed",
})

// 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)).not.toContain("Second answer")
expect(yield* store.context(sessionID)).toMatchObject([
{
type: "compaction",
reason: "auto",
summary: "## Objective\n- manual summary",
metadata: {
keepID: history.anchor.id,
recent: "[Assistant]: Second answer\n\n[User]: Third question\n\n[Assistant]: Third answer",
},
tokens: { input: 10, output: 4, reasoning: 2, cache: { read: 3, write: 2 } },
},
])
// The background call is billed once, when it finishes, and the compaction ends after it.
const types = (yield* db
.select({ type: EventTable.type })
.from(EventTable)
.where(eq(EventTable.aggregate_id, sessionID))
.orderBy(asc(EventTable.seq))
.all()
.pipe(Effect.orDie)).map((row) => row.type)
expect(types.at(-1)).toBe(Bus.versionedType(SessionEvent.Compaction.Ended.type, 1))
expect(types.toSorted()).toEqual(
[
Bus.versionedType(SessionEvent.UsageRecorded.type, 1),
Bus.versionedType(SessionEvent.Compaction.Started.type, 1),
Bus.versionedType(SessionEvent.Compaction.Ended.type, 1),
].toSorted(),
)
}),
)

it.effect("automatic compaction waits for a background summary still being written", () =>
Effect.gen(function* () {
requests = []
const release = yield* Deferred.make<void>()
beforeResponse = Deferred.await(release)
const bus = yield* Bus.Service
const compaction = yield* enablePrecompute
const session = yield* insertSession(Session.ID.make("ses_precompute_join"))
const history = precomputeHistory()
const started = yield* bus
.subscribe(SessionEvent.Compaction.Started)
.pipe(Stream.take(1), Stream.runCollect, Effect.forkChild({ startImmediately: true }))

yield* compaction.precompute(Effect.succeed(loaded(session, history.near)))
const compacting = yield* compaction
.compact({ reason: "auto", context: loaded(session, history.due) })
.pipe(Effect.forkChild({ startImmediately: true }))

// The compaction shows as running while the summary it waits for is still being written.
expect(yield* Fiber.join(started)).toHaveLength(1)
expect(compacting.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(release, undefined)

expect(yield* Fiber.join(compacting)).toEqual({ status: "completed" })
expect(requests).toHaveLength(1)
}).pipe(Effect.ensuring(Effect.sync(() => (beforeResponse = Effect.void)))),
)

it.effect("a model switch waits for the new model's first response before summarizing in the background", () =>
Effect.gen(function* () {
requests = []
const compaction = yield* enablePrecompute
const session = yield* insertSession(Session.ID.make("ses_precompute_switch"))
const history = precomputeHistory()
const switched = (messages: readonly SessionMessage.Info[]) => ({
...loaded(session, messages),
model: { ...resolved, ref: { ...resolved.ref, id: Model.ID.make("other") } },
})

// The latest response came from the previous model, so the new one has no cached prefix to summarize from.
yield* compaction.precompute(Effect.succeed(switched(history.near)))
expect(yield* compaction.compact({ reason: "auto", context: switched(history.due) })).toEqual({
status: "completed",
})
expect(requests).toHaveLength(1)
expect(JSON.stringify(requests[0].messages)).toContain("Third answer")
}),
)

it.effect("a background summary that fell too far behind is replaced from a newer response", () =>
Effect.gen(function* () {
requests = []
const compaction = yield* enablePrecompute
const session = yield* insertSession(Session.ID.make("ses_precompute_behind"))
const history = precomputeHistory()
// About 35k tokens arrived after the summarized prefix, past the 30k that may stay verbatim beside it.
const behind = [...history.near, textUser("x".repeat(140_000)), measured("Burst handled", 160_000)]

yield* compaction.precompute(Effect.succeed(loaded(session, history.near)))
yield* compaction.precompute(Effect.succeed(loaded(session, behind)))
const due = [...behind, textUser("Next question"), measured("Next answer", 182_000)]
expect(yield* compaction.compact({ reason: "auto", context: loaded(session, due) })).toEqual({
status: "completed",
})

// Compaction swapped in the newer summary, which ends before the latest response, instead of writing one.
const summarized = JSON.stringify(requests.at(-1)?.messages)
expect(summarized).toContain("x".repeat(1_000))
expect(summarized).not.toContain("Burst handled")
}),
)

it.effect("a failed background summary is retried from a newer response, then left until compaction", () =>
Effect.gen(function* () {
requests = []
const compaction = yield* enablePrecompute
const session = yield* insertSession(Session.ID.make("ses_precompute_failed"))
const history = precomputeHistory()
const later = [...history.near, textUser("Third question"), measured("Third answer", 160_000)]
const latest = [...later, textUser("Fourth question"), measured("Fourth answer", 165_000)]
const precomputeFailing = (messages: readonly SessionMessage.Info[]) =>
Effect.gen(function* () {
const reached = yield* Deferred.make<void>()
beforeResponse = Deferred.succeed(reached, undefined).pipe(Effect.andThen(Effect.die("provider down")))
yield* compaction.precompute(Effect.succeed(loaded(session, messages)))
yield* Deferred.await(reached)
yield* Effect.yieldNow

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This assumes the background fiber has failed after a single yieldNow once the response is reached. If the failure takes another scheduler turn to land, pollUnsafe() in precompute still sees it as running, and the next assertions change. Awaiting the held fiber's exit (or a deferred completed when it fails) would make this deterministic, rather than depending on scheduling.

})

yield* precomputeFailing(history.near)
expect(requests).toHaveLength(1)
// The same response would send the same request again.
yield* compaction.precompute(Effect.succeed(loaded(session, history.near)))
expect(requests).toHaveLength(1)

yield* precomputeFailing(later)
expect(requests).toHaveLength(2)
expect(JSON.stringify(requests[1].messages)).toContain("Second answer")

// Two failures in a row stop background summaries until the next compaction.
yield* compaction.precompute(Effect.succeed(loaded(session, latest)))
expect(requests).toHaveLength(2)
}).pipe(Effect.ensuring(Effect.sync(() => (beforeResponse = Effect.void)))),
)

it.effect("compaction summarizes as usual when the background summary no longer fits the conversation", () =>
Effect.gen(function* () {
const compaction = yield* enablePrecompute
const session = yield* insertSession(Session.ID.make("ses_precompute_stale"))
const history = precomputeHistory()
const compactAfterPrecompute = (due: readonly SessionMessage.Info[], context = loaded(session, due)) =>
Effect.gen(function* () {
requests = []
yield* compaction.precompute(Effect.succeed(loaded(session, history.near)))
expect(yield* compaction.compact({ reason: "auto", context })).toEqual({ status: "completed" })
// The background summary is stopped; the summary that is kept covers the whole conversation.
return JSON.stringify(requests.at(-1)?.messages)
})

// About 35k tokens arrived after the summarized prefix, past the 30k that may stay verbatim beside it.
const burst = [...history.near, textUser("x".repeat(140_000)), measured("Burst handled", 182_000)]
expect(yield* compactAfterPrecompute(burst)).toContain("Burst handled")

// A different model would replay a summary written for the previous one.
const switched = {
...loaded(session, history.due),
model: { ...resolved, ref: { ...resolved.ref, id: Model.ID.make("other") } },
}
expect(yield* compactAfterPrecompute(history.due, switched)).toContain("Third answer")
}),
)
3 changes: 3 additions & 0 deletions packages/protocol/openapi.json
Original file line number Diff line number Diff line change
Expand Up @@ -14463,6 +14463,9 @@
"todo": {
"type": "boolean"
},
"precompute_compaction": {
"type": "boolean"
},
"subagent_depth": {
"type": "integer",
"minimum": 0
Expand Down
4 changes: 4 additions & 0 deletions packages/schema/src/config/experimental.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,10 @@ export class Info extends Schema.Class<Info>("ConfigExperimental.Info")({
todo: Schema.Boolean.pipe(optional).annotate({
description: "Enable the experimental todowrite tool. Defaults to false.",
}),
precompute_compaction: Schema.Boolean.pipe(optional).annotate({
description:
"Summarize the conversation in the background as it nears the compaction threshold, so automatic compaction does not wait for the summary. Background summaries that are never used still cost tokens. Defaults to false.",
}),
subagent_depth: NonNegativeInt.pipe(optional).annotate({
description: "Maximum subagent nesting depth. Defaults to 1.",
}),
Expand Down
Loading
Loading