CLI: improve stream recovery and add configurable consecutive mistake limit

apps/cli/README.md

@@ -185,6 +185,7 @@ Tokens are valid for 90 days. The CLI will prompt you to re-authenticate when yo
 | `-m, --model <model>`             | Model to use                                                                            | `anthropic/claude-opus-4.6`              |
 | `--mode <mode>`                   | Mode to start in (code, architect, ask, debug, etc.)                                    | `code`                                   |
 | `-r, --reasoning-effort <effort>` | Reasoning effort level (unspecified, disabled, none, minimal, low, medium, high, xhigh) | `medium`                                 |
+| `--consecutive-mistake-limit <n>` | Consecutive error/repetition limit before guidance prompt (`0` disables the limit)      | `10`                                     |
 | `--ephemeral`                     | Run without persisting state (uses temporary storage)                                   | `false`                                  |
 | `--oneshot`                       | Exit upon task completion                                                               | `false`                                  |
 | `--output-format <format>`        | Output format with `--print`: `text`, `json`, or `stream-json`                          | `text`                                   |

apps/cli/src/agent/__tests__/events.test.ts

@@ -0,0 +1,35 @@
+import type { ClineMessage } from "@roo-code/types"
+
+import { detectAgentState } from "../agent-state.js"
+import { taskCompleted } from "../events.js"
+
+function createMessage(overrides: Partial<ClineMessage>): ClineMessage {
+	return { ts: Date.now() + Math.random() * 1000, type: "say", ...overrides }
+}
+
+describe("taskCompleted", () => {
+	it("returns true for completion_result", () => {
+		const previous = detectAgentState([createMessage({ type: "say", say: "text", text: "working" })])
+		const current = detectAgentState([createMessage({ type: "ask", ask: "completion_result", partial: false })])
+
+		expect(taskCompleted(previous, current)).toBe(true)
+	})
+
+	it("returns true for resume_completed_task", () => {
+		const previous = detectAgentState([createMessage({ type: "say", say: "text", text: "working" })])
+		const current = detectAgentState([createMessage({ type: "ask", ask: "resume_completed_task", partial: false })])
+
+		expect(taskCompleted(previous, current)).toBe(true)
+	})
+
+	it("returns false for recoverable idle asks", () => {
+		const previous = detectAgentState([createMessage({ type: "say", say: "text", text: "working" })])
+		const mistakeLimit = detectAgentState([
+			createMessage({ type: "ask", ask: "mistake_limit_reached", partial: false }),
+		])
+		const apiFailed = detectAgentState([createMessage({ type: "ask", ask: "api_req_failed", partial: false })])
+
+		expect(taskCompleted(previous, mistakeLimit)).toBe(false)
+		expect(taskCompleted(previous, apiFailed)).toBe(false)
+	})
+})

apps/cli/src/agent/__tests__/extension-host.test.ts

@@ -5,6 +5,8 @@ import fs from "fs"
 
 import type { ExtensionMessage, WebviewMessage } from "@roo-code/types"
 
+import { DEFAULT_FLAGS } from "@/types/index.js"
+
 import { type ExtensionHostOptions, ExtensionHost } from "../extension-host.js"
 import { ExtensionClient } from "../extension-client.js"
 import { AgentLoopState } from "../agent-state.js"
@@ -593,6 +595,20 @@ describe("ExtensionHost", () => {
 			expect(initialSettings.mode).toBe("architect")
 		})
 
+		it("should use default consecutiveMistakeLimit when not provided", () => {
+			const host = createTestHost()
+
+			const initialSettings = getPrivate<Record<string, unknown>>(host, "initialSettings")
+			expect(initialSettings.consecutiveMistakeLimit).toBe(DEFAULT_FLAGS.consecutiveMistakeLimit)
+		})
+
+		it("should set consecutiveMistakeLimit from options", () => {
+			const host = createTestHost({ consecutiveMistakeLimit: 8 })
+
+			const initialSettings = getPrivate<Record<string, unknown>>(host, "initialSettings")
+			expect(initialSettings.consecutiveMistakeLimit).toBe(8)
+		})
+
 		it("should enable auto-approval in non-interactive mode", () => {
 			const host = createTestHost({ nonInteractive: true })
 

apps/cli/src/agent/__tests__/json-event-emitter-result.test.ts

@@ -0,0 +1,129 @@
+import type { ClineMessage } from "@roo-code/types"
+import { Writable } from "stream"
+
+import type { TaskCompletedEvent } from "../events.js"
+import { JsonEventEmitter } from "../json-event-emitter.js"
+import { AgentLoopState, type AgentStateInfo } from "../agent-state.js"
+
+function createMockStdout(): { stdout: NodeJS.WriteStream; lines: () => Record<string, unknown>[] } {
+	const chunks: string[] = []
+
+	const writable = new Writable({
+		write(chunk, _encoding, callback) {
+			chunks.push(chunk.toString())
+			callback()
+		},
+	}) as unknown as NodeJS.WriteStream
+
+	const lines = () =>
+		chunks
+			.join("")
+			.split("\n")
+			.filter((line) => line.length > 0)
+			.map((line) => JSON.parse(line) as Record<string, unknown>)
+
+	return { stdout: writable, lines }
+}
+
+function emitMessage(emitter: JsonEventEmitter, message: ClineMessage): void {
+	;(emitter as unknown as { handleMessage: (msg: ClineMessage, isUpdate: boolean) => void }).handleMessage(
+		message,
+		false,
+	)
+}
+
+function emitTaskCompleted(emitter: JsonEventEmitter, event: TaskCompletedEvent): void {
+	;(emitter as unknown as { handleTaskCompleted: (taskCompleted: TaskCompletedEvent) => void }).handleTaskCompleted(
+		event,
+	)
+}
+
+function createAskCompletionMessage(ts: number, text = ""): ClineMessage {
+	return {
+		ts,
+		type: "ask",
+		ask: "completion_result",
+		partial: false,
+		text,
+	} as ClineMessage
+}
+
+function createCompletedStateInfo(message: ClineMessage): AgentStateInfo {
+	return {
+		state: AgentLoopState.IDLE,
+		isWaitingForInput: true,
+		isRunning: false,
+		isStreaming: false,
+		currentAsk: "completion_result",
+		requiredAction: "start_task",
+		lastMessageTs: message.ts,
+		lastMessage: message,
+		description: "Task completed successfully. You can provide feedback or start a new task.",
+	}
+}
+
+describe("JsonEventEmitter result emission", () => {
+	it("prefers current completion message content over stale cached completion text", () => {
+		const { stdout, lines } = createMockStdout()
+		const emitter = new JsonEventEmitter({ mode: "stream-json", stdout })
+
+		emitMessage(emitter, {
+			ts: 100,
+			type: "say",
+			say: "completion_result",
+			partial: false,
+			text: "FIRST",
+		} as ClineMessage)
+
+		const firstCompletionMessage = createAskCompletionMessage(101, "")
+		emitTaskCompleted(emitter, {
+			success: true,
+			stateInfo: createCompletedStateInfo(firstCompletionMessage),
+			message: firstCompletionMessage,
+		})
+
+		const secondCompletionMessage = createAskCompletionMessage(102, "SECOND")
+		emitTaskCompleted(emitter, {
+			success: true,
+			stateInfo: createCompletedStateInfo(secondCompletionMessage),
+			message: secondCompletionMessage,
+		})
+
+		const output = lines().filter((line) => line.type === "result")
+		expect(output).toHaveLength(2)
+		expect(output[0]?.content).toBe("FIRST")
+		expect(output[1]?.content).toBe("SECOND")
+	})
+
+	it("clears cached completion text after each result emission", () => {
+		const { stdout, lines } = createMockStdout()
+		const emitter = new JsonEventEmitter({ mode: "stream-json", stdout })
+
+		emitMessage(emitter, {
+			ts: 200,
+			type: "say",
+			say: "completion_result",
+			partial: false,
+			text: "FIRST",
+		} as ClineMessage)
+
+		const firstCompletionMessage = createAskCompletionMessage(201, "")
+		emitTaskCompleted(emitter, {
+			success: true,
+			stateInfo: createCompletedStateInfo(firstCompletionMessage),
+			message: firstCompletionMessage,
+		})
+
+		const secondCompletionMessage = createAskCompletionMessage(202, "")
+		emitTaskCompleted(emitter, {
+			success: true,
+			stateInfo: createCompletedStateInfo(secondCompletionMessage),
+			message: secondCompletionMessage,
+		})
+
+		const output = lines().filter((line) => line.type === "result")
+		expect(output).toHaveLength(2)
+		expect(output[0]?.content).toBe("FIRST")
+		expect(output[1]).not.toHaveProperty("content")
+	})
+})

apps/cli/src/agent/events.ts

@@ -260,7 +260,7 @@ export function streamingEnded(previous: AgentStateInfo, current: AgentStateInfo
  * Helper to determine if task completed.
  */
 export function taskCompleted(previous: AgentStateInfo, current: AgentStateInfo): boolean {
-	const completionAsks = ["completion_result", "api_req_failed", "mistake_limit_reached"]
+	const completionAsks = ["completion_result", "resume_completed_task"]
 	const wasNotComplete = !previous.currentAsk || !completionAsks.includes(previous.currentAsk)
 	const isNowComplete = current.currentAsk !== undefined && completionAsks.includes(current.currentAsk)
 	return wasNotComplete && isNowComplete

apps/cli/src/agent/extension-host.ts

@@ -26,7 +26,7 @@ import type {
 import { createVSCodeAPI, IExtensionHost, ExtensionHostEventMap, setRuntimeConfigValues } from "@roo-code/vscode-shim"
 import { DebugLogger, setDebugLogEnabled } from "@roo-code/core/cli"
 
-import type { SupportedProvider } from "@/types/index.js"
+import { DEFAULT_FLAGS, type SupportedProvider } from "@/types/index.js"
 import type { User } from "@/lib/sdk/index.js"
 import { getProviderSettings } from "@/lib/utils/provider.js"
 import { createEphemeralStorageDir } from "@/lib/storage/index.js"
@@ -66,6 +66,7 @@ const CLI_PACKAGE_ROOT = process.env.ROO_CLI_ROOT || findCliPackageRoot()
 export interface ExtensionHostOptions {
 	mode: string
 	reasoningEffort?: ReasoningEffortExtended | "unspecified" | "disabled"
+	consecutiveMistakeLimit?: number
 	user: User | null
 	provider: SupportedProvider
 	apiKey?: string
@@ -219,6 +220,7 @@ export class ExtensionHost extends EventEmitter implements ExtensionHostInterfac
 		// Populate initial settings.
 		const baseSettings: RooCodeSettings = {
 			mode: this.options.mode,
+			consecutiveMistakeLimit: this.options.consecutiveMistakeLimit ?? DEFAULT_FLAGS.consecutiveMistakeLimit,
 			commandExecutionTimeout: 30,
 			enableCheckpoints: false,
 			experiments: {

apps/cli/src/agent/json-event-emitter.ts

@@ -96,6 +96,7 @@ export class JsonEventEmitter {
 	private stdout: NodeJS.WriteStream
 	private events: JsonEvent[] = []
 	private unsubscribers: (() => void)[] = []
+	private pendingWrites = new Set<Promise<void>>()
 	private lastCost: JsonEventCost | undefined
 	private requestIdProvider: () => string | undefined
 	private schemaVersion: number
@@ -598,8 +599,9 @@ export class JsonEventEmitter {
 	 * Handle task completion and emit result event.
 	 */
 	private handleTaskCompleted(event: TaskCompletedEvent): void {
-		// Use tracked completion result content, falling back to event message
-		const resultContent = this.completionResultContent || event.message?.text || this.lastAssistantText
+		// Prefer the completion payload from the current event. If it is empty,
+		// fall back to the most recent tracked completion text, then assistant text.
+		const resultContent = event.message?.text || this.completionResultContent || this.lastAssistantText
 
 		this.emitEvent({
 			type: "result",
@@ -610,6 +612,10 @@ export class JsonEventEmitter {
 			cost: this.lastCost,
 		})
 
+		// Prevent stale completion content from leaking into later turns.
+		this.completionResultContent = undefined
+		this.lastAssistantText = undefined
+
 		// For "json" mode, output the final accumulated result
 		if (this.mode === "json") {
 			this.outputFinalResult(event.success, resultContent)
@@ -647,7 +653,7 @@ export class JsonEventEmitter {
 	 * Output a single JSON line (NDJSON format).
 	 */
 	private outputLine(data: unknown): void {
-		this.stdout.write(JSON.stringify(data) + "\n")
+		this.writeToStdout(JSON.stringify(data) + "\n")
 	}
 
 	/**
@@ -662,7 +668,31 @@ export class JsonEventEmitter {
 			events: this.events.filter((e) => e.type !== "result"), // Exclude the result event itself
 		}
 
-		this.stdout.write(JSON.stringify(output, null, 2) + "\n")
+		this.writeToStdout(JSON.stringify(output, null, 2) + "\n")
+	}
+
+	private writeToStdout(content: string): void {
+		const writePromise = new Promise<void>((resolve, reject) => {
+			this.stdout.write(content, (error?: Error | null) => {
+				if (error) {
+					reject(error)
+					return
+				}
+				resolve()
+			})
+		})
+
+		this.pendingWrites.add(writePromise)
+
+		void writePromise.finally(() => {
+			this.pendingWrites.delete(writePromise)
+		})
+	}
+
+	async flush(): Promise<void> {
+		while (this.pendingWrites.size > 0) {
+			await Promise.all([...this.pendingWrites])
+		}
 	}
 
 	/**

apps/cli/src/agent/message-processor.ts

@@ -343,13 +343,16 @@ export class MessageProcessor {
 
 		// Task completed
 		if (taskCompleted(previousState, currentState)) {
+			const completedSuccessfully =
+				currentState.currentAsk === "completion_result" || currentState.currentAsk === "resume_completed_task"
+
 			if (this.options.debug) {
 				debugLog("[MessageProcessor] EMIT taskCompleted", {
-					success: currentState.currentAsk === "completion_result",
+					success: completedSuccessfully,
 				})
 			}
 			const completedEvent: TaskCompletedEvent = {
-				success: currentState.currentAsk === "completion_result",
+				success: completedSuccessfully,
 				stateInfo: currentState,
 				message: currentState.lastMessage,
 			}