vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-09-01 22:24:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(core): preserve queued controls during prompt promotion

Browse source
This commit is contained in:
Kit Langton 2026-08-31 12:38:10 -04:00
parent afd7492018
commit afcaed59c9
3 changed files with 71 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/core/src/session/inbox.ts
View file

@ -517,7 +517,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
.limit(1)
.get()
.pipe(Effect.orDie)
if (!queued) return 0
if (!queued || queued.type === "compaction" || queued.type === "move") return 0
const promoted = yield* publish(db, bus, sessionID, [queued])
const arrivedSteers = yield* pendingSteers(db, sessionID)
const control = arrivedSteers.findIndex((row) => row.type === "compaction" || row.type === "move")

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

@ -128,15 +128,16 @@ const layer = Layer.effect(
}
if (!force && !continuing && (!pending || (pending.delivery === "queue" && promotable === "steer")))
return DrainResult.Complete()
return yield* restore(
const ready = yield* restore(
Effect.gen(function* () {
const selected = yield* prepareContext(sessionID)
const promoted = yield* SessionInbox.promote(
db,
bus,
sessionID,
entering && !continuing ? promotable : "steer",
)
const scope = entering && !continuing ? promotable : "steer"
const promoted = yield* SessionInbox.promote(db, bus, sessionID, scope)
if (promoted === 0) {
// Cancellation during preparation can expose a control instead of input.
const next = yield* SessionInbox.nextPromotable(db, sessionID, scope)
if (next?.type === "compaction" || next?.type === "move") return undefined
}
if (promoted > 0 && !selected.session.parentID && SessionTitle.isUntitled(selected.session))
yield* FiberMap.run(titles, sessionID, title.generate(sessionID), {
onlyIfMissing: true,
@ -145,6 +146,7 @@ const layer = Layer.effect(
return { _tag: "Ready" as const, context: yield* context.load(selected) }
}),
)
if (ready) return ready
}
}),
),

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

@ -1422,6 +1422,67 @@ describe("SessionRunnerLLM", () => {
})
})
scenario("dispatches a queued move exposed by cancellation during preparation", function* (s) {
const runner = yield* SessionRunner.Service
const prompt = yield* s.session.prompt({ sessionID, text: "Cancel me", delivery: "queue", resume: false })
const location = Location.Ref.make({ directory: AbsolutePath.make("/moved") })
yield* s.sessionInbox.admit({
id: SessionMessage.ID.create(),
sessionID,
item: {
type: "move",
payload: { location, projectID: Project.ID.global },
delivery: "queue",
},
})
s.systemLoadHook = Effect.gen(function* () {
s.systemLoadHook = Effect.void
yield* s.session.cancelInbox({ sessionID, inboxID: prompt.id }).pipe(Effect.orDie)
})
expect(yield* runner.drain({ sessionID, force: false })).toEqual(SessionRunner.DrainResult.Moved({}))
expect((yield* s.session.get(sessionID)).location).toEqual(location)
expect(s.closedTransports).toEqual([sessionID])
expect(s.requests).toHaveLength(0)
expect(yield* s.messages).toMatchObject([{ type: "location-switched", location }])
expect(yield* s.inbox).toEqual([])
expect((yield* recordedEventTypes(sessionID)).slice(-2)).toEqual([
Bus.versionedType(SessionEvent.InboxDelivered.type, 1),
Bus.versionedType(SessionEvent.Moved.type, 1),
])
})
scenario("dispatches a queued compaction exposed by cancellation during preparation", function* (s) {
const runner = yield* SessionRunner.Service
const prompt = yield* s.session.prompt({ sessionID, text: "Cancel me", delivery: "queue", resume: false })
const compaction = yield* s.sessionInbox.admitCompaction({
id: SessionMessage.ID.create(),
sessionID,
delivery: "queue",
})
s.systemLoadHook = Effect.gen(function* () {
s.systemLoadHook = Effect.void
yield* s.session.cancelInbox({ sessionID, inboxID: prompt.id }).pipe(Effect.orDie)
})
expect(yield* runner.drain({ sessionID, force: false })).toEqual(SessionRunner.DrainResult.Complete())
expect(s.requests).toHaveLength(0)
expect(yield* s.inbox).toEqual([])
expect(yield* s.messages).toMatchObject([
{
id: compaction.id,
type: "compaction",
status: "failed",
error: { type: "compaction.unavailable", message: "Nothing to compact yet" },
},
])
expect((yield* recordedEventTypes(sessionID)).slice(-3)).toEqual([
Bus.versionedType(SessionEvent.InboxDelivered.type, 1),
Bus.versionedType(SessionEvent.Compaction.Started.type, 1),
Bus.versionedType(SessionEvent.Compaction.Failed.type, 1),
])
})
scenario("delivers a queued move atomically at the idle boundary", function* (s) {
const inboxID = SessionMessage.ID.create()
yield* s.sessionInbox.admit({