diff --git a/.changeset/retry-attempt-cleanup.md b/.changeset/retry-attempt-cleanup.md new file mode 100644 index 000000000000..2ce9474c3d0d --- /dev/null +++ b/.changeset/retry-attempt-cleanup.md @@ -0,0 +1,12 @@ +--- +"@reddb-io/redcode": patch +--- + +Discard a failed provider attempt's parts before retrying and never retry past an executed tool + +When a stream failed with a retryable error, the parts it had already persisted (text, +reasoning, step-start, tool parts) stayed in the assistant message and the retried stream +appended duplicates next to them. The processor now removes what the failed attempt wrote +as soon as a retry is decided, so the message holds one copy of the answer. A failure after a +tool call already ran is no longer retried at all: replaying the request would execute the +tool a second time, so the error is surfaced as a normal terminal failure instead. diff --git a/packages/redcode/src/session/processor.ts b/packages/redcode/src/session/processor.ts index d57ecc3b8bea..01978c8dbb90 100644 --- a/packages/redcode/src/session/processor.ts +++ b/packages/redcode/src/session/processor.ts @@ -105,6 +105,16 @@ interface ProcessorContext extends Input { reasoningMap: Record /** When the provider last sent anything. A stalled turn is one where this stops moving. */ lastEventAt: number + /** + * Parts written by the provider attempt in flight. A retried stream replays from the start, so + * whatever the failed attempt persisted has to go before the next attempt appends its own. + */ + attemptParts: PartID[] + /** + * Whether a tool in this attempt reached running. A retry replays the same request, so the model + * would ask for that tool again and its side effect would happen twice. + */ + attemptExecuted: boolean } type StreamEvent = LLMEvent @@ -147,9 +157,18 @@ const layer = Layer.effect( currentText: undefined, reasoningMap: {}, lastEventAt: Date.now(), + attemptParts: [], + attemptExecuted: false, } let aborted = false + /** A fresh part id, remembered so the attempt's output can be discarded if it is retried. */ + const nextPartID = () => { + const id = PartID.ascending() + ctx.attemptParts.push(id) + return id + } + const parse = (e: unknown) => MessageV2.fromError(e, { providerID: input.model.providerID, @@ -185,6 +204,7 @@ const layer = Layer.effect( const match = yield* readToolCall(toolCallID) if (!match) return undefined const part = yield* session.updatePart(update(match.part)) + if (part.state.status !== "pending") ctx.attemptExecuted = true ctx.toolcalls[toolCallID] = { ...match.call, partID: part.id, @@ -271,7 +291,7 @@ const layer = Layer.effect( return { call: ctx.toolcalls[input.id], part } } const part = yield* session.updatePart({ - id: PartID.ascending(), + id: nextPartID(), messageID: ctx.assistantMessage.id, sessionID: ctx.assistantMessage.sessionID, type: "tool", @@ -322,7 +342,7 @@ const layer = Layer.effect( case "reasoning-start": if (value.id in ctx.reasoningMap) return ctx.reasoningMap[value.id] = { - id: PartID.ascending(), + id: nextPartID(), messageID: ctx.assistantMessage.id, sessionID: ctx.assistantMessage.sessionID, type: "reasoning", @@ -443,7 +463,7 @@ const layer = Layer.effect( case "step-start": if (!ctx.snapshot) ctx.snapshot = yield* snapshot.track() yield* session.updatePart({ - id: PartID.ascending(), + id: nextPartID(), messageID: ctx.assistantMessage.id, sessionID: ctx.sessionID, snapshot: ctx.snapshot, @@ -477,7 +497,7 @@ const layer = Layer.effect( ctx.assistantMessage.cost += usage.cost ctx.assistantMessage.tokens = usage.tokens yield* session.updatePart({ - id: PartID.ascending(), + id: nextPartID(), reason: value.reason, snapshot: completedSnapshot, messageID: ctx.assistantMessage.id, @@ -491,7 +511,7 @@ const layer = Layer.effect( const patch = yield* snapshot.patch(ctx.snapshot) if (patch.files.length) { yield* session.updatePart({ - id: PartID.ascending(), + id: nextPartID(), messageID: ctx.assistantMessage.id, sessionID: ctx.sessionID, type: "patch", @@ -518,7 +538,7 @@ const layer = Layer.effect( case "text-start": ctx.currentText = { - id: PartID.ascending(), + id: nextPartID(), messageID: ctx.assistantMessage.id, sessionID: ctx.assistantMessage.sessionID, type: "text", @@ -643,6 +663,21 @@ const layer = Layer.effect( }) }) + const discardAttempt = Effect.fn("SessionProcessor.discardAttempt")(function* () { + // Only pending tool calls can be here: one that started running makes the failure terminal. + yield* Effect.forEach(Object.keys(ctx.toolcalls), settleToolCall) + yield* Effect.forEach(ctx.attemptParts, (partID) => + session.removePart({ + sessionID: ctx.assistantMessage.sessionID, + messageID: ctx.assistantMessage.id, + partID, + }), + ) + ctx.attemptParts = [] + ctx.currentText = undefined + ctx.reasoningMap = {} + }) + const halt = Effect.fn("SessionProcessor.halt")(function* (e: unknown) { yield* Effect.logError("process", { "session.id": input.sessionID, @@ -693,6 +728,8 @@ const layer = Layer.effect( } ctx.currentText = undefined ctx.reasoningMap = {} + ctx.attemptParts = [] + ctx.attemptExecuted = false ctx.phase = undefined ctx.phaseTool = undefined yield* phase("preparing") @@ -718,21 +755,32 @@ const layer = Layer.effect( ), Effect.catchCauseIf( (cause) => !Cause.hasInterruptsOnly(cause), - (cause) => Effect.fail(Cause.squash(cause)), + (cause) => { + const error = Cause.squash(cause) + // A tool already ran in this attempt: retrying would replay the request and run it + // again, so the failure ends the turn instead and the model sees what happened. + if (ctx.attemptExecuted) return halt(error) + return Effect.fail(error) + }, ), Effect.retry( SessionRetry.policy({ provider: input.model.providerID, parse, - set: (info) => { - return status.set(ctx.sessionID, { - type: "retry", - attempt: info.attempt, - message: info.message, - action: info.action, - next: info.next, - }) - }, + // Decided to retry: the failed attempt's output goes now, before the wait, so the + // message never shows it twice. + set: (info) => + discardAttempt().pipe( + Effect.andThen( + status.set(ctx.sessionID, { + type: "retry", + attempt: info.attempt, + message: info.message, + action: info.action, + next: info.next, + }), + ), + ), }), ), Effect.catch(halt), diff --git a/packages/redcode/test/session/processor-effect.test.ts b/packages/redcode/test/session/processor-effect.test.ts index adbdaae1fcda..962a58ad2938 100644 --- a/packages/redcode/test/session/processor-effect.test.ts +++ b/packages/redcode/test/session/processor-effect.test.ts @@ -516,6 +516,127 @@ it.live("session.processor effect tests reset reasoning state across retries", ( ), ) +it.live("session.processor effect tests discard the failed attempt's parts before retrying", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + + // The first attempt gets a text chunk out before the provider fails midstream with a + // retryable error, so the message already holds a step-start and a text part. + yield* llm.push( + raw({ + chunks: [ + { id: "chatcmpl-test", object: "chat.completion.chunk", choices: [{ delta: { role: "assistant" } }] }, + { id: "chatcmpl-test", object: "chat.completion.chunk", choices: [{ delta: { content: "one" } }] }, + { error: { type: "server_error", code: "server_error", message: "xxx" } }, + ], + }), + reply().text("two").stop(), + ) + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "retry text") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + 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: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "retry text" }], + tools: {}, + }) + + const parts = yield* MessageV2.parts(msg.id) + const text = parts.filter((part): part is SessionV1.TextPart => part.type === "text") + const starts = parts.filter((part) => part.type === "step-start") + + expect(value).toBe("continue") + expect(yield* llm.calls).toBe(2) + expect(text.map((part) => part.text)).toStrictEqual(["two"]) + expect(starts).toHaveLength(1) + expect(handle.message.error).toBeUndefined() + }), + { config: (url) => providerCfg(url) }, + ), +) + +it.live("session.processor effect tests do not retry after a tool call already ran", () => + provideTmpdirServer( + ({ dir, llm }) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + let executed = 0 + + // A gateway that drops its upstream after the tool call: the edit happened, then the + // stream ends with a retryable finish reason instead of a result. + yield* llm.push(reply().tool("lookup", { query: "weather" }).finish("network_error")) + + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "tool then drop") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const mdl = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ + assistantMessage: msg, + sessionID: chat.id, + model: mdl, + }) + + 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: mdl, + agent: agent(), + system: [], + messages: [{ role: "user", content: "tool then drop" }], + tools: { + lookup: tool({ + description: "Look up information", + inputSchema: z.object({ query: z.string() }), + execute: async (input) => { + executed += 1 + return { title: "Weather lookup", output: `result:${input.query}`, metadata: {} } + }, + }), + }, + }) + + const parts = yield* MessageV2.parts(msg.id) + const calls = parts.filter((part): part is SessionV1.ToolPart => part.type === "tool") + + expect(value).toBe("stop") + expect(yield* llm.calls).toBe(1) + expect(executed).toBe(1) + expect(calls).toHaveLength(1) + expect(calls[0]?.state.status).toBe("completed") + expect(handle.message.error).toBeDefined() + }), + { config: (url) => providerCfg(url) }, + ), +) + it.live("session.processor effect tests do not retry unknown json errors", () => provideTmpdirServer( ({ dir, llm }) =>