From 9a2acdd372fcffa4dba9719e838ef5b8d9fed7d0 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sun, 21 Jun 2026 17:05:55 -0400 Subject: [PATCH 1/3] refactor(core): simplify runner transitions --- packages/core/src/session/runner/llm.ts | 132 ++++++++++++------------ 1 file changed, 65 insertions(+), 67 deletions(-) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 5d84e985a6..5d2ce003d9 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -140,29 +140,35 @@ export const layer = Layer.effect( const isQuestionRejected = (cause: Cause.Cause) => cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) - type TurnTransition = + type RebuildResult = // Request preparation observed a concurrent Session change and must restart from durable state. - | { readonly _tag: "RebuildPreparedTurn"; readonly promotion?: SessionInput.Delivery } + { readonly _tag: "Rebuild"; readonly nextPromotion: SessionInput.Delivery | undefined } + type TurnResult = + | RebuildResult // Overflow compaction completed; rebuild once through the path without overflow recovery. - | { readonly _tag: "ContinueAfterOverflowCompaction" } + | { readonly _tag: "OverflowCompacted" } + | { readonly _tag: "Complete"; readonly needsContinuation: boolean } - class TurnTransitionError extends Error { - constructor(readonly transition: TurnTransition) { - super() - } - } - - const rebuildPreparedTurn = (promotion?: SessionInput.Delivery) => - new TurnTransitionError({ _tag: "RebuildPreparedTurn", promotion }) - const continueAfterOverflowCompaction = new TurnTransitionError({ - _tag: "ContinueAfterOverflowCompaction", + const rebuild = (nextPromotion: SessionInput.Delivery | undefined): RebuildResult => ({ + _tag: "Rebuild", + nextPromotion, }) + const overflowCompacted = (): TurnResult => ({ _tag: "OverflowCompacted" }) + const complete = (needsContinuation: boolean): TurnResult => ({ _tag: "Complete", needsContinuation }) - const retryAgentMismatch = (promotion: SessionInput.Delivery | undefined) => - Effect.catchDefect((defect) => - defect instanceof SessionContextEpoch.AgentMismatch - ? Effect.die(rebuildPreparedTurn(promotion)) - : Effect.die(defect), + type ContextResult = { readonly _tag: "Ready"; readonly value: A } | RebuildResult + + const rebuildOnAgentMismatch = ( + effect: Effect.Effect, + nextPromotion: SessionInput.Delivery | undefined, + ): Effect.Effect, E> => + effect.pipe( + Effect.map((value): ContextResult => ({ _tag: "Ready", value })), + Effect.catchDefect((defect) => + defect instanceof SessionContextEpoch.AgentMismatch + ? Effect.succeed(rebuild(nextPromotion)) + : Effect.die(defect), + ), ) const sameModel = Schema.toEquivalence(Schema.UndefinedOr(ModelV2.Ref)) @@ -171,7 +177,14 @@ export const layer = Layer.effect( concurrency: "unbounded", }).pipe(Effect.map(SystemContext.combine)) - const runTurnAttempt = Effect.fn("SessionRunner.runTurn")(function* ( + type RunTurnAttempt = ( + sessionID: SessionSchema.ID, + promotion: SessionInput.Delivery | undefined, + step: number, + recoverOverflow?: typeof compaction.compactAfterOverflow, + ) => Effect.Effect + + const runTurnAttempt: RunTurnAttempt = Effect.fn("SessionRunner.runTurn")(function* ( sessionID: SessionSchema.ID, promotion: SessionInput.Delivery | undefined, step: number, @@ -181,13 +194,12 @@ export const layer = Layer.effect( if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) return yield* Effect.interrupt const agent = yield* agents.select(session.agent) - const initialized = yield* SessionContextEpoch.initialize( - db, - loadSystemContext(agent), - session.id, - session.location, - agent.id, - ).pipe(retryAgentMismatch(promotion)) + const initialization = yield* rebuildOnAgentMismatch( + SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id, session.location, agent.id), + promotion, + ) + if (initialization._tag !== "Ready") return initialization + const initialized = initialization.value const toolFibers = yield* FiberSet.make() let needsContinuation = false if (promotion) { @@ -198,19 +210,18 @@ export const layer = Layer.effect( yield* SessionInput.promoteSteers(db, events, session.id, cutoff) } } - const system = - initialized ?? - (yield* SessionContextEpoch.prepare( - db, - events, - loadSystemContext(agent), - session.id, - session.location, - agent.id, - ).pipe(retryAgentMismatch(undefined))) + const preparation = + initialized === undefined + ? yield* rebuildOnAgentMismatch( + SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id, session.location, agent.id), + undefined, + ) + : ({ _tag: "Ready", value: initialized } as const) + if (preparation._tag !== "Ready") return preparation + const system = preparation.value const current = yield* getSession(sessionID) if ((yield* agents.select(current.agent)).id !== agent.id || !sameModel(current.model, session.model)) - return yield* Effect.die(rebuildPreparedTurn()) + return rebuild(undefined) const model = yield* models.resolve(session) const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) const context = entries.map((entry) => entry.message) @@ -228,7 +239,7 @@ export const layer = Layer.effect( toolChoice: isLastStep ? "none" : undefined, }) if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) - return yield* Effect.die(rebuildPreparedTurn()) + return rebuild(undefined) const publisher = createLLMEventPublisher(events, { sessionID: session.id, agent: agent.id, @@ -242,8 +253,7 @@ export const layer = Layer.effect( const publish = (event: LLMEvent, outputPaths: ReadonlyArray = []) => withPublication(publisher.publish(event, outputPaths)) let overflowFailure: ProviderErrorEvent | undefined - if (!(yield* SessionContextEpoch.current(db, session.id, agent.id, system.revision))) - return yield* Effect.die(rebuildPreparedTurn()) + if (!(yield* SessionContextEpoch.current(db, session.id, agent.id, system.revision))) return rebuild(undefined) const providerStream = llm.stream(request).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -300,7 +310,7 @@ export const layer = Layer.effect( isContextOverflowFailure(overflowFailure ?? failure) && (yield* restore(recoverOverflow({ sessionID: session.id, entries, model, request }))) ) - return yield* Effect.die(continueAfterOverflowCompaction) + return overflowCompacted() if (overflowFailure) yield* publish(overflowFailure) const llmFailure = failure instanceof LLMError ? failure : undefined if (llmFailure && !publisher.hasProviderError()) { @@ -334,7 +344,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 complete(!publisher.hasProviderError() && needsContinuation) }), ) }, Effect.scoped) @@ -344,32 +354,20 @@ export const layer = Layer.effect( step: number, ) => Effect.Effect - const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - return yield* runTurnAttempt(sessionID, promotion, step).pipe( - Effect.catchDefect( - Effect.fnUntraced(function* (defect) { - if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) - if (defect.transition._tag === "ContinueAfterOverflowCompaction") - return yield* Effect.die("Post-compaction provider attempt cannot recover another overflow") - yield* Effect.yieldNow - return yield* runAfterOverflowCompaction(sessionID, defect.transition.promotion, step) - }), - ), - ) - }) - const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow).pipe( - Effect.catchDefect( - Effect.fnUntraced(function* (defect) { - 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, defect.transition.promotion, step) - }), - ), - ) + let nextPromotion = promotion + let recoverOverflow: typeof compaction.compactAfterOverflow | undefined = compaction.compactAfterOverflow + while (true) { + const result = yield* runTurnAttempt(sessionID, nextPromotion, step, recoverOverflow) + if (result._tag === "Complete") return result.needsContinuation + yield* Effect.yieldNow + if (result._tag === "OverflowCompacted") { + nextPromotion = undefined + recoverOverflow = undefined + continue + } + nextPromotion = result.nextPromotion + } }) const run = Effect.fn("SessionRunner.run")(function* (input: { From ebe2501d48004eb5222e194aab6b53db6c0f4b78 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sun, 21 Jun 2026 19:49:10 -0400 Subject: [PATCH 2/3] refactor(core): model runner transition data --- packages/core/src/session/runner/llm.ts | 57 +++++++++---------------- 1 file changed, 21 insertions(+), 36 deletions(-) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 5d2ce003d9..ec5c5246b2 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -8,7 +8,7 @@ import { isContextOverflowFailure, type ProviderErrorEvent, } from "@opencode-ai/llm" -import { Cause, DateTime, Effect, FiberSet, Layer, Option, Schema, Semaphore, Stream } from "effect" +import { Cause, Data, DateTime, Effect, FiberSet, Layer, Option, Schema, Semaphore, Stream } from "effect" import { AgentV2 } from "../../agent" import { Config } from "../../config" import { Database } from "../../database/database" @@ -140,34 +140,20 @@ export const layer = Layer.effect( const isQuestionRejected = (cause: Cause.Cause) => cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) - type RebuildResult = + type TurnResult = Data.TaggedEnum<{ // Request preparation observed a concurrent Session change and must restart from durable state. - { readonly _tag: "Rebuild"; readonly nextPromotion: SessionInput.Delivery | undefined } - type TurnResult = - | RebuildResult + Rebuild: { readonly nextPromotion: SessionInput.Delivery | undefined } // Overflow compaction completed; rebuild once through the path without overflow recovery. - | { readonly _tag: "OverflowCompacted" } - | { readonly _tag: "Complete"; readonly needsContinuation: boolean } + OverflowCompacted: {} + Complete: { readonly needsContinuation: boolean } + }> + const TurnResult = Data.taggedEnum() - const rebuild = (nextPromotion: SessionInput.Delivery | undefined): RebuildResult => ({ - _tag: "Rebuild", - nextPromotion, - }) - const overflowCompacted = (): TurnResult => ({ _tag: "OverflowCompacted" }) - const complete = (needsContinuation: boolean): TurnResult => ({ _tag: "Complete", needsContinuation }) - - type ContextResult = { readonly _tag: "Ready"; readonly value: A } | RebuildResult - - const rebuildOnAgentMismatch = ( - effect: Effect.Effect, - nextPromotion: SessionInput.Delivery | undefined, - ): Effect.Effect, E> => + const optionOnAgentMismatch = (effect: Effect.Effect): Effect.Effect, E> => effect.pipe( - Effect.map((value): ContextResult => ({ _tag: "Ready", value })), + Effect.map(Option.some), Effect.catchDefect((defect) => - defect instanceof SessionContextEpoch.AgentMismatch - ? Effect.succeed(rebuild(nextPromotion)) - : Effect.die(defect), + defect instanceof SessionContextEpoch.AgentMismatch ? Effect.succeed(Option.none()) : Effect.die(defect), ), ) @@ -194,11 +180,10 @@ export const layer = Layer.effect( if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) return yield* Effect.interrupt const agent = yield* agents.select(session.agent) - const initialization = yield* rebuildOnAgentMismatch( + const initialization = yield* optionOnAgentMismatch( SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id, session.location, agent.id), - promotion, ) - if (initialization._tag !== "Ready") return initialization + if (Option.isNone(initialization)) return TurnResult.Rebuild({ nextPromotion: promotion }) const initialized = initialization.value const toolFibers = yield* FiberSet.make() let needsContinuation = false @@ -212,16 +197,15 @@ export const layer = Layer.effect( } const preparation = initialized === undefined - ? yield* rebuildOnAgentMismatch( + ? yield* optionOnAgentMismatch( SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id, session.location, agent.id), - undefined, ) - : ({ _tag: "Ready", value: initialized } as const) - if (preparation._tag !== "Ready") return preparation + : Option.some(initialized) + if (Option.isNone(preparation)) return TurnResult.Rebuild({ nextPromotion: undefined }) const system = preparation.value const current = yield* getSession(sessionID) if ((yield* agents.select(current.agent)).id !== agent.id || !sameModel(current.model, session.model)) - return rebuild(undefined) + return TurnResult.Rebuild({ nextPromotion: undefined }) const model = yield* models.resolve(session) const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) const context = entries.map((entry) => entry.message) @@ -239,7 +223,7 @@ export const layer = Layer.effect( toolChoice: isLastStep ? "none" : undefined, }) if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) - return rebuild(undefined) + return TurnResult.Rebuild({ nextPromotion: undefined }) const publisher = createLLMEventPublisher(events, { sessionID: session.id, agent: agent.id, @@ -253,7 +237,8 @@ export const layer = Layer.effect( const publish = (event: LLMEvent, outputPaths: ReadonlyArray = []) => withPublication(publisher.publish(event, outputPaths)) let overflowFailure: ProviderErrorEvent | undefined - if (!(yield* SessionContextEpoch.current(db, session.id, agent.id, system.revision))) return rebuild(undefined) + if (!(yield* SessionContextEpoch.current(db, session.id, agent.id, system.revision))) + return TurnResult.Rebuild({ nextPromotion: undefined }) const providerStream = llm.stream(request).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -310,7 +295,7 @@ export const layer = Layer.effect( isContextOverflowFailure(overflowFailure ?? failure) && (yield* restore(recoverOverflow({ sessionID: session.id, entries, model, request }))) ) - return overflowCompacted() + return TurnResult.OverflowCompacted() if (overflowFailure) yield* publish(overflowFailure) const llmFailure = failure instanceof LLMError ? failure : undefined if (llmFailure && !publisher.hasProviderError()) { @@ -344,7 +329,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 complete(!publisher.hasProviderError() && needsContinuation) + return TurnResult.Complete({ needsContinuation: !publisher.hasProviderError() && needsContinuation }) }), ) }, Effect.scoped) From 497be8941e718855c2af87b3660599e4f37207ca Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sun, 21 Jun 2026 19:53:11 -0400 Subject: [PATCH 3/3] refactor(core): type runner retry control flow --- packages/core/src/session/context-epoch.ts | 20 +++--- packages/core/src/session/runner/llm.ts | 77 ++++++++++++---------- packages/core/test/session-runner.test.ts | 2 +- 3 files changed, 53 insertions(+), 46 deletions(-) diff --git a/packages/core/src/session/context-epoch.ts b/packages/core/src/session/context-epoch.ts index 1fb8df92e6..2c9a1ebeaf 100644 --- a/packages/core/src/session/context-epoch.ts +++ b/packages/core/src/session/context-epoch.ts @@ -18,7 +18,7 @@ type DatabaseService = Database.Interface["db"] class RevisionMismatch extends Error {} class LocationMismatch extends Error {} -export class AgentMismatch extends Error {} +export class AgentMismatch extends Schema.TaggedErrorClass()("SessionContextEpoch.AgentMismatch", {}) {} export class AgentReplacementBlocked extends Schema.TaggedErrorClass()( "SessionContextEpoch.AgentReplacementBlocked", { sessionID: SessionSchema.ID, previous: AgentV2.ID, current: AgentV2.ID }, @@ -45,7 +45,7 @@ export function initialize( sessionID: SessionSchema.ID, location: Location.Ref, agent: AgentV2.ID, -): Effect.Effect { +): Effect.Effect { return retryRevisionMismatch(() => initializeOnce(db, context, sessionID, location, agent)).pipe( Effect.withSpan("SessionContextEpoch.initialize"), ) @@ -58,7 +58,10 @@ export function prepare( sessionID: SessionSchema.ID, location: Location.Ref, agent: AgentV2.ID, -): Effect.Effect { +): Effect.Effect< + Prepared, + SystemContext.InitializationBlocked | ContextSnapshotDecodeError | AgentMismatch | AgentReplacementBlocked +> { return retryRevisionMismatch(() => prepareOnce(db, events, context, sessionID, location, agent)).pipe( Effect.withSpan("SessionContextEpoch.prepare"), ) @@ -153,7 +156,7 @@ const requireAgentSelection = Effect.fnUntraced(function* ( .where(eq(SessionTable.id, sessionID)) .get() .pipe(Effect.orDie) - if (!selected || (selected.agent !== null && selected.agent !== agent)) return yield* Effect.die(new AgentMismatch()) + if (!selected || (selected.agent !== null && selected.agent !== agent)) return yield* new AgentMismatch({}) }) export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacement")(function* ( @@ -212,7 +215,7 @@ const insert = Effect.fnUntraced(function* ( .get() .pipe(Effect.orDie) if (!placed) return yield* Effect.die(new LocationMismatch()) - if (placed.agent !== null && placed.agent !== agent) return yield* Effect.die(new AgentMismatch()) + if (placed.agent !== null && placed.agent !== agent) return yield* new AgentMismatch({}) const baselineSeq = yield* SessionInput.latestSeq(db, sessionID) yield* db .insert(SessionContextEpochTable) @@ -235,7 +238,7 @@ const insert = Effect.fnUntraced(function* ( }), { behavior: "immediate" }, ) - .pipe(Effect.orDie) + .pipe(Effect.catch((error) => (error instanceof AgentMismatch ? Effect.fail(error) : Effect.die(error)))) }) const replace = Effect.fnUntraced(function* ( @@ -274,7 +277,7 @@ const replace = Effect.fnUntraced(function* ( }), { behavior: "immediate" }, ) - .pipe(Effect.orDie) + .pipe(Effect.catch((error) => (error instanceof AgentMismatch ? Effect.fail(error) : Effect.die(error)))) }) const fence = Effect.fnUntraced(function* ( @@ -290,8 +293,7 @@ const fence = Effect.fnUntraced(function* ( .where(eq(SessionContextEpochTable.session_id, sessionID)) .get() .pipe(Effect.orDie) - if (!current || (current.selected !== null && current.selected !== agent)) - return yield* Effect.die(new AgentMismatch()) + if (!current || (current.selected !== null && current.selected !== agent)) return yield* new AgentMismatch({}) if (current.revision !== expectedRevision) return yield* Effect.die(new RevisionMismatch()) }) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index ec5c5246b2..8eee294594 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -36,6 +36,18 @@ import { createLLMEventPublisher } from "./publish-llm-event" import { toLLMMessages } from "./to-llm-message" import { MAX_STEPS_PROMPT } from "./max-steps" +type TurnState = Data.TaggedEnum<{ + OverflowAvailable: { readonly promotion: SessionInput.Delivery | undefined } + OverflowExhausted: {} +}> +const TurnState = Data.taggedEnum() + +type TurnResult = Data.TaggedEnum<{ + Retry: { readonly state: TurnState } + Complete: { readonly needsContinuation: boolean } +}> +const TurnResult = Data.taggedEnum() + /** * Runs one durable coding-agent Session until it settles. * @@ -140,21 +152,12 @@ export const layer = Layer.effect( const isQuestionRejected = (cause: Cause.Cause) => cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) - type TurnResult = Data.TaggedEnum<{ - // Request preparation observed a concurrent Session change and must restart from durable state. - Rebuild: { readonly nextPromotion: SessionInput.Delivery | undefined } - // Overflow compaction completed; rebuild once through the path without overflow recovery. - OverflowCompacted: {} - Complete: { readonly needsContinuation: boolean } - }> - const TurnResult = Data.taggedEnum() - - const optionOnAgentMismatch = (effect: Effect.Effect): Effect.Effect, E> => + const optionOnAgentMismatch = ( + effect: Effect.Effect, + ): Effect.Effect, E, R> => effect.pipe( - Effect.map(Option.some), - Effect.catchDefect((defect) => - defect instanceof SessionContextEpoch.AgentMismatch ? Effect.succeed(Option.none()) : Effect.die(defect), - ), + Effect.asSome, + Effect.catchTag("SessionContextEpoch.AgentMismatch", () => Effect.succeedNone), ) const sameModel = Schema.toEquivalence(Schema.UndefinedOr(ModelV2.Ref)) @@ -165,17 +168,16 @@ export const layer = Layer.effect( type RunTurnAttempt = ( sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, + state: TurnState, step: number, - recoverOverflow?: typeof compaction.compactAfterOverflow, ) => Effect.Effect const runTurnAttempt: RunTurnAttempt = Effect.fn("SessionRunner.runTurn")(function* ( sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, + state: TurnState, step: number, - recoverOverflow?: typeof compaction.compactAfterOverflow, ) { + const promotion = state._tag === "OverflowAvailable" ? state.promotion : undefined const session = yield* getSession(sessionID) if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) return yield* Effect.interrupt @@ -183,7 +185,7 @@ export const layer = Layer.effect( const initialization = yield* optionOnAgentMismatch( SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id, session.location, agent.id), ) - if (Option.isNone(initialization)) return TurnResult.Rebuild({ nextPromotion: promotion }) + if (Option.isNone(initialization)) return TurnResult.Retry({ state }) const initialized = initialization.value const toolFibers = yield* FiberSet.make() let needsContinuation = false @@ -201,11 +203,13 @@ export const layer = Layer.effect( SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id, session.location, agent.id), ) : Option.some(initialized) - if (Option.isNone(preparation)) return TurnResult.Rebuild({ nextPromotion: undefined }) + const nextState = + state._tag === "OverflowAvailable" ? TurnState.OverflowAvailable({ promotion: undefined }) : state + if (Option.isNone(preparation)) return TurnResult.Retry({ state: nextState }) const system = preparation.value const current = yield* getSession(sessionID) if ((yield* agents.select(current.agent)).id !== agent.id || !sameModel(current.model, session.model)) - return TurnResult.Rebuild({ nextPromotion: undefined }) + return TurnResult.Retry({ state: nextState }) const model = yield* models.resolve(session) const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) const context = entries.map((entry) => entry.message) @@ -223,7 +227,7 @@ export const layer = Layer.effect( toolChoice: isLastStep ? "none" : undefined, }) if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) - return TurnResult.Rebuild({ nextPromotion: undefined }) + return TurnResult.Retry({ state: nextState }) const publisher = createLLMEventPublisher(events, { sessionID: session.id, agent: agent.id, @@ -238,7 +242,7 @@ export const layer = Layer.effect( withPublication(publisher.publish(event, outputPaths)) let overflowFailure: ProviderErrorEvent | undefined if (!(yield* SessionContextEpoch.current(db, session.id, agent.id, system.revision))) - return TurnResult.Rebuild({ nextPromotion: undefined }) + return TurnResult.Retry({ state: nextState }) const providerStream = llm.stream(request).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -290,12 +294,12 @@ export const layer = Layer.effect( const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined if ( - recoverOverflow && + state._tag === "OverflowAvailable" && !publisher.hasAssistantStarted() && isContextOverflowFailure(overflowFailure ?? failure) && - (yield* restore(recoverOverflow({ sessionID: session.id, entries, model, request }))) + (yield* restore(compaction.compactAfterOverflow({ sessionID: session.id, entries, model, request }))) ) - return TurnResult.OverflowCompacted() + return TurnResult.Retry({ state: TurnState.OverflowExhausted() }) if (overflowFailure) yield* publish(overflowFailure) const llmFailure = failure instanceof LLMError ? failure : undefined if (llmFailure && !publisher.hasProviderError()) { @@ -340,18 +344,19 @@ export const layer = Layer.effect( ) => Effect.Effect const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - let nextPromotion = promotion - let recoverOverflow: typeof compaction.compactAfterOverflow | undefined = compaction.compactAfterOverflow + let state: TurnState = TurnState.OverflowAvailable({ promotion }) while (true) { - const result = yield* runTurnAttempt(sessionID, nextPromotion, step, recoverOverflow) - if (result._tag === "Complete") return result.needsContinuation - yield* Effect.yieldNow - if (result._tag === "OverflowCompacted") { - nextPromotion = undefined - recoverOverflow = undefined - continue + const result: TurnResult = yield* runTurnAttempt(sessionID, state, step) + switch (result._tag) { + case "Complete": + return result.needsContinuation + case "Retry": + state = result.state + yield* Effect.yieldNow + break + default: + return result satisfies never } - nextPromotion = result.nextPromotion } }) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index eb5ccb277d..e93fc65122 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -1142,7 +1142,7 @@ describe("SessionRunnerLLM", () => { sessionID, location, AgentV2.defaultID, - ).pipe(Effect.catchDefect(Effect.succeed)), + ).pipe(Effect.flip), ).toBeInstanceOf(SessionContextEpoch.AgentMismatch) expect(