fix(session): discard a failed attempt's parts before retrying and stop retrying past an executed tool

.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.

packages/redcode/src/session/processor.ts

@@ -105,6 +105,16 @@ interface ProcessorContext extends Input {
   reasoningMap: Record<string, SessionV1.ReasoningPart>
   /** 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),

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 }) =>