diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 15e74557c..840dd0cee 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -97,6 +97,8 @@ interface ProcessorContext extends Input { currentText: SessionV1.TextPart | undefined reasoningMap: Record outputLimitUsage: Pick | undefined + emittedText: boolean + emittedTool: boolean } type StreamEvent = LLMEvent @@ -137,14 +139,18 @@ const layer = Layer.effect( currentText: undefined, reasoningMap: {}, outputLimitUsage: undefined, + emittedText: false, + emittedTool: false, } let aborted = false const parse = (e: unknown) => - MessageV2.fromError(e, { - providerID: input.model.providerID, - aborted, - }) + SessionV1.APIError.isInstance(e) + ? e.toObject() + : MessageV2.fromError(e, { + providerID: input.model.providerID, + aborted, + }) const settleToolCall = Effect.fn("SessionProcessor.settleToolCall")(function* (toolCallID: string) { const done = ctx.toolcalls[toolCallID]?.done @@ -378,6 +384,7 @@ const layer = Layer.effect( if (ctx.assistantMessage.summary) { throw new Error(`Tool call not allowed while generating summary: ${value.name}`) } + ctx.emittedTool = true yield* ensureToolCall(value) const input = isRecord(value.input) ? value.input : { value: value.input } yield* updateToolCall(value.id, (match) => ({ @@ -567,6 +574,7 @@ const layer = Layer.effect( case "text-delta": if (!ctx.currentText) return ctx.currentText.text += value.text + ctx.emittedText ||= value.text.trim().length > 0 if (value.providerMetadata) ctx.currentText.metadata = value.providerMetadata yield* session.updatePartDelta({ sessionID: ctx.currentText.sessionID, @@ -596,6 +604,7 @@ const layer = Layer.effect( } if (value.providerMetadata) ctx.currentText.metadata = value.providerMetadata yield* session.updatePart(ctx.currentText) + ctx.emittedText ||= ctx.currentText.text.trim().length > 0 ctx.currentText = undefined return @@ -704,11 +713,14 @@ const layer = Layer.effect( }) ctx.needsCompaction = false ctx.shouldBreak = (yield* config.get()).experimental?.continue_loop_on_deny !== true + let emptyResponses = 0 return yield* Effect.gen(function* () { yield* Effect.gen(function* () { ctx.currentText = undefined ctx.reasoningMap = {} + ctx.emittedText = false + ctx.emittedTool = false yield* status.set(ctx.sessionID, { type: "busy" }) const stream = llm.stream(streamInput) @@ -717,6 +729,19 @@ const layer = Layer.effect( Stream.takeUntil(() => ctx.needsCompaction), Stream.runDrain, ) + + const empty = + !ctx.needsCompaction && + streamInput.toolChoice !== "required" && + (ctx.assistantMessage.finish === "stop" || ctx.assistantMessage.finish === "unknown") && + !ctx.emittedText && + !ctx.emittedTool + if (!empty) return + emptyResponses++ + throw new SessionV1.APIError({ + message: "Model returned reasoning without an answer or tool call", + isRetryable: emptyResponses === 1, + }) }).pipe( Effect.onInterrupt(() => Effect.gen(function* () { diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 09d108bf1..f36547950 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -280,6 +280,31 @@ const outputRetryLLM = Layer.succeed( const outputRetryEnv = LayerNode.compile(root, [...replacements, [LLM.node, outputRetryLLM]]) const itOutputRetry = testEffect(outputRetryEnv) +function reasoningOnly(text: string) { + return Stream.make( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.reasoningStart({ id: "reasoning-1" }), + LLMEvent.reasoningDelta({ id: "reasoning-1", text }), + LLMEvent.reasoningEnd({ id: "reasoning-1" }), + LLMEvent.stepFinish({ index: 0, reason: "stop" }), + LLMEvent.finish({ reason: "stop" }), + ) +} + +const semanticRetryInputs: LLM.StreamInput[] = [] +const semanticRetryStreams: Stream.Stream[] = [] +const semanticRetryLLM = Layer.succeed( + LLM.Service, + LLM.Service.of({ + stream: (input) => { + semanticRetryInputs.push(input) + return semanticRetryStreams.shift() ?? Stream.fail(new Error("missing semantic retry stream")) + }, + }), +) +const semanticRetryEnv = LayerNode.compile(root, [...replacements, [LLM.node, semanticRetryLLM]]) +const itSemanticRetry = testEffect(semanticRetryEnv) + const boot = Effect.fn("test.boot")(function* () { const processors = yield* SessionProcessor.Service const session = yield* Session.Service @@ -521,6 +546,157 @@ it.live("session.processor effect tests capture reasoning from http mock", () => ), ) +itSemanticRetry.live("session.processor effect tests retry a reasoning-only response once", () => + provideTmpdirInstance((dir) => + Effect.gen(function* () { + const { processors, session } = yield* boot() + semanticRetryInputs.length = 0 + semanticRetryStreams.length = 0 + semanticRetryStreams.push( + reasoningOnly("unfinished"), + Stream.make( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.textStart({ id: "text-1" }), + LLMEvent.textDelta({ id: "text-1", text: "done" }), + LLMEvent.textEnd({ id: "text-1" }), + LLMEvent.stepFinish({ index: 0, reason: "stop" }), + LLMEvent.finish({ reason: "stop" }), + ), + ) + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "reason") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: outputRetryModel, + }) + + const input = { + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model: outputRetryModel, + agent: agent(), + system: [], + messages: [{ role: "user", content: "reason" }], + tools: {}, + } satisfies LLM.StreamInput + + const value = yield* handle.process(input) + const parts = yield* MessageV2.parts(msg.id) + + expect(value).toBe("continue") + expect(semanticRetryInputs).toHaveLength(2) + expect(semanticRetryInputs[1]).toBe(semanticRetryInputs[0]) + expect(parts.some((part) => part.type === "reasoning" && part.text === "unfinished")).toBe(true) + expect(parts.some((part) => part.type === "text" && part.text === "done")).toBe(true) + expect(handle.message.error).toBeUndefined() + }), + ), +) + +itSemanticRetry.live("session.processor effect tests fail after two reasoning-only responses", () => + provideTmpdirInstance((dir) => + Effect.gen(function* () { + const { processors, session } = yield* boot() + semanticRetryInputs.length = 0 + semanticRetryStreams.length = 0 + semanticRetryStreams.push(reasoningOnly("one"), reasoningOnly("two")) + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "reason") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: outputRetryModel, + }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model: outputRetryModel, + agent: agent(), + system: [], + messages: [{ role: "user", content: "reason" }], + tools: {}, + }) + + expect(value).toBe("stop") + expect(semanticRetryInputs).toHaveLength(2) + expect(handle.message.error).toMatchObject({ + name: "APIError", + data: { message: "Model returned reasoning without an answer or tool call" }, + }) + }), + ), +) + +itSemanticRetry.live("session.processor effect tests do not retry unterminated nonblank text", () => + provideTmpdirInstance((dir) => + Effect.gen(function* () { + const { processors, session } = yield* boot() + semanticRetryInputs.length = 0 + semanticRetryStreams.length = 0 + semanticRetryStreams.push( + Stream.make( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.textStart({ id: "text-1" }), + LLMEvent.textDelta({ id: "text-1", text: "visible" }), + LLMEvent.stepFinish({ index: 0, reason: "stop" }), + LLMEvent.finish({ reason: "stop" }), + ), + ) + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "reason") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: outputRetryModel, + }) + + const value = yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model: outputRetryModel, + agent: agent(), + system: [], + messages: [{ role: "user", content: "reason" }], + tools: {}, + }) + const parts = yield* MessageV2.parts(msg.id) + + expect(value).toBe("continue") + expect(semanticRetryInputs).toHaveLength(1) + expect(parts.some((part) => part.type === "text" && part.text === "visible")).toBe(true) + }), + ), +) + it.live("session.processor effect tests reset reasoning state across retries", () => provideTmpdirServer( ({ dir, llm }) =>