From 372fc27df1f8e0201cb4b7283ddb0cf364365eba Mon Sep 17 00:00:00 2001 From: octane0411 Date: Sat, 7 Mar 2026 14:51:55 +0800 Subject: [PATCH 1/2] Add live dashboard updates --- README.md | 2 + docs/DEV_GUIDE.md | 6 +- docs/WORKFLOW.template.md | 17 + src/config/config-resolver.ts | 33 ++ src/config/defaults.ts | 8 + src/config/types.ts | 7 + src/observability/dashboard-server.ts | 435 ++++++++++++++++--- src/orchestrator/runtime-host.ts | 44 +- tests/agent/runner.test.ts | 5 + tests/cli/main.test.ts | 5 + tests/cli/runtime-integration.test.ts | 5 + tests/config/config-resolver.test.ts | 20 + tests/config/defaults.test.ts | 9 + tests/observability/dashboard-server.test.ts | 141 +++++- tests/orchestrator/core.test.ts | 5 + tests/orchestrator/runtime-host.test.ts | 5 + 16 files changed, 692 insertions(+), 55 deletions(-) diff --git a/README.md b/README.md index eea83933..1f537c28 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,8 @@ as `WORKFLOW.md`, then change these fields before starting Symphony: - `codex.command` If you want the dashboard, keep `server.port` in the workflow or pass `--port` on the CLI. +The web dashboard now opens with a server-rendered snapshot and continues updating live in the +browser over server-sent events. For a complete reference covering every supported field with defaults and inline documentation, see [docs/WORKFLOW.template.md](docs/WORKFLOW.template.md). diff --git a/docs/DEV_GUIDE.md b/docs/DEV_GUIDE.md index df92ea0b..f99cba66 100644 --- a/docs/DEV_GUIDE.md +++ b/docs/DEV_GUIDE.md @@ -193,6 +193,9 @@ issue to "In Review" and leave a comment summarizing what you did. | `codex.read_timeout_ms` | Max time in ms to wait for the next Codex event before declaring stream stalled | `5000` | | `codex.stall_timeout_ms` | Max silent time in ms before a running agent is declared stalled and stopped | `300000` | | `server.port` | HTTP dashboard port; omit or `null` to disable | `null` | +| `observability.dashboard_enabled` | Enable live dashboard updates when the HTTP server is running | `true` | +| `observability.refresh_ms` | Dashboard heartbeat interval in ms for time-based refreshes | `1000` | +| `observability.render_interval_ms` | Minimum spacing in ms between pushed dashboard renders | `16` | The prompt body uses **Liquid template syntax**. Available variables: - `{{ issue.identifier }}`, `{{ issue.title }}`, `{{ issue.description }}` @@ -324,5 +327,6 @@ These fields take effect on the next poll tick without restarting Symphony: **How to watch runtime state** - Structured JSON logs are the primary observability surface - Launch with `--port 3000` to access the HTTP dashboard at `http://localhost:3000` +- The dashboard serves an initial HTML snapshot, then stays current over `/api/v1/events` ---- \ No newline at end of file +--- diff --git a/docs/WORKFLOW.template.md b/docs/WORKFLOW.template.md index 759c2d68..d1312f6c 100644 --- a/docs/WORKFLOW.template.md +++ b/docs/WORKFLOW.template.md @@ -136,6 +136,23 @@ server: # Port to listen on. Set to a number to enable, or omit/null to disable. # Default: null (disabled) port: null + +# ============================================================ +# observability — Live dashboard refresh behavior (optional) +# ============================================================ +observability: + # Enable live updates for the HTTP dashboard. + # Default: true + dashboard_enabled: true + + # Heartbeat interval in milliseconds for live dashboard refreshes. + # Used to keep runtime counters current even when no orchestration state changes. + # Default: 1000 (1 s) + refresh_ms: 1000 + + # Minimum spacing between pushed dashboard renders in milliseconds. + # Default: 16 (~60 FPS upper bound) + render_interval_ms: 16 --- You are implementing work for Linear issue {{ issue.identifier }}. diff --git a/src/config/config-resolver.ts b/src/config/config-resolver.ts index 346a170a..ccd2e8bd 100644 --- a/src/config/config-resolver.ts +++ b/src/config/config-resolver.ts @@ -15,6 +15,9 @@ import { DEFAULT_MAX_CONCURRENT_AGENTS_BY_STATE, DEFAULT_MAX_RETRY_BACKOFF_MS, DEFAULT_MAX_TURNS, + DEFAULT_OBSERVABILITY_ENABLED, + DEFAULT_OBSERVABILITY_REFRESH_MS, + DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, DEFAULT_POLL_INTERVAL_MS, DEFAULT_READ_TIMEOUT_MS, DEFAULT_STALL_TIMEOUT_MS, @@ -42,6 +45,7 @@ export function resolveWorkflowConfig( const agent = asRecord(config.agent); const codex = asRecord(config.codex); const server = asRecord(config.server); + const observability = asRecord(config.observability); return { workflowPath: workflow.workflowPath, @@ -109,6 +113,17 @@ export function resolveWorkflowConfig( server: { port: readNonNegativeInteger(server.port), }, + observability: { + dashboardEnabled: + readBoolean(observability.dashboard_enabled) ?? + DEFAULT_OBSERVABILITY_ENABLED, + refreshMs: + readPositiveInteger(observability.refresh_ms) ?? + DEFAULT_OBSERVABILITY_REFRESH_MS, + renderIntervalMs: + readPositiveInteger(observability.render_interval_ms) ?? + DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, + }, }; } @@ -201,6 +216,24 @@ function readInteger(value: unknown): number | null { return null; } +function readBoolean(value: unknown): boolean | null { + if (typeof value === "boolean") { + return value; + } + + if (typeof value === "string") { + const normalized = value.trim().toLowerCase(); + if (normalized === "true") { + return true; + } + if (normalized === "false") { + return false; + } + } + + return null; +} + function readPositiveInteger(value: unknown): number | null { const parsed = readInteger(value); if (parsed === null || parsed <= 0) { diff --git a/src/config/defaults.ts b/src/config/defaults.ts index 0f4a8266..94f87ed8 100644 --- a/src/config/defaults.ts +++ b/src/config/defaults.ts @@ -27,6 +27,9 @@ export const DEFAULT_CODEX_COMMAND = "codex app-server"; export const DEFAULT_TURN_TIMEOUT_MS = 3_600_000; export const DEFAULT_READ_TIMEOUT_MS = 5_000; export const DEFAULT_STALL_TIMEOUT_MS = 300_000; +export const DEFAULT_OBSERVABILITY_ENABLED = true; +export const DEFAULT_OBSERVABILITY_REFRESH_MS = 1_000; +export const DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS = 16; export const DEFAULT_LINEAR_PAGE_SIZE = 50; export const DEFAULT_LINEAR_NETWORK_TIMEOUT_MS = 30_000; @@ -63,4 +66,9 @@ export const SPEC_DEFAULTS = Object.freeze({ readTimeoutMs: DEFAULT_READ_TIMEOUT_MS, stallTimeoutMs: DEFAULT_STALL_TIMEOUT_MS, }, + observability: { + dashboardEnabled: DEFAULT_OBSERVABILITY_ENABLED, + refreshMs: DEFAULT_OBSERVABILITY_REFRESH_MS, + renderIntervalMs: DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, + }, } as const); diff --git a/src/config/types.ts b/src/config/types.ts index e3449677..0761e7ad 100644 --- a/src/config/types.ts +++ b/src/config/types.ts @@ -44,6 +44,12 @@ export interface WorkflowServerConfig { port: number | null; } +export interface WorkflowObservabilityConfig { + dashboardEnabled: boolean; + refreshMs: number; + renderIntervalMs: number; +} + export interface ResolvedWorkflowConfig { workflowPath: string; promptTemplate: string; @@ -54,6 +60,7 @@ export interface ResolvedWorkflowConfig { agent: WorkflowAgentConfig; codex: WorkflowCodexConfig; server: WorkflowServerConfig; + observability: WorkflowObservabilityConfig; } export interface DispatchValidationFailure { diff --git a/src/observability/dashboard-server.ts b/src/observability/dashboard-server.ts index 3eaa4653..f052bc5b 100644 --- a/src/observability/dashboard-server.ts +++ b/src/observability/dashboard-server.ts @@ -5,6 +5,10 @@ import { createServer, } from "node:http"; +import { + DEFAULT_OBSERVABILITY_REFRESH_MS, + DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, +} from "../config/defaults.js"; import { ERROR_CODES } from "../errors/codes.js"; import type { RuntimeSnapshot } from "../logging/runtime-snapshot.js"; @@ -73,12 +77,16 @@ export interface DashboardServerHost { issueIdentifier: string, ): IssueDetailResponse | null | Promise; requestRefresh(): RefreshResponse | Promise; + subscribeToSnapshots?(listener: () => void): () => void; } export interface DashboardServerOptions { host: DashboardServerHost; hostname?: string; snapshotTimeoutMs?: number; + refreshMs?: number; + renderIntervalMs?: number; + liveUpdatesEnabled?: boolean; } export interface DashboardServerInstance { @@ -88,18 +96,172 @@ export interface DashboardServerInstance { close(): Promise; } +interface DashboardRenderOptions { + liveUpdatesEnabled: boolean; +} + +class DashboardLiveUpdatesController { + readonly #host: DashboardServerHost; + readonly #snapshotTimeoutMs: number; + readonly #refreshMs: number; + readonly #renderIntervalMs: number; + readonly #clients = new Set>(); + #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 { + 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 { + 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 { + const clients = [...this.#clients]; + if (clients.length === 0) { + return; + } + + await Promise.allSettled(clients.map((client) => this.writeSnapshot(client))); + } + + private async writeSnapshot(response: ServerResponse): Promise { + try { + const snapshot = 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`, + ); + } + } +} + export function createDashboardServer(options: DashboardServerOptions): Server { const hostname = options.hostname ?? "127.0.0.1"; - const handler = createDashboardRequestHandler({ + const snapshotTimeoutMs = + options.snapshotTimeoutMs ?? DEFAULT_SNAPSHOT_TIMEOUT_MS; + const liveController = new DashboardLiveUpdatesController({ host: options.host, + snapshotTimeoutMs, + refreshMs: options.refreshMs ?? DEFAULT_OBSERVABILITY_REFRESH_MS, + renderIntervalMs: + options.renderIntervalMs ?? DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, + }); + liveController.start(); + + const handler = createDashboardRequestHandler({ + ...options, hostname, - ...(options.snapshotTimeoutMs === undefined - ? {} - : { snapshotTimeoutMs: options.snapshotTimeoutMs }), + snapshotTimeoutMs, + liveController, }); - return createServer((request, response) => { + const server = createServer((request, response) => { void handler(request, response); }); + server.on("close", () => { + void liveController.close(); + }); + return server; } export async function startDashboardServer( @@ -143,11 +305,16 @@ export async function startDashboardServer( } export function createDashboardRequestHandler( - options: DashboardServerOptions, + options: DashboardServerOptions & { + liveController?: DashboardLiveUpdatesController; + }, ): (request: IncomingMessage, response: ServerResponse) => Promise { const hostname = options.hostname ?? "127.0.0.1"; const snapshotTimeoutMs = options.snapshotTimeoutMs ?? DEFAULT_SNAPSHOT_TIMEOUT_MS; + const renderOptions: DashboardRenderOptions = { + liveUpdatesEnabled: options.liveUpdatesEnabled ?? true, + }; return async (request, response) => { try { @@ -161,7 +328,7 @@ export function createDashboardRequestHandler( } const snapshot = await readSnapshot(options.host, snapshotTimeoutMs); - writeHtml(response, 200, renderDashboardHtml(snapshot)); + writeHtml(response, 200, renderDashboardHtml(snapshot, renderOptions)); return; } @@ -176,6 +343,28 @@ export function createDashboardRequestHandler( return; } + if (url.pathname === "/api/v1/events") { + if (method !== "GET") { + writeMethodNotAllowed(response, ["GET"]); + return; + } + + if (renderOptions.liveUpdatesEnabled !== true) { + writeNotFound(response, url.pathname); + return; + } + + if (options.liveController === undefined) { + writeJsonError(response, 503, ERROR_CODES.snapshotUnavailable, { + message: "Live dashboard updates are unavailable.", + }); + return; + } + + await options.liveController.handleEventsRequest(request, response); + return; + } + if (url.pathname === "/api/v1/refresh") { if (method !== "POST") { writeMethodNotAllowed(response, ["POST"]); @@ -332,40 +521,10 @@ function isSnapshotTimeoutError(error: unknown): boolean { ); } -function renderDashboardHtml(snapshot: RuntimeSnapshot): string { - const runningRows = - snapshot.running.length === 0 - ? 'No active sessions.' - : snapshot.running - .map( - (row) => ` - - ${escapeHtml(row.issue_identifier)} - ${escapeHtml(row.state)} - ${escapeHtml(row.session_id ?? "-")} - ${row.turn_count} - ${escapeHtml(row.last_event ?? "-")} - ${escapeHtml(row.last_message ?? "-")} - ${escapeHtml(row.last_event_at ?? "-")} - `, - ) - .join(""); - - const retryRows = - snapshot.retrying.length === 0 - ? 'No queued retries.' - : snapshot.retrying - .map( - (row) => ` - - ${escapeHtml(row.issue_identifier ?? row.issue_id)} - ${row.attempt} - ${escapeHtml(row.due_at)} - ${escapeHtml(row.error ?? "-")} - `, - ) - .join(""); - +function renderDashboardHtml( + snapshot: RuntimeSnapshot, + options: DashboardRenderOptions, +): string { return ` @@ -393,6 +552,32 @@ function renderDashboardHtml(snapshot: RuntimeSnapshot): string { h1, h2 { margin: 0 0 12px; } + .header { + display: flex; + align-items: center; + justify-content: space-between; + gap: 16px; + margin-bottom: 16px; + } + .status { + display: inline-flex; + align-items: center; + gap: 8px; + padding: 8px 12px; + border-radius: 999px; + background: rgba(255, 252, 247, 0.9); + border: 1px solid rgba(59, 44, 32, 0.12); + color: #5f5449; + } + .status-dot { + width: 10px; + height: 10px; + border-radius: 999px; + background: #d26c2f; + } + .status-live .status-dot { + background: #2f8f46; + } .grid { display: grid; gap: 16px; @@ -442,33 +627,45 @@ function renderDashboardHtml(snapshot: RuntimeSnapshot): string {
-

Symphony Dashboard

-

Generated at ${escapeHtml(snapshot.generated_at)}

+
+
+

Symphony Dashboard

+

Generated at ${escapeHtml(snapshot.generated_at)}

+
+
+ + ${ + options.liveUpdatesEnabled ? "Live updates connected" : "Static snapshot" + } +
+
Running
-
${snapshot.counts.running}
+
${snapshot.counts.running}
Retrying
-
${snapshot.counts.retrying}
+
${snapshot.counts.retrying}
Input Tokens
-
${snapshot.codex_totals.input_tokens}
+
${snapshot.codex_totals.input_tokens}
Output Tokens
-
${snapshot.codex_totals.output_tokens}
+
${snapshot.codex_totals.output_tokens}
Total Tokens
-
${snapshot.codex_totals.total_tokens}
+
${snapshot.codex_totals.total_tokens}
Seconds Running
-
${snapshot.codex_totals.seconds_running.toFixed(1)}
+
${snapshot.codex_totals.seconds_running.toFixed(1)}
@@ -486,7 +683,7 @@ function renderDashboardHtml(snapshot: RuntimeSnapshot): string { Last Event At - ${runningRows} + ${renderRunningRows(snapshot)} @@ -501,19 +698,153 @@ function renderDashboardHtml(snapshot: RuntimeSnapshot): string { Error - ${retryRows} + ${renderRetryRows(snapshot)}

Rate Limits

- 
${escapeHtml(JSON.stringify(snapshot.rate_limits, null, 2) ?? "null")}
+ 
${escapeHtml(
+          JSON.stringify(snapshot.rate_limits, null, 2) ?? "null",
+        )}
+ `; } +function renderRunningRows(snapshot: RuntimeSnapshot): string { + return snapshot.running.length === 0 + ? 'No active sessions.' + : snapshot.running + .map( + (row) => ` + + ${escapeHtml(row.issue_identifier)} + ${escapeHtml(row.state)} + ${escapeHtml(row.session_id ?? "-")} + ${row.turn_count} + ${escapeHtml(row.last_event ?? "-")} + ${escapeHtml(row.last_message ?? "-")} + ${escapeHtml(row.last_event_at ?? "-")} + `, + ) + .join(""); +} + +function renderRetryRows(snapshot: RuntimeSnapshot): string { + return snapshot.retrying.length === 0 + ? 'No queued retries.' + : snapshot.retrying + .map( + (row) => ` + + ${escapeHtml(row.issue_identifier ?? row.issue_id)} + ${row.attempt} + ${escapeHtml(row.due_at)} + ${escapeHtml(row.error ?? "-")} + `, + ) + .join(""); +} + function escapeHtml(value: string): string { return value .replaceAll("&", "&") diff --git a/src/orchestrator/runtime-host.ts b/src/orchestrator/runtime-host.ts index 030995fe..5f905052 100644 --- a/src/orchestrator/runtime-host.ts +++ b/src/orchestrator/runtime-host.ts @@ -120,6 +120,8 @@ export class OrchestratorRuntimeHost implements DashboardServerHost { private refreshQueued = false; + private readonly snapshotListeners = new Set<() => void>(); + constructor(options: RuntimeHostOptions) { this.config = options.config; this.tracker = options.tracker; @@ -219,6 +221,8 @@ export class OrchestratorRuntimeHost implements DashboardServerHost { : { workspaceManager: this.workspaceManager }), }); } + + this.notifySnapshotListeners(); } async pollOnce() { @@ -287,6 +291,13 @@ export class OrchestratorRuntimeHost implements DashboardServerHost { }; } + subscribeToSnapshots(listener: () => void): () => void { + this.snapshotListeners.add(listener); + return () => { + this.snapshotListeners.delete(listener); + }; + } + private async spawnWorkerExecution( issue: Issue, attempt: number | null, @@ -406,7 +417,19 @@ export class OrchestratorRuntimeHost implements DashboardServerHost { () => undefined, () => undefined, ); - return next; + return next.finally(() => { + this.notifySnapshotListeners(); + }); + } + + private notifySnapshotListeners(): void { + for (const listener of this.snapshotListeners) { + try { + listener(); + } catch { + // Observability listeners must not affect runtime correctness. + } + } } private createManagedAgentRunner(input: { @@ -470,6 +493,9 @@ export async function startRuntimeService( : await startDashboardServer({ host: runtimeHost, port: currentConfig.server.port, + refreshMs: currentConfig.observability.refreshMs, + renderIntervalMs: currentConfig.observability.renderIntervalMs, + liveUpdatesEnabled: currentConfig.observability.dashboardEnabled, }); const stopController = new AbortController(); @@ -556,6 +582,22 @@ export async function startRuntimeService( }, ); } + + if ( + dashboard !== null && + previousConfig.observability.dashboardEnabled !== + nextConfig.observability.dashboardEnabled + ) { + await logger.warn( + "workflow_reload_observability_ignored", + "Ignoring observability.dashboard_enabled change until runtime restart.", + { + outcome: "degraded", + reason: "observability_reload_requires_restart", + port: dashboard.port, + }, + ); + } }, }) : options.workflowWatcher; diff --git a/tests/agent/runner.test.ts b/tests/agent/runner.test.ts index 52afa2b8..75b6d01b 100644 --- a/tests/agent/runner.test.ts +++ b/tests/agent/runner.test.ts @@ -476,6 +476,11 @@ function createConfig(root: string, scenario: string): ResolvedWorkflowConfig { server: { port: null, }, + observability: { + dashboardEnabled: true, + refreshMs: 1_000, + renderIntervalMs: 16, + }, }; } diff --git a/tests/cli/main.test.ts b/tests/cli/main.test.ts index 6fb5b952..cb134a22 100644 --- a/tests/cli/main.test.ts +++ b/tests/cli/main.test.ts @@ -263,6 +263,11 @@ function createConfig( server: { port: null, }, + observability: { + dashboardEnabled: true, + refreshMs: 1_000, + renderIntervalMs: 16, + }, ...overrides, }; } diff --git a/tests/cli/runtime-integration.test.ts b/tests/cli/runtime-integration.test.ts index f0173bf6..caef883b 100644 --- a/tests/cli/runtime-integration.test.ts +++ b/tests/cli/runtime-integration.test.ts @@ -590,6 +590,11 @@ function createConfig( server: { port: null, }, + observability: { + dashboardEnabled: true, + refreshMs: 1_000, + renderIntervalMs: 16, + }, ...overrides, }; } diff --git a/tests/config/config-resolver.test.ts b/tests/config/config-resolver.test.ts index 9047cfef..de544340 100644 --- a/tests/config/config-resolver.test.ts +++ b/tests/config/config-resolver.test.ts @@ -13,6 +13,9 @@ import { DEFAULT_MAX_CONCURRENT_AGENTS, DEFAULT_MAX_RETRY_BACKOFF_MS, DEFAULT_MAX_TURNS, + DEFAULT_OBSERVABILITY_ENABLED, + DEFAULT_OBSERVABILITY_REFRESH_MS, + DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, DEFAULT_POLL_INTERVAL_MS, DEFAULT_READ_TIMEOUT_MS, DEFAULT_STALL_TIMEOUT_MS, @@ -51,6 +54,15 @@ describe("config-resolver", () => { expect(resolved.codex.turnTimeoutMs).toBe(DEFAULT_TURN_TIMEOUT_MS); expect(resolved.codex.readTimeoutMs).toBe(DEFAULT_READ_TIMEOUT_MS); expect(resolved.codex.stallTimeoutMs).toBe(DEFAULT_STALL_TIMEOUT_MS); + expect(resolved.observability.dashboardEnabled).toBe( + DEFAULT_OBSERVABILITY_ENABLED, + ); + expect(resolved.observability.refreshMs).toBe( + DEFAULT_OBSERVABILITY_REFRESH_MS, + ); + expect(resolved.observability.renderIntervalMs).toBe( + DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, + ); }); it("coerces env-backed fields, path-like roots, and state limits", () => { @@ -92,6 +104,11 @@ describe("config-resolver", () => { server: { port: "8080", }, + observability: { + dashboard_enabled: "false", + refresh_ms: "2500", + render_interval_ms: "33", + }, }, }, { @@ -121,6 +138,9 @@ describe("config-resolver", () => { expect(resolved.codex.readTimeoutMs).toBe(2_500); expect(resolved.codex.stallTimeoutMs).toBe(-1); expect(resolved.server.port).toBe(8080); + expect(resolved.observability.dashboardEnabled).toBe(false); + expect(resolved.observability.refreshMs).toBe(2_500); + expect(resolved.observability.renderIntervalMs).toBe(33); }); it("accepts server.port zero for ephemeral listener binding", () => { diff --git a/tests/config/defaults.test.ts b/tests/config/defaults.test.ts index a5400ef5..4dbcd953 100644 --- a/tests/config/defaults.test.ts +++ b/tests/config/defaults.test.ts @@ -8,6 +8,9 @@ import { DEFAULT_MAX_CONCURRENT_AGENTS, DEFAULT_MAX_RETRY_BACKOFF_MS, DEFAULT_MAX_TURNS, + DEFAULT_OBSERVABILITY_ENABLED, + DEFAULT_OBSERVABILITY_REFRESH_MS, + DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS, DEFAULT_POLL_INTERVAL_MS, DEFAULT_READ_TIMEOUT_MS, DEFAULT_STALL_TIMEOUT_MS, @@ -27,6 +30,9 @@ describe("SPEC_DEFAULTS", () => { expect(DEFAULT_TURN_TIMEOUT_MS).toBe(3_600_000); expect(DEFAULT_READ_TIMEOUT_MS).toBe(5_000); expect(DEFAULT_STALL_TIMEOUT_MS).toBe(300_000); + expect(DEFAULT_OBSERVABILITY_ENABLED).toBe(true); + expect(DEFAULT_OBSERVABILITY_REFRESH_MS).toBe(1_000); + expect(DEFAULT_OBSERVABILITY_RENDER_INTERVAL_MS).toBe(16); expect(DEFAULT_CODEX_COMMAND).toBe("codex app-server"); }); @@ -42,6 +48,9 @@ describe("SPEC_DEFAULTS", () => { DEFAULT_MAX_CONCURRENT_AGENTS, ); expect(SPEC_DEFAULTS.codex.command).toBe(DEFAULT_CODEX_COMMAND); + expect(SPEC_DEFAULTS.observability.dashboardEnabled).toBe( + DEFAULT_OBSERVABILITY_ENABLED, + ); expect(Object.isFrozen(SPEC_DEFAULTS)).toBe(true); }); }); diff --git a/tests/observability/dashboard-server.test.ts b/tests/observability/dashboard-server.test.ts index cffb66ce..87f97ea2 100644 --- a/tests/observability/dashboard-server.test.ts +++ b/tests/observability/dashboard-server.test.ts @@ -1,4 +1,4 @@ -import { request as httpRequest } from "node:http"; +import { type IncomingMessage, request as httpRequest } from "node:http"; import { afterEach, describe, expect, it } from "vitest"; @@ -36,6 +36,8 @@ describe("dashboard server", () => { expect(dashboard.headers["content-type"]).toContain("text/html"); expect(dashboard.body).toContain("Symphony Dashboard"); expect(dashboard.body).toContain("ABC-123"); + expect(dashboard.body).toContain("window.__SYMPHONY_SNAPSHOT__"); + expect(dashboard.body).toContain("/api/v1/events"); const state = await sendRequest(server.port, { method: "GET", @@ -238,6 +240,56 @@ describe("dashboard server", () => { }); }); + it("streams snapshot updates over server-sent events", async () => { + let snapshot = createSnapshot(); + let emitUpdate = () => {}; + const server = await startDashboardServer({ + port: 0, + renderIntervalMs: 5, + host: createHost({ + getRuntimeSnapshot: () => snapshot, + subscribeToSnapshots: (listener) => { + emitUpdate = listener; + return () => { + emitUpdate = () => {}; + }; + }, + }), + }); + servers.push(server); + + const stream = await openEventStream(server.port, "/api/v1/events"); + const initial = await stream.nextEvent(); + expect(initial.event).toBe("snapshot"); + expect(JSON.parse(initial.data)).toMatchObject({ + generated_at: "2026-03-06T10:00:00.000Z", + counts: { + running: 1, + }, + }); + + snapshot = { + ...snapshot, + generated_at: "2026-03-06T10:00:02.000Z", + counts: { + running: 2, + retrying: 1, + }, + }; + emitUpdate(); + + const next = await stream.nextEvent(); + expect(next.event).toBe("snapshot"); + expect(JSON.parse(next.data)).toMatchObject({ + generated_at: "2026-03-06T10:00:02.000Z", + counts: { + running: 2, + }, + }); + + stream.close(); + }); + it("returns a plain 404 for undefined routes", async () => { const server = await startDashboardServer({ port: 0, @@ -412,3 +464,90 @@ function sendRequest( request.end(); }); } + +async function openEventStream( + port: number, + path: string, +): Promise<{ + close(): void; + nextEvent(): Promise<{ event: string; data: string }>; +}> { + const eventQueue: Array<{ event: string; data: string }> = []; + const waitingResolvers: Array<(value: { event: string; data: string }) => void> = + []; + let buffer = ""; + let responseRef: IncomingMessage | null = null; + + const request = httpRequest({ + host: "127.0.0.1", + port, + method: "GET", + path, + }); + + await new Promise((resolve, reject) => { + request.on("response", (response) => { + responseRef = response; + response.setEncoding("utf8"); + response.on("data", (chunk) => { + buffer += chunk; + + while (buffer.includes("\n\n")) { + const separatorIndex = buffer.indexOf("\n\n"); + const rawEvent = buffer.slice(0, separatorIndex); + buffer = buffer.slice(separatorIndex + 2); + const parsed = parseServerSentEvent(rawEvent); + if (parsed === null) { + continue; + } + + const resolver = waitingResolvers.shift(); + if (resolver !== undefined) { + resolver(parsed); + continue; + } + eventQueue.push(parsed); + } + }); + resolve(); + }); + request.on("error", reject); + request.end(); + }); + + return { + close() { + responseRef?.destroy(); + request.destroy(); + }, + async nextEvent() { + const queued = eventQueue.shift(); + if (queued !== undefined) { + return queued; + } + + return await new Promise((resolve) => { + waitingResolvers.push(resolve); + }); + }, + }; +} + +function parseServerSentEvent( + payload: string, +): { event: string; data: string } | null { + const lines = payload + .split("\n") + .map((line) => line.trim()) + .filter((line) => line !== ""); + const eventLine = lines.find((line) => line.startsWith("event: ")); + const dataLine = lines.find((line) => line.startsWith("data: ")); + if (eventLine === undefined || dataLine === undefined) { + return null; + } + + return { + event: eventLine.slice("event: ".length), + data: dataLine.slice("data: ".length), + }; +} diff --git a/tests/orchestrator/core.test.ts b/tests/orchestrator/core.test.ts index 48ade6fb..3ee89d4a 100644 --- a/tests/orchestrator/core.test.ts +++ b/tests/orchestrator/core.test.ts @@ -567,6 +567,11 @@ function createConfig(overrides?: { server: { port: null, }, + observability: { + dashboardEnabled: true, + refreshMs: 1_000, + renderIntervalMs: 16, + }, }; } diff --git a/tests/orchestrator/runtime-host.test.ts b/tests/orchestrator/runtime-host.test.ts index 5e281c1b..5385860c 100644 --- a/tests/orchestrator/runtime-host.test.ts +++ b/tests/orchestrator/runtime-host.test.ts @@ -397,5 +397,10 @@ function createConfig(): ResolvedWorkflowConfig { server: { port: null, }, + observability: { + dashboardEnabled: true, + refreshMs: 1_000, + renderIntervalMs: 16, + }, }; } From 5e1b56009a8157b53b9782d25d65f11717bd9b3c Mon Sep 17 00:00:00 2001 From: octane0411 Date: Sat, 7 Mar 2026 14:55:59 +0800 Subject: [PATCH 2/2] Format live dashboard files --- src/observability/dashboard-server.ts | 12 +++++++----- tests/observability/dashboard-server.test.ts | 5 +++-- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/src/observability/dashboard-server.ts b/src/observability/dashboard-server.ts index f052bc5b..55d01490 100644 --- a/src/observability/dashboard-server.ts +++ b/src/observability/dashboard-server.ts @@ -214,15 +214,15 @@ class DashboardLiveUpdatesController { return; } - await Promise.allSettled(clients.map((client) => this.writeSnapshot(client))); + await Promise.allSettled( + clients.map((client) => this.writeSnapshot(client)), + ); } private async writeSnapshot(response: ServerResponse): Promise { try { const snapshot = await readSnapshot(this.#host, this.#snapshotTimeoutMs); - response.write( - `event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`, - ); + response.write(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`); } catch (error) { response.write( `event: error\ndata: ${JSON.stringify({ @@ -637,7 +637,9 @@ function renderDashboardHtml( }"> ${ - options.liveUpdatesEnabled ? "Live updates connected" : "Static snapshot" + options.liveUpdatesEnabled + ? "Live updates connected" + : "Static snapshot" } diff --git a/tests/observability/dashboard-server.test.ts b/tests/observability/dashboard-server.test.ts index 87f97ea2..4777117d 100644 --- a/tests/observability/dashboard-server.test.ts +++ b/tests/observability/dashboard-server.test.ts @@ -473,8 +473,9 @@ async function openEventStream( nextEvent(): Promise<{ event: string; data: string }>; }> { const eventQueue: Array<{ event: string; data: string }> = []; - const waitingResolvers: Array<(value: { event: string; data: string }) => void> = - []; + const waitingResolvers: Array< + (value: { event: string; data: string }) => void + > = []; let buffer = ""; let responseRef: IncomingMessage | null = null;