ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(core): prioritize manual compaction

Browse source
This commit is contained in:
Kit Langton 2026-07-08 11:38:43 -04:00
commit 219ded0330
4 changed files with 26 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

5
packages/core/src/session.ts
View file

@ -637,7 +637,7 @@ const layer = Layer.effect(
compact: Effect.fn("V2Session.compact")(function* (input) {
yield* result.get(input.sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admission = yield* SessionInput.admitCompaction(db, events, {
const admitted = yield* SessionInput.admitCompaction(db, events, {
id: inputID,
sessionID: input.sessionID,
}).pipe(
@ -647,9 +647,8 @@ const layer = Layer.effect(
: Effect.die(defect),
),
)
if (admission.newlyAdmitted) yield* execution.interrupt(input.sessionID)
yield* execution.wake(input.sessionID)
return admission.entry
return admitted
}),
wait: Effect.fn("V2Session.wait")(function* (sessionID) {
yield* result.get(sessionID)

13
packages/core/src/session/input.ts
View file

@ -142,12 +142,11 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
Effect.gen(function* () {
const exact = yield* find(db, input.id)
if (exact) {
if (exact.type === "compaction" && exact.sessionID === input.sessionID)
return { entry: exact, newlyAdmitted: false }
if (exact.type === "compaction" && exact.sessionID === input.sessionID) return exact
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
}
const pending = yield* pendingCompaction(db, input.sessionID)
if (pending) return { entry: pending, newlyAdmitted: false }
if (pending) return pending
return yield* events
.publish(SessionEvent.Compaction.Admitted, {
inputID: input.id,
@ -159,17 +158,13 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
return Effect.die(new Error("Compaction admission event is missing aggregate sequence"))
return pendingCompaction(db, input.sessionID).pipe(
Effect.flatMap((stored) =>
stored
? Effect.succeed({ entry: stored, newlyAdmitted: stored.id === input.id })
: Effect.die(new LifecycleConflict({ id: input.id })),
stored ? Effect.succeed(stored) : Effect.die(new LifecycleConflict({ id: input.id })),
),
)
}),
Effect.catchDefect((defect) =>
pendingCompaction(db, input.sessionID).pipe(
Effect.flatMap((stored) =>
stored ? Effect.succeed({ entry: stored, newlyAdmitted: false }) : Effect.die(defect),
),
Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect))),
),
),
)

7
packages/core/src/session/runner/llm.ts
View file

@ -624,13 +624,10 @@ const layer = Layer.effect(
}
needsContinuation = result.needsContinuation
step = result.step + 1
if (needsContinuation) {
promotion = (yield* SessionInput.pendingCompaction(db, input.sessionID)) ? undefined : "steer"
continue
}
// Manual compaction is a barrier: handle it before continuing or promoting steers.
yield* runPendingCompaction(input.sessionID)
promotion = "steer"
needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
}
yield* runPendingCompaction(input.sessionID)
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")

26
packages/core/test/session-runner.test.ts
View file

@ -1346,16 +1346,22 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("interrupts active work to run one durable compaction barrier before later prompts", () =>
it.effect("runs compaction at the next step boundary before continuation, steer, and queue", () =>
Effect.gen(function* () {
yield* setup
requests.length = 0
executions.length = 0
currentModel = recoveryModel
const session = yield* SessionV2.Service
streamGate = yield* Deferred.make<void>()
streamStarted = yield* Deferred.make<void>()
responses = [
fragmentFixture("text", "text-active", ["Active complete"]).completeEvents,
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-before-compaction", name: "echo", input: { text: "before compaction" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[LLMEvent.textDelta({ id: "summary", text: "durable summary" })],
fragmentFixture("text", "text-steer", ["Steer complete"]).completeEvents,
fragmentFixture("text", "text-queue", ["Queue complete"]).completeEvents,
@ -1365,8 +1371,12 @@ describe("SessionRunnerLLM", () => {
yield* Deferred.await(streamStarted)
const first = yield* session.compact({ sessionID })
const second = yield* session.compact({ sessionID })
expect(second.id).toBe(first.id)
const [exactRetry, coalesced] = yield* Effect.all(
[session.compact({ id: first.id, sessionID }), session.compact({ sessionID })],
{ concurrency: "unbounded" },
)
expect(exactRetry.id).toBe(first.id)
expect(coalesced.id).toBe(first.id)
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toMatchObject({
id: first.id,
})
@ -1389,10 +1399,11 @@ describe("SessionRunnerLLM", () => {
expect(yield* SessionInput.hasPending((yield* Database.Service).db, sessionID, "steer")).toBe(false)
yield* Deferred.succeed(streamGate, undefined)
expect(yield* Fiber.await(active)).toMatchObject({ _tag: "Failure" })
yield* session.wait(sessionID)
yield* Fiber.join(active)
expect(requests).toHaveLength(4)
expect(executions).toEqual(["before compaction"])
expect(userTexts(requests[0])).toContain("Active work")
expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary")
expect(userTexts(requests[2])).toContain("Steer after compaction")
expect(userTexts(requests[3])).toContain("Queue after compaction")
@ -1430,8 +1441,7 @@ describe("SessionRunnerLLM", () => {
resume: false,
})
yield* Deferred.succeed(streamGate, undefined)
expect(yield* Fiber.await(active)).toMatchObject({ _tag: "Failure" })
yield* session.wait(sessionID)
yield* Fiber.join(active)
expect(requests).toHaveLength(3)
expect(userTexts(requests[2])).toContain("Continue after failure")