From 656a2a00b486cd13a5b1e843d3a28579d4ca318a Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Fri, 11 Sep 2026 08:51:06 -0300 Subject: [PATCH] fix(session): discard a failed attempt's parts before retrying and stop retrying past an executed tool `Effect.retry(SessionRetry.policy)` wraps the whole stream consumption, but the parts a failed attempt had already persisted (text, reasoning, step-start, tool parts) were left in the assistant message, so the retried stream appended duplicates next to them. Worse, the retry replayed the same request, so a tool the model had already executed in the failed attempt (a gateway closing with `finish_reason: network_error` after a completed edit) was requested and run a second time. The processor now remembers the part ids it mints during an attempt and, as soon as the retry policy decides to retry, removes them through `Session.removePart` (projected to a DB delete) and settles the pending tool calls, before the wait starts. A failure after any tool call reached running skips the retry layer and goes straight to `halt`, so the turn ends with the error visible and no tool executes twice. Two regression tests cover this: a midstream server_error after a text chunk leaves exactly one text part and one step-start after the retry; a completed tool call followed by `network_error` makes no second provider request, keeps the tool at completed, and ends the message with an error. Claude-Session: https://claude.ai/code/session_01KAcoL6wpgVEs2ebUrUGpHS --- .changeset/retry-attempt-cleanup.md | 12 ++ packages/redcode/src/session/processor.ts | 80 +++++++++--- .../test/session/processor-effect.test.ts | 121 ++++++++++++++++++ 3 files changed, 197 insertions(+), 16 deletions(-) create mode 100644 .changeset/retry-attempt-cleanup.md 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 }) =>