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

fix(core): steer manual compaction

Browse source
This commit is contained in:
Kit Langton 2026-07-07 20:44:34 -04:00
commit 2062a5a3a8
3 changed files with 17 additions and 9 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 admitted = yield* SessionInput.admitCompaction(db, events, {
const admission = yield* SessionInput.admitCompaction(db, events, {
id: inputID,
sessionID: input.sessionID,
}).pipe(
@ -647,8 +647,9 @@ const layer = Layer.effect(
: Effect.die(defect),
),
)
if (admission.newlyAdmitted) yield* execution.interrupt(input.sessionID)
yield* execution.wake(input.sessionID)
return admitted
return admission.entry
}),
wait: Effect.fn("V2Session.wait")(function* (sessionID) {
yield* result.get(sessionID)

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

@ -142,11 +142,12 @@ 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 exact
if (exact.type === "compaction" && exact.sessionID === input.sessionID)
return { entry: exact, newlyAdmitted: false }
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
}
const pending = yield* pendingCompaction(db, input.sessionID)
if (pending) return pending
if (pending) return { entry: pending, newlyAdmitted: false }
return yield* events
.publish(SessionEvent.Compaction.Admitted, {
inputID: input.id,
@ -158,13 +159,17 @@ 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(stored) : Effect.die(new LifecycleConflict({ id: input.id })),
stored
? Effect.succeed({ entry: stored, newlyAdmitted: stored.id === input.id })
: Effect.die(new LifecycleConflict({ id: input.id })),
),
)
}),
Effect.catchDefect((defect) =>
pendingCompaction(db, input.sessionID).pipe(
Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect))),
Effect.flatMap((stored) =>
stored ? Effect.succeed({ entry: stored, newlyAdmitted: false }) : Effect.die(defect),
),
),
),
)

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

@ -1346,7 +1346,7 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("runs one durable compaction barrier before later steer and queued prompts", () =>
it.effect("interrupts active work to run one durable compaction barrier before later prompts", () =>
Effect.gen(function* () {
yield* setup
requests.length = 0
@ -1389,7 +1389,8 @@ describe("SessionRunnerLLM", () => {
expect(yield* SessionInput.hasPending((yield* Database.Service).db, sessionID, "steer")).toBe(false)
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(active)
expect(yield* Fiber.await(active)).toMatchObject({ _tag: "Failure" })
yield* session.wait(sessionID)
expect(requests).toHaveLength(4)
expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary")
@ -1429,7 +1430,8 @@ describe("SessionRunnerLLM", () => {
resume: false,
})
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(active)
expect(yield* Fiber.await(active)).toMatchObject({ _tag: "Failure" })
yield* session.wait(sessionID)
expect(requests).toHaveLength(3)
expect(userTexts(requests[2])).toContain("Continue after failure")