fix(core): safely recover malformed tool input (#37698)
This commit is contained in:
parent
584fdefe6f
commit
57ff57595a
22 changed files with 876 additions and 93 deletions
|
|
@ -30,6 +30,7 @@ import {
|
|||
type UsageInput,
|
||||
} from "@opencode-ai/ai"
|
||||
import { Auth, Endpoint, type AnyRoute } from "@opencode-ai/ai/route"
|
||||
import { ProviderShared } from "@opencode-ai/ai/protocols/shared"
|
||||
import { Cause, Context, Effect, Layer, Option, Schema, Scope, Stream } from "effect"
|
||||
import { ModelV2 } from "./model"
|
||||
import { ProviderV2 } from "./provider"
|
||||
|
|
@ -605,6 +606,7 @@ function streamPartEvents(
|
|||
LLMEvent.toolInputStart({
|
||||
id: event.id,
|
||||
name: event.toolName,
|
||||
providerExecuted: event.providerExecuted,
|
||||
providerMetadata: providerMetadata(event.providerMetadata),
|
||||
}),
|
||||
])
|
||||
|
|
@ -622,15 +624,30 @@ function streamPartEvents(
|
|||
])
|
||||
case "tool-call":
|
||||
state.toolNames[event.toolCallId] = event.toolName
|
||||
return Effect.succeed([
|
||||
LLMEvent.toolCall({
|
||||
id: event.toolCallId,
|
||||
name: event.toolName,
|
||||
input: parseToolInput(event.input),
|
||||
providerExecuted: event.providerExecuted,
|
||||
providerMetadata: providerMetadata(event.providerMetadata),
|
||||
}),
|
||||
])
|
||||
return ProviderShared.parseToolInput("aisdk", event.toolName, event.input).pipe(
|
||||
Effect.map((input) => [
|
||||
LLMEvent.toolCall({
|
||||
id: event.toolCallId,
|
||||
name: event.toolName,
|
||||
input,
|
||||
providerExecuted: event.providerExecuted,
|
||||
providerMetadata: providerMetadata(event.providerMetadata),
|
||||
}),
|
||||
]),
|
||||
Effect.catch((error) =>
|
||||
event.providerExecuted
|
||||
? Effect.fail(error)
|
||||
: Effect.succeed([
|
||||
LLMEvent.toolInputError({
|
||||
id: event.toolCallId,
|
||||
name: event.toolName,
|
||||
raw: event.input,
|
||||
message: error.reason.message,
|
||||
providerMetadata: providerMetadata(event.providerMetadata),
|
||||
}),
|
||||
]),
|
||||
),
|
||||
)
|
||||
case "tool-result":
|
||||
delete state.toolNames[event.toolCallId]
|
||||
return Effect.succeed([
|
||||
|
|
@ -685,14 +702,6 @@ function providerMetadata(value: unknown) {
|
|||
return Schema.is(ProviderMetadata)(value) ? value : undefined
|
||||
}
|
||||
|
||||
function parseToolInput(value: string) {
|
||||
try {
|
||||
return JSON.parse(value) as unknown
|
||||
} catch {
|
||||
return value
|
||||
}
|
||||
}
|
||||
|
||||
function jsonObject(input: Record<string, unknown>) {
|
||||
return Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsonValue(value)]))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -297,6 +297,12 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
|
|||
draft.cost = event.data.cost
|
||||
draft.tokens = castDraft(event.data.tokens)
|
||||
}
|
||||
if (event.data.snapshot || event.data.files)
|
||||
draft.snapshot = {
|
||||
...draft.snapshot,
|
||||
end: event.data.snapshot,
|
||||
files: event.data.files ? Array.from(event.data.files) : undefined,
|
||||
}
|
||||
})
|
||||
},
|
||||
"session.text.started": (event) => {
|
||||
|
|
|
|||
|
|
@ -80,6 +80,7 @@ const layer = Layer.effect(
|
|||
sessionID: SessionSchema.ID,
|
||||
promotion: SessionPending.Delivery | undefined,
|
||||
step: number,
|
||||
recoverMalformedToolInput: boolean,
|
||||
recoverOverflow?: typeof compaction.compact,
|
||||
assistantMessageID?: SessionMessage.ID,
|
||||
) {
|
||||
|
|
@ -195,25 +196,27 @@ const layer = Layer.effect(
|
|||
tokens: settlement.tokens,
|
||||
})
|
||||
|
||||
// Captures the end snapshot, diffs it against the step's start, and durably ends the
|
||||
// assistant step.
|
||||
const captureStepEnd = Effect.fnUntraced(function* () {
|
||||
const snapshot = yield* snapshots.capture()
|
||||
const files =
|
||||
startSnapshot && snapshot
|
||||
? yield* snapshots
|
||||
.files({ from: startSnapshot, to: snapshot })
|
||||
.pipe(Effect.catch(() => Effect.succeed(undefined)))
|
||||
: undefined
|
||||
return { snapshot, files }
|
||||
})
|
||||
|
||||
const publishStepEnd = (settlement: NonNullable<ReturnType<typeof publisher.stepSettlement>>) =>
|
||||
Effect.gen(function* () {
|
||||
const endSnapshot = yield* snapshots.capture()
|
||||
const files =
|
||||
startSnapshot && endSnapshot
|
||||
? yield* snapshots
|
||||
.files({ from: startSnapshot, to: endSnapshot })
|
||||
.pipe(Effect.catch(() => Effect.succeed(undefined)))
|
||||
: undefined
|
||||
const end = yield* captureStepEnd()
|
||||
yield* serialized(
|
||||
events.publish(SessionEvent.Step.Ended, {
|
||||
sessionID: session.id,
|
||||
assistantMessageID: yield* publisher.startAssistant(),
|
||||
finish: settlement.finish,
|
||||
...stepUsage(settlement),
|
||||
snapshot: endSnapshot,
|
||||
files,
|
||||
...end,
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
|
@ -253,7 +256,7 @@ const layer = Layer.effect(
|
|||
step: currentStep,
|
||||
})
|
||||
}
|
||||
yield* serialized(publisher.failAssistant(error))
|
||||
yield* serialized(publisher.failAssistant(error, true))
|
||||
}
|
||||
// Provider error events only arrive from the stream, so the flag is final here.
|
||||
const providerFailed = publisher.hasProviderError()
|
||||
|
|
@ -274,7 +277,7 @@ const layer = Layer.effect(
|
|||
if (settled._tag === "Failure") yield* FiberSet.clear(toolFibers)
|
||||
if (userDeclined || streamInterrupted || toolsInterrupted) {
|
||||
yield* serialized(publisher.failUnsettledTools({ type: "aborted", message: "Tool execution interrupted" }))
|
||||
yield* serialized(publisher.failAssistant({ type: "aborted", message: "Step interrupted" }))
|
||||
yield* serialized(publisher.failAssistant({ type: "aborted", message: "Step interrupted" }, true))
|
||||
}
|
||||
// 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
|
||||
|
|
@ -287,7 +290,7 @@ const layer = Layer.effect(
|
|||
const failure = infraError ?? Cause.squash(settledFailure)
|
||||
const error = toSessionError(failure)
|
||||
yield* serialized(publisher.failUnsettledTools(error))
|
||||
if (infraError !== undefined) yield* serialized(publisher.failAssistant(error))
|
||||
if (infraError !== undefined) yield* serialized(publisher.failAssistant(error, true))
|
||||
}
|
||||
|
||||
// Fail unresolved calls before the terminal step event. Local calls have joined, so
|
||||
|
|
@ -315,27 +318,49 @@ const layer = Layer.effect(
|
|||
: false
|
||||
if (hostedResultMissing && !publisher.stepSettlement())
|
||||
yield* serialized(
|
||||
publisher.failAssistant({
|
||||
type: "tool.result-missing",
|
||||
message: "Provider did not return a tool result",
|
||||
}),
|
||||
publisher.failAssistant(
|
||||
{
|
||||
type: "tool.result-missing",
|
||||
message: "Provider did not return a tool result",
|
||||
},
|
||||
true,
|
||||
),
|
||||
)
|
||||
|
||||
const stepFailure = publisher.stepFailure()
|
||||
const stepSettlement = publisher.stepSettlement()
|
||||
if (stepSettlement && !stepFailure) yield* publishStepEnd(stepSettlement)
|
||||
if (stepFailure)
|
||||
yield* serialized(publisher.publishStepFailure(stepSettlement ? stepUsage(stepSettlement) : undefined))
|
||||
if (stepFailure) {
|
||||
const end = yield* captureStepEnd()
|
||||
yield* serialized(
|
||||
publisher.publishStepFailure({
|
||||
...(stepSettlement ? stepUsage(stepSettlement) : {}),
|
||||
...end,
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
const recoveredMalformedToolInput =
|
||||
recoverMalformedToolInput &&
|
||||
publisher.hasMalformedToolInput() &&
|
||||
stream._tag === "Success" &&
|
||||
stepSettlement !== undefined &&
|
||||
!providerFailed &&
|
||||
!streamInterrupted &&
|
||||
!userDeclined &&
|
||||
!toolsInterrupted &&
|
||||
infraError === undefined
|
||||
|
||||
if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause)
|
||||
if (userDeclined) return yield* Effect.interrupt
|
||||
if ((toolsInterrupted || infraError !== undefined) && settledFailure)
|
||||
return yield* Effect.failCause(settledFailure)
|
||||
if (toolsInterrupted && settled._tag === "Failure") return yield* Effect.failCause(settled.cause)
|
||||
if (stepFailure) return yield* new StepFailedError({ error: stepFailure })
|
||||
if (stepFailure && !recoveredMalformedToolInput) return yield* new StepFailedError({ error: stepFailure })
|
||||
return {
|
||||
_tag: "Completed",
|
||||
needsContinuation,
|
||||
needsContinuation: needsContinuation || recoveredMalformedToolInput,
|
||||
malformedToolInput: recoveredMalformedToolInput,
|
||||
step: currentStep,
|
||||
} as const
|
||||
}),
|
||||
|
|
@ -346,6 +371,7 @@ const layer = Layer.effect(
|
|||
sessionID: SessionSchema.ID,
|
||||
promotion: SessionPending.Delivery | undefined,
|
||||
step: number,
|
||||
recoverMalformedToolInput: boolean,
|
||||
) {
|
||||
// Compaction restarts rebuild the request from compacted history without re-promoting.
|
||||
// Overflow recovery is one-shot: a post-compaction attempt must not recover another
|
||||
|
|
@ -356,7 +382,14 @@ const layer = Layer.effect(
|
|||
let assistantMessageID: SessionMessage.ID | undefined
|
||||
while (true) {
|
||||
const attempt = yield* Effect.suspend(() =>
|
||||
attemptStep(sessionID, currentPromotion, currentStep, recoverOverflow, assistantMessageID),
|
||||
attemptStep(
|
||||
sessionID,
|
||||
currentPromotion,
|
||||
currentStep,
|
||||
recoverMalformedToolInput,
|
||||
recoverOverflow,
|
||||
assistantMessageID,
|
||||
),
|
||||
).pipe(
|
||||
Effect.tapError((error) =>
|
||||
error instanceof SessionRunnerRetry.RetryableFailure
|
||||
|
|
@ -378,7 +411,12 @@ const layer = Layer.effect(
|
|||
.pipe(Effect.andThen(Effect.fail(error.cause)))
|
||||
}),
|
||||
)
|
||||
if (attempt._tag === "Completed") return { needsContinuation: attempt.needsContinuation, step: attempt.step }
|
||||
if (attempt._tag === "Completed")
|
||||
return {
|
||||
needsContinuation: attempt.needsContinuation,
|
||||
malformedToolInput: attempt.malformedToolInput,
|
||||
step: attempt.step,
|
||||
}
|
||||
if (attempt._tag === "RestartAfterOverflowCompaction") recoverOverflow = undefined
|
||||
yield* Effect.yieldNow
|
||||
currentPromotion = undefined
|
||||
|
|
@ -436,12 +474,18 @@ const layer = Layer.effect(
|
|||
while (shouldRun) {
|
||||
let needsContinuation = true
|
||||
let step = 1
|
||||
let canRecoverMalformedToolInput = true
|
||||
// Repeat steps while continuation is needed. A step needs continuation only
|
||||
// when it recorded local tool calls whose results the model has not yet seen;
|
||||
// a provider error suppresses it. Pending steers also continue the loop so
|
||||
// interjections are answered before the session goes idle.
|
||||
while (needsContinuation) {
|
||||
const result = yield* runStep(input.sessionID, promotion, step)
|
||||
const result = yield* runStep(
|
||||
input.sessionID,
|
||||
promotion,
|
||||
step,
|
||||
canRecoverMalformedToolInput,
|
||||
)
|
||||
// Steer/queue promotion inside runStep has already made the pending input a visible
|
||||
// user message by this point, so the first-user-message check below is reliable.
|
||||
if (!titleAttempted.has(input.sessionID)) {
|
||||
|
|
@ -449,6 +493,7 @@ const layer = Layer.effect(
|
|||
forkTitle(title.generateForFirstPrompt(yield* getSession(input.sessionID)).pipe(Effect.ignore))
|
||||
}
|
||||
needsContinuation = result.needsContinuation
|
||||
if (result.malformedToolInput) canRecoverMalformedToolInput = false
|
||||
step = result.step + 1
|
||||
if (needsContinuation) {
|
||||
yield* runPendingCompaction(input.sessionID)
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import { SessionError } from "@opencode-ai/schema/session-error"
|
|||
import { Money } from "@opencode-ai/schema/money"
|
||||
import { AgentV2 } from "../../agent"
|
||||
import { Snapshot } from "../../snapshot"
|
||||
import { RelativePath } from "../../schema"
|
||||
import { SessionUsage } from "../usage"
|
||||
|
||||
type Input = {
|
||||
|
|
@ -60,6 +61,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
let stepFailed = false
|
||||
let providerFailed = false
|
||||
let retryEvidence = false
|
||||
let malformedToolInput = false
|
||||
let stepFailure: SessionError.Error | undefined
|
||||
let stepSettlement:
|
||||
| {
|
||||
|
|
@ -112,12 +114,12 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
if (state !== undefined) current.state = { ...current.state, ...state }
|
||||
return Effect.succeed(current.ordinal)
|
||||
})
|
||||
const end = Effect.fnUntraced(function* (id: string, state?: Record<string, unknown>) {
|
||||
const end = Effect.fnUntraced(function* (id: string, state?: Record<string, unknown>, value?: string) {
|
||||
const current = chunks.get(id)
|
||||
if (!current) return yield* Effect.die(new Error(`${name} end before start: ${id}`))
|
||||
yield* ended(
|
||||
id,
|
||||
current.values.join(""),
|
||||
value ?? current.values.join(""),
|
||||
current.ordinal,
|
||||
state === undefined ? current.state : { ...current.state, ...state },
|
||||
)
|
||||
|
|
@ -175,7 +177,11 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
yield* toolInput.flush()
|
||||
})
|
||||
|
||||
const startToolInput = Effect.fnUntraced(function* (event: { readonly id: string; readonly name: string }) {
|
||||
const startToolInput = Effect.fnUntraced(function* (event: {
|
||||
readonly id: string
|
||||
readonly name: string
|
||||
readonly providerExecuted?: boolean
|
||||
}) {
|
||||
if (tools.has(event.id)) return yield* Effect.die(new Error(`Duplicate tool input start: ${event.id}`))
|
||||
const assistantMessageID = yield* startAssistant()
|
||||
tools.set(event.id, {
|
||||
|
|
@ -183,7 +189,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
name: event.name,
|
||||
called: false,
|
||||
settled: false,
|
||||
providerExecuted: false,
|
||||
providerExecuted: event.providerExecuted === true,
|
||||
})
|
||||
yield* toolInput.start(event.id)
|
||||
yield* events.publish(SessionEvent.Tool.Input.Started, {
|
||||
|
|
@ -194,13 +200,44 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
})
|
||||
})
|
||||
|
||||
const endToolInput = Effect.fnUntraced(function* (event: { readonly id: string; readonly name: string }) {
|
||||
const endToolInput = Effect.fnUntraced(function* (
|
||||
event: { readonly id: string; readonly name: string },
|
||||
value?: string,
|
||||
) {
|
||||
const tool = tools.get(event.id)
|
||||
if (!tool) return yield* Effect.die(new Error(`Tool input end before start: ${event.id}`))
|
||||
if (tool.name !== event.name)
|
||||
return yield* Effect.die(new Error(`Tool input name changed for ${event.id}: ${tool.name} -> ${event.name}`))
|
||||
if (!toolInput.has(event.id)) return yield* Effect.die(new Error(`Duplicate tool input end: ${event.id}`))
|
||||
yield* toolInput.end(event.id)
|
||||
yield* toolInput.end(event.id, undefined, value)
|
||||
})
|
||||
|
||||
const failMalformedToolInput = Effect.fnUntraced(function* (event: {
|
||||
readonly id: string
|
||||
readonly name: string
|
||||
readonly raw: string
|
||||
readonly message: string
|
||||
}) {
|
||||
if (!tools.has(event.id)) yield* startToolInput(event)
|
||||
const tool = tools.get(event.id)
|
||||
if (!tool || tool.called || tool.settled)
|
||||
return yield* Effect.die(new Error(`Malformed tool input after call settlement: ${event.id}`))
|
||||
if (tool.name !== event.name)
|
||||
return yield* Effect.die(new Error(`Tool input name changed for ${event.id}: ${tool.name} -> ${event.name}`))
|
||||
if (toolInput.has(event.id)) yield* endToolInput(event, event.raw)
|
||||
tool.settled = true
|
||||
malformedToolInput = true
|
||||
yield* events.publish(SessionEvent.Tool.Failed, {
|
||||
sessionID: input.sessionID,
|
||||
assistantMessageID: tool.assistantMessageID,
|
||||
callID: event.id,
|
||||
error: {
|
||||
type: "tool.input-json",
|
||||
message: "Tool call arguments were malformed JSON and were not executed. Retry with valid JSON.",
|
||||
},
|
||||
executed: false,
|
||||
})
|
||||
if (stepFailure === undefined) stepFailure = { type: "provider.invalid-output", message: event.message }
|
||||
})
|
||||
|
||||
const flush = Effect.fn("SessionRunner.flush")(function* () {
|
||||
|
|
@ -236,9 +273,11 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
if (replace || stepFailure === undefined) stepFailure = error
|
||||
})
|
||||
|
||||
const publishStepFailure = Effect.fnUntraced(function* (usage?: {
|
||||
readonly cost: Money.USD
|
||||
readonly tokens: ReturnType<typeof SessionUsage.tokens>
|
||||
const publishStepFailure = Effect.fnUntraced(function* (details?: {
|
||||
readonly cost?: Money.USD
|
||||
readonly tokens?: ReturnType<typeof SessionUsage.tokens>
|
||||
readonly snapshot?: Snapshot.ID
|
||||
readonly files?: readonly RelativePath[]
|
||||
}) {
|
||||
if (stepFailed || stepFailure === undefined) return
|
||||
const assistantMessageID = yield* startAssistant()
|
||||
|
|
@ -247,7 +286,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
sessionID: input.sessionID,
|
||||
assistantMessageID,
|
||||
error: stepFailure,
|
||||
...usage,
|
||||
...details,
|
||||
})
|
||||
})
|
||||
|
||||
|
|
@ -337,6 +376,10 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
case "tool-input-end":
|
||||
yield* endToolInput(event)
|
||||
return
|
||||
case "tool-input-error":
|
||||
retryEvidence = true
|
||||
yield* failMalformedToolInput(event)
|
||||
return
|
||||
case "tool-call": {
|
||||
retryEvidence = true
|
||||
if (!tools.has(event.id)) yield* startToolInput(event)
|
||||
|
|
@ -439,6 +482,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
|
|||
publishStepFailure,
|
||||
failUnsettledTools,
|
||||
hasProviderError: () => providerFailed,
|
||||
hasMalformedToolInput: () => malformedToolInput,
|
||||
hasRetryEvidence: () => retryEvidence,
|
||||
stepFailure: () => stepFailure,
|
||||
stepSettlement: () => stepSettlement,
|
||||
|
|
|
|||
|
|
@ -136,14 +136,20 @@ const assistant = (message: SessionMessage.Assistant, model: ModelV2.Ref, provid
|
|||
: item.text.length > 0
|
||||
? [{ type: "text", text: item.text }]
|
||||
: []
|
||||
const reuseToolProviderMetadata =
|
||||
sameModel &&
|
||||
(message.error === undefined ||
|
||||
(item.executed === true &&
|
||||
(item.state.status === "completed" ||
|
||||
(item.state.status === "error" && item.state.result !== undefined))))
|
||||
const call = toolCall(
|
||||
item,
|
||||
reuseProviderMetadata ? providerMetadata(providerMetadataKey, item.providerState) : undefined,
|
||||
reuseToolProviderMetadata ? providerMetadata(providerMetadataKey, item.providerState) : undefined,
|
||||
)
|
||||
if (item.executed !== true) return [call]
|
||||
const result = toolResult(
|
||||
item,
|
||||
reuseProviderMetadata
|
||||
reuseToolProviderMetadata
|
||||
? providerMetadata(providerMetadataKey, item.providerResultState ?? item.providerState)
|
||||
: undefined,
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue