import { AsyncLock, AsyncQueue } from "../../core/async";import { CONTROLLER_EVENTS } from "@local-studio/contracts/controller-events"; /** Controller event that can be serialized to an SSE frame. */export class Event { public readonly type: string; public readonly data: Record<string, unknown>; public readonly timestamp: string; public readonly id: string; public constructor(type: string, data: Record<string, unknown>) { this.type = type; this.data = data; this.timestamp = new Date().toISOString(); this.id = `${Date.now()}`; } public toSse(): string { const payload = { data: this.data, timestamp: this.timestamp }; return `id: ${this.id}\nevent: ${this.type}\ndata: ${JSON.stringify(payload)}\n\n`; }} /** SSE event manager with channels and backpressure handling. */export class EventManager { private readonly subscribers = new Map<string, Set<AsyncQueue<Event>>>(); private readonly lock = new AsyncLock(); private latestMetrics: Record<string, unknown> = {}; public async *subscribe(channel = "default", signal?: AbortSignal): AsyncIterable<Event> { const queue = new AsyncQueue<Event>(100); const release = await this.lock.acquire(); try { const existing = this.subscribers.get(channel) ?? new Set<AsyncQueue<Event>>(); existing.add(queue); this.subscribers.set(channel, existing); } finally { release(); } try { while (true) { if (signal?.aborted) break; let event: Event; try { event = await queue.shift(signal); } catch { break; } yield event; } } finally { queue.close(); const releaseCleanup = await this.lock.acquire(); try { const existing = this.subscribers.get(channel); if (existing) { existing.delete(queue); if (existing.size === 0) { this.subscribers.delete(channel); } } } finally { releaseCleanup(); } } } public async publish(event: Event, channel = "default"): Promise<void> { const release = await this.lock.acquire(); try { const subscribers = this.subscribers.get(channel); if (!subscribers || subscribers.size === 0) { return; } const deadQueues: AsyncQueue<Event>[] = []; for (const queue of subscribers) { const ok = queue.push(event); if (!ok) { deadQueues.push(queue); } } for (const dead of deadQueues) { subscribers.delete(dead); } } finally { release(); } } public async publishStatus(statusData: Record<string, unknown>): Promise<void> { await this.publish(new Event(CONTROLLER_EVENTS.STATUS, statusData)); } public async publishGpu(gpuData: Record<string, unknown>[]): Promise<void> { await this.publish(new Event(CONTROLLER_EVENTS.GPU, { gpus: gpuData, count: gpuData.length })); } public async publishMetrics(metricsData: Record<string, unknown>): Promise<void> { this.latestMetrics = { ...metricsData }; await this.publish(new Event(CONTROLLER_EVENTS.METRICS, metricsData)); } public getLatestMetrics(): Record<string, unknown> { return { ...this.latestMetrics }; } public async publishRuntimeSummary(summaryData: Record<string, unknown>): Promise<void> { await this.publish(new Event(CONTROLLER_EVENTS.RUNTIME_SUMMARY, summaryData)); } public async publishLogLine(sessionId: string, line: string): Promise<void> { await this.publish( new Event(CONTROLLER_EVENTS.LOG, { session_id: sessionId, line }), `logs:${sessionId}`, ); } public async publishLaunchProgress( recipeId: string, stage: string, message: string, progress?: number, ): Promise<void> { const payload: Record<string, unknown> = { recipe_id: recipeId, stage, message }; if (progress !== undefined) { payload["progress"] = progress; } await this.publish(new Event(CONTROLLER_EVENTS.LAUNCH_PROGRESS, payload)); }}
import { AsyncLock, AsyncQueue } from "../../core/async";
import { CONTROLLER_EVENTS } from "@local-studio/contracts/controller-events";
/** Controller event that can be serialized to an SSE frame. */
export class Event {
public readonly type: string;
public readonly data: Record<string, unknown>;
public readonly timestamp: string;
public readonly id: string;
public constructor(type: string, data: Record<string, unknown>) {
this.type = type;
this.data = data;
this.timestamp = new Date().toISOString();
this.id = `${Date.now()}`;
}
public toSse(): string {
const payload = { data: this.data, timestamp: this.timestamp };
return `id: ${this.id}\nevent: ${this.type}\ndata: ${JSON.stringify(payload)}\n\n`;
}
}
/** SSE event manager with channels and backpressure handling. */
export class EventManager {
private readonly subscribers = new Map<string, Set<AsyncQueue<Event>>>();
private readonly lock = new AsyncLock();
private latestMetrics: Record<string, unknown> = {};
public async *subscribe(channel = "default", signal?: AbortSignal): AsyncIterable<Event> {
const queue = new AsyncQueue<Event>(100);
const release = await this.lock.acquire();
try {
const existing = this.subscribers.get(channel) ?? new Set<AsyncQueue<Event>>();
existing.add(queue);
this.subscribers.set(channel, existing);
} finally {
release();
}
try {
while (true) {
if (signal?.aborted) break;
let event: Event;
try {
event = await queue.shift(signal);
} catch {
break;
}
yield event;
}
} finally {
queue.close();
const releaseCleanup = await this.lock.acquire();
try {
const existing = this.subscribers.get(channel);
if (existing) {
existing.delete(queue);
if (existing.size === 0) {
this.subscribers.delete(channel);
}
}
} finally {
releaseCleanup();
}
}
}
public async publish(event: Event, channel = "default"): Promise<void> {
const release = await this.lock.acquire();
try {
const subscribers = this.subscribers.get(channel);
if (!subscribers || subscribers.size === 0) {
return;
}
const deadQueues: AsyncQueue<Event>[] = [];
for (const queue of subscribers) {
const ok = queue.push(event);
if (!ok) {
deadQueues.push(queue);
}
}
for (const dead of deadQueues) {
subscribers.delete(dead);
}
} finally {
release();
}
}
public async publishStatus(statusData: Record<string, unknown>): Promise<void> {
await this.publish(new Event(CONTROLLER_EVENTS.STATUS, statusData));
}
public async publishGpu(gpuData: Record<string, unknown>[]): Promise<void> {
await this.publish(new Event(CONTROLLER_EVENTS.GPU, { gpus: gpuData, count: gpuData.length }));
}
public async publishMetrics(metricsData: Record<string, unknown>): Promise<void> {
this.latestMetrics = { ...metricsData };
await this.publish(new Event(CONTROLLER_EVENTS.METRICS, metricsData));
}
public getLatestMetrics(): Record<string, unknown> {
return { ...this.latestMetrics };
}
public async publishRuntimeSummary(summaryData: Record<string, unknown>): Promise<void> {
await this.publish(new Event(CONTROLLER_EVENTS.RUNTIME_SUMMARY, summaryData));
}
public async publishLogLine(sessionId: string, line: string): Promise<void> {
await this.publish(
new Event(CONTROLLER_EVENTS.LOG, { session_id: sessionId, line }),
`logs:${sessionId}`,
);
}
public async publishLaunchProgress(
recipeId: string,
stage: string,
message: string,
progress?: number,
): Promise<void> {
const payload: Record<string, unknown> = { recipe_id: recipeId, stage, message };
if (progress !== undefined) {
payload["progress"] = progress;
}
await this.publish(new Event(CONTROLLER_EVENTS.LAUNCH_PROGRESS, payload));
}
}