mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-10 19:33:39 +00:00
fix(core): continue interrupted responses (#40576)
This commit is contained in:
Aiden Cline
GitHub
parent
0cce215de4
commit
a38d44265b
2 changed files with 213 additions and 6 deletions
|
|
@ -37,6 +37,7 @@ import { SessionUsage } from "../usage"
|
|||
type CallOutcome = Data.TaggedEnum<{
|
||||
Completed: { readonly needsContinuation: boolean; readonly step: number }
|
||||
Retry: { readonly step: number }
|
||||
Continue: { readonly cause: AIError; readonly error: SessionRunnerRetry.RetryableFailure["error"]; readonly step: number }
|
||||
Restart: { readonly step: number; readonly recoveredOverflow: boolean }
|
||||
}>
|
||||
const CallOutcome = Data.taggedEnum<CallOutcome>()
|
||||
|
|
@ -91,6 +92,8 @@ const classifyToolExits = (
|
|||
const TOOLS_INTERRUPTED = { type: "aborted", message: "Tool execution interrupted" } as const
|
||||
const STEP_INTERRUPTED = { type: "aborted", message: "Step interrupted" } as const
|
||||
const RESULT_MISSING = { type: "tool.result-missing", message: "Provider did not return a tool result" } as const
|
||||
const CONTINUE_AFTER_INCOMPLETE_STREAM =
|
||||
"The previous response was interrupted. Continue from where you left off without repeating completed content."
|
||||
|
||||
const layer = Layer.effect(
|
||||
Service,
|
||||
|
|
@ -187,6 +190,20 @@ const layer = Layer.effect(
|
|||
assistantMessageID,
|
||||
).pipe(Effect.catchTag("SessionRunner.RetryableFailure", waitForRetry))
|
||||
if (outcome._tag === "Completed") return { needsContinuation: outcome.needsContinuation, step: outcome.step }
|
||||
if (outcome._tag === "Continue") {
|
||||
yield* retry(
|
||||
new SessionRunnerRetry.RetryableFailure({
|
||||
cause: outcome.cause,
|
||||
error: outcome.error,
|
||||
step: outcome.step,
|
||||
}),
|
||||
).pipe(Pull.catchDone(() => Effect.fail(outcome.cause)))
|
||||
yield* bus.publish(SessionEvent.Synthetic, {
|
||||
sessionID,
|
||||
text: CONTINUE_AFTER_INCOMPLETE_STREAM,
|
||||
})
|
||||
assistantMessageID = SessionMessage.ID.create()
|
||||
}
|
||||
if (outcome._tag === "Restart") {
|
||||
if (outcome.recoveredOverflow) recoverOverflow = false
|
||||
assistantMessageID = SessionMessage.ID.create()
|
||||
|
|
@ -426,6 +443,17 @@ const layer = Layer.effect(
|
|||
})
|
||||
}
|
||||
|
||||
const incompleteStream =
|
||||
llmFailure?.reason._tag === "InvalidProviderOutput" &&
|
||||
llmFailure.reason.classification === "incomplete-stream"
|
||||
const toolsAllowContinuation = tools.declines.length === 0 && !tools.interrupted
|
||||
if (llmError && incompleteStream && record.outputStarted && toolsAllowContinuation)
|
||||
return CallOutcome.Continue({
|
||||
cause: llmFailure,
|
||||
error: llmError,
|
||||
step: currentStep,
|
||||
})
|
||||
|
||||
if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause)
|
||||
if (tools.declines.length > 0) return yield* Effect.interrupt
|
||||
if (tools.interrupted && tools.failure) return yield* Effect.failCause(tools.failure)
|
||||
|
|
|
|||
|
|
@ -524,6 +524,9 @@ const incompleteStream = () =>
|
|||
}),
|
||||
})
|
||||
|
||||
const INCOMPLETE_STREAM_CONTINUATION =
|
||||
"The previous response was interrupted. Continue from where you left off without repeating completed content."
|
||||
|
||||
const invalidRequest = () =>
|
||||
new AIError({
|
||||
module: "test",
|
||||
|
|
@ -4007,10 +4010,11 @@ describe("SessionRunnerLLM", () => {
|
|||
}),
|
||||
)
|
||||
|
||||
it.effect("does not retry eligible failures after observable output", () =>
|
||||
it.effect("continues an incomplete stream after observable text", () =>
|
||||
Effect.gen(function* () {
|
||||
const session = yield* setup
|
||||
const failure = incompleteStream()
|
||||
yield* admit(session, "Continue partial output")
|
||||
yield* TestLLM.push(
|
||||
TestLLM.failAfter(
|
||||
failure,
|
||||
|
|
@ -4019,19 +4023,194 @@ describe("SessionRunnerLLM", () => {
|
|||
LLMEvent.textDelta({ id: "partial-rate-limit", text: "Partial" }),
|
||||
),
|
||||
)
|
||||
yield* TestLLM.push(TestLLM.text(" continuation", "continued-text"))
|
||||
|
||||
expect(yield* runPrompt(session, "Do not replay partial output").pipe(Effect.flip)).toBe(failure)
|
||||
expect(requests).toHaveLength(1)
|
||||
expect(yield* recordedEventTypes(sessionID)).not.toContain("session.retry.scheduled.1")
|
||||
expect(yield* session.context(sessionID)).toMatchObject([
|
||||
{ type: "user" },
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(requests).toHaveLength(2)
|
||||
expect(requests[1]?.messages.at(-2)).toMatchObject({
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "Partial" }],
|
||||
})
|
||||
expect(requests[1]?.messages.at(-1)).toMatchObject({
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: INCOMPLETE_STREAM_CONTINUATION,
|
||||
},
|
||||
],
|
||||
})
|
||||
const context = yield* session.context(sessionID)
|
||||
expect(context).toMatchObject([
|
||||
{ type: "user", text: "Continue partial output" },
|
||||
{
|
||||
type: "assistant",
|
||||
finish: "error",
|
||||
error: { type: "provider.invalid-output" },
|
||||
content: [{ type: "text", text: "Partial" }],
|
||||
},
|
||||
{
|
||||
type: "synthetic",
|
||||
text: INCOMPLETE_STREAM_CONTINUATION,
|
||||
},
|
||||
{ type: "assistant", finish: "stop", content: [{ type: "text", text: " continuation" }] },
|
||||
])
|
||||
const assistants = context.filter((message) => message.type === "assistant")
|
||||
expect(new Set(assistants.map((message) => message.id)).size).toBe(2)
|
||||
expect(context.find((message) => message.type === "synthetic")?.description).toBeUndefined()
|
||||
expect(yield* recordedEventTypes(sessionID)).toContain("session.retry.scheduled.1")
|
||||
yield* replaySessionProjection(sessionID)
|
||||
expect(yield* session.context(sessionID)).toMatchObject(context)
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("lowers interrupted reasoning before continuing an incomplete stream", () =>
|
||||
Effect.gen(function* () {
|
||||
const session = yield* setup
|
||||
yield* admit(session, "Continue interrupted reasoning")
|
||||
yield* TestLLM.push(
|
||||
TestLLM.failAfter(
|
||||
incompleteStream(),
|
||||
LLMEvent.stepStart({ index: 0 }),
|
||||
LLMEvent.reasoningStart({ id: "partial-reasoning" }),
|
||||
LLMEvent.reasoningDelta({ id: "partial-reasoning", text: "Partial thought" }),
|
||||
),
|
||||
)
|
||||
yield* TestLLM.push(TestLLM.text("Recovered", "reasoning-recovery"))
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(requests[1]?.messages.at(-2)).toMatchObject({
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "Partial thought" }],
|
||||
})
|
||||
expect(requests[1]?.messages.at(-1)).toMatchObject({
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: INCOMPLETE_STREAM_CONTINUATION,
|
||||
},
|
||||
],
|
||||
})
|
||||
expect(yield* session.context(sessionID)).toMatchObject([
|
||||
{ type: "user" },
|
||||
{ type: "assistant", finish: "error", content: [{ type: "reasoning", text: "Partial thought" }] },
|
||||
{ type: "synthetic" },
|
||||
{ type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] },
|
||||
])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("continues an incomplete stream after settling a local tool", () =>
|
||||
Effect.gen(function* () {
|
||||
const session = yield* setup
|
||||
yield* admit(session, "Continue after tool")
|
||||
yield* TestLLM.push(
|
||||
TestLLM.failAfter(
|
||||
incompleteStream(),
|
||||
LLMEvent.stepStart({ index: 0 }),
|
||||
LLMEvent.toolCall({ id: "call-before-close", name: "echo", input: { text: "settled" } }),
|
||||
),
|
||||
)
|
||||
yield* TestLLM.push(TestLLM.text("Recovered", "tool-recovery"))
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(executions).toEqual(["settled"])
|
||||
expect(requests[1]?.messages.slice(-3)).toMatchObject([
|
||||
{
|
||||
role: "assistant",
|
||||
content: [{ type: "tool-call", id: "call-before-close", name: "echo", input: { text: "settled" } }],
|
||||
},
|
||||
{ role: "tool", content: [{ type: "tool-result", id: "call-before-close" }] },
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: INCOMPLETE_STREAM_CONTINUATION,
|
||||
},
|
||||
],
|
||||
},
|
||||
])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("continues an incomplete stream after settling a local tool defect", () =>
|
||||
Effect.gen(function* () {
|
||||
const session = yield* setup
|
||||
yield* admit(session, "Continue after tool defect")
|
||||
yield* TestLLM.push(
|
||||
TestLLM.failAfter(
|
||||
incompleteStream(),
|
||||
LLMEvent.stepStart({ index: 0 }),
|
||||
LLMEvent.toolCall({ id: "call-defect-before-close", name: "defect", input: {} }),
|
||||
),
|
||||
)
|
||||
yield* TestLLM.push(TestLLM.text("Recovered", "tool-defect-recovery"))
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(messageRoles(requests[1])).toEqual(["user", "assistant", "tool", "user"])
|
||||
expect(yield* session.context(sessionID)).toMatchObject([
|
||||
{ type: "user" },
|
||||
{
|
||||
type: "assistant",
|
||||
content: [
|
||||
{
|
||||
type: "tool",
|
||||
id: "call-defect-before-close",
|
||||
state: { status: "error", error: { type: "unknown", message: "unexpected tool defect" } },
|
||||
},
|
||||
],
|
||||
},
|
||||
{ type: "synthetic", text: INCOMPLETE_STREAM_CONTINUATION },
|
||||
{ type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] },
|
||||
])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("stops incomplete stream continuations after five total attempts", () =>
|
||||
Effect.gen(function* () {
|
||||
const session = yield* setup
|
||||
yield* admit(session, "Exhaust partial continuations")
|
||||
const failure = incompleteStream()
|
||||
yield* TestLLM.always(
|
||||
TestLLM.failAfter(
|
||||
failure,
|
||||
LLMEvent.stepStart({ index: 0 }),
|
||||
LLMEvent.textStart({ id: "partial-exhaustion" }),
|
||||
LLMEvent.textDelta({ id: "partial-exhaustion", text: "Partial" }),
|
||||
),
|
||||
)
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
for (const [index, delay] of [2_000, 4_000, 8_000, 16_000].entries()) {
|
||||
yield* TestClock.adjust(delay)
|
||||
yield* TestLLM.wait(index + 2)
|
||||
}
|
||||
expect(yield* Fiber.join(run).pipe(Effect.flip)).toBe(failure)
|
||||
expect(requests).toHaveLength(5)
|
||||
const context = yield* session.context(sessionID)
|
||||
expect(context.filter((message) => message.type === "assistant")).toHaveLength(5)
|
||||
expect(context.filter((message) => message.type === "synthetic")).toHaveLength(4)
|
||||
}),
|
||||
)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue