clone/opencode
1
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-06-01 19:05:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Effectify remaining compaction process tests (#26776)

Browse Source
This commit is contained in:
Kit Langton
2026-05-10 20:17:01 -04:00
committed by GitHub
parent 7cea32ee08
commit 16aa67086b
1 changed files with 266 additions and 512 deletions
Download Patch File Download Diff File Expand all files Collapse all files

778
packages/opencode/test/session/compaction.test.ts
View File

@@ -1,8 +1,7 @@
import { afterEach, describe, expect, mock, test } from "bun:test" import { afterEach, describe, expect, mock, test } from "bun:test"
import { APICallError } from "ai" import { APICallError } from "ai"
import { Cause, Deferred, Effect, Exit, Layer, ManagedRuntime } from "effect" import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
import * as Stream from "effect/Stream" import * as Stream from "effect/Stream"
import z from "zod"
import { Bus } from "../../src/bus" import { Bus } from "../../src/bus"
import { Config } from "@/config/config" import { Config } from "@/config/config"
import { Image } from "@/image/image" import { Image } from "@/image/image"
@@ -10,11 +9,10 @@ import { Agent } from "../../src/agent/agent"
import { LLM } from "../../src/session/llm" import { LLM } from "../../src/session/llm"
import { SessionCompaction } from "../../src/session/compaction" import { SessionCompaction } from "../../src/session/compaction"
import { Token } from "@/util/token" import { Token } from "@/util/token"
import { WithInstance } from "../../src/project/with-instance"
import * as Log from "@opencode-ai/core/util/log" import * as Log from "@opencode-ai/core/util/log"
import { Permission } from "../../src/permission" import { Permission } from "../../src/permission"
import { Plugin } from "../../src/plugin" import { Plugin } from "../../src/plugin"
import { provideTmpdirInstance, TestInstance, tmpdir } from "../fixture/fixture" import { provideTmpdirInstance, TestInstance } from "../fixture/fixture"
import { Session as SessionNs } from "@/session/session" import { Session as SessionNs } from "@/session/session"
import { MessageV2 } from "../../src/session/message-v2" import { MessageV2 } from "../../src/session/message-v2"
import { MessageID, PartID, SessionID } from "../../src/session/schema" import { MessageID, PartID, SessionID } from "../../src/session/schema"
@@ -32,26 +30,6 @@ import { TestConfig } from "../fixture/config"
void Log.init({ print: false }) void Log.init({ print: false })
function run<A, E>(fx: Effect.Effect<A, E, SessionNs.Service>) {
return Effect.runPromise(fx.pipe(Effect.provide(SessionNs.defaultLayer)))
}
const svc = {
...SessionNs,
create(input?: SessionNs.CreateInput) {
return run(SessionNs.Service.use((svc) => svc.create(input)))
},
messages(input: z.output<typeof SessionNs.MessagesInput.zod>) {
return run(SessionNs.Service.use((svc) => svc.messages(input)))
},
updateMessage<T extends MessageV2.Info>(msg: T) {
return run(SessionNs.Service.use((svc) => svc.updateMessage(msg)))
},
updatePart<T extends MessageV2.Part>(part: T) {
return run(SessionNs.Service.use((svc) => svc.updatePart(part)))
},
}
const summary = Layer.succeed( const summary = Layer.succeed(
SessionSummary.Service, SessionSummary.Service,
SessionSummary.Service.of({ SessionSummary.Service.of({
@@ -102,50 +80,6 @@ function createModel(opts: {
const wide = () => ProviderTest.fake({ model: createModel({ context: 100_000, output: 32_000 }) }) const wide = () => ProviderTest.fake({ model: createModel({ context: 100_000, output: 32_000 }) })
async function user(sessionID: SessionID, text: string) {
const msg = await svc.updateMessage({
id: MessageID.ascending(),
role: "user",
sessionID,
agent: "build",
model: ref,
time: { created: Date.now() },
})
await svc.updatePart({
id: PartID.ascending(),
messageID: msg.id,
sessionID,
type: "text",
text,
})
return msg
}
async function assistant(sessionID: SessionID, parentID: MessageID, root: string) {
const msg: MessageV2.Assistant = {
id: MessageID.ascending(),
role: "assistant",
sessionID,
mode: "build",
agent: "build",
path: { cwd: root, root },
cost: 0,
tokens: {
output: 0,
input: 0,
reasoning: 0,
cache: { read: 0, write: 0 },
},
modelID: ref.modelID,
providerID: ref.providerID,
parentID,
time: { created: Date.now() },
finish: "end_turn",
}
await svc.updateMessage(msg)
return msg
}
function createUserMessage(sessionID: SessionID, text: string) { function createUserMessage(sessionID: SessionID, text: string) {
return Effect.gen(function* () { return Effect.gen(function* () {
const ssn = yield* SessionNs.Service const ssn = yield* SessionNs.Service
@@ -193,37 +127,40 @@ function createAssistantMessage(sessionID: SessionID, parentID: MessageID, root:
) )
} }
async function summaryAssistant(sessionID: SessionID, parentID: MessageID, root: string, text: string) { function createSummaryAssistantMessage(sessionID: SessionID, parentID: MessageID, root: string, text: string) {
const msg: MessageV2.Assistant = { return SessionNs.Service.use((ssn) =>
id: MessageID.ascending(), Effect.gen(function* () {
role: "assistant", const msg = yield* ssn.updateMessage({
sessionID, id: MessageID.ascending(),
mode: "compaction", role: "assistant",
agent: "compaction", sessionID,
path: { cwd: root, root }, mode: "compaction",
cost: 0, agent: "compaction",
tokens: { path: { cwd: root, root },
output: 0, cost: 0,
input: 0, tokens: {
reasoning: 0, output: 0,
cache: { read: 0, write: 0 }, input: 0,
}, reasoning: 0,
modelID: ref.modelID, cache: { read: 0, write: 0 },
providerID: ref.providerID, },
parentID, modelID: ref.modelID,
summary: true, providerID: ref.providerID,
time: { created: Date.now() }, parentID,
finish: "end_turn", summary: true,
} time: { created: Date.now() },
await svc.updateMessage(msg) finish: "end_turn",
await svc.updatePart({ })
id: PartID.ascending(), yield* ssn.updatePart({
messageID: msg.id, id: PartID.ascending(),
sessionID, messageID: msg.id,
type: "text", sessionID,
text, type: "text",
}) text,
return msg })
return msg
}),
)
} }
function createCompactionMarker(sessionID: SessionID) { function createCompactionMarker(sessionID: SessionID) {
@@ -248,10 +185,6 @@ function createCompactionMarker(sessionID: SessionID) {
) )
} }
async function createCompactionMarkerAsync(sessionID: SessionID) {
return run(createCompactionMarker(sessionID))
}
function fake( function fake(
input: Parameters<SessionProcessorModule.SessionProcessor.Interface["create"]>[0], input: Parameters<SessionProcessorModule.SessionProcessor.Interface["create"]>[0],
result: "continue" | "compact", result: "continue" | "compact",
@@ -283,26 +216,6 @@ function cfg(compaction?: Config.Info["compaction"]) {
}) })
} }
function runtime(
result: "continue" | "compact",
plugin = Plugin.defaultLayer,
provider = ProviderTest.fake(),
config = Config.defaultLayer,
) {
const bus = Bus.layer
return ManagedRuntime.make(
Layer.mergeAll(SessionCompaction.layer, bus).pipe(
Layer.provide(provider.layer),
Layer.provide(SessionNs.defaultLayer),
Layer.provide(layer(result)),
Layer.provide(Agent.defaultLayer),
Layer.provide(plugin),
Layer.provide(bus),
Layer.provide(config),
),
)
}
const deps = Layer.mergeAll( const deps = Layer.mergeAll(
wide().layer, wide().layer,
layer("continue"), layer("continue"),
@@ -365,10 +278,6 @@ function readCompactionPart(sessionID: SessionID) {
) )
} }
async function lastCompactionPart(sessionID: SessionID) {
return run(readCompactionPart(sessionID))
}
function llm() { function llm() {
const queue: Array< const queue: Array<
Stream.Stream<LLM.Event, unknown> | ((input: LLM.StreamInput) => Stream.Stream<LLM.Event, unknown>) Stream.Stream<LLM.Event, unknown> | ((input: LLM.StreamInput) => Stream.Stream<LLM.Event, unknown>)
@@ -391,29 +300,6 @@ function llm() {
} }
} }
function liveRuntime(layer: Layer.Layer<LLM.Service>, provider = ProviderTest.fake(), config = Config.defaultLayer) {
const bus = Bus.layer
const status = SessionStatus.layer.pipe(Layer.provide(bus))
const processor = SessionProcessorModule.SessionProcessor.layer.pipe(
Layer.provide(summary),
Layer.provide(Image.defaultLayer),
)
return ManagedRuntime.make(
Layer.mergeAll(SessionCompaction.layer.pipe(Layer.provide(processor)), processor, bus, status).pipe(
Layer.provide(provider.layer),
Layer.provide(SessionNs.defaultLayer),
Layer.provide(Snapshot.defaultLayer),
Layer.provide(layer),
Layer.provide(Permission.defaultLayer),
Layer.provide(Agent.defaultLayer),
Layer.provide(Plugin.defaultLayer),
Layer.provide(status),
Layer.provide(bus),
Layer.provide(config),
),
)
}
function reply( function reply(
text: string, text: string,
capture?: (input: LLM.StreamInput) => void, capture?: (input: LLM.StreamInput) => void,
@@ -469,23 +355,14 @@ function reply(
} }
} }
function wait(ms = 50) { function plugin(ready: Deferred.Deferred<void>) {
return new Promise((resolve) => setTimeout(resolve, ms))
}
function defer() {
let resolve!: () => void
const promise = new Promise<void>((done) => {
resolve = done
})
return { promise, resolve }
}
function plugin(ready: ReturnType<typeof defer>) {
return Layer.mock(Plugin.Service)({ return Layer.mock(Plugin.Service)({
trigger: <Name extends string, Input, Output>(name: Name, _input: Input, output: Output) => { trigger: <Name extends string, Input, Output>(name: Name, _input: Input, output: Output) => {
if (name !== "experimental.session.compacting") return Effect.succeed(output) if (name !== "experimental.session.compacting") return Effect.succeed(output)
return Effect.sync(() => ready.resolve()).pipe(Effect.andThen(Effect.never), Effect.as(output)) return Effect.sync(() => Deferred.doneUnsafe(ready, Effect.void)).pipe(
Effect.andThen(Effect.never),
Effect.as(output),
)
}, },
list: () => Effect.succeed([]), list: () => Effect.succeed([]),
init: () => Effect.void, init: () => Effect.void,
@@ -1315,154 +1192,99 @@ describe("session.compaction.process", () => {
}), }),
) )
test("stops quickly when aborted during retry backoff", async () => { itProcess.instance(
const stub = llm() "stops quickly when aborted during retry backoff",
const ready = defer() () => {
stub.push( const stub = llm()
Stream.fromAsyncIterable( stub.push(
{ Stream.fromAsyncIterable(
async *[Symbol.asyncIterator]() { {
yield { type: "start" } as LLM.Event async *[Symbol.asyncIterator]() {
throw new APICallError({ yield { type: "start" } as LLM.Event
message: "boom", throw new APICallError({
url: "https://example.com/v1/chat/completions", message: "boom",
requestBodyValues: {}, url: "https://example.com/v1/chat/completions",
statusCode: 503, requestBodyValues: {},
responseHeaders: { "retry-after-ms": "10000" }, statusCode: 503,
responseBody: '{"error":"boom"}', responseHeaders: { "retry-after-ms": "10000" },
isRetryable: true, responseBody: '{"error":"boom"}',
}) isRetryable: true,
})
},
}, },
}, (err) => err,
(err) => err, ),
), )
)
await using tmp = await tmpdir({ git: true }) return Effect.gen(function* () {
await WithInstance.provide({ const ssn = yield* SessionNs.Service
directory: tmp.path, const bus = yield* Bus.Service
fn: async () => { const ready = yield* Deferred.make<void>()
const session = await svc.create({}) const session = yield* ssn.create({})
const msg = await user(session.id, "hello") const msg = yield* createUserMessage(session.id, "hello")
const msgs = await svc.messages({ sessionID: session.id }) const msgs = yield* ssn.messages({ sessionID: session.id })
const abort = new AbortController() const off = yield* bus.subscribeCallback(SessionStatus.Event.Status, (evt) => {
const rt = liveRuntime(stub.layer, wide()) if (evt.properties.sessionID !== session.id) return
let off: (() => void) | undefined if (evt.properties.status.type !== "retry") return
let run: Promise<"continue" | "stop"> | undefined Deferred.doneUnsafe(ready, Effect.void)
try { })
off = await rt.runPromise( yield* Effect.addFinalizer(() => Effect.sync(off))
Bus.Service.use((svc) =>
svc.subscribeCallback(SessionStatus.Event.Status, (evt) => {
if (evt.properties.sessionID !== session.id) return
if (evt.properties.status.type !== "retry") return
ready.resolve()
}),
),
)
run = rt const fiber = yield* SessionCompaction.use
.runPromiseExit( .process({
SessionCompaction.Service.use((svc) => parentID: msg.id,
svc.process({ messages: msgs,
parentID: msg.id, sessionID: session.id,
messages: msgs, auto: false,
sessionID: session.id, })
auto: false, .pipe(Effect.forkChild)
}),
),
{ signal: abort.signal },
)
.then((exit) => {
if (Exit.isFailure(exit)) {
if (Cause.hasInterrupts(exit.cause) && abort.signal.aborted) return "stop"
throw Cause.squash(exit.cause)
}
return exit.value
})
await Promise.race([ yield* Deferred.await(ready).pipe(Effect.timeout("1 second"))
ready.promise, const start = Date.now()
wait(1000).then(() => { yield* Fiber.interrupt(fiber)
throw new Error("timed out waiting for retry status") const exit = yield* Fiber.await(fiber).pipe(Effect.timeout("250 millis"))
}),
])
const start = Date.now() expect(Exit.isFailure(exit)).toBe(true)
abort.abort() if (Exit.isFailure(exit)) {
const result = await Promise.race([ expect(Cause.hasInterrupts(exit.cause)).toBe(true)
run.then((value) => ({ kind: "done" as const, value, ms: Date.now() - start })), expect(Date.now() - start).toBeLessThan(250)
wait(250).then(() => ({ kind: "timeout" as const })),
])
expect(result.kind).toBe("done")
if (result.kind === "done") {
expect(result.value).toBe("stop")
expect(result.ms).toBeLessThan(250)
}
} finally {
off?.()
abort.abort()
await rt.dispose()
await run?.catch(() => undefined)
} }
}, }).pipe(Effect.provide(compactionProcessLayer({ llm: stub.layer })))
}) },
}) { git: true },
)
test("does not leave a summary assistant when aborted before processor setup", async () => { itProcess.instance(
const ready = defer() "does not leave a summary assistant when aborted before processor setup",
() =>
await using tmp = await tmpdir({ git: true }) Effect.gen(function* () {
await WithInstance.provide({ const ready = yield* Deferred.make<void>()
directory: tmp.path, return yield* Effect.gen(function* () {
fn: async () => { const ssn = yield* SessionNs.Service
const session = await svc.create({}) const session = yield* ssn.create({})
const msg = await user(session.id, "hello") const msg = yield* createUserMessage(session.id, "hello")
const msgs = await svc.messages({ sessionID: session.id }) const msgs = yield* ssn.messages({ sessionID: session.id })
const abort = new AbortController() const fiber = yield* SessionCompaction.use
const rt = runtime("continue", plugin(ready), wide()) .process({
let run: Promise<"continue" | "stop"> | undefined parentID: msg.id,
try { messages: msgs,
run = rt sessionID: session.id,
.runPromiseExit( auto: false,
SessionCompaction.Service.use((svc) =>
svc.process({
parentID: msg.id,
messages: msgs,
sessionID: session.id,
auto: false,
}),
),
{ signal: abort.signal },
)
.then((exit) => {
if (Exit.isFailure(exit)) {
if (Cause.hasInterrupts(exit.cause) && abort.signal.aborted) return "stop"
throw Cause.squash(exit.cause)
}
return exit.value
}) })
.pipe(Effect.forkChild)
await Promise.race([ yield* Deferred.await(ready).pipe(Effect.timeout("1 second"))
ready.promise, yield* Fiber.interrupt(fiber)
wait(1000).then(() => { const exit = yield* Fiber.await(fiber).pipe(Effect.timeout("250 millis"))
throw new Error("timed out waiting for compaction hook") const all = yield* ssn.messages({ sessionID: session.id })
}),
])
abort.abort() expect(Exit.isFailure(exit)).toBe(true)
expect(await run).toBe("stop") if (Exit.isFailure(exit)) expect(Cause.hasInterrupts(exit.cause)).toBe(true)
const all = await svc.messages({ sessionID: session.id })
expect(all.some((msg) => msg.info.role === "assistant" && msg.info.summary)).toBe(false) expect(all.some((msg) => msg.info.role === "assistant" && msg.info.summary)).toBe(false)
} finally { }).pipe(Effect.provide(compactionProcessLayer({ plugin: plugin(ready) })))
abort.abort() }),
await rt.dispose() { git: true },
await run?.catch(() => undefined) )
}
},
})
})
itProcess.instance( itProcess.instance(
"does not allow tool calls while generating the summary", "does not allow tool calls while generating the summary",
@@ -1533,240 +1355,172 @@ describe("session.compaction.process", () => {
{ git: true }, { git: true },
) )
test("summarizes only the head while keeping recent tail out of summary input", async () => { itProcess.instance(
const stub = llm() "summarizes only the head while keeping recent tail out of summary input",
let captured = "" () => {
stub.push( const stub = llm()
reply("summary", (input) => { let captured = ""
captured = JSON.stringify(input.messages) stub.push(
}), reply("summary", (input) => {
) captured = JSON.stringify(input.messages)
}),
)
return Effect.gen(function* () {
const ssn = yield* SessionNs.Service
const session = yield* ssn.create({})
yield* createUserMessage(session.id, "older context")
yield* createUserMessage(session.id, "keep this turn")
yield* createUserMessage(session.id, "and this one too")
yield* createCompactionMarker(session.id)
await using tmp = await tmpdir({ git: true }) const msgs = yield* ssn.messages({ sessionID: session.id })
await WithInstance.provide({ const parent = msgs.at(-1)?.info.id
directory: tmp.path, expect(parent).toBeTruthy()
fn: async () => { yield* SessionCompaction.use.process({
const session = await svc.create({}) parentID: parent!,
await user(session.id, "older context") messages: msgs,
await user(session.id, "keep this turn") sessionID: session.id,
await user(session.id, "and this one too") auto: false,
await createCompactionMarkerAsync(session.id) })
const rt = liveRuntime(stub.layer, wide()) expect(captured).toContain("older context")
try { expect(captured).not.toContain("keep this turn")
const msgs = await svc.messages({ sessionID: session.id }) expect(captured).not.toContain("and this one too")
const parent = msgs.at(-1)?.info.id expect(captured).not.toContain("What did we do so far?")
expect(parent).toBeTruthy() }).pipe(Effect.provide(compactionProcessLayer({ llm: stub.layer })))
await rt.runPromise( },
SessionCompaction.Service.use((svc) => { git: true },
svc.process({ )
parentID: parent!,
messages: msgs,
sessionID: session.id,
auto: false,
}),
),
)
expect(captured).toContain("older context") itProcess.instance(
expect(captured).not.toContain("keep this turn") "anchors repeated compactions with the previous summary",
expect(captured).not.toContain("and this one too") () => {
expect(captured).not.toContain("What did we do so far?") const stub = llm()
} finally { let captured = ""
await rt.dispose() stub.push(reply("summary one"))
} stub.push(
}, reply("summary two", (input) => {
}) captured = JSON.stringify(input.messages)
}) }),
)
test("anchors repeated compactions with the previous summary", async () => { return Effect.gen(function* () {
const stub = llm() const ssn = yield* SessionNs.Service
let captured = "" const session = yield* ssn.create({})
stub.push(reply("summary one")) yield* createUserMessage(session.id, "older context")
stub.push( yield* createUserMessage(session.id, "keep this turn")
reply("summary two", (input) => { yield* createCompactionMarker(session.id)
captured = JSON.stringify(input.messages)
}),
)
await using tmp = await tmpdir({ git: true }) let msgs = yield* ssn.messages({ sessionID: session.id })
await WithInstance.provide({ let parent = msgs.at(-1)?.info.id
directory: tmp.path, expect(parent).toBeTruthy()
fn: async () => { yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
const session = await svc.create({})
await user(session.id, "older context")
await user(session.id, "keep this turn")
await createCompactionMarkerAsync(session.id)
const rt = liveRuntime(stub.layer, wide()) yield* createUserMessage(session.id, "latest turn")
try { yield* createCompactionMarker(session.id)
let msgs = await svc.messages({ sessionID: session.id })
let parent = msgs.at(-1)?.info.id
expect(parent).toBeTruthy()
await rt.runPromise(
SessionCompaction.Service.use((svc) =>
svc.process({
parentID: parent!,
messages: msgs,
sessionID: session.id,
auto: false,
}),
),
)
await user(session.id, "latest turn") msgs = MessageV2.filterCompacted(MessageV2.stream(session.id))
await createCompactionMarkerAsync(session.id) parent = msgs.at(-1)?.info.id
expect(parent).toBeTruthy()
yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
msgs = MessageV2.filterCompacted(MessageV2.stream(session.id)) expect(captured).toContain("<previous-summary>")
parent = msgs.at(-1)?.info.id expect(captured).toContain("summary one")
expect(parent).toBeTruthy() expect(captured.match(/summary one/g)?.length).toBe(1)
await rt.runPromise( expect(captured).toContain("## Constraints & Preferences")
SessionCompaction.Service.use((svc) => expect(captured).toContain("## Progress")
svc.process({ }).pipe(Effect.provide(compactionProcessLayer({ llm: stub.layer })))
parentID: parent!, },
messages: msgs, { git: true },
sessionID: session.id, )
auto: false,
}),
),
)
expect(captured).toContain("<previous-summary>") itProcess.instance("keeps recent pre-compaction turns across repeated compactions", () => {
expect(captured).toContain("summary one")
expect(captured.match(/summary one/g)?.length).toBe(1)
expect(captured).toContain("## Constraints & Preferences")
expect(captured).toContain("## Progress")
} finally {
await rt.dispose()
}
},
})
})
test("keeps recent pre-compaction turns across repeated compactions", async () => {
const stub = llm() const stub = llm()
stub.push(reply("summary one")) stub.push(reply("summary one"))
stub.push(reply("summary two")) stub.push(reply("summary two"))
await using tmp = await tmpdir()
await WithInstance.provide({
directory: tmp.path,
fn: async () => {
const session = await svc.create({})
const u1 = await user(session.id, "one")
const u2 = await user(session.id, "two")
const u3 = await user(session.id, "three")
await createCompactionMarkerAsync(session.id)
const rt = liveRuntime(stub.layer, wide(), cfg({ tail_turns: 2, preserve_recent_tokens: 10_000 })) return Effect.gen(function* () {
try { const ssn = yield* SessionNs.Service
let msgs = await svc.messages({ sessionID: session.id }) const session = yield* ssn.create({})
let parent = msgs.at(-1)?.info.id const u1 = yield* createUserMessage(session.id, "one")
expect(parent).toBeTruthy() const u2 = yield* createUserMessage(session.id, "two")
await rt.runPromise( const u3 = yield* createUserMessage(session.id, "three")
SessionCompaction.Service.use((svc) => yield* createCompactionMarker(session.id)
svc.process({
parentID: parent!,
messages: msgs,
sessionID: session.id,
auto: false,
}),
),
)
const u4 = await user(session.id, "four") let msgs = yield* ssn.messages({ sessionID: session.id })
await createCompactionMarkerAsync(session.id) let parent = msgs.at(-1)?.info.id
expect(parent).toBeTruthy()
yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
msgs = MessageV2.filterCompacted(MessageV2.stream(session.id)) const u4 = yield* createUserMessage(session.id, "four")
parent = msgs.at(-1)?.info.id yield* createCompactionMarker(session.id)
expect(parent).toBeTruthy()
await rt.runPromise(
SessionCompaction.Service.use((svc) =>
svc.process({
parentID: parent!,
messages: msgs,
sessionID: session.id,
auto: false,
}),
),
)
const filtered = MessageV2.filterCompacted(MessageV2.stream(session.id)) msgs = MessageV2.filterCompacted(MessageV2.stream(session.id))
const ids = filtered.map((msg) => msg.info.id) parent = msgs.at(-1)?.info.id
expect(parent).toBeTruthy()
yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
expect(ids).not.toContain(u1.id) const filtered = MessageV2.filterCompacted(MessageV2.stream(session.id))
expect(ids).not.toContain(u2.id) const ids = filtered.map((msg) => msg.info.id)
expect(ids).toContain(u3.id)
expect(ids).toContain(u4.id) expect(ids).not.toContain(u1.id)
expect(filtered.some((msg) => msg.info.role === "assistant" && msg.info.summary)).toBe(true) expect(ids).not.toContain(u2.id)
expect( expect(ids).toContain(u3.id)
filtered.some((msg) => msg.info.role === "user" && msg.parts.some((part) => part.type === "compaction")), expect(ids).toContain(u4.id)
).toBe(true) expect(filtered.some((msg) => msg.info.role === "assistant" && msg.info.summary)).toBe(true)
} finally { expect(
await rt.dispose() filtered.some((msg) => msg.info.role === "user" && msg.parts.some((part) => part.type === "compaction")),
} ).toBe(true)
}, }).pipe(
}) Effect.provide(
compactionProcessLayer({ llm: stub.layer, config: cfg({ tail_turns: 2, preserve_recent_tokens: 10_000 }) }),
),
)
}) })
test("ignores previous summaries when sizing the retained tail", async () => { itProcess.instance(
await using tmp = await tmpdir() "ignores previous summaries when sizing the retained tail",
await WithInstance.provide({ Effect.gen(function* () {
directory: tmp.path, const ssn = yield* SessionNs.Service
fn: async () => { const test = yield* TestInstance
const session = await svc.create({}) const session = yield* ssn.create({})
await user(session.id, "older") yield* createUserMessage(session.id, "older")
const keep = await user(session.id, "keep this turn") const keep = yield* createUserMessage(session.id, "keep this turn")
const keepReply = await assistant(session.id, keep.id, tmp.path) const keepReply = yield* createAssistantMessage(session.id, keep.id, test.directory)
await svc.updatePart({ yield* ssn.updatePart({
id: PartID.ascending(), id: PartID.ascending(),
messageID: keepReply.id, messageID: keepReply.id,
sessionID: session.id, sessionID: session.id,
type: "text", type: "text",
text: "keep reply", text: "keep reply",
}) })
await createCompactionMarkerAsync(session.id) yield* createCompactionMarker(session.id)
const firstCompaction = (await svc.messages({ sessionID: session.id })).at(-1)?.info.id const firstCompaction = (yield* ssn.messages({ sessionID: session.id })).at(-1)?.info.id
expect(firstCompaction).toBeTruthy() expect(firstCompaction).toBeTruthy()
await summaryAssistant(session.id, firstCompaction!, tmp.path, "summary ".repeat(800)) yield* createSummaryAssistantMessage(session.id, firstCompaction!, test.directory, "summary ".repeat(800))
const recent = await user(session.id, "recent turn") const recent = yield* createUserMessage(session.id, "recent turn")
const recentReply = await assistant(session.id, recent.id, tmp.path) const recentReply = yield* createAssistantMessage(session.id, recent.id, test.directory)
await svc.updatePart({ yield* ssn.updatePart({
id: PartID.ascending(), id: PartID.ascending(),
messageID: recentReply.id, messageID: recentReply.id,
sessionID: session.id, sessionID: session.id,
type: "text", type: "text",
text: "recent reply", text: "recent reply",
}) })
await createCompactionMarkerAsync(session.id) yield* createCompactionMarker(session.id)
const msgs = yield* ssn.messages({ sessionID: session.id })
const parent = msgs.at(-1)?.info.id
expect(parent).toBeTruthy()
yield* SessionCompaction.use.process({ parentID: parent!, messages: msgs, sessionID: session.id, auto: false })
const rt = runtime("continue", Plugin.defaultLayer, wide(), cfg({ tail_turns: 2, preserve_recent_tokens: 500 })) const part = yield* readCompactionPart(session.id)
try { expect(part?.type).toBe("compaction")
const msgs = await svc.messages({ sessionID: session.id }) expect(part?.tail_start_id).toBe(keep.id)
const parent = msgs.at(-1)?.info.id }).pipe(Effect.provide(compactionProcessLayer({ config: cfg({ tail_turns: 2, preserve_recent_tokens: 500 }) }))),
expect(parent).toBeTruthy() )
await rt.runPromise(
SessionCompaction.Service.use((svc) =>
svc.process({
parentID: parent!,
messages: msgs,
sessionID: session.id,
auto: false,
}),
),
)
const part = await lastCompactionPart(session.id)
expect(part?.type).toBe("compaction")
expect(part?.tail_start_id).toBe(keep.id)
} finally {
await rt.dispose()
}
},
})
})
}) })
describe("util.token.estimate", () => { describe("util.token.estimate", () => {