Runtime Orchestrator 深度解析

24import { spawn } from 'node:child_process'\n25import { env } from '../../env.js'\n26import { pool } from '../../db/pool.js'\n27import { sub2apiConfigured, sub2apiOpenAIBaseURL } from '../../sub2api.js'\n28import { inprocClient } from './inproc-client.js'\n29import { signAgentToken } from './jwt.js'\n30import { notifyAlert } from '../../alerting.js'\n31import { Semaphore } from '../../concurrency.js'\n32\n33const KUBECTL = process.env.CUMORA_KUBECTL ?? 'kubectl'\n34/** kubectl context to target. In dev we pin to OrbStack so a stray\n35 *  `kubectl config use-context gke-...` on the dev laptop can't cause\n36 *  the dev server to spawn agent Pods in a real cluster. In production\n37 *  we leave it unset and let kubectl resolve the current context\n38 *  (typically the in-cluster SA when the server runs in K8s). Override\n39 *  the dev default with CUMORA_KUBECTL_CONTEXT (e.g. if your local\n40 *  context is named differently). */\n41const KUBECTL_CONTEXT = process.env.CUMORA_KUBECTL_CONTEXT\n42  ?? (env.NODE_ENV === 'production' ? '' : 'orbstack')\n43const IMAGE = process.env.CUMORA_AGENT_COMPUTER_IMAGE\n44  ?? 'quay.io/yetoneful/cumora-agent-computer:dev'\n45/** TTL of the JWT minted at Pod-spawn time. Pods live up to\n46 *  CUMORA_AGENT_IDLE_MS without restart, so the JWT lives longer than\n47 *  the idle ceiling to avoid mid-life expiry. */\n48const TOKEN_TTL_SECONDS = Number(process.env.CUMORA_AGENT_TOKEN_TTL_SECONDS ?? 24 * 60 * 60)

95function kubectl(args: string[], opts: KubectlOpts = {}): Promise<KubectlResult> {\n96  // Hold a permit for the FULL lifetime of the child (incl. timeout) so\n97  // the in-flight process count — not just the spawn rate — is bounded.\n98  return kubectlSem.run(() => kubectlSpawn(args, opts))\n99}\n100\n101function kubectlSpawn(args: string[], opts: KubectlOpts = {}): Promise<KubectlResult> {\n102  const timeoutMs = opts.timeoutMs ?? KUBECTL_DEFAULT_TIMEOUT_MS\n103  return new Promise((resolve) => {\n104    const ctxArgs = KUBECTL_CONTEXT === '' ? [] : ['--context', KUBECTL_CONTEXT]\n105    const child = spawn(KUBECTL, [...ctxArgs, '-n', NS, ...args], { stdio: ['pipe', 'pipe', 'pipe'] })\n106    let out = ''\n107    let err = ''\n108    let settled = false\n109    let killTimer: NodeJS.Timeout | null = null\n110    const finish = (r: KubectlResult): void => {\n111      if (settled) return\n112      settled = true\n113      if (killTimer) { clearTimeout(killTimer); killTimer = null }\n114      if (timer) { clearTimeout(timer); }\n115      resolve(r)\n116    }\n117    const timer = setTimeout(() => {\n118      if (settled) return\n119      // Try graceful first, then force.\n120      try { child.kill('SIGTERM') } catch { /* already gone */ }\n121      killTimer = setTimeout(() => {\n122        try { child.kill('SIGKILL') } catch { /* already gone */ }\n123      }, KUBECTL_KILL_GRACE_MS)\n124      // Resolve immediately rather than waiting for exit — the caller\n125      // is treating this as "kubectl is unresponsive", and the child's\n126      // exit event will fire later (we ignore it via `settled`).\n127      finish({\n128        code: 124, // curl/timeout(1) convention for "timed out"\n129        out,\n130        err: err + `\n[kubectl] killed after ${timeoutMs}ms timeout`,\n131        timedOut: true,\n132      })\n133    }, timeoutMs)\n134    timer.unref?.()\n135    child.stdout.on('data', (d: Buffer) => { out += d.toString() })\n136    child.stderr.on('data', (d: Buffer) => { err += d.toString() })\n137    child.on('error', (e) => finish({\n138      code: 1, out, err: err + (e.message ?? String(e)), timedOut: false,\n139    }))\n140    child.on('exit', (code) => finish({\n141      code: code ?? 1, out, err, timedOut: false,\n142    }))\n143    try {\n144      if (opts.stdin !== undefined) child.stdin.end(opts.stdin)\n145      else child.stdin.end()\n146    } catch (e) {\n147      // EPIPE if the child died before we wrote stdin — covered by\n148      // the error/exit handlers above.\n149      void e\n150    }\n151  })

294  return `apiVersion: v1\n295kind: Pod\n296metadata:\n297  name: ${podName(args.agentId)}\n298  labels:\n299    app: cumora-agent\n300    cumora.agent: ${yamlQuote(labelValue)}\n301spec:\n302  restartPolicy: OnFailure${saBlock}${pullSecretsBlock}\n303  containers:\n304  - name: agent-computer\n305    image: ${yamlQuote(args.image)}\n306    imagePullPolicy: IfNotPresent\n307    securityContext:\n308      # FUSE mount needs three things to work on GKE-COS:\n309      #   1. CAP_SYS_ADMIN — for the mount(2) syscall.\n310      #   2. /dev/fuse in the cgroup device whitelist — granted by the\n311      #      generic-device-plugin DaemonSet (squat/generic-device-\n312      #      plugin) which we depend on being installed in the cluster\n313      #      (see server/k8s/generic-device-plugin.note).\n314      #   3. AppArmor unconfined — the default cri-containerd profile\n315      #      denies the mount syscall even with CAP_SYS_ADMIN, so the\n316      #      pod must opt out. Without this the agent boots but\n317      #      fusermount3 fails with "Permission denied" and /workspace\n318      #      falls through to the empty backing dir — the agent loses\n319      #      access to its persona files / memory FS.\n320      appArmorProfile:\n321        type: Unconfined\n322      capabilities:\n323        add: ["SYS_ADMIN"]\n324    resources:\n325      requests:\n326        # Measured in prod, per-pod working-set ≈ 420 MiB, but only\n327        # ~263 MiB is anon (irreducible) — the other ~137 MiB is\n328        # file-backed page cache that the kernel can evict under\n329        # memory pressure. Setting memory request to 320 MiB covers\n330        # the irreducible portion with comfortable buffer; under load\n331        # the limit (1.5 GiB) absorbs bursts. CPU between wakes is\n332        # <1% — 50m is a generous reservation for a process that\n333        # spends most of its life as an idle SSE listener; bursts to\n334        # 2 cores during page render are bounded by the limit (2500m),\n335        # not the request. Density per e2-standard-2 (1180m CPU / ~6\n336        # GiB usable after kubelet reserve): ~18 pods CPU-wise, ~18\n337        # pods memory-wise — balanced, no single-resource binding.\n338        cpu: "50m"\n339        memory: "320Mi"\n340        devic.es/fuse: 1\n341      limits:\n342        # Measured under realistic workload (4 sites opened + extracted\n343        # + an adapter call): memory peaked at ~1230MiB; CPU spiked to\n344        # 2.2 cores during page rendering. 1.5Gi memory limit leaves\n345        # ~250Mi headroom over peak; 2500m CPU avoids throttling\n346        # Chromium during JS-heavy page loads.\n347        cpu: "2500m"\n348        memory: "1.5Gi"\n349        # Reserves one /dev/fuse slot from the device plugin; the\n350        # kubelet inserts the device into the container's cgroup.\n351        # Resource name comes from squat/generic-device-plugin's\n352        # default config — it uses the 'devic.es' domain.\n353        devic.es/fuse: 1\n354    volumeMounts:\n355    - name: chrome-profile\n356      mountPath: /opt/chrome-profile\n357    env:\n358    - name: CUMORA_AGENT_ID\n359      value: ${yamlQuote(args.agentId)}\n360    - name: CUMORA_AGENT_RUNTIME_URL\n361      value: ${yamlQuote(args.serverUrl)}\n362    - name: CUMORA_AGENT_IDLE_MS\n363      value: ${yamlQuote(String(args.idleMs))}\n364    - name: CUMORA_AGENT_NO_WORK_MS\n365      value: ${yamlQuote(String(args.noWorkMs))}\n366    - name: CUMORA_PERSONA_DIR\n367      value: "/workspace"\n368    - name: CUMORA_AGENT_RUNTIME_TOKEN\n369      value: |-\n370${indent(args.token)}\n371    - name: OPENAI_API_KEY\n372      value: |-\n373${indent(args.openaiKey)}\n374    - name: OPENAI_BASE_URL\n375      value: |-\n376${indent(args.openaiBaseUrl)}

429// ─── cluster-wide FUSE admission control ─────────────────────────────\n430//\n431// Triggered by the FUSE-cap incident on 2026-05-20: agent pods request\n432// `devic.es/fuse: 1` (one slot per pod), and each prod node advertises\n433// 800 slots via the generic-device-plugin DaemonSet. The app-level\n434// admission cap is intentionally lower (AGENT_POD_ADMISSION_MAX,\n435// default 40) because CPU, memory, API-server churn, and provider\n436// concurrency are the real production ceiling.\n437//\n438// `ensurePod` must REFUSE rather than blindly apply when the cluster\n439// is near saturation. Refusing lets pending wakes survive in their\n440// inbox (message.new) or get retried by their scheduler (idle /\n441// scanner); blowing past the cap, conversely, leaves pods stuck in\n442// FailedScheduling for tens of minutes with no recovery.\n443\n444export interface ClusterFuseUtilization {\n445  /** Number of cumora-agent pods currently using a fuse slot\n446   *  (Pending + Running, excluding Succeeded/Failed). */\n447  used: number\n448  /** Effective admission cap: min(cluster `devic.es/fuse` capacity,\n449   *  AGENT_POD_ADMISSION_MAX when that env is positive). */\n450  cap: number\n451  /** used / cap. 0 when cap is 0 (no nodes report fuse). */\n452  ratio: number\n453  /** True when result came from the in-memory cache. */\n454  cached: boolean\n455}\n456\n457/** Pure parser — extracted so the unit test can pin both branches\n458 *  (well-formed JSON, malformed) without spawning kubectl. */\n459export function parseClusterFuse(nodesJson: string, podsJson: string): { used: number; cap: number } {\n460  let cap = 0\n461  try {\n462    const d = JSON.parse(nodesJson) as { items?: Array<{ status?: { capacity?: Record<string, string> } }> }\n463    for (const n of d.items ?? []) {\n464      const slots = Number(n.status?.capacity?.['devic.es/fuse'] ?? 0)\n465      if (Number.isFinite(slots) && slots > 0) cap += slots\n466    }\n467  } catch { /* ignore — return 0 */ }\n468  let used = 0\n469  try {\n470    const d = JSON.parse(podsJson) as { items?: Array<{ status?: { phase?: string } }> }\n471    for (const p of d.items ?? []) {\n472      const phase = p.status?.phase\n473      // Pending pods that haven't been bound yet don't actually hold\n474      // a device slot, but they DO represent demand the scheduler is\n475      // trying to fulfill — count them so we don't over-spawn into a\n476      // queue. Succeeded/Failed pods released their slot at exit.\n477      if (phase === 'Running' || phase === 'Pending') used++\n478    }\n479  } catch { /* ignore — return 0 */ }\n480  return { used, cap }\n481}\n482\n483interface FuseUtilCache { ts: number; used: number; cap: number }\n484let fuseUtilCache: FuseUtilCache | null = null\n485\n486/** TTL on the cluster-wide fuse sample. Short enough that bursts get\n487 *  re-evaluated quickly; long enough that we don't hammer the K8s API\n488 *  on every wake. Each successful spawn bumps `used` in-memory so the\n489 *  cached value tracks intra-window growth without a refresh. */\n490const FUSE_CACHE_TTL_MS = 10_000\n491\n492/** Threshold above which `ensurePod` refuses new spawns. 0.90 leaves\n493 *  a 10% buffer for in-flight scheduling latency and inFlight-dedup\n494 *  raciness. Below 0.90 we let everything through. */\n495const FUSE_ADMISSION_THRESHOLD = 0.90\n496\n497/** Sample (or read from cache) the cluster's cumora-agent fuse usage.\n498 *  Fails open: on kubectl error returns `cap=Infinity` so callers\n499 *  never block on transient cluster-state read failures. */\n500export async function getClusterFuseUtilization(): Promise<ClusterFuseUtilization> {\n501  const now = Date.now()\n502  if (fuseUtilCache && now - fuseUtilCache.ts < FUSE_CACHE_TTL_MS) {\n503    const { used, cap } = fuseUtilCache\n504    return { used, cap, ratio: cap > 0 ? used / cap : 0, cached: true }\n505  }\n506  const [nodes, pods] = await Promise.all([\n507    kubectlWithRetry(['get', 'nodes', '-o', 'json'], { timeoutMs: 10_000 }),\n508    kubectlWithRetry(\n509      ['get', 'pods', '-l', 'app=cumora-agent', '-o', 'json'],\n510      { timeoutMs: 10_000 },\n511    ),\n512  ])\n513  if (nodes.code !== 0 || pods.code !== 0) {\n514    // Fail open — we'd rather over-spawn during a kubectl flake than\n515    // wedge every wake. Don't cache this so the next call retries.\n516    return { used: 0, cap: Number.POSITIVE_INFINITY, ratio: 0, cached: false }\n517  }\n518  const parsed = parseClusterFuse(nodes.out, pods.out)\n519  const appCap = env.AGENT_POD_ADMISSION_MAX\n520  const cap = appCap > 0 ? Math.min(parsed.cap, appCap) : parsed.cap\n521  fuseUtilCache = { ts: now, used: parsed.used, cap }\n522  return { used: parsed.used, cap, ratio: cap > 0 ? parsed.used / cap : 0, cached: false }\n523}