updates

server/db/index.js

@@ -71,6 +71,12 @@ export function getCorestore() {
 }
 
 export async function closeDb() {
+  try {
+    const { closeRemoteDbs } = await import('./remote.js')
+    await closeRemoteDbs()
+  } catch {
+    // ignore
+  }
   if (model) {
     await model.close().catch(() => {})
     model = null

server/db/model.js

@@ -101,6 +101,7 @@ export class PearDataModel extends ReadyResource {
         role: link.role || 'viewer',
         alias: link.alias || null,
         dbKeyHex: link.dbKeyHex || null,
+        discoveryKeyHex: link.discoveryKeyHex || null,
         syncMode: link.syncMode || 'both',
         linkedAt: link.linkedAt || Date.now(),
         lastSeen: link.lastSeen || null,

server/db/remote.js

@@ -0,0 +1,85 @@
+/**
+ * Open linked peer HyperDBs (read-only) after Corestore replication.
+ *
+ * Local agent always owns `peardata-meta`. Linked peers' bees are opened by
+ * `dbKeyHex` so warm `@peardata/metric-point` rows become queryable without
+ * re-scraping the remote host.
+ */
+import b4a from 'b4a'
+import { PearDataModel } from './model.js'
+import { getCorestore } from './index.js'
+import logger from '../utils/logger.js'
+
+const log = logger.child('db:remote')
+
+/** @type {Map<string, PearDataModel>} */
+const remotes = new Map()
+
+/**
+ * @param {string} dbKeyHex
+ * @returns {Promise<PearDataModel|null>}
+ */
+export async function openRemoteDb(dbKeyHex) {
+  const keyHex = String(dbKeyHex || '').toLowerCase()
+  if (!/^[0-9a-f]{64}$/.test(keyHex)) return null
+  if (remotes.has(keyHex)) return remotes.get(keyHex)
+
+  const store = getCorestore()
+  if (!store) {
+    log.warn('openRemoteDb: Corestore not open')
+    return null
+  }
+
+  const core = store.get({ key: b4a.from(keyHex, 'hex') })
+  const model = new PearDataModel(core, { writable: false, autoUpdate: true })
+  await model.ready()
+  remotes.set(keyHex, model)
+  log.info('Opened remote HyperDB', { dbKeyHex: keyHex.slice(0, 16) })
+  return model
+}
+
+/**
+ * @param {string} dbKeyHex
+ */
+export function getRemoteDb(dbKeyHex) {
+  return remotes.get(String(dbKeyHex || '').toLowerCase()) || null
+}
+
+export function listRemoteDbs() {
+  return [...remotes.entries()].map(([dbKeyHex, model]) => ({
+    dbKeyHex,
+    discoveryKeyHex: model.discoveryKeyHex,
+    length: model.db?.core?.length ?? null,
+  }))
+}
+
+/**
+ * Query warm metric points across all open remote DBs (first hit wins per call site).
+ * @param {{ chart: string, afterMs: number, beforeMs: number, limit?: number, tier?: number }} opts
+ */
+export async function queryRemoteMetricPoints(opts) {
+  /** @type {Array<{ chart: string, context: string, ts: number, tier: number, values: object }>} */
+  const all = []
+  for (const model of remotes.values()) {
+    try {
+      const rows = await model.queryMetricPoints(opts)
+      if (rows?.length) all.push(...rows)
+    } catch {
+      // ignore per-remote failures
+    }
+  }
+  all.sort((a, b) => a.ts - b.ts)
+  const limit = opts.limit || 10_000
+  return all.length > limit ? all.slice(-limit) : all
+}
+
+export async function closeRemoteDbs() {
+  for (const [key, model] of remotes) {
+    try {
+      await model.close()
+    } catch {
+      // ignore
+    }
+    remotes.delete(key)
+  }
+}

server/db/replicate.js

@@ -11,6 +11,8 @@
 import Hyperswarm from 'hyperswarm'
 import b4a from 'b4a'
 import { getCorestore, getDb } from './index.js'
+import { openRemoteDb } from './remote.js'
+import { getServerPublicKeyHex } from '../core/auth-keys.js'
 import logger from '../utils/logger.js'
 
 const log = logger.child('db:replicate')
@@ -61,6 +63,11 @@ export async function startReplication() {
     discoveryKeyHex: model.discoveryKeyHex,
     dbKeyHex: model.publicKeyHex,
   })
+
+  await rejoinLinkedPeers().catch((err) => {
+    log.warn('rejoinLinkedPeers failed', { error: err.message })
+  })
+
   return swarm
 }
 
@@ -85,6 +92,58 @@ export async function joinRemoteTopic(discoveryKeyHexOrBuf) {
   })
 }
 
+/**
+ * Persist + open a linked peer for warm pull.
+ * @param {{ discoveryKeyHex?: string|null, dbKeyHex?: string|null, syncMode?: string }} link
+ */
+export async function attachLinkedPeer(link) {
+  const mode = link.syncMode || 'both'
+  if (mode !== 'pull' && mode !== 'both') return { joined: false, opened: false }
+
+  let joined = false
+  if (isSwarmEnabled() && link.discoveryKeyHex) {
+    try {
+      await joinRemoteTopic(link.discoveryKeyHex)
+      joined = true
+    } catch (err) {
+      log.warn('joinRemoteTopic failed', { error: err.message })
+    }
+  }
+
+  let opened = false
+  if (link.dbKeyHex) {
+    const remote = await openRemoteDb(link.dbKeyHex)
+    opened = Boolean(remote)
+  }
+  return { joined, opened }
+}
+
+/**
+ * Re-join swarm topics + open remote bees from persisted peer-links (boot).
+ */
+export async function rejoinLinkedPeers() {
+  const db = getDb()
+  if (!db) return { links: 0, joined: 0, opened: 0 }
+  let localNodeId
+  try {
+    localNodeId = getServerPublicKeyHex().slice(0, 16)
+  } catch {
+    return { links: 0, joined: 0, opened: 0 }
+  }
+  const links = await db.listPeerLinks(localNodeId)
+  let joined = 0
+  let opened = 0
+  for (const link of links) {
+    const res = await attachLinkedPeer(link)
+    if (res.joined) joined++
+    if (res.opened) opened++
+  }
+  if (links.length) {
+    log.info('Rejoined linked peers', { links: links.length, joined, opened })
+  }
+  return { links: links.length, joined, opened }
+}
+
 export async function stopReplication() {
   if (!swarm) return
   try {

server/handlers/monitor.js

@@ -43,7 +43,9 @@ import {
 import { getJobs, knownJobNames } from '../services/jobs.js'
 import { formatAllMetrics } from '../rest/formatters.js'
 import { getDb } from '../db/index.js'
-import { joinRemoteTopic, isSwarmEnabled } from '../db/replicate.js'
+import { attachLinkedPeer, isSwarmEnabled } from '../db/replicate.js'
+import { listRemoteDbs } from '../db/remote.js'
+import { getParentCollector, isParentEnabled } from '../services/collectors/parent.js'
 
 /**
  * @param {import('../rpc/session.js').PeerSession} session
@@ -143,6 +145,7 @@ export function registerMonitorHandlers(session) {
       publicKeyHex: db.publicKeyHex,
       discoveryKeyHex: db.discoveryKeyHex,
       swarm: isSwarmEnabled(),
+      remotes: listRemoteDbs(),
     }
   })
 
@@ -163,20 +166,42 @@ export function registerMonitorHandlers(session) {
       role: args.role || 'viewer',
       alias: args.alias || null,
       dbKeyHex: args.dbKeyHex || null,
+      discoveryKeyHex: args.discoveryKeyHex || null,
       syncMode: args.syncMode || 'both',
     })
-    if (isSwarmEnabled() && args.discoveryKeyHex && (args.syncMode === 'pull' || args.syncMode === 'both')) {
-      try {
-        await joinRemoteTopic(args.discoveryKeyHex)
-      } catch (err) {
-        return {
-          success: true,
-          linked: true,
-          swarmWarning: err.message,
-        }
+    const attach = await attachLinkedPeer({
+      discoveryKeyHex: args.discoveryKeyHex || null,
+      dbKeyHex: args.dbKeyHex || null,
+      syncMode: args.syncMode || 'both',
+    })
+    return {
+      success: true,
+      linked: true,
+      swarmJoined: attach.joined,
+      remoteOpened: attach.opened,
+    }
+  })
+
+  session.respond('getFleetHealth', async () => {
+    if (!isParentEnabled()) {
+      return {
+        enabled: false,
+        local: anomalies.getHealth(),
+        children: [],
       }
     }
-    return { success: true, linked: true }
+    const parent = getParentCollector()
+    return {
+      enabled: true,
+      local: anomalies.getHealth(),
+      children: parent.listChildren(),
+      summary: parent.fleetSummary(),
+    }
+  })
+
+  session.respond('listChildPeers', async () => {
+    if (!isParentEnabled()) return { enabled: false, children: [] }
+    return { enabled: true, children: getParentCollector().listChildren() }
   })
 
   session.respond('unlinkPeer', async (args) => {
@@ -262,12 +287,26 @@ export function registerMonitorHandlers(session) {
     return { success: true, peerId: args.peerId }
   })
 
-  session.respond('exportSnapshot', async () => ({
-    success: true,
-    node: collector.getNodeInfo(getServerPublicKeyHex(), APP_VERSION),
-    latest: store.latestValues(),
-    health: anomalies.getHealth(),
-    alerts: listAlerts(),
-    ts: Date.now(),
-  }))
+  session.respond('exportSnapshot', async (args) => {
+    const { buildSnapshot, writeSnapshotFile, toPrometheusText, pushPrometheusText } =
+      await import('../services/export.js')
+    const snapshot = buildSnapshot()
+    const written = args?.write ? writeSnapshotFile(snapshot) : { path: null, bytes: 0 }
+    let push = null
+    if (args?.push || args?.pushUrl || process.env.PEARDATA_PUSHGATEWAY_URL) {
+      try {
+        const url = args?.pushUrl || process.env.PEARDATA_PUSHGATEWAY_URL
+        if (url) {
+          push = await pushPrometheusText(url, toPrometheusText(snapshot.latest))
+        }
+      } catch (err) {
+        push = { error: err.message }
+      }
+    }
+    return {
+      ...snapshot,
+      file: written.path,
+      push,
+    }
+  })
 }

server/pipeline.js

@@ -13,6 +13,22 @@ import {
   getProcessCollector,
   isProcessCollectorEnabled,
 } from './services/collectors/processes.js'
+import {
+  getParentCollector,
+  isParentEnabled,
+} from './services/collectors/parent.js'
+import {
+  getNginxCollector,
+  isNginxEnabled,
+} from './services/collectors/nginx.js'
+import {
+  getRedisCollector,
+  isRedisEnabled,
+} from './services/collectors/redis.js'
+import {
+  getPostgresCollector,
+  isPostgresEnabled,
+} from './services/collectors/postgres.js'
 import { getStore } from './services/store.js'
 import { getAnomalyEngine } from './services/anomaly.js'
 import {
@@ -91,10 +107,45 @@ export function startPipeline() {
     processes.start()
   }
 
+  let parent = null
+  if (isParentEnabled()) {
+    parent = getParentCollector()
+    parent.on('samples', (batch) => ingestBatch(store, anomalies, batch))
+    parent.start()
+  }
+
+  let nginx = null
+  if (isNginxEnabled()) {
+    nginx = getNginxCollector()
+    nginx.on('samples', (batch) => ingestBatch(store, anomalies, batch))
+    nginx.on('error', (err) => log.warn('Nginx collector error', { error: err.message }))
+    nginx.start()
+  }
+
+  let redis = null
+  if (isRedisEnabled()) {
+    redis = getRedisCollector()
+    redis.on('samples', (batch) => ingestBatch(store, anomalies, batch))
+    redis.on('error', (err) => log.warn('Redis collector error', { error: err.message }))
+    redis.start()
+  }
+
+  let postgres = null
+  if (isPostgresEnabled()) {
+    postgres = getPostgresCollector()
+    postgres.on('samples', (batch) => ingestBatch(store, anomalies, batch))
+    postgres.on('error', (err) => log.warn('Postgres collector error', { error: err.message }))
+    postgres.start()
+  }
+
   log.info('Metrics pipeline started', {
     hyperdb: Boolean(getDb()),
     docker: Boolean(docker),
     processes: Boolean(processes),
+    parent: Boolean(parent),
+    nginx: Boolean(nginx),
+    redis: Boolean(redis),
+    postgres: Boolean(postgres),
   })
-  return { collector, store, anomalies, docker, processes }
+  return { collector, store, anomalies, docker, processes, parent, nginx, redis, postgres }
 }

server/rest/routes.js

@@ -20,6 +20,8 @@ import { formatAllMetrics } from './formatters.js'
 import { peers } from '../core/peer-registry.js'
 import { getDb, isHyperDbEnabled } from '../db/index.js'
 import { isSwarmEnabled } from '../db/replicate.js'
+import { listRemoteDbs } from '../db/remote.js'
+import { getParentCollector, isParentEnabled } from '../services/collectors/parent.js'
 
 /**
  * @param {string} pathname
@@ -52,11 +54,23 @@ export async function handleRest(pathname, query) {
 
   // ── nodes ────────────────────────────────────────────────
   if (path === '/api/v2/nodes' || path === '/api/v3/nodes') {
-    const node = nodePayload()
-    return json({ nodes: [node], ...([node][0] && {}) })
+    const nodes = allNodePayloads()
+    return json({ nodes, count: nodes.length })
   }
   if (path === '/api/v3/node_instances') {
-    return json({ nodes: [nodePayload()] })
+    return json({ nodes: allNodePayloads() })
+  }
+  if (path === '/api/v3/fleet') {
+    if (!isParentEnabled()) {
+      return json({ enabled: false, local: nodePayload(), children: [] })
+    }
+    const parent = getParentCollector()
+    return json({
+      enabled: true,
+      local: nodePayload(),
+      children: parent.listChildren(),
+      summary: parent.fleetSummary(),
+    })
   }
 
   // ── contexts ─────────────────────────────────────────────
@@ -223,7 +237,27 @@ export async function handleRest(pathname, query) {
 
   // ── functions / settings stubs ───────────────────────────
   if (path === '/api/v3/functions' || path === '/api/v2/functions') {
-    return json({ functions: [{ name: 'collectOnce' }, { name: 'snapshot' }, { name: 'gcBuffers' }] })
+    return json({
+      functions: [
+        { name: 'collectOnce' },
+        { name: 'snapshot' },
+        { name: 'exportSnapshot' },
+        { name: 'prometheusPush' },
+        { name: 'gcBuffers' },
+      ],
+    })
+  }
+  if (path === '/api/v3/export' || path === '/api/v2/export') {
+    const { buildSnapshot, writeSnapshotFile, toPrometheusText } = await import(
+      '../services/export.js'
+    )
+    const format = (query.get('format') || 'json').toLowerCase()
+    const snapshot = buildSnapshot()
+    if (query.get('write') === '1') writeSnapshotFile(snapshot)
+    if (format === 'prometheus') {
+      return { status: 200, contentType: 'text/plain; version=0.0.4', body: toPrometheusText(snapshot.latest) }
+    }
+    return json(snapshot)
   }
   if (path === '/api/v3/settings' || path === '/api/v3/config') {
     return json({
@@ -234,16 +268,27 @@ export async function handleRest(pathname, query) {
     })
   }
   if (path === '/api/v3/stream_path') {
-    return json({
-      path: [
-        {
-          node: getServerPublicKeyHex(),
-          hostname: os.hostname(),
-          hops: 0,
-          role: 'agent',
-        },
-      ],
-    })
+    const pathNodes = [
+      {
+        node: getServerPublicKeyHex(),
+        hostname: os.hostname(),
+        hops: 0,
+        role: isParentEnabled() ? 'parent' : 'agent',
+      },
+    ]
+    if (isParentEnabled()) {
+      for (const child of getParentCollector().listChildren()) {
+        pathNodes.push({
+          node: child.publicKeyHex,
+          hostname: child.hostname || child.shortId,
+          hops: 1,
+          role: 'child',
+          connected: child.connected,
+          health: child.health,
+        })
+      }
+    }
+    return json({ path: pathNodes })
   }
 
   // ── health / root ────────────────────────────────────────
@@ -258,6 +303,7 @@ export async function handleRest(pathname, query) {
       publicKeyHex: db.publicKeyHex,
       discoveryKeyHex: db.discoveryKeyHex,
       swarm: isSwarmEnabled(),
+      remotes: listRemoteDbs(),
       collections: [
         '@peardata/node',
         '@peardata/peer-link',
@@ -331,9 +377,30 @@ function nodePayload() {
       nm: os.release(),
     },
     st: 'online',
+    role: isParentEnabled() ? 'parent' : 'agent',
   }
 }
 
+function allNodePayloads() {
+  const nodes = [nodePayload()]
+  if (isParentEnabled()) {
+    for (const child of getParentCollector().listChildren()) {
+      nodes.push({
+        nm: child.hostname || child.shortId,
+        nd: child.publicKeyHex.slice(0, 16),
+        guid: child.publicKeyHex,
+        st: child.connected ? 'online' : 'offline',
+        role: 'child',
+        health: child.health,
+        cpu: child.cpu,
+        ram: child.ram,
+        lastSeen: child.lastSeen,
+      })
+    }
+  }
+  return nodes
+}
+
 function json(body) {
   return { status: 200, contentType: 'application/json', body }
 }

server/server.js

@@ -155,6 +155,38 @@ async function shutdown() {
   } catch {
     // ignore
   }
+  try {
+    const { getParentCollector, isParentEnabled } = await import(
+      './services/collectors/parent.js'
+    )
+    if (isParentEnabled()) getParentCollector().stop()
+  } catch {
+    // ignore
+  }
+  try {
+    const { getNginxCollector, isNginxEnabled } = await import(
+      './services/collectors/nginx.js'
+    )
+    if (isNginxEnabled()) getNginxCollector().stop()
+  } catch {
+    // ignore
+  }
+  try {
+    const { getRedisCollector, isRedisEnabled } = await import(
+      './services/collectors/redis.js'
+    )
+    if (isRedisEnabled()) getRedisCollector().stop()
+  } catch {
+    // ignore
+  }
+  try {
+    const { getPostgresCollector, isPostgresEnabled } = await import(
+      './services/collectors/postgres.js'
+    )
+    if (isPostgresEnabled()) getPostgresCollector().stop()
+  } catch {
+    // ignore
+  }
   try {
     await flushWarmPending()
   } catch {

server/services/collectors/nginx.js

@@ -0,0 +1,179 @@
+/**
+ * Nginx stub_status collector (service plugin spike).
+ *
+ * Enable: PEARDATA_NGINX=1
+ * URL:    PEARDATA_NGINX_URL=http://127.0.0.1/nginx_status
+ *
+ * Charts: nginx.connections, nginx.requests
+ */
+import http from 'http'
+import https from 'https'
+import { CollectorPlugin } from './plugin.js'
+import { registerChart } from '../../../shared/metrics.js'
+import logger from '../../utils/logger.js'
+
+const log = logger.child('nginx')
+
+const CHART_CONNECTIONS = {
+  id: 'nginx.connections',
+  name: 'nginx.connections',
+  context: 'nginx.connections',
+  title: 'Nginx connections',
+  units: 'connections',
+  family: 'nginx',
+  chartType: 'line',
+  priority: 8000,
+  plugin: 'nginx',
+  dimensions: [
+    { id: 'active', name: 'active', algorithm: 'absolute' },
+    { id: 'reading', name: 'reading', algorithm: 'absolute' },
+    { id: 'writing', name: 'writing', algorithm: 'absolute' },
+    { id: 'waiting', name: 'waiting', algorithm: 'absolute' },
+  ],
+}
+
+const CHART_REQUESTS = {
+  id: 'nginx.requests',
+  name: 'nginx.requests',
+  context: 'nginx.requests',
+  title: 'Nginx requests',
+  units: 'requests/s',
+  family: 'nginx',
+  chartType: 'line',
+  priority: 8010,
+  plugin: 'nginx',
+  dimensions: [
+    { id: 'accepts', name: 'accepts', algorithm: 'incremental' },
+    { id: 'handled', name: 'handled', algorithm: 'incremental' },
+    { id: 'requests', name: 'requests', algorithm: 'incremental' },
+  ],
+}
+
+export function isNginxEnabled() {
+  const v = process.env.PEARDATA_NGINX
+  return v === '1' || v === 'on' || v === 'true'
+}
+
+/**
+ * Parse nginx stub_status body.
+ * @param {string} body
+ * @returns {{ active: number, accepts: number, handled: number, requests: number, reading: number, writing: number, waiting: number }|null}
+ */
+export function parseNginxStubStatus(body) {
+  const text = String(body || '')
+  const active = text.match(/Active connections:\s*(\d+)/i)
+  const counters = text.match(/^\s*(\d+)\s+(\d+)\s+(\d+)\s*$/m)
+  const states = text.match(/Reading:\s*(\d+)\s+Writing:\s*(\d+)\s+Waiting:\s*(\d+)/i)
+  if (!active || !counters || !states) return null
+  return {
+    active: Number(active[1]),
+    accepts: Number(counters[1]),
+    handled: Number(counters[2]),
+    requests: Number(counters[3]),
+    reading: Number(states[1]),
+    writing: Number(states[2]),
+    waiting: Number(states[3]),
+  }
+}
+
+function fetchText(url, timeoutMs = 3000) {
+  return new Promise((resolve, reject) => {
+    const mod = String(url).startsWith('https') ? https : http
+    const req = mod.get(url, (res) => {
+      let body = ''
+      res.on('data', (c) => {
+        body += c
+      })
+      res.on('end', () => {
+        if (res.statusCode && res.statusCode >= 400) {
+          reject(new Error(`HTTP ${res.statusCode}`))
+          return
+        }
+        resolve(body)
+      })
+    })
+    req.setTimeout(timeoutMs, () => {
+      req.destroy()
+      reject(new Error('timeout'))
+    })
+    req.on('error', reject)
+  })
+}
+
+export class NginxCollector extends CollectorPlugin {
+  constructor(opts = {}) {
+    super({ name: 'nginx', intervalMs: opts.intervalMs })
+    this.url = opts.url || process.env.PEARDATA_NGINX_URL || 'http://127.0.0.1/nginx_status'
+    /** @type {{ accepts: number, handled: number, requests: number, wallMs: number }|null} */
+    this._prev = null
+  }
+
+  isEnabled() {
+    return isNginxEnabled()
+  }
+
+  start() {
+    if (!this.isEnabled()) return
+    registerChart(CHART_CONNECTIONS)
+    registerChart(CHART_REQUESTS)
+    log.info('Nginx collector started', { url: this.url })
+    super.start()
+  }
+
+  async collect() {
+    const body = await fetchText(this.url)
+    const parsed = parseNginxStubStatus(body)
+    if (!parsed) throw new Error('unrecognized stub_status body')
+    const ts = Date.now()
+
+    let acceptsRate = 0
+    let handledRate = 0
+    let requestsRate = 0
+    if (this._prev && ts > this._prev.wallMs) {
+      const dt = (ts - this._prev.wallMs) / 1000
+      if (dt > 0) {
+        acceptsRate = Math.max(0, (parsed.accepts - this._prev.accepts) / dt)
+        handledRate = Math.max(0, (parsed.handled - this._prev.handled) / dt)
+        requestsRate = Math.max(0, (parsed.requests - this._prev.requests) / dt)
+      }
+    }
+    this._prev = {
+      accepts: parsed.accepts,
+      handled: parsed.handled,
+      requests: parsed.requests,
+      wallMs: ts,
+    }
+
+    return [
+      {
+        chart: 'nginx.connections',
+        context: 'nginx.connections',
+        ts,
+        values: {
+          active: parsed.active,
+          reading: parsed.reading,
+          writing: parsed.writing,
+          waiting: parsed.waiting,
+        },
+      },
+      {
+        chart: 'nginx.requests',
+        context: 'nginx.requests',
+        ts,
+        values: {
+          accepts: acceptsRate,
+          handled: handledRate,
+          requests: requestsRate,
+        },
+      },
+    ]
+  }
+}
+
+/** @type {NginxCollector|null} */
+let singleton = null
+
+export function getNginxCollector() {
+  if (!singleton) singleton = new NginxCollector()
+  return singleton
+}

server/services/collectors/parent.js

@@ -0,0 +1,334 @@
+/**
+ * Parent peer fleet aggregator (Phase 4 / M6 spike).
+ *
+ * Enable: PEARDATA_PARENT=1
+ * Children: PEARDATA_PARENT_PEERS=hex,hex  (64-char agent public keys)
+ * Optional: PEARDATA_PARENT_SEED (admin proof), PEARDATA_PARENT_POLL_MS (default 5000)
+ *
+ * Dials child agents over HyperDHT, polls getHealth + getAllMetrics,
+ * and emits namespaced fleet charts into the local pipeline:
+ *   fleet.cpu          — per-child CPU %
+ *   fleet.ram          — per-child RAM used MiB
+ *   fleet.children     — connected / configured counts
+ */
+import fs from 'fs'
+import { EventEmitter } from 'events'
+import { PearDataConnection } from '../../../client/connection.js'
+import { Methods } from '../../../shared/protocol.js'
+import { registerChart } from '../../../shared/metrics.js'
+import logger from '../../utils/logger.js'
+
+const log = logger.child('parent')
+
+export function isParentEnabled() {
+  const v = process.env.PEARDATA_PARENT
+  return v === '1' || v === 'on' || v === 'true'
+}
+
+/**
+ * @returns {string[]}
+ */
+export function parseParentPeers() {
+  const raw = process.env.PEARDATA_PARENT_PEERS || ''
+  const fromEnv = raw
+    .split(/[,\s]+/)
+    .map((s) => s.trim().toLowerCase())
+    .filter((s) => /^[0-9a-f]{64}$/.test(s))
+
+  const file = process.env.PEARDATA_PARENT_PEERS_FILE
+  if (!file) return [...new Set(fromEnv)]
+  try {
+    const text = fs.readFileSync(file, 'utf8')
+    const fromFile = text
+      .split(/\r?\n/)
+      .map((l) => l.replace(/#.*$/, '').trim().toLowerCase())
+      .filter((s) => /^[0-9a-f]{64}$/.test(s))
+    return [...new Set([...fromEnv, ...fromFile])]
+  } catch {
+    return [...new Set(fromEnv)]
+  }
+}
+
+function shortId(pk) {
+  return String(pk).slice(0, 12)
+}
+
+/**
+ * @param {string[]} childIds
+ * @param {'cpu'|'ram'} kind
+ */
+function registerFleetChart(childIds, kind) {
+  const dims = childIds.map((id) => ({
+    id,
+    name: id,
+    algorithm: 'absolute',
+  }))
+  if (!dims.length) dims.push({ id: '_none', name: '_none', algorithm: 'absolute' })
+  const def =
+    kind === 'cpu'
+      ? {
+          id: 'fleet.cpu',
+          name: 'fleet.cpu',
+          context: 'fleet.cpu',
+          title: 'Fleet CPU (children)',
+          units: 'percentage',
+          family: 'fleet',
+          chartType: 'line',
+          priority: 7000,
+          plugin: 'parent',
+          dimensions: dims,
+        }
+      : {
+          id: 'fleet.ram',
+          name: 'fleet.ram',
+          context: 'fleet.ram',
+          title: 'Fleet RAM used (children)',
+          units: 'MiB',
+          family: 'fleet',
+          chartType: 'line',
+          priority: 7010,
+          plugin: 'parent',
+          dimensions: dims,
+        }
+  registerChart(def)
+  return def
+}
+
+function registerChildrenChart() {
+  const def = {
+    id: 'fleet.children',
+    name: 'fleet.children',
+    context: 'fleet.children',
+    title: 'Fleet child agents',
+    units: 'agents',
+    family: 'fleet',
+    chartType: 'line',
+    priority: 7020,
+    plugin: 'parent',
+    dimensions: [
+      { id: 'connected', name: 'connected', algorithm: 'absolute' },
+      { id: 'configured', name: 'configured', algorithm: 'absolute' },
+    ],
+  }
+  registerChart(def)
+  return def
+}
+
+/**
+ * Extract CPU used % and RAM MiB from getAllMetrics / latest-style payload.
+ * @param {any} metrics
+ */
+export function extractChildStats(metrics) {
+  /** @type {{ cpu: number|null, ram: number|null }} */
+  const out = { cpu: null, ram: null }
+  const root = metrics?.body || metrics
+  const charts = root?.charts || root?.latest || root || {}
+  const cpu = charts['system.cpu']
+  const ram = charts['system.ram']
+  const cpuVals = cpu?.dimensions || cpu?.values || cpu
+  const ramVals = ram?.dimensions || ram?.values || ram
+  if (cpuVals && typeof cpuVals === 'object') {
+    const idle = Number(cpuVals.idle?.value ?? cpuVals.idle)
+    if (Number.isFinite(idle)) out.cpu = Math.max(0, 100 - idle)
+  }
+  if (ramVals && typeof ramVals === 'object') {
+    const used = Number(ramVals.used?.value ?? ramVals.used)
+    if (Number.isFinite(used)) out.ram = used
+  }
+  return out
+}
+
+export class ParentCollector extends EventEmitter {
+  constructor(opts = {}) {
+    super()
+    this.pollMs = opts.pollMs || Number(process.env.PEARDATA_PARENT_POLL_MS) || 5000
+    this.adminSeed = opts.adminSeed || process.env.PEARDATA_PARENT_SEED || null
+    this.peerKeys = opts.peers || parseParentPeers()
+    /** @type {Map<string, { conn: PearDataConnection|null, connected: boolean, hostname: string|null, health: string|null, cpu: number|null, ram: number|null, lastError: string|null, lastSeen: number|null }>} */
+    this.children = new Map()
+    this._timer = null
+    this._dialing = false
+  }
+
+  start() {
+    if (this._timer) return
+    for (const pk of this.peerKeys) {
+      this.children.set(pk, {
+        conn: null,
+        connected: false,
+        hostname: null,
+        health: null,
+        cpu: null,
+        ram: null,
+        lastError: null,
+        lastSeen: null,
+      })
+    }
+    registerChildrenChart()
+    registerFleetChart(this.peerKeys.map(shortId), 'cpu')
+    registerFleetChart(this.peerKeys.map(shortId), 'ram')
+    log.info('Parent collector started', { children: this.peerKeys.length, pollMs: this.pollMs })
+    this._tick()
+    this._timer = setInterval(() => this._tick(), this.pollMs)
+    if (typeof this._timer.unref === 'function') this._timer.unref()
+  }
+
+  stop() {
+    if (this._timer) {
+      clearInterval(this._timer)
+      this._timer = null
+    }
+    for (const [pk, st] of this.children) {
+      if (st.conn) {
+        st.conn.destroy().catch(() => {})
+        st.conn = null
+      }
+      st.connected = false
+      this.children.set(pk, st)
+    }
+  }
+
+  listChildren() {
+    return [...this.children.entries()].map(([publicKeyHex, st]) => ({
+      publicKeyHex,
+      shortId: shortId(publicKeyHex),
+      connected: st.connected,
+      hostname: st.hostname,
+      health: st.health,
+      cpu: st.cpu,
+      ram: st.ram,
+      lastError: st.lastError,
+      lastSeen: st.lastSeen,
+    }))
+  }
+
+  fleetSummary() {
+    const kids = this.listChildren()
+    const connected = kids.filter((k) => k.connected).length
+    const cpus = kids.map((k) => k.cpu).filter((n) => n != null)
+    const avgCpu = cpus.length ? cpus.reduce((a, b) => a + b, 0) / cpus.length : null
+    return {
+      configured: kids.length,
+      connected,
+      avgCpu,
+      status: connected === 0 ? 'offline' : connected < kids.length ? 'degraded' : 'ok',
+    }
+  }
+
+  async _ensureDial(pk) {
+    const st = this.children.get(pk)
+    if (!st) return null
+    if (st.conn?.connected) return st.conn
+    try {
+      if (st.conn) await st.conn.destroy().catch(() => {})
+      const conn = new PearDataConnection(pk, {
+        adminSeed: this.adminSeed,
+        timeoutMs: 20_000,
+      })
+      await conn.connect()
+      st.conn = conn
+      st.connected = true
+      st.lastError = null
+      conn.on('disconnected', () => {
+        st.connected = false
+        st.conn = null
+      })
+      this.children.set(pk, st)
+      log.info('Dialed child', { peer: shortId(pk) })
+      return conn
+    } catch (err) {
+      st.connected = false
+      st.conn = null
+      st.lastError = err.message
+      this.children.set(pk, st)
+      return null
+    }
+  }
+
+  async _tick() {
+    if (this._dialing) return
+    this._dialing = true
+    try {
+      const ts = Date.now()
+      /** @type {Record<string, number|null>} */
+      const cpuValues = {}
+      /** @type {Record<string, number|null>} */
+      const ramValues = {}
+      let connected = 0
+
+      for (const pk of this.peerKeys) {
+        const sid = shortId(pk)
+        cpuValues[sid] = null
+        ramValues[sid] = null
+        const conn = await this._ensureDial(pk)
+        const st = this.children.get(pk)
+        if (!conn || !st) continue
+        try {
+          const [health, node, metrics] = await Promise.all([
+            conn.request(Methods.getHealth, {}),
+            conn.request(Methods.getNodeInfo, {}),
+            conn.request(Methods.getAllMetrics, { format: 'json' }),
+          ])
+          const stats = extractChildStats(metrics)
+          st.connected = true
+          st.health = health?.status || 'ok'
+          st.hostname = node?.hostname || null
+          st.cpu = stats.cpu
+          st.ram = stats.ram
+          st.lastSeen = ts
+          st.lastError = null
+          cpuValues[sid] = stats.cpu
+          ramValues[sid] = stats.ram
+          connected++
+          this.children.set(pk, st)
+        } catch (err) {
+          st.connected = false
+          st.lastError = err.message
+          this.children.set(pk, st)
+          try {
+            await conn.destroy()
+          } catch {
+            // ignore
+          }
+          st.conn = null
+        }
+      }
+
+      const ids = this.peerKeys.map(shortId)
+      registerFleetChart(ids, 'cpu')
+      registerFleetChart(ids, 'ram')
+      registerChildrenChart()
+
+      this.emit('samples', [
+        {
+          chart: 'fleet.cpu',
+          context: 'fleet.cpu',
+          ts,
+          values: cpuValues,
+        },
+        {
+          chart: 'fleet.ram',
+          context: 'fleet.ram',
+          ts,
+          values: ramValues,
+        },
+        {
+          chart: 'fleet.children',
+          context: 'fleet.children',
+          ts,
+          values: { connected, configured: this.peerKeys.length },
+        },
+      ])
+    } finally {
+      this._dialing = false
+    }
+  }
+}
+
+/** @type {ParentCollector|null} */
+let singleton = null
+
+export function getParentCollector() {
+  if (!singleton) singleton = new ParentCollector()
+  return singleton
+}

server/services/collectors/plugin.js

@@ -0,0 +1,53 @@
+/**
+ * Minimal collector plugin base (Phase 3 / EXTENDING.md).
+ *
+ * Subclasses implement `collect()` → sample batch, call `start()` to schedule.
+ */
+import { EventEmitter } from 'events'
+import { SAMPLE_INTERVAL_MS } from '../../../shared/metrics.js'
+
+export class CollectorPlugin extends EventEmitter {
+  /**
+   * @param {{ name: string, intervalMs?: number }} opts
+   */
+  constructor(opts) {
+    super()
+    this.name = opts.name || 'plugin'
+    this.intervalMs = opts.intervalMs || Number(process.env.PEARDATA_SAMPLE_MS) || SAMPLE_INTERVAL_MS
+    this._timer = null
+  }
+
+  /** @returns {boolean} */
+  isEnabled() {
+    return false
+  }
+
+  /**
+   * @returns {Promise<Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>>|Array}
+   */
+  async collect() {
+    return []
+  }
+
+  start() {
+    if (this._timer || !this.isEnabled()) return
+    const tick = async () => {
+      try {
+        const batch = await this.collect()
+        if (batch?.length) this.emit('samples', batch)
+      } catch (err) {
+        this.emit('error', err)
+      }
+    }
+    tick()
+    this._timer = setInterval(tick, this.intervalMs)
+    if (typeof this._timer.unref === 'function') this._timer.unref()
+  }
+
+  stop() {
+    if (this._timer) {
+      clearInterval(this._timer)
+      this._timer = null
+    }
+  }
+}

server/services/collectors/postgres.js

@@ -0,0 +1,194 @@
+/**
+ * Postgres collector (service plugin spike).
+ *
+ * Enable: PEARDATA_POSTGRES=1
+ * TCP:    PEARDATA_POSTGRES_HOST / PEARDATA_POSTGRES_PORT (default 5432)
+ * Optional HTTP stats (key=value lines): PEARDATA_POSTGRES_STATS_URL
+ *
+ * Charts:
+ *   postgres.up          — 1/0 + connect latency
+ *   postgres.stats       — from HTTP stats URL when set (connections, xact, tuples)
+ */
+import net from 'net'
+import http from 'http'
+import https from 'https'
+import { CollectorPlugin } from './plugin.js'
+import { registerChart } from '../../../shared/metrics.js'
+import logger from '../../utils/logger.js'
+
+const log = logger.child('postgres')
+
+const CHART_UP = {
+  id: 'postgres.up',
+  name: 'postgres.up',
+  context: 'postgres.up',
+  title: 'Postgres availability',
+  units: 'boolean',
+  family: 'postgres',
+  chartType: 'line',
+  priority: 8200,
+  plugin: 'postgres',
+  dimensions: [
+    { id: 'up', name: 'up', algorithm: 'absolute' },
+    { id: 'latency_ms', name: 'latency_ms', algorithm: 'absolute' },
+  ],
+}
+
+const CHART_STATS = {
+  id: 'postgres.stats',
+  name: 'postgres.stats',
+  context: 'postgres.stats',
+  title: 'Postgres stats',
+  units: 'count',
+  family: 'postgres',
+  chartType: 'line',
+  priority: 8210,
+  plugin: 'postgres',
+  dimensions: [
+    { id: 'connections', name: 'connections', algorithm: 'absolute' },
+    { id: 'xact_commit', name: 'xact_commit', algorithm: 'absolute' },
+    { id: 'xact_rollback', name: 'xact_rollback', algorithm: 'absolute' },
+    { id: 'tuples_returned', name: 'tuples_returned', algorithm: 'absolute' },
+  ],
+}
+
+export function isPostgresEnabled() {
+  const v = process.env.PEARDATA_POSTGRES
+  return v === '1' || v === 'on' || v === 'true'
+}
+
+/**
+ * Parse simple key=value postgres stats (custom exporter / sidecar).
+ * @param {string} body
+ * @returns {Record<string, number>}
+ */
+export function parsePostgresStats(body) {
+  /** @type {Record<string, number>} */
+  const out = {}
+  for (const line of String(body || '').split(/\r?\n/)) {
+    const t = line.trim()
+    if (!t || t.startsWith('#')) continue
+    const m = t.match(/^([a-zA-Z0-9_]+)\s*[=:]\s*([0-9.]+)/)
+    if (m) out[m[1]] = Number(m[2])
+  }
+  return out
+}
+
+/**
+ * @param {{ host: string, port: number }} addr
+ * @param {number} [timeoutMs]
+ * @returns {Promise<{ up: number, latency_ms: number }>}
+ */
+export function probePostgresTcp(addr, timeoutMs = 3000) {
+  return new Promise((resolve) => {
+    const started = Date.now()
+    const socket = net.createConnection({ host: addr.host, port: addr.port })
+    let settled = false
+    const finish = (up) => {
+      if (settled) return
+      settled = true
+      clearTimeout(timer)
+      try {
+        socket.destroy()
+      } catch {
+        // ignore
+      }
+      resolve({ up, latency_ms: Date.now() - started })
+    }
+    const timer = setTimeout(() => finish(0), timeoutMs)
+    socket.on('connect', () => finish(1))
+    socket.on('error', () => finish(0))
+  })
+}
+
+function fetchText(url, timeoutMs = 3000) {
+  return new Promise((resolve, reject) => {
+    const mod = String(url).startsWith('https') ? https : http
+    const req = mod.get(url, (res) => {
+      let body = ''
+      res.on('data', (c) => {
+        body += c
+      })
+      res.on('end', () => {
+        if (res.statusCode && res.statusCode >= 400) {
+          reject(new Error(`HTTP ${res.statusCode}`))
+          return
+        }
+        resolve(body)
+      })
+    })
+    req.setTimeout(timeoutMs, () => {
+      req.destroy()
+      reject(new Error('timeout'))
+    })
+    req.on('error', reject)
+  })
+}
+
+export class PostgresCollector extends CollectorPlugin {
+  constructor(opts = {}) {
+    super({ name: 'postgres', intervalMs: opts.intervalMs })
+    this.host = opts.host || process.env.PEARDATA_POSTGRES_HOST || '127.0.0.1'
+    this.port = Number(opts.port || process.env.PEARDATA_POSTGRES_PORT) || 5432
+    this.statsUrl = opts.statsUrl || process.env.PEARDATA_POSTGRES_STATS_URL || ''
+  }
+
+  isEnabled() {
+    return isPostgresEnabled()
+  }
+
+  start() {
+    if (!this.isEnabled()) return
+    registerChart(CHART_UP)
+    if (this.statsUrl) registerChart(CHART_STATS)
+    log.info('Postgres collector started', {
+      host: this.host,
+      port: this.port,
+      statsUrl: this.statsUrl || null,
+    })
+    super.start()
+  }
+
+  async collect() {
+    const ts = Date.now()
+    const probe = await probePostgresTcp({ host: this.host, port: this.port })
+    /** @type {Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>} */
+    const batch = [
+      {
+        chart: 'postgres.up',
+        context: 'postgres.up',
+        ts,
+        values: { up: probe.up, latency_ms: probe.latency_ms },
+      },
+    ]
+    if (this.statsUrl) {
+      try {
+        const body = await fetchText(this.statsUrl)
+        const s = parsePostgresStats(body)
+        registerChart(CHART_STATS)
+        batch.push({
+          chart: 'postgres.stats',
+          context: 'postgres.stats',
+          ts,
+          values: {
+            connections: s.connections ?? s.numbackends ?? null,
+            xact_commit: s.xact_commit ?? null,
+            xact_rollback: s.xact_rollback ?? null,
+            tuples_returned: s.tuples_returned ?? s.tup_returned ?? null,
+          },
+        })
+      } catch (err) {
+        log.warn('Postgres stats URL failed', { error: err.message })
+      }
+    }
+    return batch
+  }
+}
+
+/** @type {PostgresCollector|null} */
+let singleton = null
+
+export function getPostgresCollector() {
+  if (!singleton) singleton = new PostgresCollector()
+  return singleton
+}

server/services/collectors/redis.js

@@ -0,0 +1,227 @@
+/**
+ * Redis INFO collector (service plugin).
+ *
+ * Enable: PEARDATA_REDIS=1
+ * Addr:   PEARDATA_REDIS_URL=redis://127.0.0.1:6379  (or host:port)
+ *
+ * Charts: redis.memory, redis.clients, redis.stats
+ */
+import net from 'net'
+import { CollectorPlugin } from './plugin.js'
+import { registerChart } from '../../../shared/metrics.js'
+import logger from '../../utils/logger.js'
+
+const log = logger.child('redis')
+
+const CHART_MEMORY = {
+  id: 'redis.memory',
+  name: 'redis.memory',
+  context: 'redis.memory',
+  title: 'Redis memory',
+  units: 'MiB',
+  family: 'redis',
+  chartType: 'area',
+  priority: 8100,
+  plugin: 'redis',
+  dimensions: [
+    { id: 'used', name: 'used', algorithm: 'absolute' },
+    { id: 'peak', name: 'peak', algorithm: 'absolute' },
+    { id: 'rss', name: 'rss', algorithm: 'absolute' },
+  ],
+}
+
+const CHART_CLIENTS = {
+  id: 'redis.clients',
+  name: 'redis.clients',
+  context: 'redis.clients',
+  title: 'Redis clients',
+  units: 'clients',
+  family: 'redis',
+  chartType: 'line',
+  priority: 8110,
+  plugin: 'redis',
+  dimensions: [
+    { id: 'connected', name: 'connected', algorithm: 'absolute' },
+    { id: 'blocked', name: 'blocked', algorithm: 'absolute' },
+  ],
+}
+
+const CHART_STATS = {
+  id: 'redis.stats',
+  name: 'redis.stats',
+  context: 'redis.stats',
+  title: 'Redis ops',
+  units: 'ops/s',
+  family: 'redis',
+  chartType: 'line',
+  priority: 8120,
+  plugin: 'redis',
+  dimensions: [
+    { id: 'ops', name: 'ops', algorithm: 'absolute' },
+    { id: 'hit_rate', name: 'hit_rate', algorithm: 'absolute' },
+  ],
+}
+
+export function isRedisEnabled() {
+  const v = process.env.PEARDATA_REDIS
+  return v === '1' || v === 'on' || v === 'true'
+}
+
+/**
+ * @param {string} urlOrHost
+ * @returns {{ host: string, port: number }}
+ */
+export function parseRedisAddr(urlOrHost) {
+  const raw = String(urlOrHost || '127.0.0.1:6379').trim()
+  if (raw.includes('://')) {
+    try {
+      const u = new URL(raw)
+      return {
+        host: u.hostname || '127.0.0.1',
+        port: Number(u.port) || 6379,
+      }
+    } catch {
+      // fall through
+    }
+  }
+  const [host, port] = raw.split(':')
+  return { host: host || '127.0.0.1', port: Number(port) || 6379 }
+}
+
+/**
+ * @param {string} body Redis INFO text
+ * @returns {Record<string, string>}
+ */
+export function parseRedisInfo(body) {
+  /** @type {Record<string, string>} */
+  const out = {}
+  for (const line of String(body || '').split(/\r?\n/)) {
+    if (!line || line.startsWith('#')) continue
+    const i = line.indexOf(':')
+    if (i < 0) continue
+    out[line.slice(0, i)] = line.slice(i + 1).trim()
+  }
+  return out
+}
+
+function bytesToMiB(n) {
+  return n / (1024 * 1024)
+}
+
+/**
+ * @param {{ host: string, port: number }} addr
+ * @param {number} [timeoutMs]
+ * @returns {Promise<string>}
+ */
+export function fetchRedisInfo(addr, timeoutMs = 3000) {
+  return new Promise((resolve, reject) => {
+    const socket = net.createConnection({ host: addr.host, port: addr.port })
+    let buf = ''
+    let settled = false
+    const done = (err, data) => {
+      if (settled) return
+      settled = true
+      clearTimeout(timer)
+      try {
+        socket.destroy()
+      } catch {
+        // ignore
+      }
+      if (err) reject(err)
+      else resolve(data)
+    }
+    const timer = setTimeout(() => done(new Error('redis timeout')), timeoutMs)
+    socket.on('connect', () => {
+      socket.write('INFO\r\n')
+    })
+    socket.on('data', (chunk) => {
+      buf += chunk.toString('utf8')
+      // RESP bulk: $<len>\r\n<body>\r\n  or plain INFO dump
+      if (buf.includes('redis_version:') || buf.includes('used_memory:')) {
+        const idx = buf.indexOf('$')
+        if (idx === 0) {
+          const nl = buf.indexOf('\r\n')
+          if (nl > 0) {
+            const body = buf.slice(nl + 2)
+            if (body.includes('redis_version:') || body.length > 200) done(null, body)
+          }
+        } else {
+          done(null, buf)
+        }
+      }
+    })
+    socket.on('error', (err) => done(err))
+    socket.on('end', () => {
+      if (buf) done(null, buf)
+      else done(new Error('redis closed with no data'))
+    })
+  })
+}
+
+export class RedisCollector extends CollectorPlugin {
+  constructor(opts = {}) {
+    super({ name: 'redis', intervalMs: opts.intervalMs })
+    this.addr = parseRedisAddr(opts.url || process.env.PEARDATA_REDIS_URL || '127.0.0.1:6379')
+  }
+
+  isEnabled() {
+    return isRedisEnabled()
+  }
+
+  start() {
+    if (!this.isEnabled()) return
+    registerChart(CHART_MEMORY)
+    registerChart(CHART_CLIENTS)
+    registerChart(CHART_STATS)
+    log.info('Redis collector started', this.addr)
+    super.start()
+  }
+
+  async collect() {
+    const raw = await fetchRedisInfo(this.addr)
+    const info = parseRedisInfo(raw)
+    const ts = Date.now()
+    const hits = Number(info.keyspace_hits) || 0
+    const misses = Number(info.keyspace_misses) || 0
+    const denom = hits + misses
+    const hitRate = denom > 0 ? (hits / denom) * 100 : 0
+    return [
+      {
+        chart: 'redis.memory',
+        context: 'redis.memory',
+        ts,
+        values: {
+          used: bytesToMiB(Number(info.used_memory) || 0),
+          peak: bytesToMiB(Number(info.used_memory_peak) || 0),
+          rss: bytesToMiB(Number(info.used_memory_rss) || 0),
+        },
+      },
+      {
+        chart: 'redis.clients',
+        context: 'redis.clients',
+        ts,
+        values: {
+          connected: Number(info.connected_clients) || 0,
+          blocked: Number(info.blocked_clients) || 0,
+        },
+      },
+      {
+        chart: 'redis.stats',
+        context: 'redis.stats',
+        ts,
+        values: {
+          ops: Number(info.instantaneous_ops_per_sec) || 0,
+          hit_rate: hitRate,
+        },
+      },
+    ]
+  }
+}
+
+/** @type {RedisCollector|null} */
+let singleton = null
+
+export function getRedisCollector() {
+  if (!singleton) singleton = new RedisCollector()
+  return singleton
+}

server/services/export.js

@@ -0,0 +1,173 @@
+/**
+ * Snapshot export + Prometheus text push (Pushgateway-compatible).
+ *
+ * Jobs:
+ *   exportSnapshot     — build JSON snapshot (optional write to PEARDATA_EXPORT_DIR)
+ *   prometheusPush     — POST exposition text to PEARDATA_PUSHGATEWAY_URL
+ *
+ * Env:
+ *   PEARDATA_EXPORT_DIR          — directory for snapshot-*.json
+ *   PEARDATA_PUSHGATEWAY_URL     — e.g. http://127.0.0.1:9091/metrics/job/peardata
+ */
+import fs from 'fs'
+import path from 'path'
+import http from 'http'
+import https from 'https'
+import os from 'os'
+import b4a from 'b4a'
+import { getStore } from './store.js'
+import { getCollector } from './collector.js'
+import { getAnomalyEngine } from './anomaly.js'
+import { listAlerts } from './alerts.js'
+import { getServerPublicKeyHex } from '../core/auth-keys.js'
+import { APP_NAME, APP_VERSION } from '../../shared/protocol.js'
+import { CHART_BY_ID } from '../../shared/metrics.js'
+import logger from '../utils/logger.js'
+
+const log = logger.child('export')
+
+/**
+ * @returns {object}
+ */
+export function buildSnapshot() {
+  const store = getStore()
+  const collector = getCollector()
+  const pk = (() => {
+    try {
+      return getServerPublicKeyHex()
+    } catch {
+      return null
+    }
+  })()
+  return {
+    success: true,
+    app: APP_NAME,
+    version: APP_VERSION,
+    hostname: os.hostname(),
+    publicKeyHex: pk,
+    node: collector.getNodeInfo(pk, APP_VERSION),
+    latest: store.latestValues(),
+    health: getAnomalyEngine().getHealth(),
+    alerts: listAlerts(),
+    ts: Date.now(),
+  }
+}
+
+/**
+ * Convert latest values to Prometheus exposition format.
+ * @param {Record<string, { ts: number, values: Record<string, number|null> }>} [latest]
+ */
+export function toPrometheusText(latest) {
+  const data = latest || getStore().latestValues()
+  const lines = [
+    `# HELP peardata_info PearData agent info`,
+    `# TYPE peardata_info gauge`,
+    `peardata_info{version="${APP_VERSION}",hostname="${os.hostname()}"} 1`,
+  ]
+  for (const [chart, point] of Object.entries(data)) {
+    const metric = `peardata_${chart.replace(/[^a-zA-Z0-9_]/g, '_')}`
+    const ctx = CHART_BY_ID.get(chart)?.context || chart
+    for (const [dim, val] of Object.entries(point.values || {})) {
+      if (val == null || Number.isNaN(val)) continue
+      lines.push(
+        `${metric}{dimension="${dim}",context="${ctx}"} ${val}`
+      )
+    }
+  }
+  return lines.join('\n') + '\n'
+}
+
+/**
+ * @param {object} snapshot
+ * @returns {{ path: string|null, bytes: number }}
+ */
+export function writeSnapshotFile(snapshot) {
+  const dir = process.env.PEARDATA_EXPORT_DIR
+  if (!dir) return { path: null, bytes: 0 }
+  fs.mkdirSync(dir, { recursive: true })
+  const file = path.join(dir, `snapshot-${snapshot.ts || Date.now()}.json`)
+  const body = JSON.stringify(snapshot, null, 2)
+  fs.writeFileSync(file, body)
+  return { path: file, bytes: body.length }
+}
+
+/**
+ * POST Prometheus text to Pushgateway (or any text receiver).
+ * @param {string} url
+ * @param {string} body
+ * @param {number} [timeoutMs]
+ */
+export function pushPrometheusText(url, body, timeoutMs = 10_000) {
+  return new Promise((resolve, reject) => {
+    const u = new URL(url)
+    const mod = u.protocol === 'https:' ? https : http
+    const req = mod.request(
+      {
+        hostname: u.hostname,
+        port: u.port || (u.protocol === 'https:' ? 443 : 80),
+        path: u.pathname + u.search,
+        method: 'POST',
+        headers: {
+          'Content-Type': 'text/plain; version=0.0.4',
+          'Content-Length': b4a.byteLength(body),
+        },
+        timeout: timeoutMs,
+      },
+      (res) => {
+        let data = ''
+        res.on('data', (c) => {
+          data += c
+        })
+        res.on('end', () => {
+          if (res.statusCode && res.statusCode >= 400) {
+            reject(new Error(`push failed HTTP ${res.statusCode}: ${data.slice(0, 200)}`))
+            return
+          }
+          resolve({ status: res.statusCode || 200, bytes: body.length })
+        })
+      }
+    )
+    req.on('timeout', () => {
+      req.destroy()
+      reject(new Error('push timeout'))
+    })
+    req.on('error', reject)
+    req.write(body)
+    req.end()
+  })
+}
+
+/**
+ * Job handler: build + optional file write.
+ */
+export async function jobExportSnapshot() {
+  const snapshot = buildSnapshot()
+  const written = writeSnapshotFile(snapshot)
+  if (written.path) log.info('Wrote snapshot', written)
+  return {
+    ok: true,
+    ts: snapshot.ts,
+    charts: Object.keys(snapshot.latest || {}).length,
+    file: written.path,
+    bytes: written.bytes,
+    snapshot: written.path ? undefined : snapshot,
+  }
+}
+
+/**
+ * Job handler: push Prometheus text to gateway.
+ * @param {{ url?: string }} [args]
+ */
+export async function jobPrometheusPush(args = {}) {
+  const url = args.url || process.env.PEARDATA_PUSHGATEWAY_URL
+  if (!url) {
+    return {
+      ok: false,
+      error: 'PEARDATA_PUSHGATEWAY_URL (or args.url) required',
+    }
+  }
+  const body = toPrometheusText()
+  const res = await pushPrometheusText(url, body)
+  log.info('Prometheus push ok', { url, ...res })
+  return { ok: true, ...res, url, lines: body.split('\n').length }
+}

server/services/jobs.js

@@ -7,6 +7,7 @@ import { Pushes } from '../../shared/protocol.js'
 import { peers } from '../core/peer-registry.js'
 import { getCollector } from './collector.js'
 import { getStore } from './store.js'
+import { jobExportSnapshot, jobPrometheusPush } from './export.js'
 
 const JOB_HANDLERS = {
   collectOnce: async () => {
@@ -17,6 +18,8 @@ const JOB_HANDLERS = {
   snapshot: async () => {
     return { ok: true, latest: getStore().latestValues() }
   },
+  exportSnapshot: jobExportSnapshot,
+  prometheusPush: jobPrometheusPush,
   gcBuffers: async () => {
     // ring buffers self-trim; placeholder for future disk GC
     return { ok: true }

server/services/store.js

@@ -9,6 +9,7 @@
 import { EventEmitter } from 'events'
 import { SAMPLE_INTERVAL_MS, CHART_BY_ID, chartSummary } from '../../shared/metrics.js'
 import { getDb } from '../db/index.js'
+import { queryRemoteMetricPoints, listRemoteDbs } from '../db/remote.js'
 
 function envInt(name, fallback) {
   const n = Number(process.env[name])
@@ -160,6 +161,26 @@ export class MetricStore extends EventEmitter {
           // keep memory result
         }
       }
+      // Linked peer warm pull (replicated Corestore / remote bee)
+      if ((!windowed.length || source !== 'hyperdb-warm') && listRemoteDbs().length) {
+        try {
+          const remote = await queryRemoteMetricPoints({
+            chart,
+            afterMs,
+            beforeMs,
+            limit: Math.max(opts.points || 60, 10_000),
+            tier: 1,
+          })
+          if (remote.length) {
+            if (!windowed.length || remote.length >= windowed.length || windowExceedsMemory) {
+              windowed = remote
+              source = 'hyperdb-remote'
+            }
+          }
+        } catch {
+          // keep prior result
+        }
+      }
     }
 
     if (!windowed.length && src.length) {