Add service resources dashboard

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<void>;
+  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<string, ActivityState>();
 
   return {
-    async check(): Promise<void> {
+    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 {

apps/server/src/agents/diff-stats-refresher.ts

@@ -1,7 +1,10 @@
 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";
 
 export type DiffStatsAgent = {
   worktreePath: string | null;
@@ -34,8 +37,11 @@ export type DiffStatsRefresherOptions = {
   getAgent: (id: string) => Promise<DiffStatsAgent | null>;
   publishEvent: (event: DiffStatsChangedEvent) => void;
   computeDiffStats?: ComputeDiffStats;
+  /** Override Git command execution while retaining the default adapter. */
+  runGitCommand?: GetDiffStatsOptions["runCommand"];
   freshnessMs?: number;
   logger?: WarnLogger;
+  tracker?: SubsystemTracker;
 };
 
 const DEFAULT_FRESHNESS_MS = 3_000;
@@ -58,25 +64,47 @@ export class DiffStatsRefresher {
 
   private readonly getAgent: (id: string) => Promise<DiffStatsAgent | null>;
   private readonly publishEvent: (event: DiffStatsChangedEvent) => void;
-  private readonly computeDiffStats: ComputeDiffStats;
+  private readonly computeDiffStats: (
+    worktreePath: string,
+    baseRef: string | null
+  ) => Promise<DiffStatsComputation>;
   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;
+    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;
   }
 
   /**
    * Schedule a refresh for the given agent. No-op when a recent compute is
    * still warm; shares the in-flight promise when one is running.
    */
   signal(agentId: string): Promise<void> {
+    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 +130,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,7 +154,9 @@ export class DiffStatsRefresher {
   }
 
   private async refresh(agentId: string): Promise<void> {
+    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
@@ -136,16 +180,29 @@ 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);
       this.logger?.warn(
         { err, agentId },
         "Diff stats refresh failed; leaving cache unchanged"
       );
       return;
     }
 
+    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)
       : undefined;

apps/server/src/db/client.ts

@@ -25,3 +25,21 @@ export function createPool(config: AppConfig): Pool {
   pool.on("error", () => {});
   return pool;
 }
+
+/**
+ * Keep observability probes isolated from the application pool. Both acquiring
+ * the single probe connection and reading a result are bounded by the driver;
+ * ServiceResources adds its own watchdog and retires the client on timeout.
+ */
+export function createServiceResourcesProbePool(config: AppConfig): Pool {
+  const pool = new Pool({
+    connectionString: config.databaseUrl,
+    max: 1,
+    idleTimeoutMillis: 30_000,
+    connectionTimeoutMillis: 2_500,
+    query_timeout: 2_500,
+    allowExitOnIdle: true,
+  });
+  pool.on("error", () => {});
+  return pool;
+}

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 {

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<boolean> {
+  return (await getSetting(pool, SERVICE_RESOURCES_COLLECTION_KEY)) === "true";
+}
+
+export async function writeServiceResourcesCollectionEnabled(
+  pool: Pool,
+  enabled: boolean
+): Promise<void> {
+  await setSetting(
+    pool,
+    SERVICE_RESOURCES_COLLECTION_KEY,
+    enabled ? "true" : "false"
+  );
+}