Address review comments

Author: EhabY

src/api/workspace.ts

@@ -18,14 +18,30 @@ import { type CoderApi } from "./coderApi";
 /** Opens a stream once; subsequent open() calls are no-ops until closed. */
 export class LazyStream<T> {
 	private stream: UnidirectionalStream<T> | null = null;
+	private opening: Promise<void> | null = null;
 
 	async open(factory: () => Promise<UnidirectionalStream<T>>): Promise<void> {
-		this.stream ??= await factory();
+		if (this.stream) return;
+
+		// Deduplicate concurrent calls; close() clears the reference to cancel.
+		if (!this.opening) {
+			const promise = factory().then((s) => {
+				if (this.opening === promise) {
+					this.stream = s;
+					this.opening = null;
+				} else {
+					s.close();
+				}
+			});
+			this.opening = promise;
+		}
+		await this.opening;
 	}
 
 	close(): void {
 		this.stream?.close();
 		this.stream = null;
+		this.opening = null;
 	}
 }
 

src/webviews/tasks/tasksPanel.ts

@@ -480,7 +480,8 @@ export class TasksPanel
 
 		const onOutput = (line: string) => {
 			const clean = stripAnsi(line);
-			if (clean.length === 0) return;
+			// Skip lines that were purely ANSI codes, but keep intentional blank lines.
+			if (line.length > 0 && clean.length === 0) return;
 			this.sendNotification({
 				type: TasksApi.workspaceLogsAppend.method,
 				data: [clean],
@@ -644,6 +645,7 @@ export class TasksPanel
 	dispose(): void {
 		this.buildLogStream.close();
 		this.agentLogStream.close();
+		this.streamingTaskId = null;
 		for (const d of this.disposables) {
 			d.dispose();
 		}

test/unit/api/workspace.test.ts

@@ -0,0 +1,88 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { LazyStream } from "@/api/workspace";
+import { type UnidirectionalStream } from "@/websocket/eventStreamConnection";
+
+function mockStream(): UnidirectionalStream<unknown> {
+	return {
+		url: "ws://test",
+		addEventListener: vi.fn(),
+		removeEventListener: vi.fn(),
+		close: vi.fn(),
+	};
+}
+
+type StreamFactory = () => Promise<UnidirectionalStream<unknown>>;
+
+/** Creates a factory whose promise can be resolved manually. */
+function deferredFactory() {
+	let resolve!: (s: UnidirectionalStream<unknown>) => void;
+	const factory: StreamFactory = vi.fn().mockReturnValue(
+		new Promise<UnidirectionalStream<unknown>>((r) => {
+			resolve = r;
+		}),
+	);
+	return {
+		factory,
+		resolve: (s?: UnidirectionalStream<unknown>) => resolve(s ?? mockStream()),
+	};
+}
+
+describe("LazyStream", () => {
+	it("opens once and ignores subsequent calls", async () => {
+		const factory: StreamFactory = vi.fn().mockResolvedValue(mockStream());
+		const lazy = new LazyStream();
+
+		await lazy.open(factory);
+		await lazy.open(factory);
+
+		expect(factory).toHaveBeenCalledOnce();
+	});
+
+	it("can reopen after close", async () => {
+		const factory: StreamFactory = vi.fn().mockResolvedValue(mockStream());
+		const lazy = new LazyStream();
+
+		await lazy.open(factory);
+		lazy.close();
+		await lazy.open(factory);
+
+		expect(factory).toHaveBeenCalledTimes(2);
+	});
+
+	it("closes the underlying stream", async () => {
+		const stream = mockStream();
+		const lazy = new LazyStream();
+
+		await lazy.open(() => Promise.resolve(stream));
+		lazy.close();
+
+		expect(stream.close).toHaveBeenCalledOnce();
+	});
+
+	it("deduplicates concurrent opens", async () => {
+		const { factory, resolve } = deferredFactory();
+		const lazy = new LazyStream();
+
+		const p1 = lazy.open(factory);
+		const p2 = lazy.open(factory);
+		resolve();
+		await Promise.all([p1, p2]);
+
+		expect(factory).toHaveBeenCalledOnce();
+	});
+
+	it("allows reopening after close during pending open", async () => {
+		const { factory, resolve } = deferredFactory();
+		const lazy = new LazyStream();
+
+		const p = lazy.open(factory);
+		lazy.close();
+		resolve();
+		await p.catch(() => {});
+
+		const factory2: StreamFactory = vi.fn().mockResolvedValue(mockStream());
+		await lazy.open(factory2);
+		expect(factory2).toHaveBeenCalledOnce();
+	});
+});