Refactor dashboard server modules

src/observability/dashboard-format.ts

@@ -0,0 +1,94 @@
+export function formatRuntimeAndTurns(
+  startedAt: string,
+  turnCount: number,
+  generatedAt: string,
+): string {
+  const runtime = formatRuntimeSeconds(
+    runtimeSecondsFromStartedAt(startedAt, generatedAt),
+  );
+  return Number.isInteger(turnCount) && turnCount > 0
+    ? `${runtime} / ${turnCount}`
+    : runtime;
+}
+
+export function formatRuntimeSeconds(seconds: number): string {
+  if (!Number.isFinite(seconds) || seconds < 0) {
+    return "0m 0s";
+  }
+  const wholeSeconds = Math.max(0, Math.trunc(seconds));
+  const mins = Math.floor(wholeSeconds / 60);
+  const secs = wholeSeconds % 60;
+  return `${mins}m ${secs}s`;
+}
+
+export function runtimeSecondsFromStartedAt(
+  startedAt: string,
+  generatedAt: string,
+): number {
+  const start = Date.parse(startedAt);
+  const generated = Date.parse(generatedAt);
+  if (
+    !Number.isFinite(start) ||
+    !Number.isFinite(generated) ||
+    generated < start
+  ) {
+    return 0;
+  }
+  return (generated - start) / 1000;
+}
+
+export function formatInteger(value: number): string {
+  return Number.isFinite(value)
+    ? Math.trunc(value).toLocaleString("en-US")
+    : "n/a";
+}
+
+export function prettyValue(value: unknown): string {
+  return value === null || value === undefined
+    ? "n/a"
+    : JSON.stringify(value, null, 2);
+}
+
+export function stateBadgeClass(state: string): string {
+  const normalized = state.toLowerCase();
+  if (
+    normalized.includes("progress") ||
+    normalized.includes("running") ||
+    normalized.includes("active")
+  ) {
+    return "state-badge state-badge-active";
+  }
+  if (
+    normalized.includes("blocked") ||
+    normalized.includes("error") ||
+    normalized.includes("failed")
+  ) {
+    return "state-badge state-badge-danger";
+  }
+  if (
+    normalized.includes("todo") ||
+    normalized.includes("queued") ||
+    normalized.includes("pending") ||
+    normalized.includes("retry")
+  ) {
+    return "state-badge state-badge-warning";
+  }
+  return "state-badge";
+}
+
+export function escapeHtml(value: string | number): string {
+  return String(value)
+    .replaceAll("&", "&amp;")
+    .replaceAll("<", "&lt;")
+    .replaceAll(">", "&gt;")
+    .replaceAll('"', "&quot;")
+    .replaceAll("'", "&#39;");
+}
+
+export function toErrorMessage(error: unknown): string {
+  if (error instanceof Error) {
+    return error.message;
+  }
+
+  return String(error);
+}

src/observability/dashboard-http.ts

@@ -0,0 +1,80 @@
+import type { IncomingMessage, ServerResponse } from "node:http";
+
+import type { RuntimeSnapshot } from "../logging/runtime-snapshot.js";
+import type { DashboardServerHost } from "./dashboard-server.js";
+
+export async function readSnapshot(
+  host: DashboardServerHost,
+  timeoutMs: number,
+): Promise<RuntimeSnapshot> {
+  return await withTimeout(host.getRuntimeSnapshot(), timeoutMs, () => {
+    return new Error(`Runtime snapshot timed out after ${timeoutMs}ms.`);
+  });
+}
+
+export function writeJson(
+  response: ServerResponse,
+  statusCode: number,
+  payload: unknown,
+): void {
+  const body = JSON.stringify(payload);
+  response.statusCode = statusCode;
+  response.setHeader("content-type", "application/json; charset=utf-8");
+  response.setHeader("content-length", Buffer.byteLength(body));
+  response.end(body);
+}
+
+export function writeHtml(
+  response: ServerResponse,
+  statusCode: number,
+  html: string,
+): void {
+  response.statusCode = statusCode;
+  response.setHeader("content-type", "text/html; charset=utf-8");
+  response.setHeader("content-length", Buffer.byteLength(html));
+  response.end(html);
+}
+
+export function writeNotFound(response: ServerResponse, path: string): void {
+  response.statusCode = 404;
+  response.setHeader("content-type", "text/plain; charset=utf-8");
+  response.end(`Not found: ${path}`);
+}
+
+export async function readRequestBody(request: IncomingMessage): Promise<void> {
+  await new Promise<void>((resolve, reject) => {
+    request.on("error", reject);
+    request.on("end", resolve);
+    request.resume();
+  });
+}
+
+export function isSnapshotTimeoutError(error: unknown): boolean {
+  return (
+    error instanceof Error &&
+    error.message.startsWith("Runtime snapshot timed out after ")
+  );
+}
+
+async function withTimeout<T>(
+  promise: Promise<T> | T,
+  timeoutMs: number,
+  createError: () => Error,
+): Promise<T> {
+  return await new Promise<T>((resolve, reject) => {
+    const timeout = setTimeout(() => {
+      reject(createError());
+    }, timeoutMs);
+
+    Promise.resolve(promise).then(
+      (value) => {
+        clearTimeout(timeout);
+        resolve(value);
+      },
+      (error) => {
+        clearTimeout(timeout);
+        reject(error);
+      },
+    );
+  });
+}

src/observability/dashboard-live-updates.ts

@@ -0,0 +1,146 @@
+import type { IncomingMessage, ServerResponse } from "node:http";
+
+import { ERROR_CODES } from "../errors/codes.js";
+import type { RuntimeSnapshot } from "../logging/runtime-snapshot.js";
+import { toErrorMessage } from "./dashboard-format.js";
+import { isSnapshotTimeoutError, readSnapshot } from "./dashboard-http.js";
+import type { DashboardServerHost } from "./dashboard-server.js";
+
+export class DashboardLiveUpdatesController {
+  readonly #host: DashboardServerHost;
+  readonly #snapshotTimeoutMs: number;
+  readonly #refreshMs: number;
+  readonly #renderIntervalMs: number;
+  readonly #clients = new Set<ServerResponse<IncomingMessage>>();
+  #flushTimer: NodeJS.Timeout | null = null;
+  #heartbeatTimer: NodeJS.Timeout | null = null;
+  #unsubscribeHost: (() => void) | null = null;
+  #closed = false;
+
+  constructor(options: {
+    host: DashboardServerHost;
+    snapshotTimeoutMs: number;
+    refreshMs: number;
+    renderIntervalMs: number;
+  }) {
+    this.#host = options.host;
+    this.#snapshotTimeoutMs = options.snapshotTimeoutMs;
+    this.#refreshMs = options.refreshMs;
+    this.#renderIntervalMs = options.renderIntervalMs;
+  }
+
+  start(): void {
+    if (typeof this.#host.subscribeToSnapshots === "function") {
+      this.#unsubscribeHost = this.#host.subscribeToSnapshots(() => {
+        this.scheduleBroadcast();
+      });
+    }
+  }
+
+  async close(): Promise<void> {
+    this.#closed = true;
+    this.#unsubscribeHost?.();
+    this.#unsubscribeHost = null;
+    this.clearTimers();
+
+    for (const client of this.#clients) {
+      client.end();
+    }
+    this.#clients.clear();
+  }
+
+  async handleEventsRequest(
+    request: IncomingMessage,
+    response: ServerResponse,
+  ): Promise<void> {
+    response.statusCode = 200;
+    response.setHeader("content-type", "text/event-stream; charset=utf-8");
+    response.setHeader("cache-control", "no-cache, no-transform");
+    response.setHeader("connection", "keep-alive");
+    response.setHeader("x-accel-buffering", "no");
+    response.write(`retry: ${this.#refreshMs}\n\n`);
+
+    this.#clients.add(response);
+    this.startHeartbeat();
+
+    const cleanup = () => {
+      this.#clients.delete(response);
+      if (this.#clients.size === 0) {
+        this.stopHeartbeat();
+      }
+    };
+
+    request.on("close", cleanup);
+    response.on("close", cleanup);
+
+    await this.writeSnapshot(response);
+  }
+
+  scheduleBroadcast(): void {
+    if (this.#closed || this.#clients.size === 0 || this.#flushTimer !== null) {
+      return;
+    }
+
+    this.#flushTimer = setTimeout(() => {
+      this.#flushTimer = null;
+      void this.broadcastSnapshot();
+    }, this.#renderIntervalMs);
+  }
+
+  private startHeartbeat(): void {
+    if (this.#heartbeatTimer !== null) {
+      return;
+    }
+
+    this.#heartbeatTimer = setInterval(() => {
+      this.scheduleBroadcast();
+    }, this.#refreshMs);
+  }
+
+  private stopHeartbeat(): void {
+    if (this.#heartbeatTimer === null) {
+      return;
+    }
+
+    clearInterval(this.#heartbeatTimer);
+    this.#heartbeatTimer = null;
+  }
+
+  private clearTimers(): void {
+    if (this.#flushTimer !== null) {
+      clearTimeout(this.#flushTimer);
+      this.#flushTimer = null;
+    }
+    this.stopHeartbeat();
+  }
+
+  private async broadcastSnapshot(): Promise<void> {
+    const clients = [...this.#clients];
+    if (clients.length === 0) {
+      return;
+    }
+
+    await Promise.allSettled(
+      clients.map((client) => this.writeSnapshot(client)),
+    );
+  }
+
+  private async writeSnapshot(response: ServerResponse): Promise<void> {
+    try {
+      const snapshot: RuntimeSnapshot = await readSnapshot(
+        this.#host,
+        this.#snapshotTimeoutMs,
+      );
+      response.write(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`);
+    } catch (error) {
+      response.write(
+        `event: error\ndata: ${JSON.stringify({
+          code: isSnapshotTimeoutError(error)
+            ? ERROR_CODES.snapshotTimedOut
+            : ERROR_CODES.snapshotUnavailable,
+          message: toErrorMessage(error),
+        })}\n\n`,
+      );
+    }
+  }
+}