Merge pull request RooCodeInc#1435 from RooVetGit/cte/terminal-reliability

Theoretical handling of stream non-availability or delayed availability

Author: cte

src/integrations/terminal/TerminalManager.ts

@@ -82,7 +82,10 @@ declare module "vscode" {
 	// https://github.com/microsoft/vscode/blob/f0417069c62e20f3667506f4b7e53ca0004b4e3e/src/vscode-dts/vscode.d.ts#L10794
 	interface Window {
 		onDidStartTerminalShellExecution?: (
-			listener: (e: any) => any,
+			listener: (e: {
+				terminal: vscode.Terminal
+				execution: { read(): AsyncIterable<string>; commandLine: { value: string } }
+			}) => any,
 			thisArgs?: any,
 			disposables?: vscode.Disposable[],
 		) => vscode.Disposable
@@ -203,57 +206,77 @@ export class TerminalManager {
 	constructor() {
 		let startDisposable: vscode.Disposable | undefined
 		let endDisposable: vscode.Disposable | undefined
+
 		try {
 			// onDidStartTerminalShellExecution
 			startDisposable = (vscode.window as vscode.Window).onDidStartTerminalShellExecution?.(async (e) => {
 				// Get a handle to the stream as early as possible:
 				const stream = e?.execution.read()
 				const terminalInfo = TerminalRegistry.getTerminalInfoByTerminal(e.terminal)
-				if (stream && terminalInfo) {
-					const process = this.processes.get(terminalInfo.id)
-					if (process) {
-						terminalInfo.stream = stream
-						terminalInfo.running = true
-						terminalInfo.streamClosed = false
-						process.emit("stream_available", terminalInfo.id, stream)
-					}
-				} else {
-					console.error("[TerminalManager] Stream failed, not registered for terminal")
-				}
 
-				console.info("[TerminalManager] Shell execution started:", {
+				console.info("[TerminalManager] shell execution started", {
 					hasExecution: !!e?.execution,
+					hasStream: !!stream,
 					command: e?.execution?.commandLine?.value,
 					terminalId: terminalInfo?.id,
 				})
+
+				if (terminalInfo) {
+					const process = this.processes.get(terminalInfo.id)
+
+					if (process) {
+						if (stream) {
+							terminalInfo.stream = stream
+							terminalInfo.running = true
+							terminalInfo.streamClosed = false
+							console.log(`[TerminalManager] stream_available -> ${terminalInfo.id}`)
+							process.emit("stream_available", terminalInfo.id, stream)
+						} else {
+							process.emit("stream_unavailable", terminalInfo.id)
+							console.error(`[TerminalManager] stream_unavailable -> ${terminalInfo.id}`)
+						}
+					}
+				} else {
+					console.error("[TerminalManager] terminalInfo not available")
+				}
 			})
 
 			// onDidEndTerminalShellExecution
 			endDisposable = (vscode.window as vscode.Window).onDidEndTerminalShellExecution?.(async (e) => {
 				const exitDetails = this.interpretExitCode(e?.exitCode)
-				console.info("[TerminalManager] Shell execution ended:", {
-					...exitDetails,
-				})
+				console.info("[TerminalManager] Shell execution ended:", { ...exitDetails })
+				let emitted = false
 
-				// Signal completion to any waiting processes
+				// Signal completion to any waiting processes.
 				for (const id of this.terminalIds) {
 					const info = TerminalRegistry.getTerminal(id)
+
 					if (info && info.terminal === e.terminal) {
 						info.running = false
 						const process = this.processes.get(id)
+
 						if (process) {
+							console.log(`[TerminalManager] emitting shell_execution_complete -> ${id}`)
+							emitted = true
 							process.emit("shell_execution_complete", id, exitDetails)
 						}
+
 						break
 					}
 				}
+
+				if (!emitted) {
+					console.log(`[TerminalManager#onDidStartTerminalShellExecution] no terminal found`)
+				}
 			})
 		} catch (error) {
-			console.error("[TerminalManager] Error setting up shell execution handlers:", error)
+			console.error("[TerminalManager] failed to configure shell execution handlers", error)
 		}
+
 		if (startDisposable) {
 			this.disposables.push(startDisposable)
 		}
+
 		if (endDisposable) {
 			this.disposables.push(endDisposable)
 		}
@@ -366,9 +389,6 @@ export class TerminalManager {
 	}
 
 	disposeAll() {
-		// for (const info of this.terminals) {
-		// 	//info.terminal.dispose() // dont want to dispose terminals when task is aborted
-		// }
 		this.terminalIds.clear()
 		this.processes.clear()
 		this.disposables.forEach((disposable) => disposable.dispose())

src/integrations/terminal/TerminalProcess.ts

@@ -47,6 +47,7 @@ export interface TerminalProcessEvents {
 	 */
 	shell_execution_complete: [id: number, exitDetails: ExitCodeDetails]
 	stream_available: [id: number, stream: AsyncIterable<string>]
+	stream_unavailable: [id: number]
 	/**
 	 * Emitted when an execution fails to emit a "line" event for a given period of time.
 	 * @param id The terminal ID
@@ -85,15 +86,24 @@ export class TerminalProcess extends EventEmitter<TerminalProcessEvents> {
 			const terminalInfo = TerminalRegistry.getTerminalInfoByTerminal(terminal)
 
 			if (!terminalInfo) {
-				console.error("[TerminalProcess] Terminal not found in registry")
+				console.error("[TerminalProcess#run] terminal not found in registry")
 				this.emit("no_shell_integration")
 				this.emit("completed")
 				this.emit("continue")
 				return
 			}
 
-			// When executeCommand() is called, onDidStartTerminalShellExecution will fire in TerminalManager
-			// which creates a new stream via execution.read() and emits 'stream_available'
+			this.once("stream_unavailable", (id: number) => {
+				if (id === terminalInfo.id) {
+					console.error(`[TerminalProcess#run] stream_unavailable`)
+					this.emit("completed")
+					this.emit("continue")
+				}
+			})
+
+			// When `executeCommand()` is called, `onDidStartTerminalShellExecution`
+			// will fire in `TerminalManager` which creates a new stream via
+			// `execution.read()` and emits `stream_available`.
 			const streamAvailable = new Promise<AsyncIterable<string>>((resolve) => {
 				this.once("stream_available", (id: number, stream: AsyncIterable<string>) => {
 					if (id === terminalInfo.id) {
@@ -111,15 +121,35 @@ export class TerminalProcess extends EventEmitter<TerminalProcessEvents> {
 				})
 			})
 
-			// readLine needs to know if streamClosed, so store this for later
+			// `readLine()` needs to know if streamClosed, so store this for later.
+			// NOTE: This doesn't seem to be used anywhere.
 			this.terminalInfo = terminalInfo
 
-			// Execute command
+			// Execute command.
 			terminal.shellIntegration.executeCommand(command)
 			this.isHot = true
 
-			// Wait for stream to be available
-			const stream = await streamAvailable
+			// Wait for stream to be available.
+			// const stream = await streamAvailable
+
+			// Wait for stream to be available.
+			let stream: AsyncIterable<string>
+
+			try {
+				stream = await Promise.race([
+					streamAvailable,
+					new Promise<never>((_, reject) => {
+						setTimeout(
+							() => reject(new Error("Timeout waiting for terminal stream to become available")),
+							10_000,
+						)
+					}),
+				])
+			} catch (error) {
+				console.error(`[TerminalProcess#run] timed out waiting for stream`)
+				this.emit("stream_stalled", terminalInfo.id)
+				stream = await streamAvailable
+			}
 
 			let preOutput = ""
 			let commandOutputStarted = false
@@ -256,6 +286,7 @@ export class TerminalProcess extends EventEmitter<TerminalProcessEvents> {
 	}
 
 	public continue() {
+		console.log(`[TerminalProcess#continue] flushing all`)
 		this.flushAll()
 		this.isListening = false
 		this.removeAllListeners("line")