mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-08 18:30:00 -04:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| fe64ef154c | |||
| f57e0a2134 | |||
| d89a0d0bf7 |
@@ -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
|
||||||
}),
|
}),
|
||||||
|
|||||||
+22
-2
@@ -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: {
|
||||||
|
|||||||
Reference in New Issue
Block a user