feat(deploy): live SSE progress for deploy_component

bin/cliOperations.ts

@@ -11,11 +11,19 @@ import * as fs from 'fs-extra';
 import * as YAML from 'yaml';
 import { streamPackagedDirectory } from '../components/packageComponent.ts';
 import { buildMultipartBody } from './multipartBuilder.ts';
+import { parseSSE, renderDeployProgress } from './sseConsumer.ts';
 import { getHdbPid } from '../utility/processManagement/processManagement.js';
 import { initConfig, getConfigPath } from '../config/configUtils.js';
 
 const OP_ALIASES = { deploy: 'deploy_component', package: 'package_component' };
 
+// Operations whose responses should be consumed as text/event-stream so live phase events
+// (extract, install, load, replicate, restart) render as they happen instead of after the
+// whole deploy completes. Add an operation here only after wiring its server-side
+// SSE_PROGRESS_OPERATIONS entry — otherwise the server returns the buffered JSON path and
+// the SSE parser sees no events.
+const SSE_OPERATIONS = new Set(['deploy_component']);
+
 // Properties on `req` that the CLI itself uses for transport/UX, not the operations API.
 // They never get serialized into the request body.
 const TRANSPORT_ONLY_FIELDS = new Set([
@@ -191,6 +199,11 @@ async function cliOperations(req: any, skipResponseLog = false) {
 				options.headers.Authorization = `Bearer ${tokens.operation_token}`;
 			}
 		}
+		const useSse = SSE_OPERATIONS.has(req.operation);
+		if (useSse) {
+			options.headers.Accept = 'text/event-stream';
+			options.streamResponse = true;
+		}
 		let body;
 		if (req._multipart) {
 			const packageStream = req._packageStream;
@@ -216,13 +229,43 @@ async function cliOperations(req: any, skipResponseLog = false) {
 		let response: any = await httpRequest(options, body);
 
 		let responseData;
-		try {
-			responseData = JSON.parse(response.body);
-		} catch {
-			responseData = {
-				status: response.statusCode + ' ' + (response.statusMessage || 'Unknown'),
-				body: response.body,
-			};
+		if (useSse && response.headers['content-type']?.startsWith('text/event-stream')) {
+			// Consume SSE: render phase events live, capture the final result from the `done`
+			// event (or the error message from the `error` event). The HTTP status stays 200
+			// until end-of-stream; failures are signaled in-band.
+			const state = {};
+			let finalResult;
+			let sseError;
+			for await (const message of parseSSE(response)) {
+				renderDeployProgress(message, state, process.stderr);
+				if (message.event === 'done') {
+					try {
+						finalResult = JSON.parse(message.data)?.result;
+					} catch {
+						finalResult = message.data;
+					}
+				} else if (message.event === 'error') {
+					try {
+						sseError = JSON.parse(message.data);
+					} catch {
+						sseError = { message: message.data };
+					}
+				}
+			}
+			if (sseError) {
+				console.error(`error: ${sseError.message ?? sseError}`);
+				process.exit(1);
+			}
+			responseData = finalResult ?? { message: 'Deploy completed (no result payload).' };
+		} else {
+			try {
+				responseData = JSON.parse(response.body);
+			} catch {
+				responseData = {
+					status: response.statusCode + ' ' + (response.statusMessage || 'Unknown'),
+					body: response.body,
+				};
+			}
 		}
 
 		let responseLog;

bin/sseConsumer.ts

@@ -0,0 +1,123 @@
+import type { Readable } from 'node:stream';
+
+export interface SSEMessage {
+	event: string;
+	data: string;
+	id?: string;
+	retry?: number;
+}
+
+/**
+ * Parse a Readable carrying Server-Sent Events into structured messages.
+ *
+ * Yields one `SSEMessage` per blank-line-terminated record. Handles split data: lines,
+ * CRLF or LF line endings, and arbitrary chunk boundaries — the underlying Node http
+ * Readable does not guarantee chunks align with SSE record boundaries.
+ */
+export async function* parseSSE(stream: Readable): AsyncGenerator<SSEMessage> {
+	let buffer = '';
+	for await (const chunk of stream) {
+		buffer += chunk.toString('utf8');
+		while (true) {
+			const recordEnd = buffer.indexOf('\n\n');
+			const crlfEnd = buffer.indexOf('\r\n\r\n');
+			let endIdx = -1;
+			let delimLen = 0;
+			if (recordEnd !== -1 && (crlfEnd === -1 || recordEnd < crlfEnd)) {
+				endIdx = recordEnd;
+				delimLen = 2;
+			} else if (crlfEnd !== -1) {
+				endIdx = crlfEnd;
+				delimLen = 4;
+			}
+			if (endIdx === -1) break;
+			const record = buffer.slice(0, endIdx);
+			buffer = buffer.slice(endIdx + delimLen);
+			const msg = parseRecord(record);
+			if (msg) yield msg;
+		}
+	}
+	// Any trailing record without a terminating blank line is treated as a final message,
+	// matching the looser behavior browsers exhibit on connection close.
+	if (buffer.trim()) {
+		const msg = parseRecord(buffer);
+		if (msg) yield msg;
+	}
+}
+
+function parseRecord(record: string): SSEMessage | null {
+	const lines = record.split(/\r?\n/);
+	let event = 'message';
+	let id: string | undefined;
+	let retry: number | undefined;
+	const dataLines: string[] = [];
+	for (const line of lines) {
+		if (line === '' || line.startsWith(':')) continue;
+		const colon = line.indexOf(':');
+		const field = colon === -1 ? line : line.slice(0, colon);
+		// Per spec, a leading space after the colon is stripped.
+		let value = colon === -1 ? '' : line.slice(colon + 1);
+		if (value.startsWith(' ')) value = value.slice(1);
+		switch (field) {
+			case 'event':
+				event = value;
+				break;
+			case 'data':
+				dataLines.push(value);
+				break;
+			case 'id':
+				id = value;
+				break;
+			case 'retry': {
+				const n = Number(value);
+				if (Number.isFinite(n)) retry = n;
+				break;
+			}
+		}
+	}
+	if (dataLines.length === 0 && event === 'message') return null;
+	return { event, data: dataLines.join('\n'), id, retry };
+}
+
+interface RenderState {
+	currentPhase?: string;
+}
+
+/**
+ * Render SSE deploy events as terse, line-oriented progress to stderr (so stdout stays
+ * reserved for the final JSON/YAML response document). Phase transitions print once.
+ */
+export function renderDeployProgress(message: SSEMessage, state: RenderState, output: NodeJS.WritableStream): void {
+	let parsed: unknown;
+	try {
+		parsed = JSON.parse(message.data);
+	} catch {
+		parsed = message.data;
+	}
+	switch (message.event) {
+		case 'phase': {
+			const p = parsed as { phase?: string; status?: string; rolling?: boolean };
+			const label = p.phase ?? '?';
+			if (p.status === 'start') {
+				if (state.currentPhase !== label) {
+					output.write(`${label}…\n`);
+					state.currentPhase = label;
+				}
+			} else if (p.status === 'done') {
+				output.write(`${label} done\n`);
+			} else if (p.status === 'error') {
+				const msg = (parsed as { message?: string }).message ?? 'failed';
+				output.write(`${label} ERROR: ${msg}\n`);
+			}
+			break;
+		}
+		case 'error': {
+			const e = parsed as { message?: string; code?: string | number };
+			output.write(`error: ${e.message ?? message.data}${e.code ? ` (${e.code})` : ''}\n`);
+			break;
+		}
+		case 'done':
+			// Caller picks up the final result via the SSE iterator; nothing to render here.
+			break;
+	}
+}

components/Application.ts

@@ -434,6 +434,10 @@ export class Application {
 	dirPath: string;
 	logger: Logger;
 	packageManagerPrefix: string; // can be used to configure a package manager prefix, specifically "sfw".
+	// Optional progress emitter for SSE-style reporting. Set by the operations API when the
+	// caller requested `Accept: text/event-stream`. Undefined for the historical
+	// single-response code path; phase-event emissions are all optional-chained off this.
+	progress?: { emit(event: string, data: unknown): void };
 
 	constructor({ name, payload, packageIdentifier, install }: ApplicationOptions) {
 		this.name = name;
@@ -476,7 +480,14 @@ export function derivePackageIdentifier(packageIdentifier: string) {
  * @returns A promise that resolves when all preparation steps complete.
  */
 export function prepareApplication(application: Application) {
-	return extractApplication(application).then(() => installApplication(application));
+	return extractApplication(application).then(() => {
+		// extractApplication finished; the next phase is install. We emit the boundary here so
+		// the SSE consumer sees `extract done → install start` in order even though Application
+		// itself isn't aware of which phase comes next.
+		application.progress?.emit('phase', { phase: 'extract', status: 'done' });
+		application.progress?.emit('phase', { phase: 'install', status: 'start' });
+		return installApplication(application);
+	});
 }
 
 /**

components/operations.js

@@ -348,6 +348,11 @@ async function packageComponent(req) {
  * @returns {Promise<string>}
  */
 async function deployComponent(req) {
+	// `req.progress` is a ProgressEmitter set by handlePostRequest when the client sends
+	// `Accept: text/event-stream`. Non-SSE callers leave it undefined; the optional-chained
+	// calls below are no-ops in that case, keeping the historical single-response path intact.
+	const progress = req.progress;
+
 	if (req.project) {
 		req.project = path.parse(req.project).name;
 	} else if (req.package) {
@@ -393,12 +398,21 @@ async function deployComponent(req) {
 			timeout: req.install_timeout,
 		},
 	});
+	if (progress) application.progress = progress;
 
-	await prepareApplication(application);
+	progress?.emit('phase', { phase: 'extract', status: 'start' });
+	try {
+		await prepareApplication(application);
+	} catch (err) {
+		progress?.emit('phase', { phase: 'extract_or_install', status: 'error', message: err?.message ?? String(err) });
+		throw err;
+	}
+	progress?.emit('phase', { phase: 'install', status: 'done' });
 
 	// now we attempt to actually load the component in case there is
 	// an error we can immediately detect and report, but app code should not run on the main thread
 	if (!isMainThread && !process.env.HARPER_SAFE_MODE) {
+		progress?.emit('phase', { phase: 'load', status: 'start' });
 		const pseudoResources = new Resources();
 		pseudoResources.isWorker = true;
 
@@ -415,16 +429,24 @@ async function deployComponent(req) {
 			req.project
 		);
 
-		if (lastError) throw lastError;
+		if (lastError) {
+			progress?.emit('phase', { phase: 'load', status: 'error', message: lastError?.message ?? String(lastError) });
+			throw lastError;
+		}
+		progress?.emit('phase', { phase: 'load', status: 'done' });
 	}
 	const rollingRestart = req.restart === 'rolling';
 	// if doing a rolling restart set restart to false so that other nodes don't also restart.
 	req.restart = rollingRestart ? false : req.restart;
+	progress?.emit('phase', { phase: 'replicate', status: 'start' });
 	let response = await server.replication.replicateOperation(req);
+	progress?.emit('phase', { phase: 'replicate', status: 'done' });
 	if (req.restart === true) {
+		progress?.emit('phase', { phase: 'restart', status: 'start' });
 		manageThreads.restartWorkers('http');
 		response.message = `Successfully deployed: ${application.name}, restarting Harper`;
 	} else if (rollingRestart) {
+		progress?.emit('phase', { phase: 'restart', status: 'start', rolling: true });
 		const serverUtilities = require('../server/serverHelpers/serverUtilities.ts');
 		const jobResponse = await serverUtilities.executeJob({
 			operation: 'restart_service',

server/serverHelpers/progressEmitter.ts

@@ -0,0 +1,79 @@
+import { PassThrough, Readable } from 'node:stream';
+
+export interface ProgressEvent {
+	event: string;
+	data: unknown;
+}
+
+export type ProgressListener = (event: ProgressEvent) => void;
+
+/**
+ * Lightweight pub-sub used to report phase/install/replicate events from a long-running
+ * operation back to the HTTP layer. We deliberately don't use Node's EventEmitter here:
+ * we only need broadcast semantics for a small set of event types, and we want the
+ * `emit(event, data)` shape that matches the SSE wire format directly.
+ */
+export class ProgressEmitter {
+	private listeners: ProgressListener[] = [];
+
+	emit(event: string, data: unknown): void {
+		// Snapshot before iteration so a listener that unsubscribes itself during dispatch
+		// doesn't shift indexes underneath us.
+		const snapshot = this.listeners.slice();
+		for (const listener of snapshot) {
+			try {
+				listener({ event, data });
+			} catch {
+				// A buggy listener must never break the operation. Swallow and continue.
+			}
+		}
+	}
+
+	subscribe(listener: ProgressListener): () => void {
+		this.listeners.push(listener);
+		return () => {
+			const i = this.listeners.indexOf(listener);
+			if (i !== -1) this.listeners.splice(i, 1);
+		};
+	}
+}
+
+/**
+ * Wrap a long-running operation so its progress events stream back as Server-Sent Events.
+ *
+ * The returned Readable emits one SSE message per `emitter.emit(...)` call, then a final
+ * `done` (or `error`) event with the operation's result, then ends. The caller is
+ * expected to set Content-Type: text/event-stream on the response.
+ */
+export function createSSEResponseStream(emitter: ProgressEmitter, operation: () => Promise<unknown>): Readable {
+	const stream = new PassThrough();
+
+	const unsubscribe = emitter.subscribe((event) => {
+		writeSSE(stream, event);
+	});
+
+	operation()
+		.then((result) => {
+			writeSSE(stream, { event: 'done', data: { result } });
+		})
+		.catch((err) => {
+			writeSSE(stream, {
+				event: 'error',
+				data: {
+					message: err?.message ?? String(err),
+					code: err?.statusCode ?? err?.code,
+				},
+			});
+		})
+		.finally(() => {
+			unsubscribe();
+			stream.end();
+		});
+
+	return stream;
+}
+
+function writeSSE(stream: PassThrough, event: ProgressEvent): void {
+	const data = typeof event.data === 'string' ? event.data : JSON.stringify(event.data);
+	stream.write(`event: ${event.event}\ndata: ${data}\n\n`);
+}