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

fix(core): fail drains on tool output persistence failures

Browse source 
Restructure runner turn settlement: replace TurnTransitionError defect
control flow with tagged turn results and a plain restart loop, name the
settlement evidence, and merge duplicated interrupt branches.

Split settled tool-fiber failures: tool implementation defects keep
surfacing as model-visible tool errors and the turn continues, while
typed ToolOutputStore errors now fail the assistant and the drain
instead of being silently absorbed.
This commit is contained in:
Kit Langton 2026-07-02 16:19:46 -04:00
commit 0d1ef5adb6
2 changed files with 137 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -32,7 +32,7 @@ import { SessionInput } from "../input"
import { SessionSchema } from "../schema"
import { SessionStore } from "../store"
import { SessionTitle } from "../title"
import { type RunError, Service } from "./index"
import { Service } from "./index"
import { SessionRunnerModel } from "./model"
import { createLLMEventPublisher } from "./publish-llm-event"
import { toLLMMessages } from "./to-llm-message"
@ -153,22 +153,6 @@ const layer = Layer.effect(
const isQuestionRejected = (cause: Cause.Cause<unknown>) =>
cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError)
type TurnTransition =
// Automatic compaction completed; rebuild the request from compacted history.
| { readonly _tag: "ContinueAfterCompaction"; readonly step: number }
// Overflow compaction completed; rebuild once through the path without overflow recovery.
| { readonly _tag: "ContinueAfterOverflowCompaction"; readonly step: number }
class TurnTransitionError extends Error {
constructor(readonly transition: TurnTransition) {
super()
}
}
const continueAfterCompaction = (step: number) => new TurnTransitionError({ _tag: "ContinueAfterCompaction", step })
const continueAfterOverflowCompaction = (step: number) =>
new TurnTransitionError({ _tag: "ContinueAfterOverflowCompaction", step })
const loadSystemContext = (agent: AgentV2.Selection) =>
Effect.all([systemContext.load(), skillGuidance.load(agent), referenceGuidance.load(), mcpGuidance.load(agent)], {
concurrency: "unbounded",
@ -218,8 +202,9 @@ const layer = Layer.effect(
tools: toolMaterialization?.definitions ?? [],
toolChoice: isLastStep ? "none" : undefined,
})
// Automatic compaction completed; rebuild the request from compacted history.
if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request }))
return yield* Effect.die(continueAfterCompaction(currentStep))
return { _tag: "RestartAfterCompaction", step: currentStep } as const
const startSnapshot = yield* snapshots.capture()
const publisher = createLLMEventPublisher(events, {
sessionID: session.id,
@ -284,44 +269,71 @@ const layer = Layer.effect(
return yield* Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
// Gather the evidence: how did the provider stream end?
const stream = yield* restore(providerStream).pipe(Effect.exit)
const failure =
stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined
const streamFailure = Option.getOrUndefined(Exit.findErrorOption(stream))
// Note: Exit.hasInterrupts is a type guard whose false branch unsoundly narrows
// away non-interrupt failures, so both interrupt checks stay Cause-based.
const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)
// A context overflow before any assistant output is recoverable: compact and
// restart the turn instead of surfacing the provider error.
if (
recoverOverflow &&
!publisher.hasAssistantStarted() &&
isContextOverflowFailure(overflowFailure ?? failure) &&
isContextOverflowFailure(overflowFailure ?? streamFailure) &&
(yield* restore(recoverOverflow({ sessionID: session.id, messages: context, request })))
)
return yield* Effect.die(continueAfterOverflowCompaction(currentStep))
return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
// An unrecovered held-back overflow becomes the turn's durable provider error. A
// thrown LLM failure fails hosted tool calls and the assistant unless a provider
// error was already recorded from the stream.
if (overflowFailure) yield* publish(overflowFailure)
const llmFailure = failure instanceof LLMError ? failure : undefined
const llmFailure = streamFailure instanceof LLMError ? streamFailure : undefined
if (llmFailure && !publisher.hasProviderError()) {
yield* publication.withPermit(publisher.failUnsettledTools("Provider did not return a tool result", true))
yield* publication.withPermit(publisher.failAssistant(llmFailure.reason.message))
}
if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers)
// Provider error events only arrive from the stream, so the flag is final here.
const providerFailed = publisher.hasProviderError()
// Settle tool fibers: an interrupted stream abandons unstarted tool work first.
if (streamInterrupted) yield* FiberSet.clear(toolFibers)
const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit)
const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)
const toolsInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)
if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) {
const questionDismissed = settled._tag === "Failure" && isQuestionRejected(settled.cause)
if (questionDismissed || streamInterrupted || toolsInterrupted) {
yield* FiberSet.clear(toolFibers)
yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
yield* publication.withPermit(publisher.failAssistant("Provider turn interrupted"))
return yield* Effect.interrupt
// Match V1: dismissing a question halts the loop like an interruption.
if (questionDismissed) return yield* Effect.interrupt
}
if (streamInterrupted || toolsInterrupted) {
yield* FiberSet.clear(toolFibers)
yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
yield* publication.withPermit(publisher.failAssistant("Provider turn interrupted"))
}
if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) {
const failure = Cause.squash(settled.cause)
// A settled tool fiber failure is one of two things. A defect from a tool
// implementation becomes a failed tool call the model can read, and the turn still
// settles so the model may recover. A typed infrastructure failure (tool output
// could not be persisted) also fails the assistant and then fails the drain.
const settledFailure = settled._tag === "Failure" && !toolsInterrupted ? settled.cause : undefined
const infraError =
settledFailure === undefined ? undefined : Option.getOrUndefined(Cause.findErrorOption(settledFailure))
if (settledFailure !== undefined) {
const failure = infraError ?? Cause.squash(settledFailure)
const message = failure instanceof Error ? failure.message : String(failure)
yield* publication.withPermit(publisher.failUnsettledTools(`Tool execution failed: ${message}`))
if (infraError !== undefined)
yield* publication.withPermit(publisher.failAssistant(`Tool execution failed: ${message}`))
}
const stepSettlement = publisher.stepSettlement()
if (stepSettlement && !streamInterrupted && !toolsInterrupted && !publisher.hasProviderError()) {
if (
stepSettlement &&
!streamInterrupted &&
!toolsInterrupted &&
infraError === undefined &&
!providerFailed
) {
const endSnapshot = yield* snapshots.capture()
const files =
startSnapshot && endSnapshot
@ -342,49 +354,43 @@ const layer = Layer.effect(
}),
)
}
if (publisher.hasProviderError())
yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
if (stream._tag === "Success" && !publisher.hasProviderError())
// A provider error orphans recorded local calls; a clean stream can still leave
// hosted calls without results.
if (providerFailed) yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted"))
if (stream._tag === "Success" && !providerFailed)
yield* publication.withPermit(publisher.failUnsettledTools("Provider did not return a tool result", true))
if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause)
if (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause))
if (settled._tag === "Failure" && (toolsInterrupted || infraError !== undefined))
return yield* Effect.failCause(settled.cause)
return { needsContinuation: !publisher.hasProviderError() && needsContinuation, step: currentStep }
return {
_tag: "Completed",
needsContinuation: !providerFailed && needsContinuation,
step: currentStep,
} as const
}),
)
}, Effect.scoped)
type RunTurn = (
const runTurn = Effect.fnUntraced(function* (
sessionID: SessionSchema.ID,
promotion: SessionInput.Delivery | undefined,
step: number,
) => Effect.Effect<{ readonly needsContinuation: boolean; readonly step: number }, RunError>
const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) {
return yield* runTurnAttempt(sessionID, promotion, step).pipe(
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, undefined, defect.transition.step)
}),
),
)
})
const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) {
return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow).pipe(
Effect.catchDefect(
Effect.fnUntraced(function* (defect) {
if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect)
yield* Effect.yieldNow
if (defect.transition._tag === "ContinueAfterOverflowCompaction")
return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step)
return yield* runTurn(sessionID, undefined, defect.transition.step)
}),
),
)
) {
// Compaction restarts rebuild the request from compacted history without re-promoting.
// Overflow recovery is one-shot: a post-compaction attempt must not recover another
// overflow, so the recovery hook is dropped after it fires.
let recoverOverflow: typeof compaction.compactAfterOverflow | undefined = compaction.compactAfterOverflow
let currentPromotion = promotion
let currentStep = step
while (true) {
const attempt = yield* runTurnAttempt(sessionID, currentPromotion, currentStep, recoverOverflow)
if (attempt._tag === "Completed") return { needsContinuation: attempt.needsContinuation, step: attempt.step }
if (attempt._tag === "RestartAfterOverflowCompaction") recoverOverflow = undefined
yield* Effect.yieldNow
currentPromotion = undefined
currentStep = attempt.step
}
})
const drain = Effect.fnUntraced(function* (input: {
@ -425,18 +431,15 @@ const layer = Layer.effect(
Effect.gen(function* () {
const failure =
Exit.isFailure(exit) && !Cause.hasInterrupts(exit.cause) ? Cause.squash(exit.cause) : undefined
yield* events.publish(
SessionEvent.ExecutionSettled,
{
sessionID: input.sessionID,
timestamp: yield* DateTime.now,
outcome: Exit.isSuccess(exit) ? "success" : Cause.hasInterrupts(exit.cause) ? "interrupted" : "failure",
error:
failure !== undefined
? { type: "unknown", message: failure instanceof Error ? failure.message : String(failure) }
: undefined,
},
)
yield* events.publish(SessionEvent.ExecutionSettled, {
sessionID: input.sessionID,
timestamp: yield* DateTime.now,
outcome: Exit.isSuccess(exit) ? "success" : Cause.hasInterrupts(exit.cause) ? "interrupted" : "failure",
error:
failure !== undefined
? { type: "unknown", message: failure instanceof Error ? failure.message : String(failure) }
: undefined,
})
}).pipe(
Effect.catchCause(() => Effect.void),
Effect.asVoid,

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

@ -154,6 +154,14 @@ const echo = Layer.effectDiscard(
output: Schema.Struct({}),
execute: () => Effect.die("unexpected tool defect"),
}),
// BigInt output with no model content forces ToolOutputStore.bound onto its
// JSON.stringify encode path, which fails with a typed StorageError.
storefail: Tool.make({
description: "Produce output that cannot be persisted",
input: Schema.Struct({}),
output: Schema.Any,
execute: () => Effect.succeed({ big: 1n }),
}),
}),
),
)
@ -671,7 +679,7 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(1)
expect(requests[0]?.model).toBe(model)
expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect"])
expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect", "storefail"])
expect(requests[0]?.messages.map((message) => ({ role: message.role, content: message.content }))).toEqual([
{ role: "user", content: [{ type: "text", text: "First" }] },
{ role: "user", content: [{ type: "text", text: "Second" }] },
@ -1456,7 +1464,7 @@ describe("SessionRunnerLLM", () => {
yield* session.resume(sessionID)
expect(requests).toHaveLength(1)
expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect"])
expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect", "storefail"])
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "user", text: "Use tools" },
{
@ -2700,6 +2708,49 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("fails the drain when tool output persistence fails", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Call storefail" }), resume: false })
requests.length = 0
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-storefail", name: "storefail", input: {} }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[],
]
const exit = yield* session.resume(sessionID).pipe(Effect.exit)
expect(Exit.isFailure(exit)).toBe(true)
expect(requests).toHaveLength(1)
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "user", text: "Call storefail" },
{
type: "assistant",
content: [
{
type: "tool",
id: "call-storefail",
state: {
status: "error",
error: {
type: "unknown",
message: expect.stringContaining("Tool execution failed: Failed to encode tool output"),
},
},
},
],
},
])
}),
)
it.effect("interrupts runner continuation when a question is dismissed", () =>
Effect.gen(function* () {
yield* setup