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

fix(core): reset steps for promoted prompts (#33452)

Browse source
This commit is contained in:
Kit Langton 2026-06-23 01:01:14 +02:00 • committed by GitHub
commit dc468bdcfd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 143 additions and 121 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -79,7 +79,7 @@ import { MAX_STEPS_PROMPT } from "./max-steps"
* - [ ] Update title, summaries, compaction state, and cleanup in bounded background work.
*
* Use `llm.stream(request)` for each provider turn. Keep tool execution and continuation here.
* Durable activity recovery remains a separate future slice with an explicit retry policy.
* Durable continuation recovery remains a separate future slice with an explicit retry policy.
*
* The current slice loads V2 history, translates it, resolves a model through a core service, and persists one
* provider turn. Registry definitions are advertised, local tool calls are settled durably, and an
@ -142,9 +142,9 @@ export const layer = Layer.effect(
type TurnTransition =
// Automatic compaction completed; rebuild the request from compacted history.
| { readonly _tag: "ContinueAfterCompaction" }
| { readonly _tag: "ContinueAfterCompaction"; readonly step: number }
// Overflow compaction completed; rebuild once through the path without overflow recovery.
| { readonly _tag: "ContinueAfterOverflowCompaction" }
| { readonly _tag: "ContinueAfterOverflowCompaction"; readonly step: number }
class TurnTransitionError extends Error {
constructor(readonly transition: TurnTransition) {
@ -152,10 +152,9 @@ export const layer = Layer.effect(
}
}
const continueAfterCompaction = new TurnTransitionError({ _tag: "ContinueAfterCompaction" })
const continueAfterOverflowCompaction = new TurnTransitionError({
_tag: "ContinueAfterOverflowCompaction",
})
const continueAfterCompaction = (step: number) => new TurnTransitionError({ _tag: "ContinueAfterCompaction", step })
const continueAfterOverflowCompaction = (step: number) =>
new TurnTransitionError({ _tag: "ContinueAfterOverflowCompaction", step })
const loadSystemContext = (agent: AgentV2.Selection) =>
Effect.all([systemContext.load(), skillGuidance.load(agent), referenceGuidance.load()], {
@ -175,20 +174,23 @@ export const layer = Layer.effect(
const initialized = yield* SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id)
const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>()
let needsContinuation = false
let currentStep = step
if (promotion) {
const cutoff = yield* EventV2.latestSequence(db, session.id)
if (promotion === "steer") yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
let promoted = 0
if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
if (promotion === "queue") {
yield* SessionInput.promoteNextQueued(db, events, session.id)
yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id))
promoted += yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
}
if (promoted > 0) currentStep = 1
}
const system =
initialized ?? (yield* SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id))
const model = yield* models.resolve(session)
const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq)
const context = entries.map((entry) => entry.message)
const isLastStep = agent.info?.steps !== undefined && step >= agent.info.steps
const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps
const toolMaterialization = isLastStep ? undefined : yield* tools.materialize(agent.info?.permissions)
const promptCacheKey = /^ses_[0-9a-f]{64}$/.test(session.id) ? session.id.slice(4) : session.id
const request = LLM.request({
@ -202,7 +204,7 @@ export const layer = Layer.effect(
toolChoice: isLastStep ? "none" : undefined,
})
if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request }))
return yield* Effect.die(continueAfterCompaction)
return yield* Effect.die(continueAfterCompaction(currentStep))
const publisher = createLLMEventPublisher(events, {
sessionID: session.id,
agent: agent.id,
@ -272,7 +274,7 @@ export const layer = Layer.effect(
isContextOverflowFailure(overflowFailure ?? failure) &&
(yield* restore(recoverOverflow({ sessionID: session.id, entries, model, request })))
)
return yield* Effect.die(continueAfterOverflowCompaction)
return yield* Effect.die(continueAfterOverflowCompaction(currentStep))
if (overflowFailure) yield* publish(overflowFailure)
const llmFailure = failure instanceof LLMError ? failure : undefined
if (llmFailure && !publisher.hasProviderError()) {
@ -306,7 +308,7 @@ export const layer = Layer.effect(
yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true))
if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause)
if (settled._tag === "Failure") return yield* Effect.failCause(settled.cause)
return !publisher.hasProviderError() && needsContinuation
return { needsContinuation: !publisher.hasProviderError() && needsContinuation, step: currentStep }
}),
)
}, Effect.scoped)
@ -314,7 +316,7 @@ export const layer = Layer.effect(
sessionID: SessionSchema.ID,
promotion: SessionInput.Delivery | undefined,
step: number,
) => Effect.Effect<boolean, RunError>
) => Effect.Effect<{ readonly needsContinuation: boolean; readonly step: number }, RunError>
const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) {
return yield* runTurnAttempt(sessionID, promotion, step).pipe(
@ -324,7 +326,7 @@ export const layer = Layer.effect(
if (defect.transition._tag === "ContinueAfterOverflowCompaction")
return yield* Effect.die("Post-compaction provider attempt cannot recover another overflow")
yield* Effect.yieldNow
return yield* runAfterOverflowCompaction(sessionID, undefined, step)
return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step)
}),
),
)
@ -337,8 +339,8 @@ export const layer = Layer.effect(
if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect)
yield* Effect.yieldNow
if (defect.transition._tag === "ContinueAfterOverflowCompaction")
return yield* runAfterOverflowCompaction(sessionID, undefined, step)
return yield* runTurn(sessionID, undefined, step)
return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step)
return yield* runTurn(sessionID, undefined, defect.transition.step)
}),
),
)
@ -353,16 +355,19 @@ export const layer = Layer.effect(
if (!input.force && !hasSteer && !hasQueue) return
yield* failInterruptedTools(input.sessionID)
let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let openActivity = input.force || hasSteer || hasQueue
while (openActivity) {
let shouldRun = input.force || hasSteer || hasQueue
while (shouldRun) {
let needsContinuation = true
for (let step = 1; needsContinuation; step++) {
needsContinuation = yield* runTurn(input.sessionID, promotion, step)
let step = 1
while (needsContinuation) {
const result = yield* runTurn(input.sessionID, promotion, step)
needsContinuation = result.needsContinuation
step = result.step + 1
promotion = "steer"
if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
}
openActivity = yield* SessionInput.hasPending(db, input.sessionID, "queue")
promotion = openActivity ? "queue" : undefined
shouldRun = yield* SessionInput.hasPending(db, input.sessionID, "queue")
promotion = shouldRun ? "queue" : undefined
}
})

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

@ -1851,7 +1851,7 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("starts queued input after the active activity settles", () =>
it.effect("promotes queued input after continuation ends", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
@ -1883,7 +1883,7 @@ describe("SessionRunnerLLM", () => {
yield* Deferred.await(streamStarted)
yield* session.prompt({
sessionID,
prompt: new Prompt({ text: "Wait until the next activity" }),
prompt: new Prompt({ text: "Wait until continuation ends" }),
delivery: "queue",
})
yield* Deferred.succeed(streamGate, undefined)
@ -1894,7 +1894,7 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(3)
expect(userTexts(requests[0]!)).toEqual(["Start working"])
expect(userTexts(requests[1]!)).toEqual(["Start working"])
expect(userTexts(requests[2]!)).toEqual(["Start working", "Wait until the next activity"])
expect(userTexts(requests[2]!)).toEqual(["Start working", "Wait until continuation ends"])
}),
)
@ -1984,7 +1984,7 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("runs queued active inputs as separate FIFO activities", () =>
it.effect("promotes queued inputs one at a time in FIFO order", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
@ -2027,14 +2027,14 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("opens queued input after idle steering activity settles", () =>
it.effect("promotes queued input after steering continuation ends", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Start steering activity" }), resume: false })
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Start steering" }), resume: false })
yield* session.prompt({
sessionID,
prompt: new Prompt({ text: "Queue later activity" }),
prompt: new Prompt({ text: "Queue for later" }),
delivery: "queue",
resume: false,
})
@ -2056,12 +2056,12 @@ describe("SessionRunnerLLM", () => {
yield* session.resume(sessionID)
expect(requests).toHaveLength(2)
expect(userTexts(requests[0]!)).toEqual(["Start steering activity"])
expect(userTexts(requests[1]!)).toEqual(["Start steering activity", "Queue later activity"])
expect(userTexts(requests[0]!)).toEqual(["Start steering"])
expect(userTexts(requests[1]!)).toEqual(["Start steering", "Queue for later"])
}),
)
it.effect("coalesces steers into the active queued activity before starting the next queued activity", () =>
it.effect("promotes steers before the next queued input", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
@ -2101,8 +2101,8 @@ describe("SessionRunnerLLM", () => {
streamGate = secondGate
yield* Deferred.succeed(firstGate, undefined)
while (requests.length < 2) yield* Effect.yieldNow
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Steer first queued activity" }) })
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Also steer first queued activity" }) })
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Steer before next queued input" }) })
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Also steer before next queued input" }) })
yield* Deferred.succeed(secondGate, undefined)
yield* Fiber.join(first)
streamGate = undefined
@ -2113,14 +2113,14 @@ describe("SessionRunnerLLM", () => {
expect(userTexts(requests[2]!)).toEqual([
"Start working",
"Queue first",
"Steer first queued activity",
"Also steer first queued activity",
"Steer before next queued input",
"Also steer before next queued input",
])
expect(userTexts(requests[3]!)).toEqual([
"Start working",
"Queue first",
"Steer first queued activity",
"Also steer first queued activity",
"Steer before next queued input",
"Also steer before next queued input",
"Queue second",
])
}),
@ -2354,13 +2354,13 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("starts the first queued activity when woken while idle", () =>
it.effect("promotes the first queued input when woken while idle", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({
sessionID,
prompt: new Prompt({ text: "Wait for fresh activity" }),
prompt: new Prompt({ text: "Wait in queue" }),
delivery: "queue",
resume: false,
})
@ -2370,30 +2370,7 @@ describe("SessionRunnerLLM", () => {
yield* Effect.yieldNow
expect(requests).toHaveLength(1)
expect(userTexts(requests[0]!)).toEqual(["Wait for fresh activity"])
}),
)
it.effect("does not spend one activity step budget across queued activities", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const queued = Array.from({ length: 26 }, (_, index) => `Queued activity ${index + 1}`)
for (const text of queued) {
yield* session.prompt({ sessionID, prompt: new Prompt({ text }), delivery: "queue", resume: false })
}
requests.length = 0
responses = queued.map(() => [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.stepFinish({ index: 0, reason: "stop" }),
LLMEvent.finish({ reason: "stop" }),
])
yield* session.resume(sessionID)
expect(requests).toHaveLength(queued.length)
expect(userTexts(requests.at(-1)!)).toEqual(queued)
expect(userTexts(requests[0]!)).toEqual(["Wait in queue"])
}),
)
@ -2768,7 +2745,7 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("interrupts a blocked provider turn without local tool activity", () =>
it.effect("interrupts a blocked provider turn without local tool execution", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
@ -2828,38 +2805,6 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("continues past 25 local tool steps when the agent has no step limit", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Loop forever" }), resume: false })
requests.length = 0
authorizations.length = 0
executions.length = 0
streamGate = undefined
streamStarted = undefined
responses = [
...Array.from({ length: 25 }, (_, index) => [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: `call-echo-${index}`, name: "echo", input: { text: `${index}` } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
]),
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.stepFinish({ index: 0, reason: "stop" }),
LLMEvent.finish({ reason: "stop" }),
],
]
yield* session.resume(sessionID)
expect(requests).toHaveLength(26)
expect(executions).toHaveLength(25)
}),
)
it.effect("forces a text response on an agent's configured final step", () =>
Effect.gen(function* () {
yield* setup
@ -2908,6 +2853,58 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("resets the configured step allowance when steering input promotes", () =>
Effect.gen(function* () {
yield* setup
const agents = yield* AgentV2.Service
yield* agents.transform((editor) =>
editor.update(AgentV2.ID.make("build"), (agent) => {
agent.steps = 2
}),
)
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Start work" }), resume: false })
requests.length = 0
executions.length = 0
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-before-steer", name: "echo", input: { text: "before" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-after-steer", name: "echo", input: { text: "after" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.stepFinish({ index: 0, reason: "stop" }),
LLMEvent.finish({ reason: "stop" }),
],
]
streamGate = yield* Deferred.make<void>()
streamStarted = yield* Deferred.make<void>()
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Change direction" }) })
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(run)
streamGate = undefined
streamStarted = undefined
expect(requests).toHaveLength(3)
expect(requests[1]?.toolChoice).toBeUndefined()
expect(requests[1]?.tools).not.toEqual([])
expect(requests[2]?.toolChoice).toMatchObject({ type: "none" })
expect(executions).toEqual(["before", "after"])
}),
)
it.effect("projects provider errors as terminal assistant step failures", () =>
Effect.gen(function* () {
yield* setup

4
packages/sdk/js/src/v2/gen/sdk.gen.ts
View file

@ -5344,7 +5344,7 @@ export class Session3 extends HeyApiClient {
/**
* Switch session agent
*
* Switch the agent used by subsequent session activity.
* Switch the agent used by subsequent provider turns.
*/
public switchAgent<ThrowOnError extends boolean = false>(
parameters: {
@ -5383,7 +5383,7 @@ export class Session3 extends HeyApiClient {
/**
* Switch session model
*
* Switch the model used by subsequent session activity.
* Switch the model used by subsequent provider turns.
*/
public switchModel<ThrowOnError extends boolean = false>(
parameters: {

4
packages/sdk/openapi.json
View file

@ -10383,7 +10383,7 @@
}
}
},
"description": "Switch the agent used by subsequent session activity.",
"description": "Switch the agent used by subsequent provider turns.",
"summary": "Switch session agent",
"requestBody": {
"content": {
@ -10468,7 +10468,7 @@
}
}
},
"description": "Switch the model used by subsequent session activity.",
"description": "Switch the model used by subsequent provider turns.",
"summary": "Switch session model",
"requestBody": {
"content": {

4
packages/server/src/groups/session.ts
View file

@ -152,7 +152,7 @@ export const SessionGroup = HttpApiGroup.make("server.session")
OpenApi.annotations({
identifier: "v2.session.switchAgent",
summary: "Switch session agent",
description: "Switch the agent used by subsequent session activity.",
description: "Switch the agent used by subsequent provider turns.",
}),
),
)
@ -168,7 +168,7 @@ export const SessionGroup = HttpApiGroup.make("server.session")
OpenApi.annotations({
identifier: "v2.session.switchModel",
summary: "Switch session model",
description: "Switch the model used by subsequent session activity.",
description: "Switch the model used by subsequent provider turns.",
}),
),
)