From 3e1842ba59e404761f8f8c3ab4c83c78e0a0201b Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Wed, 15 Jul 2026 23:10:11 -0600 Subject: [PATCH 01/12] feat: add service resources dashboard --- apps/server/src/agents/activity-monitor.ts | 11 +- .../server/src/agents/diff-stats-refresher.ts | 43 +- apps/server/src/jobs/service.ts | 7 + .../src/observability/service-resources.ts | 765 ++++++++++++++++++ .../src/observability/subsystem-tracker.ts | 180 +++++ apps/server/src/release-auto-check.ts | 11 + apps/server/src/routes/resources.ts | 22 + apps/server/src/server.ts | 98 +++ .../src/server/agent-lifecycle-runtime.ts | 11 +- apps/server/src/server/ui-events.ts | 16 + apps/server/src/shared/git/base-ref.ts | 17 +- apps/server/src/shared/git/diff-stats.ts | 15 +- apps/server/src/stream-manager.ts | 22 + .../server/src/terminal/copy-mode-observer.ts | 29 +- apps/server/test/diff-stats-refresher.test.ts | 12 + apps/server/test/diff-stats.test.ts | 7 +- apps/server/test/service-resources.test.ts | 150 ++++ apps/server/test/subsystem-tracker.test.ts | 110 +++ apps/server/test/system-routes.test.ts | 36 + .../app/service-resources-chart-config.ts | 13 + .../app/service-resources-chart.tsx | 144 ++++ .../app/service-resources-dashboard.tsx | 241 ++++++ .../app/service-resources-format.ts | 46 ++ .../app/service-resources-settings.tsx | 109 +++ .../app/service-resources-subsystems.tsx | 123 +++ apps/web/src/components/app/settings-pane.tsx | 2 + apps/web/src/components/app/settings-state.ts | 4 + apps/web/src/hooks/use-service-resources.ts | 129 +++ docs/service-resources-dashboard-plan.md | 476 +++++++++++ e2e/settings.spec.ts | 36 + 30 files changed, 2868 insertions(+), 17 deletions(-) create mode 100644 apps/server/src/observability/service-resources.ts create mode 100644 apps/server/src/observability/subsystem-tracker.ts create mode 100644 apps/server/src/routes/resources.ts create mode 100644 apps/server/test/service-resources.test.ts create mode 100644 apps/server/test/subsystem-tracker.test.ts create mode 100644 apps/web/src/components/app/service-resources-chart-config.ts create mode 100644 apps/web/src/components/app/service-resources-chart.tsx create mode 100644 apps/web/src/components/app/service-resources-dashboard.tsx create mode 100644 apps/web/src/components/app/service-resources-format.ts create mode 100644 apps/web/src/components/app/service-resources-settings.tsx create mode 100644 apps/web/src/components/app/service-resources-subsystems.tsx create mode 100644 apps/web/src/hooks/use-service-resources.ts create mode 100644 docs/service-resources-dashboard-plan.md diff --git a/apps/server/src/agents/activity-monitor.ts b/apps/server/src/agents/activity-monitor.ts index 73155440..e079a1af 100644 --- a/apps/server/src/agents/activity-monitor.ts +++ b/apps/server/src/agents/activity-monitor.ts @@ -44,7 +44,7 @@ export type ActivityMonitorDeps = { export type ActivityMonitor = { /** Run one activity-check pass across all running agents. */ - check(): Promise; + check(): Promise<{ agentsScanned: number; corrections: number }>; /** Drop tracked state for an agent (e.g. on stop/archive). */ forget(agentId: string): void; }; @@ -71,8 +71,9 @@ export function createActivityMonitor( const state = new Map(); return { - async check(): Promise { + async check(): Promise<{ agentsScanned: number; corrections: number }> { const { pool, logger } = deps; + let corrections = 0; const result = await pool.query( `SELECT id, @@ -132,6 +133,8 @@ export function createActivityMonitor( { agentId: row.id }, "Activity monitor: correction skipped — event was updated concurrently" ); + } else { + corrections += 1; } } else if (!paneChanged && eventType === "working") { const staleDurationMs = now - prev.lastChangeAt; @@ -151,6 +154,8 @@ export function createActivityMonitor( { agentId: row.id }, "Activity monitor: correction skipped — event was updated concurrently" ); + } else { + corrections += 1; } } } @@ -168,6 +173,8 @@ export function createActivityMonitor( state.delete(id); } } + + return { agentsScanned: result.rows.length, corrections }; }, forget(agentId: string): void { diff --git a/apps/server/src/agents/diff-stats-refresher.ts b/apps/server/src/agents/diff-stats-refresher.ts index 16367432..9e37f9a3 100644 --- a/apps/server/src/agents/diff-stats-refresher.ts +++ b/apps/server/src/agents/diff-stats-refresher.ts @@ -2,6 +2,7 @@ import { getDiffStats as defaultGetDiffStats, type DiffStats, } from "../shared/git/diff-stats.js"; +import type { SubsystemTracker } from "../observability/subsystem-tracker.js"; export type DiffStatsAgent = { worktreePath: string | null; @@ -36,6 +37,7 @@ export type DiffStatsRefresherOptions = { computeDiffStats?: ComputeDiffStats; freshnessMs?: number; logger?: WarnLogger; + tracker?: SubsystemTracker; }; const DEFAULT_FRESHNESS_MS = 3_000; @@ -61,13 +63,28 @@ export class DiffStatsRefresher { private readonly computeDiffStats: ComputeDiffStats; private readonly freshnessMs: number; private readonly logger: WarnLogger | null; + private readonly tracker: SubsystemTracker | null; + private signals = 0; + private dedupedSignals = 0; constructor(options: DiffStatsRefresherOptions) { this.getAgent = options.getAgent; this.publishEvent = options.publishEvent; - this.computeDiffStats = options.computeDiffStats ?? defaultGetDiffStats; + this.computeDiffStats = + options.computeDiffStats ?? + (async (worktreePath, baseRef) => { + let commandError: unknown = null; + const stats = await defaultGetDiffStats(worktreePath, baseRef, { + onError: (error) => { + commandError = error; + }, + }); + if (commandError !== null) throw commandError; + return stats; + }); this.freshnessMs = options.freshnessMs ?? DEFAULT_FRESHNESS_MS; this.logger = options.logger ?? null; + this.tracker = options.tracker ?? null; } /** @@ -75,8 +92,12 @@ export class DiffStatsRefresher { * still warm; shares the in-flight promise when one is running. */ signal(agentId: string): Promise { + this.signals += 1; const existing = this.inFlight.get(agentId); - if (existing) return existing; + if (existing) { + this.dedupedSignals += 1; + return existing; + } const last = this.lastSignaledAt.get(agentId) ?? 0; const now = Date.now(); @@ -102,6 +123,20 @@ export class DiffStatsRefresher { return this.cache.get(agentId) ?? null; } + getMetrics(): { + cacheEntries: number; + inFlight: number; + signals: number; + dedupedSignals: number; + } { + return { + cacheEntries: this.cache.size, + inFlight: this.inFlight.size, + signals: this.signals, + dedupedSignals: this.dedupedSignals, + }; + } + /** * Drop any cached state for an agent (archive/delete cleanup). */ @@ -112,6 +147,7 @@ export class DiffStatsRefresher { } private async refresh(agentId: string): Promise { + const trackedRun = this.tracker?.start(); let nextStats: DiffStats | null = null; try { const agent = await this.getAgent(agentId); @@ -139,6 +175,7 @@ export class DiffStatsRefresher { nextStats = await this.computeDiffStats(path, baseRef); } } catch (err) { + trackedRun?.fail(err); this.logger?.warn( { err, agentId }, "Diff stats refresh failed; leaving cache unchanged" @@ -146,6 +183,8 @@ export class DiffStatsRefresher { return; } + trackedRun?.succeed({ files: nextStats?.files ?? 0 }); + const previous = this.cache.has(agentId) ? this.cache.get(agentId) : undefined; diff --git a/apps/server/src/jobs/service.ts b/apps/server/src/jobs/service.ts index 59a61894..c1976688 100644 --- a/apps/server/src/jobs/service.ts +++ b/apps/server/src/jobs/service.ts @@ -109,6 +109,13 @@ export class JobService { this.onRunStateChangeCallbacks.push(cb); } + getRuntimeMetrics(): { scheduledJobs: number; activeMonitors: number } { + return { + scheduledJobs: this.schedulers.size, + activeMonitors: this.monitors.size, + }; + } + private emitRunStateChange(run: JobRunRecord): void { for (const cb of this.onRunStateChangeCallbacks) { try { diff --git a/apps/server/src/observability/service-resources.ts b/apps/server/src/observability/service-resources.ts new file mode 100644 index 00000000..3673e78d --- /dev/null +++ b/apps/server/src/observability/service-resources.ts @@ -0,0 +1,765 @@ +import os from "node:os"; +import { monitorEventLoopDelay } from "node:perf_hooks"; + +import type { Pool } from "pg"; + +import { runCommand } from "../shared/lib/run-command.js"; +import type { + SubsystemHealthState, + SubsystemSnapshot, + SubsystemTracker, +} from "./subsystem-tracker.js"; + +const SAMPLE_INTERVAL_MS = 5_000; +const EXTERNAL_SAMPLE_INTERVAL_MS = 10_000; +const MAX_SAMPLES = 720; +const HTTP_WINDOW_MS = 60_000; +const HTTP_BUCKET_MS = 5_000; +const HTTP_BUCKET_COUNT = HTTP_WINDOW_MS / HTTP_BUCKET_MS; +const MAX_HTTP_DURATIONS_PER_BUCKET = 128; +const DATABASE_PROBE_TIMEOUT_MS = 3_000; + +export type ResourceSample = { + at: number; + serverCpuPercent: number; + serverRssBytes: number; + serverHeapBytes: number; + agentCpuPercent: number | null; + agentRssBytes: number | null; + hostLoad1: number; +}; + +type HttpBucket = { + startedAt: number; + requests: number; + errors: number; + durationsMs: number[]; +}; + +export type HttpRequestToken = { + startedAt: number; + finished: boolean; +}; + +export type WorkloadSnapshot = { + runningAgents: number; + sseClients: number; + streams: number; + streamViewers: number; + terminalObservers: number; + terminalViewers: number; + scheduledJobs: number; + jobMonitors: number; + gitRefreshesInFlight: number; + uiEventsPublished: number; + uiWriteFailures: number; + terminalPolls: number; + terminalPollFailures: number; +}; + +export type ServiceResourcesDeps = { + pool: Pool; + listAgentSessions: () => Promise>; + getWorkloads: () => WorkloadSnapshot; + subsystemTrackers: SubsystemTracker[]; + processTreeSupported?: boolean; +}; + +type AgentProcessSnapshot = { + supported: boolean; + cpuPercent: number | null; + rssBytes: number | null; + processCount: number | null; + sampledAt: number | null; + error: string | null; +}; + +export type ServiceResourcesResponse = { + generatedAt: number; + processStartedAt: number; + availableHistoryMs: number; + sampleIntervalMs: number; + overall: { + state: "healthy" | "degraded" | "unavailable" | "unknown"; + reasons: Array<{ code: string; message: string }>; + }; + capabilities: { + processTreeMetrics: "available" | "unsupported" | "error"; + eventLoopMetrics: "available"; + }; + current: { + server: { + cpuPercent: number; + rssBytes: number; + heapUsedBytes: number; + heapTotalBytes: number; + externalBytes: number; + uptimeSeconds: number; + }; + host: { + load1: number; + load5: number; + load15: number; + cpuCount: number; + totalMemoryBytes: number; + freeMemoryBytes: number; + }; + agents: AgentProcessSnapshot; + database: { + state: "healthy" | "unavailable" | "unknown"; + latencyMs: number | null; + sampledAt: number | null; + pool: { total: number; idle: number; waiting: number; max: number }; + }; + eventLoop: { p95DelayMs: number }; + http: { + requestsPerMinute: number; + inFlight: number; + errorRatePercent: number; + p95DurationMs: number | null; + }; + workloads: WorkloadSnapshot; + }; + subsystems: SubsystemSnapshot[]; + series: ResourceSample[]; +}; + +function round(value: number, digits = 1): number { + const factor = 10 ** digits; + return Math.round(value * factor) / factor; +} + +function percentile95(values: number[]): number | null { + if (values.length === 0) return null; + const sorted = [...values].sort((a, b) => a - b); + return sorted[Math.max(0, Math.ceil(sorted.length * 0.95) - 1)] ?? null; +} + +function ownerSubsystemState(input: { + active: number; + lastSucceededAt: number | null; + lastFailedAt: number | null; +}): SubsystemHealthState { + if ( + input.lastFailedAt !== null && + (input.lastSucceededAt === null || + input.lastFailedAt >= input.lastSucceededAt) + ) { + return "degraded"; + } + if (input.lastSucceededAt !== null) return "healthy"; + return input.active > 0 ? "running" : "idle"; +} + +function operationalSubsystem(input: { + id: string; + label: string; + description: string; + state: SubsystemHealthState; + runs?: number; + failures?: number; + lastDurationMs?: number | null; + metadata?: Record; +}): SubsystemSnapshot { + return { + id: input.id, + label: input.label, + description: input.description, + state: input.state, + statusReason: input.state === "degraded" ? "failure" : null, + expectedCadenceMs: null, + lastStartedAt: null, + lastCompletedAt: null, + lastSucceededAt: null, + lastFailedAt: null, + lastDurationMs: input.lastDurationMs ?? null, + p95DurationMs: input.lastDurationMs ?? null, + inFlight: 0, + runs: input.runs ?? 0, + failures: input.failures ?? 0, + lastError: null, + metadata: input.metadata ?? {}, + }; +} + +export class ServiceResources { + private readonly eventLoopDelay = monitorEventLoopDelay({ resolution: 20 }); + private samples: ResourceSample[] = []; + private timer: NodeJS.Timeout | null = null; + private running = false; + private generation = 0; + private previousCpu = process.cpuUsage(); + private previousCpuAt = performance.now(); + private currentCpuPercent = 0; + private lastExternalSampleAt = 0; + private agentProcesses: AgentProcessSnapshot; + private database: ServiceResourcesResponse["current"]["database"]; + private databaseProbe: Promise< + ServiceResourcesResponse["current"]["database"] + > | null = null; + private httpInFlight = 0; + private httpBuckets: HttpBucket[] = []; + private runningAgentCount = 0; + private workloads: WorkloadSnapshot; + private previousOwnerCounters: { + uiEventsPublished: number; + uiWriteFailures: number; + terminalPolls: number; + terminalPollFailures: number; + } | null = null; + private ownerHealth = { + uiEvents: { + lastSucceededAt: null as number | null, + lastFailedAt: null as number | null, + }, + terminalObservers: { + lastSucceededAt: null as number | null, + lastFailedAt: null as number | null, + }, + }; + + constructor(private readonly deps: ServiceResourcesDeps) { + this.agentProcesses = { + supported: + deps.processTreeSupported ?? + (process.platform === "darwin" || process.platform === "linux"), + cpuPercent: null, + rssBytes: null, + processCount: null, + sampledAt: null, + error: null, + }; + this.workloads = { ...this.deps.getWorkloads(), runningAgents: 0 }; + this.database = { + state: "unknown", + latencyMs: null, + sampledAt: null, + pool: this.poolSnapshot(), + }; + } + + start(): void { + if (this.running) return; + this.running = true; + const generation = ++this.generation; + this.previousCpu = process.cpuUsage(); + this.previousCpuAt = performance.now(); + this.eventLoopDelay.enable(); + void this.runSampleLoop(generation); + } + + stop(): void { + this.running = false; + this.generation += 1; + if (this.timer) clearTimeout(this.timer); + this.timer = null; + this.eventLoopDelay.disable(); + } + + requestStarted(): HttpRequestToken { + this.httpInFlight += 1; + return { startedAt: performance.now(), finished: false }; + } + + requestFinished(token: HttpRequestToken, statusCode: number): void { + if (token.finished) return; + token.finished = true; + this.httpInFlight = Math.max(0, this.httpInFlight - 1); + const now = Date.now(); + const bucket = this.getHttpBucket(now); + bucket.requests += 1; + if (statusCode >= 500) bucket.errors += 1; + bucket.durationsMs.push(Math.max(0, performance.now() - token.startedAt)); + if (bucket.durationsMs.length > MAX_HTTP_DURATIONS_PER_BUCKET) { + bucket.durationsMs.splice( + 0, + bucket.durationsMs.length - MAX_HTTP_DURATIONS_PER_BUCKET + ); + } + } + + getHttpObservationStorageSize(): number { + return this.httpBuckets.reduce( + (sum, bucket) => sum + bucket.durationsMs.length, + 0 + ); + } + + getSnapshot(windowMs = 60 * 60 * 1000): ServiceResourcesResponse { + const now = Date.now(); + const memory = process.memoryUsage(); + const [load1, load5, load15] = os.loadavg(); + const httpBuckets = this.getActiveHttpBuckets(now); + const requestCount = httpBuckets.reduce( + (sum, bucket) => sum + bucket.requests, + 0 + ); + const errors = httpBuckets.reduce((sum, bucket) => sum + bucket.errors, 0); + const durations = httpBuckets.flatMap((bucket) => bucket.durationsMs); + const trackedSubsystems = this.deps.subsystemTrackers.map((tracker) => + tracker.snapshot(now) + ); + const workloads = { ...this.workloads }; + const httpP95 = percentile95(durations); + const operationalSubsystems = [ + operationalSubsystem({ + id: "api-server", + label: "API server", + description: "Handles authenticated HTTP, SSE, and WebSocket traffic.", + state: errors > 0 ? "degraded" : "healthy", + runs: requestCount, + failures: errors, + lastDurationMs: httpP95, + metadata: { + requestsPerMinute: requestCount, + inFlight: this.httpInFlight, + }, + }), + operationalSubsystem({ + id: "database", + label: "Database", + description: "PostgreSQL connectivity and connection-pool capacity.", + state: + this.database.state === "unavailable" + ? "degraded" + : this.database.state === "unknown" + ? "unknown" + : this.database.pool.waiting > 0 + ? "degraded" + : "healthy", + lastDurationMs: this.database.latencyMs, + metadata: { + poolTotal: this.database.pool.total, + poolIdle: this.database.pool.idle, + poolWaiting: this.database.pool.waiting, + }, + }), + operationalSubsystem({ + id: "job-schedulers", + label: "Job schedulers", + description: "Cron schedules and monitors for active automation runs.", + state: + workloads.scheduledJobs > 0 || workloads.jobMonitors > 0 + ? "healthy" + : "idle", + metadata: { + scheduledJobs: workloads.scheduledJobs, + activeMonitors: workloads.jobMonitors, + }, + }), + operationalSubsystem({ + id: "ui-event-stream", + label: "UI event stream", + description: "Connected browser clients receiving server-sent events.", + state: ownerSubsystemState({ + active: workloads.sseClients, + ...this.ownerHealth.uiEvents, + }), + runs: workloads.uiEventsPublished, + failures: workloads.uiWriteFailures, + metadata: { + clients: workloads.sseClients, + eventsPublished: workloads.uiEventsPublished, + writeFailures: workloads.uiWriteFailures, + }, + }), + operationalSubsystem({ + id: "terminal-observers", + label: "Terminal observers", + description: "Viewer-driven terminal copy-mode observation.", + state: ownerSubsystemState({ + active: workloads.terminalObservers, + ...this.ownerHealth.terminalObservers, + }), + runs: workloads.terminalPolls, + failures: workloads.terminalPollFailures, + metadata: { + observers: workloads.terminalObservers, + viewers: workloads.terminalViewers, + polls: workloads.terminalPolls, + pollFailures: workloads.terminalPollFailures, + }, + }), + ]; + const subsystems = [...operationalSubsystems, ...trackedSubsystems]; + const reasons: Array<{ code: string; message: string }> = []; + + if (this.database.state === "unavailable") { + reasons.push({ + code: "DB_PROBE_FAILED", + message: "The latest database probe failed.", + }); + } + if (this.database.pool.waiting > 0) { + reasons.push({ + code: "DB_POOL_WAITING", + message: `${this.database.pool.waiting} database request${this.database.pool.waiting === 1 ? " is" : "s are"} waiting for a connection.`, + }); + } + const delayed = subsystems.filter((item) => item.state === "degraded"); + if (delayed.length > 0) { + reasons.push({ + code: "SUBSYSTEM_DEGRADED", + message: `${delayed.map((item) => item.label).join(", ")} ${delayed.length === 1 ? "needs" : "need"} attention.`, + }); + } + const eventLoopP95 = round(this.eventLoopDelay.percentile(95) / 1e6, 1); + if (eventLoopP95 > 100) { + reasons.push({ + code: "EVENT_LOOP_DELAY_HIGH", + message: `Event-loop p95 delay is ${eventLoopP95} ms.`, + }); + } + + const overallState = + this.database.state === "unavailable" + ? "unavailable" + : reasons.length > 0 + ? "degraded" + : this.samples.length === 0 + ? "unknown" + : "healthy"; + + const series = this.samples.filter((sample) => now - sample.at <= windowMs); + + return { + generatedAt: now, + processStartedAt: now - process.uptime() * 1000, + availableHistoryMs: + series.length > 1 ? series[series.length - 1]!.at - series[0]!.at : 0, + sampleIntervalMs: SAMPLE_INTERVAL_MS, + overall: { state: overallState, reasons }, + capabilities: { + processTreeMetrics: !this.agentProcesses.supported + ? "unsupported" + : this.agentProcesses.error + ? "error" + : "available", + eventLoopMetrics: "available", + }, + current: { + server: { + cpuPercent: this.currentCpuPercent, + rssBytes: memory.rss, + heapUsedBytes: memory.heapUsed, + heapTotalBytes: memory.heapTotal, + externalBytes: memory.external, + uptimeSeconds: process.uptime(), + }, + host: { + load1, + load5, + load15, + cpuCount: os.cpus().length, + totalMemoryBytes: os.totalmem(), + freeMemoryBytes: os.freemem(), + }, + agents: { ...this.agentProcesses }, + database: { ...this.database, pool: this.poolSnapshot() }, + eventLoop: { p95DelayMs: eventLoopP95 }, + http: { + requestsPerMinute: requestCount, + inFlight: this.httpInFlight, + errorRatePercent: + requestCount > 0 ? round((errors / requestCount) * 100, 1) : 0, + p95DurationMs: httpP95, + }, + workloads, + }, + subsystems, + series, + }; + } + + private async runSampleLoop(generation: number): Promise { + try { + await this.sample(generation); + } catch { + // Sampling is best-effort; the last committed snapshot remains valid. + } finally { + if (!this.isActive(generation)) return; + this.timer = setTimeout(() => { + this.timer = null; + void this.runSampleLoop(generation); + }, SAMPLE_INTERVAL_MS); + this.timer.unref?.(); + } + } + + private isActive(generation: number): boolean { + return this.running && this.generation === generation; + } + + private async sample(generation: number): Promise { + const now = Date.now(); + const cpuNow = process.cpuUsage(); + const wallNow = performance.now(); + const cpuMicros = + cpuNow.user - + this.previousCpu.user + + cpuNow.system - + this.previousCpu.system; + const wallMicros = Math.max(1, (wallNow - this.previousCpuAt) * 1000); + const currentCpuPercent = round((cpuMicros / wallMicros) * 100, 1); + let database = this.database; + let agentProcesses = this.agentProcesses; + let runningAgentCount = this.runningAgentCount; + let lastExternalSampleAt = this.lastExternalSampleAt; + + if (now - this.lastExternalSampleAt >= EXTERNAL_SAMPLE_INTERVAL_MS) { + const [databaseResult, agentResult] = await Promise.all([ + this.sampleDatabase(), + this.sampleAgentProcesses(), + ]); + database = databaseResult; + agentProcesses = agentResult.processes; + runningAgentCount = agentResult.runningAgentCount; + lastExternalSampleAt = now; + } + + if (!this.isActive(generation)) return; + + const workloads = { + ...this.deps.getWorkloads(), + runningAgents: runningAgentCount, + }; + this.updateOwnerHealth(workloads, now); + this.workloads = workloads; + this.currentCpuPercent = currentCpuPercent; + this.previousCpu = cpuNow; + this.previousCpuAt = wallNow; + this.database = database; + this.agentProcesses = agentProcesses; + this.runningAgentCount = runningAgentCount; + this.lastExternalSampleAt = lastExternalSampleAt; + + const memory = process.memoryUsage(); + this.samples.push({ + at: now, + serverCpuPercent: currentCpuPercent, + serverRssBytes: memory.rss, + serverHeapBytes: memory.heapUsed, + agentCpuPercent: agentProcesses.cpuPercent, + agentRssBytes: agentProcesses.rssBytes, + hostLoad1: os.loadavg()[0], + }); + if (this.samples.length > MAX_SAMPLES) { + this.samples = this.samples.slice(-MAX_SAMPLES); + } + } + + private async sampleDatabase(): Promise< + ServiceResourcesResponse["current"]["database"] + > { + if (!this.databaseProbe) { + const started = performance.now(); + const probe = this.deps.pool + .query("SELECT 1") + .then(() => ({ + state: "healthy" as const, + latencyMs: round(performance.now() - started, 1), + sampledAt: Date.now(), + pool: this.poolSnapshot(), + })) + .catch(() => ({ + state: "unavailable" as const, + latencyMs: null, + sampledAt: Date.now(), + pool: this.poolSnapshot(), + })); + this.databaseProbe = probe; + void probe.finally(() => { + if (this.databaseProbe === probe) this.databaseProbe = null; + }); + } + + const probe = this.databaseProbe; + return new Promise((resolve) => { + const timeout = setTimeout(() => { + resolve({ + state: "unavailable", + latencyMs: null, + sampledAt: Date.now(), + pool: this.poolSnapshot(), + }); + }, DATABASE_PROBE_TIMEOUT_MS); + void probe.then((snapshot) => { + clearTimeout(timeout); + resolve(snapshot); + }); + }); + } + + private poolSnapshot() { + return { + total: this.deps.pool.totalCount, + idle: this.deps.pool.idleCount, + waiting: this.deps.pool.waitingCount, + max: Number(this.deps.pool.options.max ?? 10), + }; + } + + private async sampleAgentProcesses(): Promise<{ + processes: AgentProcessSnapshot; + runningAgentCount: number; + }> { + try { + const agents = await this.deps.listAgentSessions(); + const runningAgentCount = agents.length; + if (!this.agentProcesses.supported) { + return { + processes: { ...this.agentProcesses, error: null }, + runningAgentCount, + }; + } + const sessions = new Set( + agents + .map((agent) => agent.tmuxSession?.trim()) + .filter((value): value is string => Boolean(value)) + ); + if (sessions.size === 0) { + return { + processes: { + supported: true, + cpuPercent: 0, + rssBytes: 0, + processCount: 0, + sampledAt: Date.now(), + error: null, + }, + runningAgentCount, + }; + } + + const [panes, processes] = await Promise.all([ + runCommand( + "tmux", + ["list-panes", "-a", "-F", "#{session_name}\t#{pane_pid}"], + { allowedExitCodes: [0, 1], timeoutMs: 3_000 } + ), + runCommand("ps", ["-axo", "pid=,ppid=,%cpu=,rss="], { + timeoutMs: 3_000, + }), + ]); + + const roots = new Set(); + for (const line of panes.stdout.split("\n")) { + const [session, pidText] = line.trim().split("\t"); + const pid = Number(pidText); + if (session && sessions.has(session) && Number.isFinite(pid)) { + roots.add(pid); + } + } + + const rows = processes.stdout + .split("\n") + .map((line) => line.trim().split(/\s+/)) + .filter((parts) => parts.length >= 4) + .map(([pid, ppid, cpu, rss]) => ({ + pid: Number(pid), + ppid: Number(ppid), + cpu: Number(cpu), + rss: Number(rss), + })) + .filter((row) => Number.isFinite(row.pid) && Number.isFinite(row.ppid)); + const included = new Set(roots); + let changed = true; + while (changed) { + changed = false; + for (const row of rows) { + if (!included.has(row.pid) && included.has(row.ppid)) { + included.add(row.pid); + changed = true; + } + } + } + const owned = rows.filter((row) => included.has(row.pid)); + return { + processes: { + supported: true, + cpuPercent: round( + owned.reduce((sum, row) => sum + (row.cpu || 0), 0), + 1 + ), + rssBytes: owned.reduce((sum, row) => sum + (row.rss || 0) * 1024, 0), + processCount: owned.length, + sampledAt: Date.now(), + error: null, + }, + runningAgentCount, + }; + } catch { + return { + processes: { + ...this.agentProcesses, + sampledAt: Date.now(), + error: "Process sampling failed", + }, + runningAgentCount: this.runningAgentCount, + }; + } + } + + private updateOwnerHealth(workloads: WorkloadSnapshot, now: number): void { + const previous = this.previousOwnerCounters; + if (previous) { + const published = Math.max( + 0, + workloads.uiEventsPublished - previous.uiEventsPublished + ); + const writeFailures = Math.max( + 0, + workloads.uiWriteFailures - previous.uiWriteFailures + ); + if (writeFailures > 0) { + this.ownerHealth.uiEvents.lastFailedAt = now; + } else if (published > 0) { + this.ownerHealth.uiEvents.lastSucceededAt = now; + } + + const polls = Math.max( + 0, + workloads.terminalPolls - previous.terminalPolls + ); + const pollFailures = Math.max( + 0, + workloads.terminalPollFailures - previous.terminalPollFailures + ); + if (pollFailures > 0) { + this.ownerHealth.terminalObservers.lastFailedAt = now; + } else if (polls > 0) { + this.ownerHealth.terminalObservers.lastSucceededAt = now; + } + } + this.previousOwnerCounters = { + uiEventsPublished: workloads.uiEventsPublished, + uiWriteFailures: workloads.uiWriteFailures, + terminalPolls: workloads.terminalPolls, + terminalPollFailures: workloads.terminalPollFailures, + }; + } + + private getHttpBucket(now: number): HttpBucket { + const startedAt = Math.floor(now / HTTP_BUCKET_MS) * HTTP_BUCKET_MS; + let bucket = this.httpBuckets.find((item) => item.startedAt === startedAt); + if (!bucket) { + bucket = { startedAt, requests: 0, errors: 0, durationsMs: [] }; + this.httpBuckets.push(bucket); + } + this.httpBuckets = this.getActiveHttpBuckets(now); + return bucket; + } + + private getActiveHttpBuckets(now: number): HttpBucket[] { + const currentBucket = Math.floor(now / HTTP_BUCKET_MS) * HTTP_BUCKET_MS; + const oldestBucket = + currentBucket - (HTTP_BUCKET_COUNT - 1) * HTTP_BUCKET_MS; + return this.httpBuckets.filter( + (bucket) => bucket.startedAt >= oldestBucket + ); + } +} + +export function isSubsystemDegraded(state: SubsystemHealthState): boolean { + return state === "degraded"; +} diff --git a/apps/server/src/observability/subsystem-tracker.ts b/apps/server/src/observability/subsystem-tracker.ts new file mode 100644 index 00000000..41861d1a --- /dev/null +++ b/apps/server/src/observability/subsystem-tracker.ts @@ -0,0 +1,180 @@ +export type SubsystemHealthState = + | "healthy" + | "degraded" + | "running" + | "idle" + | "disabled" + | "unknown"; + +export type SubsystemSnapshot = { + id: string; + label: string; + description: string; + state: SubsystemHealthState; + statusReason: "failure" | "stale" | "stuck" | null; + expectedCadenceMs: number | null; + lastStartedAt: number | null; + lastCompletedAt: number | null; + lastSucceededAt: number | null; + lastFailedAt: number | null; + lastDurationMs: number | null; + p95DurationMs: number | null; + inFlight: number; + runs: number; + failures: number; + lastError: string | null; + metadata: Record; +}; + +export type SubsystemRun = { + succeed(metadata?: Record): void; + fail(error: unknown, metadata?: Record): void; +}; + +export type SubsystemTrackerOptions = { + id: string; + label: string; + description: string; + expectedCadenceMs?: number | null; +}; + +const MAX_DURATIONS = 60; + +function publicErrorSummary(error: unknown): string { + const raw = error instanceof Error ? error.message : String(error); + return /timed?\s*out|timeout|abort/i.test(raw) + ? "Operation timed out" + : "Operation failed"; +} + +export class SubsystemTracker { + private readonly options: Required; + private durations: number[] = []; + private activeStarts: number[] = []; + private inFlight = 0; + private runs = 0; + private failures = 0; + private disabled = false; + private lastStartedAt: number | null = null; + private lastCompletedAt: number | null = null; + private lastSucceededAt: number | null = null; + private lastFailedAt: number | null = null; + private lastDurationMs: number | null = null; + private lastError: string | null = null; + private metadata: Record = {}; + + constructor(options: SubsystemTrackerOptions) { + this.options = { + ...options, + expectedCadenceMs: options.expectedCadenceMs ?? null, + }; + } + + setDisabled(disabled: boolean): void { + this.disabled = disabled; + } + + start(): SubsystemRun { + const startedAt = Date.now(); + this.lastStartedAt = startedAt; + this.activeStarts.push(startedAt); + this.inFlight += 1; + let completed = false; + + const finish = ( + succeeded: boolean, + error: unknown, + metadata?: Record + ) => { + if (completed) return; + completed = true; + const completedAt = Date.now(); + const duration = Math.max(0, completedAt - startedAt); + const activeIndex = this.activeStarts.indexOf(startedAt); + if (activeIndex >= 0) this.activeStarts.splice(activeIndex, 1); + this.inFlight = Math.max(0, this.inFlight - 1); + this.runs += 1; + this.lastCompletedAt = completedAt; + this.lastDurationMs = duration; + this.durations.push(duration); + if (this.durations.length > MAX_DURATIONS) { + this.durations = this.durations.slice(-MAX_DURATIONS); + } + if (metadata) this.metadata = { ...metadata }; + + if (succeeded) { + this.lastSucceededAt = completedAt; + this.lastError = null; + } else { + this.failures += 1; + this.lastFailedAt = completedAt; + this.lastError = publicErrorSummary(error); + } + }; + + return { + succeed: (metadata) => finish(true, null, metadata), + fail: (error, metadata) => finish(false, error, metadata), + }; + } + + snapshot(now = Date.now()): SubsystemSnapshot { + let state: SubsystemHealthState; + let statusReason: SubsystemSnapshot["statusReason"] = null; + const oldestActiveStart = + this.activeStarts.length > 0 ? Math.min(...this.activeStarts) : null; + const activeRunIsStuck = + this.options.expectedCadenceMs !== null && + oldestActiveStart !== null && + now - oldestActiveStart > this.options.expectedCadenceMs * 2; + if (this.disabled) { + state = "disabled"; + } else if (activeRunIsStuck) { + state = "degraded"; + statusReason = "stuck"; + } else if (this.inFlight > 0) { + state = "running"; + } else if (this.runs === 0) { + state = this.options.expectedCadenceMs === null ? "idle" : "unknown"; + } else if ( + this.lastFailedAt !== null && + (this.lastSucceededAt === null || + this.lastFailedAt > this.lastSucceededAt) + ) { + state = "degraded"; + statusReason = "failure"; + } else if ( + this.options.expectedCadenceMs !== null && + this.lastSucceededAt !== null && + now - this.lastSucceededAt > this.options.expectedCadenceMs * 2 + ) { + state = "degraded"; + statusReason = "stale"; + } else { + state = "healthy"; + } + + const sorted = [...this.durations].sort((a, b) => a - b); + const p95Index = Math.max(0, Math.ceil(sorted.length * 0.95) - 1); + + return { + id: this.options.id, + label: this.options.label, + description: this.options.description, + state, + statusReason, + expectedCadenceMs: this.options.expectedCadenceMs, + lastStartedAt: this.lastStartedAt, + lastCompletedAt: this.lastCompletedAt, + lastSucceededAt: this.lastSucceededAt, + lastFailedAt: this.lastFailedAt, + lastDurationMs: this.lastDurationMs, + p95DurationMs: sorted.length > 0 ? sorted[p95Index] : null, + inFlight: this.inFlight, + runs: this.runs, + failures: this.failures, + lastError: this.lastError, + metadata: { ...this.metadata }, + }; + } +} diff --git a/apps/server/src/release-auto-check.ts b/apps/server/src/release-auto-check.ts index 2b49161e..cec41a29 100644 --- a/apps/server/src/release-auto-check.ts +++ b/apps/server/src/release-auto-check.ts @@ -8,6 +8,7 @@ import { type ReleaseInfoSnapshot, } from "./release-info.js"; import { pruneCacheExcept } from "./release-tarball-cache.js"; +import type { SubsystemTracker } from "./observability/subsystem-tracker.js"; export const AUTOMATIC_UPDATE_MODE_KEY = "automatic_update_mode"; export const AUTOMATIC_UPDATE_MODES = ["off", "check"] as const; @@ -67,6 +68,7 @@ export type AutoCheckRuntimeDeps = { * dismissed-by-tag localStorage atom). */ broadcast: AutoCheckBroadcaster; logger: Logger; + tracker?: SubsystemTracker; }; export type AutoCheckRuntime = ReturnType; @@ -130,27 +132,36 @@ export function createAutoCheckRuntime(deps: AutoCheckRuntimeDeps) { // before the first await observes the in-flight state and coalesces // onto the same promise. const promise = (async (): Promise => { + const trackedRun = deps.tracker?.start(); if (deps.isApplyInProgress()) { + trackedRun?.succeed({ skipped: 1 }); return { ok: "skipped", reason: "apply in progress" }; } const mode = await readAutomaticUpdateMode(deps.pool).catch( () => DEFAULT_MODE ); if (mode === "off") { + deps.tracker?.setDisabled(true); + trackedRun?.succeed({ skipped: 1 }); return { ok: "skipped", reason: "mode=off" }; } + deps.tracker?.setDisabled(false); deps.logger.info({ reason }, "auto-update: running release check"); const result = await computeReleaseInfo(deps.computeDeps, { logger: deps.logger, }); if (!result.ok) { + trackedRun?.fail(new Error(result.error)); deps.logger.warn( { error: result.error }, "auto-update: release check failed; keeping previous snapshot" ); return { ok: false, reason: result.error }; } + trackedRun?.succeed({ + updateAvailable: result.snapshot.updateAvailable ? 1 : 0, + }); snapshot = result.snapshot; emitBroadcast(); diff --git a/apps/server/src/routes/resources.ts b/apps/server/src/routes/resources.ts new file mode 100644 index 00000000..abeee2b7 --- /dev/null +++ b/apps/server/src/routes/resources.ts @@ -0,0 +1,22 @@ +import type { FastifyInstance } from "fastify"; + +import type { ServiceResources } from "../observability/service-resources.js"; + +export async function registerResourceRoutes( + app: FastifyInstance, + resources: ServiceResources +): Promise { + app.get("/api/v1/system/resources", async (request, reply) => { + const query = request.query as { window?: unknown }; + const windows: Record = { + "15m": 15 * 60 * 1000, + "1h": 60 * 60 * 1000, + }; + const requested = typeof query.window === "string" ? query.window : "1h"; + const windowMs = windows[requested]; + if (!windowMs) { + return reply.code(400).send({ error: 'window must be "15m" or "1h".' }); + } + return resources.getSnapshot(windowMs); + }); +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 51a49d5a..30f35e84 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -116,6 +116,7 @@ import { registerReleaseRoutes } from "./routes/release.js"; import { createAutoCheckRuntime } from "./release-auto-check.js"; import { registerStaticRoutes } from "./routes/static.js"; import { registerSystemRoutes } from "./routes/system.js"; +import { registerResourceRoutes } from "./routes/resources.js"; import { dateTruncTz, loadScopedActivityEvents, @@ -147,6 +148,11 @@ import { UiEventBroker, type UiEvent } from "./server/ui-events.js"; import { createActivityMonitor } from "./agents/activity-monitor.js"; import { createAutoRenamePrompter } from "./agents/auto-rename-prompter.js"; import { DiffStatsRefresher } from "./agents/diff-stats-refresher.js"; +import { SubsystemTracker } from "./observability/subsystem-tracker.js"; +import { + ServiceResources, + type HttpRequestToken, +} from "./observability/service-resources.js"; const config = loadConfig(); const app = Fastify({ @@ -159,6 +165,31 @@ const focusTracker = new FocusTracker(); const slackNotifier = new SlackNotifier(pool, app.log); slackNotifier.setFocusCheck((agentId) => focusTracker.isFocused(agentId)); const uiEventBroker = new UiEventBroker(); +const reconciliationTracker = new SubsystemTracker({ + id: "agent-reconciliation", + label: "Agent reconciliation", + description: + "Checks running agent sessions and corrects stale lifecycle state.", + expectedCadenceMs: 30_000, +}); +const activityTracker = new SubsystemTracker({ + id: "activity-monitor", + label: "Activity monitor", + description: "Compares agent-reported state with recent terminal activity.", + expectedCadenceMs: 30_000, +}); +const gitRefreshTracker = new SubsystemTracker({ + id: "git-diff-refreshes", + label: "Git diff refreshes", + description: + "Computes cached diff statistics when agent activity requests a refresh.", +}); +const updateCheckTracker = new SubsystemTracker({ + id: "update-checker", + label: "Update checker", + description: "Checks the configured release channel for Dispatch updates.", + expectedCadenceMs: 6 * 60 * 60 * 1000, +}); const diffStatsRefresher = new DiffStatsRefresher({ getAgent: async (id) => { const agent = await agentManager.getAgent(id); @@ -171,6 +202,7 @@ const diffStatsRefresher = new DiffStatsRefresher({ }, publishEvent: (event) => uiEventBroker.publish(event), logger: app.log, + tracker: gitRefreshTracker, }); agentManager.attachDiffStatsRefresher(diffStatsRefresher); const terminalTokenStore = new TerminalTokenStore(60_000); @@ -261,6 +293,7 @@ const autoCheckRuntime = createAutoCheckRuntime({ }); }, logger: app.log, + tracker: updateCheckTracker, }); const activityMonitor = createActivityMonitor({ @@ -286,7 +319,49 @@ const agentLifecycleRuntime = createAgentLifecycleRuntime({ activityMonitor, withStreamFlag, publishUiEvent: (event) => uiEventBroker.publish(event as UiEvent), + reconciliationTracker, + activityTracker, }); +const serviceResources = new ServiceResources({ + pool, + listAgentSessions: async () => { + const agents = await agentManager.listAgents(); + return agents + .filter((agent) => + ["creating", "running", "stopping"].includes(agent.status) + ) + .map((agent) => ({ tmuxSession: agent.tmuxSession })); + }, + getWorkloads: () => { + const streamMetrics = streamManager.getMetrics(); + const observerMetrics = copyModeObserverManager.getMetrics(); + const jobMetrics = jobService.getRuntimeMetrics(); + const gitMetrics = diffStatsRefresher.getMetrics(); + const uiMetrics = uiEventBroker.getMetrics(); + return { + runningAgents: 0, + sseClients: uiMetrics.clients, + streams: streamMetrics.streams, + streamViewers: streamMetrics.viewers, + terminalObservers: observerMetrics.observers, + terminalViewers: observerMetrics.viewers, + scheduledJobs: jobMetrics.scheduledJobs, + jobMonitors: jobMetrics.activeMonitors, + gitRefreshesInFlight: gitMetrics.inFlight, + uiEventsPublished: uiMetrics.eventsPublished, + uiWriteFailures: uiMetrics.writeFailures, + terminalPolls: observerMetrics.pollCount, + terminalPollFailures: observerMetrics.pollFailures, + }; + }, + subsystemTrackers: [ + reconciliationTracker, + activityTracker, + gitRefreshTracker, + updateCheckTracker, + ], +}); +const resourceRequestStarts = new WeakMap(); const notificationRuntime = createNotificationRuntime({ agentManager, jobService, @@ -389,6 +464,26 @@ async function registerRoutes() { return payload; }); + app.addHook("onRequest", async (request) => { + if (!request.url.startsWith("/api/")) return; + resourceRequestStarts.set(request, serviceResources.requestStarted()); + }); + const finishResourceRequest = (request: object, statusCode: number) => { + const token = resourceRequestStarts.get(request); + if (!token) return; + serviceResources.requestFinished(token, statusCode); + resourceRequestStarts.delete(request); + }; + app.addHook("onResponse", async (request, reply) => { + finishResourceRequest(request, reply.statusCode); + }); + app.addHook("onRequestAbort", async (request) => { + finishResourceRequest(request, 499); + }); + app.addHook("onTimeout", async (request) => { + finishResourceRequest(request, 504); + }); + // --------------------------------------------------------------------------- // Auth hook — runs before every /api/ route except auth + health endpoints // --------------------------------------------------------------------------- @@ -528,6 +623,7 @@ async function registerRoutes() { rewriteForColor: (color) => staticTheme.rewriteForColor(color as IconColor), publishUiEvent: (event) => uiEventBroker.publish(event as UiEvent), }); + await registerResourceRoutes(app, serviceResources); await registerBrainRoutes(app, { brainStore, @@ -696,6 +792,7 @@ export async function initializeApp(options?: { await runMigrations(); } config.authToken = await getOrCreateAuthToken(pool); + serviceResources.start(); const shouldReconcileState = options?.reconcileState ?? true; if (shouldReconcileState) { await agentManager.reconcileAgents(); @@ -762,6 +859,7 @@ async function cleanupAppResources(): Promise { agentLifecycleRuntime.stopReconcileLoop(); authRuntime.stopSessionCleanupTimer(); autoCheckRuntime.stopScheduler(); + serviceResources.stop(); notificationRuntime.clearPendingWebNotifications(); diff --git a/apps/server/src/server/agent-lifecycle-runtime.ts b/apps/server/src/server/agent-lifecycle-runtime.ts index 0ec871ef..8a3c722e 100644 --- a/apps/server/src/server/agent-lifecycle-runtime.ts +++ b/apps/server/src/server/agent-lifecycle-runtime.ts @@ -3,6 +3,7 @@ import type { FastifyBaseLogger } from "fastify"; import type { ActivityMonitor } from "../agents/activity-monitor.js"; import type { AgentManager, AgentRecord } from "../agents/manager.js"; import type { StreamManager } from "../stream-manager.js"; +import type { SubsystemTracker } from "../observability/subsystem-tracker.js"; type CreateAgentLifecycleRuntimeDeps = { agentManager: AgentManager; @@ -14,6 +15,8 @@ type CreateAgentLifecycleRuntimeDeps = { agent: T ) => T & { hasStream: boolean }; publishUiEvent: (event: unknown) => void; + reconciliationTracker?: SubsystemTracker; + activityTracker?: SubsystemTracker; }; export function createAgentLifecycleRuntime( @@ -97,6 +100,7 @@ export function createAgentLifecycleRuntime( }, async runAgentStatusReconciliation(): Promise { + const reconciliationRun = deps.reconciliationTracker?.start(); try { const reconciled = await agentManager.reconcileAgentStatuses(); for (const agent of reconciled) { @@ -147,16 +151,21 @@ export function createAgentLifecycleRuntime( }); } } + reconciliationRun?.succeed({ corrections: reconciled.length }); } catch (error) { + reconciliationRun?.fail(error); appLog.warn({ err: error }, "Agent status reconciliation failed."); } // Activity monitor: compare self-reported status against tmux pane // activity and auto-correct mismatches (runs on the same cadence). if (activityMonitor) { + const activityRun = deps.activityTracker?.start(); try { - await activityMonitor.check(); + const result = await activityMonitor.check(); + activityRun?.succeed(result); } catch (error) { + activityRun?.fail(error); appLog.warn({ err: error }, "Activity monitor check failed."); } } diff --git a/apps/server/src/server/ui-events.ts b/apps/server/src/server/ui-events.ts index d425b19b..21b29fea 100644 --- a/apps/server/src/server/ui-events.ts +++ b/apps/server/src/server/ui-events.ts @@ -64,6 +64,8 @@ export type UiEvent = export class UiEventBroker { private clients = new Set(); private nextId = 1; + private eventsPublished = 0; + private writeFailures = 0; subscribe(stream: NodeJS.WritableStream): () => void { this.clients.add(stream); @@ -77,9 +79,22 @@ export class UiEventBroker { } publish(event: UiEvent): void { + this.eventsPublished += 1; this.write(event); } + getMetrics(): { + clients: number; + eventsPublished: number; + writeFailures: number; + } { + return { + clients: this.clients.size, + eventsPublished: this.eventsPublished, + writeFailures: this.writeFailures, + }; + } + sendSnapshot(stream: NodeJS.WritableStream, agents: AgentRecord[]): void { this.write({ type: "snapshot", agents }, stream); } @@ -95,6 +110,7 @@ export class UiEventBroker { try { client.write(payload); } catch { + this.writeFailures += 1; this.clients.delete(client); } } diff --git a/apps/server/src/shared/git/base-ref.ts b/apps/server/src/shared/git/base-ref.ts index f87eca9a..7ef68c04 100644 --- a/apps/server/src/shared/git/base-ref.ts +++ b/apps/server/src/shared/git/base-ref.ts @@ -11,6 +11,8 @@ export type ResolveBaseRefOptions = { runCommand?: CommandRunner; /** When false, do not infer the target branch from @{upstream}. */ allowUpstreamFallback?: boolean; + /** Receives command execution errors while resolution keeps its null fallback. */ + onError?: (error: unknown) => void; }; function normalizeBranchName(ref: string | null | undefined): string | null { @@ -51,7 +53,8 @@ async function resolveTargetBranch( } } } - } catch { + } catch (error) { + options.onError?.(error); // No upstream configured — fall through to origin/main. } } @@ -120,13 +123,13 @@ export async function resolveBaseRef( options ); for (const ref of candidateRefsForBranch(branchName)) { - const found = await refExists(run, worktreePath, ref); + const found = await refExists(run, worktreePath, ref, options.onError); if (found) return found; } return ( - (await refExists(run, worktreePath, "origin/main")) ?? - (await refExists(run, worktreePath, "main")) + (await refExists(run, worktreePath, "origin/main", options.onError)) ?? + (await refExists(run, worktreePath, "main", options.onError)) ); } @@ -142,7 +145,8 @@ function isSafeRef(ref: string): boolean { async function refExists( run: CommandRunner, worktreePath: string, - ref: string + ref: string, + onError?: (error: unknown) => void ): Promise { // The `isSafeRef` guard at every entry point keeps `-`-prefixed values // from reaching git as flags. We tried adding a `--` end-of-options @@ -157,7 +161,8 @@ async function refExists( { allowedExitCodes: [0, 1, 128], timeoutMs: 5_000 } ); return result.exitCode === 0 && result.stdout.trim() ? ref : null; - } catch { + } catch (error) { + onError?.(error); return null; } } diff --git a/apps/server/src/shared/git/diff-stats.ts b/apps/server/src/shared/git/diff-stats.ts index af0ecc6c..bfef6a10 100644 --- a/apps/server/src/shared/git/diff-stats.ts +++ b/apps/server/src/shared/git/diff-stats.ts @@ -22,6 +22,8 @@ export type GetDiffStatsOptions = { runCommand?: CommandRunner; /** Include staged, unstaged, and untracked working-tree changes. */ includeUncommitted?: boolean; + /** Receives command/probe failures while the public result remains null. */ + onError?: (error: unknown) => void; }; /** @@ -51,6 +53,7 @@ export async function getDiffStats( try { const resolvedBase = await resolveBaseRef(worktreePath, baseRef, { runCommand: run, + onError: options.onError, }); if (!resolvedBase) return null; @@ -85,7 +88,8 @@ export async function getDiffStats( const ignoredPaths = await getGitIgnoredPaths( worktreePath, trackedPaths, - run + run, + options.onError ); let added = 0; @@ -123,7 +127,8 @@ export async function getDiffStats( files: seenFiles.size, computedAt: Date.now(), }; - } catch { + } catch (error) { + options.onError?.(error); return null; } } @@ -145,7 +150,8 @@ const CHECK_IGNORE_BATCH_SIZE = 500; async function getGitIgnoredPaths( worktreePath: string, paths: string[], - run: CommandRunner + run: CommandRunner, + onError?: (error: unknown) => void ): Promise> { if (paths.length === 0) return new Set(); const ignored = new Set(); @@ -164,7 +170,8 @@ async function getGitIgnoredPaths( } } return ignored; - } catch { + } catch (error) { + onError?.(error); return ignored; } } diff --git a/apps/server/src/stream-manager.ts b/apps/server/src/stream-manager.ts index 3f618de7..4a6e748e 100644 --- a/apps/server/src/stream-manager.ts +++ b/apps/server/src/stream-manager.ts @@ -20,6 +20,8 @@ export class StreamManager { private sessions = new Map(); private onStateChange: OnStateChange; private onStreamEnd?: OnStreamEnd; + private framesSent = 0; + private bytesSent = 0; constructor(onStateChange: OnStateChange, onStreamEnd?: OnStreamEnd) { this.onStateChange = onStateChange; @@ -200,6 +202,24 @@ export class StreamManager { return session !== undefined && session.status === "live"; } + getMetrics(): { + streams: number; + viewers: number; + framesSent: number; + bytesSent: number; + } { + let viewers = 0; + for (const session of this.sessions.values()) { + viewers += session.viewers.size; + } + return { + streams: this.sessions.size, + viewers, + framesSent: this.framesSent, + bytesSent: this.bytesSent, + }; + } + stopAll(): void { for (const agentId of [...this.sessions.keys()]) { this.stopStream(agentId); @@ -237,6 +257,8 @@ export class StreamManager { }; if (v.writable !== false) { v.write(frameChunk); + this.framesSent += 1; + this.bytesSent += frameChunk.length; if (typeof v.flush === "function") { v.flush(); } diff --git a/apps/server/src/terminal/copy-mode-observer.ts b/apps/server/src/terminal/copy-mode-observer.ts index d50188f2..732915e6 100644 --- a/apps/server/src/terminal/copy-mode-observer.ts +++ b/apps/server/src/terminal/copy-mode-observer.ts @@ -37,9 +37,29 @@ const DETACH_GRACE_MS = 2_000; export class CopyModeObserverManager { private readonly observers = new Map(); + private pollCount = 0; + private pollFailures = 0; constructor(private readonly publishTerminalState: PublishTerminalState) {} + getMetrics(): { + observers: number; + viewers: number; + pollCount: number; + pollFailures: number; + } { + let viewers = 0; + for (const observer of this.observers.values()) { + viewers += observer.viewers.size; + } + return { + observers: this.observers.size, + viewers, + pollCount: this.pollCount, + pollFailures: this.pollFailures, + }; + } + attachViewer( agentId: string, sessionName: string, @@ -207,8 +227,15 @@ export class CopyModeObserverManager { } observer.pollInFlight = (async () => { + this.pollCount += 1; const observedAt = Date.now(); - const state = await observer.terminal.getCopyModeState(); + let state; + try { + state = await observer.terminal.getCopyModeState(); + } catch (error) { + this.pollFailures += 1; + throw error; + } const now = Date.now(); if (!state.inCopyMode) { diff --git a/apps/server/test/diff-stats-refresher.test.ts b/apps/server/test/diff-stats-refresher.test.ts index 30077b02..4bbde3cb 100644 --- a/apps/server/test/diff-stats-refresher.test.ts +++ b/apps/server/test/diff-stats-refresher.test.ts @@ -6,6 +6,7 @@ import { type DiffStatsChangedEvent, } from "../src/agents/diff-stats-refresher.js"; import type { DiffStats } from "../src/shared/git/diff-stats.js"; +import { SubsystemTracker } from "../src/observability/subsystem-tracker.js"; type AgentMap = Map; @@ -398,12 +399,18 @@ describe("DiffStatsRefresher", () => { throw new Error("git exploded"); }); const warn = vi.fn(); + const tracker = new SubsystemTracker({ + id: "git", + label: "Git", + description: "Refreshes diffs", + }); const refresher = new DiffStatsRefresher({ getAgent: async (id) => agents.get(id) ?? null, publishEvent: (event) => events.push(event), computeDiffStats: compute, freshnessMs: 1_000, logger: { warn }, + tracker, }); await refresher.signal("a1"); @@ -415,5 +422,10 @@ describe("DiffStatsRefresher", () => { expect(refresher.getStats("a1")).toMatchObject({ added: 5 }); expect(events).toHaveLength(1); expect(warn).toHaveBeenCalled(); + expect(tracker.snapshot()).toMatchObject({ + state: "degraded", + failures: 1, + lastError: "Operation failed", + }); }); }); diff --git a/apps/server/test/diff-stats.test.ts b/apps/server/test/diff-stats.test.ts index 59dc15b3..e6738546 100644 --- a/apps/server/test/diff-stats.test.ts +++ b/apps/server/test/diff-stats.test.ts @@ -320,9 +320,14 @@ describe("getDiffStats", () => { const runCommand = vi.fn(async () => { throw new Error("git: command not found"); }); + const onError = vi.fn(); - const result = await getDiffStats(tempRoot, "main", { runCommand }); + const result = await getDiffStats(tempRoot, "main", { + runCommand, + onError, + }); expect(result).toBeNull(); + expect(onError).toHaveBeenCalledWith(expect.any(Error)); }); it("rejects refs that start with `-` so a crafted base branch can't be parsed as a git option", async () => { diff --git a/apps/server/test/service-resources.test.ts b/apps/server/test/service-resources.test.ts new file mode 100644 index 00000000..797c2171 --- /dev/null +++ b/apps/server/test/service-resources.test.ts @@ -0,0 +1,150 @@ +import type { Pool } from "pg"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + ServiceResources, + type WorkloadSnapshot, +} from "../src/observability/service-resources.js"; + +function createPool(query = vi.fn(async () => ({ rows: [{ ok: 1 }] }))): Pool { + return { + query, + totalCount: 1, + idleCount: 1, + waitingCount: 0, + options: { max: 10 }, + } as unknown as Pool; +} + +function workloads(): WorkloadSnapshot { + return { + runningAgents: 0, + sseClients: 0, + streams: 0, + streamViewers: 0, + terminalObservers: 0, + terminalViewers: 0, + scheduledJobs: 0, + jobMonitors: 0, + gitRefreshesInFlight: 0, + uiEventsPublished: 0, + uiWriteFailures: 0, + terminalPolls: 0, + terminalPollFailures: 0, + }; +} + +describe("ServiceResources", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-15T12:00:00Z")); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("serializes slow samples and prevents commits after stop", async () => { + let resolveQuery: (() => void) | null = null; + const query = vi.fn( + () => + new Promise<{ rows: never[] }>((resolve) => { + resolveQuery = () => resolve({ rows: [] }); + }) + ); + const resources = new ServiceResources({ + pool: createPool(query), + listAgentSessions: async () => [], + getWorkloads: workloads, + subsystemTrackers: [], + processTreeSupported: false, + }); + + resources.start(); + resources.start(); + await Promise.resolve(); + expect(query).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(13_500); + expect(query).toHaveBeenCalledTimes(1); + + const samplesBeforeStop = resources.getSnapshot().series.length; + resources.stop(); + resolveQuery?.(); + await Promise.resolve(); + await Promise.resolve(); + expect(resources.getSnapshot().series).toHaveLength(samplesBeforeStop); + }); + + it("counts running agents when process-tree metrics are unsupported", async () => { + const resources = new ServiceResources({ + pool: createPool(), + listAgentSessions: async () => [ + { tmuxSession: "agent-one" }, + { tmuxSession: "agent-two" }, + ], + getWorkloads: workloads, + subsystemTrackers: [], + processTreeSupported: false, + }); + + resources.start(); + await vi.advanceTimersByTimeAsync(0); + expect(resources.getSnapshot()).toMatchObject({ + capabilities: { processTreeMetrics: "unsupported" }, + current: { workloads: { runningAgents: 2 } }, + }); + resources.stop(); + }); + + it("bounds request timing storage and finalizes requests exactly once", () => { + const resources = new ServiceResources({ + pool: createPool(), + listAgentSessions: async () => [], + getWorkloads: workloads, + subsystemTrackers: [], + processTreeSupported: false, + }); + + for (let index = 0; index < 5_000; index += 1) { + const token = resources.requestStarted(); + resources.requestFinished(token, index % 10 === 0 ? 500 : 200); + resources.requestFinished(token, 500); + } + + expect(resources.getHttpObservationStorageSize()).toBeLessThanOrEqual(128); + expect(resources.getSnapshot().current.http).toMatchObject({ + requestsPerMinute: 5_000, + inFlight: 0, + errorRatePercent: 10, + }); + }); + + it("degrades owner subsystems when recent writes or polls fail", async () => { + const current = workloads(); + current.sseClients = 1; + current.terminalObservers = 1; + const resources = new ServiceResources({ + pool: createPool(), + listAgentSessions: async () => [], + getWorkloads: () => ({ ...current }), + subsystemTrackers: [], + processTreeSupported: false, + }); + + resources.start(); + await vi.advanceTimersByTimeAsync(0); + current.uiEventsPublished += 1; + current.uiWriteFailures += 1; + current.terminalPolls += 1; + current.terminalPollFailures += 1; + await vi.advanceTimersByTimeAsync(5_000); + + const byId = new Map( + resources.getSnapshot().subsystems.map((item) => [item.id, item]) + ); + expect(byId.get("ui-event-stream")?.state).toBe("degraded"); + expect(byId.get("terminal-observers")?.state).toBe("degraded"); + resources.stop(); + }); +}); diff --git a/apps/server/test/subsystem-tracker.test.ts b/apps/server/test/subsystem-tracker.test.ts new file mode 100644 index 00000000..c91bc3eb --- /dev/null +++ b/apps/server/test/subsystem-tracker.test.ts @@ -0,0 +1,110 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { SubsystemTracker } from "../src/observability/subsystem-tracker.js"; + +describe("SubsystemTracker", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-15T12:00:00Z")); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("moves from unknown through running to healthy", () => { + const tracker = new SubsystemTracker({ + id: "reconcile", + label: "Reconciliation", + description: "Checks sessions", + expectedCadenceMs: 30_000, + }); + + expect(tracker.snapshot().state).toBe("unknown"); + const run = tracker.start(); + expect(tracker.snapshot().state).toBe("running"); + vi.advanceTimersByTime(125); + run.succeed({ corrections: 2 }); + + expect(tracker.snapshot()).toMatchObject({ + state: "healthy", + runs: 1, + failures: 0, + lastDurationMs: 125, + metadata: { corrections: 2 }, + }); + }); + + it("degrades on failure without exposing exception details", () => { + const tracker = new SubsystemTracker({ + id: "git", + label: "Git", + description: "Refreshes diffs", + }); + const run = tracker.start(); + run.fail( + new Error( + "git -C /Users/private/repo failed https://user:pass@example.test token=secret cookie=session" + ) + ); + + expect(tracker.snapshot()).toMatchObject({ + state: "degraded", + statusReason: "failure", + failures: 1, + lastError: "Operation failed", + }); + }); + + it("uses a stable timeout summary for timeout-like failures", () => { + const tracker = new SubsystemTracker({ + id: "db", + label: "Database", + description: "Checks connectivity", + }); + tracker.start().fail(new Error("query timeout password=hunter2")); + expect(tracker.snapshot().lastError).toBe("Operation timed out"); + }); + + it("marks recurring work stale after twice its cadence", () => { + const tracker = new SubsystemTracker({ + id: "loop", + label: "Loop", + description: "Recurring work", + expectedCadenceMs: 1_000, + }); + tracker.start().succeed(); + vi.advanceTimersByTime(2_001); + expect(tracker.snapshot()).toMatchObject({ + state: "degraded", + statusReason: "stale", + }); + }); + + it("degrades a recurring run that never settles", () => { + const tracker = new SubsystemTracker({ + id: "loop", + label: "Loop", + description: "Recurring work", + expectedCadenceMs: 1_000, + }); + tracker.start(); + vi.advanceTimersByTime(2_001); + expect(tracker.snapshot()).toMatchObject({ + state: "degraded", + statusReason: "stuck", + inFlight: 1, + }); + }); + + it("reports intentionally disabled work without degrading it", () => { + const tracker = new SubsystemTracker({ + id: "updates", + label: "Updates", + description: "Checks releases", + expectedCadenceMs: 1_000, + }); + tracker.setDisabled(true); + expect(tracker.snapshot().state).toBe("disabled"); + }); +}); diff --git a/apps/server/test/system-routes.test.ts b/apps/server/test/system-routes.test.ts index de4ea86f..0717a215 100644 --- a/apps/server/test/system-routes.test.ts +++ b/apps/server/test/system-routes.test.ts @@ -63,6 +63,42 @@ describe("GET /api/v1/system/defaults", () => { }); }); +describe("GET /api/v1/system/resources", () => { + it("returns a bounded operational snapshot", async () => { + const res = await ctx.app.inject({ + method: "GET", + url: "/api/v1/system/resources?window=15m", + headers: { cookie: sessionCookie }, + }); + expect(res.statusCode).toBe(200); + const body = res.json(); + expect(body.sampleIntervalMs).toBe(5_000); + expect(body.current.server.rssBytes).toBeGreaterThan(0); + expect(body.current.database.pool.max).toBeGreaterThan(0); + expect(Array.isArray(body.series)).toBe(true); + expect(body.series.length).toBeLessThanOrEqual(720); + expect(body.subsystems.map((item: { id: string }) => item.id)).toEqual( + expect.arrayContaining([ + "api-server", + "database", + "agent-reconciliation", + "activity-monitor", + "git-diff-refreshes", + ]) + ); + }); + + it("rejects unsupported history windows", async () => { + const res = await ctx.app.inject({ + method: "GET", + url: "/api/v1/system/resources?window=24h", + headers: { cookie: sessionCookie }, + }); + expect(res.statusCode).toBe(400); + expect(res.json().error).toMatch(/window/); + }); +}); + describe("GET /api/v1/system/path-info", () => { it("rejects missing path parameter", async () => { const res = await ctx.app.inject({ diff --git a/apps/web/src/components/app/service-resources-chart-config.ts b/apps/web/src/components/app/service-resources-chart-config.ts new file mode 100644 index 00000000..7a971005 --- /dev/null +++ b/apps/web/src/components/app/service-resources-chart-config.ts @@ -0,0 +1,13 @@ +import type { ChartConfig } from "@/components/ui/chart"; + +export const cpuChartConfig = { + serverCpuPercent: { label: "Dispatch CPU", color: "hsl(var(--chart-1))" }, + agentCpuPercent: { label: "Agent CPU", color: "hsl(var(--chart-3))" }, + hostLoad1: { label: "Host load (1m)", color: "hsl(var(--chart-2))" }, +} satisfies ChartConfig; + +export const memoryChartConfig = { + serverRssMb: { label: "Dispatch RSS", color: "hsl(var(--chart-1))" }, + serverHeapMb: { label: "JS heap", color: "hsl(var(--chart-4))" }, + agentRssMb: { label: "Agent RSS", color: "hsl(var(--chart-3))" }, +} satisfies ChartConfig; diff --git a/apps/web/src/components/app/service-resources-chart.tsx b/apps/web/src/components/app/service-resources-chart.tsx new file mode 100644 index 00000000..af389d2f --- /dev/null +++ b/apps/web/src/components/app/service-resources-chart.tsx @@ -0,0 +1,144 @@ +import { Area, AreaChart, CartesianGrid, XAxis, YAxis } from "recharts"; + +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { + ChartContainer, + ChartTooltip, + ChartTooltipContent, + type ChartConfig, +} from "@/components/ui/chart"; + +export function ResourceChart({ + title, + description, + data, + config, + keys, + unit, + secondaryKey, +}: { + title: string; + description: string; + data: Array>; + config: ChartConfig; + keys: string[]; + unit: string; + secondaryKey?: string; +}) { + const allKeys = secondaryKey ? [...keys, secondaryKey] : keys; + return ( + + + {title} + {description} + + + {data.length < 2 ? ( +
+ Collecting history… +
+ ) : ( + + + + {allKeys.map((key) => ( + + + + + ))} + + + + new Date(Number(value)).toLocaleTimeString([], { + hour: "numeric", + minute: "2-digit", + }) + } + /> + `${Math.round(Number(value))}${unit}`} + /> + {secondaryKey && ( + `${Number(value).toFixed(1)} load`} + /> + )} + [ + item.dataKey === secondaryKey + ? `${Number(value).toFixed(2)} load` + : `${Number(value).toFixed(1)}${unit}`, + config[String(item.dataKey)]?.label ?? name, + ]} + /> + } + labelFormatter={(value) => + new Date(Number(value)).toLocaleTimeString() + } + /> + {allKeys.map((key) => ( + + ))} + + + )} + + + ); +} diff --git a/apps/web/src/components/app/service-resources-dashboard.tsx b/apps/web/src/components/app/service-resources-dashboard.tsx new file mode 100644 index 00000000..b94c152a --- /dev/null +++ b/apps/web/src/components/app/service-resources-dashboard.tsx @@ -0,0 +1,241 @@ +import { + Activity, + CircleGauge, + Cpu, + Database, + GitCompareArrows, + HardDrive, + MemoryStick, + Users, +} from "lucide-react"; + +import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; +import type { ServiceResourcesResponse } from "@/hooks/use-service-resources"; +import { ResourceChart } from "./service-resources-chart"; +import { + cpuChartConfig, + memoryChartConfig, +} from "./service-resources-chart-config"; +import { + formatBytes, + formatMs, + formatUptime, +} from "./service-resources-format"; +import { ServiceResourcesSubsystems } from "./service-resources-subsystems"; + +function SummaryCard({ + icon: Icon, + label, + value, + detail, + scope, +}: { + icon: typeof Cpu; + label: string; + value: string; + detail: string; + scope: "Dispatch" | "Agents" | "Dependency" | "Host"; +}) { + return ( + + +
+
+ +
+ + {scope} + +
+
{label}
+
+ {value} +
+
+ {detail} +
+ + + ); +} + +export function ServiceResourcesDashboard({ + data, +}: { + data: ServiceResourcesResponse; +}) { + const current = data.current; + const chartData = data.series.map((sample) => ({ + ...sample, + serverRssMb: sample.serverRssBytes / 1024 / 1024, + serverHeapMb: sample.serverHeapBytes / 1024 / 1024, + agentRssMb: + sample.agentRssBytes === null ? null : sample.agentRssBytes / 1024 / 1024, + })); + + return ( + <> + {data.overall.reasons.length > 0 && ( + + + +
+
+ Dispatch needs attention +
+ {data.overall.reasons.map((reason) => ( +

+ {reason.message} +

+ ))} +
+ + + )} + +
+ + + + + + +
+ +
+ + +
+ + + +
+ + + + {" "} + Workload + + + + {[ + ["Running agents", current.workloads.runningAgents], + ["SSE clients", current.workloads.sseClients], + ["Terminal viewers", current.workloads.terminalViewers], + ["Scheduled jobs", current.workloads.scheduledJobs], + ["Git refreshes", current.workloads.gitRefreshesInFlight], + ].map(([label, value]) => ( +
+
{label}
+
{value}
+
+ ))} + + + + + + Capacity + + + +
+ Service uptime + + {formatUptime(current.server.uptimeSeconds)} + +
+
+ Database pool + + {current.database.pool.total} total ·{" "} + {current.database.pool.idle} idle + +
+
+ Requests (1 min) + + {current.http.requestsPerMinute} + +
+
+ Host CPUs + {current.host.cpuCount} +
+ + +
+ + ); +} diff --git a/apps/web/src/components/app/service-resources-format.ts b/apps/web/src/components/app/service-resources-format.ts new file mode 100644 index 00000000..5d90c18f --- /dev/null +++ b/apps/web/src/components/app/service-resources-format.ts @@ -0,0 +1,46 @@ +import type { ResourceHealthState } from "@/hooks/use-service-resources"; + +export function formatBytes(bytes: number | null): string { + if (bytes === null) return "—"; + if (bytes < 1024) return `${bytes} B`; + const units = ["KB", "MB", "GB", "TB"]; + let value = bytes / 1024; + let unit = units[0]; + for (let index = 1; index < units.length && value >= 1024; index += 1) { + value /= 1024; + unit = units[index]; + } + return `${value >= 10 ? value.toFixed(0) : value.toFixed(1)} ${unit}`; +} + +export function formatMs(value: number | null): string { + if (value === null) return "—"; + if (value < 1) return "<1 ms"; + if (value < 1000) return `${Math.round(value)} ms`; + return `${(value / 1000).toFixed(1)} s`; +} + +export function formatUptime(seconds: number): string { + const days = Math.floor(seconds / 86_400); + const hours = Math.floor((seconds % 86_400) / 3_600); + const minutes = Math.floor((seconds % 3_600) / 60); + if (days > 0) return `${days}d ${hours}h`; + if (hours > 0) return `${hours}h ${minutes}m`; + return `${minutes}m`; +} + +export function stateLabel(state: ResourceHealthState): string { + if (state === "degraded") return "Needs attention"; + return state.replace("_", " "); +} + +export function stateBadgeVariant(state: ResourceHealthState) { + if (state === "healthy" || state === "running") return "running" as const; + if (state === "degraded" || state === "unavailable") return "error" as const; + if (state === "unknown") return "stopped" as const; + return "default" as const; +} + +export function metadataLabel(key: string): string { + return key.replace(/([a-z])([A-Z])/g, "$1 $2").toLowerCase(); +} diff --git a/apps/web/src/components/app/service-resources-settings.tsx b/apps/web/src/components/app/service-resources-settings.tsx new file mode 100644 index 00000000..71b5b9ca --- /dev/null +++ b/apps/web/src/components/app/service-resources-settings.tsx @@ -0,0 +1,109 @@ +import { useState } from "react"; +import { RefreshCw } from "lucide-react"; + +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent } from "@/components/ui/card"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import { + type ResourceWindow, + useServiceResources, +} from "@/hooks/use-service-resources"; +import { cn } from "@/lib/utils"; +import { ServiceResourcesDashboard } from "./service-resources-dashboard"; +import { stateBadgeVariant, stateLabel } from "./service-resources-format"; + +export function ServiceResourcesSettings(): JSX.Element { + const [window, setWindow] = useState("1h"); + const { data, error, isFetching, refetch, dataUpdatedAt } = + useServiceResources(window); + const stale = dataUpdatedAt > 0 && Date.now() - dataUpdatedAt > 15_000; + + return ( +
+
+
+
+

Service resources

+ {data && ( + + {stateLabel(data.overall.state)} + + )} +
+

+ Live health and resource use for Dispatch, its agents, dependencies, + and host. +

+
+
+ {stale && ( + Data is stale + )} + + {dataUpdatedAt > 0 + ? `Updated ${new Date(dataUpdatedAt).toLocaleTimeString()}` + : ""} + + + +
+
+ + {!data && !error && ( +
+ Loading service resources… +
+ )} + {!data && error && ( + + +
+ Resources are unavailable +
+

{error.message}

+ + + + )} + {data && } +
+ ); +} diff --git a/apps/web/src/components/app/service-resources-subsystems.tsx b/apps/web/src/components/app/service-resources-subsystems.tsx new file mode 100644 index 00000000..7678e4fa --- /dev/null +++ b/apps/web/src/components/app/service-resources-subsystems.tsx @@ -0,0 +1,123 @@ +import { useState } from "react"; +import { ChevronDown, ChevronRight, Server } from "lucide-react"; + +import { Badge } from "@/components/ui/badge"; +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import type { SubsystemSnapshot } from "@/hooks/use-service-resources"; +import { + formatMs, + metadataLabel, + stateBadgeVariant, + stateLabel, +} from "./service-resources-format"; + +function reasonLabel(reason: SubsystemSnapshot["statusReason"]): string | null { + if (reason === "stuck") return "A run exceeded twice its expected cadence."; + if (reason === "stale") return "No recent successful run was observed."; + if (reason === "failure") return "The latest observed operation failed."; + return null; +} + +function SubsystemRow({ subsystem }: { subsystem: SubsystemSnapshot }) { + const [expanded, setExpanded] = useState(false); + const metadata = Object.entries(subsystem.metadata); + const statusReason = reasonLabel(subsystem.statusReason); + return ( +
+ + {expanded && ( +
+

{subsystem.description}

+ {statusReason && ( +

{statusReason}

+ )} +
+ {subsystem.p95DurationMs !== null && ( + p95 {formatMs(subsystem.p95DurationMs)} + )} + {subsystem.failures > 0 && ( + {subsystem.failures} failures + )} + {metadata.map(([key, value]) => ( + + {metadataLabel(key)} {value.toLocaleString()} + + ))} +
+ {subsystem.lastError && ( +

{subsystem.lastError}

+ )} +
+ )} +
+ ); +} + +export function ServiceResourcesSubsystems({ + subsystems, +}: { + subsystems: SubsystemSnapshot[]; +}) { + return ( + + + + + Runtime health + + + Live state for Dispatch loops, dependencies, and connection managers. + + + +
+ Subsystem + Duration + Activity + State +
+ {subsystems.map((subsystem) => ( + + ))} + + + ); +} diff --git a/apps/web/src/components/app/settings-pane.tsx b/apps/web/src/components/app/settings-pane.tsx index c2182555..3624d9fe 100644 --- a/apps/web/src/components/app/settings-pane.tsx +++ b/apps/web/src/components/app/settings-pane.tsx @@ -13,6 +13,7 @@ import { ReleasesAdmin } from "@/components/app/release-admin"; import { UpdatesSection } from "@/components/app/release-manager"; import { SecuritySettings } from "@/components/app/security-settings"; import { ServiceStatus } from "@/components/app/service-status"; +import { ServiceResourcesSettings } from "@/components/app/service-resources-settings"; import { type ServiceState } from "@/components/app/types"; import { WorktreeLocationSettings } from "@/components/app/worktree-location-settings"; import { type IconColorId } from "@/hooks/use-icon-color"; @@ -206,6 +207,7 @@ export function SettingsContent({ )} {activeSection === "notifications" && } {activeSection === "connections" && } + {activeSection === "resources" && } {activeSection === "updates" && ( )} diff --git a/apps/web/src/components/app/settings-state.ts b/apps/web/src/components/app/settings-state.ts index e9784ff8..a6c844b3 100644 --- a/apps/web/src/components/app/settings-state.ts +++ b/apps/web/src/components/app/settings-state.ts @@ -5,6 +5,7 @@ import { BookOpenText, Cable, Package, + Gauge, Settings, Users, } from "lucide-react"; @@ -17,6 +18,7 @@ export type SettingsSection = | "connections" | "notifications" | "updates" + | "resources" | "help" | "releases"; @@ -29,6 +31,7 @@ const BASE_SECTIONS: Array<{ { id: "agents", label: "Agents", icon: Users }, { id: "connections", label: "Connections", icon: Cable }, { id: "notifications", label: "Notifications", icon: Bell }, + { id: "resources", label: "Resources", icon: Gauge }, { id: "updates", label: "Updates", icon: ArrowDownToLine }, ]; @@ -49,6 +52,7 @@ const ALL_VALID_SECTIONS: SettingsSection[] = [ "connections", "notifications", "updates", + "resources", "help", "releases", ]; diff --git a/apps/web/src/hooks/use-service-resources.ts b/apps/web/src/hooks/use-service-resources.ts new file mode 100644 index 00000000..e4dc8480 --- /dev/null +++ b/apps/web/src/hooks/use-service-resources.ts @@ -0,0 +1,129 @@ +import { useQuery } from "@tanstack/react-query"; + +import { api } from "@/lib/api"; + +export type ResourceHealthState = + | "healthy" + | "degraded" + | "unavailable" + | "running" + | "idle" + | "disabled" + | "unknown"; + +export type ResourceSample = { + at: number; + serverCpuPercent: number; + serverRssBytes: number; + serverHeapBytes: number; + agentCpuPercent: number | null; + agentRssBytes: number | null; + hostLoad1: number; +}; + +export type SubsystemSnapshot = { + id: string; + label: string; + description: string; + state: ResourceHealthState; + statusReason: "failure" | "stale" | "stuck" | null; + expectedCadenceMs: number | null; + lastStartedAt: number | null; + lastCompletedAt: number | null; + lastSucceededAt: number | null; + lastFailedAt: number | null; + lastDurationMs: number | null; + p95DurationMs: number | null; + inFlight: number; + runs: number; + failures: number; + lastError: string | null; + metadata: Record; +}; + +export type ServiceResourcesResponse = { + generatedAt: number; + processStartedAt: number; + availableHistoryMs: number; + sampleIntervalMs: number; + overall: { + state: "healthy" | "degraded" | "unavailable" | "unknown"; + reasons: Array<{ code: string; message: string }>; + }; + capabilities: { + processTreeMetrics: "available" | "unsupported" | "error"; + eventLoopMetrics: "available"; + }; + current: { + server: { + cpuPercent: number; + rssBytes: number; + heapUsedBytes: number; + heapTotalBytes: number; + externalBytes: number; + uptimeSeconds: number; + }; + host: { + load1: number; + load5: number; + load15: number; + cpuCount: number; + totalMemoryBytes: number; + freeMemoryBytes: number; + }; + agents: { + supported: boolean; + cpuPercent: number | null; + rssBytes: number | null; + processCount: number | null; + sampledAt: number | null; + error: string | null; + }; + database: { + state: "healthy" | "unavailable" | "unknown"; + latencyMs: number | null; + sampledAt: number | null; + pool: { total: number; idle: number; waiting: number; max: number }; + }; + eventLoop: { p95DelayMs: number }; + http: { + requestsPerMinute: number; + inFlight: number; + errorRatePercent: number; + p95DurationMs: number | null; + }; + workloads: { + runningAgents: number; + sseClients: number; + streams: number; + streamViewers: number; + terminalObservers: number; + terminalViewers: number; + scheduledJobs: number; + jobMonitors: number; + gitRefreshesInFlight: number; + uiEventsPublished: number; + uiWriteFailures: number; + terminalPolls: number; + terminalPollFailures: number; + }; + }; + subsystems: SubsystemSnapshot[]; + series: ResourceSample[]; +}; + +export type ResourceWindow = "15m" | "1h"; + +export function useServiceResources(window: ResourceWindow) { + return useQuery({ + queryKey: ["service-resources", window], + queryFn: () => + api( + `/api/v1/system/resources?window=${encodeURIComponent(window)}` + ), + refetchInterval: () => + typeof document !== "undefined" && document.hidden ? false : 5_000, + refetchIntervalInBackground: false, + placeholderData: (previous) => previous, + }); +} diff --git a/docs/service-resources-dashboard-plan.md b/docs/service-resources-dashboard-plan.md new file mode 100644 index 00000000..7f1ca4d4 --- /dev/null +++ b/docs/service-resources-dashboard-plan.md @@ -0,0 +1,476 @@ +# Service Resources Dashboard + +## Goal + +Add a **Resources** page under Dispatch Settings that answers three operator +questions quickly: + +1. Is Dispatch healthy right now? +2. What is consuming host resources? +3. Which Dispatch loop or dependency is slow, failing, or falling behind? + +The page is an operational view of the Dispatch installation, not another +agent-usage analytics page. Existing Activity metrics remain the place for +tokens, working time, and agent productivity. + +## Current architecture findings + +Dispatch is primarily one Bun/Fastify server process. That process coordinates +several kinds of work: + +- a PostgreSQL connection pool with a maximum of 10 connections; +- tmux-backed agent sessions and their CLI process trees; +- a 30-second agent reconciliation pass; +- pane activity checks that run as part of the reconciliation cadence; +- signal-driven Git diff-stat computations, deduplicated in flight and cached + for 3 seconds; +- per-job cron schedulers and active-run monitors; +- an automatic release check after startup and every six hours; +- SSE clients, terminal WebSockets, and optional browser screencast streams; +- viewer-driven terminal copy-mode polling; +- periodic tmux diagnostic capture and log maintenance. + +There is no continuously running Git watcher today. The closest unit is +`DiffStatsRefresher`, which runs Git subprocesses when agent activity or a UI +request signals it. The dashboard should call this **Git diff refreshes** and +show request, duration, cache/deduplication, and failure information. + +Postgres and the tmux server are dependencies that may be shared with other +applications or Dispatch instances. Their entire host-process CPU and memory +must not be presented as Dispatch-owned usage. Dispatch can accurately report +its database pool and query health, its own tmux sessions, and resource totals +for process trees rooted at those sessions. + +## Proposed information architecture + +Add `/settings/resources` with a `Gauge`-style icon after Notifications and +before Updates. Use the existing Settings shell and shadcn cards, badges, +tooltips, and chart primitives. + +```text +┌ Overall health ───── Uptime ───── Last sample ─────────────────────┐ +│ Dispatch CPU │ Dispatch RSS │ Agent CPU/RSS │ DB │ Event-loop lag │ +├ Recent history ────────────────────────────────────────────────────┤ +│ CPU: Dispatch / Agents / host load │ Memory: RSS / heap / Agents │ +├ Runtime health ────────────────────────────────────────────────────┤ +│ API Healthy 18 req/min p95 24 ms ▸ │ +│ Database Healthy 2/10 pooled 3 ms ▸ │ +│ Reconciliation Healthy 12 s ago 180 ms ▸ │ +│ Git diff refresh Degraded 2 failures p95 1.4 s ▸ │ +├ Capacity and storage ──────────────────────────────────────────────┤ +│ Agents 8 running │ SSE 3 │ Streams 1 │ DB 84 MB │ Media 1.2 GB │ +└────────────────────────────────────────────────────────────────────┘ +``` + +### 1. Overview + +The first viewport should contain: + +- overall state: Healthy, Needs attention, or Unavailable; +- last sample time and process uptime; +- Dispatch CPU; +- Dispatch resident memory (RSS); +- agent process CPU and RSS aggregate; +- database round-trip latency and pool use; +- event-loop delay; +- active workload summary: running agents, SSE clients, terminal sockets, + stream sessions, and scheduled jobs. + +CPU must be labeled as **one-core percentage** so an expensive process can +legitimately exceed 100% on a multicore host. Also show host load separately; +do not blend the two measurements. + +Memory should distinguish: + +- Dispatch server RSS and JavaScript heap; +- Dispatch agent process-tree RSS; +- host total/free memory as context. + +Host free memory is informational, not a health verdict. Operating systems use +otherwise-free memory for caches, so a simple `used / total` threshold would +create false alarms. + +### 2. Recent resource history + +Show two compact charts using the in-memory samples collected since the server +started: + +- CPU: Dispatch server, agent process trees, and host load; +- memory: Dispatch RSS, agent process-tree RSS, and JavaScript heap. + +The MVP should keep a one-hour ring buffer sampled every five seconds (720 +samples). It should say **History since last service start** and show the +actual available duration. Do not write five-second samples to Postgres: that +would make observability increase database churn and grow permanent storage. + +A later release can add opt-in, downsampled persistence if operators need +cross-restart trends. + +### 3. Runtime health + +Use a dense, scannable table rather than a grid of equally prominent cards. +Each row represents a real runtime unit and reports only fields that make sense +for that unit. + +| Runtime unit | Useful live fields | Health signal | +| ----------------------- | ----------------------------------------------------------------- | ------------------------------------------------------ | +| API server | requests/min, in flight, error rate, p50/p95 duration | recent 5xx burst or sustained latency/event-loop delay | +| Database | probe latency, pool total/idle/waiting | failed probe, waiting clients, sustained slow probe | +| Agent reconciliation | cadence, last success, last duration, agents scanned, corrections | failed or stale beyond twice its cadence | +| Activity monitor | last success, duration, panes scanned, corrections | failure or stale beyond twice its cadence | +| Git diff refreshes | requests, in flight, completed, failures, p95 duration, last run | repeated failures, timeout, or growing in-flight work | +| Job schedulers | enabled schedules, active monitors, next run | scheduler error or missed run | +| Update checker | mode, last result, last duration, next run | last attempt failed; `off` is Disabled, not unhealthy | +| UI event stream | connected clients, events sent, write failures | repeated write failures | +| Terminal observers | viewers, active observers, fast/slow poll counts | repeated tmux probe failures | +| Browser streams | live streams, viewers, frames/second, bytes/second | CDP disconnect/error | +| Diagnostics maintenance | last capture/prune, last duration, failure | stale or failed beyond expected cadence | + +Rows should use the states `healthy`, `degraded`, `unavailable`, `running`, +`idle`, `disabled`, and `unknown`. Before a scheduled unit has run, show +Unknown or Scheduled rather than green. A unit intentionally disabled should +never lower overall health. + +Expanding a row can reveal its last error, recent duration, expected cadence, +and a plain-language description. Error strings must be sanitized; do not +expose environment variables, command arguments, auth tokens, or full private +paths. + +### 4. Capacity and storage + +Show operational totals that help explain pressure: + +- agents by lifecycle state and number of agent process trees found; +- Postgres pool use (`total`, `idle`, `waiting`, configured max); +- active SSE clients, terminal WebSockets/viewers, copy-mode observers, and + browser streams/viewers; +- scheduled jobs and in-flight job monitors; +- current database size via `pg_database_size(current_database())`; +- media, logs, diagnostics, and release-cache directory size. + +Directory sizes must be sampled asynchronously on a slow cadence (at least 60 +seconds), cached, bounded to the known Dispatch-owned roots, and tolerant of +missing or unreadable files. They must never be recursively scanned during an +HTTP request. + +Label every figure by scope: + +- **Dispatch**: server process and Dispatch-owned files; +- **Agents**: process trees rooted at Dispatch tmux panes; +- **Dependency**: database health/pool and tmux service state; +- **Host**: whole-machine context. + +This prevents users from reading shared Postgres or tmux resource totals as +resources caused solely by Dispatch. + +### 5. Diagnostics actions (post-MVP) + +Useful follow-on actions are: + +- copy a sanitized JSON snapshot; +- download a bounded diagnostics bundle containing the current snapshot, + recent subsystem results, and recent server log tail; +- open the existing update or help pages when a known issue has a documented + remedy. + +The first version should remain read-only. Restarting the service, killing an +agent, clearing caches, or pruning files are materially different operations +and should not be hidden inside a resource dashboard. + +## Metric definitions and collection + +### Cheap five-second sample + +Collect on a single unref'ed server timer and keep the result in memory: + +- `process.memoryUsage()` for RSS, heap used/total, external, and array + buffers; +- `process.cpuUsage(previous)` divided by elapsed wall time for one-core CPU; +- `process.uptime()`; +- `os.loadavg()`, `os.totalmem()`, and `os.freemem()`; +- event-loop delay from `node:perf_hooks`' `monitorEventLoopDelay`, after + confirming Bun runtime compatibility; +- cached counters and gauges owned by server components. + +Use separate slower cadences for collection that crosses a process or storage +boundary: + +- database probe and database pool snapshot every 10 seconds; +- one OS process snapshot every 10 seconds for known Dispatch tmux pane roots + and descendants; +- storage and database-size sampling every 60 seconds. + +Process sampling should use one `ps` invocation per sample, parse the full +parent map once, and aggregate only known roots. Never run `ps` once per agent. +If the platform does not provide the required fields, return an explicit +`unsupported` capability and keep the rest of the page working. + +### Request metrics + +Use Fastify hooks to record request start/end into rolling, bounded buckets: + +- request count and in-flight gauge; +- status-class counts; +- total duration distribution; +- optionally normalized route-level duration for the slowest few routes. + +Do not retain raw URLs because IDs create unbounded cardinality and query +strings can contain sensitive data. Use Fastify's normalized route template, +exclude or separately tag the resources endpoint, and keep only aggregate +histogram buckets rather than every request. + +### Subsystem metrics + +Create a small shared tracker with explicit lifecycle methods: + +```ts +type SubsystemRunTracker = { + start(metadata?: Record): RunHandle; + snapshot(now?: number): SubsystemSnapshot; +}; + +type RunHandle = { + succeed(metadata?: Record): void; + fail(error: unknown): void; +}; +``` + +Each snapshot should include: + +- expected cadence or `null` for signal-driven work; +- last started/completed/succeeded/failed timestamps; +- current in-flight count and peak; +- bounded success/failure counters; +- last and rolling p50/p95 duration; +- sanitized last error summary; +- unit-specific numeric metadata such as agents scanned or corrections made. + +Explicit instrumentation at the actual call sites is preferable to inferring +loop health from logs. The tracker must not swallow errors or change existing +control flow. + +### Health evaluation + +Evaluate health on the server so web and future CLI clients share semantics. +Start with conservative rules: + +- Unavailable: the DB probe fails or the sampler itself cannot produce a + current snapshot; +- Degraded: a recurring subsystem's last successful completion is older than + twice its expected cadence, a recent run failed with no later success, DB + pool waiters persist, or event-loop/API latency remains high across multiple + samples; +- Healthy: required probes pass and no required subsystem is degraded; +- Unknown: insufficient data; +- Disabled: intentionally not scheduled. + +Use sustained windows rather than one-sample CPU or latency spikes. Initial +thresholds should be constants with tests and returned in the API metadata so +the UI can explain why a state is degraded. + +## API proposal + +Add an authenticated, read-only endpoint: + +```text +GET /api/v1/system/resources?window=1h +``` + +The endpoint returns the cached current snapshot plus a downsampled series. It +does not perform subprocess, directory, or database-size collection inline. + +Suggested top-level contract: + +```ts +type ServiceResourcesResponse = { + generatedAt: string; + processStartedAt: string; + availableHistoryMs: number; + sampleIntervalMs: number; + overall: { state: HealthState; reasons: HealthReason[] }; + capabilities: { + processTreeMetrics: "available" | "unsupported" | "error"; + eventLoopMetrics: "available" | "unsupported"; + storageMetrics: "available" | "partial" | "error"; + }; + current: { + host: HostMetrics; + server: ProcessMetrics; + agents: AgentProcessMetrics; + database: DatabaseMetrics; + http: HttpMetrics; + workloads: WorkloadMetrics; + storage: StorageMetrics; + }; + subsystems: SubsystemSnapshot[]; + series: ResourceSample[]; +}; +``` + +Return stable reason codes plus display text, for example +`DB_PROBE_FAILED`, `RECONCILER_STALE`, and `EVENT_LOOP_DELAY_HIGH`. This lets +the web UI test behavior without parsing prose. + +The existing `/api/v1/health` should remain a minimal readiness check used by +launchd/update flows. Do not make it depend on the heavier sampler or expand it +into the dashboard payload. + +## Backend design + +Introduce an observability module with a narrow ownership boundary: + +```text +apps/server/src/observability/ + service-resources.ts orchestration, ring buffer, public snapshot + runtime-sampler.ts process, host, event-loop sampling + process-tree-sampler.ts one bounded OS process snapshot + subsystem-tracker.ts run/counter instrumentation + health-evaluator.ts stable state and reason rules + storage-sampler.ts slow cached Dispatch-owned disk scan +``` + +Add `apps/server/src/routes/resources.ts` rather than continuing to grow the +already broad system settings route. Construct one `ServiceResources` runtime +in `server.ts`, pass trackers or lightweight counter callbacks into the +components that own the work, and stop its timers in `cleanupAppResources()`. + +Components need read-only snapshot methods for current gauges where a counter +callback would be awkward: + +- `UiEventBroker`: connected client count and publish/write counters; +- `StreamManager`: stream/viewer/frame/byte counts; +- `CopyModeObserverManager`: observer/viewer/poll counts; +- `JobService`: scheduler and active monitor counts; +- `DiffStatsRefresher`: cache/in-flight counts plus run tracker integration. + +Avoid a global mutable metrics bag. Each feature remains the owner of its +gauges and receives only the tracker it needs; `ServiceResources` composes +snapshots at sample time. + +## Frontend design + +Add: + +```text +apps/web/src/hooks/use-service-resources.ts +apps/web/src/components/app/service-resources-settings.tsx +apps/web/src/components/app/service-resources-dashboard.tsx +apps/web/src/components/app/service-resources-chart.tsx +apps/web/src/components/app/service-resources-chart-config.ts +apps/web/src/components/app/service-resources-subsystems.tsx +apps/web/src/components/app/service-resources-format.ts +``` + +The React Query hook should: + +- poll every five seconds only while `/settings/resources` is mounted and the + document is visible; +- retain the previous successful payload during a transient refresh failure; +- expose staleness separately from service-reported health; +- stop polling when the page unmounts. + +Use responsive layout: + +- desktop: six summary cards, two side-by-side charts, full subsystem table; +- narrow screens: two-column cards, stacked charts, subsystem rows that reveal + detail on tap; +- mobile: single-column cards and list-style subsystem details without a wide + horizontal table. + +The existing API/DB dots in the Settings sidebar can remain. Once the resource +endpoint exists, clicking or focusing those statuses could link to Resources, +but the dashboard must not lift feature state into `App.tsx`; the page owns its +query and presentation. + +## Implementation phases + +### Phase 1: observable foundation and live overview + +1. Add ring-buffer, CPU/memory/host sampler, event-loop sampler, health + evaluator, and lifecycle cleanup. +2. Add DB probe/pool and workload gauges that are already cheap and available. +3. Add the authenticated resources route and contract tests. +4. Add the Settings route/nav item, overview cards, capability labels, loading, + stale, empty, and unavailable states. +5. Add one-hour CPU/memory charts and responsive Playwright coverage. + +This phase delivers a useful dashboard without modifying every subsystem. + +### Phase 2: real subsystem health + +1. Instrument reconciliation and activity monitoring separately. +2. Instrument Git diff refresh requests, dedupes, durations, and failures. +3. Instrument job schedulers/monitors and automatic update checks. +4. Expose UI event, terminal observer, and browser stream gauges/counters. +5. Add the runtime-health table and stable health-reason tests. + +### Phase 3: resource attribution and storage + +1. Implement cross-platform, single-pass process-tree sampling for Dispatch + agent tmux pane roots. +2. Add slow cached storage sampling and database size. +3. Add capacity/storage UI and partial/unsupported states. +4. Validate overhead under idle, many-agent, and active-stream scenarios. + +### Phase 4: operator diagnostics + +1. Add sanitized JSON snapshot copy/download. +2. Add a bounded diagnostics bundle if operator demand justifies it. +3. Consider opt-in downsampled persistence only after validating a real need + for history across restarts. + +## Validation strategy + +### Backend + +- unit-test ring-buffer eviction and downsampling; +- unit-test CPU delta math and unavailable platform fields; +- unit-test health transitions, especially Unknown/Disabled and stale cadence; +- unit-test error sanitization and bounded cardinality; +- route-test auth, payload shape, query validation, and partial capabilities; +- use fake clocks for all cadence and staleness tests; +- verify sampler and tracker failures never break the service loop they + observe. + +### Frontend + +- component-test loading, healthy, degraded, partial, stale, and unavailable + payloads; +- verify units and scopes are explicit (`Dispatch`, `Agents`, `Dependency`, + `Host`); +- verify polling pauses when hidden and stops on unmount; +- Playwright: open Settings, select Resources, observe a successful refresh, + expand a subsystem, and validate the mobile layout; +- capture and share a screenshot of the changed UI flow. + +### Performance acceptance + +Before shipping, compare an idle service with and without metrics enabled: + +- sampler CPU should remain negligible at the five-second cadence; +- history must remain bounded regardless of uptime; +- only one OS process listing may run per sample; +- no filesystem recursion or database-size query may run per HTTP request; +- opening multiple dashboard tabs must not multiply backend sampling work; +- the resources endpoint p95 should be dominated by JSON serialization of the + cached snapshot, not collection work. + +## Recommended MVP cut + +Ship Phases 1 and 2 plus the process-tree portion of Phase 3 as the first +user-visible release. They provide immediate answers about server CPU/memory, +API/DB health, current workload, the loops most likely to explain degraded +behavior, and whether resource pressure belongs to the control plane or its +agents. Keep disk scanning and database-size reporting for the next increment; +they are useful capacity signals but less important for live diagnosis. + +The MVP is successful when an operator can distinguish these cases without +opening a terminal: + +- Dispatch itself is consuming CPU or memory; +- running agent processes are consuming the resources instead; +- Postgres is reachable but its pool is saturated; +- the event loop/API is slow; +- reconciliation, activity monitoring, or Git diff refreshes are failing or + stale; +- everything is healthy and the observed host pressure is outside Dispatch. diff --git a/e2e/settings.spec.ts b/e2e/settings.spec.ts index bb24e325..ee39a7ae 100644 --- a/e2e/settings.spec.ts +++ b/e2e/settings.spec.ts @@ -139,6 +139,42 @@ test.describe("Settings pane", () => { await expect(page.getByText("Browser extension connected")).toBeVisible(); }); + test("shows live service resources and expands subsystem details", async ({ + page, + }) => { + await loadApp(page); + + await page.getByTestId("settings-button").click(); + await page + .getByTestId("sidebar-shell") + .getByText("Resources", { exact: true }) + .click(); + + const dashboard = page.getByTestId("service-resources-dashboard"); + await expect(dashboard).toBeVisible({ timeout: 10_000 }); + await expect(page).toHaveURL(/\/settings\/resources$/); + await expect( + dashboard.getByTestId("resource-card-dispatch-cpu") + ).toBeVisible(); + await expect(dashboard.getByTestId("resource-card-database")).toBeVisible(); + await expect( + dashboard.getByText(/host load uses the right load axis/i) + ).toBeVisible(); + await expect( + dashboard.getByText("Browser streams", { exact: true }) + ).toHaveCount(0); + + const databaseRow = dashboard.getByTestId("subsystem-database"); + await databaseRow.click(); + await expect(databaseRow).toHaveAttribute("aria-expanded", "true"); + await expect(dashboard.getByText("pool total")).toBeVisible(); + + await dashboard.getByTestId("refresh-service-resources").click(); + await expect( + dashboard.getByTestId("refresh-service-resources") + ).toBeEnabled({ timeout: 10_000 }); + }); + test("agent type settings filter the create-agent dialog", async ({ page, }) => { From ad25ff50a14a81dccac5688ef49ca3bf0541b116 Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Wed, 15 Jul 2026 23:23:19 -0600 Subject: [PATCH 02/12] fix: close resources review gaps --- .../server/src/agents/diff-stats-refresher.ts | 50 ++++++--- .../src/observability/service-resources.ts | 41 +++++-- apps/server/src/shared/git/diff-stats.ts | 51 +++++++-- apps/server/test/diff-stats-refresher.test.ts | 104 ++++++++++++++++++ apps/server/test/service-resources.test.ts | 26 +++++ 5 files changed, 238 insertions(+), 34 deletions(-) diff --git a/apps/server/src/agents/diff-stats-refresher.ts b/apps/server/src/agents/diff-stats-refresher.ts index 9e37f9a3..2a34112e 100644 --- a/apps/server/src/agents/diff-stats-refresher.ts +++ b/apps/server/src/agents/diff-stats-refresher.ts @@ -1,6 +1,8 @@ import { - getDiffStats as defaultGetDiffStats, + getDiffStatsComputation as defaultGetDiffStatsComputation, type DiffStats, + type DiffStatsComputation, + type GetDiffStatsOptions, } from "../shared/git/diff-stats.js"; import type { SubsystemTracker } from "../observability/subsystem-tracker.js"; @@ -35,6 +37,8 @@ export type DiffStatsRefresherOptions = { getAgent: (id: string) => Promise; publishEvent: (event: DiffStatsChangedEvent) => void; computeDiffStats?: ComputeDiffStats; + /** Override Git command execution while retaining the default adapter. */ + runGitCommand?: GetDiffStatsOptions["runCommand"]; freshnessMs?: number; logger?: WarnLogger; tracker?: SubsystemTracker; @@ -60,7 +64,10 @@ export class DiffStatsRefresher { private readonly getAgent: (id: string) => Promise; private readonly publishEvent: (event: DiffStatsChangedEvent) => void; - private readonly computeDiffStats: ComputeDiffStats; + private readonly computeDiffStats: ( + worktreePath: string, + baseRef: string | null + ) => Promise; private readonly freshnessMs: number; private readonly logger: WarnLogger | null; private readonly tracker: SubsystemTracker | null; @@ -70,18 +77,18 @@ export class DiffStatsRefresher { constructor(options: DiffStatsRefresherOptions) { this.getAgent = options.getAgent; this.publishEvent = options.publishEvent; - this.computeDiffStats = - options.computeDiffStats ?? - (async (worktreePath, baseRef) => { - let commandError: unknown = null; - const stats = await defaultGetDiffStats(worktreePath, baseRef, { - onError: (error) => { - commandError = error; - }, - }); - if (commandError !== null) throw commandError; - return stats; - }); + const customComputeDiffStats = options.computeDiffStats; + this.computeDiffStats = customComputeDiffStats + ? async (worktreePath, baseRef) => { + const stats = await customComputeDiffStats(worktreePath, baseRef); + return stats + ? { kind: "success", stats } + : { kind: "no-data", stats: null }; + } + : (worktreePath, baseRef) => + defaultGetDiffStatsComputation(worktreePath, baseRef, { + runCommand: options.runGitCommand, + }); this.freshnessMs = options.freshnessMs ?? DEFAULT_FRESHNESS_MS; this.logger = options.logger ?? null; this.tracker = options.tracker ?? null; @@ -149,6 +156,7 @@ export class DiffStatsRefresher { private async refresh(agentId: string): Promise { const trackedRun = this.tracker?.start(); let nextStats: DiffStats | null = null; + let computation: DiffStatsComputation = { kind: "no-data", stats: null }; try { const agent = await this.getAgent(agentId); // Prefer the dispatch-managed worktreePath. Older rows can be missing @@ -172,7 +180,9 @@ export class DiffStatsRefresher { (agent?.worktreePath || gitContextWorktreePath ? DEFAULT_WORKTREE_BASE_BRANCH : null); - nextStats = await this.computeDiffStats(path, baseRef); + computation = await this.computeDiffStats(path, baseRef); + if (computation.kind === "failure") throw computation.error; + nextStats = computation.stats; } } catch (err) { trackedRun?.fail(err); @@ -183,7 +193,15 @@ export class DiffStatsRefresher { return; } - trackedRun?.succeed({ files: nextStats?.files ?? 0 }); + if (computation.kind === "partial") { + trackedRun?.fail(computation.error); + this.logger?.warn( + { err: computation.error, agentId }, + "Diff stats refreshed with a best-effort Git probe failure" + ); + } else { + trackedRun?.succeed({ files: nextStats?.files ?? 0 }); + } const previous = this.cache.has(agentId) ? this.cache.get(agentId) diff --git a/apps/server/src/observability/service-resources.ts b/apps/server/src/observability/service-resources.ts index 3673e78d..117e1be2 100644 --- a/apps/server/src/observability/service-resources.ts +++ b/apps/server/src/observability/service-resources.ts @@ -63,6 +63,8 @@ export type ServiceResourcesDeps = { getWorkloads: () => WorkloadSnapshot; subsystemTrackers: SubsystemTracker[]; processTreeSupported?: boolean; + /** Override the platform process probes in focused tests. */ + runProcessCommand?: typeof runCommand; }; type AgentProcessSnapshot = { @@ -603,15 +605,31 @@ export class ServiceResources { processes: AgentProcessSnapshot; runningAgentCount: number; }> { + let agents: Array<{ tmuxSession: string | null }>; + try { + agents = await this.deps.listAgentSessions(); + } catch { + return { + processes: { + ...this.agentProcesses, + sampledAt: Date.now(), + error: "Agent session sampling failed", + }, + runningAgentCount: this.runningAgentCount, + }; + } + + // Session ownership is platform-independent. Commit its fresh value even + // when the optional tmux/ps process probe below is unavailable or fails. + const runningAgentCount = agents.length; + if (!this.agentProcesses.supported) { + return { + processes: { ...this.agentProcesses, error: null }, + runningAgentCount, + }; + } + try { - const agents = await this.deps.listAgentSessions(); - const runningAgentCount = agents.length; - if (!this.agentProcesses.supported) { - return { - processes: { ...this.agentProcesses, error: null }, - runningAgentCount, - }; - } const sessions = new Set( agents .map((agent) => agent.tmuxSession?.trim()) @@ -631,13 +649,14 @@ export class ServiceResources { }; } + const run = this.deps.runProcessCommand ?? runCommand; const [panes, processes] = await Promise.all([ - runCommand( + run( "tmux", ["list-panes", "-a", "-F", "#{session_name}\t#{pane_pid}"], { allowedExitCodes: [0, 1], timeoutMs: 3_000 } ), - runCommand("ps", ["-axo", "pid=,ppid=,%cpu=,rss="], { + run("ps", ["-axo", "pid=,ppid=,%cpu=,rss="], { timeoutMs: 3_000, }), ]); @@ -695,7 +714,7 @@ export class ServiceResources { sampledAt: Date.now(), error: "Process sampling failed", }, - runningAgentCount: this.runningAgentCount, + runningAgentCount, }; } } diff --git a/apps/server/src/shared/git/diff-stats.ts b/apps/server/src/shared/git/diff-stats.ts index bfef6a10..6c620885 100644 --- a/apps/server/src/shared/git/diff-stats.ts +++ b/apps/server/src/shared/git/diff-stats.ts @@ -9,6 +9,12 @@ export type DiffStats = { computedAt: number; }; +export type DiffStatsComputation = + | { kind: "success"; stats: DiffStats } + | { kind: "no-data"; stats: null } + | { kind: "partial"; stats: DiffStats; error: unknown } + | { kind: "failure"; stats: null; error: unknown }; + const GIT_TIMEOUT_MS = 15_000; type CommandRunner = ( @@ -48,22 +54,50 @@ export async function getDiffStats( baseRef: string | null, options: GetDiffStatsOptions = {} ): Promise { + const result = await getDiffStatsComputation(worktreePath, baseRef, options); + return result.stats; +} + +/** + * Internal, discriminated form used by observability-aware callers. It keeps + * best-effort probe failures distinct from fatal Git failures without changing + * the public getDiffStats null/usable-stats contract. + */ +export async function getDiffStatsComputation( + worktreePath: string, + baseRef: string | null, + options: GetDiffStatsOptions = {} +): Promise { const run = options.runCommand ?? runCommand; const includeUncommitted = options.includeUncommitted !== false; + const probeErrors: unknown[] = []; + const recordError = (error: unknown) => { + probeErrors.push(error); + options.onError?.(error); + }; try { const resolvedBase = await resolveBaseRef(worktreePath, baseRef, { runCommand: run, - onError: options.onError, + onError: recordError, }); - if (!resolvedBase) return null; + if (!resolvedBase) { + return probeErrors.length > 0 + ? { kind: "failure", stats: null, error: probeErrors[0] } + : { kind: "no-data", stats: null }; + } const mergeBase = await run( "git", ["-C", worktreePath, "merge-base", "HEAD", resolvedBase], { allowedExitCodes: [0, 1, 128], timeoutMs: 5_000 } ); + if (mergeBase.exitCode === 128) { + const error = new Error("Git merge-base failed"); + recordError(error); + return { kind: "failure", stats: null, error }; + } if (mergeBase.exitCode !== 0 || !mergeBase.stdout.trim()) { - return null; + return { kind: "no-data", stats: null }; } const mergeBaseSha = mergeBase.stdout.trim(); @@ -89,7 +123,7 @@ export async function getDiffStats( worktreePath, trackedPaths, run, - options.onError + recordError ); let added = 0; @@ -121,15 +155,18 @@ export async function getDiffStats( added += lines; } - return { + const stats = { added, deleted, files: seenFiles.size, computedAt: Date.now(), }; + return probeErrors.length > 0 + ? { kind: "partial", stats, error: probeErrors[0] } + : { kind: "success", stats }; } catch (error) { - options.onError?.(error); - return null; + recordError(error); + return { kind: "failure", stats: null, error }; } } diff --git a/apps/server/test/diff-stats-refresher.test.ts b/apps/server/test/diff-stats-refresher.test.ts index 4bbde3cb..a913c7cc 100644 --- a/apps/server/test/diff-stats-refresher.test.ts +++ b/apps/server/test/diff-stats-refresher.test.ts @@ -6,10 +6,19 @@ import { type DiffStatsChangedEvent, } from "../src/agents/diff-stats-refresher.js"; import type { DiffStats } from "../src/shared/git/diff-stats.js"; +import type { RunCommandResult } from "../src/shared/lib/run-command.js"; import { SubsystemTracker } from "../src/observability/subsystem-tracker.js"; type AgentMap = Map; +function gitResult( + exitCode: number, + stdout = "", + stderr = "" +): RunCommandResult { + return { exitCode, stdout, stderr }; +} + function setupAgents(entries: Array<[string, DiffStatsAgent]>): AgentMap { return new Map(entries); } @@ -428,4 +437,99 @@ describe("DiffStatsRefresher", () => { lastError: "Operation failed", }); }); + + it("marks merge-base exit 128 as a default-adapter failure", async () => { + const agents = setupAgents([ + ["a1", { worktreePath: "/tmp/wt", cwd: null, baseBranch: "main" }], + ]); + const events: DiffStatsChangedEvent[] = []; + const tracker = new SubsystemTracker({ + id: "git", + label: "Git", + description: "Refreshes diffs", + }); + const runGitCommand = vi.fn( + async (_command: string, args: string[]): Promise => { + const key = args.join(" "); + if (key === "-C /tmp/wt rev-parse --verify --quiet origin/main") { + return gitResult(0, "origin/main\n"); + } + if (key === "-C /tmp/wt merge-base HEAD origin/main") { + return gitResult(128, "", "fatal: bad revision"); + } + throw new Error(`Unexpected command: ${key}`); + } + ); + const refresher = new DiffStatsRefresher({ + getAgent: async (id) => agents.get(id) ?? null, + publishEvent: (event) => events.push(event), + runGitCommand, + tracker, + }); + + await refresher.signal("a1"); + + expect(events).toHaveLength(0); + expect(tracker.snapshot()).toMatchObject({ + state: "degraded", + failures: 1, + lastError: "Operation failed", + }); + }); + + it("publishes usable default-adapter stats after check-ignore fails", async () => { + const agents = setupAgents([ + ["a1", { worktreePath: "/tmp/wt", cwd: null, baseBranch: "main" }], + ]); + const events: DiffStatsChangedEvent[] = []; + const warn = vi.fn(); + const tracker = new SubsystemTracker({ + id: "git", + label: "Git", + description: "Refreshes diffs", + }); + const runGitCommand = vi.fn( + async (_command: string, args: string[]): Promise => { + const key = args.join(" "); + if (key === "-C /tmp/wt rev-parse --verify --quiet origin/main") { + return gitResult(0, "origin/main\n"); + } + if (key === "-C /tmp/wt merge-base HEAD origin/main") { + return gitResult(0, "abcd1234\n"); + } + if (key === "-C /tmp/wt diff abcd1234 --numstat") { + return gitResult(0, "3\t1\tsrc/foo.ts\n"); + } + if (key === "-C /tmp/wt ls-files --others --exclude-standard") { + return gitResult(0); + } + if (args.includes("check-ignore")) { + throw new Error("check-ignore unavailable"); + } + throw new Error(`Unexpected command: ${key}`); + } + ); + const refresher = new DiffStatsRefresher({ + getAgent: async (id) => agents.get(id) ?? null, + publishEvent: (event) => events.push(event), + runGitCommand, + logger: { warn }, + tracker, + }); + + await refresher.signal("a1"); + + expect(refresher.getStats("a1")).toMatchObject({ + added: 3, + deleted: 1, + files: 1, + }); + expect(events).toHaveLength(1); + expect(warn).toHaveBeenCalled(); + expect(tracker.snapshot()).toMatchObject({ + state: "degraded", + failures: 1, + lastError: "Operation failed", + }); + }); }); diff --git a/apps/server/test/service-resources.test.ts b/apps/server/test/service-resources.test.ts index 797c2171..98302e66 100644 --- a/apps/server/test/service-resources.test.ts +++ b/apps/server/test/service-resources.test.ts @@ -97,6 +97,32 @@ describe("ServiceResources", () => { resources.stop(); }); + it("keeps a fresh running-agent count when process probing fails", async () => { + const runProcessCommand = vi.fn(async () => { + throw new Error("tmux unavailable"); + }); + const resources = new ServiceResources({ + pool: createPool(), + listAgentSessions: async () => [ + { tmuxSession: "agent-one" }, + { tmuxSession: "agent-two" }, + ], + getWorkloads: workloads, + subsystemTrackers: [], + processTreeSupported: true, + runProcessCommand, + }); + + resources.start(); + await vi.advanceTimersByTimeAsync(0); + expect(resources.getSnapshot()).toMatchObject({ + capabilities: { processTreeMetrics: "error" }, + current: { workloads: { runningAgents: 2 } }, + }); + expect(runProcessCommand).toHaveBeenCalled(); + resources.stop(); + }); + it("bounds request timing storage and finalizes requests exactly once", () => { const resources = new ServiceResources({ pool: createPool(), From 19ed02ab4b2ed10966dc1bae13c3f12258580c26 Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Thu, 16 Jul 2026 09:03:26 -0600 Subject: [PATCH 03/12] fix: unref database probe timeout --- apps/server/src/observability/service-resources.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/observability/service-resources.ts b/apps/server/src/observability/service-resources.ts index 117e1be2..49794114 100644 --- a/apps/server/src/observability/service-resources.ts +++ b/apps/server/src/observability/service-resources.ts @@ -585,6 +585,7 @@ export class ServiceResources { pool: this.poolSnapshot(), }); }, DATABASE_PROBE_TIMEOUT_MS); + timeout.unref?.(); void probe.then((snapshot) => { clearTimeout(timeout); resolve(snapshot); From e92b5181c3e2e79e1ab7c9d85c5e82d848f5ecf7 Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Thu, 16 Jul 2026 09:24:09 -0600 Subject: [PATCH 04/12] fix: polish resources dashboard semantics --- .../app/service-resources-chart.tsx | 4 +- .../app/service-resources-dashboard.tsx | 42 ++++++++++++------- .../app/service-resources-settings.tsx | 19 ++++++--- .../app/service-resources-subsystems.tsx | 6 ++- e2e/settings.spec.ts | 11 +++++ 5 files changed, 59 insertions(+), 23 deletions(-) diff --git a/apps/web/src/components/app/service-resources-chart.tsx b/apps/web/src/components/app/service-resources-chart.tsx index af389d2f..7a8ac83f 100644 --- a/apps/web/src/components/app/service-resources-chart.tsx +++ b/apps/web/src/components/app/service-resources-chart.tsx @@ -101,10 +101,10 @@ export function ResourceChart({ `${Number(value).toFixed(1)} load`} + tickFormatter={(value) => Number(value).toFixed(1)} /> )} @@ -123,12 +137,16 @@ export function ServiceResourcesDashboard({ value={ current.agents.rssBytes === null ? "Unavailable" - : formatBytes(current.agents.rssBytes) + : current.agents.processCount === 0 + ? "None" + : formatBytes(current.agents.rssBytes) } detail={ current.agents.cpuPercent === null ? "Process-tree sampling unavailable" - : `${current.agents.cpuPercent.toFixed(1)}% CPU · ${current.agents.processCount ?? 0} processes` + : current.agents.processCount === 0 + ? "No agent processes running" + : `${current.agents.cpuPercent.toFixed(1)}% CPU · ${current.agents.processCount ?? 0} processes` } scope="Agents" /> @@ -150,7 +168,7 @@ export function ServiceResourcesDashboard({ icon={HardDrive} label="Host memory free" value={formatBytes(current.host.freeMemoryBytes)} - detail={`${current.host.load1.toFixed(2)} one-minute load`} + detail={`${current.host.load1.toFixed(2)} load / ${current.host.cpuCount} CPUs`} scope="Host" /> @@ -186,19 +204,15 @@ export function ServiceResourcesDashboard({ - {[ - ["Running agents", current.workloads.runningAgents], - ["SSE clients", current.workloads.sseClients], - ["Terminal viewers", current.workloads.terminalViewers], - ["Scheduled jobs", current.workloads.scheduledJobs], - ["Git refreshes", current.workloads.gitRefreshesInFlight], - ].map(([label, value]) => ( + {workloadItems.map((item) => (
-
{label}
-
{value}
+
+ {item.label} +
+
{item.value}
))} diff --git a/apps/web/src/components/app/service-resources-settings.tsx b/apps/web/src/components/app/service-resources-settings.tsx index 71b5b9ca..caf3c8ab 100644 --- a/apps/web/src/components/app/service-resources-settings.tsx +++ b/apps/web/src/components/app/service-resources-settings.tsx @@ -45,14 +45,23 @@ export function ServiceResourcesSettings(): JSX.Element { and host.

-
+
{stale && ( Data is stale )} - - {dataUpdatedAt > 0 - ? `Updated ${new Date(dataUpdatedAt).toLocaleTimeString()}` - : ""} + + {dataUpdatedAt > 0 && ( + <> + Updated + {new Date(dataUpdatedAt).toLocaleTimeString([], { + hour: "numeric", + minute: "2-digit", + })} + + )} -
diff --git a/apps/web/src/hooks/use-service-resources.ts b/apps/web/src/hooks/use-service-resources.ts index ff9b72c2..da509184 100644 --- a/apps/web/src/hooks/use-service-resources.ts +++ b/apps/web/src/hooks/use-service-resources.ts @@ -129,7 +129,7 @@ export function useServiceResources(window: ResourceWindow) { `/api/v1/system/resources?window=${encodeURIComponent(window)}` ), refetchInterval: () => - typeof document !== "undefined" && document.hidden ? false : 5_000, + typeof document !== "undefined" && document.hidden ? false : 15_000, refetchIntervalInBackground: false, placeholderData: (previous) => previous, }); diff --git a/e2e/settings.spec.ts b/e2e/settings.spec.ts index d9a3a2f6..47aa2711 100644 --- a/e2e/settings.spec.ts +++ b/e2e/settings.spec.ts @@ -167,6 +167,12 @@ test.describe("Settings pane", () => { await expect( dashboard.getByText(/host load uses the right load axis/i) ).toBeVisible(); + await expect( + dashboard.getByText(/History resets when Dispatch restarts/i) + ).toBeVisible(); + await expect( + dashboard.getByTestId("refresh-service-resources") + ).toHaveCount(0); await expect( dashboard.getByText("Browser streams", { exact: true }) ).toHaveCount(0); @@ -180,11 +186,6 @@ test.describe("Settings pane", () => { dashboard.getByTestId("subsystem-stat-trend-database-poolTotal") ).toBeVisible(); - await dashboard.getByTestId("refresh-service-resources").click(); - await expect( - dashboard.getByTestId("refresh-service-resources") - ).toBeEnabled({ timeout: 10_000 }); - await page.setViewportSize({ width: 390, height: 844 }); await expect(dashboard.getByTestId("resources-updated-at")).toBeVisible(); }); From 57543e5c8cccecb80f3eb8b3e8dffe4e9c7eabdb Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Thu, 16 Jul 2026 22:30:07 -0600 Subject: [PATCH 09/12] feat: add runtime resource collection toggle --- .../service-resources-settings.ts | 23 +++++ .../src/observability/service-resources.ts | 28 +++++- apps/server/src/routes/resources.ts | 17 +++- apps/server/src/server.ts | 7 +- apps/server/test/service-resources.test.ts | 43 +++++++++ apps/server/test/system-routes.test.ts | 52 +++++++++++ .../app/service-resources-settings.tsx | 93 +++++++++++++++---- apps/web/src/hooks/use-service-resources.ts | 43 ++++++++- e2e/settings.spec.ts | 21 ++++- 9 files changed, 299 insertions(+), 28 deletions(-) create mode 100644 apps/server/src/observability/service-resources-settings.ts diff --git a/apps/server/src/observability/service-resources-settings.ts b/apps/server/src/observability/service-resources-settings.ts new file mode 100644 index 00000000..cf6041ba --- /dev/null +++ b/apps/server/src/observability/service-resources-settings.ts @@ -0,0 +1,23 @@ +import type { Pool } from "pg"; + +import { getSetting, setSetting } from "../db/settings.js"; + +export const SERVICE_RESOURCES_COLLECTION_KEY = + "service_resources_collection_enabled"; + +export async function readServiceResourcesCollectionEnabled( + pool: Pool +): Promise { + return (await getSetting(pool, SERVICE_RESOURCES_COLLECTION_KEY)) === "true"; +} + +export async function writeServiceResourcesCollectionEnabled( + pool: Pool, + enabled: boolean +): Promise { + await setSetting( + pool, + SERVICE_RESOURCES_COLLECTION_KEY, + enabled ? "true" : "false" + ); +} diff --git a/apps/server/src/observability/service-resources.ts b/apps/server/src/observability/service-resources.ts index 3429c43b..38369576 100644 --- a/apps/server/src/observability/service-resources.ts +++ b/apps/server/src/observability/service-resources.ts @@ -52,6 +52,7 @@ type HttpSnapshot = { export type HttpRequestToken = { startedAt: number; finished: boolean; + generation: number; }; export type WorkloadSnapshot = { @@ -90,6 +91,7 @@ type AgentProcessSnapshot = { }; export type ServiceResourcesResponse = { + collectionEnabled: boolean; generatedAt: number; processStartedAt: number; availableHistoryMs: number; @@ -271,14 +273,37 @@ export class ServiceResources { this.eventLoopDelay.disable(); } + setCollectionEnabled(enabled: boolean): void { + if (enabled) { + this.start(); + return; + } + this.stop(); + this.samples = []; + this.httpBuckets = []; + this.httpInFlight = 0; + } + + isCollectionEnabled(): boolean { + return this.running; + } + requestStarted(): HttpRequestToken { + if (!this.running) { + return { startedAt: 0, finished: true, generation: this.generation }; + } this.httpInFlight += 1; - return { startedAt: performance.now(), finished: false }; + return { + startedAt: performance.now(), + finished: false, + generation: this.generation, + }; } requestFinished(token: HttpRequestToken, statusCode: number): void { if (token.finished) return; token.finished = true; + if (!this.running || token.generation !== this.generation) return; this.httpInFlight = Math.max(0, this.httpInFlight - 1); const now = Date.now(); const bucket = this.getHttpBucket(now); @@ -348,6 +373,7 @@ export class ServiceResources { const series = this.samples.filter((sample) => now - sample.at <= windowMs); return { + collectionEnabled: this.running, generatedAt: now, processStartedAt: now - process.uptime() * 1000, availableHistoryMs: diff --git a/apps/server/src/routes/resources.ts b/apps/server/src/routes/resources.ts index abeee2b7..10d05128 100644 --- a/apps/server/src/routes/resources.ts +++ b/apps/server/src/routes/resources.ts @@ -1,10 +1,12 @@ import type { FastifyInstance } from "fastify"; +import type { Pool } from "pg"; import type { ServiceResources } from "../observability/service-resources.js"; +import { writeServiceResourcesCollectionEnabled } from "../observability/service-resources-settings.js"; export async function registerResourceRoutes( app: FastifyInstance, - resources: ServiceResources + deps: { pool: Pool; resources: ServiceResources } ): Promise { app.get("/api/v1/system/resources", async (request, reply) => { const query = request.query as { window?: unknown }; @@ -17,6 +19,17 @@ export async function registerResourceRoutes( if (!windowMs) { return reply.code(400).send({ error: 'window must be "15m" or "1h".' }); } - return resources.getSnapshot(windowMs); + return deps.resources.getSnapshot(windowMs); + }); + + app.post("/api/v1/system/resources/settings", async (request, reply) => { + const body = request.body as { enabled?: unknown } | null; + if (typeof body?.enabled !== "boolean") { + return reply.code(400).send({ error: "enabled must be a boolean." }); + } + + await writeServiceResourcesCollectionEnabled(deps.pool, body.enabled); + deps.resources.setCollectionEnabled(body.enabled); + return { collectionEnabled: deps.resources.isCollectionEnabled() }; }); } diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 30f35e84..1e1852eb 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -153,6 +153,7 @@ import { ServiceResources, type HttpRequestToken, } from "./observability/service-resources.js"; +import { readServiceResourcesCollectionEnabled } from "./observability/service-resources-settings.js"; const config = loadConfig(); const app = Fastify({ @@ -623,7 +624,7 @@ async function registerRoutes() { rewriteForColor: (color) => staticTheme.rewriteForColor(color as IconColor), publishUiEvent: (event) => uiEventBroker.publish(event as UiEvent), }); - await registerResourceRoutes(app, serviceResources); + await registerResourceRoutes(app, { pool, resources: serviceResources }); await registerBrainRoutes(app, { brainStore, @@ -792,7 +793,9 @@ export async function initializeApp(options?: { await runMigrations(); } config.authToken = await getOrCreateAuthToken(pool); - serviceResources.start(); + serviceResources.setCollectionEnabled( + await readServiceResourcesCollectionEnabled(pool) + ); const shouldReconcileState = options?.reconcileState ?? true; if (shouldReconcileState) { await agentManager.reconcileAgents(); diff --git a/apps/server/test/service-resources.test.ts b/apps/server/test/service-resources.test.ts index b3c69530..26e88787 100644 --- a/apps/server/test/service-resources.test.ts +++ b/apps/server/test/service-resources.test.ts @@ -163,6 +163,7 @@ describe("ServiceResources", () => { subsystemTrackers: [], processTreeSupported: false, }); + resources.start(); for (let index = 0; index < 5_000; index += 1) { const token = resources.requestStarted(); @@ -176,6 +177,48 @@ describe("ServiceResources", () => { inFlight: 0, errorRatePercent: 10, }); + resources.stop(); + }); + + it("disables sampling and clears retained observations at runtime", async () => { + const resources = new ServiceResources({ + pool: createPool(), + listAgentSessions: async () => [], + getWorkloads: workloads, + subsystemTrackers: [], + processTreeSupported: false, + }); + + const disabledToken = resources.requestStarted(); + resources.requestFinished(disabledToken, 200); + expect(resources.getSnapshot()).toMatchObject({ + collectionEnabled: false, + series: [], + }); + expect(resources.getHttpObservationStorageSize()).toBe(0); + + resources.setCollectionEnabled(true); + await vi.advanceTimersByTimeAsync(0); + const enabledToken = resources.requestStarted(); + resources.requestFinished(enabledToken, 200); + expect(resources.getSnapshot().collectionEnabled).toBe(true); + expect(resources.getSnapshot().series).toHaveLength(1); + expect(resources.getHttpObservationStorageSize()).toBe(1); + + resources.setCollectionEnabled(false); + expect(resources.getSnapshot()).toMatchObject({ + collectionEnabled: false, + series: [], + }); + expect(resources.getHttpObservationStorageSize()).toBe(0); + + resources.setCollectionEnabled(true); + const staleToken = resources.requestStarted(); + resources.setCollectionEnabled(false); + resources.setCollectionEnabled(true); + resources.requestFinished(staleToken, 200); + expect(resources.getHttpObservationStorageSize()).toBe(0); + resources.setCollectionEnabled(false); }); it("degrades owner subsystems when recent writes or polls fail", async () => { diff --git a/apps/server/test/system-routes.test.ts b/apps/server/test/system-routes.test.ts index 69f3f99a..9f2c70e2 100644 --- a/apps/server/test/system-routes.test.ts +++ b/apps/server/test/system-routes.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import { readServiceResourcesCollectionEnabled } from "../src/observability/service-resources-settings.js"; import { useInjectApp } from "./helpers/inject-app.js"; vi.mock("../src/shared/lib/run-command.js", () => ({ @@ -72,6 +73,7 @@ describe("GET /api/v1/system/resources", () => { }); expect(res.statusCode).toBe(200); const body = res.json(); + expect(body.collectionEnabled).toBe(false); expect(body.sampleIntervalMs).toBe(5_000); expect(body.current.server.rssBytes).toBeGreaterThan(0); expect(body.current.database.pool.max).toBeGreaterThan(0); @@ -106,6 +108,56 @@ describe("GET /api/v1/system/resources", () => { expect(res.statusCode).toBe(400); expect(res.json().error).toMatch(/window/); }); + + it("toggles resource collection at runtime and persists the setting", async () => { + const enable = await ctx.app.inject({ + method: "POST", + url: "/api/v1/system/resources/settings", + headers: { cookie: sessionCookie }, + payload: { enabled: true }, + }); + expect(enable.statusCode).toBe(200); + expect(enable.json()).toEqual({ collectionEnabled: true }); + expect(await readServiceResourcesCollectionEnabled(ctx.pool)).toBe(true); + + const enabledSnapshot = await ctx.app.inject({ + method: "GET", + url: "/api/v1/system/resources", + headers: { cookie: sessionCookie }, + }); + expect(enabledSnapshot.json().collectionEnabled).toBe(true); + + const disable = await ctx.app.inject({ + method: "POST", + url: "/api/v1/system/resources/settings", + headers: { cookie: sessionCookie }, + payload: { enabled: false }, + }); + expect(disable.statusCode).toBe(200); + expect(disable.json()).toEqual({ collectionEnabled: false }); + expect(await readServiceResourcesCollectionEnabled(ctx.pool)).toBe(false); + + const disabledSnapshot = await ctx.app.inject({ + method: "GET", + url: "/api/v1/system/resources", + headers: { cookie: sessionCookie }, + }); + expect(disabledSnapshot.json()).toMatchObject({ + collectionEnabled: false, + series: [], + }); + }); + + it("rejects invalid resource collection settings", async () => { + const res = await ctx.app.inject({ + method: "POST", + url: "/api/v1/system/resources/settings", + headers: { cookie: sessionCookie }, + payload: { enabled: "yes" }, + }); + expect(res.statusCode).toBe(400); + expect(res.json().error).toMatch(/enabled/); + }); }); describe("GET /api/v1/system/path-info", () => { diff --git a/apps/web/src/components/app/service-resources-settings.tsx b/apps/web/src/components/app/service-resources-settings.tsx index 29841f8b..b8a8916e 100644 --- a/apps/web/src/components/app/service-resources-settings.tsx +++ b/apps/web/src/components/app/service-resources-settings.tsx @@ -1,8 +1,10 @@ import { useState } from "react"; +import { Activity } from "lucide-react"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Card, CardContent } from "@/components/ui/card"; +import { Checkbox } from "@/components/ui/checkbox"; import { Select, SelectContent, @@ -12,6 +14,7 @@ import { } from "@/components/ui/select"; import { type ResourceWindow, + useSetServiceResourcesCollection, useServiceResources, } from "@/hooks/use-service-resources"; import { ServiceResourcesDashboard } from "./service-resources-dashboard"; @@ -20,7 +23,12 @@ import { stateBadgeVariant, stateLabel } from "./service-resources-format"; export function ServiceResourcesSettings(): JSX.Element { const [window, setWindow] = useState("1h"); const { data, error, refetch, dataUpdatedAt } = useServiceResources(window); - const stale = dataUpdatedAt > 0 && Date.now() - dataUpdatedAt > 45_000; + const collectionMutation = useSetServiceResourcesCollection(window); + const collectionEnabled = data?.collectionEnabled ?? false; + const stale = + collectionEnabled && + dataUpdatedAt > 0 && + Date.now() - dataUpdatedAt > 45_000; return (

Service resources

{data && ( - - {stateLabel(data.overall.state)} + + {stateLabel( + collectionEnabled ? data.overall.state : "disabled" + )} )}

- Live health and resource use for Dispatch, its agents, dependencies, - and host. History resets when Dispatch restarts. + Health and resource insights for Dispatch, its agents, dependencies, + and host. Collected history resets when Dispatch restarts.

{stale && ( Data is stale )} - - {dataUpdatedAt > 0 && ( - <> - Updated - {new Date(dataUpdatedAt).toLocaleTimeString([], { - hour: "numeric", - minute: "2-digit", - })} - - )} - + {collectionEnabled && ( + + {dataUpdatedAt > 0 && ( + <> + Updated + {new Date(dataUpdatedAt).toLocaleTimeString([], { + hour: "numeric", + minute: "2-digit", + })} + + )} + + )} + setWindow(value as ResourceWindow)} @@ -107,6 +88,29 @@ export function ServiceResourcesSettings(): JSX.Element {
+
+ + collectionMutation.mutate(checked)} + disabled={!data || collectionMutation.isPending} + aria-label="Collect resource metrics" + data-testid="resource-collection-toggle" + /> +
+ {collectionMutation.error && (

{collectionMutation.error.message} @@ -131,24 +135,6 @@ export function ServiceResourcesSettings(): JSX.Element { )} - {data && !collectionEnabled && ( - - -

-
- -
-
- Metric collection is off -
-

- Turn on Collect metrics to sample service resources every 5 - seconds and retain up to one hour of in-memory history. -

-
- - - )} {data && collectionEnabled && } ); diff --git a/apps/web/src/components/ui/switch.tsx b/apps/web/src/components/ui/switch.tsx new file mode 100644 index 00000000..8bdb26f8 --- /dev/null +++ b/apps/web/src/components/ui/switch.tsx @@ -0,0 +1,30 @@ +import * as React from "react"; +import * as SwitchPrimitive from "@radix-ui/react-switch"; + +import { cn } from "@/lib/utils"; + +const Switch = React.forwardRef< + React.ElementRef, + React.ComponentPropsWithoutRef +>(({ className, ...props }, ref) => ( + + + +)); +Switch.displayName = SwitchPrimitive.Root.displayName; + +export { Switch }; diff --git a/e2e/settings.spec.ts b/e2e/settings.spec.ts index 80e2cdba..60227133 100644 --- a/e2e/settings.spec.ts +++ b/e2e/settings.spec.ts @@ -163,9 +163,6 @@ test.describe("Settings pane", () => { "resource-collection-toggle" ); await expect(collectionToggle).not.toBeChecked(); - await expect( - dashboard.getByTestId("resource-collection-disabled") - ).toBeVisible(); await expect( dashboard.getByTestId("resource-card-dispatch-cpu") ).toHaveCount(0); @@ -176,9 +173,13 @@ test.describe("Settings pane", () => { dashboard.getByTestId("resource-card-dispatch-cpu") ).toBeVisible({ timeout: 10_000 }); await expect(dashboard.getByTestId("resource-card-database")).toBeVisible(); + const agentProcessesCard = dashboard.getByTestId( + "resource-card-agent-processes" + ); + await expect(agentProcessesCard).toBeVisible(); await expect( - dashboard.getByTestId("resource-card-agent-processes").getByText("None") - ).toBeVisible(); + agentProcessesCard.getByText("0 B", { exact: true }) + ).toHaveCount(0); await expect(dashboard.getByText(/load \/ \d+ CPUs/)).toBeVisible(); await expect(dashboard.getByText("Connected browsers")).toBeVisible(); await expect(dashboard.getByText("Active terminal views")).toBeVisible(); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7641ba31..21262d3b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -138,6 +138,9 @@ importers: "@radix-ui/react-slot": specifier: ^1.1.0 version: 1.2.4(@types/react@18.3.28)(react@18.3.1) + "@radix-ui/react-switch": + specifier: ^1.2.6 + version: 1.2.6(@types/react-dom@18.3.7(@types/react@18.3.28))(@types/react@18.3.28)(react-dom@18.3.1(react@18.3.1))(react@18.3.1) "@radix-ui/react-tooltip": specifier: ^1.2.8 version: 1.2.8(@types/react-dom@18.3.7(@types/react@18.3.28))(@types/react@18.3.28)(react-dom@18.3.1(react@18.3.1))(react@18.3.1) @@ -2842,6 +2845,22 @@ packages: "@types/react": optional: true + "@radix-ui/react-switch@1.2.6": + resolution: + { + integrity: sha512-bByzr1+ep1zk4VubeEVViV592vu2lHE2BZY5OnzehZqOOgogN80+mNtCqPkhn2gklJqOpxWgPoYTSnhBCqpOXQ==, + } + peerDependencies: + "@types/react": "*" + "@types/react-dom": "*" + react: ^16.8 || ^17.0 || ^18.0 || ^19.0 || ^19.0.0-rc + react-dom: ^16.8 || ^17.0 || ^18.0 || ^19.0 || ^19.0.0-rc + peerDependenciesMeta: + "@types/react": + optional: true + "@types/react-dom": + optional: true + "@radix-ui/react-tooltip@1.2.8": resolution: { @@ -11425,6 +11444,21 @@ snapshots: optionalDependencies: "@types/react": 18.3.28 + "@radix-ui/react-switch@1.2.6(@types/react-dom@18.3.7(@types/react@18.3.28))(@types/react@18.3.28)(react-dom@18.3.1(react@18.3.1))(react@18.3.1)": + dependencies: + "@radix-ui/primitive": 1.1.3 + "@radix-ui/react-compose-refs": 1.1.2(@types/react@18.3.28)(react@18.3.1) + "@radix-ui/react-context": 1.1.2(@types/react@18.3.28)(react@18.3.1) + "@radix-ui/react-primitive": 2.1.3(@types/react-dom@18.3.7(@types/react@18.3.28))(@types/react@18.3.28)(react-dom@18.3.1(react@18.3.1))(react@18.3.1) + "@radix-ui/react-use-controllable-state": 1.2.2(@types/react@18.3.28)(react@18.3.1) + "@radix-ui/react-use-previous": 1.1.1(@types/react@18.3.28)(react@18.3.1) + "@radix-ui/react-use-size": 1.1.1(@types/react@18.3.28)(react@18.3.1) + react: 18.3.1 + react-dom: 18.3.1(react@18.3.1) + optionalDependencies: + "@types/react": 18.3.28 + "@types/react-dom": 18.3.7(@types/react@18.3.28) + "@radix-ui/react-tooltip@1.2.8(@types/react-dom@18.3.7(@types/react@18.3.28))(@types/react@18.3.28)(react-dom@18.3.1(react@18.3.1))(react@18.3.1)": dependencies: "@radix-ui/primitive": 1.1.3 From 566f49d8f8052b5194497e0d03a2f361093cdcc5 Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Fri, 17 Jul 2026 08:52:44 -0600 Subject: [PATCH 11/12] fix: address resource dashboard UX review --- .../app/service-resources-chart.tsx | 7 ++++ .../app/service-resources-settings.tsx | 4 +- .../app/service-resources-subsystems.tsx | 2 +- apps/web/src/components/ui/switch.tsx | 2 +- apps/web/src/hooks/use-service-resources.ts | 41 +++++++++++-------- e2e/settings.spec.ts | 4 +- 6 files changed, 37 insertions(+), 23 deletions(-) diff --git a/apps/web/src/components/app/service-resources-chart.tsx b/apps/web/src/components/app/service-resources-chart.tsx index 7a8ac83f..96ecb6ca 100644 --- a/apps/web/src/components/app/service-resources-chart.tsx +++ b/apps/web/src/components/app/service-resources-chart.tsx @@ -9,6 +9,8 @@ import { } from "@/components/ui/card"; import { ChartContainer, + ChartLegend, + ChartLegendContent, ChartTooltip, ChartTooltipContent, type ChartConfig, @@ -123,6 +125,11 @@ export function ResourceChart({ new Date(Number(value)).toLocaleTimeString() } /> + + } + /> {allKeys.map((key) => ( ("1h"); const { data, error, refetch, dataUpdatedAt } = useServiceResources(window); - const collectionMutation = useSetServiceResourcesCollection(window); + const collectionMutation = useSetServiceResourcesCollection(); const collectionEnabled = data?.collectionEnabled ?? false; const stale = collectionEnabled && @@ -51,7 +51,7 @@ export function ServiceResourcesSettings(): JSX.Element {
{stale ? ( diff --git a/apps/web/src/components/app/service-resources-subsystems.tsx b/apps/web/src/components/app/service-resources-subsystems.tsx index a1860721..5df6d7c0 100644 --- a/apps/web/src/components/app/service-resources-subsystems.tsx +++ b/apps/web/src/components/app/service-resources-subsystems.tsx @@ -158,7 +158,7 @@ function SubsystemRow({