fix(agent-core): mark resampled tool-call-markup leaks as non-replayable

packages/agent-core/src/agent-loop.ts

@@ -208,21 +208,32 @@ async function runLoop(
 				pendingMessages = [];
 			}
 
-			// Stream assistant response
+			// Stream assistant response. message_end is emitted below (not inside
+			// streamAssistantResponse) so a leaked attempt can be classified before
+			// it is ever recorded as replayable.
 			let message = await streamAssistantResponse(currentContext, config, signal, emit, streamFunction);
 
 			// Serving-side tool parsers can fail and leak the model's tool-call
 			// markup into plain text: the turn then carries no executable call
 			// and the loop would end even though the model meant to act.
 			// Resample the identical context a bounded number of times; the
 			// leaked attempt is dropped from the request context while its
-			// message events above remain for observability.
+			// message events above remain for observability. Each resampled
+			// attempt is finalized as a non-retryable error turn (never as its raw
+			// "stop" content) so transform-messages.ts's error/aborted replay skip
+			// keeps the leaked markup out of the next request and out of any later
+			// resumed session — without this, the raw `<tool_call>` text would be
+			// replayed to the model, teaching it the broken format.
 			const leakRetryLimit = config.toolCallLeakRetries ?? DEFAULT_TOOL_CALL_LEAK_RETRIES;
 			for (let attempt = 0; attempt < leakRetryLimit && isToolCallMarkupLeak(message); attempt++) {
 				if (currentContext.messages[currentContext.messages.length - 1] !== message) break;
+				await emit({ type: "message_end", message: markLeakedAttemptNonReplayable(message) });
 				currentContext.messages.pop();
 				message = await streamAssistantResponse(currentContext, config, signal, emit, streamFunction);
 			}
+			// Final attempt for this turn (resampled success, or the last leaked
+			// attempt once the retry budget is exhausted) is committed as-is.
+			await emit({ type: "message_end", message });
 			newMessages.push(message);
 
 			if (message.stopReason === "error" || message.stopReason === "aborted") {
@@ -288,6 +299,13 @@ async function runLoop(
 /**
  * Stream an assistant response from the LLM.
  * This is where AgentMessage[] gets transformed to Message[] for the LLM.
+ *
+ * Emits `message_start` / `message_update` as the response streams in, and
+ * pushes the finalized message onto `context.messages`, but deliberately does
+ * NOT emit `message_end` for it — the caller (runLoop) decides how to
+ * finalize each attempt (as-is, or reclassified as a non-replayable error
+ * turn) before that event, which is what lands the message in Agent state
+ * and session history.
  */
 /** Bounded default for resampling turns whose tool call leaked into text. */
 const DEFAULT_TOOL_CALL_LEAK_RETRIES = 2;
@@ -315,6 +333,26 @@ function isToolCallMarkupLeak(message: AssistantMessage): boolean {
 	return sawMarkup;
 }
 
+/**
+ * Reclassify a leaked-markup attempt that is about to be resampled as a
+ * failed, non-retryable turn. `providers`' `transformMessages` already skips
+ * assistant messages with stopReason "error" (or "aborted") when building the
+ * next request, so marking it this way removes the raw `<tool_call>` text
+ * from every future request built from this context — including a later
+ * resumed session — while the message itself (and its own message_start /
+ * message_end events) still lands in state and persisted history for
+ * observability. The errorMessage text intentionally does not match
+ * providers' transient-error retry patterns, so this can never be picked up
+ * by an automatic error-retry / autopilot-resume path.
+ */
+function markLeakedAttemptNonReplayable(message: AssistantMessage): AssistantMessage {
+	return {
+		...message,
+		stopReason: "error",
+		errorMessage: "Tool-call markup leaked into text; response was resampled.",
+	};
+}
+
 async function streamAssistantResponse(
 	context: AgentContext,
 	config: AgentLoopConfig,
@@ -391,7 +429,6 @@ async function streamAssistantResponse(
 				if (!addedPartial) {
 					await emit({ type: "message_start", message: { ...finalMessage } });
 				}
-				await emit({ type: "message_end", message: finalMessage });
 				return finalMessage;
 			}
 		}
@@ -404,7 +441,6 @@ async function streamAssistantResponse(
 		context.messages.push(finalMessage);
 		await emit({ type: "message_start", message: { ...finalMessage } });
 	}
-	await emit({ type: "message_end", message: finalMessage });
 	return finalMessage;
 }
 

packages/agent-core/test/agent-loop.test.ts

@@ -1649,29 +1649,92 @@ describe("tool-call markup leak retry", () => {
 		// context to the first call.
 		expect(requestMessages[1]).toEqual(requestMessages[0]);
 		// Both attempts remain observable in the event stream.
-		expect(events.filter((e) => e.type === "message_end" && (e as any).message.role === "assistant")).toHaveLength(2);
+		const assistantMessageEnds = events.filter(
+			(e) => e.type === "message_end" && (e as any).message.role === "assistant",
+		) as Extract<AgentEvent, { type: "message_end" }>[];
+		expect(assistantMessageEnds).toHaveLength(2);
+		// The leaked attempt is reclassified as a non-retryable error turn so it can
+		// never be replayed to the model (transform-messages.ts skips stopReason
+		// "error"/"aborted" assistant messages), while the resampled turn is
+		// committed unmarked with its real content intact.
+		const leakedEndMessage = assistantMessageEnds[0].message as AssistantMessage;
+		expect(leakedEndMessage.stopReason).toBe("error");
+		expect(leakedEndMessage.errorMessage).toMatch(/leak/i);
+		const finalEndMessage = assistantMessageEnds[1].message as AssistantMessage;
+		expect(finalEndMessage.stopReason).toBe("stop");
+		expect(finalEndMessage.content).toEqual([{ type: "text", text: "done cleanly" }]);
 		// Only one turn ends.
 		expect(events.filter((e) => e.type === "turn_end")).toHaveLength(1);
 	});
 
+	it("excludes the marked leaked attempt from a later turn built from persisted messages", async () => {
+		// Simulates what happens across a resumed session: everything observed via
+		// message_end (including the leaked attempt) is what gets persisted and fed
+		// back in as `messages` for the next turn. A replay-aware convertToLlm (like
+		// providers' transformMessages) must not see the raw leaked markup.
+		const leaked = createAssistantMessage([{ type: "text", text: LEAK_TEXT }]);
+		const good = createAssistantMessage([{ type: "text", text: "done cleanly" }]);
+		const { streamFn } = scriptedStream([leaked, good]);
+
+		const persisted: AgentMessage[] = [];
+		const context: AgentContext = { systemPrompt: "", messages: [], tools: [] };
+		const config: AgentLoopConfig = { model: createModel(), convertToLlm: identityConverter };
+
+		const stream = agentLoop([createUserMessage("go")], context, config, undefined, streamFn);
+		for await (const event of stream) {
+			if (event.type === "message_end") persisted.push(event.message);
+		}
+
+		// The leaked attempt IS in persisted history (kept for observability)...
+		const persistedAssistants = persisted.filter((m) => m.role === "assistant") as AssistantMessage[];
+		expect(persistedAssistants).toHaveLength(2);
+		expect(persistedAssistants[0].stopReason).toBe("error");
+
+		// ...but a replay-skip filter equivalent to providers' transformMessages
+		// (skip assistant messages with stopReason "error"/"aborted") removes the
+		// leaked markup entirely from what a later turn would send to the model.
+		const replayableForNextTurn = persisted.filter(
+			(m) =>
+				!(
+					m.role === "assistant" &&
+					((m as AssistantMessage).stopReason === "error" || (m as AssistantMessage).stopReason === "aborted")
+				),
+		);
+		const serialized = JSON.stringify(replayableForNextTurn);
+		expect(serialized).not.toContain("tool_call");
+		expect(serialized).toContain("done cleanly");
+	});
+
 	it("stops after the bounded retries and commits the last attempt", async () => {
 		const leaked = createAssistantMessage([{ type: "text", text: LEAK_TEXT }]);
 		const { streamFn, calls, requestMessages } = scriptedStream([leaked, leaked, leaked]);
 
 		const context: AgentContext = { systemPrompt: "", messages: [], tools: [] };
 		const config: AgentLoopConfig = { model: createModel(), convertToLlm: identityConverter };
 
+		const events: AgentEvent[] = [];
 		const stream = agentLoop([createUserMessage("go")], context, config, undefined, streamFn);
-		for await (const _event of stream) {
-			// drain
-		}
+		for await (const event of stream) events.push(event);
 		const messages = await stream.result();
 
 		// 1 initial + 2 default retries.
 		expect(calls()).toBe(3);
 		expect(requestMessages[1]).toEqual(requestMessages[0]);
 		expect(requestMessages[2]).toEqual(requestMessages[0]);
+		// Budget exhausted: the final leaked attempt stays as-is (not marked), so
+		// it keeps carrying the raw leaked text as its own committed content.
 		expect((messages.at(-1) as AssistantMessage).content).toEqual([{ type: "text", text: LEAK_TEXT }]);
+		expect((messages.at(-1) as AssistantMessage).stopReason).toBe("stop");
+
+		const assistantMessageEnds = events.filter(
+			(e) => e.type === "message_end" && (e as any).message.role === "assistant",
+		) as Extract<AgentEvent, { type: "message_end" }>[];
+		expect(assistantMessageEnds).toHaveLength(3);
+		// The two resampled (superseded) attempts are marked non-retryable...
+		expect((assistantMessageEnds[0].message as AssistantMessage).stopReason).toBe("error");
+		expect((assistantMessageEnds[1].message as AssistantMessage).stopReason).toBe("error");
+		// ...but the last, budget-exhausted attempt is committed unmarked.
+		expect((assistantMessageEnds[2].message as AssistantMessage).stopReason).toBe("stop");
 	});
 
 	it("can be disabled with toolCallLeakRetries: 0", async () => {

packages/coding-agent/test/agent-session-retry.test.ts

@@ -148,6 +148,98 @@ describe("AgentSession retry", () => {
 		expect(created.session.isRetrying).toBe(false);
 	});
 
+	it("does not auto-retry a resampled tool-call markup leak", async () => {
+		// Regression: agent-loop's own leak-retry resamples a turn whose text
+		// leaked raw `<tool_call>` markup, and marks the leaked attempt with
+		// stopReason "error" so it is never replayed. That marker must not be
+		// mistaken for a transient provider error and trigger AgentSession's
+		// separate auto-retry path on top of the already-resampled turn.
+		const LEAK_TEXT = "<tool_call> <function=run_command> ls </function> </tool_call>";
+		let callCount = 0;
+		const requestMessages: unknown[][] = [];
+		const model = stepModel();
+		const agent = new Agent({
+			getApiKey: () => "test-key",
+			initialState: { model, systemPrompt: "Test", tools: [] },
+			streamFn: (_model, llmContext: { messages: unknown[] }) => {
+				callCount++;
+				requestMessages.push([...llmContext.messages]);
+				const stream = new MockAssistantStream();
+				queueMicrotask(() => {
+					// Call 1 leaks; call 2 resamples cleanly (both inside the first
+					// prompt()); call 3 is a later, independent prompt() built from
+					// the persisted (marked) history.
+					const msg = callCount === 1 ? createAssistantMessage(LEAK_TEXT) : createAssistantMessage("done cleanly");
+					stream.push({ type: "start", partial: msg });
+					stream.push({ type: "done", reason: "stop", message: msg });
+				});
+				return stream;
+			},
+		});
+
+		const sessionManager = SessionManager.inMemory();
+		const settingsManager = SettingsManager.create(tempDir, tempDir);
+		const authStorage = AuthStorage.create(join(tempDir, "auth.json"));
+		const modelRegistry = await createModelRegistry(authStorage, tempDir);
+		await authStorage.modify("step", async () => ({ type: "api_key", key: "test-key" }));
+		settingsManager.applyOverrides({ retry: { enabled: true, maxRetries: 3, baseDelayMs: 1 } });
+
+		session = new AgentSession({
+			agent,
+			sessionManager,
+			settingsManager,
+			cwd: tempDir,
+			modelRuntime: getModelRuntime(modelRegistry),
+			resourceLoader: createTestResourceLoader(),
+		});
+
+		const events: string[] = [];
+		session.subscribe((event) => {
+			if (event.type === "auto_retry_start") events.push(`start:${event.attempt}`);
+			if (event.type === "auto_retry_end") events.push(`end:success=${event.success}`);
+		});
+
+		await session.prompt("Test");
+
+		// agent-loop resampled once on its own (2 model calls); AgentSession's
+		// separate auto-retry never fired on top of it.
+		expect(callCount).toBe(2);
+		expect(events).toEqual([]);
+		expect(session.isRetrying).toBe(false);
+
+		const messages = session.agent.state.messages;
+		const leakedMessage = messages.find(
+			(m) =>
+				m.role === "assistant" &&
+				(m as AssistantMessage).content.some((c) => c.type === "text" && c.text === LEAK_TEXT),
+		) as AssistantMessage | undefined;
+		expect(leakedMessage?.stopReason).toBe("error");
+
+		// A later, independent prompt() rebuilds its request from persisted
+		// history. `@step-harness/providers`' internal transformMessages (used by
+		// every real provider adapter just before the wire request) is what
+		// actually strips stopReason "error"/"aborted" assistant messages from
+		// that history - it isn't exported for reuse here, so this asserts the
+		// same invariant with an equivalent hand-rolled filter: the raw
+		// (unfiltered) request still carries the marked leak object (proving
+		// coding-agent's own convertToLlm does not itself strip it - matching
+		// production, where the real provider layer does that job), but the
+		// filter a replay path applies removes the leaked text entirely.
+		await session.prompt("Again");
+		expect(callCount).toBe(3);
+
+		const rawThirdRequest = requestMessages[2] as AssistantMessage[];
+		const serializedRaw = JSON.stringify(rawThirdRequest);
+		expect(serializedRaw).toContain(LEAK_TEXT);
+
+		const replayableThirdRequest = rawThirdRequest.filter(
+			(m) => !(m.role === "assistant" && (m.stopReason === "error" || m.stopReason === "aborted")),
+		);
+		const serializedReplayable = JSON.stringify(replayableThirdRequest);
+		expect(serializedReplayable).not.toContain(LEAK_TEXT);
+		expect(serializedReplayable).toContain("done cleanly");
+	});
+
 	it("exhausts max retries and emits failure", async () => {
 		const created = await createSession({ failCount: 99, maxRetries: 2 });
 		const events: string[] = [];