Compare commits

...

3 Commits

Author SHA1 Message Date
Kit Langton fe64ef154c refactor(core): classify step settlement once 2026-07-03 16:52:52 -04:00
Kit Langton f57e0a2134 refactor(core): derive continuation and held overflow from step ledger 2026-07-03 16:46:17 -04:00
Kit Langton d89a0d0bf7 refactor(core): rename LLM event publisher to step ledger 2026-07-03 16:44:55 -04:00
3 changed files with 92 additions and 59 deletions
+68 -55
View File
@@ -1,15 +1,6 @@
export * as SessionRunnerLLM from "./llm" export * as SessionRunnerLLM from "./llm"
import { import { LLM, LLMClient, LLMError, LLMEvent, Message, SystemPart, isContextOverflowFailure } from "@opencode-ai/llm"
LLM,
LLMClient,
LLMError,
LLMEvent,
Message,
SystemPart,
isContextOverflowFailure,
type ProviderErrorEvent,
} from "@opencode-ai/llm"
import { Cause, DateTime, Effect, Exit, FiberSet, Layer, Option, Semaphore, Stream } from "effect" import { Cause, DateTime, Effect, Exit, FiberSet, Layer, Option, Semaphore, Stream } from "effect"
import { AgentV2 } from "../../agent" import { AgentV2 } from "../../agent"
import { Config } from "../../config" import { Config } from "../../config"
@@ -36,7 +27,7 @@ import { SessionStore } from "../store"
import { SessionTitle } from "../title" import { SessionTitle } from "../title"
import { Service } from "./index" import { Service } from "./index"
import { SessionRunnerModel } from "./model" import { SessionRunnerModel } from "./model"
import { createLLMEventPublisher } from "./publish-llm-event" import { createStepLedger } from "./step-ledger"
import { toLLMMessages } from "./to-llm-message" import { toLLMMessages } from "./to-llm-message"
import { MAX_STEPS_PROMPT } from "./max-steps" import { MAX_STEPS_PROMPT } from "./max-steps"
import { SessionRunnerSystemPrompt } from "./system-prompt" import { SessionRunnerSystemPrompt } from "./system-prompt"
@@ -185,7 +176,6 @@ const layer = Layer.effect(
session.id, session.id,
) )
const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>() const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>()
let needsContinuation = false
let currentStep = step let currentStep = step
if (promotion) { if (promotion) {
let promoted = 0 let promoted = 0
@@ -222,7 +212,7 @@ const layer = Layer.effect(
if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request })) if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request }))
return { _tag: "RestartAfterCompaction", step: currentStep } as const return { _tag: "RestartAfterCompaction", step: currentStep } as const
const startSnapshot = yield* snapshots.capture() const startSnapshot = yield* snapshots.capture()
const publisher = createLLMEventPublisher(events, { const ledger = createStepLedger(events, {
sessionID: session.id, sessionID: session.id,
agent: agent.id, agent: agent.id,
// The selected catalog identity, not model.id: route-level ids are provider API // The selected catalog identity, not model.id: route-level ids are provider API
@@ -235,26 +225,25 @@ const layer = Layer.effect(
// mid-event. // mid-event.
const serialized = <A, E, R>(effect: Effect.Effect<A, E, R>) => publication.withPermit(effect) const serialized = <A, E, R>(effect: Effect.Effect<A, E, R>) => publication.withPermit(effect)
const publish = (event: LLMEvent, outputPaths: ReadonlyArray<string> = []) => const publish = (event: LLMEvent, outputPaths: ReadonlyArray<string> = []) =>
serialized(publisher.publish(event, outputPaths)) serialized(ledger.publish(event, outputPaths))
let overflowFailure: ProviderErrorEvent | undefined
const providerStream = llm.stream(request).pipe( const providerStream = llm.stream(request).pipe(
Stream.runForEach((event) => Stream.runForEach((event) =>
Effect.gen(function* () { Effect.gen(function* () {
if (overflowFailure || publisher.hasProviderError()) return if (ledger.heldOverflow() || ledger.hasProviderError()) return
if (LLMEvent.is.providerError(event)) { if (LLMEvent.is.providerError(event)) {
if (isContextOverflowFailure(event) && !publisher.hasAssistantStarted()) { if (isContextOverflowFailure(event) && !ledger.hasAssistantStarted()) {
overflowFailure = event ledger.holdOverflow(event)
return return
} }
} }
yield* publish(event) yield* publish(event)
if (event.type !== "tool-call" || event.providerExecuted) return if (event.type !== "tool-call" || event.providerExecuted) return
if (!toolMaterialization) { if (!toolMaterialization) {
yield* serialized(publisher.failUnsettledTools("Tools are disabled after the maximum agent steps")) yield* serialized(ledger.failUnsettledTools("Tools are disabled after the maximum agent steps"))
return return
} }
needsContinuation = true ledger.admitLocalToolCall(event.id)
const assistantMessageID = yield* publisher.assistantMessageID(event.id) const assistantMessageID = yield* ledger.assistantMessageID(event.id)
yield* Effect.uninterruptibleMask((restore) => yield* Effect.uninterruptibleMask((restore) =>
restore( restore(
toolMaterialization.settle({ toolMaterialization.settle({
@@ -279,12 +268,12 @@ const layer = Layer.effect(
).pipe(FiberSet.run(toolFibers)) ).pipe(FiberSet.run(toolFibers))
}), }),
), ),
Effect.ensuring(serialized(publisher.flush())), Effect.ensuring(serialized(ledger.flush())),
) )
// Captures the end snapshot, diffs it against the step's start, and durably ends the // Captures the end snapshot, diffs it against the step's start, and durably ends the
// assistant step. // assistant step.
const publishStepEnd = (settlement: NonNullable<ReturnType<typeof publisher.stepSettlement>>) => const publishStepEnd = (settlement: NonNullable<ReturnType<typeof ledger.stepSettlement>>) =>
Effect.gen(function* () { Effect.gen(function* () {
const endSnapshot = yield* snapshots.capture() const endSnapshot = yield* snapshots.capture()
const files = const files =
@@ -296,7 +285,7 @@ const layer = Layer.effect(
yield* serialized( yield* serialized(
events.publish(SessionEvent.Step.Ended, { events.publish(SessionEvent.Step.Ended, {
sessionID: session.id, sessionID: session.id,
assistantMessageID: yield* publisher.startAssistant(), assistantMessageID: yield* ledger.startAssistant(),
finish: settlement.finish, finish: settlement.finish,
cost: 0, cost: 0,
tokens: settlement.tokens, tokens: settlement.tokens,
@@ -319,8 +308,8 @@ const layer = Layer.effect(
// restart the step instead of surfacing the provider error. // restart the step instead of surfacing the provider error.
if ( if (
recoverOverflow && recoverOverflow &&
!publisher.hasAssistantStarted() && !ledger.hasAssistantStarted() &&
isContextOverflowFailure(overflowFailure ?? streamFailure) && isContextOverflowFailure(ledger.heldOverflow() ?? streamFailure) &&
(yield* restore(recoverOverflow({ sessionID: session.id, messages: context, request }))) (yield* restore(recoverOverflow({ sessionID: session.id, messages: context, request })))
) )
return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
@@ -328,28 +317,21 @@ const layer = Layer.effect(
// An unrecovered held-back overflow becomes the step's durable provider error. A // An unrecovered held-back overflow becomes the step's durable provider error. A
// thrown LLM failure fails hosted tool calls and the assistant unless a provider // thrown LLM failure fails hosted tool calls and the assistant unless a provider
// error was already recorded from the stream. // error was already recorded from the stream.
if (overflowFailure) yield* publish(overflowFailure) const heldOverflow = ledger.heldOverflow()
if (heldOverflow) yield* publish(heldOverflow)
const llmFailure = streamFailure instanceof LLMError ? streamFailure : undefined const llmFailure = streamFailure instanceof LLMError ? streamFailure : undefined
if (llmFailure && !publisher.hasProviderError()) { if (llmFailure && !ledger.hasProviderError()) {
yield* serialized(publisher.failUnsettledTools("Provider did not return a tool result", true)) yield* serialized(ledger.failUnsettledTools("Provider did not return a tool result", true))
yield* serialized(publisher.failAssistant(llmFailure.reason.message)) yield* serialized(ledger.failAssistant(llmFailure.reason.message))
} }
// Provider error events only arrive from the stream, so the flag is final here. // Provider error events only arrive from the stream, so the flag is final here.
const providerFailed = publisher.hasProviderError() const providerFailed = ledger.hasProviderError()
// Settle tool fibers: an interrupted stream abandons unstarted tool work first. // Settle tool fibers: an interrupted stream abandons unstarted tool work first.
if (streamInterrupted) yield* FiberSet.clear(toolFibers) if (streamInterrupted) yield* FiberSet.clear(toolFibers)
const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit)
const toolsInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) const toolsInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)
const questionDismissed = settled._tag === "Failure" && isQuestionRejected(settled.cause) const questionDismissed = settled._tag === "Failure" && isQuestionRejected(settled.cause)
if (questionDismissed || streamInterrupted || toolsInterrupted) {
yield* FiberSet.clear(toolFibers)
yield* serialized(publisher.failUnsettledTools("Tool execution interrupted"))
yield* serialized(publisher.failAssistant("Step interrupted"))
// Match V1: dismissing a question halts the loop like an interruption.
if (questionDismissed) return yield* Effect.interrupt
}
// A settled tool fiber failure is one of two things. A defect from a tool // 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 step still // implementation becomes a failed tool call the model can read, and the step still
// settles so the model may recover. A typed infrastructure failure (tool output // settles so the model may recover. A typed infrastructure failure (tool output
@@ -357,30 +339,61 @@ const layer = Layer.effect(
const settledFailure = settled._tag === "Failure" && !toolsInterrupted ? settled.cause : undefined const settledFailure = settled._tag === "Failure" && !toolsInterrupted ? settled.cause : undefined
const infraError = const infraError =
settledFailure === undefined ? undefined : Option.getOrUndefined(Cause.findErrorOption(settledFailure)) settledFailure === undefined ? undefined : Option.getOrUndefined(Cause.findErrorOption(settledFailure))
if (settledFailure !== undefined) {
const failure = infraError ?? Cause.squash(settledFailure) const settlement =
const message = failure instanceof Error ? failure.message : String(failure) questionDismissed || streamInterrupted || toolsInterrupted
yield* serialized(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) ? { _tag: "Interrupted" as const, questionDismissed }
if (infraError !== undefined) : providerFailed
yield* serialized(publisher.failAssistant(`Tool execution failed: ${message}`)) ? { _tag: "ProviderFailed" as const }
: infraError !== undefined
? { _tag: "ToolInfraFailed" as const, cause: infraError }
: { _tag: "Clean" as const }
if (settlement._tag === "Interrupted") {
yield* FiberSet.clear(toolFibers)
yield* serialized(ledger.failUnsettledTools("Tool execution interrupted"))
yield* serialized(ledger.failAssistant("Step interrupted"))
// Match V1: dismissing a question halts the loop like an interruption.
if (settlement.questionDismissed) return yield* Effect.interrupt
} }
const stepSettlement = publisher.stepSettlement() if (settlement._tag === "ToolInfraFailed") {
const stepEndedCleanly = const message = settlement.cause instanceof Error ? settlement.cause.message : String(settlement.cause)
!streamInterrupted && !toolsInterrupted && infraError === undefined && !providerFailed yield* serialized(ledger.failUnsettledTools(`Tool execution failed: ${message}`))
if (stepSettlement && stepEndedCleanly) yield* publishStepEnd(stepSettlement) yield* serialized(ledger.failAssistant(`Tool execution failed: ${message}`))
// A provider error orphans recorded local calls; a clean stream can still leave }
// hosted calls without results.
if (providerFailed) yield* serialized(publisher.failUnsettledTools("Tool execution interrupted")) if (settlement._tag === "Clean") {
if (stream._tag === "Success" && !providerFailed) if (settledFailure !== undefined) {
yield* serialized(publisher.failUnsettledTools("Provider did not return a tool result", true)) const failure = Cause.squash(settledFailure)
const message = failure instanceof Error ? failure.message : String(failure)
yield* serialized(ledger.failUnsettledTools(`Tool execution failed: ${message}`))
}
const stepSettlement = ledger.stepSettlement()
if (stepSettlement) yield* publishStepEnd(stepSettlement)
}
// A provider error orphans recorded local calls. A tool fiber that failed
// alongside the provider error still settles with its own failure message first.
if (settlement._tag === "ProviderFailed") {
if (settledFailure !== undefined) {
const failure = infraError ?? Cause.squash(settledFailure)
const message = failure instanceof Error ? failure.message : String(failure)
yield* serialized(ledger.failUnsettledTools(`Tool execution failed: ${message}`))
}
yield* serialized(ledger.failUnsettledTools("Tool execution interrupted"))
}
// A clean stream can still leave hosted calls without results.
if (stream._tag === "Success" && settlement._tag !== "ProviderFailed")
yield* serialized(ledger.failUnsettledTools("Provider did not return a tool result", true))
if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause)
if (settled._tag === "Failure" && (toolsInterrupted || infraError !== undefined)) if (settled._tag === "Failure" && (toolsInterrupted || infraError !== undefined))
return yield* Effect.failCause(settled.cause) return yield* Effect.failCause(settled.cause)
return { return {
_tag: "Completed", _tag: "Completed",
needsContinuation: !providerFailed && needsContinuation, needsContinuation: !providerFailed && ledger.hasLocalToolCalls(),
step: currentStep, step: currentStep,
} as const } as const
}), }),
@@ -1,4 +1,11 @@
import { ToolOutput, type LLMEvent, type ProviderMetadata, type ToolResultValue, type Usage } from "@opencode-ai/llm" import {
ToolOutput,
type LLMEvent,
type ProviderErrorEvent,
type ProviderMetadata,
type ToolResultValue,
type Usage,
} from "@opencode-ai/llm"
import { DateTime, Effect } from "effect" import { DateTime, Effect } from "effect"
import { EventV2 } from "../../event" import { EventV2 } from "../../event"
import { ModelV2 } from "../../model" import { ModelV2 } from "../../model"
@@ -51,7 +58,7 @@ const settledOutput = (value: ToolOutput | undefined, result: ToolResultValue):
} }
/** Persist one step without executing tools or starting a continuation step. */ /** Persist one step without executing tools or starting a continuation step. */
export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) => { export const createStepLedger = (events: EventV2.Interface, input: Input) => {
const tools = new Map< const tools = new Map<
string, string,
{ {
@@ -69,7 +76,9 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
let assistantActive = false let assistantActive = false
let assistantFailed = false let assistantFailed = false
let providerFailed = false let providerFailed = false
let heldOverflowEvent: ProviderErrorEvent | undefined
let stepSettlement: { readonly finish: string; readonly tokens: ReturnType<typeof tokens> } | undefined let stepSettlement: { readonly finish: string; readonly tokens: ReturnType<typeof tokens> } | undefined
const localToolCalls = new Set<string>()
const startAssistant = Effect.fnUntraced(function* () { const startAssistant = Effect.fnUntraced(function* () {
if (assistantMessageID !== undefined) return assistantMessageID if (assistantMessageID !== undefined) return assistantMessageID
@@ -229,6 +238,11 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
return tool ? Effect.succeed(tool.assistantMessageID) : Effect.die(new Error(`Unknown tool call: ${callID}`)) return tool ? Effect.succeed(tool.assistantMessageID) : Effect.die(new Error(`Unknown tool call: ${callID}`))
} }
const admitLocalToolCall = (callID: string) => {
const tool = tools.get(callID)
if (tool?.called && !tool.providerExecuted) localToolCalls.add(callID)
}
const publish = Effect.fn("SessionRunner.publishLLMEvent")(function* ( const publish = Effect.fn("SessionRunner.publishLLMEvent")(function* (
event: LLMEvent, event: LLMEvent,
outputPaths: ReadonlyArray<string> = [], outputPaths: ReadonlyArray<string> = [],
@@ -397,8 +411,14 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
flush, flush,
failAssistant, failAssistant,
failUnsettledTools, failUnsettledTools,
admitLocalToolCall,
holdOverflow: (event: ProviderErrorEvent) => {
heldOverflowEvent = event
},
heldOverflow: () => heldOverflowEvent,
hasActiveAssistant: () => assistantActive, hasActiveAssistant: () => assistantActive,
hasAssistantStarted: () => assistantMessageID !== undefined, hasAssistantStarted: () => assistantMessageID !== undefined,
hasLocalToolCalls: () => localToolCalls.size > 0,
hasProviderError: () => providerFailed, hasProviderError: () => providerFailed,
stepSettlement: () => stepSettlement, stepSettlement: () => stepSettlement,
startAssistant, startAssistant,
@@ -7,7 +7,7 @@ import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionV2 } from "@opencode-ai/core/session" import { SessionV2 } from "@opencode-ai/core/session"
import { ModelV2 } from "@opencode-ai/core/model" import { ModelV2 } from "@opencode-ai/core/model"
import { ProviderV2 } from "@opencode-ai/core/provider" import { ProviderV2 } from "@opencode-ai/core/provider"
import { createLLMEventPublisher } from "@opencode-ai/core/session/runner/publish-llm-event" import { createStepLedger } from "@opencode-ai/core/session/runner/step-ledger"
const sessionID = SessionV2.ID.make("ses_tool_event_test") const sessionID = SessionV2.ID.make("ses_tool_event_test")
const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB" const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB"
@@ -40,7 +40,7 @@ const capture = () => {
}) })
return { return {
published, published,
publisher: createLLMEventPublisher(events, { publisher: createStepLedger(events, {
sessionID, sessionID,
agent: "build", agent: "build",
model: { model: {