From 4f6a2c5b696c9d53a6537deb4c52910603b7e8d1 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:00:04 -0400 Subject: [PATCH 01/10] refactor(core): isolate provider turn runner --- packages/core/src/session/runner/llm.ts | 287 +----------------- packages/core/src/session/runner/run-turn.ts | 288 +++++++++++++++++++ 2 files changed, 295 insertions(+), 280 deletions(-) create mode 100644 packages/core/src/session/runner/run-turn.ts diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 88ba79098a..c895558e02 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -1,37 +1,12 @@ -import { - LLM, - LLMClient, - LLMError, - LLMEvent, - SystemPart, - isContextOverflowFailure, - type ProviderErrorEvent, -} from "@opencode-ai/llm" -import { Cause, DateTime, Effect, FiberSet, Layer, Option, Schema, Semaphore, Stream } from "effect" -import { AgentV2 } from "../../agent" -import { Config } from "../../config" +import { DateTime, Effect, Layer } from "effect" import { Database } from "../../database/database" import { EventV2 } from "../../event" -import { Location } from "../../location" -import { ModelV2 } from "../../model" -import { ProviderV2 } from "../../provider" -import { QuestionV2 } from "../../question" -import { SystemContext } from "../../system-context/index" -import { SystemContextRegistry } from "../../system-context/registry" -import { SkillGuidance } from "../../skill/guidance" -import { ToolRegistry } from "../../tool/registry" -import { ToolOutputStore } from "../../tool-output-store" -import { SessionContextEpoch } from "../context-epoch" -import { SessionCompaction } from "../compaction" import { SessionEvent } from "../event" -import { SessionHistory } from "../history" import { SessionInput } from "../input" import { SessionSchema } from "../schema" import { SessionStore } from "../store" -import { type RunError, Service, StepLimitExceededError } from "./index" -import { SessionRunnerModel } from "./model" -import { createLLMEventPublisher } from "./publish-llm-event" -import { toLLMMessages } from "./to-llm-message" +import { Service, StepLimitExceededError } from "./index" +import { RunTurn } from "./run-turn" /** * Runs one durable coding-agent Session until it settles. @@ -75,8 +50,9 @@ import { toLLMMessages } from "./to-llm-message" * - [ ] Coalesce streamed deltas and add covering projected-history indexes. * - [ ] 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. + * `RunTurn` owns provider-turn preparation, streaming, tool settlement, and continuation signals. + * This module owns durable activity scheduling and bounded continuation. Durable activity 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 a @@ -90,22 +66,9 @@ export const layer = Layer.effect( Service, Effect.gen(function* () { const events = yield* EventV2.Service - const llm = yield* LLMClient.Service - const agents = yield* AgentV2.Service - const tools = yield* ToolRegistry.Service - const models = yield* SessionRunnerModel.Service const store = yield* SessionStore.Service - const location = yield* Location.Service - const systemContext = yield* SystemContextRegistry.Service - const skillGuidance = yield* SkillGuidance.Service - const config = yield* Config.Service const db = (yield* Database.Service).db - const compaction = SessionCompaction.make({ events, llm, config: yield* config.entries() }) - const getSession = Effect.fn("SessionRunner.getSession")(function* (sessionID: SessionSchema.ID) { - const session = yield* store.get(sessionID) - if (!session) return yield* Effect.die(`Session not found: ${sessionID}`) - return session - }) + const runTurn = yield* RunTurn.make const getContext = Effect.fn("SessionRunner.getContext")(function* (sessionID: SessionSchema.ID) { return yield* store.context(sessionID) @@ -132,242 +95,6 @@ export const layer = Layer.effect( } }) - const awaitToolFibers = (fibers: FiberSet.FiberSet) => - Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers)) - - // Match V1: dismissing a question halts the loop instead of becoming model-facing tool output. - const isQuestionRejected = (cause: Cause.Cause) => - cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) - - type TurnTransition = - // Request preparation observed a concurrent Session change and must restart from durable state. - | { readonly _tag: "RebuildPreparedTurn"; readonly promotion?: SessionInput.Delivery } - // Overflow compaction completed; rebuild once through the path without overflow recovery. - | { readonly _tag: "ContinueAfterOverflowCompaction" } - - 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 retryAgentMismatch = (promotion: SessionInput.Delivery | undefined) => - Effect.catchDefect((defect) => - defect instanceof SessionContextEpoch.AgentMismatch - ? Effect.die(rebuildPreparedTurn(promotion)) - : Effect.die(defect), - ) - - const sameModel = Schema.toEquivalence(Schema.UndefinedOr(ModelV2.Ref)) - const loadSystemContext = (agent: AgentV2.Selection) => - Effect.all([systemContext.load(), skillGuidance.load(agent)], { concurrency: "unbounded" }).pipe( - Effect.map(SystemContext.combine), - ) - - const runTurnAttempt = Effect.fn("SessionRunner.runTurn")(function* ( - sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, - recoverOverflow?: typeof compaction.compactAfterOverflow, - ) { - const session = yield* getSession(sessionID) - 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 toolFibers = yield* FiberSet.make() - let needsContinuation = false - if (promotion) { - const cutoff = yield* SessionInput.latestSeq(db, session.id) - if (promotion === "steer") 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) - } - } - const system = - initialized ?? - (yield* SessionContextEpoch.prepare( - db, - events, - loadSystemContext(agent), - session.id, - session.location, - agent.id, - ).pipe(retryAgentMismatch(undefined))) - const current = yield* getSession(sessionID) - if ((yield* agents.select(current.agent)).id !== agent.id || !sameModel(current.model, session.model)) - return yield* Effect.die(rebuildPreparedTurn()) - const model = yield* models.resolve(session) - const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) - const context = entries.map((entry) => entry.message) - const toolMaterialization = 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({ - model, - providerOptions: { openai: { promptCacheKey } }, - system: [agent.info?.system, system.baseline] - .filter((part): part is string => part !== undefined && part.length > 0) - .map(SystemPart.make), - messages: toLLMMessages(context, model), - tools: toolMaterialization.definitions, - }) - if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) - return yield* Effect.die(rebuildPreparedTurn()) - const publisher = createLLMEventPublisher(events, { - sessionID: session.id, - agent: agent.id, - model: { - id: ModelV2.ID.make(model.id), - providerID: ProviderV2.ID.make(model.provider), - ...(session.model?.variant === undefined ? {} : { variant: session.model.variant }), - }, - }) - const withPublication = Semaphore.makeUnsafe(1).withPermit - 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()) - const providerStream = llm.stream(request).pipe( - Stream.runForEach((event) => - Effect.gen(function* () { - if (overflowFailure || publisher.hasProviderError()) return - if (LLMEvent.is.providerError(event)) { - if (isContextOverflowFailure(event) && !publisher.hasAssistantStarted()) { - overflowFailure = event - return - } - } - yield* publish(event) - if (event.type !== "tool-call" || event.providerExecuted) return - needsContinuation = true - const assistantMessageID = yield* publisher.assistantMessageID(event.id) - yield* Effect.uninterruptibleMask((restore) => - restore( - toolMaterialization.settle({ - sessionID: session.id, - agent: agent.id, - assistantMessageID, - call: event, - }), - ).pipe( - Effect.flatMap((settlement) => - publish( - LLMEvent.toolResult({ - id: event.id, - name: event.name, - result: settlement.result, - output: settlement.output, - }), - settlement.outputPaths ?? [], - ), - ), - ), - ).pipe(FiberSet.run(toolFibers)) - }), - ), - Effect.ensuring(withPublication(publisher.flush())), - ) - - return yield* Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - const stream = yield* restore(providerStream).pipe(Effect.exit) - const failure = - stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined - if ( - recoverOverflow && - !publisher.hasAssistantStarted() && - isContextOverflowFailure(overflowFailure ?? failure) && - (yield* restore(recoverOverflow({ sessionID: session.id, entries, model, request }))) - ) - return yield* Effect.die(continueAfterOverflowCompaction) - if (overflowFailure) yield* publish(overflowFailure) - const llmFailure = failure instanceof LLMError ? failure : undefined - if (llmFailure && !publisher.hasProviderError()) { - yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) - yield* withPublication( - events.publish(SessionEvent.Step.Failed, { - sessionID: session.id, - timestamp: yield* DateTime.now, - assistantMessageID: yield* publisher.startAssistant(), - error: { type: "unknown", message: llmFailure.reason.message }, - }), - ) - } - if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers) - const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) - if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { - yield* FiberSet.clear(toolFibers) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) - return yield* Effect.interrupt - } - if ( - (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) || - (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)) - ) { - yield* FiberSet.clear(toolFibers) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) - } - if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) { - const failure = Cause.squash(settled.cause) - const message = failure instanceof Error ? failure.message : String(failure) - yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) - } - if (publisher.hasProviderError()) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) - if (stream._tag === "Success" && !publisher.hasProviderError()) - 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 - }), - ) - }, Effect.scoped) - type RunTurn = ( - sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, - ) => Effect.Effect - - const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion) { - return yield* runTurnAttempt(sessionID, promotion).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) - }), - ), - ) - }) - - const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion) { - return yield* runTurnAttempt(sessionID, promotion, 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) - return yield* runTurn(sessionID, defect.transition.promotion) - }), - ), - ) - }) - const run = Effect.fn("SessionRunner.run")(function* (input: { readonly sessionID: SessionSchema.ID readonly force?: boolean diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts new file mode 100644 index 0000000000..29b4956b93 --- /dev/null +++ b/packages/core/src/session/runner/run-turn.ts @@ -0,0 +1,288 @@ +export * as RunTurn from "./run-turn" + +import { + LLM, + LLMClient, + LLMError, + LLMEvent, + SystemPart, + isContextOverflowFailure, + type ProviderErrorEvent, +} from "@opencode-ai/llm" +import { Cause, DateTime, Effect, FiberSet, Option, Schema, Semaphore, Stream } from "effect" +import { AgentV2 } from "../../agent" +import { Config } from "../../config" +import { Database } from "../../database/database" +import { EventV2 } from "../../event" +import { Location } from "../../location" +import { ModelV2 } from "../../model" +import { ProviderV2 } from "../../provider" +import { QuestionV2 } from "../../question" +import { SkillGuidance } from "../../skill/guidance" +import { SystemContext } from "../../system-context/index" +import { SystemContextRegistry } from "../../system-context/registry" +import { ToolOutputStore } from "../../tool-output-store" +import { ToolRegistry } from "../../tool/registry" +import { SessionCompaction } from "../compaction" +import { SessionContextEpoch } from "../context-epoch" +import { SessionEvent } from "../event" +import { SessionHistory } from "../history" +import { SessionInput } from "../input" +import { SessionSchema } from "../schema" +import { SessionStore } from "../store" +import type { RunError } from "./index" +import { SessionRunnerModel } from "./model" +import { createLLMEventPublisher } from "./publish-llm-event" +import { toLLMMessages } from "./to-llm-message" + +export type Run = ( + sessionID: SessionSchema.ID, + promotion: SessionInput.Delivery | undefined, +) => Effect.Effect + +type TurnTransition = + | { readonly _tag: "RebuildPreparedTurn"; readonly promotion?: SessionInput.Delivery } + | { readonly _tag: "ContinueAfterOverflowCompaction" } + +class TurnTransitionError extends Error { + constructor(readonly transition: TurnTransition) { + super() + } +} + +export const make = Effect.gen(function* () { + const events = yield* EventV2.Service + const llm = yield* LLMClient.Service + const agents = yield* AgentV2.Service + const tools = yield* ToolRegistry.Service + const models = yield* SessionRunnerModel.Service + const store = yield* SessionStore.Service + const location = yield* Location.Service + const systemContext = yield* SystemContextRegistry.Service + const skillGuidance = yield* SkillGuidance.Service + const config = yield* Config.Service + const db = (yield* Database.Service).db + const compaction = SessionCompaction.make({ events, llm, config: yield* config.entries() }) + + const getSession = Effect.fn("SessionRunner.getSession")(function* (sessionID: SessionSchema.ID) { + const session = yield* store.get(sessionID) + if (!session) return yield* Effect.die(`Session not found: ${sessionID}`) + return session + }) + const awaitToolFibers = (fibers: FiberSet.FiberSet) => + Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers)) + const isQuestionRejected = (cause: Cause.Cause) => + cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) + const rebuildPreparedTurn = (promotion?: SessionInput.Delivery) => + new TurnTransitionError({ _tag: "RebuildPreparedTurn", promotion }) + const continueAfterOverflowCompaction = new TurnTransitionError({ + _tag: "ContinueAfterOverflowCompaction", + }) + const retryAgentMismatch = (promotion: SessionInput.Delivery | undefined) => + Effect.catchDefect((defect) => + defect instanceof SessionContextEpoch.AgentMismatch + ? Effect.die(rebuildPreparedTurn(promotion)) + : Effect.die(defect), + ) + const sameModel = Schema.toEquivalence(Schema.UndefinedOr(ModelV2.Ref)) + const loadSystemContext = (agent: AgentV2.Selection) => + Effect.all([systemContext.load(), skillGuidance.load(agent)], { concurrency: "unbounded" }).pipe( + Effect.map(SystemContext.combine), + ) + + const runAttempt = Effect.fn("SessionRunner.runTurn")(function* ( + sessionID: SessionSchema.ID, + promotion: SessionInput.Delivery | undefined, + recoverOverflow?: typeof compaction.compactAfterOverflow, + ) { + const session = yield* getSession(sessionID) + 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 toolFibers = yield* FiberSet.make() + let needsContinuation = false + if (promotion) { + const cutoff = yield* SessionInput.latestSeq(db, session.id) + if (promotion === "steer") 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) + } + } + const system = + initialized ?? + (yield* SessionContextEpoch.prepare( + db, + events, + loadSystemContext(agent), + session.id, + session.location, + agent.id, + ).pipe(retryAgentMismatch(undefined))) + const current = yield* getSession(sessionID) + if ((yield* agents.select(current.agent)).id !== agent.id || !sameModel(current.model, session.model)) + return yield* Effect.die(rebuildPreparedTurn()) + const model = yield* models.resolve(session) + const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) + const context = entries.map((entry) => entry.message) + const toolMaterialization = 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({ + model, + providerOptions: { openai: { promptCacheKey } }, + system: [agent.info?.system, system.baseline] + .filter((part): part is string => part !== undefined && part.length > 0) + .map(SystemPart.make), + messages: toLLMMessages(context, model), + tools: toolMaterialization.definitions, + }) + if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) + return yield* Effect.die(rebuildPreparedTurn()) + const publisher = createLLMEventPublisher(events, { + sessionID: session.id, + agent: agent.id, + model: { + id: ModelV2.ID.make(model.id), + providerID: ProviderV2.ID.make(model.provider), + ...(session.model?.variant === undefined ? {} : { variant: session.model.variant }), + }, + }) + const withPublication = Semaphore.makeUnsafe(1).withPermit + 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()) + const providerStream = llm.stream(request).pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (overflowFailure || publisher.hasProviderError()) return + if (LLMEvent.is.providerError(event)) { + if (isContextOverflowFailure(event) && !publisher.hasAssistantStarted()) { + overflowFailure = event + return + } + } + yield* publish(event) + if (event.type !== "tool-call" || event.providerExecuted) return + needsContinuation = true + const assistantMessageID = yield* publisher.assistantMessageID(event.id) + yield* Effect.uninterruptibleMask((restore) => + restore( + toolMaterialization.settle({ + sessionID: session.id, + agent: agent.id, + assistantMessageID, + call: event, + }), + ).pipe( + Effect.flatMap((settlement) => + publish( + LLMEvent.toolResult({ + id: event.id, + name: event.name, + result: settlement.result, + output: settlement.output, + }), + settlement.outputPaths ?? [], + ), + ), + ), + ).pipe(FiberSet.run(toolFibers)) + }), + ), + Effect.ensuring(withPublication(publisher.flush())), + ) + + return yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const stream = yield* restore(providerStream).pipe(Effect.exit) + const failure = + stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined + if ( + recoverOverflow && + !publisher.hasAssistantStarted() && + isContextOverflowFailure(overflowFailure ?? failure) && + (yield* restore(recoverOverflow({ sessionID: session.id, entries, model, request }))) + ) + return yield* Effect.die(continueAfterOverflowCompaction) + if (overflowFailure) yield* publish(overflowFailure) + const llmFailure = failure instanceof LLMError ? failure : undefined + if (llmFailure && !publisher.hasProviderError()) { + yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) + yield* withPublication( + events.publish(SessionEvent.Step.Failed, { + sessionID: session.id, + timestamp: yield* DateTime.now, + assistantMessageID: yield* publisher.startAssistant(), + error: { type: "unknown", message: llmFailure.reason.message }, + }), + ) + } + if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers) + const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) + if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { + yield* FiberSet.clear(toolFibers) + yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + return yield* Effect.interrupt + } + if ( + (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) || + (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)) + ) { + yield* FiberSet.clear(toolFibers) + yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + } + if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) { + const failure = Cause.squash(settled.cause) + const message = failure instanceof Error ? failure.message : String(failure) + yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) + } + if (publisher.hasProviderError()) + yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + if (stream._tag === "Success" && !publisher.hasProviderError()) + 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 + }), + ) + }, Effect.scoped) + + const runAfterOverflowCompaction: Run = Effect.fnUntraced(function* (sessionID, promotion) { + return yield* runAttempt(sessionID, promotion).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) + }), + ), + ) + }) + + const run: Run = Effect.fnUntraced(function* (sessionID, promotion) { + return yield* runAttempt(sessionID, promotion, 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) + return yield* run(sessionID, defect.transition.promotion) + }), + ), + ) + }) + + return run +}) From ad048465937530344058b4759c560fdb2fa782ec Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:06:12 -0400 Subject: [PATCH 02/10] refactor(core): clarify provider turn phases --- packages/core/src/session/runner/run-turn.ts | 242 ++++++++++++------- 1 file changed, 151 insertions(+), 91 deletions(-) diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index 29b4956b93..6a0c6c35b8 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -1,5 +1,15 @@ export * as RunTurn from "./run-turn" +/** + * Drives one logical provider turn to settlement. + * + * A logical turn may rebuild its immutable preparation when concurrent Session, + * agent, model, or Context Epoch changes make a prepared request stale. Each + * prepared attempt invokes `llm.stream` at most once. A pre-output context + * overflow may compact and rebuild once; later rebuilds do not restore that + * recovery budget. + */ + import { LLM, LLMClient, @@ -40,15 +50,10 @@ export type Run = ( promotion: SessionInput.Delivery | undefined, ) => Effect.Effect -type TurnTransition = - | { readonly _tag: "RebuildPreparedTurn"; readonly promotion?: SessionInput.Delivery } - | { readonly _tag: "ContinueAfterOverflowCompaction" } - -class TurnTransitionError extends Error { - constructor(readonly transition: TurnTransition) { - super() - } -} +const TurnTransition = Schema.TaggedUnion({ + RebuildPreparedTurn: { promotion: SessionInput.Delivery.pipe(Schema.optional) }, + ContinueAfterOverflowCompaction: {}, +}) export const make = Effect.gen(function* () { const events = yield* EventV2.Service @@ -74,14 +79,12 @@ export const make = Effect.gen(function* () { const isQuestionRejected = (cause: Cause.Cause) => cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) const rebuildPreparedTurn = (promotion?: SessionInput.Delivery) => - new TurnTransitionError({ _tag: "RebuildPreparedTurn", promotion }) - const continueAfterOverflowCompaction = new TurnTransitionError({ - _tag: "ContinueAfterOverflowCompaction", - }) + TurnTransition.cases.RebuildPreparedTurn.make({ promotion }) + const continueAfterOverflowCompaction = TurnTransition.cases.ContinueAfterOverflowCompaction.make({}) const retryAgentMismatch = (promotion: SessionInput.Delivery | undefined) => Effect.catchDefect((defect) => defect instanceof SessionContextEpoch.AgentMismatch - ? Effect.die(rebuildPreparedTurn(promotion)) + ? Effect.fail(rebuildPreparedTurn(promotion)) : Effect.die(defect), ) const sameModel = Schema.toEquivalence(Schema.UndefinedOr(ModelV2.Ref)) @@ -90,10 +93,15 @@ export const make = Effect.gen(function* () { Effect.map(SystemContext.combine), ) - const runAttempt = Effect.fn("SessionRunner.runTurn")(function* ( + /** + * Promotes admitted input and builds one coherent immutable request snapshot. + * + * Rebuild transitions before promotion preserve the requested delivery; + * transitions after promotion clear it so queued input cannot be promoted twice. + */ + const prepareTurn = Effect.fn("SessionRunner.prepareTurn")(function* ( sessionID: SessionSchema.ID, promotion: SessionInput.Delivery | undefined, - recoverOverflow?: typeof compaction.compactAfterOverflow, ) { const session = yield* getSession(sessionID) if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) @@ -106,8 +114,6 @@ export const make = Effect.gen(function* () { session.location, agent.id, ).pipe(retryAgentMismatch(promotion)) - const toolFibers = yield* FiberSet.make() - let needsContinuation = false if (promotion) { const cutoff = yield* SessionInput.latestSeq(db, session.id) if (promotion === "steer") yield* SessionInput.promoteSteers(db, events, session.id, cutoff) @@ -128,7 +134,7 @@ export const make = Effect.gen(function* () { ).pipe(retryAgentMismatch(undefined))) 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 yield* Effect.fail(rebuildPreparedTurn()) const model = yield* models.resolve(session) const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) const context = entries.map((entry) => entry.message) @@ -144,47 +150,76 @@ export const make = Effect.gen(function* () { tools: toolMaterialization.definitions, }) if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) - return yield* Effect.die(rebuildPreparedTurn()) + return yield* Effect.fail(rebuildPreparedTurn()) + return { session, agent, model, entries, request, system, toolMaterialization } + }) + + type PreparedTurn = Effect.Success> + + /** + * Allocates the mutable state shared by provider consumption and settlement. + * Publication is serialized because provider events and local tool results may + * arrive concurrently but mutate one durable publisher state machine. + */ + const makeRuntime = Effect.fnUntraced(function* (prepared: PreparedTurn) { const publisher = createLLMEventPublisher(events, { - sessionID: session.id, - agent: agent.id, + sessionID: prepared.session.id, + agent: prepared.agent.id, model: { - id: ModelV2.ID.make(model.id), - providerID: ProviderV2.ID.make(model.provider), - ...(session.model?.variant === undefined ? {} : { variant: session.model.variant }), + id: ModelV2.ID.make(prepared.model.id), + providerID: ProviderV2.ID.make(prepared.model.provider), + ...(prepared.session.model?.variant === undefined ? {} : { variant: prepared.session.model.variant }), }, }) const withPublication = Semaphore.makeUnsafe(1).withPermit - 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()) - const providerStream = llm.stream(request).pipe( + return { + publisher, + withPublication, + publish: (event: LLMEvent, outputPaths: ReadonlyArray = []) => + withPublication(publisher.publish(event, outputPaths)), + toolFibers: yield* FiberSet.make(), + needsContinuation: false, + overflowFailure: undefined as ProviderErrorEvent | undefined, + } + }) + + type TurnRuntime = Effect.Success> + + /** + * Consumes exactly one provider stream. + * + * Every event is durably published before a local tool starts. Tool settlement + * is registered with the turn FiberSet before interruption can resume. A + * recoverable pre-output overflow is withheld until the settlement phase. + */ + const consumeProvider = (prepared: PreparedTurn, runtime: TurnRuntime) => + llm.stream(prepared.request).pipe( Stream.runForEach((event) => Effect.gen(function* () { - if (overflowFailure || publisher.hasProviderError()) return - if (LLMEvent.is.providerError(event)) { - if (isContextOverflowFailure(event) && !publisher.hasAssistantStarted()) { - overflowFailure = event - return - } + if (runtime.overflowFailure || runtime.publisher.hasProviderError()) return + if ( + LLMEvent.is.providerError(event) && + isContextOverflowFailure(event) && + !runtime.publisher.hasAssistantStarted() + ) { + runtime.overflowFailure = event + return } - yield* publish(event) + yield* runtime.publish(event) if (event.type !== "tool-call" || event.providerExecuted) return - needsContinuation = true - const assistantMessageID = yield* publisher.assistantMessageID(event.id) + runtime.needsContinuation = true + const assistantMessageID = yield* runtime.publisher.assistantMessageID(event.id) yield* Effect.uninterruptibleMask((restore) => restore( - toolMaterialization.settle({ - sessionID: session.id, - agent: agent.id, + prepared.toolMaterialization.settle({ + sessionID: prepared.session.id, + agent: prepared.agent.id, assistantMessageID, call: event, }), ).pipe( Effect.flatMap((settlement) => - publish( + runtime.publish( LLMEvent.toolResult({ id: event.id, name: event.name, @@ -195,94 +230,119 @@ export const make = Effect.gen(function* () { ), ), ), - ).pipe(FiberSet.run(toolFibers)) + ).pipe(FiberSet.run(runtime.toolFibers)) }), ), - Effect.ensuring(withPublication(publisher.flush())), + Effect.ensuring(runtime.withPublication(runtime.publisher.flush())), ) + /** + * Runs one prepared provider attempt and settles every local tool it starts. + * + * The interruption mask keeps the handoff from stream completion to tool + * settlement atomic, while provider consumption and tool work remain interruptible. + */ + const runAttempt = Effect.fn("SessionRunner.runTurn")(function* ( + sessionID: SessionSchema.ID, + promotion: SessionInput.Delivery | undefined, + recoverOverflow?: typeof compaction.compactAfterOverflow, + ) { + const prepared = yield* prepareTurn(sessionID, promotion) + const runtime = yield* makeRuntime(prepared) + if (!(yield* SessionContextEpoch.current(db, prepared.session.id, prepared.agent.id, prepared.system.revision))) + return yield* Effect.fail(rebuildPreparedTurn()) + return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { - const stream = yield* restore(providerStream).pipe(Effect.exit) + const stream = yield* restore(consumeProvider(prepared, runtime)).pipe(Effect.exit) const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined if ( recoverOverflow && - !publisher.hasAssistantStarted() && - isContextOverflowFailure(overflowFailure ?? failure) && - (yield* restore(recoverOverflow({ sessionID: session.id, entries, model, request }))) + !runtime.publisher.hasAssistantStarted() && + isContextOverflowFailure(runtime.overflowFailure ?? failure) && + (yield* restore( + recoverOverflow({ + sessionID: prepared.session.id, + entries: prepared.entries, + model: prepared.model, + request: prepared.request, + }), + )) ) - return yield* Effect.die(continueAfterOverflowCompaction) - if (overflowFailure) yield* publish(overflowFailure) + return yield* Effect.fail(continueAfterOverflowCompaction) + if (runtime.overflowFailure) yield* runtime.publish(runtime.overflowFailure) const llmFailure = failure instanceof LLMError ? failure : undefined - if (llmFailure && !publisher.hasProviderError()) { - yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) - yield* withPublication( + if (llmFailure && !runtime.publisher.hasProviderError()) { + yield* runtime.withPublication( + runtime.publisher.failUnsettledTools("Provider did not return a tool result", true), + ) + yield* runtime.withPublication( events.publish(SessionEvent.Step.Failed, { - sessionID: session.id, + sessionID: prepared.session.id, timestamp: yield* DateTime.now, - assistantMessageID: yield* publisher.startAssistant(), + assistantMessageID: yield* runtime.publisher.startAssistant(), error: { type: "unknown", message: llmFailure.reason.message }, }), ) } - if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers) - const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) + if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(runtime.toolFibers) + const settled = yield* restore(awaitToolFibers(runtime.toolFibers)).pipe(Effect.exit) if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { - yield* FiberSet.clear(toolFibers) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + yield* FiberSet.clear(runtime.toolFibers) + yield* runtime.withPublication(runtime.publisher.failUnsettledTools("Tool execution interrupted")) return yield* Effect.interrupt } if ( (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) || (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)) ) { - yield* FiberSet.clear(toolFibers) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + yield* FiberSet.clear(runtime.toolFibers) + yield* runtime.withPublication(runtime.publisher.failUnsettledTools("Tool execution interrupted")) } if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) { const failure = Cause.squash(settled.cause) const message = failure instanceof Error ? failure.message : String(failure) - yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) + yield* runtime.withPublication(runtime.publisher.failUnsettledTools(`Tool execution failed: ${message}`)) } - if (publisher.hasProviderError()) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) - if (stream._tag === "Success" && !publisher.hasProviderError()) - yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) + if (runtime.publisher.hasProviderError()) + yield* runtime.withPublication(runtime.publisher.failUnsettledTools("Tool execution interrupted")) + if (stream._tag === "Success" && !runtime.publisher.hasProviderError()) + yield* runtime.withPublication( + runtime.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 !runtime.publisher.hasProviderError() && runtime.needsContinuation }), ) }, Effect.scoped) - const runAfterOverflowCompaction: Run = Effect.fnUntraced(function* (sessionID, promotion) { - return yield* runAttempt(sessionID, promotion).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") + /** Rebuilds stale attempts while preserving the single overflow-recovery budget. */ + const runState = Effect.fnUntraced(function* ( + sessionID: SessionSchema.ID, + promotion: SessionInput.Delivery | undefined, + canRecoverOverflow: boolean, + ): Effect.fn.Return { + return yield* runAttempt( + sessionID, + promotion, + canRecoverOverflow ? compaction.compactAfterOverflow : undefined, + ).pipe( + Effect.catchTags({ + ContinueAfterOverflowCompaction: Effect.fnUntraced(function* () { yield* Effect.yieldNow - return yield* runAfterOverflowCompaction(sessionID, defect.transition.promotion) + return yield* runState(sessionID, undefined, false) }), - ), + RebuildPreparedTurn: Effect.fnUntraced(function* (transition) { + yield* Effect.yieldNow + return yield* runState(sessionID, transition.promotion, canRecoverOverflow) + }), + }), ) }) - const run: Run = Effect.fnUntraced(function* (sessionID, promotion) { - return yield* runAttempt(sessionID, promotion, 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) - return yield* run(sessionID, defect.transition.promotion) - }), - ), - ) - }) + const run: Run = (sessionID, promotion) => runState(sessionID, promotion, true) return run }) From 5462e08ac1256c30a2d1b0f07b4ca1a8b9995f2e Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:13:24 -0400 Subject: [PATCH 03/10] docs(core): clarify turn rebuild promotion --- packages/core/src/session/runner/run-turn.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index 6a0c6c35b8..ea35b2efb8 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -96,8 +96,10 @@ export const make = Effect.gen(function* () { /** * Promotes admitted input and builds one coherent immutable request snapshot. * - * Rebuild transitions before promotion preserve the requested delivery; - * transitions after promotion clear it so queued input cannot be promoted twice. + * If initialization becomes stale before input is promoted, the rebuilt + * attempt must still perform that promotion. Once promotion has completed, + * later rebuilds read it from durable history instead of promoting again; + * repeating a queue promotion could open the next queued prompt too early. */ const prepareTurn = Effect.fn("SessionRunner.prepareTurn")(function* ( sessionID: SessionSchema.ID, From 1cb6f7b091bc6b8afd5a1c4a6822c964a1931a56 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:17:03 -0400 Subject: [PATCH 04/10] refactor(core): simplify provider turn retries --- packages/core/src/session/runner/run-turn.ts | 207 ++++++++----------- 1 file changed, 91 insertions(+), 116 deletions(-) diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index ea35b2efb8..cf9f5b8b05 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -1,13 +1,12 @@ export * as RunTurn from "./run-turn" /** - * Drives one logical provider turn to settlement. + * Sends the next request to the model and finishes every tool call it starts. * - * A logical turn may rebuild its immutable preparation when concurrent Session, - * agent, model, or Context Epoch changes make a prepared request stale. Each - * prepared attempt invokes `llm.stream` at most once. A pre-output context - * overflow may compact and rebuild once; later rebuilds do not restore that - * recovery budget. + * Before sending, it makes admitted input visible, loads the latest Session + * history and instructions, and compacts oversized history. If the model rejects + * the request for being too large before producing output, it may compact and + * try once more. Returns `true` when tool results require another model request. */ import { @@ -50,9 +49,9 @@ export type Run = ( promotion: SessionInput.Delivery | undefined, ) => Effect.Effect -const TurnTransition = Schema.TaggedUnion({ - RebuildPreparedTurn: { promotion: SessionInput.Delivery.pipe(Schema.optional) }, - ContinueAfterOverflowCompaction: {}, +const AttemptResult = Schema.TaggedUnion({ + Complete: { needsContinuation: Schema.Boolean }, + CompactedOverflow: {}, }) export const make = Effect.gen(function* () { @@ -78,92 +77,89 @@ export const make = Effect.gen(function* () { Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers)) const isQuestionRejected = (cause: Cause.Cause) => cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) - const rebuildPreparedTurn = (promotion?: SessionInput.Delivery) => - TurnTransition.cases.RebuildPreparedTurn.make({ promotion }) - const continueAfterOverflowCompaction = TurnTransition.cases.ContinueAfterOverflowCompaction.make({}) - const retryAgentMismatch = (promotion: SessionInput.Delivery | undefined) => - Effect.catchDefect((defect) => - defect instanceof SessionContextEpoch.AgentMismatch - ? Effect.fail(rebuildPreparedTurn(promotion)) - : Effect.die(defect), + const stale = Symbol("stale turn preparation") + const retryAgentMismatch = (effect: Effect.Effect) => + effect.pipe( + Effect.catchDefect((defect) => + defect instanceof SessionContextEpoch.AgentMismatch ? Effect.succeed(stale) : Effect.die(defect), + ), ) - const sameModel = Schema.toEquivalence(Schema.UndefinedOr(ModelV2.Ref)) const loadSystemContext = (agent: AgentV2.Selection) => Effect.all([systemContext.load(), skillGuidance.load(agent)], { concurrency: "unbounded" }).pipe( Effect.map(SystemContext.combine), ) /** - * Promotes admitted input and builds one coherent immutable request snapshot. + * Builds the next model request from durable Session state. * - * If initialization becomes stale before input is promoted, the rebuilt - * attempt must still perform that promotion. Once promotion has completed, - * later rebuilds read it from durable history instead of promoting again; - * repeating a queue promotion could open the next queued prompt too early. + * Initial instructions must be available before admitted input becomes visible. + * Once input is promoted, retries load it from history instead of promoting + * again. This matters for queued input because promotion opens the next item. */ const prepareTurn = Effect.fn("SessionRunner.prepareTurn")(function* ( sessionID: SessionSchema.ID, promotion: SessionInput.Delivery | undefined, ) { - const session = yield* getSession(sessionID) - 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)) - if (promotion) { - const cutoff = yield* SessionInput.latestSeq(db, session.id) - if (promotion === "steer") 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) + let pendingPromotion = promotion + while (true) { + const session = yield* getSession(sessionID) + 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* retryAgentMismatch( + SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id, session.location, agent.id), + ) + if (initialized === stale) continue + if (pendingPromotion) { + const cutoff = yield* SessionInput.latestSeq(db, session.id) + if (pendingPromotion === "steer") yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + if (pendingPromotion === "queue") { + yield* SessionInput.promoteNextQueued(db, events, session.id) + yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + } + pendingPromotion = undefined } + const prepared = + initialized ?? + (yield* retryAgentMismatch( + SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id, session.location, agent.id), + )) + if (prepared === stale) continue + const system = prepared + const model = yield* models.resolve(session) + const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) + const toolMaterialization = yield* tools.materialize(agent.info?.permissions) + const request = LLM.request({ + model, + providerOptions: { + openai: { promptCacheKey: /^ses_[0-9a-f]{64}$/.test(session.id) ? session.id.slice(4) : session.id }, + }, + system: [agent.info?.system, system.baseline] + .filter((part): part is string => part !== undefined && part.length > 0) + .map(SystemPart.make), + messages: toLLMMessages( + entries.map((entry) => entry.message), + model, + ), + tools: toolMaterialization.definitions, + }) + if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) { + continue + } + if (!(yield* SessionContextEpoch.current(db, session.id, agent.id, system.revision))) { + continue + } + return { session, agent, model, entries, request, toolMaterialization } } - const system = - initialized ?? - (yield* SessionContextEpoch.prepare( - db, - events, - loadSystemContext(agent), - session.id, - session.location, - agent.id, - ).pipe(retryAgentMismatch(undefined))) - const current = yield* getSession(sessionID) - if ((yield* agents.select(current.agent)).id !== agent.id || !sameModel(current.model, session.model)) - return yield* Effect.fail(rebuildPreparedTurn()) - const model = yield* models.resolve(session) - const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) - const context = entries.map((entry) => entry.message) - const toolMaterialization = 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({ - model, - providerOptions: { openai: { promptCacheKey } }, - system: [agent.info?.system, system.baseline] - .filter((part): part is string => part !== undefined && part.length > 0) - .map(SystemPart.make), - messages: toLLMMessages(context, model), - tools: toolMaterialization.definitions, - }) - if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) - return yield* Effect.fail(rebuildPreparedTurn()) - return { session, agent, model, entries, request, system, toolMaterialization } }) - type PreparedTurn = Effect.Success> + type RequestSnapshot = Effect.Success> /** - * Allocates the mutable state shared by provider consumption and settlement. - * Publication is serialized because provider events and local tool results may - * arrive concurrently but mutate one durable publisher state machine. + * Provider events and tool results can arrive concurrently. They share one + * permit so their durable Session events are written in order. */ - const makeRuntime = Effect.fnUntraced(function* (prepared: PreparedTurn) { + const startTurn = Effect.fnUntraced(function* (prepared: RequestSnapshot) { const publisher = createLLMEventPublisher(events, { sessionID: prepared.session.id, agent: prepared.agent.id, @@ -185,16 +181,14 @@ export const make = Effect.gen(function* () { } }) - type TurnRuntime = Effect.Success> + type ActiveTurn = Effect.Success> /** - * Consumes exactly one provider stream. - * - * Every event is durably published before a local tool starts. Tool settlement - * is registered with the turn FiberSet before interruption can resume. A - * recoverable pre-output overflow is withheld until the settlement phase. + * Reads one model response. A tool call is recorded before its side effect + * starts. An overflow error is held back briefly so successful compaction does + * not leave a terminal error in Session history. */ - const consumeProvider = (prepared: PreparedTurn, runtime: TurnRuntime) => + const consumeProvider = (prepared: RequestSnapshot, runtime: ActiveTurn) => llm.stream(prepared.request).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -239,10 +233,8 @@ export const make = Effect.gen(function* () { ) /** - * Runs one prepared provider attempt and settles every local tool it starts. - * - * The interruption mask keeps the handoff from stream completion to tool - * settlement atomic, while provider consumption and tool work remain interruptible. + * The model response and tools remain interruptible. The short handoff after + * the response ends is protected so no started tool is forgotten before cleanup. */ const runAttempt = Effect.fn("SessionRunner.runTurn")(function* ( sessionID: SessionSchema.ID, @@ -250,10 +242,7 @@ export const make = Effect.gen(function* () { recoverOverflow?: typeof compaction.compactAfterOverflow, ) { const prepared = yield* prepareTurn(sessionID, promotion) - const runtime = yield* makeRuntime(prepared) - if (!(yield* SessionContextEpoch.current(db, prepared.session.id, prepared.agent.id, prepared.system.revision))) - return yield* Effect.fail(rebuildPreparedTurn()) - + const runtime = yield* startTurn(prepared) return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { const stream = yield* restore(consumeProvider(prepared, runtime)).pipe(Effect.exit) @@ -272,7 +261,7 @@ export const make = Effect.gen(function* () { }), )) ) - return yield* Effect.fail(continueAfterOverflowCompaction) + return AttemptResult.cases.CompactedOverflow.make({}) if (runtime.overflowFailure) yield* runtime.publish(runtime.overflowFailure) const llmFailure = failure instanceof LLMError ? failure : undefined if (llmFailure && !runtime.publisher.hasProviderError()) { @@ -315,36 +304,22 @@ export const make = Effect.gen(function* () { ) if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) if (settled._tag === "Failure") return yield* Effect.failCause(settled.cause) - return !runtime.publisher.hasProviderError() && runtime.needsContinuation + return AttemptResult.cases.Complete.make({ + needsContinuation: !runtime.publisher.hasProviderError() && runtime.needsContinuation, + }) }), ) }, Effect.scoped) - /** Rebuilds stale attempts while preserving the single overflow-recovery budget. */ - const runState = Effect.fnUntraced(function* ( - sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, - canRecoverOverflow: boolean, - ): Effect.fn.Return { - return yield* runAttempt( - sessionID, - promotion, - canRecoverOverflow ? compaction.compactAfterOverflow : undefined, - ).pipe( - Effect.catchTags({ - ContinueAfterOverflowCompaction: Effect.fnUntraced(function* () { - yield* Effect.yieldNow - return yield* runState(sessionID, undefined, false) - }), - RebuildPreparedTurn: Effect.fnUntraced(function* (transition) { - yield* Effect.yieldNow - return yield* runState(sessionID, transition.promotion, canRecoverOverflow) - }), - }), - ) + const run: Run = Effect.fnUntraced(function* (sessionID, promotion) { + const first = yield* runAttempt(sessionID, promotion, compaction.compactAfterOverflow) + if (first._tag === "Complete") return first.needsContinuation + const second = yield* runAttempt(sessionID, undefined) + return AttemptResult.match(second, { + Complete: (result) => result.needsContinuation, + CompactedOverflow: () => false, + }) }) - const run: Run = (sessionID, promotion) => runState(sessionID, promotion, true) - return run }) From 652ec048e7ef39b30851d462a2d068b98ec07f5d Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:24:53 -0400 Subject: [PATCH 05/10] refactor(core): deepen provider turn runner --- packages/core/src/session/runner/llm.ts | 2 +- packages/core/src/session/runner/run-turn.ts | 166 +++++++++---------- 2 files changed, 76 insertions(+), 92 deletions(-) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index c895558e02..704bf2a4bd 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -108,7 +108,7 @@ export const layer = Layer.effect( while (openActivity) { let needsContinuation = true for (let step = 0; step < MAX_STEPS; step++) { - needsContinuation = yield* runTurn(input.sessionID, promotion) + needsContinuation = yield* runTurn({ sessionID: input.sessionID, delivery: promotion }) promotion = "steer" if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer") if (!needsContinuation) break diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index cf9f5b8b05..9790a90764 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -44,10 +44,12 @@ import { SessionRunnerModel } from "./model" import { createLLMEventPublisher } from "./publish-llm-event" import { toLLMMessages } from "./to-llm-message" -export type Run = ( - sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, -) => Effect.Effect +export interface Input { + readonly sessionID: SessionSchema.ID + readonly delivery?: SessionInput.Delivery +} + +export type Run = (input: Input) => Effect.Effect const AttemptResult = Schema.TaggedUnion({ Complete: { needsContinuation: Schema.Boolean }, @@ -96,11 +98,11 @@ export const make = Effect.gen(function* () { * Once input is promoted, retries load it from history instead of promoting * again. This matters for queued input because promotion opens the next item. */ - const prepareTurn = Effect.fn("SessionRunner.prepareTurn")(function* ( + const buildRequest = Effect.fn("SessionRunner.buildRequest")(function* ( sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, + delivery: SessionInput.Delivery | undefined, ) { - let pendingPromotion = promotion + let pendingDelivery = delivery while (true) { const session = yield* getSession(sessionID) if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) @@ -110,14 +112,14 @@ export const make = Effect.gen(function* () { SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id, session.location, agent.id), ) if (initialized === stale) continue - if (pendingPromotion) { + if (pendingDelivery) { const cutoff = yield* SessionInput.latestSeq(db, session.id) - if (pendingPromotion === "steer") yield* SessionInput.promoteSteers(db, events, session.id, cutoff) - if (pendingPromotion === "queue") { + if (pendingDelivery === "steer") yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + if (pendingDelivery === "queue") { yield* SessionInput.promoteNextQueued(db, events, session.id) yield* SessionInput.promoteSteers(db, events, session.id, cutoff) } - pendingPromotion = undefined + pendingDelivery = undefined } const prepared = initialized ?? @@ -153,13 +155,20 @@ export const make = Effect.gen(function* () { } }) - type RequestSnapshot = Effect.Success> + type RequestSnapshot = Effect.Success> /** - * Provider events and tool results can arrive concurrently. They share one - * permit so their durable Session events are written in order. + * Reads one model response and finishes every tool it starts. + * + * Provider events and tool results share one permit so durable events stay in + * order. Tool calls are recorded before their side effects begin. A pre-output + * overflow is held back so successful compaction does not leave a terminal + * error in Session history. */ - const startTurn = Effect.fnUntraced(function* (prepared: RequestSnapshot) { + const streamAndSettle = Effect.fn("SessionRunner.streamAndSettle")(function* ( + prepared: RequestSnapshot, + recoverOverflow?: typeof compaction.compactAfterOverflow, + ) { const publisher = createLLMEventPublisher(events, { sessionID: prepared.session.id, agent: prepared.agent.id, @@ -170,41 +179,23 @@ export const make = Effect.gen(function* () { }, }) const withPublication = Semaphore.makeUnsafe(1).withPermit - return { - publisher, - withPublication, - publish: (event: LLMEvent, outputPaths: ReadonlyArray = []) => - withPublication(publisher.publish(event, outputPaths)), - toolFibers: yield* FiberSet.make(), - needsContinuation: false, - overflowFailure: undefined as ProviderErrorEvent | undefined, - } - }) - - type ActiveTurn = Effect.Success> - - /** - * Reads one model response. A tool call is recorded before its side effect - * starts. An overflow error is held back briefly so successful compaction does - * not leave a terminal error in Session history. - */ - const consumeProvider = (prepared: RequestSnapshot, runtime: ActiveTurn) => - llm.stream(prepared.request).pipe( + const publish = (event: LLMEvent, outputPaths: ReadonlyArray = []) => + withPublication(publisher.publish(event, outputPaths)) + const toolFibers = yield* FiberSet.make() + let needsContinuation = false + let overflowFailure: ProviderErrorEvent | undefined + const providerStream = llm.stream(prepared.request).pipe( Stream.runForEach((event) => Effect.gen(function* () { - if (runtime.overflowFailure || runtime.publisher.hasProviderError()) return - if ( - LLMEvent.is.providerError(event) && - isContextOverflowFailure(event) && - !runtime.publisher.hasAssistantStarted() - ) { - runtime.overflowFailure = event + if (overflowFailure || publisher.hasProviderError()) return + if (LLMEvent.is.providerError(event) && isContextOverflowFailure(event) && !publisher.hasAssistantStarted()) { + overflowFailure = event return } - yield* runtime.publish(event) + yield* publish(event) if (event.type !== "tool-call" || event.providerExecuted) return - runtime.needsContinuation = true - const assistantMessageID = yield* runtime.publisher.assistantMessageID(event.id) + needsContinuation = true + const assistantMessageID = yield* publisher.assistantMessageID(event.id) yield* Effect.uninterruptibleMask((restore) => restore( prepared.toolMaterialization.settle({ @@ -215,7 +206,7 @@ export const make = Effect.gen(function* () { }), ).pipe( Effect.flatMap((settlement) => - runtime.publish( + publish( LLMEvent.toolResult({ id: event.id, name: event.name, @@ -226,32 +217,23 @@ export const make = Effect.gen(function* () { ), ), ), - ).pipe(FiberSet.run(runtime.toolFibers)) + ).pipe(FiberSet.run(toolFibers)) }), ), - Effect.ensuring(runtime.withPublication(runtime.publisher.flush())), + Effect.ensuring(withPublication(publisher.flush())), ) - /** - * The model response and tools remain interruptible. The short handoff after - * the response ends is protected so no started tool is forgotten before cleanup. - */ - const runAttempt = Effect.fn("SessionRunner.runTurn")(function* ( - sessionID: SessionSchema.ID, - promotion: SessionInput.Delivery | undefined, - recoverOverflow?: typeof compaction.compactAfterOverflow, - ) { - const prepared = yield* prepareTurn(sessionID, promotion) - const runtime = yield* startTurn(prepared) + // Keep cleanup protected after the response ends so no started tool is + // forgotten, while the response stream and tool work remain interruptible. return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { - const stream = yield* restore(consumeProvider(prepared, runtime)).pipe(Effect.exit) + const stream = yield* restore(providerStream).pipe(Effect.exit) const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined if ( recoverOverflow && - !runtime.publisher.hasAssistantStarted() && - isContextOverflowFailure(runtime.overflowFailure ?? failure) && + !publisher.hasAssistantStarted() && + isContextOverflowFailure(overflowFailure ?? failure) && (yield* restore( recoverOverflow({ sessionID: prepared.session.id, @@ -262,63 +244,65 @@ export const make = Effect.gen(function* () { )) ) return AttemptResult.cases.CompactedOverflow.make({}) - if (runtime.overflowFailure) yield* runtime.publish(runtime.overflowFailure) + if (overflowFailure) yield* publish(overflowFailure) const llmFailure = failure instanceof LLMError ? failure : undefined - if (llmFailure && !runtime.publisher.hasProviderError()) { - yield* runtime.withPublication( - runtime.publisher.failUnsettledTools("Provider did not return a tool result", true), - ) - yield* runtime.withPublication( + if (llmFailure && !publisher.hasProviderError()) { + yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) + yield* withPublication( events.publish(SessionEvent.Step.Failed, { sessionID: prepared.session.id, timestamp: yield* DateTime.now, - assistantMessageID: yield* runtime.publisher.startAssistant(), + assistantMessageID: yield* publisher.startAssistant(), error: { type: "unknown", message: llmFailure.reason.message }, }), ) } - if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(runtime.toolFibers) - const settled = yield* restore(awaitToolFibers(runtime.toolFibers)).pipe(Effect.exit) + if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers) + const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { - yield* FiberSet.clear(runtime.toolFibers) - yield* runtime.withPublication(runtime.publisher.failUnsettledTools("Tool execution interrupted")) + yield* FiberSet.clear(toolFibers) + yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) return yield* Effect.interrupt } if ( (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) || (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)) ) { - yield* FiberSet.clear(runtime.toolFibers) - yield* runtime.withPublication(runtime.publisher.failUnsettledTools("Tool execution interrupted")) + yield* FiberSet.clear(toolFibers) + yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) } if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) { const failure = Cause.squash(settled.cause) const message = failure instanceof Error ? failure.message : String(failure) - yield* runtime.withPublication(runtime.publisher.failUnsettledTools(`Tool execution failed: ${message}`)) + yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) } - if (runtime.publisher.hasProviderError()) - yield* runtime.withPublication(runtime.publisher.failUnsettledTools("Tool execution interrupted")) - if (stream._tag === "Success" && !runtime.publisher.hasProviderError()) - yield* runtime.withPublication( - runtime.publisher.failUnsettledTools("Provider did not return a tool result", true), - ) + if (publisher.hasProviderError()) + yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + if (stream._tag === "Success" && !publisher.hasProviderError()) + 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 AttemptResult.cases.Complete.make({ - needsContinuation: !runtime.publisher.hasProviderError() && runtime.needsContinuation, + needsContinuation: !publisher.hasProviderError() && needsContinuation, }) }), ) }, Effect.scoped) - const run: Run = Effect.fnUntraced(function* (sessionID, promotion) { - const first = yield* runAttempt(sessionID, promotion, compaction.compactAfterOverflow) - if (first._tag === "Complete") return first.needsContinuation - const second = yield* runAttempt(sessionID, undefined) - return AttemptResult.match(second, { - Complete: (result) => result.needsContinuation, - CompactedOverflow: () => false, - }) + const run: Run = Effect.fn("SessionRunner.runTurn")(function* (input) { + let pendingDelivery = input.delivery + let canRecoverOverflow = true + while (true) { + const request = yield* buildRequest(input.sessionID, pendingDelivery) + pendingDelivery = undefined + const result = yield* streamAndSettle(request, canRecoverOverflow ? compaction.compactAfterOverflow : undefined) + const next = AttemptResult.match(result, { + Complete: (completed) => completed.needsContinuation, + CompactedOverflow: () => undefined, + }) + if (next !== undefined) return next + canRecoverOverflow = false + } }) return run From b795f415af455a8d6c0ad0b5a05e8430dfb5ac4e Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:27:19 -0400 Subject: [PATCH 06/10] refactor(core): clarify provider turn helpers --- packages/core/src/session/runner/run-turn.ts | 103 +++++++++---------- 1 file changed, 51 insertions(+), 52 deletions(-) diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index 9790a90764..cbcda8ffd4 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -49,8 +49,6 @@ export interface Input { readonly delivery?: SessionInput.Delivery } -export type Run = (input: Input) => Effect.Effect - const AttemptResult = Schema.TaggedUnion({ Complete: { needsContinuation: Schema.Boolean }, CompactedOverflow: {}, @@ -90,6 +88,11 @@ export const make = Effect.gen(function* () { Effect.all([systemContext.load(), skillGuidance.load(agent)], { concurrency: "unbounded" }).pipe( Effect.map(SystemContext.combine), ) + const promoteDelivery = Effect.fnUntraced(function* (sessionID: SessionSchema.ID, delivery: SessionInput.Delivery) { + const cutoff = yield* SessionInput.latestSeq(db, sessionID) + if (delivery === "queue") yield* SessionInput.promoteNextQueued(db, events, sessionID) + yield* SessionInput.promoteSteers(db, events, sessionID, cutoff) + }) /** * Builds the next model request from durable Session state. @@ -113,12 +116,7 @@ export const make = Effect.gen(function* () { ) if (initialized === stale) continue if (pendingDelivery) { - const cutoff = yield* SessionInput.latestSeq(db, session.id) - if (pendingDelivery === "steer") yield* SessionInput.promoteSteers(db, events, session.id, cutoff) - if (pendingDelivery === "queue") { - yield* SessionInput.promoteNextQueued(db, events, session.id) - yield* SessionInput.promoteSteers(db, events, session.id, cutoff) - } + yield* promoteDelivery(session.id, pendingDelivery) pendingDelivery = undefined } const prepared = @@ -167,7 +165,7 @@ export const make = Effect.gen(function* () { */ const streamAndSettle = Effect.fn("SessionRunner.streamAndSettle")(function* ( prepared: RequestSnapshot, - recoverOverflow?: typeof compaction.compactAfterOverflow, + canRecoverOverflow: boolean, ) { const publisher = createLLMEventPublisher(events, { sessionID: prepared.session.id, @@ -181,9 +179,37 @@ export const make = Effect.gen(function* () { const withPublication = Semaphore.makeUnsafe(1).withPermit const publish = (event: LLMEvent, outputPaths: ReadonlyArray = []) => withPublication(publisher.publish(event, outputPaths)) + const failUnsettled = (message: string, providerExecuted = false) => + withPublication(publisher.failUnsettledTools(message, providerExecuted)) const toolFibers = yield* FiberSet.make() let needsContinuation = false let overflowFailure: ProviderErrorEvent | undefined + const startTool = Effect.fnUntraced(function* (event: Extract) { + needsContinuation = true + const assistantMessageID = yield* publisher.assistantMessageID(event.id) + yield* Effect.uninterruptibleMask((restore) => + restore( + prepared.toolMaterialization.settle({ + sessionID: prepared.session.id, + agent: prepared.agent.id, + assistantMessageID, + call: event, + }), + ).pipe( + Effect.flatMap((settlement) => + publish( + LLMEvent.toolResult({ + id: event.id, + name: event.name, + result: settlement.result, + output: settlement.output, + }), + settlement.outputPaths ?? [], + ), + ), + ), + ).pipe(FiberSet.run(toolFibers)) + }) const providerStream = llm.stream(prepared.request).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -194,30 +220,7 @@ export const make = Effect.gen(function* () { } yield* publish(event) if (event.type !== "tool-call" || event.providerExecuted) return - needsContinuation = true - const assistantMessageID = yield* publisher.assistantMessageID(event.id) - yield* Effect.uninterruptibleMask((restore) => - restore( - prepared.toolMaterialization.settle({ - sessionID: prepared.session.id, - agent: prepared.agent.id, - assistantMessageID, - call: event, - }), - ).pipe( - Effect.flatMap((settlement) => - publish( - LLMEvent.toolResult({ - id: event.id, - name: event.name, - result: settlement.result, - output: settlement.output, - }), - settlement.outputPaths ?? [], - ), - ), - ), - ).pipe(FiberSet.run(toolFibers)) + yield* startTool(event) }), ), Effect.ensuring(withPublication(publisher.flush())), @@ -231,11 +234,11 @@ export const make = Effect.gen(function* () { const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined if ( - recoverOverflow && + canRecoverOverflow && !publisher.hasAssistantStarted() && isContextOverflowFailure(overflowFailure ?? failure) && (yield* restore( - recoverOverflow({ + compaction.compactAfterOverflow({ sessionID: prepared.session.id, entries: prepared.entries, model: prepared.model, @@ -247,7 +250,7 @@ export const make = Effect.gen(function* () { if (overflowFailure) yield* publish(overflowFailure) const llmFailure = failure instanceof LLMError ? failure : undefined if (llmFailure && !publisher.hasProviderError()) { - yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) + yield* failUnsettled("Provider did not return a tool result", true) yield* withPublication( events.publish(SessionEvent.Step.Failed, { sessionID: prepared.session.id, @@ -257,29 +260,25 @@ export const make = Effect.gen(function* () { }), ) } - if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers) + const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) + if (streamInterrupted) yield* FiberSet.clear(toolFibers) const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { yield* FiberSet.clear(toolFibers) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + yield* failUnsettled("Tool execution interrupted") return yield* Effect.interrupt } - if ( - (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) || - (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)) - ) { - yield* FiberSet.clear(toolFibers) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) - } - if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) { + const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) + if (toolInterrupted) yield* FiberSet.clear(toolFibers) + if (streamInterrupted || toolInterrupted || publisher.hasProviderError()) + yield* failUnsettled("Tool execution interrupted") + if (settled._tag === "Failure" && !toolInterrupted) { const failure = Cause.squash(settled.cause) const message = failure instanceof Error ? failure.message : String(failure) - yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) + yield* failUnsettled(`Tool execution failed: ${message}`) } - if (publisher.hasProviderError()) - yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) if (stream._tag === "Success" && !publisher.hasProviderError()) - yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) + yield* failUnsettled("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 AttemptResult.cases.Complete.make({ @@ -289,13 +288,13 @@ export const make = Effect.gen(function* () { ) }, Effect.scoped) - const run: Run = Effect.fn("SessionRunner.runTurn")(function* (input) { + const run = Effect.fn("SessionRunner.runTurn")(function* (input: Input): Effect.fn.Return { let pendingDelivery = input.delivery let canRecoverOverflow = true while (true) { const request = yield* buildRequest(input.sessionID, pendingDelivery) pendingDelivery = undefined - const result = yield* streamAndSettle(request, canRecoverOverflow ? compaction.compactAfterOverflow : undefined) + const result = yield* streamAndSettle(request, canRecoverOverflow) const next = AttemptResult.match(result, { Complete: (completed) => completed.needsContinuation, CompactedOverflow: () => undefined, From cacb08c1be68c7436f3f812a9fb24bf3d931ad51 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:30:40 -0400 Subject: [PATCH 07/10] refactor(core): isolate provider turn settlement --- packages/core/src/session/runner/run-turn.ts | 149 ++++++++---------- .../session/runner/settle-provider-turn.ts | 71 +++++++++ 2 files changed, 133 insertions(+), 87 deletions(-) create mode 100644 packages/core/src/session/runner/settle-provider-turn.ts diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index cbcda8ffd4..404fa4dfed 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -18,7 +18,7 @@ import { isContextOverflowFailure, type ProviderErrorEvent, } from "@opencode-ai/llm" -import { Cause, DateTime, Effect, FiberSet, Option, Schema, Semaphore, Stream } from "effect" +import { DateTime, Effect, Schema, Semaphore, Stream } from "effect" import { AgentV2 } from "../../agent" import { Config } from "../../config" import { Database } from "../../database/database" @@ -26,11 +26,9 @@ import { EventV2 } from "../../event" import { Location } from "../../location" import { ModelV2 } from "../../model" import { ProviderV2 } from "../../provider" -import { QuestionV2 } from "../../question" import { SkillGuidance } from "../../skill/guidance" import { SystemContext } from "../../system-context/index" import { SystemContextRegistry } from "../../system-context/registry" -import { ToolOutputStore } from "../../tool-output-store" import { ToolRegistry } from "../../tool/registry" import { SessionCompaction } from "../compaction" import { SessionContextEpoch } from "../context-epoch" @@ -42,6 +40,7 @@ import { SessionStore } from "../store" import type { RunError } from "./index" import { SessionRunnerModel } from "./model" import { createLLMEventPublisher } from "./publish-llm-event" +import { SettleProviderTurn } from "./settle-provider-turn" import { toLLMMessages } from "./to-llm-message" export interface Input { @@ -73,10 +72,6 @@ export const make = Effect.gen(function* () { if (!session) return yield* Effect.die(`Session not found: ${sessionID}`) return session }) - const awaitToolFibers = (fibers: FiberSet.FiberSet) => - Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers)) - const isQuestionRejected = (cause: Cause.Cause) => - cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) const stale = Symbol("stale turn preparation") const retryAgentMismatch = (effect: Effect.Effect) => effect.pipe( @@ -181,21 +176,19 @@ export const make = Effect.gen(function* () { withPublication(publisher.publish(event, outputPaths)) const failUnsettled = (message: string, providerExecuted = false) => withPublication(publisher.failUnsettledTools(message, providerExecuted)) - const toolFibers = yield* FiberSet.make() let needsContinuation = false let overflowFailure: ProviderErrorEvent | undefined - const startTool = Effect.fnUntraced(function* (event: Extract) { + const toolEffect = Effect.fnUntraced(function* (event: Extract) { needsContinuation = true const assistantMessageID = yield* publisher.assistantMessageID(event.id) - yield* Effect.uninterruptibleMask((restore) => - restore( - prepared.toolMaterialization.settle({ - sessionID: prepared.session.id, - agent: prepared.agent.id, - assistantMessageID, - call: event, - }), - ).pipe( + return yield* prepared.toolMaterialization + .settle({ + sessionID: prepared.session.id, + agent: prepared.agent.id, + assistantMessageID, + call: event, + }) + .pipe( Effect.flatMap((settlement) => publish( LLMEvent.toolResult({ @@ -207,85 +200,67 @@ export const make = Effect.gen(function* () { settlement.outputPaths ?? [], ), ), - ), - ).pipe(FiberSet.run(toolFibers)) + ) }) - const providerStream = llm.stream(prepared.request).pipe( - Stream.runForEach((event) => + const result = yield* SettleProviderTurn.run({ + stream: (runTool) => + llm.stream(prepared.request).pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (overflowFailure || publisher.hasProviderError()) return + if ( + LLMEvent.is.providerError(event) && + isContextOverflowFailure(event) && + !publisher.hasAssistantStarted() + ) { + overflowFailure = event + return + } + yield* publish(event) + if (event.type !== "tool-call" || event.providerExecuted) return + yield* runTool(toolEffect(event)) + }), + ), + Effect.ensuring(withPublication(publisher.flush())), + ), + recoverOverflow: (failure, restore) => Effect.gen(function* () { - if (overflowFailure || publisher.hasProviderError()) return - if (LLMEvent.is.providerError(event) && isContextOverflowFailure(event) && !publisher.hasAssistantStarted()) { - overflowFailure = event - return - } - yield* publish(event) - if (event.type !== "tool-call" || event.providerExecuted) return - yield* startTool(event) - }), - ), - Effect.ensuring(withPublication(publisher.flush())), - ) - - // Keep cleanup protected after the response ends so no started tool is - // forgotten, while the response stream and tool work remain interruptible. - return yield* Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - const stream = yield* restore(providerStream).pipe(Effect.exit) - const failure = - stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined - if ( - canRecoverOverflow && - !publisher.hasAssistantStarted() && - isContextOverflowFailure(overflowFailure ?? failure) && - (yield* restore( + if (!canRecoverOverflow || publisher.hasAssistantStarted()) return false + if (!isContextOverflowFailure(overflowFailure ?? failure)) return false + return yield* restore( compaction.compactAfterOverflow({ sessionID: prepared.session.id, entries: prepared.entries, model: prepared.model, request: prepared.request, }), - )) - ) - return AttemptResult.cases.CompactedOverflow.make({}) - if (overflowFailure) yield* publish(overflowFailure) - const llmFailure = failure instanceof LLMError ? failure : undefined - if (llmFailure && !publisher.hasProviderError()) { - yield* failUnsettled("Provider did not return a tool result", true) - yield* withPublication( - events.publish(SessionEvent.Step.Failed, { - sessionID: prepared.session.id, - timestamp: yield* DateTime.now, - assistantMessageID: yield* publisher.startAssistant(), - error: { type: "unknown", message: llmFailure.reason.message }, - }), ) - } - const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) - if (streamInterrupted) yield* FiberSet.clear(toolFibers) - const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) - if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { - yield* FiberSet.clear(toolFibers) - yield* failUnsettled("Tool execution interrupted") - return yield* Effect.interrupt - } - const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) - if (toolInterrupted) yield* FiberSet.clear(toolFibers) - if (streamInterrupted || toolInterrupted || publisher.hasProviderError()) - yield* failUnsettled("Tool execution interrupted") - if (settled._tag === "Failure" && !toolInterrupted) { - const failure = Cause.squash(settled.cause) - const message = failure instanceof Error ? failure.message : String(failure) - yield* failUnsettled(`Tool execution failed: ${message}`) - } - if (stream._tag === "Success" && !publisher.hasProviderError()) - yield* failUnsettled("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 AttemptResult.cases.Complete.make({ + }), + projectProviderFailure: (failure) => + Effect.gen(function* () { + if (overflowFailure) yield* publish(overflowFailure) + if (failure instanceof LLMError && !publisher.hasProviderError()) { + yield* failUnsettled("Provider did not return a tool result", true) + yield* withPublication( + events.publish(SessionEvent.Step.Failed, { + sessionID: prepared.session.id, + timestamp: yield* DateTime.now, + assistantMessageID: yield* publisher.startAssistant(), + error: { type: "unknown", message: failure.reason.message }, + }), + ) + } + }), + hasProviderError: publisher.hasProviderError, + failUnsettled, + }) + return SettleProviderTurn.Result.match(result, { + RecoveredOverflow: () => AttemptResult.cases.CompactedOverflow.make({}), + Complete: () => + AttemptResult.cases.Complete.make({ needsContinuation: !publisher.hasProviderError() && needsContinuation, - }) - }), - ) + }), + }) }, Effect.scoped) const run = Effect.fn("SessionRunner.runTurn")(function* (input: Input): Effect.fn.Return { diff --git a/packages/core/src/session/runner/settle-provider-turn.ts b/packages/core/src/session/runner/settle-provider-turn.ts new file mode 100644 index 0000000000..2e6302bfb2 --- /dev/null +++ b/packages/core/src/session/runner/settle-provider-turn.ts @@ -0,0 +1,71 @@ +export * as SettleProviderTurn from "./settle-provider-turn" + +import { Cause, Effect, FiberSet, Option, Schema } from "effect" +import { QuestionV2 } from "../../question" +import { ToolOutputStore } from "../../tool-output-store" + +export const Result = Schema.TaggedUnion({ + Complete: {}, + RecoveredOverflow: {}, +}) + +const isQuestionRejected = (cause: Cause.Cause) => + cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) + +const awaitTools = (fibers: FiberSet.FiberSet) => + Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers)) + +/** + * Runs one provider response together with every local tool it starts. + * + * The provider and tools remain interruptible. Once the provider stops, cleanup + * cannot be interrupted before every started tool is observed, cancelled, or + * settled and its original failure is propagated. + */ +export const run = Effect.fn("SessionRunner.settleProviderTurn")(function* (input: { + readonly stream: ( + runTool: (effect: Effect.Effect) => Effect.Effect, + ) => Effect.Effect + readonly recoverOverflow: ( + failure: unknown, + restore: (effect: Effect.Effect) => Effect.Effect, + ) => Effect.Effect + readonly projectProviderFailure: (failure: unknown) => Effect.Effect + readonly hasProviderError: () => boolean + readonly failUnsettled: (message: string, providerExecuted?: boolean) => Effect.Effect +}) { + const tools = yield* FiberSet.make() + return yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const stream = yield* restore(input.stream((effect) => effect.pipe(FiberSet.run(tools)))).pipe(Effect.exit) + const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined + if (yield* input.recoverOverflow(failure, restore)) return Result.cases.RecoveredOverflow.make({}) + + yield* input.projectProviderFailure(failure) + const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) + if (streamInterrupted) yield* FiberSet.clear(tools) + const settled = yield* restore(awaitTools(tools)).pipe(Effect.exit) + if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { + yield* FiberSet.clear(tools) + yield* input.failUnsettled("Tool execution interrupted") + return yield* Effect.interrupt + } + + const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) + if (toolInterrupted) yield* FiberSet.clear(tools) + if (streamInterrupted || toolInterrupted || input.hasProviderError()) + yield* input.failUnsettled("Tool execution interrupted") + if (settled._tag === "Failure" && !toolInterrupted) { + const failure = Cause.squash(settled.cause) + yield* input.failUnsettled( + `Tool execution failed: ${failure instanceof Error ? failure.message : String(failure)}`, + ) + } + if (stream._tag === "Success" && !input.hasProviderError()) + yield* input.failUnsettled("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 Result.cases.Complete.make({}) + }), + ) +}, Effect.scoped) From ad639f89fbf4ec71e16727ad69cd944eda120ae0 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:33:09 -0400 Subject: [PATCH 08/10] refactor(core): localize provider turn settlement --- packages/core/src/session/runner/run-turn.ts | 121 +++++++++++------- .../session/runner/settle-provider-turn.ts | 71 ---------- 2 files changed, 76 insertions(+), 116 deletions(-) delete mode 100644 packages/core/src/session/runner/settle-provider-turn.ts diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index 404fa4dfed..273e84830f 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -18,7 +18,7 @@ import { isContextOverflowFailure, type ProviderErrorEvent, } from "@opencode-ai/llm" -import { DateTime, Effect, Schema, Semaphore, Stream } from "effect" +import { Cause, DateTime, Effect, FiberSet, Option, Schema, Semaphore, Stream } from "effect" import { AgentV2 } from "../../agent" import { Config } from "../../config" import { Database } from "../../database/database" @@ -26,9 +26,11 @@ import { EventV2 } from "../../event" import { Location } from "../../location" import { ModelV2 } from "../../model" import { ProviderV2 } from "../../provider" +import { QuestionV2 } from "../../question" import { SkillGuidance } from "../../skill/guidance" import { SystemContext } from "../../system-context/index" import { SystemContextRegistry } from "../../system-context/registry" +import { ToolOutputStore } from "../../tool-output-store" import { ToolRegistry } from "../../tool/registry" import { SessionCompaction } from "../compaction" import { SessionContextEpoch } from "../context-epoch" @@ -40,7 +42,6 @@ import { SessionStore } from "../store" import type { RunError } from "./index" import { SessionRunnerModel } from "./model" import { createLLMEventPublisher } from "./publish-llm-event" -import { SettleProviderTurn } from "./settle-provider-turn" import { toLLMMessages } from "./to-llm-message" export interface Input { @@ -53,6 +54,12 @@ const AttemptResult = Schema.TaggedUnion({ CompactedOverflow: {}, }) +const isQuestionRejected = (cause: Cause.Cause) => + cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) + +const awaitTools = (fibers: FiberSet.FiberSet) => + Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers)) + export const make = Effect.gen(function* () { const events = yield* EventV2.Service const llm = yield* LLMClient.Service @@ -202,42 +209,48 @@ export const make = Effect.gen(function* () { ), ) }) - const result = yield* SettleProviderTurn.run({ - stream: (runTool) => - llm.stream(prepared.request).pipe( - Stream.runForEach((event) => - Effect.gen(function* () { - if (overflowFailure || publisher.hasProviderError()) return - if ( - LLMEvent.is.providerError(event) && - isContextOverflowFailure(event) && - !publisher.hasAssistantStarted() - ) { - overflowFailure = event - return - } - yield* publish(event) - if (event.type !== "tool-call" || event.providerExecuted) return - yield* runTool(toolEffect(event)) - }), - ), - Effect.ensuring(withPublication(publisher.flush())), - ), - recoverOverflow: (failure, restore) => + const settleProviderTurn = Effect.fnUntraced(function* () { + const tools = yield* FiberSet.make() + return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { - if (!canRecoverOverflow || publisher.hasAssistantStarted()) return false - if (!isContextOverflowFailure(overflowFailure ?? failure)) return false - return yield* restore( - compaction.compactAfterOverflow({ - sessionID: prepared.session.id, - entries: prepared.entries, - model: prepared.model, - request: prepared.request, - }), + const stream = yield* restore( + llm.stream(prepared.request).pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (overflowFailure || publisher.hasProviderError()) return + if ( + LLMEvent.is.providerError(event) && + isContextOverflowFailure(event) && + !publisher.hasAssistantStarted() + ) { + overflowFailure = event + return + } + yield* publish(event) + if (event.type !== "tool-call" || event.providerExecuted) return + yield* toolEffect(event).pipe(FiberSet.run(tools)) + }), + ), + Effect.ensuring(withPublication(publisher.flush())), + ), + ).pipe(Effect.exit) + const failure = + stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined + if ( + canRecoverOverflow && + !publisher.hasAssistantStarted() && + isContextOverflowFailure(overflowFailure ?? failure) && + (yield* restore( + compaction.compactAfterOverflow({ + sessionID: prepared.session.id, + entries: prepared.entries, + model: prepared.model, + request: prepared.request, + }), + )) ) - }), - projectProviderFailure: (failure) => - Effect.gen(function* () { + return AttemptResult.cases.CompactedOverflow.make({}) + if (overflowFailure) yield* publish(overflowFailure) if (failure instanceof LLMError && !publisher.hasProviderError()) { yield* failUnsettled("Provider did not return a tool result", true) @@ -250,17 +263,35 @@ export const make = Effect.gen(function* () { }), ) } + const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) + if (streamInterrupted) yield* FiberSet.clear(tools) + const settled = yield* restore(awaitTools(tools)).pipe(Effect.exit) + if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { + yield* FiberSet.clear(tools) + yield* failUnsettled("Tool execution interrupted") + return yield* Effect.interrupt + } + const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) + if (toolInterrupted) yield* FiberSet.clear(tools) + if (streamInterrupted || toolInterrupted || publisher.hasProviderError()) + yield* failUnsettled("Tool execution interrupted") + if (settled._tag === "Failure" && !toolInterrupted) { + const failure = Cause.squash(settled.cause) + yield* failUnsettled( + `Tool execution failed: ${failure instanceof Error ? failure.message : String(failure)}`, + ) + } + if (stream._tag === "Success" && !publisher.hasProviderError()) + yield* failUnsettled("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 AttemptResult.cases.Complete.make({ + needsContinuation: !publisher.hasProviderError() && needsContinuation, + }) }), - hasProviderError: publisher.hasProviderError, - failUnsettled, - }) - return SettleProviderTurn.Result.match(result, { - RecoveredOverflow: () => AttemptResult.cases.CompactedOverflow.make({}), - Complete: () => - AttemptResult.cases.Complete.make({ - needsContinuation: !publisher.hasProviderError() && needsContinuation, - }), + ) }) + return yield* settleProviderTurn() }, Effect.scoped) const run = Effect.fn("SessionRunner.runTurn")(function* (input: Input): Effect.fn.Return { diff --git a/packages/core/src/session/runner/settle-provider-turn.ts b/packages/core/src/session/runner/settle-provider-turn.ts deleted file mode 100644 index 2e6302bfb2..0000000000 --- a/packages/core/src/session/runner/settle-provider-turn.ts +++ /dev/null @@ -1,71 +0,0 @@ -export * as SettleProviderTurn from "./settle-provider-turn" - -import { Cause, Effect, FiberSet, Option, Schema } from "effect" -import { QuestionV2 } from "../../question" -import { ToolOutputStore } from "../../tool-output-store" - -export const Result = Schema.TaggedUnion({ - Complete: {}, - RecoveredOverflow: {}, -}) - -const isQuestionRejected = (cause: Cause.Cause) => - cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError) - -const awaitTools = (fibers: FiberSet.FiberSet) => - Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers)) - -/** - * Runs one provider response together with every local tool it starts. - * - * The provider and tools remain interruptible. Once the provider stops, cleanup - * cannot be interrupted before every started tool is observed, cancelled, or - * settled and its original failure is propagated. - */ -export const run = Effect.fn("SessionRunner.settleProviderTurn")(function* (input: { - readonly stream: ( - runTool: (effect: Effect.Effect) => Effect.Effect, - ) => Effect.Effect - readonly recoverOverflow: ( - failure: unknown, - restore: (effect: Effect.Effect) => Effect.Effect, - ) => Effect.Effect - readonly projectProviderFailure: (failure: unknown) => Effect.Effect - readonly hasProviderError: () => boolean - readonly failUnsettled: (message: string, providerExecuted?: boolean) => Effect.Effect -}) { - const tools = yield* FiberSet.make() - return yield* Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - const stream = yield* restore(input.stream((effect) => effect.pipe(FiberSet.run(tools)))).pipe(Effect.exit) - const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined - if (yield* input.recoverOverflow(failure, restore)) return Result.cases.RecoveredOverflow.make({}) - - yield* input.projectProviderFailure(failure) - const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) - if (streamInterrupted) yield* FiberSet.clear(tools) - const settled = yield* restore(awaitTools(tools)).pipe(Effect.exit) - if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { - yield* FiberSet.clear(tools) - yield* input.failUnsettled("Tool execution interrupted") - return yield* Effect.interrupt - } - - const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) - if (toolInterrupted) yield* FiberSet.clear(tools) - if (streamInterrupted || toolInterrupted || input.hasProviderError()) - yield* input.failUnsettled("Tool execution interrupted") - if (settled._tag === "Failure" && !toolInterrupted) { - const failure = Cause.squash(settled.cause) - yield* input.failUnsettled( - `Tool execution failed: ${failure instanceof Error ? failure.message : String(failure)}`, - ) - } - if (stream._tag === "Success" && !input.hasProviderError()) - yield* input.failUnsettled("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 Result.cases.Complete.make({}) - }), - ) -}, Effect.scoped) From 576a541f671a2733e0165a38d60244e3f9958528 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 22:34:27 -0400 Subject: [PATCH 09/10] refactor(core): clarify provider turn handlers --- packages/core/src/session/runner/run-turn.ts | 122 ++++++++++--------- 1 file changed, 63 insertions(+), 59 deletions(-) diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index 273e84830f..1658bfd0fa 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -18,7 +18,7 @@ import { isContextOverflowFailure, type ProviderErrorEvent, } from "@opencode-ai/llm" -import { Cause, DateTime, Effect, FiberSet, Option, Schema, Semaphore, Stream } from "effect" +import { Cause, DateTime, Effect, Exit, FiberSet, Option, Schema, Semaphore, Stream } from "effect" import { AgentV2 } from "../../agent" import { Config } from "../../config" import { Database } from "../../database/database" @@ -209,34 +209,71 @@ export const make = Effect.gen(function* () { ), ) }) + const handleProviderEvent = Effect.fnUntraced(function* ( + event: LLMEvent, + tools: FiberSet.FiberSet, + ) { + if (overflowFailure || publisher.hasProviderError()) return + if (LLMEvent.is.providerError(event) && isContextOverflowFailure(event) && !publisher.hasAssistantStarted()) { + overflowFailure = event + return + } + yield* publish(event) + if (event.type !== "tool-call" || event.providerExecuted) return + yield* toolEffect(event).pipe(FiberSet.run(tools)) + }) + const providerStream = (tools: FiberSet.FiberSet) => + llm.stream(prepared.request).pipe( + Stream.runForEach((event) => handleProviderEvent(event, tools)), + Effect.ensuring(withPublication(publisher.flush())), + ) + const projectProviderFailure = Effect.fnUntraced(function* (failure: unknown) { + if (overflowFailure) yield* publish(overflowFailure) + if (!(failure instanceof LLMError) || publisher.hasProviderError()) return + yield* failUnsettled("Provider did not return a tool result", true) + yield* withPublication( + events.publish(SessionEvent.Step.Failed, { + sessionID: prepared.session.id, + timestamp: yield* DateTime.now, + assistantMessageID: yield* publisher.startAssistant(), + error: { type: "unknown", message: failure.reason.message }, + }), + ) + }) + const finishTools = Effect.fnUntraced(function* ( + stream: Exit.Exit, + tools: FiberSet.FiberSet, + wait: Effect.Effect, + ) { + const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) + if (streamInterrupted) yield* FiberSet.clear(tools) + const settled = yield* wait.pipe(Effect.exit) + if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { + yield* FiberSet.clear(tools) + yield* failUnsettled("Tool execution interrupted") + return yield* Effect.interrupt + } + const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) + if (toolInterrupted) yield* FiberSet.clear(tools) + if (streamInterrupted || toolInterrupted || publisher.hasProviderError()) + yield* failUnsettled("Tool execution interrupted") + if (settled._tag === "Failure" && !toolInterrupted) { + const failure = Cause.squash(settled.cause) + yield* failUnsettled(`Tool execution failed: ${failure instanceof Error ? failure.message : String(failure)}`) + } + if (stream._tag === "Success" && !publisher.hasProviderError()) + yield* failUnsettled("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) + }) const settleProviderTurn = Effect.fnUntraced(function* () { const tools = yield* FiberSet.make() return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { - const stream = yield* restore( - llm.stream(prepared.request).pipe( - Stream.runForEach((event) => - Effect.gen(function* () { - if (overflowFailure || publisher.hasProviderError()) return - if ( - LLMEvent.is.providerError(event) && - isContextOverflowFailure(event) && - !publisher.hasAssistantStarted() - ) { - overflowFailure = event - return - } - yield* publish(event) - if (event.type !== "tool-call" || event.providerExecuted) return - yield* toolEffect(event).pipe(FiberSet.run(tools)) - }), - ), - Effect.ensuring(withPublication(publisher.flush())), - ), - ).pipe(Effect.exit) + const stream = yield* restore(providerStream(tools)).pipe(Effect.exit) const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined - if ( + const recovered = canRecoverOverflow && !publisher.hasAssistantStarted() && isContextOverflowFailure(overflowFailure ?? failure) && @@ -248,43 +285,10 @@ export const make = Effect.gen(function* () { request: prepared.request, }), )) - ) - return AttemptResult.cases.CompactedOverflow.make({}) + if (recovered) return AttemptResult.cases.CompactedOverflow.make({}) - if (overflowFailure) yield* publish(overflowFailure) - if (failure instanceof LLMError && !publisher.hasProviderError()) { - yield* failUnsettled("Provider did not return a tool result", true) - yield* withPublication( - events.publish(SessionEvent.Step.Failed, { - sessionID: prepared.session.id, - timestamp: yield* DateTime.now, - assistantMessageID: yield* publisher.startAssistant(), - error: { type: "unknown", message: failure.reason.message }, - }), - ) - } - const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) - if (streamInterrupted) yield* FiberSet.clear(tools) - const settled = yield* restore(awaitTools(tools)).pipe(Effect.exit) - if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { - yield* FiberSet.clear(tools) - yield* failUnsettled("Tool execution interrupted") - return yield* Effect.interrupt - } - const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) - if (toolInterrupted) yield* FiberSet.clear(tools) - if (streamInterrupted || toolInterrupted || publisher.hasProviderError()) - yield* failUnsettled("Tool execution interrupted") - if (settled._tag === "Failure" && !toolInterrupted) { - const failure = Cause.squash(settled.cause) - yield* failUnsettled( - `Tool execution failed: ${failure instanceof Error ? failure.message : String(failure)}`, - ) - } - if (stream._tag === "Success" && !publisher.hasProviderError()) - yield* failUnsettled("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) + yield* projectProviderFailure(failure) + yield* finishTools(stream, tools, restore(awaitTools(tools))) return AttemptResult.cases.Complete.make({ needsContinuation: !publisher.hasProviderError() && needsContinuation, }) From 7ad5c27e7ce5b2eb1963bfdb945d23c200cc18f6 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 6 Jun 2026 23:01:43 -0400 Subject: [PATCH 10/10] refactor(core): simplify tool settlement outcomes --- packages/core/src/session/runner/run-turn.ts | 54 ++++++++++++-------- 1 file changed, 34 insertions(+), 20 deletions(-) diff --git a/packages/core/src/session/runner/run-turn.ts b/packages/core/src/session/runner/run-turn.ts index 1658bfd0fa..196d591861 100644 --- a/packages/core/src/session/runner/run-turn.ts +++ b/packages/core/src/session/runner/run-turn.ts @@ -240,31 +240,45 @@ export const make = Effect.gen(function* () { }), ) }) + const finishToolsSuccessfully = Effect.fnUntraced(function* (stream: Exit.Exit) { + if (Exit.hasInterrupts(stream) || publisher.hasProviderError()) yield* failUnsettled("Tool execution interrupted") + if (Exit.isSuccess(stream) && !publisher.hasProviderError()) + yield* failUnsettled("Provider did not return a tool result", true) + if (Exit.isFailure(stream)) return yield* Effect.failCause(stream.cause) + }) + const finishToolsAfterFailure = Effect.fnUntraced(function* ( + stream: Exit.Exit, + tools: FiberSet.FiberSet, + cause: Cause.Cause, + ) { + if (isQuestionRejected(cause)) { + yield* FiberSet.clear(tools) + yield* failUnsettled("Tool execution interrupted") + return yield* Effect.interrupt + } + const interrupted = Cause.hasInterrupts(cause) + if (interrupted) yield* FiberSet.clear(tools) + if (Exit.hasInterrupts(stream) || interrupted || publisher.hasProviderError()) + yield* failUnsettled("Tool execution interrupted") + if (!interrupted) { + const failure = Cause.squash(cause) + yield* failUnsettled(`Tool execution failed: ${failure instanceof Error ? failure.message : String(failure)}`) + } + if (Exit.isFailure(stream)) return yield* Effect.failCause(stream.cause) + return yield* Effect.failCause(cause) + }) const finishTools = Effect.fnUntraced(function* ( stream: Exit.Exit, tools: FiberSet.FiberSet, wait: Effect.Effect, ) { - const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) - if (streamInterrupted) yield* FiberSet.clear(tools) - const settled = yield* wait.pipe(Effect.exit) - if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { - yield* FiberSet.clear(tools) - yield* failUnsettled("Tool execution interrupted") - return yield* Effect.interrupt - } - const toolInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) - if (toolInterrupted) yield* FiberSet.clear(tools) - if (streamInterrupted || toolInterrupted || publisher.hasProviderError()) - yield* failUnsettled("Tool execution interrupted") - if (settled._tag === "Failure" && !toolInterrupted) { - const failure = Cause.squash(settled.cause) - yield* failUnsettled(`Tool execution failed: ${failure instanceof Error ? failure.message : String(failure)}`) - } - if (stream._tag === "Success" && !publisher.hasProviderError()) - yield* failUnsettled("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) + if (Exit.hasInterrupts(stream)) yield* FiberSet.clear(tools) + return yield* wait.pipe( + Effect.matchCauseEffect({ + onFailure: (cause) => finishToolsAfterFailure(stream, tools, cause), + onSuccess: () => finishToolsSuccessfully(stream), + }), + ) }) const settleProviderTurn = Effect.fnUntraced(function* () { const tools = yield* FiberSet.make()