feat: feed worker metrics into runtime router

.runtime/portal/RUNBOOK.txt

@@ -38,6 +38,8 @@ Streaming runtime operations:
   node scripts/runtime-worker-drain.mjs status
   node scripts/runtime-worker-drain.mjs reconcile
   node scripts/runtime-worker-drain.mjs reconcile --apply
+  node scripts/runtime-worker-metrics.mjs status
+  node scripts/runtime-worker-metrics.mjs sample
   node scripts/runtime-worker-drain.mjs drain goosed-3
   node scripts/runtime-worker-drain.mjs undrain goosed-3
   node scripts/check-tool-runtime.mjs

.runtime/portal/scripts/runtime-worker-drain.mjs

@@ -96,7 +96,11 @@ for (const id of [...new Set(workers)].sort()) {
     activeStreams > 0 &&
     lastStreamStartedAt &&
     staleAgeMs > staleMs &&
-    (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt),
+    (
+      !lastStreamEndedAt ||
+      lastStreamEndedAt >= lastStreamStartedAt ||
+      now - lastStreamEndedAt > staleMs
+    ),
   );
   if (action === 'reconcile' && stale && apply) {
     await client

.runtime/portal/scripts/runtime-worker-metrics.mjs

@@ -0,0 +1,171 @@
+#!/usr/bin/env node
+import fs from 'node:fs';
+import path from 'node:path';
+import { execFileSync } from 'node:child_process';
+import { createClient } from 'redis';
+
+function loadEnvFile(filePath) {
+  if (!fs.existsSync(filePath)) return;
+  for (const line of fs.readFileSync(filePath, 'utf8').split('\n')) {
+    const trimmed = line.trim();
+    if (!trimmed || trimmed.startsWith('#')) continue;
+    const idx = trimmed.indexOf('=');
+    if (idx < 0) continue;
+    const key = trimmed.slice(0, idx).trim();
+    const value = trimmed.slice(idx + 1).trim();
+    if (!process.env[key]) process.env[key] = value;
+  }
+}
+
+function sh(command) {
+  return execFileSync('/bin/zsh', ['-lc', command], { encoding: 'utf8' }).trim();
+}
+
+function parsePercent(value) {
+  const n = Number(String(value ?? '').replace('%', '').trim());
+  return Number.isFinite(n) ? n : 0;
+}
+
+function parseBytes(value) {
+  const match = String(value ?? '').trim().match(/^([\d.]+)\s*([kmgt]?i?b)?$/i);
+  if (!match) return 0;
+  const n = Number(match[1]);
+  if (!Number.isFinite(n)) return 0;
+  const unit = (match[2] || 'b').toLowerCase();
+  const scale = {
+    b: 1,
+    kb: 1e3,
+    mb: 1e6,
+    gb: 1e9,
+    tb: 1e12,
+    kib: 1024,
+    mib: 1024 ** 2,
+    gib: 1024 ** 3,
+    tib: 1024 ** 4,
+  }[unit] ?? 1;
+  return Math.round(n * scale);
+}
+
+function parseMemUsage(value) {
+  const [used, limit] = String(value ?? '').split('/').map((part) => part.trim());
+  return {
+    usedBytes: parseBytes(used),
+    limitBytes: parseBytes(limit),
+  };
+}
+
+function targetWorkerIds() {
+  const targets = (process.env.TKMIND_API_TARGETS || process.env.TKMIND_API_TARGET || '')
+    .split(',')
+    .map((value) => value.trim())
+    .filter(Boolean);
+  return targets.map((target, index) => ({
+    id: `goosed-${index + 1}`,
+    target,
+    container: `goosed-prod-${index + 1}`,
+  }));
+}
+
+function readDockerStats(container) {
+  try {
+    const line = sh(`docker stats --no-stream --format '{{json .}}' ${JSON.stringify(container)} 2>/dev/null`);
+    return line ? JSON.parse(line) : null;
+  } catch {
+    return null;
+  }
+}
+
+function readDockerInspect(container) {
+  try {
+    const line = sh(`docker inspect --format '{{json .State}}' ${JSON.stringify(container)} 2>/dev/null`);
+    return line ? JSON.parse(line) : null;
+  } catch {
+    return null;
+  }
+}
+
+function readContainerFdCount(container) {
+  try {
+    return Number(sh(`docker exec ${JSON.stringify(container)} sh -lc 'ls /proc/1/fd 2>/dev/null | wc -l'`)) || 0;
+  } catch {
+    return 0;
+  }
+}
+
+function workerKey(namespace, id, field) {
+  return [namespace, 'worker', id, field].join(':');
+}
+
+loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(process.cwd(), '.env'));
+
+const redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL || 'redis://127.0.0.1:6379/0';
+const namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE || 'memind:runtime';
+const action = process.argv[2] || 'sample';
+const dryRun = process.argv.includes('--dry-run') || action === 'status';
+const fdWarn = Number(process.env.GOOSED_FD_WARN ?? 180);
+const now = Date.now();
+
+if (!['sample', 'status'].includes(action)) {
+  console.error('Usage: node scripts/runtime-worker-metrics.mjs <sample|status> [--dry-run]');
+  process.exit(2);
+}
+
+const workers = [];
+for (const worker of targetWorkerIds()) {
+  const stats = readDockerStats(worker.container);
+  const state = readDockerInspect(worker.container);
+  const fdCount = readContainerFdCount(worker.container);
+  const mem = parseMemUsage(stats?.MemUsage);
+  const cpuLoad = parsePercent(stats?.CPUPerc);
+  const memoryPressure = parsePercent(stats?.MemPerc);
+  const fdPressure = fdWarn > 0 ? Number((fdCount / fdWarn).toFixed(4)) : 0;
+  workers.push({
+    ...worker,
+    ok: Boolean(stats && state?.Running),
+    health: state?.Health?.Status ?? state?.Status ?? null,
+    hostPid: Number(state?.Pid ?? 0) || null,
+    cpuLoad,
+    memoryPressure,
+    fdPressure,
+    fdCount,
+    pids: Number(stats?.PIDs ?? 0),
+    memUsedBytes: mem.usedBytes,
+    memLimitBytes: mem.limitBytes,
+    sampledAt: now,
+  });
+}
+
+if (!dryRun) {
+  const client = createClient({ url: redisUrl });
+  client.on('error', (err) => {
+    console.error(`Redis error: ${err instanceof Error ? err.message : err}`);
+  });
+  await client.connect();
+  for (const worker of workers) {
+    await client
+      .multi()
+      .set(workerKey(namespace, worker.id, 'cpu_load'), String(worker.cpuLoad))
+      .set(workerKey(namespace, worker.id, 'memory_pressure'), String(worker.memoryPressure))
+      .set(workerKey(namespace, worker.id, 'fd_pressure'), String(worker.fdPressure))
+      .set(workerKey(namespace, worker.id, 'fd_count'), String(worker.fdCount))
+      .set(workerKey(namespace, worker.id, 'container_pids'), String(worker.pids))
+      .set(workerKey(namespace, worker.id, 'container_name'), worker.container)
+      .set(workerKey(namespace, worker.id, 'container_health'), worker.health ?? '')
+      .set(workerKey(namespace, worker.id, 'container_host_pid'), String(worker.hostPid ?? ''))
+      .set(workerKey(namespace, worker.id, 'metrics_sampled_at'), String(worker.sampledAt))
+      .set(workerKey(namespace, worker.id, 'heartbeat'), String(now), { EX: 90 })
+      .exec();
+  }
+  await client.quit();
+}
+
+console.log(JSON.stringify({
+  ok: workers.every((worker) => worker.ok),
+  action,
+  dryRun,
+  namespace,
+  redisWrites: !dryRun,
+  workers,
+}, null, 2));
+
+process.exit(workers.every((worker) => worker.ok) ? 0 : 1);

.runtime/portal/server.mjs

@@ -8710,12 +8710,14 @@ function createRuntimeRouter({
       key("worker", workerId, "ewma_first_token_ms"),
       key("worker", workerId, "error_rate"),
       key("worker", workerId, "memory_pressure"),
+      key("worker", workerId, "cpu_load"),
+      key("worker", workerId, "fd_pressure"),
       key("worker", workerId, "drain")
     ]);
-    if (/^(1|true|yes)$/i.test(String(values[5] ?? ""))) return Number.POSITIVE_INFINITY;
-    return readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2;
+    if (/^(1|true|yes)$/i.test(String(values[7] ?? ""))) return Number.POSITIVE_INFINITY;
+    return readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2 + readNumber(values[5]) * 2 + readNumber(values[6]) * 2;
   };
-  const workerScoreFromValues = (values = []) => readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2;
+  const workerScoreFromValues = (values = []) => readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2 + readNumber(values[12]) * 2 + readNumber(values[13]) * 2;
   return {
     async pickTarget(fallbackPick, orderedTargets = targets) {
       const client = await getClient();
@@ -8783,7 +8785,13 @@ function createRuntimeRouter({
           key("worker", workerId, "stream_abort_count"),
           key("worker", workerId, "stream_error_count"),
           key("worker", workerId, "last_stream_started_at"),
-          key("worker", workerId, "last_stream_ended_at")
+          key("worker", workerId, "last_stream_ended_at"),
+          key("worker", workerId, "cpu_load"),
+          key("worker", workerId, "fd_pressure"),
+          key("worker", workerId, "fd_count"),
+          key("worker", workerId, "container_pids"),
+          key("worker", workerId, "container_health"),
+          key("worker", workerId, "metrics_sampled_at")
         ]).catch(() => []) : [];
         workers.push({
           id: workerId,
@@ -8800,6 +8808,12 @@ function createRuntimeRouter({
           streamErrorCount: readNumber(values?.[9]),
           lastStreamStartedAt: values?.[10] ? Number(values[10]) : null,
           lastStreamEndedAt: values?.[11] ? Number(values[11]) : null,
+          cpuLoad: readNumber(values?.[12]),
+          fdPressure: readNumber(values?.[13]),
+          fdCount: readNumber(values?.[14]),
+          containerPids: readNumber(values?.[15]),
+          containerHealth: values?.[16] ?? null,
+          metricsSampledAt: values?.[17] ? Number(values[17]) : null,
           score: workerScoreFromValues(values)
         });
       }

scripts/build-portal-runtime.mjs

@@ -275,6 +275,10 @@ async function writeMetadata() {
     path.join(root, 'scripts', 'runtime-worker-drain.mjs'),
     path.join(runtimeRoot, 'scripts', 'runtime-worker-drain.mjs'),
   );
+  await fs.copyFile(
+    path.join(root, 'scripts', 'runtime-worker-metrics.mjs'),
+    path.join(runtimeRoot, 'scripts', 'runtime-worker-metrics.mjs'),
+  );
   await fs.copyFile(
     path.join(root, 'scripts', 'check-tool-runtime.mjs'),
     path.join(runtimeRoot, 'scripts', 'check-tool-runtime.mjs'),
@@ -326,6 +330,8 @@ async function writeMetadata() {
       '  node scripts/runtime-worker-drain.mjs status',
       '  node scripts/runtime-worker-drain.mjs reconcile',
       '  node scripts/runtime-worker-drain.mjs reconcile --apply',
+      '  node scripts/runtime-worker-metrics.mjs status',
+      '  node scripts/runtime-worker-metrics.mjs sample',
       '  node scripts/runtime-worker-drain.mjs drain goosed-3',
       '  node scripts/runtime-worker-drain.mjs undrain goosed-3',
       '  node scripts/check-tool-runtime.mjs',

scripts/runtime-worker-drain.mjs

@@ -96,7 +96,11 @@ for (const id of [...new Set(workers)].sort()) {
     activeStreams > 0 &&
     lastStreamStartedAt &&
     staleAgeMs > staleMs &&
-    (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt),
+    (
+      !lastStreamEndedAt ||
+      lastStreamEndedAt >= lastStreamStartedAt ||
+      now - lastStreamEndedAt > staleMs
+    ),
   );
   if (action === 'reconcile' && stale && apply) {
     await client

scripts/runtime-worker-metrics.mjs

@@ -0,0 +1,171 @@
+#!/usr/bin/env node
+import fs from 'node:fs';
+import path from 'node:path';
+import { execFileSync } from 'node:child_process';
+import { createClient } from 'redis';
+
+function loadEnvFile(filePath) {
+  if (!fs.existsSync(filePath)) return;
+  for (const line of fs.readFileSync(filePath, 'utf8').split('\n')) {
+    const trimmed = line.trim();
+    if (!trimmed || trimmed.startsWith('#')) continue;
+    const idx = trimmed.indexOf('=');
+    if (idx < 0) continue;
+    const key = trimmed.slice(0, idx).trim();
+    const value = trimmed.slice(idx + 1).trim();
+    if (!process.env[key]) process.env[key] = value;
+  }
+}
+
+function sh(command) {
+  return execFileSync('/bin/zsh', ['-lc', command], { encoding: 'utf8' }).trim();
+}
+
+function parsePercent(value) {
+  const n = Number(String(value ?? '').replace('%', '').trim());
+  return Number.isFinite(n) ? n : 0;
+}
+
+function parseBytes(value) {
+  const match = String(value ?? '').trim().match(/^([\d.]+)\s*([kmgt]?i?b)?$/i);
+  if (!match) return 0;
+  const n = Number(match[1]);
+  if (!Number.isFinite(n)) return 0;
+  const unit = (match[2] || 'b').toLowerCase();
+  const scale = {
+    b: 1,
+    kb: 1e3,
+    mb: 1e6,
+    gb: 1e9,
+    tb: 1e12,
+    kib: 1024,
+    mib: 1024 ** 2,
+    gib: 1024 ** 3,
+    tib: 1024 ** 4,
+  }[unit] ?? 1;
+  return Math.round(n * scale);
+}
+
+function parseMemUsage(value) {
+  const [used, limit] = String(value ?? '').split('/').map((part) => part.trim());
+  return {
+    usedBytes: parseBytes(used),
+    limitBytes: parseBytes(limit),
+  };
+}
+
+function targetWorkerIds() {
+  const targets = (process.env.TKMIND_API_TARGETS || process.env.TKMIND_API_TARGET || '')
+    .split(',')
+    .map((value) => value.trim())
+    .filter(Boolean);
+  return targets.map((target, index) => ({
+    id: `goosed-${index + 1}`,
+    target,
+    container: `goosed-prod-${index + 1}`,
+  }));
+}
+
+function readDockerStats(container) {
+  try {
+    const line = sh(`docker stats --no-stream --format '{{json .}}' ${JSON.stringify(container)} 2>/dev/null`);
+    return line ? JSON.parse(line) : null;
+  } catch {
+    return null;
+  }
+}
+
+function readDockerInspect(container) {
+  try {
+    const line = sh(`docker inspect --format '{{json .State}}' ${JSON.stringify(container)} 2>/dev/null`);
+    return line ? JSON.parse(line) : null;
+  } catch {
+    return null;
+  }
+}
+
+function readContainerFdCount(container) {
+  try {
+    return Number(sh(`docker exec ${JSON.stringify(container)} sh -lc 'ls /proc/1/fd 2>/dev/null | wc -l'`)) || 0;
+  } catch {
+    return 0;
+  }
+}
+
+function workerKey(namespace, id, field) {
+  return [namespace, 'worker', id, field].join(':');
+}
+
+loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(process.cwd(), '.env'));
+
+const redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL || 'redis://127.0.0.1:6379/0';
+const namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE || 'memind:runtime';
+const action = process.argv[2] || 'sample';
+const dryRun = process.argv.includes('--dry-run') || action === 'status';
+const fdWarn = Number(process.env.GOOSED_FD_WARN ?? 180);
+const now = Date.now();
+
+if (!['sample', 'status'].includes(action)) {
+  console.error('Usage: node scripts/runtime-worker-metrics.mjs <sample|status> [--dry-run]');
+  process.exit(2);
+}
+
+const workers = [];
+for (const worker of targetWorkerIds()) {
+  const stats = readDockerStats(worker.container);
+  const state = readDockerInspect(worker.container);
+  const fdCount = readContainerFdCount(worker.container);
+  const mem = parseMemUsage(stats?.MemUsage);
+  const cpuLoad = parsePercent(stats?.CPUPerc);
+  const memoryPressure = parsePercent(stats?.MemPerc);
+  const fdPressure = fdWarn > 0 ? Number((fdCount / fdWarn).toFixed(4)) : 0;
+  workers.push({
+    ...worker,
+    ok: Boolean(stats && state?.Running),
+    health: state?.Health?.Status ?? state?.Status ?? null,
+    hostPid: Number(state?.Pid ?? 0) || null,
+    cpuLoad,
+    memoryPressure,
+    fdPressure,
+    fdCount,
+    pids: Number(stats?.PIDs ?? 0),
+    memUsedBytes: mem.usedBytes,
+    memLimitBytes: mem.limitBytes,
+    sampledAt: now,
+  });
+}
+
+if (!dryRun) {
+  const client = createClient({ url: redisUrl });
+  client.on('error', (err) => {
+    console.error(`Redis error: ${err instanceof Error ? err.message : err}`);
+  });
+  await client.connect();
+  for (const worker of workers) {
+    await client
+      .multi()
+      .set(workerKey(namespace, worker.id, 'cpu_load'), String(worker.cpuLoad))
+      .set(workerKey(namespace, worker.id, 'memory_pressure'), String(worker.memoryPressure))
+      .set(workerKey(namespace, worker.id, 'fd_pressure'), String(worker.fdPressure))
+      .set(workerKey(namespace, worker.id, 'fd_count'), String(worker.fdCount))
+      .set(workerKey(namespace, worker.id, 'container_pids'), String(worker.pids))
+      .set(workerKey(namespace, worker.id, 'container_name'), worker.container)
+      .set(workerKey(namespace, worker.id, 'container_health'), worker.health ?? '')
+      .set(workerKey(namespace, worker.id, 'container_host_pid'), String(worker.hostPid ?? ''))
+      .set(workerKey(namespace, worker.id, 'metrics_sampled_at'), String(worker.sampledAt))
+      .set(workerKey(namespace, worker.id, 'heartbeat'), String(now), { EX: 90 })
+      .exec();
+  }
+  await client.quit();
+}
+
+console.log(JSON.stringify({
+  ok: workers.every((worker) => worker.ok),
+  action,
+  dryRun,
+  namespace,
+  redisWrites: !dryRun,
+  workers,
+}, null, 2));
+
+process.exit(workers.every((worker) => worker.ok) ? 0 : 1);

tkmind-proxy.mjs

@@ -132,15 +132,19 @@ function createRuntimeRouter({
       key('worker', workerId, 'ewma_first_token_ms'),
       key('worker', workerId, 'error_rate'),
       key('worker', workerId, 'memory_pressure'),
+      key('worker', workerId, 'cpu_load'),
+      key('worker', workerId, 'fd_pressure'),
       key('worker', workerId, 'drain'),
     ]);
-    if (/^(1|true|yes)$/i.test(String(values[5] ?? ''))) return Number.POSITIVE_INFINITY;
+    if (/^(1|true|yes)$/i.test(String(values[7] ?? ''))) return Number.POSITIVE_INFINITY;
     return (
       readNumber(values[0]) * 3 +
       readNumber(values[1]) +
       readNumber(values[2]) * 0.01 +
       readNumber(values[3]) * 5 +
-      readNumber(values[4]) * 2
+      readNumber(values[4]) * 2 +
+      readNumber(values[5]) * 2 +
+      readNumber(values[6]) * 2
     );
   };
   const workerScoreFromValues = (values = []) => (
@@ -148,7 +152,9 @@ function createRuntimeRouter({
     readNumber(values[1]) +
     readNumber(values[2]) * 0.01 +
     readNumber(values[3]) * 5 +
-    readNumber(values[4]) * 2
+    readNumber(values[4]) * 2 +
+    readNumber(values[12]) * 2 +
+    readNumber(values[13]) * 2
   );
   return {
     async pickTarget(fallbackPick, orderedTargets = targets) {
@@ -233,6 +239,12 @@ function createRuntimeRouter({
                 key('worker', workerId, 'stream_error_count'),
                 key('worker', workerId, 'last_stream_started_at'),
                 key('worker', workerId, 'last_stream_ended_at'),
+                key('worker', workerId, 'cpu_load'),
+                key('worker', workerId, 'fd_pressure'),
+                key('worker', workerId, 'fd_count'),
+                key('worker', workerId, 'container_pids'),
+                key('worker', workerId, 'container_health'),
+                key('worker', workerId, 'metrics_sampled_at'),
               ])
               .catch(() => [])
           : [];
@@ -251,6 +263,12 @@ function createRuntimeRouter({
           streamErrorCount: readNumber(values?.[9]),
           lastStreamStartedAt: values?.[10] ? Number(values[10]) : null,
           lastStreamEndedAt: values?.[11] ? Number(values[11]) : null,
+          cpuLoad: readNumber(values?.[12]),
+          fdPressure: readNumber(values?.[13]),
+          fdCount: readNumber(values?.[14]),
+          containerPids: readNumber(values?.[15]),
+          containerHealth: values?.[16] ?? null,
+          metricsSampledAt: values?.[17] ? Number(values[17]) : null,
           score: workerScoreFromValues(values),
         });
       }